Lars
e25ff86420
Some checks failed
Deploy Development / deploy (push) Successful in 45s
Test Suite / pytest-backend (push) Failing after 1m10s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 1s
Test Suite / compose-smoke (push) Has been skipped
Beantwortet die Lücke zwischen parallelen Tabellen und späterem Steering Core: Graph-Read-Model, minimale Flows, sichtbarer Steuerungszustand. Co-authored-by: Cursor <cursoragent@cursor.com>
91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""AP0.10 — Integration: Snapshot, Transitions, Attention."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from tests.factories import provision_user_in_tenant
|
|
from tests.test_initiatives_actions import (
|
|
_auth,
|
|
_create_action,
|
|
_create_initiative,
|
|
_login,
|
|
)
|
|
|
|
|
|
def test_steering_snapshot_graph(client):
|
|
user = provision_user_in_tenant(tenant_role="member")
|
|
token = _login(client, user)
|
|
initiative_id = _create_initiative(client, token).json()["id"]
|
|
action = _create_action(client, token, initiative_id, title="Linked Action").json()
|
|
|
|
blocker = client.post(
|
|
f"/api/initiatives/{initiative_id}/blockers",
|
|
json={"title": "B", "action_id": action["id"]},
|
|
headers=_auth(token),
|
|
)
|
|
assert blocker.status_code == 201
|
|
|
|
snap = client.get(
|
|
f"/api/initiatives/{initiative_id}/steering-snapshot",
|
|
headers=_auth(token),
|
|
)
|
|
assert snap.status_code == 200
|
|
data = snap.json()
|
|
assert data["operating_phase"] in (
|
|
"execute",
|
|
"triage",
|
|
"structure",
|
|
"review",
|
|
"adapt",
|
|
"commit",
|
|
"capture",
|
|
"verify",
|
|
"closure",
|
|
)
|
|
linked = next(a for a in data["actions"] if a["id"] == action["id"])
|
|
assert len(linked["blockers"]) == 1
|
|
assert linked["open_blocker_count"] == 1
|
|
|
|
|
|
def test_blocker_resolve_unblocks_action(client):
|
|
user = provision_user_in_tenant(tenant_role="member")
|
|
token = _login(client, user)
|
|
initiative_id = _create_initiative(client, token).json()["id"]
|
|
action = _create_action(client, token, initiative_id).json()["id"]
|
|
|
|
created = client.post(
|
|
f"/api/initiatives/{initiative_id}/blockers",
|
|
json={
|
|
"title": "Block",
|
|
"action_id": action,
|
|
"set_action_blocked": True,
|
|
},
|
|
headers=_auth(token),
|
|
)
|
|
blocker_id = created.json()["id"]
|
|
blocked = client.get(f"/api/actions/{action}", headers=_auth(token)).json()
|
|
assert blocked["status"] == "blocked"
|
|
|
|
client.patch(
|
|
f"/api/blockers/{blocker_id}",
|
|
json={"status": "resolved"},
|
|
headers=_auth(token),
|
|
)
|
|
unblocked = client.get(f"/api/actions/{action}", headers=_auth(token)).json()
|
|
assert unblocked["status"] == "open"
|
|
|
|
|
|
def test_attention_action_review_required_without_review(client):
|
|
user = provision_user_in_tenant(tenant_role="member")
|
|
token = _login(client, user)
|
|
initiative_id = _create_initiative(client, token).json()["id"]
|
|
action = _create_action(client, token, initiative_id).json()
|
|
client.patch(
|
|
f"/api/actions/{action['id']}",
|
|
json={"status": "review_required"},
|
|
headers=_auth(token),
|
|
)
|
|
|
|
res = client.get("/api/workspace/attention", headers=_auth(token))
|
|
assert res.status_code == 200
|
|
assert any(i["kind"] == "action_review_required" for i in res.json())
|