126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
|
"""Unit tests — Epic roll-up read model (P4)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from steering.read_models.epic_rollup import (
|
||
|
|
compute_epic_rollup,
|
||
|
|
epic_rollup_attention_items,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _item(
|
||
|
|
*,
|
||
|
|
id: str,
|
||
|
|
title: str,
|
||
|
|
item_kind: str = "story",
|
||
|
|
status: str = "accepted",
|
||
|
|
parent_backlog_id: str | None = None,
|
||
|
|
converted_action_id: str | None = None,
|
||
|
|
) -> dict:
|
||
|
|
return {
|
||
|
|
"id": id,
|
||
|
|
"title": title,
|
||
|
|
"item_kind": item_kind,
|
||
|
|
"status": status,
|
||
|
|
"parent_backlog_id": parent_backlog_id,
|
||
|
|
"converted_action_id": converted_action_id,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_epic_rollup_empty_epic():
|
||
|
|
rollups = compute_epic_rollup(
|
||
|
|
backlog_items=[_item(id="e1", title="Epic", item_kind="epic")],
|
||
|
|
actions=[],
|
||
|
|
)
|
||
|
|
assert len(rollups) == 1
|
||
|
|
assert rollups[0]["status"] == "empty"
|
||
|
|
assert rollups[0]["child_total"] == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_epic_rollup_intake_only():
|
||
|
|
rollups = compute_epic_rollup(
|
||
|
|
backlog_items=[
|
||
|
|
_item(id="e1", title="Checkout", item_kind="epic"),
|
||
|
|
_item(id="s1", title="Login", parent_backlog_id="e1"),
|
||
|
|
_item(id="s2", title="Pay", parent_backlog_id="e1"),
|
||
|
|
],
|
||
|
|
actions=[],
|
||
|
|
)
|
||
|
|
assert rollups[0]["status"] == "intake_only"
|
||
|
|
assert rollups[0]["child_total"] == 2
|
||
|
|
assert rollups[0]["child_uncommitted"] == 2
|
||
|
|
assert rollups[0]["completion_ratio"] == 0.0
|
||
|
|
|
||
|
|
|
||
|
|
def test_epic_rollup_partial_and_done():
|
||
|
|
rollups = compute_epic_rollup(
|
||
|
|
backlog_items=[
|
||
|
|
_item(id="e1", title="Epic", item_kind="epic"),
|
||
|
|
_item(
|
||
|
|
id="s1",
|
||
|
|
title="Done story",
|
||
|
|
parent_backlog_id="e1",
|
||
|
|
status="converted",
|
||
|
|
converted_action_id="a1",
|
||
|
|
),
|
||
|
|
_item(id="s2", title="Open story", parent_backlog_id="e1"),
|
||
|
|
],
|
||
|
|
actions=[
|
||
|
|
{"id": "a1", "status": "done"},
|
||
|
|
{"id": "a2", "status": "in_progress"},
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert rollups[0]["status"] == "in_progress"
|
||
|
|
assert rollups[0]["actions_done"] == 1
|
||
|
|
assert rollups[0]["child_uncommitted"] == 1
|
||
|
|
assert rollups[0]["completion_ratio"] == 0.5
|
||
|
|
|
||
|
|
|
||
|
|
def test_epic_rollup_all_done():
|
||
|
|
rollups = compute_epic_rollup(
|
||
|
|
backlog_items=[
|
||
|
|
_item(id="e1", title="Epic", item_kind="epic"),
|
||
|
|
_item(
|
||
|
|
id="s1",
|
||
|
|
title="A",
|
||
|
|
parent_backlog_id="e1",
|
||
|
|
status="converted",
|
||
|
|
converted_action_id="a1",
|
||
|
|
),
|
||
|
|
],
|
||
|
|
actions=[{"id": "a1", "status": "done"}],
|
||
|
|
)
|
||
|
|
assert rollups[0]["status"] == "done"
|
||
|
|
assert rollups[0]["completion_ratio"] == 1.0
|
||
|
|
|
||
|
|
|
||
|
|
def test_epic_rollup_attention_awaiting_commit():
|
||
|
|
rollups = compute_epic_rollup(
|
||
|
|
backlog_items=[
|
||
|
|
_item(id="e1", title="Epic", item_kind="epic"),
|
||
|
|
_item(id="s1", title="Story", parent_backlog_id="e1"),
|
||
|
|
],
|
||
|
|
actions=[],
|
||
|
|
)
|
||
|
|
items = epic_rollup_attention_items(initiative_id="init-1", rollups=rollups)
|
||
|
|
assert any(i["reason_code"] == "epic_awaiting_commit" for i in items)
|
||
|
|
|
||
|
|
|
||
|
|
def test_epic_rollup_attention_split_focus():
|
||
|
|
rollups = compute_epic_rollup(
|
||
|
|
backlog_items=[
|
||
|
|
_item(id="e1", title="Epic", item_kind="epic"),
|
||
|
|
_item(
|
||
|
|
id="s1",
|
||
|
|
title="Running",
|
||
|
|
parent_backlog_id="e1",
|
||
|
|
status="converted",
|
||
|
|
converted_action_id="a1",
|
||
|
|
),
|
||
|
|
_item(id="s2", title="Waiting", parent_backlog_id="e1"),
|
||
|
|
],
|
||
|
|
actions=[{"id": "a1", "status": "in_progress"}],
|
||
|
|
)
|
||
|
|
items = epic_rollup_attention_items(initiative_id="init-1", rollups=rollups)
|
||
|
|
assert any(i["reason_code"] == "epic_uncommitted_during_execution" for i in items)
|