commit 1385902ba8b0b7c600cd85ca244ce007ee45531b
Author: Jens Kristoffersson <jenkr@vgregion.se>
AuthorDate: Tue Sep 8 15:22:20 2026 +0200
Commit: Jens Kristoffersson <jenkr@vgregion.se>
CommitDate: Tue Sep 8 15:22:20 2026 +0200
Subfloden (fas 1): call_flow-nod + lokala subgrafer i motorn
- rpa/schema.py: ny nodtyp call_flow (target + in/out-mappning). validate_graph
validerar aven flow["subflows"] (varje subgrafs noder/kanter), kollar att
#lokala mal finns och flaggar cirkulara lokala call_flow-anrop.
flow_json_schema/step_reference_markdown dokumenterar bada.
- rpa/graph_engine.py: run_graph tar resolve_call + _call_stack. call_flow-noden
laddar malgrafen, seedar en fräsch vars_ ur inputs-default + in-mappningen
(_interpolate), kor rekursivt, kopierar out-mappningen tillbaka. Cykel- och
djupvakt (MAX_CALL_DEPTH=20). graph_needs_browser/first_browser_node tittar in
i call_flow-mal nar en upplosare finns.
- rpa/flow_engine.py: run_flow bygger resolve_call (lokala subflows + externa
flows.load_flow), ny param entry_subflow (kor bara en lokal subgraf).
Foraldrad modul-docstring uppdaterad.
- rpa/flows.py: _normalize normaliserar subflows (id slug:as ur namn); save_flow
skriver subflows nar de finns.
- rpa/runs.py: entry_subflow vidarebefordras.
- Tester: call_flow (in/ut, default-fallback, saknat mal, ingen upplosare,
direkt cykel, djupvakt, on_error), subflow round-trip + valideringsfel,
entry_subflow, call_flow + skip_browser. 106 grona (+21).
---
rpa/flow_engine.py | 76 ++++++++++++++++++---------
rpa/flows.py | 21 +++++++-
rpa/graph_engine.py | 79 ++++++++++++++++++++++++++--
rpa/runs.py | 5 +-
rpa/schema.py | 128 +++++++++++++++++++++++++++++++++++++++++++++
tests/test_flows.py | 63 ++++++++++++++++++++++
tests/test_graph_engine.py | 98 +++++++++++++++++++++++++++++++++-
tests/test_run_vars.py | 34 +++++++++++-
tests/test_schema.py | 26 +++++++++
tests/test_skip_browser.py | 10 ++++
10 files changed, 505 insertions(+), 35 deletions(-)
diff --git a/rpa/flow_engine.py b/rpa/flow_engine.py
index 8a8a6b5..41f81e1 100644
--- a/rpa/flow_engine.py
+++ b/rpa/flow_engine.py
@@ -1,6 +1,9 @@
-"""Tolkar och kör användardefinierade flöden (lista av steg som JSON).
+"""Kör de enskilda stegen i ett flöde. Själva grafvandringen (kanter, if/loop/
+call_flow) ligger i rpa/graph_engine.py; run_flow() här är dispatchern som
+normaliserar indata (gammal {steps}-lista migreras) och sätter upp webbläsare/
+trace/close-mode. `_execute_step` kör en enskild nod.
-Stegtyper:
+Nodtyper (fält per typ):
goto {url}
wait {seconds}
wait_for_load {}
@@ -51,7 +54,7 @@ Stegtyper:
träffarna, till hjälp vid felsökning i UI:ts
variabelpanel - loop-steget bryr sig bara om
selector/count)
- loop {over, steps: [...], item_var?, index_var?}
+ loop {over, item_var?, index_var?} (utgångar each/done - se längre ner)
-- itererar över "list"-resultat (DOM-element,
scope=item fungerar som vanligt) ELLER över en
vanlig listvariabel (t.ex. från "Lägg till i
@@ -88,15 +91,15 @@ Stegtyper:
-- ser ut. Sparar i data/downloads/ (filename,
-- default "utskrift.pdf", stödjer {variabel}) och
-- lägger ev. sökvägen i var.
- if {branches: [{condition, steps: [...]}, ...], else?: [...]}
- -- första grenen vars condition är sann körs
- -- (branches[0] = "if", övriga = "else if");
- -- körs ingen körs "else" (om satt)
- jump_to_label {target} -- hoppar till steget vars "label" matchar "target", i
- -- SAMMA stegdel (bara syskon, inte in i/ut ur
- -- en loop/if); FlowError om etiketten saknas
- -- eller om för många hopp sker (skydd mot
- -- oändlig loop)
+ if {branches: [{condition}, ...]} -- utgångar case0..caseN (första sanna) + else;
+ -- grenarnas noder kopplas med kanter (graph_engine)
+ loop {over, item_var?, index_var?} -- utgångar each (per varv) / done; kroppen kopplar
+ -- tillbaka till loopens in
+ call_flow {target, in?: [{name, value}], out?: [{from, to}]}
+ -- kör ett annat flöde ("target" = flödesnamn) eller
+ -- en lokal subgraf ("#<subflow-id>") som ett steg;
+ -- 'in' mappar mot subflödets inputs ({variabel} ok),
+ -- 'out' kopierar subflödets slutvariabler tillbaka
condition (för "if"-grenar): {kind, ...}
kind: "element_exists" | "element_not_exists" -- {by, value, role?, scope?, nth?}
@@ -109,15 +112,12 @@ interpolering: text-fält (by=text/role-värden, "type"/"set_var"/list_append-li
namngivet fält ur en dict-variabel, t.ex. en rad från "Extrahera rad"), vilket
ersätts med textvärdet vid körning. Fungerar INTE i CSS-selectors (by=css),
URL:er eller filnamn - utom capture_pdf/save_attachment-stegens "filename".
-label: valfri nyckel på vilket steg som helst - namnger steget som mål för "jump_to_label"
-enabled: valfri nyckel på vilket steg som helst - satt till false hoppar
- körningen över steget (om det är en "loop"/"if" skippas hela
- blocket inklusive dess inre steg).
-on_error: valfri nyckel på vilket steg som helst - "stop" (default) avbryter
- körningen om steget kastar fel/timeout; "continue" loggar felet och
- går vidare till nästa steg (gör steget "hit or miss"). Sätts det på
- en "loop"/"if" gäller det hela blocket - ett ofångat fel i ett inre
- steg låter blocket avbrytas men körningen fortsätter efter det.
+label: valfri nyckel på vilken nod som helst - namnger noden (visas i körloggen)
+enabled: valfri nyckel på vilken nod som helst - satt till false hoppar
+ körningen över noden och följer "out".
+on_error: valfri nyckel på vilken nod som helst - "stop" (default) avbryter
+ körningen om noden kastar fel/timeout; "continue" loggar felet och
+ följer "out" ändå (gör noden "hit or miss").
"""
import base64
@@ -947,6 +947,7 @@ def run_flow(
trace_path: Path | None = None,
settings: dict | None = None,
initial_vars: dict | None = None,
+ entry_subflow: str | None = None,
) -> None:
"""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
@@ -975,7 +976,7 @@ 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
+ from . import flows as _flows, graph_engine, migrate
settings = settings or {}
download_dir = download_dir or config.DOWNLOAD_DIR
@@ -992,15 +993,37 @@ def run_flow(
graph = {"nodes": [], "edges": [], "start": None}
nodes, edges, start = graph["nodes"], graph["edges"], graph.get("start")
+ # Lokala subgrafer (call_flow target "#<id>") + valfri "kör bara subgrafen".
+ subflows = flow.get("subflows", []) if isinstance(flow, dict) else []
+ by_id = {s["id"]: s for s in subflows if isinstance(s, dict) and s.get("id")}
+ if entry_subflow:
+ sf = by_id.get(entry_subflow)
+ if sf is None:
+ log(f"FEL: okänd subgraf '{entry_subflow}'.")
+ if on_finished is not None:
+ on_finished(FlowError(f"Okänd subgraf: {entry_subflow}"), {})
+ return
+ nodes, edges, start = sf.get("nodes", []), sf.get("edges", []), sf.get("start")
+ log(f"Kör subgrafen '{sf.get('name') or entry_subflow}'.")
+
+ def resolve_call(target: str):
+ if target.startswith("#"):
+ s = by_id.get(target[1:])
+ if s is None:
+ raise FlowError(f"Okänd lokal subgraf: {target}")
+ return s.get("nodes", []), s.get("edges", []), s.get("start"), s.get("inputs", [])
+ f = _flows.load_flow(target)
+ return f["nodes"], f["edges"], f.get("start"), f.get("inputs", [])
+
# Rena desktop-/variabelflöden startar ingen webbläsare. Sätts explicit via
# 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)
+ skip_browser = bool(settings.get("skip_browser")) or not graph_engine.graph_needs_browser(nodes, resolve_call)
if skip_browser:
log("Kör utan webbläsare (skip_browser / inga webbläsarnoder).")
_run_and_finish(
None, None, nodes, edges, start, log, download_dir, close_mode, close_event,
- on_finished, None, settings, initial_vars,
+ on_finished, None, settings, initial_vars, resolve_call,
)
return
@@ -1015,13 +1038,13 @@ def run_flow(
page = get_page(context)
_run_and_finish(
context, page, nodes, edges, start, log, download_dir, close_mode, close_event,
- on_finished, trace_path, settings, initial_vars,
+ on_finished, trace_path, settings, initial_vars, resolve_call,
)
def _run_and_finish(
context, page, nodes, edges, start, log, download_dir, close_mode, close_event, on_finished,
- trace_path, settings, initial_vars=None,
+ trace_path, settings, initial_vars=None, resolve_call=None,
) -> None:
from . import graph_engine
@@ -1035,6 +1058,7 @@ def _run_and_finish(
graph_engine.run_graph(
nodes, edges, log, page=page, download_dir=download_dir,
vars_=vars_, desktop_holder=desktop_holder, start=start,
+ resolve_call=resolve_call,
)
except Exception as exc: # noqa: BLE001
error = exc
diff --git a/rpa/flows.py b/rpa/flows.py
index 16662bf..55d73a0 100644
--- a/rpa/flows.py
+++ b/rpa/flows.py
@@ -19,6 +19,12 @@ from .migrate import flow_to_graph
from .schema import validate_graph
NAME_RE = re.compile(r"^[a-zA-Z0-9_\-åäöÅÄÖ ]{1,80}$")
+_SLUG_RE = re.compile(r"[^a-z0-9]+")
+
+
+def _slug(text: str) -> str:
+ s = _SLUG_RE.sub("-", str(text).strip().lower()).strip("-")
+ return s or "sub"
class InvalidFlowName(ValueError):
@@ -58,9 +64,22 @@ def _normalize(flow: dict) -> dict:
flow.setdefault("layout", {})
flow.setdefault("inputs", [])
flow.setdefault("input_sets", {})
+ flow["subflows"] = [_normalize_subflow(sf) for sf in flow.get("subflows") or [] if isinstance(sf, dict)]
return flow
+def _normalize_subflow(sf: dict) -> dict:
+ name = str(sf.get("name") or sf.get("id") or "Subflöde")
+ return {
+ "id": str(sf.get("id") or _slug(name)),
+ "name": name,
+ "nodes": sf.get("nodes", []),
+ "edges": sf.get("edges", []),
+ "start": sf.get("start"),
+ "inputs": sf.get("inputs", []),
+ }
+
+
def load_flow(name: str) -> dict:
path = flow_path(name)
if not path.exists():
@@ -94,7 +113,7 @@ def save_flow(name: str, flow: dict, *, strict: bool = False) -> None:
}
if g.get("start"):
out["start"] = g["start"]
- for key in ("inputs", "input_sets", "layout"):
+ for key in ("inputs", "input_sets", "layout", "subflows"):
val = g.get(key)
if val:
out[key] = val
diff --git a/rpa/graph_engine.py b/rpa/graph_engine.py
index af38b67..acac73e 100644
--- a/rpa/graph_engine.py
+++ b/rpa/graph_engine.py
@@ -19,9 +19,11 @@ from .flow_engine import (
_NO_BROWSER_STEPS,
_evaluate_condition,
_execute_step,
+ _interpolate,
)
MAX_NODE_STEPS = 100_000 # skydd mot oändliga loop-back-cykler
+MAX_CALL_DEPTH = 20 # skydd mot djupa/cirkulära call_flow-anrop
def _adjacency(edges: list) -> dict[str, dict[str, list[str]]]:
@@ -42,8 +44,10 @@ def _start_node(nodes: list, edges: list, explicit: str | None) -> str | None:
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."""
+def graph_needs_browser(nodes: list, resolve=None, _seen: frozenset = frozenset()) -> bool:
+ """Platt variant av flow_engine._flow_needs_browser för nod-grafen. `resolve`
+ (target -> (nodes, edges, start, inputs)) låter oss titta in i call_flow-mål;
+ saknas den antar vi konservativt att webbläsaren behövs."""
for n in nodes:
if n.get("enabled") is False:
continue
@@ -57,6 +61,17 @@ def graph_needs_browser(nodes: list) -> bool:
):
return True
continue
+ if t == "call_flow":
+ target = str(n.get("target", ""))
+ if resolve is None or target in _seen:
+ return True
+ try:
+ sub_nodes, *_ = resolve(target)
+ except Exception: # noqa: BLE001 - okänt mål: låt körningen ge felet
+ return True
+ if graph_needs_browser(sub_nodes, resolve, _seen | {target}):
+ return True
+ continue
return True
return False
@@ -76,13 +91,21 @@ def first_browser_node(nodes: list) -> str | None:
):
return f"if{(' ' + n['label']) if n.get('label') else ''}"
continue
+ if t == "call_flow":
+ return f"call_flow → {n.get('target', '?')}"
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:
+def run_graph(
+ nodes, edges, log, *, page, download_dir, vars_, desktop_holder, start=None,
+ resolve_call=None, _call_stack=(),
+) -> None:
"""Kör grafen. Undantag från ett steg propagerar om inte nodens on_error =
- "continue". Muterar vars_ / desktop_holder."""
+ "continue". Muterar vars_ / desktop_holder.
+
+ resolve_call(target) -> (nodes, edges, start, inputs) används av call_flow-noder
+ för att köra ett annat flöde/en lokal subgraf som ett steg."""
nodes_by_id = {n["id"]: n for n in nodes}
out = _adjacency(edges)
@@ -164,6 +187,19 @@ def run_graph(nodes, edges, log, *, page, download_dir, vars_, desktop_holder, s
cursor = nxt if nxt is not None else cursor # ingen kropp -> loopa vidare
continue
+ # --- call_flow: kör ett annat flöde / lokal subgraf som ett steg ---
+ if t == "call_flow":
+ try:
+ _run_call_flow(node, log, page, download_dir, vars_, desktop_holder,
+ resolve_call, _call_stack)
+ except Exception as exc: # noqa: BLE001
+ if node.get("on_error") == "continue":
+ log(f"call_flow misslyckades - fortsätter (on_error=continue): {exc}")
+ else:
+ raise
+ cursor = go(cursor, "out")
+ continue
+
# --- vanligt steg (desktop hanteras internt av _execute_step) ---
try:
_execute_step(node, t, page, log, vars_, item_locator, download_dir, desktop_holder)
@@ -178,6 +214,41 @@ def run_graph(nodes, edges, log, *, page, download_dir, vars_, desktop_holder, s
log("Flöde klart.")
+def _run_call_flow(node, log, page, download_dir, vars_, desktop_holder, resolve_call, call_stack) -> None:
+ target = str(node.get("target", "")).strip()
+ if not target:
+ raise FlowError("call_flow saknar 'target'.")
+ if resolve_call is None:
+ raise FlowError(f"call_flow '{target}' kan inte köras här (ingen flödesupplösare).")
+ if target in call_stack:
+ raise FlowError(f"Cirkulärt call_flow-anrop: {' -> '.join([*call_stack, target])}.")
+ if len(call_stack) >= MAX_CALL_DEPTH:
+ raise FlowError(f"För djupa call_flow-anrop (>{MAX_CALL_DEPTH}) - avbryter.")
+
+ sub_nodes, sub_edges, sub_start, sub_inputs = resolve_call(target)
+
+ sub_vars: dict = {}
+ for inp in sub_inputs or []:
+ name = inp.get("name") if isinstance(inp, dict) else None
+ if name and inp.get("default") not in (None, ""):
+ sub_vars[str(name)] = str(inp["default"])
+ for m in node.get("in", []) or []:
+ if isinstance(m, dict) and m.get("name"):
+ sub_vars[str(m["name"])] = _interpolate(str(m.get("value", "")), vars_)
+
+ log(f"call_flow -> {target} (in: {', '.join(sorted(sub_vars)) or '-'})")
+ run_graph(
+ sub_nodes, sub_edges, log, page=page, download_dir=download_dir,
+ vars_=sub_vars, desktop_holder=desktop_holder, start=sub_start,
+ resolve_call=resolve_call, _call_stack=(*call_stack, target),
+ )
+
+ for m in node.get("out", []) or []:
+ if isinstance(m, dict) and m.get("from") and m.get("to"):
+ vars_[str(m["to"])] = sub_vars.get(str(m["from"]))
+ log(f"call_flow <- {target} klart")
+
+
def _loop_init(node: dict, vars_: dict) -> dict:
over_name = node.get("over")
over = vars_.get(over_name)
diff --git a/rpa/runs.py b/rpa/runs.py
index eb70f0e..2b2510b 100644
--- a/rpa/runs.py
+++ b/rpa/runs.py
@@ -25,9 +25,11 @@ def start_run(
close_mode: str = "always",
settings: dict | None = None,
initial_vars: dict | None = None,
+ entry_subflow: str | None = None,
) -> str:
"""`flow` = nod/kant-grafen ({nodes, edges, start}) eller gammalt {steps} /
- en steglista (migreras internt)."""
+ en steglista (migreras internt). entry_subflow kör bara den lokala subgrafen
+ med det id:t (för att testa en subflow-flik)."""
run_id = uuid.uuid4().hex[:12]
close_event = threading.Event()
trace_path = config.TRACE_DIR / f"{run_id}.zip"
@@ -59,6 +61,7 @@ def start_run(
trace_path=trace_path,
settings=settings,
initial_vars=initial_vars,
+ entry_subflow=entry_subflow,
)
except Exception as exc: # noqa: BLE001
_append_log(run_id, f"Oväntat fel: {exc}")
diff --git a/rpa/schema.py b/rpa/schema.py
index b339a13..833ce71 100644
--- a/rpa/schema.py
+++ b/rpa/schema.py
@@ -458,6 +458,16 @@ STEP_TYPES: list[dict] = [
"consumes_vars": ["over"],
"produces_vars": ["item_var", "index_var"],
},
+ {
+ "type": "call_flow", "category": "control",
+ "label_sv": "Anropa subflöde", "label_en": "Call subflow",
+ "help": "Kör ett annat flöde som ett steg. 'target' = ett externt sparat flödesnamn ELLER '#<subflow-id>' för en lokal subgraf i samma fil. 'in' mappar värden till subflödets inputs (stödjer {variabel}); 'out' kopierar subflödets slutvariabler tillbaka. Vanliga handtag in/out.",
+ "fields": [
+ {"key": "target", "kind": "text", "required": True,
+ "label_sv": "Målflöde (namn eller #subflow-id)",
+ "label_en": "Target flow (name or #subflow-id)"},
+ ],
+ },
# ---- Filer ----
{
"type": "download", "category": "files",
@@ -732,9 +742,79 @@ def validate_graph(flow: Any, *, strict: bool = True) -> list[str]:
if key not in allowed:
errors.append(f"settings.{key}: okänd inställning.")
+ _validate_subflows(flow, errors, strict)
return errors
+def _iter_graphs(flow: dict):
+ """(label, graph-dict) för huvudgrafen och varje subflow."""
+ yield "", flow
+ for sf in flow.get("subflows", []) or []:
+ if isinstance(sf, dict):
+ yield f"subflow {sf.get('name') or sf.get('id')!r}: ", sf
+
+
+def _validate_subflows(flow: dict, errors: list[str], strict: bool) -> None:
+ subflows = flow.get("subflows")
+ if subflows in (None, []):
+ return
+ if not isinstance(subflows, list):
+ errors.append("'subflows' måste vara en lista.")
+ return
+
+ local_ids: set[str] = set()
+ for i, sf in enumerate(subflows):
+ if not isinstance(sf, dict):
+ errors.append(f"subflows[{i}]: måste vara ett objekt.")
+ continue
+ sid = sf.get("id")
+ if not sid or not isinstance(sid, str):
+ errors.append(f"subflows[{i}].id: saknas.")
+ elif sid in local_ids:
+ errors.append(f"subflows[{i}].id: dubblett {sid!r}.")
+ else:
+ local_ids.add(sid)
+ # nod/kant-validering per subgraf (utan settings/subflows-rekursion)
+ for err in validate_graph(
+ {"nodes": sf.get("nodes", []), "edges": sf.get("edges", []), "start": sf.get("start")},
+ strict=strict,
+ ):
+ errors.append(f"subflow {sf.get('name') or sid!r}: {err}")
+
+ # call_flow-mål: #lokala id måste finnas; lokala anrop får inte bilda cykel
+ calls: dict[str, set[str]] = {}
+ for label, graph in _iter_graphs(flow):
+ owner = graph.get("id", "#main")
+ calls.setdefault(owner, set())
+ for node in graph.get("nodes", []) or []:
+ if not isinstance(node, dict) or node.get("type") != "call_flow":
+ continue
+ target = str(node.get("target", ""))
+ if target.startswith("#"):
+ tid = target[1:]
+ if tid not in local_ids:
+ errors.append(f"{label}call_flow: okänd lokal subflow {target!r}.")
+ else:
+ calls[owner].add(tid)
+
+ # cykeldetektering bland lokala anrop
+ WHITE, GREY, BLACK = 0, 1, 2
+ color = {k: WHITE for k in calls}
+
+ def visit(n: str) -> bool:
+ color[n] = GREY
+ for m in calls.get(n, ()):
+ if color.get(m, WHITE) == GREY or (color.get(m, WHITE) == WHITE and visit(m)):
+ return True
+ color[n] = BLACK
+ return False
+
+ for n in list(calls):
+ if color[n] == WHITE and visit(n):
+ errors.append("subflows: cirkulärt call_flow-anrop mellan lokala subgrafer.")
+ break
+
+
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:
@@ -777,6 +857,19 @@ def _validate_node_fields(node: dict, label: str, errors: list[str], strict: boo
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.")
+ elif t == "call_flow":
+ if strict and _is_blank(node.get("target")):
+ errors.append(f"{label}.target: obligatoriskt för 'call_flow'.")
+ for key, req in (("in", ("name",)), ("out", ("from", "to"))):
+ entries = node.get(key)
+ if entries in (None, []):
+ continue
+ if not isinstance(entries, list):
+ errors.append(f"{label}.{key}: måste vara en lista.")
+ continue
+ for ei, m in enumerate(entries):
+ if not isinstance(m, dict) or any(_is_blank(m.get(k)) for k in req):
+ errors.append(f"{label}.{key}[{ei}]: kräver {', '.join(req)}.")
def _validate_condition(cond: Any, path: str, errors: list[str]) -> None:
@@ -827,6 +920,18 @@ def flow_json_schema() -> dict:
},
},
"fields": {"type": "array", "items": {"type": "object"}},
+ "in": {
+ "type": "array",
+ "description": "call_flow: värden till subflödets inputs.",
+ "items": {"type": "object", "properties": {"name": {"type": "string"}, "value": {"type": "string"}},
+ "required": ["name"]},
+ },
+ "out": {
+ "type": "array",
+ "description": "call_flow: subflödets variabler -> anroparens variabler.",
+ "items": {"type": "object", "properties": {"from": {"type": "string"}, "to": {"type": "string"}},
+ "required": ["from", "to"]},
+ },
}
for key in sorted(all_field_keys):
node_props.setdefault(key, {"type": ["string", "number", "boolean"]})
@@ -842,6 +947,22 @@ def flow_json_schema() -> dict:
"type": "object",
"properties": {s["key"]: _setting_schema(s) for s in SETTINGS_SCHEMA},
},
+ "subflows": {
+ "type": "array",
+ "description": "Lokala namngivna subgrafer som kan anropas med call_flow (target '#<id>').",
+ "items": {
+ "type": "object",
+ "properties": {
+ "id": {"type": "string"},
+ "name": {"type": "string"},
+ "nodes": {"type": "array", "items": {"$ref": "#/$defs/node"}},
+ "edges": {"type": "array", "items": {"$ref": "#/$defs/edge"}},
+ "start": {"type": "string"},
+ "inputs": {"type": "array", "items": {"type": "object"}},
+ },
+ "required": ["id", "name", "nodes", "edges"],
+ },
+ },
"explanation": {"type": "string", "description": "Kort förklaring på svenska av vad flödet gör."},
},
"required": ["nodes", "edges"],
@@ -899,7 +1020,12 @@ def step_reference_markdown() -> str:
"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"
+ "- **`call_flow`**: kör ett annat flöde som ett steg. `target` = externt flödesnamn eller "
+ "`#<subflow-id>`. `in: [{name, value}]` mappar mot subflödets `inputs` ({variabel} tillåts); "
+ "`out: [{from, to}]` kopierar subflödets slutvariabler tillbaka. Vanliga handtag `in`/`out`.\n"
"- Ingen `jump_to_label` – rita en kant i stället.\n"
+ "\nValfri toppnyckel `subflows: [{id, name, nodes, edges, start, inputs}]` – lokala namngivna "
+ "subgrafer som `call_flow` kan anropa med `target: \"#<id>\"`.\n"
)
by_cat: dict[str, list[dict]] = {}
for spec in STEP_TYPES:
@@ -930,6 +1056,8 @@ def step_reference_markdown() -> str:
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")
+ if spec["type"] == "call_flow":
+ out.append(" - `in`: lista av {name, value}; `out`: lista av {from, to}")
out.append("\n### Villkor (för `if`-grenar)\n")
for c in CONDITION_KINDS:
diff --git a/tests/test_flows.py b/tests/test_flows.py
index e774cec..bbbf2d9 100644
--- a/tests/test_flows.py
+++ b/tests/test_flows.py
@@ -113,3 +113,66 @@ def test_layout_omitted_when_empty():
raw = json.loads((flows.config.FLOWS_DIR / "nolayout.json").read_text(encoding="utf-8"))
assert "layout" not in raw
+
+
+# --- subflöden ---
+def _with_subflows(**extra):
+ return graph(
+ [{"id": "n1", "type": "call_flow", "target": "#hjalp",
+ "in": [{"name": "x", "value": "1"}], "out": [{"from": "y", "to": "r"}]}],
+ subflows=[{
+ "id": "hjalp", "name": "Hjälpare",
+ "nodes": [{"id": "s1", "type": "set_var", "var": "y", "text": "{x}"}],
+ "edges": [], "start": "s1", "inputs": [{"name": "x"}],
+ }],
+ **extra,
+ )
+
+
+def test_subflows_round_trip():
+ flows.save_flow("m", _with_subflows())
+ loaded = flows.load_flow("m")
+ assert loaded["subflows"][0]["id"] == "hjalp"
+ assert loaded["subflows"][0]["nodes"][0]["type"] == "set_var"
+ assert loaded["nodes"][0]["type"] == "call_flow"
+
+
+def test_subflow_id_slugged_from_name_when_missing():
+ g = graph([], subflows=[{"name": "Hämta Bilagor",
+ "nodes": [], "edges": []}])
+ flows.save_flow("s", g)
+ assert flows.load_flow("s")["subflows"][0]["id"] == "h-mta-bilagor"
+
+
+def test_subflows_omitted_when_empty():
+ import json
+
+ flows.save_flow("nosub", graph([], subflows=[]))
+ raw = json.loads((flows.config.FLOWS_DIR / "nosub.json").read_text(encoding="utf-8"))
+ assert "subflows" not in raw
+
+
+def test_call_flow_to_unknown_local_subflow_rejected():
+ with pytest.raises(flows.InvalidFlow):
+ flows.save_flow("bad", graph(
+ [{"id": "n1", "type": "call_flow", "target": "#saknas"}],
+ subflows=[{"id": "finns", "name": "F", "nodes": [], "edges": []}],
+ ))
+
+
+def test_cyclic_local_subflows_rejected():
+ with pytest.raises(flows.InvalidFlow):
+ flows.save_flow("cyc", graph(
+ [{"id": "n1", "type": "call_flow", "target": "#a"}],
+ subflows=[
+ {"id": "a", "name": "A", "nodes": [{"id": "x", "type": "call_flow", "target": "#b"}], "edges": []},
+ {"id": "b", "name": "B", "nodes": [{"id": "y", "type": "call_flow", "target": "#a"}], "edges": []},
+ ],
+ ))
+
+
+def test_broken_subflow_node_rejected():
+ with pytest.raises(flows.InvalidFlow):
+ flows.save_flow("bs", graph([], subflows=[
+ {"id": "s", "name": "S", "nodes": [{"id": "n", "type": "nonesuch"}], "edges": []},
+ ]))
diff --git a/tests/test_graph_engine.py b/tests/test_graph_engine.py
index 20c6f95..d120746 100644
--- a/tests/test_graph_engine.py
+++ b/tests/test_graph_engine.py
@@ -6,13 +6,13 @@ from rpa import graph_engine
from rpa.flow_engine import FlowError
-def run(nodes, edges, vars_=None, start=None):
+def run(nodes, edges, vars_=None, start=None, resolve_call=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,
+ desktop_holder={"session": None}, start=start, resolve_call=resolve_call,
)
return v, logs
@@ -200,6 +200,100 @@ def test_on_error_continue_swallows_and_proceeds():
assert any("on_error=continue" in m for m in logs)
+# --- call_flow ---
+def _resolver(graphs):
+ """graphs: {target: (nodes, edges, start, inputs)}"""
+ def resolve(target):
+ if target not in graphs:
+ raise FlowError(f"okänt mål {target}")
+ return graphs[target]
+ return resolve
+
+
+def test_call_flow_maps_in_and_out():
+ sub = (
+ [{"id": "s1", "type": "set_var", "var": "y", "text": "{x}{x}"}],
+ [], "s1", [{"name": "x"}],
+ )
+ v, _ = run(
+ [
+ {"id": "a", "type": "set_var", "var": "v", "text": "ab"},
+ {"id": "c", "type": "call_flow", "target": "#dbl",
+ "in": [{"name": "x", "value": "{v}"}], "out": [{"from": "y", "to": "r"}]},
+ {"id": "d", "type": "set_var", "var": "after", "text": "ja"},
+ ],
+ [E("a", "c"), E("c", "d")],
+ resolve_call=_resolver({"#dbl": sub}),
+ )
+ assert v["r"] == "abab"
+ assert v["after"] == "ja" # fortsätter via out efteråt
+ assert "y" not in v # subflödets namnrymd läcker inte
+
+
+def test_call_flow_uses_input_default_when_unmapped():
+ sub = ([{"id": "s1", "type": "set_var", "var": "y", "text": "{x}"}], [], "s1",
+ [{"name": "x", "default": "fallback"}])
+ v, _ = run(
+ [{"id": "c", "type": "call_flow", "target": "#s", "out": [{"from": "y", "to": "r"}]}],
+ [],
+ resolve_call=_resolver({"#s": sub}),
+ )
+ assert v["r"] == "fallback"
+
+
+def test_call_flow_no_out_mapping_keeps_caller_clean():
+ sub = ([{"id": "s1", "type": "set_var", "var": "y", "text": "1"}], [], "s1", [])
+ v, _ = run(
+ [{"id": "c", "type": "call_flow", "target": "#s"}],
+ [],
+ resolve_call=_resolver({"#s": sub}),
+ )
+ assert v == {}
+
+
+def test_call_flow_missing_target_raises():
+ with pytest.raises(FlowError, match="okänt mål|target"):
+ run([{"id": "c", "type": "call_flow", "target": "#nope"}], [],
+ resolve_call=_resolver({}))
+
+
+def test_call_flow_without_resolver_raises():
+ with pytest.raises(FlowError, match="flödesupplösare"):
+ run([{"id": "c", "type": "call_flow", "target": "x"}], [])
+
+
+def test_call_flow_direct_cycle_raises():
+ # subgrafen anropar sig själv
+ sub_nodes = [{"id": "s1", "type": "call_flow", "target": "#loop"}]
+ graphs = {"#loop": (sub_nodes, [], "s1", [])}
+ with pytest.raises(FlowError, match="[Cc]irkulärt"):
+ run([{"id": "c", "type": "call_flow", "target": "#loop"}], [],
+ resolve_call=_resolver(graphs))
+
+
+def test_call_flow_depth_guard(monkeypatch):
+ monkeypatch.setattr(graph_engine, "MAX_CALL_DEPTH", 3)
+ # varje nivå anropar nästa: #l0 -> #l1 -> #l2 -> #l3 ...
+ graphs = {}
+ for i in range(10):
+ graphs[f"#l{i}"] = ([{"id": "n", "type": "call_flow", "target": f"#l{i + 1}"}], [], "n", [])
+ with pytest.raises(FlowError, match="djupa call_flow"):
+ run([{"id": "c", "type": "call_flow", "target": "#l0"}], [],
+ resolve_call=_resolver(graphs))
+
+
+def test_call_flow_on_error_continue():
+ v, logs = run(
+ [
+ {"id": "c", "type": "call_flow", "target": "#missing", "on_error": "continue"},
+ {"id": "d", "type": "set_var", "var": "r", "text": "nådde hit"},
+ ],
+ [E("c", "d")],
+ resolve_call=_resolver({}),
+ )
+ assert v["r"] == "nådde hit"
+
+
# --- budget ---
def test_infinite_loopback_hits_budget(monkeypatch):
monkeypatch.setattr(graph_engine, "MAX_NODE_STEPS", 200)
diff --git a/tests/test_run_vars.py b/tests/test_run_vars.py
index 7fc104c..6271fe9 100644
--- a/tests/test_run_vars.py
+++ b/tests/test_run_vars.py
@@ -47,8 +47,40 @@ def test_start_run_propagates_initial_vars(monkeypatch):
kw.get("on_finished") and kw["on_finished"](None, {})
monkeypatch.setattr(runs, "run_flow", fake_run_flow)
- runs.start_run([], initial_vars={"k": "v"})
+ runs.start_run([], initial_vars={"k": "v"}, entry_subflow="sub1")
import time
time.sleep(0.1)
assert seen.get("initial_vars") == {"k": "v"}
+ assert seen.get("entry_subflow") == "sub1"
+
+
+def test_entry_subflow_runs_only_that_subgraph():
+ flow = {
+ "format": "graph",
+ "nodes": [{"id": "a", "type": "set_var", "var": "main_ran", "text": "yes"}],
+ "edges": [], "start": "a", "settings": {},
+ "subflows": [{
+ "id": "calc", "name": "Calc",
+ "nodes": [{"id": "s", "type": "set_var", "var": "y", "text": "{x}!"}],
+ "edges": [], "start": "s", "inputs": [{"name": "x"}],
+ }],
+ }
+ result: dict = {}
+ flow_engine.run_flow(
+ flow, log=lambda _m: None, initial_vars={"x": "hej"}, entry_subflow="calc",
+ on_finished=lambda err, v: result.update({"err": err, "vars": v}),
+ )
+ assert result["err"] is None
+ assert result["vars"]["y"] == "hej!"
+ assert "main_ran" not in result["vars"]
+
+
+def test_entry_subflow_unknown_id_errors():
+ result: dict = {}
+ flow_engine.run_flow(
+ {"format": "graph", "nodes": [], "edges": [], "subflows": []},
+ log=lambda _m: None, entry_subflow="nope",
+ on_finished=lambda err, v: result.update({"err": err}),
+ )
+ assert result["err"] is not None
diff --git a/tests/test_schema.py b/tests/test_schema.py
index 3e55bb9..32a3ba5 100644
--- a/tests/test_schema.py
+++ b/tests/test_schema.py
@@ -95,6 +95,32 @@ def test_loop_node_handles():
assert schema.node_handles("loop")["outputs"] == ["each", "done"]
+def test_call_flow_in_step_types_with_default_handles():
+ assert "call_flow" in schema.STEP_TYPE_NAMES
+ assert schema.node_handles("call_flow")["outputs"] == ["out"]
+
+
+def test_call_flow_target_required_strict_only():
+ flow = {"nodes": [{"id": "n1", "type": "call_flow"}], "edges": []}
+ assert any("target" in e for e in schema.validate_graph(flow, strict=True))
+ assert schema.validate_graph(flow, strict=False) == []
+
+
+def test_call_flow_bad_io_entry_rejected():
+ flow = {"nodes": [{"id": "n1", "type": "call_flow", "target": "x",
+ "out": [{"from": "a"}]}], "edges": []}
+ assert any(".out[0]" in e for e in schema.validate_graph(flow, strict=False))
+
+
+def test_subflow_graph_is_validated():
+ flow = {
+ "nodes": [], "edges": [],
+ "subflows": [{"id": "s", "name": "S",
+ "nodes": [{"id": "n", "type": "goto"}], "edges": []}],
+ }
+ assert any("subflow" in e and "url" in e for e in schema.validate_graph(flow, strict=True))
+
+
def test_unknown_setting_rejected():
errors = schema.validate_graph({"nodes": [], "edges": [], "settings": {"turbo": True}})
assert any("turbo" in e for e in errors)
diff --git a/tests/test_skip_browser.py b/tests/test_skip_browser.py
index 20a2ec0..7f6c3f9 100644
--- a/tests/test_skip_browser.py
+++ b/tests/test_skip_browser.py
@@ -45,6 +45,16 @@ def test_if_with_element_condition_needs_browser():
])) is True
+def test_call_flow_needs_browser_depends_on_target():
+ resolve_pure = lambda tgt: ([{"id": "s", "type": "set_var", "var": "y", "text": "1"}], [], "s", [])
+ resolve_web = lambda tgt: ([{"id": "s", "type": "goto", "url": "https://x"}], [], "s", [])
+ call = [{"id": "c", "type": "call_flow", "target": "#s"}]
+ assert graph_engine.graph_needs_browser(call, resolve_pure) is False
+ assert graph_engine.graph_needs_browser(call, resolve_web) is True
+ # utan upplösare: konservativt True
+ assert graph_engine.graph_needs_browser(call) is True
+
+
def test_disabled_browser_node_does_not_force_browser():
assert graph_engine.graph_needs_browser([
{"id": "a", "type": "desktop_launch", "path": "x"},