103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""Tests for move_task / set_project_goal — event-sourced PM ops.
|
|
|
|
Both mutations must go through events (`TaskUpdated.project_id` /
|
|
`ProjectUpdated.goal`) so a `replay()` reconstructs the new state; a direct
|
|
projection UPDATE would be lost. These tests assert the projection changed
|
|
AND an event was logged.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from claire.hlc import HLCGenerator
|
|
from claire.web import service
|
|
|
|
|
|
def _event_count(conn: sqlite3.Connection) -> int:
|
|
return conn.execute("SELECT COUNT(*) FROM events").fetchone()[0]
|
|
|
|
|
|
def _task_project_id(conn: sqlite3.Connection, task_id: str) -> str:
|
|
return conn.execute(
|
|
"SELECT project_id FROM tasks WHERE id = ?", (task_id,)
|
|
).fetchone()["project_id"]
|
|
|
|
|
|
def test_move_task_by_project_name(conn: sqlite3.Connection, gen: HLCGenerator) -> None:
|
|
src = service.create_project(conn, gen, name="alpha")
|
|
dst = service.create_project(conn, gen, name="beta")
|
|
task = service.add_task(conn, gen, project="alpha", title="ship it")
|
|
assert _task_project_id(conn, str(task.id)) == str(src.id)
|
|
before = _event_count(conn)
|
|
|
|
moved = service.move_task(conn, gen, task_ref=str(task.id), project="beta")
|
|
|
|
assert moved.id == task.id
|
|
assert moved.project_id == dst.id
|
|
assert _event_count(conn) == before + 1
|
|
assert _task_project_id(conn, str(task.id)) == str(dst.id)
|
|
|
|
|
|
def test_move_task_by_project_id(conn: sqlite3.Connection, gen: HLCGenerator) -> None:
|
|
service.create_project(conn, gen, name="alpha")
|
|
dst = service.create_project(conn, gen, name="beta")
|
|
task = service.add_task(conn, gen, project="alpha", title="ship it")
|
|
before = _event_count(conn)
|
|
|
|
moved = service.move_task(conn, gen, task_ref=str(task.id), project=str(dst.id))
|
|
|
|
assert moved.project_id == dst.id
|
|
assert _event_count(conn) == before + 1
|
|
assert _task_project_id(conn, str(task.id)) == str(dst.id)
|
|
|
|
|
|
def test_move_task_unknown_task(conn: sqlite3.Connection, gen: HLCGenerator) -> None:
|
|
service.create_project(conn, gen, name="beta")
|
|
with pytest.raises(service.NotFound):
|
|
service.move_task(
|
|
conn, gen,
|
|
task_ref="00000000-0000-0000-0000-000000000000",
|
|
project="beta",
|
|
)
|
|
|
|
|
|
def test_move_task_unknown_project(conn: sqlite3.Connection, gen: HLCGenerator) -> None:
|
|
service.create_project(conn, gen, name="alpha")
|
|
task = service.add_task(conn, gen, project="alpha", title="ship it")
|
|
with pytest.raises(service.NotFound):
|
|
service.move_task(conn, gen, task_ref=str(task.id), project="nope")
|
|
|
|
|
|
def test_set_project_goal(conn: sqlite3.Connection, gen: HLCGenerator) -> None:
|
|
proj = service.create_project(conn, gen, name="alpha")
|
|
before = _event_count(conn)
|
|
|
|
updated = service.set_project_goal(
|
|
conn, gen, name_or_id="alpha", goal="ship the orchestrator"
|
|
)
|
|
|
|
assert updated.id == proj.id
|
|
assert updated.goal == "ship the orchestrator"
|
|
assert _event_count(conn) == before + 1
|
|
row = conn.execute(
|
|
"SELECT goal FROM projects WHERE id = ?", (str(proj.id),)
|
|
).fetchone()
|
|
assert row["goal"] == "ship the orchestrator"
|
|
|
|
|
|
def test_set_project_goal_strips(conn: sqlite3.Connection, gen: HLCGenerator) -> None:
|
|
service.create_project(conn, gen, name="alpha")
|
|
|
|
updated = service.set_project_goal(
|
|
conn, gen, name_or_id="alpha", goal=" trimmed goal "
|
|
)
|
|
|
|
assert updated.goal == "trimmed goal"
|
|
|
|
|
|
def test_set_project_goal_unknown(conn: sqlite3.Connection, gen: HLCGenerator) -> None:
|
|
with pytest.raises(service.NotFound):
|
|
service.set_project_goal(conn, gen, name_or_id="nope", goal="x")
|