"""Task execution history (app/crud/task_history.py): every run kept with its steps, iterations and values, surviving the run's own rollback, with secrets masked, sizes capped, tenants apart and old runs cleaned up.""" import json import uuid from datetime import datetime, timedelta, timezone import httpx import pytest from sqlalchemy import select, text from app.core import jobs from app.core.auth import create_access_token from app.core.database import engine from app.core.jobs import tenant_session from app.crud import task_history, tasks as task_engine from app.crud.secrets import MASK, Scrubber from app.crud.task_history import MAX_STRING_CHARS, MAX_VALUE_CHARS, RunTrigger, _clean, _Limit from app.crud.tasks import run_task from app.models.automation import Task, TaskRun, TaskRunStep from app.models.identity import Role, Tenant, User, UserRole from tests.conftest import AsyncSessionLocal SECRET = "sk_live_9f8e7d6c5b4a" START = {"key": "start", "type": "start", "label": "Start"} def _node(key, type_, next_=None, **extra): return {"key": key, "type": type_, "label": extra.pop("label", key.title()), "next_nodes": [{"to": next_}] if next_ else [], **extra} def _set(key, next_=None, **values): params = [{"name": k, "type": "any", "value": v} for k, v in values.items()] return _node(key, "set_variable", next_, input_params=params) async def _task(client, headers, nodes, variables=None): resp = await client.post("/tasks", headers=headers, json={"name": "History task", "nodes": nodes, "variables": variables or []}) assert resp.status_code == 201, resp.text return resp.json()["id"] async def _run(client, headers, task_id, context=None, **body): resp = await client.post(f"/tasks/{task_id}/run", headers=headers, json={"context": context or {}, "allow_draft": True, **body}) assert resp.status_code == 200, resp.text return resp.json() async def _steps(client, headers, task_id, run_id): resp = await client.get(f"/tasks/{task_id}/runs/{run_id}/steps", headers=headers, params={"limit": 500}) assert resp.status_code == 200, resp.text return resp.json()["items"] async def _step(client, headers, task_id, run_id, seq): resp = await client.get(f"/tasks/{task_id}/runs/{run_id}/steps/{seq}", headers=headers) assert resp.status_code == 200, resp.text return resp.json() async def _everything_stored(tenant_id) -> str: """Every stored byte of the tenant's history, as the database has it.""" async with AsyncSessionLocal() as db: runs = (await db.execute(text("SELECT to_jsonb(r)::text FROM task_runs r WHERE tenant_id = :t"), {"t": tenant_id})).scalars().all() steps = (await db.execute(text("SELECT to_jsonb(s)::text FROM task_run_steps s WHERE tenant_id = :t"), {"t": tenant_id})).scalars().all() assert runs return "\n".join([*runs, *steps]) async def test_a_successful_run_is_kept_with_each_steps_values(client, auth_headers, test_tenant): nodes = [ _node("start", "start", "greet"), _node("greet", "set_variable", "code", label="Greet", input_params=[{"name": "greeting", "type": "string", "value": "Hello {{who}}"}]), _node( "code", "custom_code", label="Shout", language="python", code="import sys, json\nctx = json.load(sys.stdin)\nprint(json.dumps({'loud': ctx['text'].upper()}))", input_params=[{"name": "text", "type": "string", "source_field": "greeting"}], output_params=[{"name": "shouted", "type": "string", "source": "loud"}], ), ] task_id = await _task(client, auth_headers, nodes) result = await _run(client, auth_headers, task_id, {"who": "Asha"}) assert result["ok"] is True, result page = (await client.get(f"/tasks/{task_id}/runs", headers=auth_headers)).json() assert [r["id"] for r in page["items"]] == [result["run_id"]] and page["has_more"] is False run = page["items"][0] assert run["status"] == "succeeded" and run["trigger"] == "designer" and run["task_status"] == "draft" assert run["started_by"] == str(test_tenant["admin_id"]) and run["started_by_name"] == "Pytest Admin" assert run["step_count"] == 3 and run["failed_steps"] == 0 and run["finished_at"] >= run["started_at"] detail = (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}", headers=auth_headers)).json() assert detail["input"]["who"] == "Asha" and detail["output"]["shouted"] == "HELLO ASHA" steps = await _steps(client, auth_headers, task_id, result["run_id"]) assert [(s["node_key"], s["node_label"], s["ok"]) for s in steps] == [("start", "Start", True), ("greet", "Greet", True), ("code", "Shout", True)] # The list itself carries no values: they come when a step is opened. assert "inputs" not in steps[0] shout = await _step(client, auth_headers, task_id, result["run_id"], 3) assert shout["inputs"] == {"text": "Hello Asha"} assert shout["outputs"] == {"loud": "HELLO ASHA"} and shout["saved"] == {"shouted": "HELLO ASHA"} assert shout["node_type"] == "custom_code" and shout["duration_ms"] >= 0 async def test_the_browsers_source_is_kept_as_a_label(client, auth_headers): task_id = await _task(client, auth_headers, [_node("start", "start")]) await client.post(f"/tasks/{task_id}/publish", headers=auth_headers) source = str(uuid.uuid4()) result = await _run(client, auth_headers, task_id, allow_draft=False, source={"kind": "page", "id": source}) run = (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}", headers=auth_headers)).json() assert run["trigger"] == "page" and run["trigger_ref"] == source and run["task_status"] == "published" assert (await client.get(f"/tasks/{task_id}/runs", headers=auth_headers, params={"trigger": "schedule"})).json()["items"] == [] assert len((await client.get(f"/tasks/{task_id}/runs", headers=auth_headers, params={"trigger": "page", "status": "succeeded"})).json()["items"]) == 1 async def test_a_failed_step_is_kept_with_its_inputs(client, auth_headers): nodes = [ _node("start", "start", "go"), _node("go", "custom_code", label="Boom", language="python", code="import sys\nprint('bad thing')\nsys.exit(3)", input_params=[{"name": "order", "type": "string", "value": "A-1"}]), ] task_id = await _task(client, auth_headers, nodes) result = await _run(client, auth_headers, task_id) assert result["ok"] is False run = (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}", headers=auth_headers)).json() assert run["status"] == "failed" and run["failed_steps"] == 1 and run["detail"].startswith("Boom:") failed = (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}/steps", headers=auth_headers, params={"failed_only": True})).json()["items"] assert [s["node_key"] for s in failed] == ["go"] assert (await _step(client, auth_headers, task_id, result["run_id"], failed[0]["seq"]))["inputs"] == {"order": "A-1"} async def test_a_failed_loop_is_summed_up_by_the_step_inside_it(client, auth_headers): nodes = [ _node("start", "start", "each"), _node("each", "loop", label="Each", loop_mode="repeat", input_params=[{"name": "count", "type": "int", "value": "2"}]) | {"next_nodes": [{"to": "bad", "branch": "body"}]}, _node("bad", "custom_code", "each", label="Bad step", language="python", code="import sys\nsys.exit(1)"), ] task_id = await _task(client, auth_headers, nodes) result = await _run(client, auth_headers, task_id) run = (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}", headers=auth_headers)).json() assert run["status"] == "failed" and run["failed_steps"] == 2 and run["detail"].startswith("Bad step:") async def _committed_task(tenant_id, nodes) -> uuid.UUID: async with AsyncSessionLocal() as db: async with db.begin(): task = Task(tenant_id=tenant_id, name="Direct", nodes=nodes, variables=[], status="published") db.add(task) await db.flush() return task.id async def test_history_survives_the_runs_rollback_and_an_error_inside_a_step(test_tenant, monkeypatch): tenant_id = test_tenant["tenant_id"] task_id = await _committed_task(tenant_id, [_node("start", "start", "write"), _node("write", "custom_code", code="x", input_params=[{"name": "n", "type": "int", "value": "7"}])]) async def explode(db, tenant, node, inputs): raise RuntimeError("the database went away") monkeypatch.setitem(task_engine._HANDLERS, "custom_code", explode) trigger = RunTrigger("schedule", ref="job-1", name="Nightly") with pytest.raises(RuntimeError): async with tenant_session(tenant_id) as db: task = (await db.execute(select(Task).where(Task.id == task_id))).scalar_one() await run_task(db, tenant_id, task, {}, trigger=trigger) async with AsyncSessionLocal() as db: run = (await db.execute(select(TaskRun).where(TaskRun.id == trigger.run_id))).scalar_one() steps = (await db.execute(select(TaskRunStep).where(TaskRunStep.run_id == run.id).order_by(TaskRunStep.seq))).scalars().all() assert run.status == "error" and "the database went away" in run.detail assert run.trigger == "schedule" and run.trigger_name == "Nightly" and run.finished_at is not None assert [(s.node_key, s.ok) for s in steps] == [("start", True), ("write", False)] assert steps[1].inputs == {"n": 7} and "RuntimeError" in steps[1].detail async def test_a_run_in_a_transaction_that_holds_its_task_doesnt_wait_on_itself(test_tenant): """The task is uncommitted and locked by the run's own transaction, which history can't see. It must save anyway, without waiting.""" tenant_id = test_tenant["tenant_id"] trigger = RunTrigger("other") with pytest.raises(ZeroDivisionError): async with tenant_session(tenant_id) as db: task = Task(tenant_id=tenant_id, name="Not committed", nodes=[_node("start", "start")], variables=[], status="draft") db.add(task) await db.flush() ok, _, _ = await run_task(db, tenant_id, task, {}, trigger=trigger) assert ok 1 / 0 async with AsyncSessionLocal() as db: run = await db.get(TaskRun, trigger.run_id) assert run is not None and run.status == "succeeded" and run.step_count == 1 def _nested_nodes(): return [ _node("start", "start", "pages"), _node( "pages", "loop", label="Each page", loop_mode="repeat", input_params=[{"name": "count", "type": "int", "value": "2"}], loop_item_var="p", loop_index_var="page", next_nodes=[], ) | {"next_nodes": [{"to": "fetch", "branch": "body"}, {"to": "end", "branch": "done"}]}, _set("fetch", "companies", rows="{{page}}"), _node("companies", "loop", label="Each company", loop_mode="for_each", input_params=[{"name": "items", "type": "list", "source_field": "list"}], loop_item_var="company", loop_index_var="i") | {"next_nodes": [{"to": "save", "branch": "body"}, {"to": "pages", "branch": "done"}]}, _set("save", "companies", saved="{{company.name}}"), _set("end", None, finished="yes"), ] async def test_nested_loops_group_by_iteration(client, auth_headers): task_id = await _task(client, auth_headers, _nested_nodes()) result = await _run(client, auth_headers, task_id, {"list": [{"name": "Acme"}, {"name": "Birla"}, {"name": "Cipla"}]}) assert result["ok"] is True, result run_id = result["run_id"] url = f"/tasks/{task_id}/runs/{run_id}/timeline" top = (await client.get(url, headers=auth_headers)).json()["items"] assert [(i["kind"], i["node_key"] or i["label"]) for i in top] == [ ("step", "start"), ("step", "pages"), ("iteration", "Each page 1/2"), ("iteration", "Each page 2/2"), ("step", "pages"), ("step", "end"), ] # start, pages, fetch, companies, 3 × save, companies done — per page. assert top[2]["steps"] == 6 and top[2]["failed"] == 0 and top[2]["path"] == "Each page 1/2" page2 = (await client.get(url, headers=auth_headers, params={"parent": "Each page 2/2"})).json()["items"] assert [(i["kind"], i["node_key"] or i["label"]) for i in page2] == [ ("step", "fetch"), ("step", "companies"), ("iteration", "Each company 1/3"), ("iteration", "Each company 2/3"), ("iteration", "Each company 3/3"), ("step", "companies"), ] assert page2[3]["path"] == "Each page 2/2 › Each company 2/3" inner = (await client.get(url, headers=auth_headers, params={"parent": page2[3]["path"]})).json()["items"] assert [i["node_key"] for i in inner] == ["save"] and inner[0]["iteration"] == "Each page 2/2 › Each company 2/3" values = await _step(client, auth_headers, task_id, run_id, inner[0]["seq"]) assert values["outputs"] == {"saved": "Birla"} async def test_every_step_is_kept_past_the_traces_cap_and_a_level_pages(client, auth_headers, test_tenant): nodes = [ _node("start", "start", "loop"), _node("loop", "loop", label="Each", loop_mode="repeat", loop_max_iterations=2000, input_params=[{"name": "count", "type": "int", "value": "1100"}]) | {"next_nodes": [{"to": "a", "branch": "body"}]}, _set("a", "loop", v="{{index}}"), ] task_id = await _task(client, auth_headers, nodes) result = await _run(client, auth_headers, task_id) assert result["ok"] is True assert len(result["trace"]) == 1001 # capped, with its "not shown" line, as before run = (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}", headers=auth_headers)).json() assert run["step_count"] == 1103 and run["lost_steps"] == 0 seen, after = [], 0 while True: page = (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}/timeline", headers=auth_headers, params={"after_seq": after, "limit": 300})).json() seen += page["items"] after = page["items"][-1]["seq"] if not page["has_more"]: break assert len(seen) == 3 + 1100 and [i["seq"] for i in seen] == sorted(i["seq"] for i in seen) assert sum(i["kind"] == "iteration" for i in seen) == 1100 async def test_secrets_never_reach_stored_history(client, auth_headers, test_tenant, monkeypatch): """A long secret, a short one the Scrubber won't touch, an input read from each, echoes in a result, a key and a number, and a template rendered into the request shown as sent.""" for name, value in (("PAY_KEY", SECRET), ("PIN", "x7q"), ("NUM", "918273645")): assert (await client.post("/secrets", headers=auth_headers, json={"name": name, "value": value})).status_code == 201 captured = {} def handler(request: httpx.Request) -> httpx.Response: captured["auth"] = request.headers["authorization"] captured["url"] = str(request.url) return httpx.Response(200, json={"echo": request.headers["authorization"], SECRET: "as a key", "number": 918273645, "pin": "x7q and more", "access_token": "tok-abcdef", "safe": "kept"}) real_client = httpx.AsyncClient monkeypatch.setattr(task_engine.httpx, "AsyncClient", lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw)) nodes = [ _node("start", "start", "call"), _node( "call", "api_rest", "fail", url="https://api.example.test/pay?key={{secret.PAY_KEY}}&order={{order}}", method="POST", headers={"Authorization": "Bearer {{secret.PAY_KEY}}", "X-Pin": "{{secret.PIN}}"}, body_template='{"pin": "{{secret.PIN}}"}', input_params=[ {"name": "api_key", "type": "string", "source_field": "secret.PAY_KEY"}, {"name": "pin", "type": "string", "source_field": "secret.PIN"}, {"name": "wrapped", "type": "string", "value": "Bearer {{secret.PAY_KEY}}"}, {"name": "order", "type": "string", "value": "A-1"}, ], output_params=[{"name": "reply", "type": "any", "source": "*"}], ), _node("fail", "custom_code", language="python", code=f"import sys, json\nctx = json.load(sys.stdin)\nprint('leak', ctx['k'])\nsys.exit(1)", input_params=[{"name": "k", "type": "string", "source_field": "secret.PAY_KEY"}]), ] task_id = await _task(client, auth_headers, nodes) result = await _run(client, auth_headers, task_id, {"mine": SECRET, "password": "hunter2hunter"}) assert result["ok"] is False assert SECRET in captured["auth"] and SECRET in captured["url"] # the real call really had it stored = await _everything_stored(test_tenant["tenant_id"]) for leak in (SECRET, "x7q", "918273645", "hunter2hunter", "tok-abcdef"): at = stored.find(leak) assert at < 0, f"{leak} stored: {stored[max(0, at - 120):at + 60]}" call = await _step(client, auth_headers, task_id, result["run_id"], 2) assert call["inputs"]["api_key"] == MASK and call["inputs"]["pin"] == MASK and call["inputs"]["wrapped"] == MASK assert call["inputs"]["order"] == "A-1" sent = call["inputs"]["$sent"] assert sent["url"] == f"https://api.example.test/pay?key={MASK}&order=A-1" and sent["method"] == "POST" assert sent["headers"]["X-Pin"] == MASK and sent["headers"]["Authorization"] == MASK body = call["outputs"]["body"] assert body["safe"] == "kept" and body["access_token"] == MASK and body["number"] == MASK assert all(SECRET not in k for k in body) def test_masking_comes_before_cutting(): scrub = Scrubber({"K": SECRET}) # The cut would fall in the middle of the secret. text_value = "a" * (MAX_STRING_CHARS - 5) + SECRET + "b" * 100 limit = _Limit(MAX_VALUE_CHARS) cleaned = _clean(text_value, scrub, limit) assert limit.cut and SECRET[:5] not in cleaned and cleaned.endswith("more characters not stored)") def test_big_values_stay_readable_and_bounded(): scrub = Scrubber([]) rows = [{"id": i, "name": f"Company {i}", "note": "x" * 100} for i in range(5000)] limit = _Limit(MAX_VALUE_CHARS) cleaned = _clean({"status": 200, "body": rows}, scrub, limit) assert limit.cut and cleaned["status"] == 200 and cleaned["body"][0]["name"] == "Company 0" assert "more not stored" in cleaned["body"][-1] assert len(json.dumps(cleaned)) < MAX_VALUE_CHARS * 1.5 # What jsonb refuses is removed; what JSON can't say is said in words. odd = _clean({"nul": "a\x00b", "pair": "\ud83d", "when": datetime(2026, 1, 1), "raw": b"abc", "nan": float("nan")}, scrub, _Limit(1000)) assert odd["nul"] == "ab" and odd["raw"] == "<3 bytes>" and odd["nan"] == "nan" and "2026" in odd["when"] json.dumps(odd, allow_nan=False) deep: dict = {} for _ in range(100): deep = {"x": deep} assert "too deeply" in json.dumps(_clean(deep, scrub, _Limit(10_000))) async def test_a_credential_found_under_its_key_is_masked_everywhere_after(client, auth_headers, test_tenant): """A login step's token isn't a Secret, but it is one: not in the step's detail (which quotes values), nor where a later step passes it on.""" nodes = [ _node("start", "start", "login"), _set("login", "use", access_token="tok-preview-abcdef"), _node("use", "set_variable", input_params=[{"name": "auth", "type": "string", "value": "Bearer {{access_token}}"}]), ] task_id = await _task(client, auth_headers, nodes) result = await _run(client, auth_headers, task_id) assert result["ok"] is True and result["context"]["auth"] == "Bearer tok-preview-abcdef" # the run itself is unchanged assert "tok-preview-abcdef" not in await _everything_stored(test_tenant["tenant_id"]) used = await _step(client, auth_headers, task_id, result["run_id"], 3) assert used["inputs"] == {"auth": f"Bearer {MASK}"} and MASK in used["detail"] async def test_a_run_past_its_size_limit_keeps_steps_without_values(client, auth_headers, monkeypatch): monkeypatch.setattr(task_history, "MAX_RUN_CHARS", 300) nodes = [_node("start", "start", "a"), _set("a", "b", v="x" * 500), _set("b", None, w="y")] task_id = await _task(client, auth_headers, nodes) result = await _run(client, auth_headers, task_id) assert result["ok"] is True last = await _step(client, auth_headers, task_id, result["run_id"], 3) assert last["node_key"] == "b" and last["ok"] is True and last["inputs"] is None and last["truncated"] is True async def test_a_history_failure_never_fails_or_stalls_the_run(client, auth_headers, monkeypatch): def broken(): raise OSError("history database unavailable") monkeypatch.setattr(task_history, "_Session", broken) task_id = await _task(client, auth_headers, [_node("start", "start", "a"), _set("a", None, v="1")]) result = await _run(client, auth_headers, task_id) assert result["ok"] is True and result["context"]["v"] == "1" monkeypatch.undo() assert (await client.get(f"/tasks/{task_id}/runs", headers=auth_headers)).json()["items"] == [] assert (await client.get(f"/tasks/{task_id}/runs/{result['run_id']}", headers=auth_headers)).status_code == 404 async def test_steps_are_saved_while_a_long_run_is_still_going(test_tenant, monkeypatch): """A run that dies outright (no chance to finish) still leaves what it did.""" monkeypatch.setattr(task_history, "FLUSH_STEPS", 2) tenant_id = test_tenant["tenant_id"] task_id = await _committed_task(tenant_id, [_node("start", "start", "a"), _set("a", "b", v="1"), _set("b", "c", v="2"), _set("c", "d", v="3"), _set("d", None, v="4")]) seen: list[int] = [] real = task_engine._exec_set_variable trigger = RunTrigger("other") async def spying(db, tenant, node, inputs): async with AsyncSessionLocal() as other: seen.append((await other.execute(text("SELECT count(*) FROM task_run_steps WHERE run_id = :r"), {"r": trigger.run_id})).scalar()) status = (await other.execute(text("SELECT status FROM task_runs WHERE id = :r"), {"r": trigger.run_id})).scalar() if node["key"] == "d": assert status == "running" return await real(db, tenant, node, inputs) monkeypatch.setitem(task_engine._HANDLERS, "set_variable", spying) async with tenant_session(tenant_id) as db: task = (await db.execute(select(Task).where(Task.id == task_id))).scalar_one() ok, _, _ = await run_task(db, tenant_id, task, {}, trigger=trigger) assert ok and seen[-1] >= 2 @pytest.fixture async def second_tenant(): unique = uuid.uuid4().hex[:8] async with AsyncSessionLocal() as db: async with db.begin(): tenant = Tenant(name=f"Pytest Tenant {unique}", status="active", license_set_at=datetime.now(timezone.utc)) db.add(tenant) await db.flush() role = Role(tenant_id=tenant.id, name="Admin", is_system_role=True, action_matrix={}) user = User(tenant_id=tenant.id, name="Other", email=f"pytest-other-{unique}@example.com", password_hash="x", status="active") db.add_all([role, user]) await db.flush() db.add(UserRole(user_id=user.id, role_id=role.id)) ids = tenant.id, user.id, role.id yield ids async with AsyncSessionLocal() as db: async with db.begin(): await db.execute(text("DELETE FROM user_roles WHERE user_id = :u"), {"u": ids[1]}) await db.execute(text("DELETE FROM users WHERE id = :u"), {"u": ids[1]}) await db.execute(text("DELETE FROM roles WHERE id = :r"), {"r": ids[2]}) await db.execute(text("DELETE FROM tenants WHERE id = :t"), {"t": ids[0]}) async def test_tenants_cant_see_each_others_history(client, auth_headers, test_tenant, second_tenant): task_id = await _task(client, auth_headers, [_node("start", "start", "a"), _set("a", None, v="private")]) run_id = (await _run(client, auth_headers, task_id))["run_id"] other = {"Authorization": f"Bearer {create_access_token(tenant_id=second_tenant[0], user_id=second_tenant[1], vendor_id=None)}"} assert (await client.get(f"/tasks/{task_id}/runs", headers=other)).json()["items"] == [] for path in (f"runs/{run_id}", f"runs/{run_id}/timeline", f"runs/{run_id}/steps", f"runs/{run_id}/steps/1"): assert (await client.get(f"/tasks/{task_id}/{path}", headers=other)).status_code == 404 # The database itself, with no tenant filter in the query. async with engine.connect() as conn: async with conn.begin(): await conn.execute(text("SELECT set_config('app.current_tenant_id', :v, true)"), {"v": str(second_tenant[0])}) assert (await conn.execute(text("SELECT count(*) FROM task_runs"))).scalar() == 0 assert (await conn.execute(text("SELECT count(*) FROM task_run_steps"))).scalar() == 0 async with conn.begin(): await conn.execute(text("SELECT set_config('app.current_tenant_id', :v, true)"), {"v": str(test_tenant["tenant_id"])}) assert (await conn.execute(text("SELECT count(*) FROM task_runs"))).scalar() == 1 async def test_only_people_who_may_view_tasks_can_read_history(client, auth_headers, test_tenant): task_id = await _task(client, auth_headers, [_node("start", "start")]) run_id = (await _run(client, auth_headers, task_id))["run_id"] paths = ("runs", f"runs/{run_id}", f"runs/{run_id}/timeline", f"runs/{run_id}/steps", f"runs/{run_id}/steps/1") vendor = {"Authorization": f"Bearer {create_access_token(tenant_id=test_tenant['tenant_id'], user_id=test_tenant['admin_id'], vendor_id=uuid.uuid4())}"} for path in paths: assert (await client.get(f"/tasks/{task_id}/{path}", headers=vendor)).status_code in (401, 403), path unique = uuid.uuid4().hex[:8] async with AsyncSessionLocal() as db: async with db.begin(): role = Role(tenant_id=test_tenant["tenant_id"], name=f"No tasks {unique}", is_system_role=False, action_matrix={"masters": ["view"]}) viewer_role = Role(tenant_id=test_tenant["tenant_id"], name=f"Task viewer {unique}", is_system_role=False, action_matrix={"tasks": ["view"]}) none = User(tenant_id=test_tenant["tenant_id"], name="No tasks", email=f"pytest-none-{unique}@example.com", password_hash="x", status="active") viewer = User(tenant_id=test_tenant["tenant_id"], name="Viewer", email=f"pytest-viewer-{unique}@example.com", password_hash="x", status="active") db.add_all([role, viewer_role, none, viewer]) await db.flush() db.add_all([UserRole(user_id=none.id, role_id=role.id), UserRole(user_id=viewer.id, role_id=viewer_role.id)]) none_id, viewer_id, role_ids = none.id, viewer.id, (role.id, viewer_role.id) try: denied = {"Authorization": f"Bearer {create_access_token(tenant_id=test_tenant['tenant_id'], user_id=none_id, vendor_id=None)}"} allowed = {"Authorization": f"Bearer {create_access_token(tenant_id=test_tenant['tenant_id'], user_id=viewer_id, vendor_id=None)}"} for path in paths: assert (await client.get(f"/tasks/{task_id}/{path}", headers=denied)).status_code == 403, path assert (await client.get(f"/tasks/{task_id}/{path}", headers=allowed)).status_code == 200, path # Viewing doesn't extend to changing how long history is kept. assert (await client.put("/tasks/history-settings", headers=allowed, json={"retention_days": 1})).status_code == 403 finally: async with AsyncSessionLocal() as db: async with db.begin(): await db.execute(text("DELETE FROM user_roles WHERE user_id = ANY(:u)"), {"u": [none_id, viewer_id]}) await db.execute(text("DELETE FROM users WHERE id = ANY(:u)"), {"u": [none_id, viewer_id]}) await db.execute(text("DELETE FROM roles WHERE id = ANY(:r)"), {"r": list(role_ids)}) async def _old_run(tenant_id, task_id, *, age_days=0, status="succeeded", quiet_hours=0): now = datetime.now(timezone.utc) run_id = uuid.uuid4() async with AsyncSessionLocal() as db: async with db.begin(): db.add( TaskRun( id=run_id, tenant_id=tenant_id, task_id=task_id, task_name="Old", task_status="published", trigger="other", status=status, started_at=now - timedelta(days=age_days), updated_at=now - timedelta(hours=quiet_hours), ) ) await db.flush() db.add(TaskRunStep(run_id=run_id, seq=1, tenant_id=tenant_id, node_key="start", node_type="start", ok=True, detail="started", started_at=now)) return run_id async def test_retention_removes_old_runs_and_their_steps(client, auth_headers, test_tenant): tenant_id, task_id = test_tenant["tenant_id"], uuid.uuid4() settings = await client.get("/tasks/history-settings", headers=auth_headers) assert settings.json()["retention_days"] == 30 updated = await client.put("/tasks/history-settings", headers=auth_headers, json={"retention_days": 7}) assert updated.status_code == 200 and updated.json()["retention_days"] == 7 assert (await client.put("/tasks/history-settings", headers=auth_headers, json={"retention_days": 0})).status_code == 422 old = await _old_run(tenant_id, task_id, age_days=8) kept = await _old_run(tenant_id, task_id, age_days=6) dead = await _old_run(tenant_id, task_id, age_days=2, status="running", quiet_hours=2) live = await _old_run(tenant_id, task_id, status="running") await task_history.queue_cleanup() await task_history.queue_cleanup() # waiting already: still one job async with AsyncSessionLocal() as db: async with db.begin(): queued = (await db.execute(text("SELECT count(*) FROM jobs WHERE dedupe_key = :k AND status = 'queued' AND queue = 'pytest'"), {"k": f"{task_history.CLEANUP_JOB}:pytest"})).scalar() assert queued == 1 await db.execute(text("UPDATE jobs SET run_after = now() WHERE dedupe_key = :k AND status = 'queued' AND queue = 'pytest'"), {"k": f"{task_history.CLEANUP_JOB}:pytest"}) assert await jobs.run_due([task_history.CLEANUP_JOB]) == 1 async with AsyncSessionLocal() as db: rows = dict((await db.execute(select(TaskRun.id, TaskRun.status).where(TaskRun.tenant_id == tenant_id))).all()) steps = (await db.execute(select(TaskRunStep.run_id).where(TaskRunStep.tenant_id == tenant_id))).scalars().all() assert old not in rows and kept in rows and live in rows assert rows[dead] == "interrupted" and rows[live] == "running" assert old not in steps and kept in steps async def test_the_cap_per_task_keeps_the_newest(test_tenant): tenant_id, task_id = test_tenant["tenant_id"], uuid.uuid4() runs = [await _old_run(tenant_id, task_id, age_days=d) for d in (1, 2, 3, 4)] other_task = await _old_run(tenant_id, uuid.uuid4(), age_days=5) assert await task_history.clean_tenant(tenant_id, 30, cap=2) == 2 async with AsyncSessionLocal() as db: left = set((await db.execute(select(TaskRun.id).where(TaskRun.tenant_id == tenant_id))).scalars().all()) assert left == {runs[0], runs[1], other_task}