commits
tags
"""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, 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, resolve_call=resolve_call,
)
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)
# --- 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)
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")],
)