"""Task control flow (app/crud/tasks.py engine + app/crud/task_logic.py):
If / Else nodes, Loop nodes (for each / repeat / while), variables by path
everywhere, and output sources."""

import pytest

from app.crud.params import apply_output_params, get_path, resolve_input_params
from app.crud.task_logic import compare, rules_pass


def test_paths_reach_into_records_and_lists():
    ctx = {"record": {"status": "open", "vendor": {"id": 7}}, "rows": [{"amount": 5}, {"amount": 9}], "a.b": "literal"}
    assert get_path(ctx, "record.vendor.id") == 7
    assert get_path(ctx, "rows.1.amount") == 9 and get_path(ctx, "rows[0].amount") == 5
    assert get_path(ctx, "a.b") == "literal"  # an exact key wins over a path
    assert get_path(ctx, "record.missing.deeper") is None

    ok, _, inputs = resolve_input_params(
        [{"name": "vid", "type": "int", "source_field": "record.vendor.id"}, {"name": "flag", "type": "boolean", "source_field": "f"}, {"name": "rec", "type": "object", "source_field": "record"}],
        {**ctx, "f": "yes"},
    )
    assert ok and inputs == {"vid": 7, "flag": True, "rec": ctx["record"]}


def test_output_sources_and_nullable_objects():
    ok, _, out = apply_output_params(
        [{"name": "record", "type": "object", "source": "*"}, {"name": "first_amount", "type": "number", "source": "0.amount"}],
        [{"amount": 2.5}],
    )
    assert not ok  # "*" on a list isn't an object — the declared type is enforced
    ok, _, out = apply_output_params([{"name": "record", "type": "object", "source": "*"}, {"name": "status", "type": "string"}], {"status": "open"})
    assert ok and out == {"record": {"status": "open"}, "status": "open"}
    ok, _, out = apply_output_params([{"name": "record", "type": "object", "source": "*"}], None)
    assert ok and out == {"record": None}  # "not found" is a value you can branch on


@pytest.mark.parametrize(
    "op, actual, expected, result",
    [
        ("equals", 5, "5", True),
        ("equals", True, "true", True),
        ("gt", "150", "100", True),
        ("lte", 3.5, 3.5, True),
        ("in", "b", "a, b, c", True),
        ("not_in", "z", ["a", "b"], True),
        ("contains", "Pending review", "REVIEW", True),
        ("contains", ["x", "y"], "y", True),
        ("starts_with", "INV-001", "inv", True),
        ("is_empty", "", None, True),
        ("is_empty", [], None, True),
        ("is_not_empty", {"a": 1}, None, True),
        ("is_true", "yes", None, True),
        ("gt", None, 1, False),
    ],
)
def test_loose_comparisons(op, actual, expected, result):
    assert compare(op, actual, expected) is result


def test_rules_all_any_and_variable_values():
    ctx = {"item": {"amount": 150}, "limit": 100}
    big = {"driver_field": "item.amount", "operator": "gt", "value": "{{limit}}"}
    never = {"driver_field": "item.amount", "operator": "lt", "value": 0}
    assert rules_pass([big], "all", ctx) and not rules_pass([big, never], "all", ctx) and rules_pass([big, never], "any", ctx)


async def _master(client, auth_headers, rows):
    master = await client.post(
        "/masters",
        headers=auth_headers,
        json={"name": "Loop orders", "field_schema": [{"name": "order_no", "type": "varchar"}, {"name": "amount", "type": "integer"}, {"name": "priority", "type": "varchar"}]},
    )
    assert master.status_code == 201, master.text
    for r in rows:
        resp = await client.post(f"/masters/{master.json()['id']}/records", headers=auth_headers, json={"field_values": r})
        assert resp.status_code == 201, resp.text
    return master.json()["id"]


async def _run(client, auth_headers, nodes, context=None):
    task = await client.post("/tasks", headers=auth_headers, json={"name": "Flow test", "nodes": nodes})
    assert task.status_code == 201, task.text
    resp = await client.post(f"/tasks/{task.json()['id']}/run", headers=auth_headers, json={"context": context or {}, "allow_draft": True})
    assert resp.status_code == 200, resp.text
    return resp.json()


