144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
"""Service-layer + replay tests for the Decisions log (migration 0009)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from claire.db import migrate, open_db
|
|
from claire.domain import DecisionMaker
|
|
from claire.events import replay
|
|
from claire.hlc import HLCGenerator
|
|
from claire.web import service
|
|
|
|
|
|
def _setup() -> tuple:
|
|
conn = open_db(":memory:")
|
|
migrate(conn)
|
|
gen = HLCGenerator("test-machine")
|
|
return conn, gen
|
|
|
|
|
|
def test_record_decision_claire_minimal() -> None:
|
|
conn, gen = _setup()
|
|
d = service.record_decision(
|
|
conn, gen,
|
|
made_by=DecisionMaker.CLAIRE,
|
|
text="Use a 2-gate review workflow.",
|
|
)
|
|
assert d.made_by is DecisionMaker.CLAIRE
|
|
assert d.text == "Use a 2-gate review workflow."
|
|
assert d.rationale is None
|
|
assert d.project_id is None
|
|
assert d.task_id is None
|
|
assert d.created_hlc
|
|
|
|
|
|
def test_record_decision_user_with_rationale_and_links() -> None:
|
|
conn, gen = _setup()
|
|
proj = service.create_project(conn, gen, name="p")
|
|
task = service.add_task(conn, gen, project="p", title="t")
|
|
d = service.record_decision(
|
|
conn, gen,
|
|
made_by=DecisionMaker.USER,
|
|
text="Defer mobile to phase 2.",
|
|
rationale="Web app is unblocking more users right now.",
|
|
project="p",
|
|
task_ref=str(task.id),
|
|
)
|
|
assert d.made_by is DecisionMaker.USER
|
|
assert d.rationale == "Web app is unblocking more users right now."
|
|
assert d.project_id == proj.id
|
|
assert d.task_id == task.id
|
|
|
|
|
|
def test_record_decision_strips_and_rejects_empty_text() -> None:
|
|
conn, gen = _setup()
|
|
with pytest.raises(service.InvalidInput):
|
|
service.record_decision(
|
|
conn, gen, made_by=DecisionMaker.CLAIRE, text=" ",
|
|
)
|
|
with pytest.raises(service.InvalidInput):
|
|
service.record_decision(
|
|
conn, gen, made_by=DecisionMaker.CLAIRE, text="",
|
|
)
|
|
|
|
|
|
def test_record_decision_unknown_project_raises() -> None:
|
|
conn, gen = _setup()
|
|
with pytest.raises(service.NotFound):
|
|
service.record_decision(
|
|
conn, gen,
|
|
made_by=DecisionMaker.CLAIRE, text="x",
|
|
project="no-such-project",
|
|
)
|
|
|
|
|
|
def test_record_decision_unknown_task_raises() -> None:
|
|
conn, gen = _setup()
|
|
with pytest.raises(service.NotFound):
|
|
service.record_decision(
|
|
conn, gen,
|
|
made_by=DecisionMaker.CLAIRE, text="x",
|
|
task_ref="00000000-0000-0000-0000-000000000000",
|
|
)
|
|
|
|
|
|
def test_list_decisions_filters() -> None:
|
|
conn, gen = _setup()
|
|
service.create_project(conn, gen, name="pa")
|
|
service.create_project(conn, gen, name="pb")
|
|
task_a = service.add_task(conn, gen, project="pa", title="ta")
|
|
task_b = service.add_task(conn, gen, project="pb", title="tb")
|
|
|
|
d1 = service.record_decision(
|
|
conn, gen, made_by=DecisionMaker.CLAIRE,
|
|
text="claire on pa", project="pa",
|
|
)
|
|
d2 = service.record_decision(
|
|
conn, gen, made_by=DecisionMaker.USER,
|
|
text="user on pa task", project="pa", task_ref=str(task_a.id),
|
|
)
|
|
d3 = service.record_decision(
|
|
conn, gen, made_by=DecisionMaker.CLAIRE,
|
|
text="claire on pb task", project="pb", task_ref=str(task_b.id),
|
|
)
|
|
|
|
# All — newest first.
|
|
all_ = service.list_decisions(conn)
|
|
assert [d.id for d in all_] == [d3.id, d2.id, d1.id]
|
|
|
|
# Filter by project name → id resolution.
|
|
proj_a = service.read.get_project(conn, "pa")
|
|
assert proj_a is not None
|
|
by_a = service.list_decisions(conn, project_id=proj_a.id)
|
|
assert {d.id for d in by_a} == {d1.id, d2.id}
|
|
|
|
# Filter by task.
|
|
by_task = service.list_decisions(conn, task_id=task_a.id)
|
|
assert [d.id for d in by_task] == [d2.id]
|
|
|
|
# Filter by made_by.
|
|
by_user = service.list_decisions(conn, made_by=DecisionMaker.USER)
|
|
assert [d.id for d in by_user] == [d2.id]
|
|
by_claire = service.list_decisions(conn, made_by=DecisionMaker.CLAIRE)
|
|
assert [d.id for d in by_claire] == [d3.id, d1.id]
|
|
|
|
|
|
def test_decision_survives_replay() -> None:
|
|
conn, gen = _setup()
|
|
service.create_project(conn, gen, name="p")
|
|
task = service.add_task(conn, gen, project="p", title="t")
|
|
d = service.record_decision(
|
|
conn, gen, made_by=DecisionMaker.CLAIRE,
|
|
text="design call", rationale="why",
|
|
project="p", task_ref=str(task.id),
|
|
)
|
|
# Wipe + replay event log; the decision row must come back identical.
|
|
replay(conn)
|
|
fresh = service.read.get_decision(conn, d.id)
|
|
assert fresh is not None
|
|
assert fresh.id == d.id
|
|
assert fresh.made_by is DecisionMaker.CLAIRE
|
|
assert fresh.text == "design call"
|
|
assert fresh.rationale == "why"
|
|
assert fresh.task_id == task.id
|