6c195bc2e7
- POST /api/approvals/{id}/decision (approve starts continuation flow
with task facts; reject closes; 404/409/422 guards)
- Tenant scoping: list filters + cross-tenant reads 404
- Error mapping covered: LFX down 503, LFX drift 502, corrupt flow 422
- 34 tests green
255 lines
9.1 KiB
Python
255 lines
9.1 KiB
Python
"""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, continuation_flow_id=None):
|
|
"""Run approval_handoff over threshold; returns (run, task)."""
|
|
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},
|
|
"callback_base_url": live_server,
|
|
}
|
|
if continuation_flow_id is not None:
|
|
body["continuation_flow_id"] = continuation_flow_id
|
|
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):
|
|
_, task = await _handoff_task(client, live_server)
|
|
response = await client.post(
|
|
f"/api/approvals/{task['id']}/decision",
|
|
json={"decision": "approve", "approver": "sachbearbeiter", "note": "ok"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["task"]["status"] == "approved"
|
|
assert body["task"]["approver"] == "sachbearbeiter"
|
|
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):
|
|
cont = await _create_flow(client, f"cont-{uuid4().hex[:8]}")
|
|
await _version_and_deploy(client, cont["id"], fixture="simple_rule.json")
|
|
|
|
_, task = await _handoff_task(
|
|
client, live_server, continuation_flow_id=cont["id"]
|
|
)
|
|
assert task["continuation_flow_id"] == cont["id"]
|
|
|
|
response = await client.post(
|
|
f"/api/approvals/{task['id']}/decision",
|
|
json={"decision": "approve", "approver": "chef"},
|
|
)
|
|
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_reject_task(client, lfx_server, live_server):
|
|
_, task = await _handoff_task(client, live_server)
|
|
response = await client.post(
|
|
f"/api/approvals/{task['id']}/decision",
|
|
json={"decision": "reject", "approver": "chef", "note": "zu teuer"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["task"]["status"] == "rejected"
|
|
assert body["task"]["decision_note"] == "zu teuer"
|
|
assert body["continuation_run"] is None
|
|
|
|
|
|
async def test_decision_guards(client, lfx_server, live_server):
|
|
_, task = await _handoff_task(client, live_server)
|
|
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)
|
|
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):
|
|
from app.flows import store as flow_store
|
|
|
|
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):
|
|
flow = await _create_flow(client, f"tenant-{uuid4().hex[:8]}")
|
|
await _version_and_deploy(client, flow["id"])
|
|
|
|
for tenant in ("pfarramt-a", "pfarramt-b"):
|
|
response = await client.post(
|
|
f"/api/flows/{flow['id']}/run",
|
|
json={"input": {"betrag": 1}, "tenant_id": tenant},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
response = await client.get(
|
|
f"/api/flows/{flow['id']}/runs", params={"tenant_id": "pfarramt-a"}
|
|
)
|
|
runs_a = response.json()
|
|
assert len(runs_a) == 1
|
|
assert runs_a[0]["tenant_id"] == "pfarramt-a"
|
|
|
|
response = await client.get(f"/api/flows/{flow['id']}/runs")
|
|
assert len(response.json()) == 2
|
|
|
|
# Cross-tenant read with scoping 404s.
|
|
response = await client.get(
|
|
f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-b"}
|
|
)
|
|
assert response.status_code == 404
|
|
response = await client.get(
|
|
f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-a"}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
async def test_lfx_down_maps_to_503(client, lfx_server, monkeypatch):
|
|
from app.settings import settings
|
|
|
|
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):
|
|
from app.flows.lfx_client import LfxClient
|
|
from app.settings import settings
|
|
|
|
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):
|
|
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
|