async def test_loop_over_data_list_with_if_else_updates_each_record(client, auth_headers):
    master_id = await _master(client, auth_headers, [{"order_no": "A-1", "amount": 50}, {"order_no": "A-2", "amount": 150}, {"order_no": "A-3", "amount": 300}])
    datasource = {"datasource_type": "master", "master_id": master_id}
    nodes = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "list"}]},
        {"key": "list", "type": "data_list", "label": "All orders", **datasource, "output_params": [{"name": "orders", "type": "list"}], "next_nodes": [{"to": "each"}]},
        {
            "key": "each", "type": "loop", "label": "Each order", "loop_mode": "for_each",
            "input_params": [{"name": "items", "type": "list", "source_field": "orders"}],
            "loop_collect": ["priority"],
            "output_params": [{"name": "priorities", "type": "list", "source": "results"}, {"name": "processed", "type": "int", "source": "count"}],
            "next_nodes": [{"to": "big", "branch": "body"}, {"to": "summary", "branch": "done"}],
        },
        {
            "key": "big", "type": "condition", "label": "Big order?",
            "rules": [{"driver_field": "item.amount", "operator": "gt", "value": "{{threshold}}"}],
            "next_nodes": [{"to": "mark_high", "branch": "true"}, {"to": "mark_low", "branch": "false"}],
        },
        {
            "key": "mark_high", "type": "data_update", "label": "Mark high", **datasource, "match_field": "order_no",
            "input_params": [{"name": "match_value", "type": "string", "source_field": "item.order_no"}, {"name": "priority", "type": "string", "source_field": "high_label"}],
            "output_params": [{"name": "priority", "type": "string"}],
            "next_nodes": [{"to": "each"}],  # back to the loop = next item
        },
        {
            "key": "mark_low", "type": "data_update", "label": "Mark low", **datasource, "match_field": "order_no",
            "input_params": [{"name": "match_value", "type": "string", "source_field": "item.order_no"}, {"name": "priority", "type": "string", "source_field": "low_label"}],
            "output_params": [{"name": "priority", "type": "string"}],
        },
        {
            "key": "summary", "type": "generate_sequence", "label": "Batch number", "sequence_key": f"loop-test-{master_id}", "sequence_prefix": "BATCH-",
            "output_params": [{"name": "batch", "type": "string"}],
        },
    ]
    result = await _run(client, auth_headers, nodes, {"threshold": 100, "high_label": "high", "low_label": "low"})
    assert result["ok"], result["trace"]
    ctx = result["context"]
    assert ctx["processed"] == 3 and ctx["priorities"] == ["low", "high", "high"] and ctx["batch"] == "BATCH-0001"
    assert "item" not in ctx and "index" not in ctx  # loop variables don't leak past the loop

    records = await client.get(f"/masters/{master_id}/records", headers=auth_headers)
    by_no = {r["field_values"]["order_no"]: r["field_values"].get("priority") for r in records.json()}
    assert by_no == {"A-1": "low", "A-2": "high", "A-3": "high"}

    iterations = [t["iteration"] for t in result["trace"] if t["node_key"] == "big"]
    assert iterations == ["Each order 1/3", "Each order 2/3", "Each order 3/3"]
    assert [t["detail"].split(" —")[0] for t in result["trace"] if t["node_key"] == "big"] == ["false", "true", "true"]


