state-hub/tests/test_legacy_meter.py

from __future__ import annotations


async def _create_domain(client, slug="legacy-domain", name="Legacy Domain"):
    r = await client.post("/domains/", json={"slug": slug, "name": name})
    assert r.status_code == 201, r.text
    return r.json()


async def _create_topic(client, domain_slug="legacy-domain", slug="legacy-topic", title="Legacy Topic"):
    r = await client.post("/topics/", json={
        "slug": slug,
        "title": title,
        "domain": domain_slug,
    })
    assert r.status_code == 201, r.text
    return r.json()


async def _create_workplan(client, topic_id, slug="legacy-wp", title="Legacy WP"):
    r = await client.post("/workplans/", json={
        "topic_id": topic_id,
        "slug": slug,
        "title": title,
        "status": "ready",
    })
    assert r.status_code == 201, r.text
    return r.json()


def _summary_by_key(summary):
    return {item["interface"]["interface_key"]: item for item in summary["interfaces"]}


class TestWorkplanAliasesAndLegacyMeter:
    async def test_preferred_workplan_routes_do_not_meter_legacy_usage(self, client):
        await _create_domain(client)
        topic = await _create_topic(client)
        wp = await _create_workplan(client, topic["id"])

        r = await client.get(f"/workplans/?topic_id={topic['id']}")
        assert r.status_code == 200
        assert [row["id"] for row in r.json()] == [wp["id"]]

        r = await client.patch(f"/workplans/{wp['id']}", json={"status": "active"})
        assert r.status_code == 200
        assert r.json()["status"] == "active"

        r = await client.get("/legacy-meter/summary")
        assert r.status_code == 200
        assert r.json()["interfaces"] == []

    async def test_legacy_workstream_route_is_metered_with_identity_buckets(self, client):
        await _create_domain(client)
        topic = await _create_topic(client)
        await _create_workplan(client, topic["id"])

        r = await client.get(
            f"/workstreams/?topic_id={topic['id']}",
            headers={
                "X-StateHub-Tenant": "tenant-a",
                "X-StateHub-User": "alice",
                "X-StateHub-Component": "old-client",
            },
        )
        assert r.status_code == 200
        assert r.headers["Deprecation"] == "true"
        assert r.headers["X-StateHub-Replacement"] == "/workplans/"

        summary = (await client.get("/legacy-meter/summary")).json()
        item = _summary_by_key(summary)["rest_api:GET /workstreams/"]
        assert item["window"]["calls"] == 1
        assert item["window"]["tenants"] == {"tenant-a": 1}
        assert item["window"]["users"] == {"alice": 1}
        assert item["window"]["components"] == {"old-client": 1}
        assert item["retirement_candidate"] is False

    async def test_weekly_review_reports_unused_verified_legacy_interface(self, client):
        r = await client.post("/legacy-meter/interfaces", json={
            "interface_key": "rest_api:GET /obsolete",
            "interface_kind": "rest_api",
            "replacement_ref": "/workplans/",
            "replacement_verified": True,
        })
        assert r.status_code == 201, r.text

        review = (await client.get("/legacy-meter/weekly-review")).json()
        assert review["activity_core_handoff"]["scheduler_owner"] == "activity-core"
        candidates = _summary_by_key({"interfaces": review["retirement_candidates"]})
        assert "rest_api:GET /obsolete" in candidates
        assert candidates["rest_api:GET /obsolete"]["retirement_reason"] == "no measured usage in review window"

    async def test_recent_usage_blocks_weekly_retirement_candidate(self, client):
        r = await client.post("/legacy-meter/usage", json={
            "interface_key": "rest_api:GET /old-but-used",
            "interface_kind": "rest_api",
            "replacement_ref": "/workplans/",
            "replacement_verified": True,
            "tenant_key": "tenant-a",
            "user_key": "alice",
            "component_key": "old-client",
        })
        assert r.status_code == 201, r.text

        review = (await client.get("/legacy-meter/weekly-review")).json()
        items = _summary_by_key(review)
        item = items["rest_api:GET /old-but-used"]
        assert item["window"]["calls"] == 1
        assert item["retirement_candidate"] is False
        assert item["retirement_reason"] == "1 call(s) in review window"

    async def test_legacy_completion_event_is_metered_and_workplan_event_is_preferred(self, client):
        await _create_domain(client)
        topic = await _create_topic(client)
        wp = await _create_workplan(client, topic["id"])

        r = await client.patch(
            f"/workstreams/{wp['id']}",
            json={"status": "finished"},
            headers={"X-StateHub-Component": "old-client"},
        )
        assert r.status_code == 200
        assert r.json()["status"] == "finished"

        summary = (await client.get("/legacy-meter/summary")).json()
        items = _summary_by_key(summary)
        assert items["rest_api:PATCH /workstreams/{workstream_id}"]["window"]["components"] == {
            "old-client": 1
        }
        event = items["event_subject:org.statehub.workstream.completed"]
        assert event["interface"]["replacement_ref"] == "org.statehub.workplan.completed"
        assert event["window"]["components"] == {"state-hub.events": 1}

    async def test_workplan_dependency_and_execution_aliases(self, client):
        await _create_domain(client)
        topic = await _create_topic(client)
        first = await _create_workplan(client, topic["id"], slug="first", title="First")
        second = await _create_workplan(client, topic["id"], slug="second", title="Second")

        r = await client.post(
            f"/workplans/{first['id']}/dependencies/",
            json={"to_workstream_id": second["id"]},
        )
        assert r.status_code == 201, r.text

        r = await client.get(f"/workplans/{first['id']}/dependencies/")
        assert r.status_code == 200
        assert len(r.json()) == 1

        r = await client.patch(
            f"/execution/workplans/{first['id']}/intent",
            json={"execution_state": "queued", "launch_mode": "queued"},
        )
        assert r.status_code == 200
        assert r.json()["execution_state"] == "queued"

        r = await client.patch(
            f"/execution/workstreams/{first['id']}/intent",
            json={"execution_state": "manual", "launch_mode": "manual"},
            headers={"X-StateHub-Component": "old-executor"},
        )
        assert r.status_code == 200
        assert r.headers["X-StateHub-Replacement"] == "/execution/workplans/{workplan_id}/intent"

        summary = (await client.get("/legacy-meter/summary")).json()
        item = _summary_by_key(summary)["rest_api:PATCH /execution/workstreams/{workstream_id}/intent"]
        assert item["window"]["components"] == {"old-executor": 1}