Lars
afa68150c0
Some checks failed
Deploy Development / deploy (push) Successful in 33s
Test Suite / pytest-backend (push) Failing after 28s
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 7s
Test Suite / compose-smoke (push) Has been skipped
317 lines
9.9 KiB
Python
317 lines
9.9 KiB
Python
"""Initiatives and actions tests (AP0.5)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from auth import AUTH_HEADER
|
|
from db import get_connection
|
|
from services.actors import create_actor
|
|
from tests.factories import provision_user_in_tenant
|
|
|
|
|
|
def _login(client, user: dict) -> str:
|
|
res = client.post(
|
|
"/api/auth/login",
|
|
json={"email": user["email"], "password": user["password"]},
|
|
)
|
|
assert res.status_code == 200
|
|
return res.json()["token"]
|
|
|
|
|
|
def _auth(token: str) -> dict:
|
|
return {AUTH_HEADER: token}
|
|
|
|
|
|
def _create_initiative(client, token: str, *, title: str = "Test Vorhaben", **kwargs):
|
|
body = {"title": title, **kwargs}
|
|
return client.post("/api/initiatives", json=body, headers=_auth(token))
|
|
|
|
|
|
def _create_action(client, token: str, initiative_id: str, *, title: str = "Test Maßnahme", **kwargs):
|
|
body = {"title": title, **kwargs}
|
|
return client.post(
|
|
f"/api/initiatives/{initiative_id}/actions",
|
|
json=body,
|
|
headers=_auth(token),
|
|
)
|
|
|
|
|
|
def test_initiative_crud(client):
|
|
user = provision_user_in_tenant(tenant_role="member")
|
|
token = _login(client, user)
|
|
|
|
created = _create_initiative(
|
|
client,
|
|
token,
|
|
title="Kairo MVP",
|
|
goal="Erster Slice",
|
|
priority="high",
|
|
)
|
|
assert created.status_code == 201
|
|
initiative = created.json()
|
|
assert initiative["title"] == "Kairo MVP"
|
|
assert initiative["goal"] == "Erster Slice"
|
|
assert initiative["status"] == "active"
|
|
assert initiative["priority"] == "high"
|
|
assert initiative["owner_actor_id"] == user["actor_id"]
|
|
assert initiative["tenant_id"] == user["tenant_id"]
|
|
|
|
listed = client.get("/api/initiatives", headers=_auth(token))
|
|
assert listed.status_code == 200
|
|
assert any(i["id"] == initiative["id"] for i in listed.json())
|
|
|
|
fetched = client.get(f"/api/initiatives/{initiative['id']}", headers=_auth(token))
|
|
assert fetched.status_code == 200
|
|
|
|
updated = client.patch(
|
|
f"/api/initiatives/{initiative['id']}",
|
|
json={"status": "paused", "goal": "Angepasst"},
|
|
headers=_auth(token),
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["status"] == "paused"
|
|
assert updated.json()["goal"] == "Angepasst"
|
|
|
|
deleted = client.delete(f"/api/initiatives/{initiative['id']}", headers=_auth(token))
|
|
assert deleted.status_code == 204
|
|
assert client.get(f"/api/initiatives/{initiative['id']}", headers=_auth(token)).status_code == 404
|
|
|
|
|
|
def test_action_crud_and_status_change(client):
|
|
user = provision_user_in_tenant(tenant_role="member")
|
|
token = _login(client, user)
|
|
initiative_id = _create_initiative(client, token).json()["id"]
|
|
|
|
created = _create_action(
|
|
client,
|
|
token,
|
|
initiative_id,
|
|
title="Status prüfen",
|
|
assigned_actor_ids=[user["actor_id"]],
|
|
)
|
|
assert created.status_code == 201
|
|
action = created.json()
|
|
assert action["status"] == "open"
|
|
assert action["assigned_actor_ids"] == [user["actor_id"]]
|
|
|
|
listed = client.get(f"/api/initiatives/{initiative_id}/actions", headers=_auth(token))
|
|
assert listed.status_code == 200
|
|
assert len(listed.json()) == 1
|
|
|
|
updated = client.patch(
|
|
f"/api/actions/{action['id']}",
|
|
json={"status": "in_progress"},
|
|
headers=_auth(token),
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["status"] == "in_progress"
|
|
|
|
done = client.patch(
|
|
f"/api/actions/{action['id']}",
|
|
json={"status": "done"},
|
|
headers=_auth(token),
|
|
)
|
|
assert done.status_code == 200
|
|
assert done.json()["status"] == "done"
|
|
|
|
|
|
def test_action_assignments(client):
|
|
user = provision_user_in_tenant(tenant_role="admin")
|
|
token = _login(client, user)
|
|
agent = create_actor(tenant_id=user["tenant_id"], actor_type="agent", name="Helper Agent")
|
|
initiative_id = _create_initiative(client, token).json()["id"]
|
|
action_id = _create_action(client, token, initiative_id).json()["id"]
|
|
|
|
assigned = client.put(
|
|
f"/api/actions/{action_id}/assignments",
|
|
json={"actor_ids": [user["actor_id"], agent["id"]]},
|
|
headers=_auth(token),
|
|
)
|
|
assert assigned.status_code == 200
|
|
body = assigned.json()
|
|
assert set(body["assigned_actor_ids"]) == {user["actor_id"], agent["id"]}
|
|
|
|
|
|
def test_my_open_actions(client):
|
|
user = provision_user_in_tenant(tenant_role="member")
|
|
token = _login(client, user)
|
|
initiative_id = _create_initiative(client, token).json()["id"]
|
|
|
|
open_action = _create_action(
|
|
client,
|
|
token,
|
|
initiative_id,
|
|
title="Offen",
|
|
assigned_actor_ids=[user["actor_id"]],
|
|
).json()
|
|
in_progress = _create_action(
|
|
client,
|
|
token,
|
|
initiative_id,
|
|
title="In Arbeit",
|
|
assigned_actor_ids=[user["actor_id"]],
|
|
).json()
|
|
client.patch(
|
|
f"/api/actions/{in_progress['id']}",
|
|
json={"status": "in_progress"},
|
|
headers=_auth(token),
|
|
)
|
|
done_action = _create_action(
|
|
client,
|
|
token,
|
|
initiative_id,
|
|
title="Erledigt",
|
|
assigned_actor_ids=[user["actor_id"]],
|
|
).json()
|
|
client.patch(
|
|
f"/api/actions/{done_action['id']}",
|
|
json={"status": "done"},
|
|
headers=_auth(token),
|
|
)
|
|
unassigned = _create_action(
|
|
client,
|
|
token,
|
|
initiative_id,
|
|
title="Nicht zugewiesen",
|
|
).json()
|
|
|
|
open_res = client.get("/api/actions/me/open", headers=_auth(token))
|
|
assert open_res.status_code == 200
|
|
open_ids = {a["id"] for a in open_res.json()}
|
|
assert open_action["id"] in open_ids
|
|
assert in_progress["id"] in open_ids
|
|
assert done_action["id"] not in open_ids
|
|
assert unassigned["id"] not in open_ids
|
|
for item in open_res.json():
|
|
assert "initiative_title" in item
|
|
|
|
|
|
def test_tenant_isolation_initiatives(client):
|
|
user_a = provision_user_in_tenant()
|
|
user_b = provision_user_in_tenant()
|
|
token_a = _login(client, user_a)
|
|
token_b = _login(client, user_b)
|
|
|
|
initiative_id = _create_initiative(client, token_a, title="Tenant A only").json()["id"]
|
|
|
|
assert client.get(f"/api/initiatives/{initiative_id}", headers=_auth(token_b)).status_code == 404
|
|
assert (
|
|
client.patch(
|
|
f"/api/initiatives/{initiative_id}",
|
|
json={"title": "Hack"},
|
|
headers=_auth(token_b),
|
|
).status_code
|
|
== 404
|
|
)
|
|
assert client.get("/api/initiatives", headers=_auth(token_b)).json() == []
|
|
|
|
|
|
def test_tenant_isolation_actions(client):
|
|
user_a = provision_user_in_tenant()
|
|
user_b = provision_user_in_tenant()
|
|
token_a = _login(client, user_a)
|
|
token_b = _login(client, user_b)
|
|
|
|
initiative_id = _create_initiative(client, token_a).json()["id"]
|
|
action_id = _create_action(
|
|
client,
|
|
token_a,
|
|
initiative_id,
|
|
assigned_actor_ids=[user_a["actor_id"]],
|
|
).json()["id"]
|
|
|
|
assert client.get(f"/api/actions/{action_id}", headers=_auth(token_b)).status_code == 404
|
|
assert (
|
|
client.patch(
|
|
f"/api/actions/{action_id}",
|
|
json={"status": "done"},
|
|
headers=_auth(token_b),
|
|
).status_code
|
|
== 404
|
|
)
|
|
assert client.get("/api/actions/me/open", headers=_auth(token_b)).json() == []
|
|
|
|
|
|
def test_cross_tenant_actor_assignment_rejected(client):
|
|
user_a = provision_user_in_tenant()
|
|
user_b = provision_user_in_tenant()
|
|
token_a = _login(client, user_a)
|
|
initiative_id = _create_initiative(client, token_a).json()["id"]
|
|
action_id = _create_action(client, token_a, initiative_id).json()["id"]
|
|
|
|
res = client.put(
|
|
f"/api/actions/{action_id}/assignments",
|
|
json={"actor_ids": [user_b["actor_id"]]},
|
|
headers=_auth(token_a),
|
|
)
|
|
assert res.status_code == 400
|
|
|
|
|
|
def test_audit_on_create_and_status_change(client):
|
|
user = provision_user_in_tenant()
|
|
token = _login(client, user)
|
|
initiative_id = _create_initiative(client, token, title="Audit Test").json()["id"]
|
|
action_id = _create_action(client, token, initiative_id).json()["id"]
|
|
client.patch(
|
|
f"/api/actions/{action_id}",
|
|
json={"status": "blocked"},
|
|
headers=_auth(token),
|
|
)
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT action FROM audit_log
|
|
WHERE tenant_id = %s::uuid
|
|
AND action IN (
|
|
'initiative.created', 'action.created', 'action.status_changed'
|
|
)
|
|
ORDER BY created_at
|
|
""",
|
|
(user["tenant_id"],),
|
|
)
|
|
actions = {row[0] for row in cur.fetchall()}
|
|
finally:
|
|
conn.close()
|
|
|
|
assert "initiative.created" in actions
|
|
assert "action.created" in actions
|
|
assert "action.status_changed" in actions
|
|
|
|
|
|
def test_unauthenticated_rejected(client):
|
|
assert client.get("/api/initiatives").status_code == 401
|
|
assert client.get("/api/actions/me/open").status_code == 401
|
|
|
|
|
|
@pytest.fixture()
|
|
def enforce_capabilities(monkeypatch):
|
|
monkeypatch.setenv("CAPABILITY_ENFORCE", "enforce")
|
|
|
|
|
|
def test_member_has_initiative_capabilities(client):
|
|
member = provision_user_in_tenant(tenant_role="member", portal_role="user")
|
|
token = _login(client, member)
|
|
ctx = client.get("/api/me/context", headers=_auth(token)).json()
|
|
assert "kairo.initiative.read" in ctx["capabilities"]
|
|
assert "kairo.initiative.manage" in ctx["capabilities"]
|
|
assert "kairo.action.read" in ctx["capabilities"]
|
|
assert "kairo.action.manage" in ctx["capabilities"]
|
|
|
|
|
|
def test_invalid_status_rejected(client):
|
|
user = provision_user_in_tenant()
|
|
token = _login(client, user)
|
|
initiative_id = _create_initiative(client, token).json()["id"]
|
|
action_id = _create_action(client, token, initiative_id).json()["id"]
|
|
|
|
res = client.patch(
|
|
f"/api/actions/{action_id}",
|
|
json={"status": "invalid_status"},
|
|
headers=_auth(token),
|
|
)
|
|
assert res.status_code == 422
|