async def test_data_get_record_object_and_template_variables(client, auth_headers):
    master_id = await _master(client, auth_headers, [{"order_no": "B-7", "amount": 70}])
    nodes = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "get"}]},
        {
            "key": "get", "type": "data_get", "label": "Find order", "master_id": master_id, "match_field": "order_no",
            "input_params": [{"name": "match_value", "type": "string", "source_field": "orderNo"}],
            "output_params": [{"name": "order", "type": "object", "source": "*"}],
            "next_nodes": [{"to": "found"}],
        },
        {
            "key": "found", "type": "condition", "label": "Found?", "rules": [{"driver_field": "order", "operator": "is_not_empty"}],
            "next_nodes": [{"to": "number", "branch": "true"}],
        },
        {
            # No declared inputs: the prefix template reads variables straight from the run.
            "key": "number", "type": "generate_sequence", "label": "Ref", "sequence_key": f"tmpl-{master_id}",
            "sequence_prefix": "REF-", "output_params": [{"name": "ref", "type": "string"}],
        },
    ]
    found = await _run(client, auth_headers, nodes, {"orderNo": "B-7"})
    assert found["ok"], found["trace"]
    assert found["context"]["order"]["amount"] == 70 and found["context"]["ref"].startswith("REF-")

    missing = await _run(client, auth_headers, nodes, {"orderNo": "nope"})
    assert missing["ok"] and missing["context"]["order"] is None and "ref" not in missing["context"]
    assert [t["node_key"] for t in missing["trace"]] == ["start", "get", "found"]


async def test_repeat_and_while_loops_and_failures(client, auth_headers):
    def seq(key, name):
        return {"key": key, "type": "generate_sequence", "label": key, "sequence_key": f"{key}-{name}", "output_params": [{"name": "n", "type": "int"}]}

    import uuid

    tag = uuid.uuid4().hex[:8]
    repeat = await _run(
        client, auth_headers,
        [
            {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "loop"}]},
            {"key": "loop", "type": "loop", "label": "Three times", "loop_mode": "repeat", "loop_collect": ["n", "index"],
             "input_params": [{"name": "count", "type": "int", "source_field": "times"}],
             "output_params": [{"name": "all", "type": "list", "source": "results"}], "next_nodes": [{"to": "s", "branch": "body"}]},
            seq("s", tag),
        ],
        {"times": 3},
    )
    assert repeat["ok"], repeat["trace"]
    assert repeat["context"]["all"] == [{"n": 1, "index": 0}, {"n": 2, "index": 1}, {"n": 3, "index": 2}]

    # While n < 4: the counter node runs until the rule turns false.
    tag2 = uuid.uuid4().hex[:8]
    whiled = await _run(
        client, auth_headers,
        [
            {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "loop"}]},
            {"key": "loop", "type": "loop", "label": "Until four", "loop_mode": "while", "rules": [{"driver_field": "n", "operator": "lt", "value": 4}],
             "output_params": [{"name": "rounds", "type": "int", "source": "count"}], "next_nodes": [{"to": "s", "branch": "body"}]},
            seq("s", tag2),
        ],
        {"n": 0},
    )
    assert whiled["ok"] and whiled["context"]["rounds"] == 4 and whiled["context"]["n"] == 4

    runaway = await _run(
        client, auth_headers,
        [
            {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "loop"}]},
            {"key": "loop", "type": "loop", "label": "Forever", "loop_mode": "while", "loop_max_iterations": 5, "rules": [{"driver_field": "go", "operator": "is_true"}]},
        ],
        {"go": True},
    )
    assert not runaway["ok"] and "never became false" in runaway["trace"][-1]["detail"]

    too_many = await _run(
        client, auth_headers,
        [
            {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "loop"}]},
            {"key": "loop", "type": "loop", "label": "Each", "loop_max_iterations": 2, "input_params": [{"name": "items", "type": "list", "source_field": "xs"}]},
        ],
        {"xs": [1, 2, 3]},
    )
    assert not too_many["ok"] and "over this loop's limit" in too_many["trace"][-1]["detail"]


async def test_a_failing_iteration_stops_or_is_counted(client, auth_headers):
    body_fails = {"key": "bad", "type": "data_get", "label": "Needs a datasource"}  # not configured -> fails
    base = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "loop"}]},
        {"key": "loop", "type": "loop", "label": "Each", "input_params": [{"name": "items", "type": "list", "source_field": "xs"}],
         "output_params": [{"name": "failed", "type": "int"}], "next_nodes": [{"to": "bad", "branch": "body"}]},
        body_fails,
    ]
    stopped = await _run(client, auth_headers, base, {"xs": [1, 2]})
    assert not stopped["ok"] and "stopped at iteration 1" in stopped["trace"][-1]["detail"]

    base[1] = {**base[1], "loop_continue_on_error": True}
    counted = await _run(client, auth_headers, base, {"xs": [1, 2]})
    assert counted["ok"] and counted["context"]["failed"] == 2

