commit d117d4379ffff7371e93bb9d1f403fdf4d20fcc6
Author: Jens Kristoffersson <jenkr@vgregion.se>
AuthorDate: Tue Sep 8 13:43:46 2026 +0200
Commit: Jens Kristoffersson <jenkr@vgregion.se>
CommitDate: Tue Sep 8 13:43:46 2026 +0200
Akta grafmotor (fas 1): ny nod/kant-motor + migrering
Floden lagras och kors nu som en riktig DAG av noder och kanter i stallet
for en nastlad {steps}-lista. Grunden for fri kabeldragning i UI:t.
- rpa/graph_engine.py (ny): run_graph() foljer kanterna. if -> case{i}/else,
loop -> each (per varv) / done, med loopback fran kroppen till loop-nodens in.
Aateranvander _execute_step/_evaluate_condition fran flow_engine - inget dupliceras.
graph_needs_browser/first_browser_node ersatter de gamla steg-varianterna.
- rpa/migrate.py (ny): steps_to_graph()/flow_to_graph() konverterar gamla
nastlade floden och block. jump_to_label blir en vanlig kant.
- rpa/flow_engine.py: run_flow() ar nu en dispatcher (list/{steps} migreras,
{nodes,edges} kors direkt). _run_steps och loop/if-grenarna i _execute_step
borttagna. _NO_BROWSER_STEPS behalls.
- rpa/flows.py + rpa/blocks.py: laser/skriver grafform, auto-migrerar gammalt
format vid inlasning. validate_graph i stallet for validate_flow.
- rpa/schema.py: validate_graph(), node_handles(), grafbaserad flow_json_schema()
och step_reference_markdown(). jump_to_label borttaget.
- rpa/runs.py + webui/app.py: skickar hela flodet till start_run/run_flow.
- Tester: test_graph_engine, test_migrate (nya); test_schema, test_flows,
test_blocks, test_skip_browser, test_desktop_capture omskrivna for grafform.
test_assist vantar pa fas 2.
---
rpa/blocks.py | 31 +++--
rpa/flow_engine.py | 224 ++++++--------------------------
rpa/flows.py | 67 ++++++----
rpa/graph_engine.py | 199 ++++++++++++++++++++++++++++
rpa/migrate.py | 146 +++++++++++++++++++++
rpa/runs.py | 6 +-
rpa/schema.py | 293 ++++++++++++++++++++++++------------------
tests/test_blocks.py | 23 +++-
tests/test_desktop_capture.py | 5 +-
tests/test_flows.py | 63 ++++++---
tests/test_graph_engine.py | 213 ++++++++++++++++++++++++++++++
tests/test_migrate.py | 104 +++++++++++++++
tests/test_schema.py | 151 ++++++++++++----------
tests/test_skip_browser.py | 74 +++++------
webui/app.py | 2 +-
15 files changed, 1119 insertions(+), 482 deletions(-)
diff --git a/rpa/blocks.py b/rpa/blocks.py
index 323bc30..6232f0d 100644
--- a/rpa/blocks.py
+++ b/rpa/blocks.py
@@ -1,6 +1,7 @@
-"""Block-bibliotek: återanvändbara flödesdelar (`{name, description, steps,
-settings?}`). Medföljande block ligger i repo-mappen `blocks/` (läs-bara),
-användarens egna i `data/blocks/`. Spegel av rpa/flows.py.
+"""Block-bibliotek: återanvändbara flödesdelar i grafform
+(`{name, description, nodes, edges, settings?}`). Medföljande block ligger i
+repo-mappen `blocks/` (läs-bara), användarens egna i `data/blocks/`.
+Spegel av rpa/flows.py. Gamla `{steps}`-block migreras vid inläsning.
"""
from __future__ import annotations
@@ -10,7 +11,8 @@ import re
from pathlib import Path
from . import config
-from .schema import validate_flow
+from .migrate import steps_to_graph
+from .schema import validate_graph
NAME_RE = re.compile(r"^[a-zA-Z0-9_\-åäöÅÄÖ ]{1,80}$")
@@ -37,10 +39,15 @@ def _check_name(name: str) -> str:
def _read(path: Path, builtin: bool) -> dict:
data = json.loads(path.read_text(encoding="utf-8"))
+ if data.get("nodes") is None and "steps" in data:
+ g = steps_to_graph(data.get("steps", []))
+ data = {**data, "nodes": g["nodes"], "edges": g["edges"], "start": g["start"]}
return {
"name": data.get("name") or path.stem,
"description": data.get("description", ""),
- "steps": data.get("steps", []),
+ "nodes": data.get("nodes", []),
+ "edges": data.get("edges", []),
+ "start": data.get("start"),
"settings": data.get("settings", {}),
"builtin": builtin,
}
@@ -61,7 +68,7 @@ def list_blocks() -> list[dict]:
out[b["name"]] = {
"name": b["name"],
"description": b["description"],
- "step_count": len(b["steps"]),
+ "step_count": len(b["nodes"]),
"builtin": builtin,
}
return sorted(out.values(), key=lambda b: b["name"].lower())
@@ -78,13 +85,21 @@ def load_block(name: str) -> dict:
def save_block(name: str, block: dict) -> None:
_check_name(name)
+ if block.get("nodes") is None and "steps" in block:
+ g = steps_to_graph(block.get("steps", []))
+ block = {**block, "nodes": g["nodes"], "edges": g["edges"]}
normalized = {
"name": name,
"description": str(block.get("description", "")),
- "steps": block.get("steps", []),
+ "nodes": block.get("nodes", []),
+ "edges": block.get("edges", []),
"settings": block.get("settings", {}),
}
- errors = validate_flow({"steps": normalized["steps"], "settings": normalized["settings"]}, strict=False)
+ # ett block är ett fragment - ingen krav på exakt en start
+ errors = [
+ e for e in validate_graph(normalized, strict=False)
+ if "startnod" not in e.lower()
+ ]
if errors:
raise InvalidBlock(errors)
config.BLOCKS_DIR.mkdir(parents=True, exist_ok=True)
diff --git a/rpa/flow_engine.py b/rpa/flow_engine.py
index 6c6d61c..8a8a6b5 100644
--- a/rpa/flow_engine.py
+++ b/rpa/flow_engine.py
@@ -157,67 +157,13 @@ _DESKTOP_STEPS = frozenset({
"desktop_screenshot", "desktop_click_image", "desktop_wait_for_image", "desktop_click_xy",
})
-# Stegtyper som inte rör webbläsaren alls (utöver desktop_*). loop/if är
-# strukturella - deras inre steg (och if-villkor) avgör själva om de behöver den.
+# Nodtyper som inte rör webbläsaren alls (utöver desktop_*). "loop" hanteras
+# separat (dess kropp = egna noder). "if" avgörs av sina villkor.
_NO_BROWSER_STEPS = frozenset({
- "set_var", "increment", "clear_var", "save_var", "load_var", "run_script",
- "wait", "jump_to_label", "loop", "if",
+ "set_var", "increment", "clear_var", "save_var", "load_var", "run_script", "wait",
})
-def _condition_needs_browser(cond: dict) -> bool:
- return cond.get("kind", "element_exists") in ("element_exists", "element_not_exists")
-
-
-def _step_needs_browser(step: dict) -> bool:
- """True om steget (eller något nästlat steg / if-villkor) kräver Playwright.
- Inaktiverade steg räknas inte - de hoppas över vid körning."""
- if step.get("enabled") is False:
- return False
- t = step.get("type")
- if t in _DESKTOP_STEPS or (t in _NO_BROWSER_STEPS and t not in ("loop", "if")):
- return False
- if t == "loop":
- return any(_step_needs_browser(s) for s in step.get("steps", []))
- if t == "if":
- for br in step.get("branches", []):
- if _condition_needs_browser(br.get("condition", {})):
- return True
- if any(_step_needs_browser(s) for s in br.get("steps", [])):
- return True
- return any(_step_needs_browser(s) for s in (step.get("else") or []))
- return True
-
-
-def _flow_needs_browser(steps: list) -> bool:
- return any(_step_needs_browser(s) for s in steps)
-
-
-def _first_browser_step(steps: list) -> str | None:
- """Beskrivning av det första steget som gör att webbläsaren måste startas -
- för att kunna säga det i körloggen. Ignorerar inaktiverade steg? Nej, motorn
- startar webbläsaren oavsett enabled (enkelhet)."""
- for step in steps:
- if not _step_needs_browser(step):
- continue
- t = step.get("type")
- if t == "loop":
- inner = _first_browser_step(step.get("steps", []))
- return f"loop -> {inner}" if inner else t
- if t == "if":
- for br in step.get("branches", []):
- if _condition_needs_browser(br.get("condition", {})):
- return f"if ({br['condition'].get('kind')})"
- inner = _first_browser_step(br.get("steps", []))
- if inner:
- return f"if -> {inner}"
- inner = _first_browser_step(step.get("else") or [])
- return f"if/else -> {inner}" if inner else t
- label = step.get("label")
- return f"{t}" + (f" '{label}'" if label else "")
- return None
-
-
_BAD_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
@@ -427,12 +373,6 @@ def _evaluate_condition(page: Page, item_locator, vars_: dict, condition: dict)
raise FlowError(f"Okänt villkor: {kind!r}")
-# Skydd mot oändliga hopp via jump_to_label - delas (muteras) genom hela
-# körningens rekursiva _run_steps-anrop.
-MAX_JUMPS_PER_RUN = 1000
-
-
-
def _truthy(v) -> bool:
return v is True or str(v).strip().lower() == "true"
@@ -509,9 +449,10 @@ def _execute_desktop_step(step, t, log, vars_, desktop_holder) -> None:
raise FlowError(f"Ohanterad desktop-stegtyp: {t!r}")
-def _execute_step(step, t, page, log, vars_, item_locator, download_dir, jump_budget, desktop_holder) -> None:
- """Kör ett enskilt steg. Undantag propagerar - anroparen avgör (via
- stegets "on_error") om körningen ska stanna eller fortsätta."""
+def _execute_step(step, t, page, log, vars_, item_locator, download_dir, desktop_holder) -> None:
+ """Kör ett enskilt steg (icke-behållar). loop/if hanteras av grafmotorn.
+ Undantag propagerar - anroparen avgör (via stegets "on_error") om körningen
+ ska stanna eller fortsätta."""
if t in _DESKTOP_STEPS:
_execute_desktop_step(step, t, log, vars_, desktop_holder)
return
@@ -764,61 +705,6 @@ def _execute_step(step, t, page, log, vars_, item_locator, download_dir, jump_bu
vars_[var_name] = {"selector": selector, "count": count, "preview": preview}
log(f"Hittade {count} post(er) -> variabel '{var_name}'")
- elif t == "loop":
- over_name = step.get("over")
- over = vars_.get(over_name)
- if over is None:
- raise FlowError(
- f"Loop refererar okänd variabel '{over_name}' "
- "(kör ett 'Lista poster'/'Lägg till i lista'/'Läs in variabel'-steg "
- "innan loopen och ange samma namn)."
- )
-
- index_var = step.get("index_var") or "index"
-
- if isinstance(over, dict) and "selector" in over and "count" in over:
- # DOM-baserad loop, från "Lista poster": item_locator = den aktuella
- # rad-/postens element, scope=item i inre steg söker inom den.
- selector = over["selector"]
- count = over["count"]
- for post_idx in range(count):
- log(f"-- Post {post_idx + 1}/{count} --")
- vars_[index_var] = post_idx + 1
- current_item = page.locator(selector).nth(post_idx)
- _run_steps(page, step.get("steps", []), log, vars_, current_item, download_dir, jump_budget, desktop_holder)
- elif isinstance(over, list):
- # Loop över en vanlig listvariabel (t.ex. från "Lägg till i lista" eller
- # "Läs in variabel"). Inget DOM-element per varv - varje värde läggs i
- # variabeln item_var (default "item") och kan användas som {item} i
- # by=text/role-fält för att hitta motsvarande element på sidan.
- item_var = step.get("item_var") or "item"
- count = len(over)
- for post_idx, item_value in enumerate(over):
- log(f"-- Post {post_idx + 1}/{count} --")
- vars_[index_var] = post_idx + 1
- vars_[item_var] = item_value
- _run_steps(page, step.get("steps", []), log, vars_, item_locator, download_dir, jump_budget, desktop_holder)
- else:
- raise FlowError(
- f"Variabel '{over_name}' kan inte loopas över (varken en lista eller "
- "ett 'Lista poster'-resultat)."
- )
-
- elif t == "if":
- matched_steps = None
- for branch in step.get("branches", []):
- if _evaluate_condition(page, item_locator, vars_, branch.get("condition", {})):
- matched_steps = branch.get("steps", [])
- break
- if matched_steps is not None:
- log("Villkor uppfyllt, kör gren")
- _run_steps(page, matched_steps, log, vars_, item_locator, download_dir, jump_budget, desktop_holder)
- elif step.get("else") is not None:
- log("Inget villkor uppfyllt, kör 'else'-gren")
- _run_steps(page, step["else"], log, vars_, item_locator, download_dir, jump_budget, desktop_holder)
- else:
- log("Inget villkor uppfyllt, ingen 'else'-gren - fortsätter")
-
elif t == "download":
loc = _resolve_locator(page, item_locator, step, vars_)
log(f"Laddar ner via: {_describe_selector(step, vars_)}{_describe_target(loc)}")
@@ -1038,55 +924,6 @@ def _execute_step(step, t, page, log, vars_, item_locator, download_dir, jump_bu
raise FlowError(f"Okänd stegtyp: {t!r}")
-def _run_steps(
- page: Page | None,
- steps: list,
- log: LogFn,
- vars_: dict,
- item_locator,
- download_dir: Path,
- jump_budget: dict,
- desktop_holder: dict,
-) -> None:
- i = 0
- while i < len(steps):
- step = steps[i]
- t = step.get("type")
-
- if step.get("enabled") is False:
- log(f"(Inaktiverat, hoppar över: {t})")
- i += 1
- continue
-
- if t == "jump_to_label":
- target_label = step.get("target", "")
- target_index = next((j for j, s in enumerate(steps) if s.get("label") == target_label), None)
- if target_index is None:
- raise FlowError(
- f"Hittar ingen etikett '{target_label}' bland stegen på samma nivå "
- "(kan inte hoppa in i/ut ur en loop eller if-gren)."
- )
- jump_budget["remaining"] -= 1
- if jump_budget["remaining"] <= 0:
- raise FlowError(
- "För många hopp i den här körningen - avbryter "
- "(troligen en oändlig hopp-loop)."
- )
- log(f"Hoppar till etikett '{target_label}'")
- i = target_index
- continue
-
- try:
- _execute_step(step, t, page, log, vars_, item_locator, download_dir, jump_budget, desktop_holder)
- except Exception as _step_exc:
- if step.get("on_error") == "continue":
- log(f"Steget '{t}' misslyckades - fortsätter (on_error=continue): {_step_exc}")
- else:
- raise
-
- i += 1
-
-
def _notify_run_complete(error: Exception | None) -> None:
try:
from plyer import notification
@@ -1101,7 +938,7 @@ def _notify_run_complete(error: Exception | None) -> None:
def run_flow(
- steps: list,
+ flow,
log: LogFn = print,
download_dir: Path = None,
close_mode: str = "always",
@@ -1111,7 +948,9 @@ def run_flow(
settings: dict | None = None,
initial_vars: dict | None = None,
) -> None:
- """Kör ett flöde.
+ """Kör ett flöde. `flow` kan vara nod/kant-grafen ({nodes, edges, start}),
+ ett helt flöde ({steps, ...} eller {nodes, ...}), eller en rå steglista
+ (gammalt format) - gamla format migreras internt till grafen.
initial_vars seedar variabellagret innan första steget - används för
indata-variabler (flow["inputs"] / värde-set) som matas in vid körning.
@@ -1136,24 +975,38 @@ def run_flow(
rpa/browser.py. settings["notify_on_complete"] = true visar en
skrivbordsnotis när körningen är klar (via plyer, samma som main.py).
"""
+ from . import graph_engine, migrate
+
settings = settings or {}
download_dir = download_dir or config.DOWNLOAD_DIR
+ # Normalisera indata till nod/kant-grafen.
+ if isinstance(flow, list):
+ graph = migrate.steps_to_graph(flow)
+ elif isinstance(flow, dict) and flow.get("nodes") is None and "steps" in flow:
+ graph = migrate.steps_to_graph(flow.get("steps", []), flow.get("layout"))
+ elif isinstance(flow, dict) and flow.get("nodes") is not None:
+ graph = {"nodes": flow["nodes"], "edges": flow.get("edges", []), "start": flow.get("start")}
+ settings = settings or flow.get("settings", {})
+ else:
+ graph = {"nodes": [], "edges": [], "start": None}
+ nodes, edges, start = graph["nodes"], graph["edges"], graph.get("start")
+
# Rena desktop-/variabelflöden startar ingen webbläsare. Sätts explicit via
- # settings["skip_browser"] eller upptäcks automatiskt (inga webbläsarsteg).
- skip_browser = bool(settings.get("skip_browser")) or not _flow_needs_browser(steps)
+ # settings["skip_browser"] eller upptäcks automatiskt (inga webbläsarnoder).
+ skip_browser = bool(settings.get("skip_browser")) or not graph_engine.graph_needs_browser(nodes)
if skip_browser:
- log("Kör utan webbläsare (skip_browser / inga webbläsarsteg).")
+ log("Kör utan webbläsare (skip_browser / inga webbläsarnoder).")
_run_and_finish(
- None, None, steps, log, download_dir, close_mode, close_event,
+ None, None, nodes, edges, start, log, download_dir, close_mode, close_event,
on_finished, None, settings, initial_vars,
)
return
- reason = _first_browser_step(steps)
+ reason = graph_engine.first_browser_node(nodes)
if reason:
- log(f"Startar webbläsare (krävs av steg: {reason}).")
+ log(f"Startar webbläsare (krävs av: {reason}).")
with sync_playwright() as p:
context = launch_context(p, settings)
@@ -1161,29 +1014,28 @@ def run_flow(
context.tracing.start(screenshots=True, snapshots=True, sources=True)
page = get_page(context)
_run_and_finish(
- context, page, steps, log, download_dir, close_mode, close_event,
+ context, page, nodes, edges, start, log, download_dir, close_mode, close_event,
on_finished, trace_path, settings, initial_vars,
)
def _run_and_finish(
- context, page, steps, log, download_dir, close_mode, close_event, on_finished,
+ context, page, nodes, edges, start, log, download_dir, close_mode, close_event, on_finished,
trace_path, settings, initial_vars=None,
) -> None:
+ from . import graph_engine
+
error: Exception | None = None
# Indata-variabler är strängar (samma som övriga variabler i motorn).
vars_: dict = {str(k): (v if isinstance(v, str) else str(v)) for k, v in (initial_vars or {}).items()}
if vars_:
log(f"Indata-variabler: {', '.join(sorted(vars_))}")
- jump_budget = {"remaining": MAX_JUMPS_PER_RUN}
desktop_holder: dict = {"session": None}
try:
- _run_steps(
- page, steps, log, vars_=vars_, item_locator=None,
- download_dir=download_dir, jump_budget=jump_budget,
- desktop_holder=desktop_holder,
+ graph_engine.run_graph(
+ nodes, edges, log, page=page, download_dir=download_dir,
+ vars_=vars_, desktop_holder=desktop_holder, start=start,
)
- log("Flöde klart.")
except Exception as exc: # noqa: BLE001
error = exc
log(f"FEL: {exc}")
diff --git a/rpa/flows.py b/rpa/flows.py
index 597f7e2..16662bf 100644
--- a/rpa/flows.py
+++ b/rpa/flows.py
@@ -2,6 +2,10 @@
data/flows/. Utbrutet ur webui så att både webui:t och MCP-servern
(rpa/mcp_server.py) delar exakt samma regler (namnvalidering, schema-validering
vid spara).
+
+Flöden lagras som nod/kant-grafen ({format:"graph", nodes, edges, start, ...}).
+Gamla {steps}-flöden läses fortfarande - de migreras vid inläsning och skrivs
+om till grafformat vid nästa sparning.
"""
from __future__ import annotations
@@ -11,7 +15,8 @@ import re
from pathlib import Path
from . import config
-from .schema import validate_flow
+from .migrate import flow_to_graph
+from .schema import validate_graph
NAME_RE = re.compile(r"^[a-zA-Z0-9_\-åäöÅÄÖ ]{1,80}$")
@@ -25,7 +30,7 @@ class FlowNotFound(FileNotFoundError):
class InvalidFlow(ValueError):
- """Flödet strider mot schemat (rpa/schema.py:validate_flow)."""
+ """Flödet strider mot schemat (rpa/schema.py:validate_graph)."""
def __init__(self, errors: list[str]):
self.errors = errors
@@ -42,12 +47,13 @@ def list_flows() -> list[str]:
return sorted(p.stem for p in config.FLOWS_DIR.glob("*.json"))
-def load_flow(name: str) -> dict:
- path = flow_path(name)
- if not path.exists():
- raise FlowNotFound(f"Flödet finns inte: {name}")
- flow = json.loads(path.read_text(encoding="utf-8"))
- flow.setdefault("steps", [])
+def _normalize(flow: dict) -> dict:
+ """Alltid grafform, med alla nycklar defaultade."""
+ flow = flow_to_graph(flow) # migrerar {steps} vid behov, no-op annars
+ flow.setdefault("format", "graph")
+ flow.setdefault("nodes", [])
+ flow.setdefault("edges", [])
+ flow.setdefault("start", None)
flow.setdefault("settings", {})
flow.setdefault("layout", {})
flow.setdefault("inputs", [])
@@ -55,38 +61,45 @@ def load_flow(name: str) -> dict:
return flow
+def load_flow(name: str) -> dict:
+ path = flow_path(name)
+ if not path.exists():
+ raise FlowNotFound(f"Flödet finns inte: {name}")
+ return _normalize(json.loads(path.read_text(encoding="utf-8")))
+
+
def save_flow(name: str, flow: dict, *, strict: bool = False) -> None:
- """Sparar ett flöde. Kastar InvalidFlow om det strider mot schemat.
+ """Sparar ett flöde (grafform). Kastar InvalidFlow om det strider mot schemat.
strict=False (default) tillåter halvfärdiga flöden med tomma obligatoriska
fält - webui:t autosparar under redigering. MCP-servern använder strict=True."""
path = flow_path(name)
- normalized: dict = {
- "steps": flow.get("steps", []),
- "settings": flow.get("settings", {}),
- }
- errors = validate_flow(normalized, strict=strict)
+ g = _normalize(flow)
+
+ errors = validate_graph(g, strict=strict)
if errors:
raise InvalidFlow(errors)
- # layout = frivilliga nodpositioner från canvas-UI:t. Ingår inte i
- # valideringen (motorn/schemat bryr sig inte) men sparas om den finns.
- layout = flow.get("layout")
- if isinstance(layout, dict) and layout:
- normalized["layout"] = layout
-
- # inputs = deklarerade indata-variabler, input_sets = namngivna värde-set.
- inputs = flow.get("inputs")
+ inputs = g.get("inputs")
if isinstance(inputs, list) and inputs:
bad = [i for i, x in enumerate(inputs) if not isinstance(x, dict) or not str(x.get("name", "")).strip()]
if bad:
raise InvalidFlow([f"inputs[{i}]: varje indata måste ha ett 'name'." for i in bad])
- normalized["inputs"] = inputs
- input_sets = flow.get("input_sets")
- if isinstance(input_sets, dict) and input_sets:
- normalized["input_sets"] = input_sets
- path.write_text(json.dumps(normalized, ensure_ascii=False, indent=2), encoding="utf-8")
+ out: dict = {
+ "format": "graph",
+ "nodes": g["nodes"],
+ "edges": g["edges"],
+ "settings": g["settings"],
+ }
+ if g.get("start"):
+ out["start"] = g["start"]
+ for key in ("inputs", "input_sets", "layout"):
+ val = g.get(key)
+ if val:
+ out[key] = val
+
+ path.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
def delete_flow(name: str) -> bool:
diff --git a/rpa/graph_engine.py b/rpa/graph_engine.py
new file mode 100644
index 0000000..af38b67
--- /dev/null
+++ b/rpa/graph_engine.py
@@ -0,0 +1,199 @@
+"""Grafmotorn: kör ett nod/kant-flöde genom att följa kanterna.
+
+Startar på `start`-noden (eller den enda utan inkommande kant), följer `out`-
+handtaget efter varje vanlig nod. `if`-noder följer `case{i}`/`else`, `loop`-
+noder `each` (per varv) / `done` (efter sista varvet); loop-kroppen kopplar
+tillbaka till loop-nodens `in`.
+
+Stegprimitiverna (_execute_step, _evaluate_condition, ...) återanvänds från
+rpa/flow_engine.py - inget dupliceras.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+
+from .flow_engine import (
+ FlowError,
+ _DESKTOP_STEPS,
+ _NO_BROWSER_STEPS,
+ _evaluate_condition,
+ _execute_step,
+)
+
+MAX_NODE_STEPS = 100_000 # skydd mot oändliga loop-back-cykler
+
+
+def _adjacency(edges: list) -> dict[str, dict[str, list[str]]]:
+ out: dict[str, dict[str, list[str]]] = defaultdict(lambda: defaultdict(list))
+ for e in edges:
+ h = e.get("sourceHandle") or "out"
+ out[e["source"]][h].append(e["target"])
+ return out
+
+
+def _start_node(nodes: list, edges: list, explicit: str | None) -> str | None:
+ if explicit:
+ return explicit
+ targeted = {e["target"] for e in edges}
+ starts = [n["id"] for n in nodes if n["id"] not in targeted]
+ if starts:
+ return starts[0]
+ return nodes[0]["id"] if nodes else None
+
+
+def graph_needs_browser(nodes: list) -> bool:
+ """Platt variant av flow_engine._flow_needs_browser för nod-grafen."""
+ for n in nodes:
+ if n.get("enabled") is False:
+ continue
+ t = n.get("type")
+ if t in _DESKTOP_STEPS or t in _NO_BROWSER_STEPS or t == "loop":
+ continue
+ if t == "if":
+ for br in n.get("branches", []):
+ if br.get("condition", {}).get("kind", "element_exists") in (
+ "element_exists", "element_not_exists"
+ ):
+ return True
+ continue
+ return True
+ return False
+
+
+def first_browser_node(nodes: list) -> str | None:
+ for n in nodes:
+ if n.get("enabled") is False:
+ continue
+ t = n.get("type")
+ if t in _DESKTOP_STEPS or t in _NO_BROWSER_STEPS or t == "loop":
+ continue
+ if t == "if":
+ if any(
+ br.get("condition", {}).get("kind", "element_exists")
+ in ("element_exists", "element_not_exists")
+ for br in n.get("branches", [])
+ ):
+ return f"if{(' ' + n['label']) if n.get('label') else ''}"
+ continue
+ return t + (f" '{n['label']}'" if n.get("label") else "")
+ return None
+
+
+def run_graph(nodes, edges, log, *, page, download_dir, vars_, desktop_holder, start=None) -> None:
+ """Kör grafen. Undantag från ett steg propagerar om inte nodens on_error =
+ "continue". Muterar vars_ / desktop_holder."""
+ nodes_by_id = {n["id"]: n for n in nodes}
+ out = _adjacency(edges)
+
+ def go(node_id: str, handle: str) -> str | None:
+ targets = out.get(node_id, {}).get(handle, [])
+ return targets[0] if targets else None
+
+ cursor = _start_node(nodes, edges, start)
+ loop_state: dict[str, dict] = {}
+ # [(loop_node_id, locator|None)] - översta = aktuell "item"-scope
+ item_stack: list[list] = []
+ steps_run = 0
+
+ while cursor is not None:
+ steps_run += 1
+ if steps_run > MAX_NODE_STEPS:
+ raise FlowError("För många steg i körningen - avbryter (troligen en oändlig loop).")
+
+ node = nodes_by_id.get(cursor)
+ if node is None:
+ raise FlowError(f"En kant pekar på en okänd nod: {cursor!r}")
+ t = node.get("type")
+
+ if node.get("enabled") is False:
+ log(f"(Inaktiverat, hoppar över: {t})")
+ cursor = go(cursor, "out")
+ continue
+
+ item_locator = item_stack[-1][1] if item_stack else None
+
+ # --- if ---
+ if t == "if":
+ chosen = "else"
+ for i, br in enumerate(node.get("branches", [])):
+ try:
+ if _evaluate_condition(page, item_locator, vars_, br.get("condition", {})):
+ chosen = f"case{i}"
+ log(f"if: gren {i} matchade")
+ break
+ except Exception as exc: # noqa: BLE001
+ if node.get("on_error") == "continue":
+ log(f"if-villkoret misslyckades - fortsätter (else): {exc}")
+ break
+ raise
+ else:
+ log("if: ingen gren matchade -> else")
+ cursor = go(cursor, chosen)
+ continue
+
+ # --- loop ---
+ if t == "loop":
+ st = loop_state.get(cursor)
+ if st is None:
+ st = _loop_init(node, vars_)
+ loop_state[cursor] = st
+ item_stack.append([cursor, item_locator]) # ärver ev. yttre scope
+
+ total = st["total"]
+ if st["i"] >= total:
+ loop_state.pop(cursor, None)
+ if item_stack and item_stack[-1][0] == cursor:
+ item_stack.pop()
+ log(f"loop klar ({total} varv)")
+ cursor = go(cursor, "done")
+ continue
+
+ idx = st["i"]
+ st["i"] += 1
+ vars_[st["index_var"]] = idx + 1
+ if st["kind"] == "dom":
+ loc = page.locator(st["selector"]).nth(idx)
+ if item_stack and item_stack[-1][0] == cursor:
+ item_stack[-1][1] = loc
+ else:
+ vars_[st["item_var"]] = st["items"][idx]
+ log(f"-- Varv {idx + 1}/{total} --")
+
+ nxt = go(cursor, "each")
+ cursor = nxt if nxt is not None else cursor # ingen kropp -> loopa vidare
+ continue
+
+ # --- vanligt steg (desktop hanteras internt av _execute_step) ---
+ try:
+ _execute_step(node, t, page, log, vars_, item_locator, download_dir, desktop_holder)
+ except Exception as exc: # noqa: BLE001
+ if node.get("on_error") == "continue":
+ log(f"Steget '{t}' misslyckades - fortsätter (on_error=continue): {exc}")
+ else:
+ raise
+
+ cursor = go(cursor, "out")
+
+ log("Flöde klart.")
+
+
+def _loop_init(node: dict, vars_: dict) -> dict:
+ over_name = node.get("over")
+ over = vars_.get(over_name)
+ if over is None:
+ raise FlowError(
+ f"Loop refererar okänd variabel '{over_name}' (kör 'Lista poster' / "
+ "'Lägg till i lista' / 'Läs in variabel' innan loopen)."
+ )
+ index_var = node.get("index_var") or "index"
+ if isinstance(over, dict) and "selector" in over and "count" in over:
+ return {"kind": "dom", "selector": over["selector"], "total": over["count"],
+ "i": 0, "index_var": index_var}
+ if isinstance(over, list):
+ return {"kind": "list", "items": over, "total": len(over), "i": 0,
+ "index_var": index_var, "item_var": node.get("item_var") or "item"}
+ raise FlowError(
+ f"Variabel '{over_name}' kan inte loopas över (varken en lista eller ett "
+ "'Lista poster'-resultat)."
+ )
diff --git a/rpa/migrate.py b/rpa/migrate.py
new file mode 100644
index 0000000..373db32
--- /dev/null
+++ b/rpa/migrate.py
@@ -0,0 +1,146 @@
+"""Migrerar det gamla nästlade `{steps}`-formatet till nod/kant-grafen som
+grafmotorn (rpa/graph_engine.py) kör.
+
+Gammalt: steps: [ {type, ...}, {type:"loop", steps:[...]}, {type:"if", branches:[{condition, steps:[]}], else:[...]} ]
+Nytt: nodes: [{id, type, position, ...fält}], edges: [{id, source, target, sourceHandle, targetHandle}]
+
+Handtag: vanlig nod in/out; `if` -> in + case0..caseN + else; `loop` -> in + each + done.
+jump_to_label försvinner - föregångarens kant pekas om till etikett-noden.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+_CONTAINER_KEYS = ("steps", "branches", "else")
+
+
+def _uid() -> str:
+ return uuid.uuid4().hex[:12]
+
+
+def _flat_node(step: dict, pos: dict | None) -> dict:
+ """Nod med stegets platta fält (utan behållar-nycklar). `if` behåller
+ branches men bara deras condition (inte steps)."""
+ node = {k: v for k, v in step.items() if k not in _CONTAINER_KEYS}
+ node.setdefault("id", _uid())
+ node["position"] = pos or {"x": 0, "y": 0}
+ if step.get("type") == "if":
+ node["branches"] = [{"condition": b.get("condition", {})} for b in step.get("branches", [])]
+ return node
+
+
+class _Builder:
+ def __init__(self, layout: dict | None):
+ self.nodes: list[dict] = []
+ self.edges: list[dict] = []
+ self.layout = layout or {}
+ self._y = 0
+
+ def _pos(self, step: dict) -> dict:
+ sid = step.get("id")
+ if sid and sid in self.layout:
+ return self.layout[sid]
+ self._y += 130
+ return {"x": 0, "y": self._y}
+
+ def _edge(self, src: str, src_h: str, tgt: str, tgt_h: str = "in") -> None:
+ self.edges.append({
+ "id": f"e-{_uid()}",
+ "source": src, "sourceHandle": src_h,
+ "target": tgt, "targetHandle": tgt_h,
+ })
+
+ def chain(self, steps: list) -> tuple[str | None, list[tuple[str, str]]]:
+ """Bygger noder för en steglista. Returnerar (entry_id, open_ends) där
+ open_ends är [(node_id, handle), ...] som ska kopplas vidare."""
+ label_map: dict[str, str] = {}
+ prepared: list[dict] = []
+ for s in steps:
+ s = dict(s)
+ s.setdefault("id", _uid())
+ prepared.append(s)
+ if s.get("label"):
+ label_map[s["label"]] = s["id"]
+
+ entry: str | None = None
+ open_ends: list[tuple[str, str]] = []
+
+ for s in prepared:
+ if s.get("type") == "jump_to_label":
+ tgt = label_map.get(s.get("target", ""))
+ if tgt:
+ for nid, h in open_ends:
+ self._edge(nid, h, tgt)
+ open_ends = [] # framåtflödet stannar här
+ continue
+
+ b_entry, b_open = self._block(s)
+ if entry is None:
+ entry = b_entry
+ for nid, h in open_ends:
+ self._edge(nid, h, b_entry)
+ open_ends = b_open
+
+ return entry, open_ends
+
+ def _block(self, step: dict) -> tuple[str, list[tuple[str, str]]]:
+ t = step.get("type")
+ node = _flat_node(step, self._pos(step))
+ self.nodes.append(node)
+ nid = node["id"]
+
+ if t == "loop":
+ body_entry, body_open = self.chain(step.get("steps", []))
+ if body_entry:
+ self._edge(nid, "each", body_entry)
+ for bn, bh in body_open: # kroppens slut loopar tillbaka
+ self._edge(bn, bh, nid)
+ return nid, [(nid, "done")]
+
+ if t == "if":
+ open_ends: list[tuple[str, str]] = []
+ for i, br in enumerate(step.get("branches", [])):
+ b_entry, b_open = self.chain(br.get("steps", []))
+ if b_entry:
+ self._edge(nid, f"case{i}", b_entry)
+ open_ends.extend(b_open)
+ else:
+ open_ends.append((nid, f"case{i}"))
+ els = step.get("else")
+ if els is not None:
+ e_entry, e_open = self.chain(els)
+ if e_entry:
+ self._edge(nid, "else", e_entry)
+ open_ends.extend(e_open)
+ else:
+ open_ends.append((nid, "else"))
+ else:
+ # inget else i gamla modellen = "fortsätt" -> else-handtaget öppet
+ open_ends.append((nid, "else"))
+ return nid, open_ends
+
+ return nid, [(nid, "out")]
+
+
+def steps_to_graph(steps: list, layout: dict | None = None) -> dict:
+ b = _Builder(layout)
+ entry, _open = b.chain(steps or [])
+ return {"nodes": b.nodes, "edges": b.edges, "start": entry}
+
+
+def flow_to_graph(flow: dict) -> dict:
+ """Ett helt flöde: behåll settings/inputs/input_sets, konvertera steps."""
+ if flow.get("nodes") is not None:
+ return flow # redan graf
+ g = steps_to_graph(flow.get("steps", []), flow.get("layout"))
+ return {
+ "format": "graph",
+ "nodes": g["nodes"],
+ "edges": g["edges"],
+ "start": g["start"],
+ "settings": flow.get("settings", {}),
+ "inputs": flow.get("inputs", []),
+ "input_sets": flow.get("input_sets", {}),
+ "layout": {},
+ }
diff --git a/rpa/runs.py b/rpa/runs.py
index 9e05a62..eb70f0e 100644
--- a/rpa/runs.py
+++ b/rpa/runs.py
@@ -21,11 +21,13 @@ def _append_log(run_id: str, message: str) -> None:
def start_run(
- steps: list,
+ flow,
close_mode: str = "always",
settings: dict | None = None,
initial_vars: dict | None = None,
) -> str:
+ """`flow` = nod/kant-grafen ({nodes, edges, start}) eller gammalt {steps} /
+ en steglista (migreras internt)."""
run_id = uuid.uuid4().hex[:12]
close_event = threading.Event()
trace_path = config.TRACE_DIR / f"{run_id}.zip"
@@ -48,7 +50,7 @@ def start_run(
def target() -> None:
try:
run_flow(
- steps,
+ flow,
log=lambda m: _append_log(run_id, m),
download_dir=config.DOWNLOAD_DIR,
close_mode=close_mode,
diff --git a/rpa/schema.py b/rpa/schema.py
index b25ee67..b339a13 100644
--- a/rpa/schema.py
+++ b/rpa/schema.py
@@ -439,21 +439,14 @@ STEP_TYPES: list[dict] = [
{
"type": "if", "category": "control",
"label_sv": "Om / annars om / annars", "label_en": "If / else if / else",
- "help": "Första grenen vars villkor är sant körs. Valfri 'else'-gren om ingen matchar.",
- "container": "branches", "fields": [],
- },
- {
- "type": "jump_to_label", "category": "control",
- "label_sv": "Hoppa till etikett", "label_en": "Jump to label",
- "help": "Hoppar till steget vars 'label' matchar 'target', bland syskonsteg på samma nivå (inte in i/ut ur loop/if).",
- "fields": [{"key": "target", "kind": "text", "required": True,
- "label_sv": "Hoppa till etikett", "label_en": "Jump to label"}],
+ "help": "Utvärderar grenarnas villkor uppifrån. Första sanna grenens utgång följs (case0, case1, …); matchar ingen följs 'else'. Villkoren ligger i noden; grenarnas steg är noder du kopplar till respektive utgång.",
+ "control": "if", "fields": [],
},
{
"type": "loop", "category": "control",
"label_sv": "Loop (för varje post)", "label_en": "Loop (for each item)",
- "help": "Itererar över ett 'list'-resultat (DOM-poster, scope=item i inre steg) ELLER en vanlig listvariabel (värdet i item_var, {item} i by=text/role).",
- "container": "steps",
+ "help": "Itererar över ett 'list'-resultat (DOM-poster) ELLER en vanlig listvariabel. Utgången 'each' körs en gång per varv (item/index sätts); kroppens sista nod ska kopplas TILLBAKA till loopens 'in'. Efter sista varvet följs 'done'.",
+ "control": "loop",
"fields": [
{"key": "over", "kind": "text", "required": True,
"label_sv": "Loopa över variabel", "label_en": "Loop over variable"},
@@ -647,110 +640,143 @@ STEP_TYPE_NAMES = [s["type"] for s in STEP_TYPES]
STEP_BY_TYPE = {s["type"]: s for s in STEP_TYPES}
+# --------------------------------------------------------------------------
+# Handtag per nodtyp (för canvasen + kant-validering)
+# --------------------------------------------------------------------------
+
+def node_handles(node_or_type) -> dict:
+ """{"inputs": ["in"], "outputs": [...]}. `if`-noder får ett utgångshandtag
+ per gren (case0..caseN) + else; `loop` får each + done."""
+ if isinstance(node_or_type, dict):
+ t = node_or_type.get("type")
+ n_branches = len(node_or_type.get("branches", []) or [1])
+ else:
+ t = node_or_type
+ n_branches = 1
+ spec = STEP_BY_TYPE.get(t, {})
+ ctrl = spec.get("control")
+ if ctrl == "if":
+ return {"inputs": ["in"], "outputs": [f"case{i}" for i in range(n_branches)] + ["else"]}
+ if ctrl == "loop":
+ return {"inputs": ["in"], "outputs": ["each", "done"]}
+ return {"inputs": ["in"], "outputs": ["out"]}
+
+
# --------------------------------------------------------------------------
# Validering
# --------------------------------------------------------------------------
-def validate_flow(flow: Any, *, strict: bool = True) -> list[str]:
- """Strukturell validering av ett flöde. Returnerar en lista felmeddelanden
- (tom lista = giltigt). Fångar det som skulle krascha eller tyst missförstås
- av flow_engine - inte allt semantiskt (t.ex. att en variabel faktiskt finns
- vid körning).
+def validate_graph(flow: Any, *, strict: bool = True) -> list[str]:
+ """Strukturell validering av nod/kant-grafen. Tom lista = giltigt.
- strict=False hoppar över "obligatoriskt fält saknas" - används av webui:t som
- autosparar halvfärdiga flöden under redigering. MCP-servern och NL-till-flöde
- använder strict=True."""
+ strict=False hoppar över "obligatoriskt fält saknas" - webui:t autosparar
+ halvfärdiga flöden. MCP och NL-till-flöde använder strict=True."""
errors: list[str] = []
if not isinstance(flow, dict):
- return ["Flödet måste vara ett objekt med nycklarna 'steps' och (valfritt) 'settings'."]
-
- steps = flow.get("steps", [])
- if not isinstance(steps, list):
- errors.append("'steps' måste vara en lista.")
- else:
- _validate_steps(steps, "steps", errors, strict)
+ return ["Flödet måste vara ett objekt med 'nodes' och 'edges'."]
+
+ nodes = flow.get("nodes", [])
+ edges = flow.get("edges", [])
+ if not isinstance(nodes, list):
+ return ["'nodes' måste vara en lista."]
+ if not isinstance(edges, list):
+ errors.append("'edges' måste vara en lista.")
+ edges = []
+
+ ids: set[str] = set()
+ handles_by_id: dict[str, dict] = {}
+ for i, node in enumerate(nodes):
+ p = f"nodes[{i}]"
+ if not isinstance(node, dict):
+ errors.append(f"{p}: noden måste vara ett objekt.")
+ continue
+ nid = node.get("id")
+ if not nid or not isinstance(nid, str):
+ errors.append(f"{p}.id: saknas.")
+ continue
+ if nid in ids:
+ errors.append(f"{p}.id: dubblett {nid!r}.")
+ ids.add(nid)
+ handles_by_id[nid] = node_handles(node)
+ _validate_node_fields(node, f"nod {nid!r}", errors, strict)
+
+ for i, e in enumerate(edges):
+ p = f"edges[{i}]"
+ if not isinstance(e, dict):
+ errors.append(f"{p}: kanten måste vara ett objekt.")
+ continue
+ s, t = e.get("source"), e.get("target")
+ if s not in ids:
+ errors.append(f"{p}: source pekar på okänd nod {s!r}.")
+ if t not in ids:
+ errors.append(f"{p}: target pekar på okänd nod {t!r}.")
+ sh = e.get("sourceHandle") or "out"
+ if s in handles_by_id and sh not in handles_by_id[s]["outputs"]:
+ errors.append(f"{p}: nod {s!r} har inget utgångshandtag {sh!r}.")
+
+ # exakt en start
+ if nodes:
+ targeted = {e.get("target") for e in edges if isinstance(e, dict)}
+ starts = [n["id"] for n in nodes if isinstance(n, dict) and n.get("id") not in targeted]
+ if flow.get("start") and flow["start"] not in ids:
+ errors.append(f"start: okänd nod {flow['start']!r}.")
+ if not flow.get("start") and len(starts) > 1:
+ errors.append(f"Flera startnoder ({len(starts)}) - koppla ihop dem eller sätt 'start'.")
+ if not starts and not flow.get("start"):
+ errors.append("Ingen startnod (alla noder har inkommande kanter).")
settings = flow.get("settings", {})
- if not isinstance(settings, dict):
- errors.append("'settings' måste vara ett objekt.")
- else:
+ if isinstance(settings, dict):
allowed = {s["key"] for s in SETTINGS_SCHEMA}
for key in settings:
if key not in allowed:
- errors.append(f"settings.{key}: okänd inställning (tillåtna: {', '.join(sorted(allowed))}).")
+ errors.append(f"settings.{key}: okänd inställning.")
return errors
-def _validate_steps(steps: list, path: str, errors: list[str], strict: bool = True) -> None:
- for i, step in enumerate(steps):
- p = f"{path}[{i}]"
- if not isinstance(step, dict):
- errors.append(f"{p}: steget måste vara ett objekt.")
- continue
- t = step.get("type")
- if t not in STEP_BY_TYPE:
- errors.append(f"{p}.type: okänd stegtyp {t!r} (tillåtna: {', '.join(STEP_TYPE_NAMES)}).")
- continue
-
- spec = STEP_BY_TYPE[t]
+def _validate_node_fields(node: dict, label: str, errors: list[str], strict: bool) -> None:
+ t = node.get("type")
+ if t not in STEP_BY_TYPE:
+ errors.append(f"{label}.type: okänd nodtyp {t!r} (tillåtna: {', '.join(STEP_TYPE_NAMES)}).")
+ return
+ spec = STEP_BY_TYPE[t]
- # Gemensamma fält.
- if "on_error" in step and step["on_error"] not in ("stop", "continue"):
- errors.append(f"{p}.on_error: måste vara 'stop' eller 'continue'.")
- if "enabled" in step and not isinstance(step["enabled"], bool):
- errors.append(f"{p}.enabled: måste vara true/false.")
+ if "on_error" in node and node["on_error"] not in ("stop", "continue"):
+ errors.append(f"{label}.on_error: måste vara 'stop' eller 'continue'.")
+ if "enabled" in node and not isinstance(node["enabled"], bool):
+ errors.append(f"{label}.enabled: måste vara true/false.")
- # Obligatoriska fält.
- if strict:
- for f in spec["fields"]:
- if f.get("required") and _is_blank(step.get(f["key"])):
- errors.append(f"{p}.{f['key']}: obligatoriskt för stegtypen '{t}'.")
-
- # Enum-fält (drivs av fältets egna 'options').
+ if strict:
for f in spec["fields"]:
- if f.get("kind") != "select" or not f.get("options"):
- continue
- val = step.get(f["key"])
- if val not in (None, "") and val not in f["options"]:
- errors.append(f"{p}.{f['key']}: måste vara ett av {', '.join(o for o in f['options'] if o)}.")
-
- # Behållarsteg.
- if t == "loop":
- inner = step.get("steps", [])
- if not isinstance(inner, list):
- errors.append(f"{p}.steps: måste vara en lista.")
- else:
- _validate_steps(inner, f"{p}.steps", errors, strict)
- elif t == "if":
- branches = step.get("branches")
- if not isinstance(branches, list) or not branches:
- errors.append(f"{p}.branches: 'if' kräver minst en gren.")
- else:
- for bi, branch in enumerate(branches):
- bp = f"{p}.branches[{bi}]"
- if not isinstance(branch, dict):
- errors.append(f"{bp}: grenen måste vara ett objekt.")
- continue
- _validate_condition(branch.get("condition", {}), f"{bp}.condition", errors)
- binner = branch.get("steps", [])
- if not isinstance(binner, list):
- errors.append(f"{bp}.steps: måste vara en lista.")
- else:
- _validate_steps(binner, f"{bp}.steps", errors, strict)
- if step.get("else") is not None:
- if not isinstance(step["else"], list):
- errors.append(f"{p}.else: måste vara en lista.")
- else:
- _validate_steps(step["else"], f"{p}.else", errors, strict)
- elif t == "extract_row":
- fields = step.get("fields", [])
- if not isinstance(fields, list) or not fields:
- errors.append(f"{p}.fields: extract_row kräver minst en kolumn ({{name, by, value}}).")
- else:
- for fi, col in enumerate(fields):
- if not isinstance(col, dict) or _is_blank(col.get("name")):
- errors.append(f"{p}.fields[{fi}].name: kolumnen måste ha ett namn.")
+ if f.get("required") and _is_blank(node.get(f["key"])):
+ errors.append(f"{label}.{f['key']}: obligatoriskt för '{t}'.")
+
+ for f in spec["fields"]:
+ if f.get("kind") != "select" or not f.get("options"):
+ continue
+ val = node.get(f["key"])
+ if val not in (None, "") and val not in f["options"]:
+ errors.append(f"{label}.{f['key']}: måste vara ett av {', '.join(o for o in f['options'] if o)}.")
+
+ if t == "if":
+ branches = node.get("branches")
+ if not isinstance(branches, list) or not branches:
+ errors.append(f"{label}.branches: 'if' kräver minst en gren.")
+ else:
+ for bi, br in enumerate(branches):
+ if not isinstance(br, dict):
+ errors.append(f"{label}.branches[{bi}]: måste vara ett objekt.")
+ continue
+ _validate_condition(br.get("condition", {}), f"{label}.branches[{bi}].condition", errors)
+ elif t == "extract_row":
+ fields = node.get("fields", [])
+ if not isinstance(fields, list) or not fields:
+ errors.append(f"{label}.fields: extract_row kräver minst en kolumn.")
+ else:
+ for fi, col in enumerate(fields):
+ if not isinstance(col, dict) or _is_blank(col.get("name")):
+ errors.append(f"{label}.fields[{fi}].name: kolumnen måste ha ett namn.")
def _validate_condition(cond: Any, path: str, errors: list[str]) -> None:
@@ -776,9 +802,8 @@ def _is_blank(value: Any) -> bool:
# --------------------------------------------------------------------------
def flow_json_schema() -> dict:
- """JSON Schema för ett helt flöde ({steps, settings}), rekursivt för loop/if.
- Används som vägledning till LLM:en - den precisa kontrollen görs av
- validate_flow()."""
+ """JSON Schema för ett flöde i grafform ({nodes, edges}). Vägledning till
+ LLM:en - den precisa kontrollen görs av validate_graph()."""
all_field_keys: set[str] = set()
for spec in STEP_TYPES:
for f in spec["fields"]:
@@ -786,43 +811,54 @@ def flow_json_schema() -> dict:
for f in COMMON_STEP_FIELDS:
all_field_keys.add(f["key"])
- step_props: dict[str, Any] = {
+ node_props: dict[str, Any] = {
+ "id": {"type": "string"},
"type": {"type": "string", "enum": STEP_TYPE_NAMES},
"label": {"type": "string"},
"enabled": {"type": "boolean"},
"on_error": {"type": "string", "enum": ["stop", "continue"]},
- "steps": {"type": "array", "items": {"$ref": "#/$defs/step"}},
+ "position": {"type": "object", "properties": {"x": {"type": "number"}, "y": {"type": "number"}}},
"branches": {
"type": "array",
"items": {
"type": "object",
- "properties": {
- "condition": {"$ref": "#/$defs/condition"},
- "steps": {"type": "array", "items": {"$ref": "#/$defs/step"}},
- },
- "required": ["condition", "steps"],
+ "properties": {"condition": {"$ref": "#/$defs/condition"}},
+ "required": ["condition"],
},
},
- "else": {"type": "array", "items": {"$ref": "#/$defs/step"}},
"fields": {"type": "array", "items": {"type": "object"}},
}
for key in sorted(all_field_keys):
- step_props.setdefault(key, {"type": ["string", "number", "boolean"]})
+ node_props.setdefault(key, {"type": ["string", "number", "boolean"]})
return {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
- "steps": {"type": "array", "items": {"$ref": "#/$defs/step"}},
+ "nodes": {"type": "array", "items": {"$ref": "#/$defs/node"}},
+ "edges": {"type": "array", "items": {"$ref": "#/$defs/edge"}},
+ "start": {"type": "string", "description": "Id på startnoden."},
"settings": {
"type": "object",
"properties": {s["key"]: _setting_schema(s) for s in SETTINGS_SCHEMA},
},
"explanation": {"type": "string", "description": "Kort förklaring på svenska av vad flödet gör."},
},
- "required": ["steps"],
+ "required": ["nodes", "edges"],
"$defs": {
- "step": {"type": "object", "properties": step_props, "required": ["type"]},
+ "node": {"type": "object", "properties": node_props, "required": ["id", "type"]},
+ "edge": {
+ "type": "object",
+ "properties": {
+ "id": {"type": "string"},
+ "source": {"type": "string"},
+ "target": {"type": "string"},
+ "sourceHandle": {"type": "string",
+ "description": "out (default), eller case0/case1/…/else för if, each/done för loop"},
+ "targetHandle": {"type": "string", "description": "in (default)"},
+ },
+ "required": ["source", "target"],
+ },
"condition": {
"type": "object",
"properties": {
@@ -853,11 +889,17 @@ def step_reference_markdown() -> str:
"""Människo/LLM-läsbar referens över alla stegtyper, genererad ur STEP_TYPES.
Bäddas in i systemprompten i rpa/assist.py."""
out: list[str] = []
- out.append("## Stegtyper\n")
+ out.append("## Flödesmodell (nod/kant-graf)\n")
out.append(
- "Ett flöde är JSON: {\"steps\": [ ...steg... ], \"settings\": { ... }}. "
- "Varje steg är {\"type\": \"<typ>\", ...fält...}. Alla fältvärden är strängar "
- "om inget annat sägs. Fält markerade (obl.) måste vara ifyllda.\n"
+ "Ett flöde är JSON: `{\"nodes\": [...], \"edges\": [...], \"start\": \"<nod-id>\", \"settings\": {...}}`.\n"
+ "En nod: `{\"id\": \"n1\", \"type\": \"<typ>\", ...fält...}`. Alla fältvärden är strängar om inget annat sägs.\n"
+ "En kant: `{\"source\": \"n1\", \"target\": \"n2\", \"sourceHandle\": \"out\", \"targetHandle\": \"in\"}`.\n"
+ "Kör-ordningen bestäms av kanterna. Efter en vanlig nod följs `out`.\n"
+ "- **`if`**: fält `branches: [{condition}]`. Utgångar `case0`, `case1`, … (första sanna) + `else`. "
+ "Koppla varje grens noder till respektive utgång.\n"
+ "- **`loop`**: utgång `each` (körs per varv, sätter `item`/`index`) och `done` (efter sista varvet). "
+ "Kroppens SISTA nod måste ha en kant TILLBAKA till loop-nodens `in`.\n"
+ "- Ingen `jump_to_label` – rita en kant i stället.\n"
)
by_cat: dict[str, list[dict]] = {}
for spec in STEP_TYPES:
@@ -882,10 +924,10 @@ def step_reference_markdown() -> str:
if f.get("help"):
bits.append("– " + f["help"])
out.append(f" - {' '.join(bits)}")
- if spec.get("container") == "steps":
- out.append(" - `steps`: lista av inre steg")
- if spec.get("container") == "branches":
- out.append(" - `branches`: lista av {condition, steps}; valfri `else`: lista av steg")
+ if spec.get("control") == "if":
+ out.append(" - `branches`: lista av {condition}; utgångar case0…caseN + else")
+ if spec.get("control") == "loop":
+ out.append(" - utgångar: `each` (per varv), `done` (klar). Kroppen kopplar tillbaka till `in`.")
if spec["type"] == "extract_row":
out.append(" - `fields`: lista av {name, by, value, role?} - name blir kolumnrubrik")
@@ -909,12 +951,13 @@ def step_reference_markdown() -> str:
return "\n".join(out)
-def engine_step_types_from_source(flow_engine_source: str) -> set[str]:
- """Plockar ut stegtyperna som motorn faktiskt hanterar, för drift-vakten i
- tests/test_schema.py. Matchar `if t == "..."` / `elif t == "..."` samt
- desktop_*-literalerna i `_DESKTOP_STEPS`-mängden."""
- found = set(re.findall(r'\b(?:el)?if t == "([a-z_]+)":', flow_engine_source))
- found |= set(re.findall(r'"(desktop_[a-z_]+)"', flow_engine_source))
+def engine_step_types_from_source(*engine_sources: str) -> set[str]:
+ """Plockar ut nodtyperna som motorn faktiskt hanterar, för drift-vakten i
+ tests/test_schema.py. Skanna flow_engine.py + graph_engine.py: matchar
+ `if t == "..."` / `elif t == "..."` samt desktop_*-literalerna."""
+ src = "\n".join(engine_sources)
+ found = set(re.findall(r'\b(?:el)?if t == "([a-z_]+)":', src))
+ found |= set(re.findall(r'"(desktop_[a-z_]+)"', src))
return found
diff --git a/tests/test_blocks.py b/tests/test_blocks.py
index 43ba088..6914dc0 100644
--- a/tests/test_blocks.py
+++ b/tests/test_blocks.py
@@ -24,13 +24,23 @@ def test_crud_round_trip():
assert meta == [{"name": "mitt block", "description": "test", "step_count": 1, "builtin": False}]
b = blocks.load_block("mitt block")
- assert b["steps"][0]["url"] == "https://x.se"
+ assert b["nodes"][0]["url"] == "https://x.se"
assert b["builtin"] is False
assert blocks.delete_block("mitt block") is True
assert blocks.list_blocks() == []
+def test_save_block_accepts_graph_form():
+ blocks.save_block("grafblock", {
+ "nodes": [{"id": "a", "type": "wait", "seconds": 1}, {"id": "b", "type": "wait", "seconds": 2}],
+ "edges": [{"id": "e", "source": "a", "target": "b"}],
+ })
+ b = blocks.load_block("grafblock")
+ assert [n["id"] for n in b["nodes"]] == ["a", "b"]
+ assert len(b["edges"]) == 1
+
+
def test_builtin_and_user_union(tmp_dirs):
_user, builtin = tmp_dirs
import json
@@ -50,7 +60,7 @@ def test_user_block_shadows_builtin(tmp_dirs):
(builtin / "x.json").write_text(json.dumps({"name": "x", "steps": [{"type": "wait", "seconds": 1}]}), encoding="utf-8")
blocks.save_block("x", {"steps": [{"type": "wait", "seconds": 9}]})
- assert blocks.load_block("x")["steps"][0]["seconds"] == 9
+ assert blocks.load_block("x")["nodes"][0]["seconds"] == 9
assert blocks.load_block("x")["builtin"] is False
@@ -62,12 +72,15 @@ def test_invalid_block_rejected():
def test_shipped_example_blocks_are_valid():
- """De medföljande blocken i repo-mappen blocks/ ska validera."""
+ """De medföljande blocken i repo-mappen blocks/ ska validera (som fragment)."""
from rpa import config as real_config
- from rpa.schema import validate_flow
+ from rpa.migrate import flow_to_graph
+ from rpa.schema import validate_graph
for p in (real_config.ROOT_DIR / "blocks").glob("*.json"):
import json
data = json.loads(p.read_text(encoding="utf-8"))
- assert validate_flow({"steps": data["steps"], "settings": data.get("settings", {})}, strict=False) == [], p.name
+ g = flow_to_graph(data)
+ errors = [e for e in validate_graph(g, strict=False) if "startnod" not in e.lower()]
+ assert errors == [], p.name
diff --git a/tests/test_desktop_capture.py b/tests/test_desktop_capture.py
index aed4d4c..a004eba 100644
--- a/tests/test_desktop_capture.py
+++ b/tests/test_desktop_capture.py
@@ -19,7 +19,8 @@ def test_image_steps_use_image_field_kind():
def test_image_field_kind_does_not_break_validation():
flow = {
- "steps": [{"type": "desktop_click_image", "image": "knapp.png", "confidence": 0.9}],
+ "nodes": [{"id": "n1", "type": "desktop_click_image", "image": "knapp.png", "confidence": 0.9}],
+ "edges": [],
"settings": {},
}
- assert schema.validate_flow(flow) == []
+ assert schema.validate_graph(flow) == []
diff --git a/tests/test_flows.py b/tests/test_flows.py
index 122c36a..e774cec 100644
--- a/tests/test_flows.py
+++ b/tests/test_flows.py
@@ -11,14 +11,18 @@ def tmp_flows_dir(tmp_path, monkeypatch):
return d
+def graph(nodes, edges=None, **extra):
+ return {"format": "graph", "nodes": nodes, "edges": edges or [], "settings": {}, **extra}
+
+
def test_crud_round_trip():
assert flows.list_flows() == []
- flows.save_flow("mitt flöde", {"steps": [{"type": "goto", "url": "https://x.se"}], "settings": {}})
+ flows.save_flow("mitt flöde", graph([{"id": "n1", "type": "goto", "url": "https://x.se"}]))
assert flows.list_flows() == ["mitt flöde"]
loaded = flows.load_flow("mitt flöde")
- assert loaded["steps"][0]["url"] == "https://x.se"
+ assert loaded["nodes"][0]["url"] == "https://x.se"
assert loaded["settings"] == {}
assert flows.delete_flow("mitt flöde") is True
@@ -26,6 +30,19 @@ def test_crud_round_trip():
assert flows.delete_flow("mitt flöde") is False
+def test_old_steps_flow_is_migrated_on_load(tmp_flows_dir):
+ import json
+
+ (tmp_flows_dir / "gammalt.json").write_text(
+ json.dumps({"steps": [{"type": "goto", "url": "https://x.se"}], "settings": {}}),
+ encoding="utf-8",
+ )
+ loaded = flows.load_flow("gammalt")
+ assert loaded["format"] == "graph"
+ assert [n["type"] for n in loaded["nodes"]] == ["goto"]
+ assert loaded["nodes"][0]["url"] == "https://x.se"
+
+
def test_invalid_name_rejected():
with pytest.raises(flows.InvalidFlowName):
flows.flow_path("../secret")
@@ -39,7 +56,7 @@ def test_load_missing_raises():
def test_save_lenient_by_default_strict_on_request():
- half = {"steps": [{"type": "goto"}], "settings": {}} # url saknas
+ half = graph([{"id": "n1", "type": "goto"}]) # url saknas
flows.save_flow("wip", half) # lenient - ok
with pytest.raises(flows.InvalidFlow):
flows.save_flow("wip", half, strict=True)
@@ -47,28 +64,32 @@ def test_save_lenient_by_default_strict_on_request():
def test_save_rejects_structural_errors_even_lenient():
with pytest.raises(flows.InvalidFlow):
- flows.save_flow("bad", {"steps": [{"type": "nonesuch"}]})
+ flows.save_flow("bad", graph([{"id": "n1", "type": "nonesuch"}]))
+
+
+def test_save_rejects_dangling_edge():
+ with pytest.raises(flows.InvalidFlow):
+ flows.save_flow("bad", graph(
+ [{"id": "n1", "type": "wait", "seconds": 1}],
+ [{"id": "e1", "source": "n1", "target": "ghost"}],
+ ))
-def test_layout_and_step_ids_round_trip():
- flow = {
- "steps": [{"type": "goto", "id": "abc123", "url": "https://x.se"}],
- "settings": {},
- "layout": {"abc123": {"x": 40, "y": 120}},
- }
+def test_node_positions_round_trip():
+ flow = graph([{"id": "abc123", "type": "goto", "url": "https://x.se",
+ "position": {"x": 40, "y": 120}}])
flows.save_flow("laid out", flow)
loaded = flows.load_flow("laid out")
- assert loaded["steps"][0]["id"] == "abc123"
- assert loaded["layout"] == {"abc123": {"x": 40, "y": 120}}
+ assert loaded["nodes"][0]["id"] == "abc123"
+ assert loaded["nodes"][0]["position"] == {"x": 40, "y": 120}
def test_inputs_and_sets_round_trip():
- flow = {
- "steps": [{"type": "goto", "url": "https://x.se/{org_id}"}],
- "settings": {},
- "inputs": [{"name": "org_id", "label": "Org", "default": "4"}],
- "input_sets": {"Test": {"org_id": "9"}},
- }
+ flow = graph(
+ [{"id": "n1", "type": "goto", "url": "https://x.se/{org_id}"}],
+ inputs=[{"name": "org_id", "label": "Org", "default": "4"}],
+ input_sets={"Test": {"org_id": "9"}},
+ )
flows.save_flow("params", flow)
loaded = flows.load_flow("params")
assert loaded["inputs"][0]["name"] == "org_id"
@@ -77,17 +98,17 @@ def test_inputs_and_sets_round_trip():
def test_input_without_name_rejected():
with pytest.raises(flows.InvalidFlow):
- flows.save_flow("bad", {"steps": [], "settings": {}, "inputs": [{"label": "x"}]})
+ flows.save_flow("bad", graph([], inputs=[{"label": "x"}]))
def test_missing_flow_defaults_inputs():
- flows.save_flow("plain", {"steps": [], "settings": {}})
+ flows.save_flow("plain", graph([]))
loaded = flows.load_flow("plain")
assert loaded["inputs"] == [] and loaded["input_sets"] == {}
def test_layout_omitted_when_empty():
- flows.save_flow("nolayout", {"steps": [], "settings": {}, "layout": {}})
+ flows.save_flow("nolayout", graph([], layout={}))
import json
raw = json.loads((flows.config.FLOWS_DIR / "nolayout.json").read_text(encoding="utf-8"))
diff --git a/tests/test_graph_engine.py b/tests/test_graph_engine.py
new file mode 100644
index 0000000..20c6f95
--- /dev/null
+++ b/tests/test_graph_engine.py
@@ -0,0 +1,213 @@
+"""Grafmotorn (rpa/graph_engine.py) - körning genom att följa kanter."""
+
+import pytest
+
+from rpa import graph_engine
+from rpa.flow_engine import FlowError
+
+
+def run(nodes, edges, vars_=None, start=None):
+ v = dict(vars_ or {})
+ logs: list[str] = []
+ graph_engine.run_graph(
+ nodes, edges, logs.append,
+ page=None, download_dir=None, vars_=v,
+ desktop_holder={"session": None}, start=start,
+ )
+ return v, logs
+
+
+def E(src, tgt, h="out"):
+ return {"id": f"{src}-{h}-{tgt}", "source": src, "sourceHandle": h, "target": tgt}
+
+
+# --- linjärt ---
+def test_linear_chain_runs_in_order():
+ v, _ = run(
+ [
+ {"id": "a", "type": "set_var", "var": "x", "text": "1"},
+ {"id": "b", "type": "increment", "var": "x", "by": 4},
+ {"id": "c", "type": "set_var", "var": "y", "text": "klar"},
+ ],
+ [E("a", "b"), E("b", "c")],
+ )
+ assert v["x"] == 5 and v["y"] == "klar"
+
+
+def test_start_is_node_without_incoming_edge():
+ v, _ = run(
+ [
+ {"id": "b", "type": "set_var", "var": "r", "text": "B"},
+ {"id": "a", "type": "set_var", "var": "r", "text": "A"},
+ ],
+ [E("a", "b")],
+ )
+ assert v["r"] == "B" # a -> b, b vinner
+
+
+def test_explicit_start_overrides():
+ v, _ = run(
+ [
+ {"id": "a", "type": "set_var", "var": "r", "text": "A"},
+ {"id": "b", "type": "set_var", "var": "r", "text": "B"},
+ ],
+ [],
+ start="b",
+ )
+ assert v["r"] == "B"
+
+
+# --- if ---
+IF_NODES = [
+ {"id": "f", "type": "if", "branches": [
+ {"condition": {"kind": "var_equals", "var": "läge", "text": "prod"}},
+ {"condition": {"kind": "var_equals", "var": "läge", "text": "test"}},
+ ]},
+ {"id": "p", "type": "set_var", "var": "r", "text": "PROD"},
+ {"id": "t", "type": "set_var", "var": "r", "text": "TEST"},
+ {"id": "e", "type": "set_var", "var": "r", "text": "ANNAT"},
+]
+IF_EDGES = [E("f", "p", "case0"), E("f", "t", "case1"), E("f", "e", "else")]
+
+
+@pytest.mark.parametrize("läge,expected", [("prod", "PROD"), ("test", "TEST"), ("x", "ANNAT")])
+def test_if_picks_matching_branch(läge, expected):
+ v, _ = run(IF_NODES, IF_EDGES, {"läge": läge})
+ assert v["r"] == expected
+
+
+def test_if_no_match_no_else_ends_that_path():
+ v, _ = run(
+ [{"id": "f", "type": "if", "branches": [
+ {"condition": {"kind": "var_equals", "var": "x", "text": "1"}}]},
+ {"id": "c", "type": "set_var", "var": "r", "text": "case"}],
+ [E("f", "c", "case0")],
+ {"x": "0"},
+ )
+ assert "r" not in v
+
+
+def test_if_branches_merge_to_shared_node():
+ v, _ = run(
+ IF_NODES + [{"id": "m", "type": "increment", "var": "n", "by": 1}],
+ IF_EDGES + [E("p", "m"), E("t", "m"), E("e", "m")],
+ {"läge": "prod", "n": "0"},
+ )
+ assert v["r"] == "PROD" and v["n"] == 1
+
+
+# --- loop ---
+def test_loop_over_list_var_sets_item_and_index():
+ v, logs = run(
+ [
+ {"id": "L", "type": "loop", "over": "poster", "item_var": "it", "index_var": "i"},
+ {"id": "n", "type": "increment", "var": "antal", "by": 1},
+ {"id": "last", "type": "set_var", "var": "sista", "text": "{it}"},
+ {"id": "done", "type": "set_var", "var": "klar", "text": "ja"},
+ ],
+ [E("L", "n", "each"), E("n", "last"), E("last", "L"), E("L", "done", "done")],
+ {"poster": ["a", "b", "c"], "antal": "0"},
+ )
+ assert v["antal"] == 3
+ assert v["sista"] == "c"
+ assert v["klar"] == "ja"
+ assert v["i"] == 3
+ assert any("3 varv" in m for m in logs)
+
+
+def test_loop_over_empty_list_goes_straight_to_done():
+ v, _ = run(
+ [
+ {"id": "L", "type": "loop", "over": "poster", "item_var": "it"},
+ {"id": "n", "type": "increment", "var": "antal", "by": 1},
+ {"id": "d", "type": "set_var", "var": "klar", "text": "ja"},
+ ],
+ [E("L", "n", "each"), E("n", "L"), E("L", "d", "done")],
+ {"poster": [], "antal": "0"},
+ )
+ assert v["antal"] == "0" and v["klar"] == "ja" # increment kördes aldrig
+
+
+def test_loop_missing_variable_raises():
+ with pytest.raises(FlowError, match="okänd variabel"):
+ run(
+ [{"id": "L", "type": "loop", "over": "saknas"},
+ {"id": "d", "type": "set_var", "var": "x", "text": "1"}],
+ [E("L", "d", "done")],
+ )
+
+
+def test_loop_over_non_iterable_raises():
+ with pytest.raises(FlowError, match="kan inte loopas"):
+ run(
+ [{"id": "L", "type": "loop", "over": "v"},
+ {"id": "d", "type": "set_var", "var": "x", "text": "1"}],
+ [E("L", "d", "done")],
+ {"v": "en sträng"},
+ )
+
+
+def test_nested_loops_accumulate_pairs():
+ v, _ = run(
+ [
+ {"id": "OUT", "type": "loop", "over": "rader", "item_var": "r", "index_var": "ri"},
+ {"id": "IN", "type": "loop", "over": "kol", "item_var": "c", "index_var": "ci"},
+ {"id": "app", "type": "set_var", "var": "sista", "text": "{r}{c}"},
+ {"id": "cnt", "type": "increment", "var": "antal", "by": 1},
+ {"id": "fin", "type": "set_var", "var": "klar", "text": "ja"},
+ ],
+ [
+ E("OUT", "IN", "each"),
+ E("IN", "app", "each"), E("app", "cnt"), E("cnt", "IN"),
+ E("IN", "OUT", "done"),
+ E("OUT", "fin", "done"),
+ ],
+ {"rader": ["1", "2"], "kol": ["a", "b"], "antal": "0"},
+ )
+ assert v["antal"] == 4
+ assert v["sista"] == "2b"
+ assert v["klar"] == "ja"
+
+
+# --- disabled ---
+def test_disabled_node_is_skipped():
+ v, _ = run(
+ [
+ {"id": "a", "type": "set_var", "var": "x", "text": "1"},
+ {"id": "b", "type": "increment", "var": "x", "by": 9, "enabled": False},
+ {"id": "c", "type": "increment", "var": "x", "by": 1},
+ ],
+ [E("a", "b"), E("b", "c")],
+ )
+ assert v["x"] == 2
+
+
+# --- felpropagering ---
+def test_error_stops_by_default():
+ with pytest.raises(FlowError):
+ run([{"id": "a", "type": "save_var", "var": "obefintlig", "filename": "x.json"}], [])
+
+
+def test_on_error_continue_swallows_and_proceeds():
+ v, logs = run(
+ [
+ {"id": "a", "type": "save_var", "var": "obefintlig", "filename": "x.json", "on_error": "continue"},
+ {"id": "b", "type": "set_var", "var": "r", "text": "nådde hit"},
+ ],
+ [E("a", "b")],
+ )
+ assert v["r"] == "nådde hit"
+ assert any("on_error=continue" in m for m in logs)
+
+
+# --- budget ---
+def test_infinite_loopback_hits_budget(monkeypatch):
+ monkeypatch.setattr(graph_engine, "MAX_NODE_STEPS", 200)
+ with pytest.raises(FlowError, match="För många steg"):
+ run(
+ [
+ {"id": "a", "type": "increment", "var": "x", "by": 1},
+ {"id": "b", "type": "increment", "var": "x", "by": 1},
+ ],
+ [E("a", "b"), E("b", "a")],
+ )
diff --git a/tests/test_migrate.py b/tests/test_migrate.py
new file mode 100644
index 0000000..ff1236a
--- /dev/null
+++ b/tests/test_migrate.py
@@ -0,0 +1,104 @@
+"""Migrering: gammalt nästlat {steps} -> nod/kant-graf, som sedan körs och ger
+samma variabelutfall som den gamla listmotorn gav."""
+
+import pytest
+
+from rpa import graph_engine, migrate
+from rpa.flow_engine import FlowError
+
+
+def run_steps(steps, vars_=None):
+ g = migrate.steps_to_graph(steps)
+ v = dict(vars_ or {})
+ logs: list[str] = []
+ graph_engine.run_graph(
+ g["nodes"], g["edges"], logs.append,
+ page=None, download_dir=None, vars_=v,
+ desktop_holder={"session": None}, start=g["start"],
+ )
+ return v, g, logs
+
+
+def test_linear_steps_preserve_order_and_ids():
+ v, g, _ = run_steps([
+ {"type": "set_var", "id": "s1", "var": "x", "text": "1"},
+ {"type": "increment", "id": "s2", "var": "x", "by": 2},
+ ])
+ assert [n["id"] for n in g["nodes"]] == ["s1", "s2"]
+ assert g["start"] == "s1"
+ assert v["x"] == 3
+
+
+def test_if_else_merges_and_continues():
+ steps = [
+ {"type": "set_var", "var": "r", "text": "start"},
+ {"type": "if", "branches": [
+ {"condition": {"kind": "var_equals", "var": "läge", "text": "prod"},
+ "steps": [{"type": "set_var", "var": "r", "text": "PROD"}]},
+ ], "else": [{"type": "set_var", "var": "r", "text": "ELSE"}]},
+ {"type": "set_var", "var": "efter", "text": "körd"},
+ ]
+ for läge, expected in (("prod", "PROD"), ("test", "ELSE")):
+ v, _, _ = run_steps(steps, {"läge": läge})
+ assert v["r"] == expected
+ assert v["efter"] == "körd" # båda grenar mynnar i efter-noden
+
+
+def test_if_without_else_falls_through():
+ steps = [
+ {"type": "if", "branches": [
+ {"condition": {"kind": "var_equals", "var": "x", "text": "1"},
+ "steps": [{"type": "set_var", "var": "r", "text": "träff"}]},
+ ]},
+ {"type": "set_var", "var": "efter", "text": "körd"},
+ ]
+ v, _, _ = run_steps(steps, {"x": "0"})
+ assert "r" not in v and v["efter"] == "körd"
+
+
+def test_loop_body_loops_back_and_done_continues():
+ steps = [
+ {"type": "loop", "over": "poster", "item_var": "it", "index_var": "i", "steps": [
+ {"type": "increment", "var": "antal", "by": 1},
+ {"type": "set_var", "var": "sista", "text": "{it}"},
+ ]},
+ {"type": "save_var", "var": "sista", "filename": "x.json"},
+ ]
+ v, g, _ = run_steps(steps, {"poster": ["a", "b"], "antal": "0"})
+ assert v["antal"] == 2 and v["sista"] == "b" and v["i"] == 2
+ # loop-noden har en each-kant och en done-kant, kroppen loopar tillbaka till in
+ loop_id = next(n["id"] for n in g["nodes"] if n["type"] == "loop")
+ handles = {e["sourceHandle"] for e in g["edges"] if e["source"] == loop_id}
+ assert {"each", "done"} <= handles
+ assert any(e["target"] == loop_id for e in g["edges"]) # loopback
+
+
+def test_jump_to_label_becomes_a_plain_edge():
+ steps = [
+ {"type": "set_var", "id": "a", "var": "r", "text": "1"},
+ {"type": "jump_to_label", "target": "slut"},
+ {"type": "set_var", "id": "skip", "var": "r", "text": "HOPPAD"},
+ {"type": "set_var", "id": "b", "var": "r", "text": "2", "label": "slut"},
+ ]
+ v, g, _ = run_steps(steps)
+ assert "jump_to_label" not in {n["type"] for n in g["nodes"]}
+ assert v["r"] == "2" # 'skip'-noden nås aldrig
+
+
+def test_flat_node_drops_container_keys():
+ g = migrate.steps_to_graph([
+ {"type": "loop", "over": "x", "steps": [{"type": "wait", "seconds": 0}]},
+ {"type": "if", "branches": [{"condition": {"kind": "var_equals", "var": "a", "text": "b"},
+ "steps": [{"type": "wait", "seconds": 0}]}], "else": []},
+ ])
+ for n in g["nodes"]:
+ assert "steps" not in n
+ if n["type"] == "if":
+ assert all(set(b) == {"condition"} for b in n["branches"])
+ else:
+ assert "branches" not in n
+
+
+def test_flow_to_graph_is_noop_for_graph_form():
+ graph = {"nodes": [{"id": "a", "type": "wait", "seconds": 1}], "edges": []}
+ assert migrate.flow_to_graph(graph) is graph
diff --git a/tests/test_schema.py b/tests/test_schema.py
index 0093a47..3e55bb9 100644
--- a/tests/test_schema.py
+++ b/tests/test_schema.py
@@ -1,53 +1,46 @@
from pathlib import Path
-from rpa import schema
+from rpa import migrate, schema
ROOT = Path(__file__).resolve().parent.parent
-GOOD_FLOW = {
- "steps": [
- {"type": "goto", "url": "https://example.com/"},
- {"type": "wait_for_login", "timeout": 120},
- {"type": "list", "value": "a", "var": "links"},
- {
- "type": "loop",
- "over": "links",
- "steps": [
- {"type": "extract_text", "by": "css", "value": ":scope", "scope": "item", "var": "lnk"},
- {"type": "list_append", "var": "collected", "text": "{lnk}"},
- ],
- },
- {
- "type": "if",
- "branches": [
- {"condition": {"kind": "var_contains", "var": "collected", "text": "x"},
- "steps": [{"type": "screenshot", "name": "hit"}]},
- ],
- "else": [{"type": "screenshot", "name": "miss"}],
- },
- {"type": "save_var", "var": "collected", "filename": "out.csv"},
- ],
- "settings": {"headless": True},
-}
-
-
-DESKTOP_FLOW = {
- "steps": [
- {"type": "desktop_launch", "path": "notepad.exe"},
- {"type": "desktop_type", "by": "control_type", "value": "Edit", "text": "hej {namn}"},
- {"type": "desktop_press_key", "keys": "^s"},
- {"type": "desktop_click_xy", "x": 10, "y": 20},
- ],
- "settings": {},
-}
-
-
-def test_good_flow_validates():
- assert schema.validate_flow(GOOD_FLOW) == []
-
-
-def test_desktop_flow_validates():
- assert schema.validate_flow(DESKTOP_FLOW) == []
+
+def g(steps, settings=None):
+ """Bygg en graf ur en steglista (via migreringen) + valfria settings."""
+ graph = migrate.steps_to_graph(steps)
+ graph["settings"] = settings or {}
+ return graph
+
+
+GOOD = g([
+ {"type": "goto", "url": "https://example.com/"},
+ {"type": "wait_for_login", "timeout": 120},
+ {"type": "list", "value": "a", "var": "links"},
+ {"type": "loop", "over": "links", "steps": [
+ {"type": "extract_text", "by": "css", "value": ":scope", "scope": "item", "var": "lnk"},
+ {"type": "list_append", "var": "collected", "text": "{lnk}"},
+ ]},
+ {"type": "if", "branches": [
+ {"condition": {"kind": "var_contains", "var": "collected", "text": "x"},
+ "steps": [{"type": "screenshot", "name": "hit"}]},
+ ], "else": [{"type": "screenshot", "name": "miss"}]},
+ {"type": "save_var", "var": "collected", "filename": "out.csv"},
+], {"headless": True})
+
+DESKTOP = g([
+ {"type": "desktop_launch", "path": "notepad.exe"},
+ {"type": "desktop_type", "by": "control_type", "value": "Edit", "text": "hej {namn}"},
+ {"type": "desktop_press_key", "keys": "^s"},
+ {"type": "desktop_click_xy", "x": 10, "y": 20},
+])
+
+
+def test_good_graph_validates():
+ assert schema.validate_graph(GOOD) == []
+
+
+def test_desktop_graph_validates():
+ assert schema.validate_graph(DESKTOP) == []
def test_desktop_steps_have_category():
@@ -56,53 +49,79 @@ def test_desktop_steps_have_category():
assert all(s["category"] == "desktop" for s in desktop)
-def test_unknown_step_type_rejected():
- errors = schema.validate_flow({"steps": [{"type": "teleport"}]})
- assert any("okänd stegtyp" in e for e in errors)
+def test_jump_to_label_removed():
+ assert "jump_to_label" not in schema.STEP_TYPE_NAMES
+
+
+def test_unknown_node_type_rejected():
+ errors = schema.validate_graph({"nodes": [{"id": "n1", "type": "teleport"}], "edges": []})
+ assert any("okänd nodtyp" in e for e in errors)
def test_missing_required_field_strict_only():
- flow = {"steps": [{"type": "goto"}]} # url saknas
- assert any("url" in e for e in schema.validate_flow(flow, strict=True))
- assert schema.validate_flow(flow, strict=False) == []
+ flow = {"nodes": [{"id": "n1", "type": "goto"}], "edges": []}
+ assert any("url" in e for e in schema.validate_graph(flow, strict=True))
+ assert schema.validate_graph(flow, strict=False) == []
def test_bad_enum_rejected_even_non_strict():
- flow = {"steps": [{"type": "click", "value": "Knapp", "by": "telepathy"}]}
- assert any(".by" in e for e in schema.validate_flow(flow, strict=False))
+ flow = {"nodes": [{"id": "n1", "type": "click", "value": "Knapp", "by": "telepathy"}], "edges": []}
+ assert any(".by" in e for e in schema.validate_graph(flow, strict=False))
+
+
+def test_edge_to_unknown_node_rejected():
+ flow = {"nodes": [{"id": "n1", "type": "wait", "seconds": 1}],
+ "edges": [{"source": "n1", "target": "ghost"}]}
+ assert any("okänd nod" in e for e in schema.validate_graph(flow))
+
+
+def test_bad_source_handle_rejected():
+ flow = {"nodes": [{"id": "a", "type": "wait", "seconds": 1}, {"id": "b", "type": "wait", "seconds": 1}],
+ "edges": [{"source": "a", "target": "b", "sourceHandle": "case9"}]}
+ assert any("utgångshandtag" in e for e in schema.validate_graph(flow))
+
+
+def test_if_node_handles_and_branch_validation():
+ ifnode = {"id": "f", "type": "if", "branches": [{"condition": {"kind": "var_equals", "var": "x", "text": "1"}}]}
+ h = schema.node_handles(ifnode)
+ assert h["outputs"] == ["case0", "else"]
+ assert schema.validate_graph({"nodes": [ifnode], "edges": []}) == []
+ # inga grenar -> fel
+ assert any("kräver minst en gren" in e for e in schema.validate_graph(
+ {"nodes": [{"id": "f", "type": "if", "branches": []}], "edges": []}))
-def test_if_requires_branch():
- errors = schema.validate_flow({"steps": [{"type": "if", "branches": []}]})
- assert any("kräver minst en gren" in e for e in errors)
+def test_loop_node_handles():
+ assert schema.node_handles("loop")["outputs"] == ["each", "done"]
def test_unknown_setting_rejected():
- errors = schema.validate_flow({"steps": [], "settings": {"turbo": True}})
+ errors = schema.validate_graph({"nodes": [], "edges": [], "settings": {"turbo": True}})
assert any("turbo" in e for e in errors)
-def test_engine_and_schema_step_types_match():
- """Drift-vakt: varje stegtyp som flow_engine hanterar ska finnas i STEP_TYPES
- och tvärtom."""
- src = (ROOT / "rpa" / "flow_engine.py").read_text(encoding="utf-8")
- engine_types = schema.engine_step_types_from_source(src)
+def test_engine_and_schema_node_types_match():
+ """Drift-vakt: varje nodtyp motorn hanterar ska finnas i STEP_TYPES och tvärtom."""
+ fe = (ROOT / "rpa" / "flow_engine.py").read_text(encoding="utf-8")
+ ge = (ROOT / "rpa" / "graph_engine.py").read_text(encoding="utf-8")
+ engine_types = schema.engine_step_types_from_source(fe, ge)
schema_types = set(schema.STEP_TYPE_NAMES)
assert engine_types == schema_types, (
- f"bara i motorn: {engine_types - schema_types}; "
- f"bara i schema: {schema_types - engine_types}"
+ f"bara i motorn: {engine_types - schema_types}; bara i schema: {schema_types - engine_types}"
)
-def test_json_schema_is_serializable_and_recursive():
+def test_json_schema_is_serializable_graph():
import json
js = schema.flow_json_schema()
json.dumps(js)
- assert js["$defs"]["step"]["properties"]["steps"]["items"] == {"$ref": "#/$defs/step"}
+ assert "node" in js["$defs"] and "edge" in js["$defs"]
+ assert js["required"] == ["nodes", "edges"]
def test_reference_markdown_mentions_every_step():
md = schema.step_reference_markdown()
for name in schema.STEP_TYPE_NAMES:
assert f"`{name}`" in md
+ assert "sourceHandle" in md and "each" in md
diff --git a/tests/test_skip_browser.py b/tests/test_skip_browser.py
index c124cb0..20a2ec0 100644
--- a/tests/test_skip_browser.py
+++ b/tests/test_skip_browser.py
@@ -2,70 +2,68 @@
import pytest
-from rpa import flow_engine
+from rpa import flow_engine, graph_engine, migrate
-# --- ren logik: _flow_needs_browser --------------------------------------
+def nodes(steps):
+ return migrate.steps_to_graph(steps)["nodes"]
+
def test_var_only_flow_needs_no_browser():
- steps = [
+ assert graph_engine.graph_needs_browser(nodes([
{"type": "set_var", "var": "x", "text": "1"},
{"type": "increment", "var": "x"},
{"type": "save_var", "var": "x", "filename": "x.json"},
{"type": "wait", "seconds": 0},
- ]
- assert flow_engine._flow_needs_browser(steps) is False
+ ])) is False
def test_desktop_only_flow_needs_no_browser():
- steps = [
+ assert graph_engine.graph_needs_browser(nodes([
{"type": "desktop_launch", "path": "notepad.exe"},
{"type": "loop", "over": "x", "steps": [{"type": "desktop_click", "by": "name", "value": "OK"}]},
{"type": "if", "branches": [{"condition": {"kind": "var_equals", "var": "x", "text": "1"},
"steps": [{"type": "desktop_screenshot"}]}]},
- ]
- assert flow_engine._flow_needs_browser(steps) is False
+ ])) is False
def test_goto_needs_browser():
- assert flow_engine._flow_needs_browser([{"type": "goto", "url": "https://x"}]) is True
+ assert graph_engine.graph_needs_browser(nodes([{"type": "goto", "url": "https://x"}])) is True
-def test_loop_with_browser_body_needs_browser():
- steps = [{"type": "loop", "over": "x", "steps": [{"type": "click", "by": "text", "value": "A"}]}]
- assert flow_engine._flow_needs_browser(steps) is True
+def test_loop_body_browser_node_is_its_own_node():
+ # click blir en egen nod -> upptäcks oberoende av loopen
+ assert graph_engine.graph_needs_browser(nodes([
+ {"type": "loop", "over": "x", "steps": [{"type": "click", "by": "text", "value": "A"}]},
+ ])) is True
def test_if_with_element_condition_needs_browser():
- steps = [{"type": "if", "branches": [{"condition": {"kind": "element_exists", "by": "css", "value": ".x"},
- "steps": [{"type": "desktop_screenshot"}]}]}]
- assert flow_engine._flow_needs_browser(steps) is True
+ assert graph_engine.graph_needs_browser(nodes([
+ {"type": "if", "branches": [{"condition": {"kind": "element_exists", "by": "css", "value": ".x"},
+ "steps": [{"type": "desktop_screenshot"}]}]},
+ ])) is True
-def test_disabled_browser_step_does_not_force_browser():
- steps = [
- {"type": "desktop_launch", "path": "notepad.exe"},
- {"type": "click", "by": "text", "value": "X", "enabled": False},
- ]
- assert flow_engine._flow_needs_browser(steps) is False
+def test_disabled_browser_node_does_not_force_browser():
+ assert graph_engine.graph_needs_browser([
+ {"id": "a", "type": "desktop_launch", "path": "x"},
+ {"id": "b", "type": "click", "by": "text", "value": "Y", "enabled": False},
+ ]) is False
-def test_first_browser_step_names_the_trigger():
- steps = [
- {"type": "desktop_launch", "path": "x"},
- {"type": "click", "by": "text", "value": "Y", "label": "knappen"},
+def test_first_browser_node_names_the_trigger():
+ ns = [
+ {"id": "a", "type": "desktop_launch", "path": "x"},
+ {"id": "b", "type": "click", "by": "text", "value": "Y", "label": "knappen"},
]
- assert flow_engine._first_browser_step(steps) == "click 'knappen'"
- assert flow_engine._first_browser_step([{"type": "set_var", "var": "a", "text": "1"}]) is None
-
+ assert graph_engine.first_browser_node(ns) == "click 'knappen'"
+ assert graph_engine.first_browser_node([{"id": "a", "type": "set_var", "var": "a", "text": "1"}]) is None
-# --- integration: webbläsaren startas inte -------------------------------
def test_run_flow_skips_browser_for_var_only(tmp_path, monkeypatch):
- def boom(*a, **k):
- raise AssertionError("launch_context anropades trots att flödet inte behöver webbläsaren")
-
- monkeypatch.setattr(flow_engine, "launch_context", boom)
+ monkeypatch.setattr(flow_engine, "launch_context",
+ lambda *a, **k: (_ for _ in ()).throw(AssertionError("browser launched!")))
monkeypatch.setattr(flow_engine.config, "EXPORT_DIR", tmp_path)
logs: list[str] = []
@@ -84,11 +82,9 @@ def test_run_flow_skips_browser_for_var_only(tmp_path, monkeypatch):
assert any("utan webbläsare" in m for m in logs)
-def test_run_flow_forced_skip_then_browser_step_errors(monkeypatch):
- monkeypatch.setattr(
- flow_engine, "launch_context",
- lambda *a, **k: (_ for _ in ()).throw(AssertionError("should not launch")),
- )
+def test_run_flow_forced_skip_then_browser_node_errors(monkeypatch):
+ monkeypatch.setattr(flow_engine, "launch_context",
+ lambda *a, **k: (_ for _ in ()).throw(AssertionError("should not launch")))
result: dict = {}
flow_engine.run_flow(
[{"type": "goto", "url": "https://example.com/"}],
@@ -96,5 +92,5 @@ def test_run_flow_forced_skip_then_browser_step_errors(monkeypatch):
settings={"skip_browser": True},
on_finished=lambda err, v: result.update({"err": err}),
)
- assert isinstance(result["err"], flow_engine.FlowError)
+ assert result["err"] is not None
assert "kräver webbläsaren" in str(result["err"])
diff --git a/webui/app.py b/webui/app.py
index 50f26b4..f456c60 100644
--- a/webui/app.py
+++ b/webui/app.py
@@ -256,7 +256,7 @@ def run_flow_endpoint(name: str, options: RunOptions = RunOptions()) -> dict:
close_mode = options.close_mode if options.close_mode in CLOSE_MODES else "always"
initial_vars = {str(k): str(v) for k, v in (options.initial_vars or {}).items()}
run_id = start_run(
- flow.get("steps", []),
+ flow,
close_mode=close_mode,
settings=flow.get("settings", {}),
initial_vars=initial_vars,