Kairo-Jinkendo/backend/tests/test_a1_progression_join.py

"""AP-A1-PM — progression join graph tests."""

from __future__ import annotations

from services.progression_graph import compute_progression_lanes
from tests.factories import provision_user_in_tenant
from tests.test_initiatives_actions import _auth, _create_initiative, _login


def _item(item_id: str, item_type: str = "maturity_stage", **kwargs):
    return {
        "id": item_id,
        "item_type": item_type,
        "status": kwargs.get("status", "planned"),
        "title": kwargs.get("title", item_id),
        "sort_order": kwargs.get("sort_order", 0),
    }


def _requires(from_id: str, to_id: str) -> dict:
    return {
        "from_item_id": from_id,
        "to_item_id": to_id,
        "dependency_type": "requires",
    }


def test_progression_lanes_parallel_sources_and_post_join():
    items = [
        _item("d1", status="active", title="Dehnung — Einstieg"),
        _item("h1", status="active", title="Hüftrotation — Einstieg"),
        _item("join", item_type="join_gate", title="Join"),
        _item("d2", title="Dehnung — nach Join"),
        _item("h2", title="Hüftrotation — nach Join"),
    ]
    deps = [
        _requires("join", "d1"),
        _requires("join", "h1"),
        _requires("d2", "join"),
        _requires("h2", "join"),
    ]
    lanes = compute_progression_lanes(items=items, dependencies=deps)
    assert lanes["d1"].startswith("source:")
    assert lanes["h1"].startswith("source:")
    assert lanes["h1"] != lanes["d1"]
    assert lanes["d2"] == "post-join:d2"
    assert lanes["h2"] == "post-join:h2"
    assert lanes["d2"] != lanes["h2"]


def test_progression_lanes_sequential_chain_shares_lane():
    items = [
        _item("g1", status="active"),
        _item("g2"),
        _item("g3"),
    ]
    deps = [_requires("g2", "g1"), _requires("g3", "g2")]
    lanes = compute_progression_lanes(items=items, dependencies=deps)
    assert lanes["g1"] == lanes["g2"] == lanes["g3"]


def test_progression_join_switches_and_activates_outputs(client):
    user = provision_user_in_tenant(tenant_role="admin")
    token = _login(client, user)

    created = _create_initiative(
        client,
        token,
        title="Mawashi Progression",
        archetype_key="initiative.maturity_journey",
    )
    initiative_id = created.json()["id"]

    from services.archetype_starter_kit import apply_a1_progression_demo

    demo = apply_a1_progression_demo(
        tenant_id=user["tenant_id"],
        initiative_id=initiative_id,
        user_id=user["id"],
    )
    assert demo["applied"] is True

    progression = client.get(
        f"/api/initiatives/{initiative_id}/roadmap/progression-state",
        headers=_auth(token),
    )
    assert progression.status_code == 200
    body = progression.json()
    assert len(body["active_work_gates"]) == 2
    assert len(body["joins"]) == 1
    assert body["joins"][0]["pending_inputs"]

    items = client.get(
        f"/api/initiatives/{initiative_id}/roadmap/items",
        headers=_auth(token),
    ).json()
    by_title = {item["title"]: item for item in items}
    d1 = by_title["Dehnung — Einstieg"]
    h1 = by_title["Hüftrotation — Einstieg"]
    join = by_title["Mawashi Geri zur Hüfte"]

    for gate in (d1, h1):
        client.post(
            f"/api/initiatives/{initiative_id}/evidence",
            json={
                "title": f"Nachweis {gate['title']}",
                "roadmap_item_id": gate["id"],
                "status": "accepted",
            },
            headers=_auth(token),
        )
        verify = client.post(
            f"/api/roadmap-items/{gate['id']}/verify-reached",
            headers=_auth(token),
        )
        assert verify.status_code == 200

    items_after = client.get(
        f"/api/initiatives/{initiative_id}/roadmap/items",
        headers=_auth(token),
    ).json()
    by_title_after = {item["title"]: item for item in items_after}
    assert by_title_after[join["title"]]["status"] == "reached"
    assert by_title_after["Dehnung — nach Join"]["status"] == "active"
    assert by_title_after["Hüftrotation — nach Join"]["status"] == "active"

    recurring = client.get(
        f"/api/initiatives/{initiative_id}/recurring",
        headers=_auth(token),
    ).json()
    active_titles = {r["title"] for r in recurring if r["status"] == "active"}
    assert "Enddehnung Mawashi-Vorbereitung" in active_titles
    assert "Explosive Hüftrotation" in active_titles


def test_multi_active_work_gates_both_in_next_actions(client):
    user = provision_user_in_tenant(tenant_role="admin")
    token = _login(client, user)

    created = _create_initiative(
        client,
        token,
        title="Multi Active",
        archetype_key="initiative.maturity_journey",
    )
    initiative_id = created.json()["id"]

    from services.archetype_starter_kit import apply_a1_progression_demo

    apply_a1_progression_demo(
        tenant_id=user["tenant_id"],
        initiative_id=initiative_id,
        user_id=user["id"],
    )

    snap = client.get(
        f"/api/initiatives/{initiative_id}/steering-snapshot",
        headers=_auth(token),
    )
    assert snap.status_code == 200
    recurring_items = [
        a for a in snap.json().get("next_actions") or [] if a.get("kind") == "recurring_due"
    ]
    assert len(recurring_items) >= 2