def test_fixed_value_inputs():
    ctx = {"payment_label": "RTI/7", "amount": 1000, "record": {"office": "Kollam"}}
    ok, _, inputs = resolve_input_params(
        [
            {"name": "gateway", "type": "string", "source_field": "", "value": "Razorpay"},
            {"name": "note", "type": "string", "value": "Paid via {{payment_label}} at {{record.office}}{{missing}}"},
            {"name": "amount", "type": "number", "value": "{{amount}}"},  # one placeholder keeps the variable's type
            {"name": "blank", "type": "string", "value": "{{missing}}", "required": False},  # empty + optional = skipped
        ],
        ctx,
    )
    assert ok and inputs == {"gateway": "Razorpay", "note": "Paid via RTI/7 at Kollam", "amount": 1000}
    ok, detail, _ = resolve_input_params([{"name": "who", "type": "string", "value": "{{nobody}}"}], ctx)
    assert not ok and "who" in detail


async def test_data_update_writes_fixed_text(client, auth_headers):
    master_id = await _master(client, auth_headers, [])
    nodes = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "save"}]},
        {
            "key": "save", "type": "data_update", "label": "Save", "datasource_type": "master", "master_id": master_id,
            "input_params": [
                {"name": "order_no", "type": "string", "required": False, "value": "ORD-{{n}}"},
                {"name": "priority", "type": "string", "required": False, "value": "Razorpay"},
            ],
            "output_params": [{"name": "saved", "type": "object", "source": "*"}],
        },
    ]
    result = await _run(client, auth_headers, nodes, {"n": 42})
    assert result["ok"] is True, result
    saved = result["context"]["saved"]
    assert saved["order_no"] == "ORD-42" and saved["priority"] == "Razorpay"

async def test_data_update_records_the_running_user(client, auth_headers, test_tenant):
    import uuid as _uuid

    from sqlalchemy import select

    from tests.conftest import AsyncSessionLocal
    from app.crud.tasks import _exec_data_update, NodeInputs
    from app.models.master import MasterRecord

    master_id = await _master(client, auth_headers, [])
    save = {
        "key": "save", "type": "data_update", "label": "Save", "datasource_type": "master", "master_id": master_id, "match_field": "order_no",
        "input_params": [
            {"name": "match_value", "type": "string", "required": False, "source_field": "order_no"},
            {"name": "order_no", "type": "string", "required": False, "source_field": "order_no"},
            {"name": "priority", "type": "string", "required": False, "source_field": "priority"},
        ],
    }
    nodes = [{"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "save"}]}, save]
    admin_id = test_tenant["admin_id"]

    # A caller claiming to be someone else is ignored — the run is always for the real caller.
    created = await _run(client, auth_headers, nodes, {"order_no": "A-1", "priority": "low", "currentUserId": str(_uuid.uuid4())})
    assert created["ok"] is True, created
    updated = await _run(client, auth_headers, nodes, {"order_no": "A-1", "priority": "high"})
    assert updated["ok"] is True, updated

    async with AsyncSessionLocal() as db:
        rows = (await db.execute(select(MasterRecord).where(MasterRecord.master_id == _uuid.UUID(master_id)))).scalars().all()
        assert [(r.field_values["priority"], r.created_by, r.updated_by) for r in rows] == [("high", admin_id, admin_id)]

        # A scheduled run's saved currentUserId that isn't a user of this tenant records no one (no FK failure).
        ok, _, _ = await _exec_data_update(
            db, test_tenant["tenant_id"], {**save, "match_field": None}, NodeInputs({"order_no": "B-1"}, {"currentUserId": str(_uuid.uuid4())})
        )
        assert ok
        await db.flush()
        stray = (await db.execute(select(MasterRecord).where(MasterRecord.master_id == _uuid.UUID(master_id), MasterRecord.created_by.is_(None)))).scalars().all()
        assert [r.field_values["order_no"] for r in stray] == ["B-1"]
        await db.rollback()