"""Step 5 tests: decisions + continuations, tenant isolation, error paths.""" import json from pathlib import Path from uuid import UUID, uuid4 from app.flows.schemas import ApprovalCreate FIXTURES = Path(__file__).parent / "fixtures" / "flows" def load_fixture(name: str) -> dict: return json.loads((FIXTURES / name).read_text()) async def _create_flow(client, name): response = await client.post("/api/flows", json={"name": name}) assert response.status_code == 201, response.text return response.json() async def _version_and_deploy(client, flow_id, fixture="simple_rule.json"): response = await client.post( f"/api/flows/{flow_id}/versions", json={"langflow_json": load_fixture(fixture)}, ) assert response.status_code == 201, response.text response = await client.post(f"/api/flows/{flow_id}/versions/1/deploy") assert response.status_code == 200, response.text return response.json() async def _handoff_task(client, live_server, monkeypatch): """Run approval_handoff over threshold; returns (run, task).""" from app.settings import settings monkeypatch.setattr(settings, "callback_base_url", live_server) flow = await _create_flow(client, f"handoff-{uuid4().hex[:8]}") await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json") body = {"input": {"betrag": 250000, "schwelle": 200000}} run = (await client.post(f"/api/flows/{flow['id']}/run", json=body)).json() assert run["status"] == "pending_approval" tasks = (await client.get("/api/approvals")).json() task = next(t for t in tasks if t["run_id"] == run["id"]) return run, task async def test_approve_without_continuation(client, lfx_server, live_server, login, monkeypatch): await login() _, task = await _handoff_task(client, live_server, monkeypatch) response = await client.post( f"/api/approvals/{task['id']}/decision", json={"decision": "approve", "note": "ok"}, ) assert response.status_code == 200, response.text body = response.json() assert body["task"]["status"] == "approved" assert body["task"]["approver"] == "admin@example.com" assert body["task"]["decided_at"] is not None assert body["continuation_run"] is None async def test_approve_with_continuation_runs_facts( client, lfx_server, live_server, login, monkeypatch ): await login() cont = await _create_flow(client, f"cont-{uuid4().hex[:8]}") await _version_and_deploy(client, cont["id"], fixture="simple_rule.json") # Continuation is authoring-time wiring on the definition (admin-only); # run bodies cannot set it. response = await client.post( "/api/flows", json={"name": f"handoff-{uuid4().hex[:8]}", "continuation_flow_id": cont["id"]}, ) assert response.status_code == 201, response.text flow = response.json() await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json") from app.settings import settings monkeypatch.setattr(settings, "callback_base_url", live_server) run = ( await client.post( f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 250000, "schwelle": 200000}}, ) ).json() assert run["status"] == "pending_approval" tasks = (await client.get("/api/approvals")).json() task = next(t for t in tasks if t["run_id"] == run["id"]) assert task["continuation_flow_id"] == cont["id"] response = await client.post( f"/api/approvals/{task['id']}/decision", json={"decision": "approve"}, ) assert response.status_code == 200, response.text body = response.json() assert body["task"]["status"] == "approved" cont_run = body["continuation_run"] assert cont_run is not None assert cont_run["status"] == "completed" assert cont_run["tenant_id"] == task["tenant_id"] # Facts (betrag 250000 > Schwelle) flowed into the continuation. assert "Sonderfall" in cont_run["output"]["result"] async def test_client_continuation_is_ignored(client, lfx_server, live_server, login, monkeypatch): """A run body cannot chain arbitrary flows (G).""" from app.settings import settings await login() monkeypatch.setattr(settings, "callback_base_url", live_server) cont = await _create_flow(client, f"cont-{uuid4().hex[:8]}") await _version_and_deploy(client, cont["id"], fixture="simple_rule.json") flow = await _create_flow(client, f"handoff-{uuid4().hex[:8]}") await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json") run = ( await client.post( f"/api/flows/{flow['id']}/run", json={ "input": {"betrag": 250000, "schwelle": 200000}, "continuation_flow_id": cont["id"], }, ) ).json() assert run["status"] == "pending_approval" tasks = (await client.get("/api/approvals")).json() task = next(t for t in tasks if t["run_id"] == run["id"]) assert task["continuation_flow_id"] is None async def test_reject_task(client, lfx_server, live_server, login, monkeypatch): await login() _, task = await _handoff_task(client, live_server, monkeypatch) # Any logged-in user (not just admins) can decide; identity is the session. await login(email="chef@example.com", role="member") response = await client.post( f"/api/approvals/{task['id']}/decision", json={"decision": "reject", "note": "zu teuer"}, ) assert response.status_code == 200, response.text body = response.json() assert body["task"]["status"] == "rejected" assert body["task"]["approver"] == "chef@example.com" assert body["task"]["decision_note"] == "zu teuer" assert body["continuation_run"] is None async def test_decision_guards(client, lfx_server, live_server, login, monkeypatch): await login() _, task = await _handoff_task(client, live_server, monkeypatch) url = f"/api/approvals/{task['id']}/decision" response = await client.post(url, json={"decision": "approve"}) assert response.status_code == 200 response = await client.post(url, json={"decision": "reject"}) assert response.status_code == 409 response = await client.post(f"/api/approvals/{uuid4()}/decision", json={"decision": "approve"}) assert response.status_code == 404 _, task2 = await _handoff_task(client, live_server, monkeypatch) response = await client.post( f"/api/approvals/{task2['id']}/decision", json={"decision": "maybe"} ) assert response.status_code == 422 response = await client.post( f"/api/approvals/{task2['id']}/decision", json={"decision": "approve", "note": "x"}, ) # continuation unset -> fine, resolves to no continuation run assert response.status_code == 200 async def test_decision_bad_continuation_reference(client, session, login): from app.flows import store as flow_store await login() flow = await _create_flow(client, f"badcont-{uuid4().hex[:8]}") version = ( await client.post( f"/api/flows/{flow['id']}/versions", json={"langflow_json": load_fixture("simple_rule.json")}, ) ).json() run = flow_store.create_run( session, flow_store.get_version(session, UUID(version["id"])), input={}, ) task = flow_store.create_approval_task(session, ApprovalCreate(run_id=run.id, title="t")) # plant unknown continuation directly (store allows it pre-validation) task.continuation_flow_id = uuid4() session.add(task) session.commit() response = await client.post(f"/api/approvals/{task.id}/decision", json={"decision": "approve"}) assert response.status_code == 422 async def test_tenant_isolation(client, lfx_server, login): """Strict session scoping: explicit ?tenant_id= cannot widen reads (E).""" await login() flow = await _create_flow(client, f"tenant-{uuid4().hex[:8]}") await _version_and_deploy(client, flow["id"]) await login(email="a@example.com", role="member", tenant_id="pfarramt-a") response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}}) assert response.status_code == 200 run_a = response.json() assert run_a["tenant_id"] == "pfarramt-a" await login(email="b@example.com", role="member", tenant_id="pfarramt-b") response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}}) assert response.status_code == 200 # A forged tenant_id query param is ignored: b sees only b's runs. response = await client.get(f"/api/flows/{flow['id']}/runs", params={"tenant_id": "pfarramt-a"}) runs_b = response.json() assert len(runs_b) == 1 assert runs_b[0]["tenant_id"] == "pfarramt-b" # Cross-tenant run read 404s with or without the forged param. response = await client.get(f"/api/flows/runs/{run_a['id']}") assert response.status_code == 404 response = await client.get( f"/api/flows/runs/{run_a['id']}", params={"tenant_id": "pfarramt-a"} ) assert response.status_code == 404 await login(email="a@example.com", role="member", tenant_id="pfarramt-a") response = await client.get(f"/api/flows/runs/{run_a['id']}") assert response.status_code == 200 async def test_lfx_down_maps_to_503(client, lfx_server, monkeypatch, login): from app.settings import settings await login() flow = await _create_flow(client, f"down-{uuid4().hex[:8]}") await client.post( f"/api/flows/{flow['id']}/versions", json={"langflow_json": load_fixture("simple_rule.json")}, ) monkeypatch.setattr(settings, "lfx_url", "http://127.0.0.1:9") response = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy") assert response.status_code == 503 # Deploy for real, then break the URL again for the run call. monkeypatch.setattr(settings, "lfx_url", lfx_server) deployed = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy") assert deployed.status_code == 200 monkeypatch.setattr(settings, "lfx_url", "http://127.0.0.1:9") response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}}) assert response.status_code == 503 async def test_lfx_drift_maps_to_502(client, lfx_server, login): from app.flows.lfx_client import LfxClient from app.settings import settings await login() flow = await _create_flow(client, f"drift-{uuid4().hex[:8]}") deployed = await _version_and_deploy(client, flow["id"]) await LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None).delete_flow( deployed["lfx_flow_id"] ) response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}}) assert response.status_code == 502 async def test_corrupt_flow_json_deploy_maps_to_422(client, lfx_server, login): await login() flow = await _create_flow(client, f"corrupt-{uuid4().hex[:8]}") bogus = { "name": "bogus", "data": { "nodes": [ { "id": "Nope-x1", "data": { "id": "Nope-x1", "type": "Nope", "showNode": True, "node": {"template": {}, "outputs": []}, }, "type": "genericNode", } ], "edges": [], }, } response = await client.post(f"/api/flows/{flow['id']}/versions", json={"langflow_json": bogus}) assert response.status_code == 201, response.text response = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy") assert response.status_code == 422