foxygit / RPA-Remisser Log in
commits tags

/tests/test_skip_browser.py · 4.72 KB

raw
"""skip_browser: rena desktop-/variabelflöden startar ingen webbläsare."""

import pytest

from rpa import flow_engine, graph_engine, migrate


def nodes(steps):
    return migrate.steps_to_graph(steps)["nodes"]


def test_var_only_flow_needs_no_browser():
    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},
    ])) is False


def test_desktop_only_flow_needs_no_browser():
    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"}]}]},
    ])) is False


def test_goto_needs_browser():
    assert graph_engine.graph_needs_browser(nodes([{"type": "goto", "url": "https://x"}])) 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():
    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_ocr_prompt_lookup_dont_need_browser():
    assert graph_engine.graph_needs_browser(nodes([
        {"type": "ocr_read", "source": "pdf", "file": "r.pdf", "var": "t"},
        {"type": "prompt", "mode": "confirm", "message": "?"},
        {"type": "lookup", "from_var": "cfg", "key": "k", "var": "v"},
        {"type": "merge_json", "file": "x.json", "key": "k", "value": "v"},
    ])) is False


def test_read_url_needs_browser():
    assert graph_engine.graph_needs_browser(nodes([{"type": "read_url", "var": "u"}])) 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"},
        {"id": "b", "type": "click", "by": "text", "value": "Y", "enabled": False},
    ]) is False


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 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


def test_run_flow_skips_browser_for_var_only(tmp_path, monkeypatch):
    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] = []
    result: dict = {}
    flow_engine.run_flow(
        [
            {"type": "set_var", "var": "a", "text": "hej"},
            {"type": "save_var", "var": "a", "filename": "a.json"},
        ],
        log=logs.append,
        on_finished=lambda err, v: result.update({"err": err, "vars": v}),
    )
    assert result["err"] is None
    assert result["vars"]["a"] == "hej"
    assert (tmp_path / "a.json").exists()
    assert any("utan webbläsare" in m for m in logs)


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/"}],
        log=lambda _m: None,
        settings={"skip_browser": True},
        on_finished=lambda err, v: result.update({"err": err}),
    )
    assert result["err"] is not None
    assert "kräver webbläsaren" in str(result["err"])