FabledGrimoire/tests/test_system_usage_readout.py

"""Rulings are counted where someone will read them.

`system_usage_events` is read in two places, and these pin both readouts:

- PER SYSTEM: `usage_for_systems` — one GROUP BY, a zero shape for a System
  nothing touched — carried by the Systems list, the REST detail and MCP
  `get_system`.
- IN AGGREGATE: `retrieval_summary`'s `system_usage` block — counts by source,
  the areas named, and deliberately NO pull-through ratio.

Integration for both aggregates, not mocked: a GROUP BY the database rejects
is swallowed by the fail-open guard and reads as zero.
"""
from unittest.mock import AsyncMock, patch

import pytest
import pytest_asyncio

from grimoire.models import async_session
from grimoire.services.system_usage import empty_system_usage
# Bound at import, before conftest's autouse stub replaces the module attribute.
from grimoire.services.system_usage import usage_for_systems as real_usage_for_systems
from tests.helpers import ensure_user


# ── the payloads carry it ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_get_system_carries_its_usage():
    from tests.helpers import fake_system

    shown = {**empty_system_usage(), "surfaced_count": 4, "pull_count": 1}
    with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
         patch("grimoire.mcp.tools.systems.systems_svc") as svc, \
         patch("grimoire.services.system_usage.usage_for_systems",
               AsyncMock(return_value={3: shown})):
        svc.get_system = AsyncMock(return_value=fake_system(id=3))
        svc.list_records_for_system = AsyncMock(return_value=[])
        from grimoire.mcp.tools.systems import get_system
        result = await get_system(system_id=3)
    assert result["usage"] == shown


def test_the_zero_shape_matches_the_client_type():
    """The four keys `RecordUsage` in frontend/src/types/usage.ts declares —
    a fifth here would be a field the chip never reads, a missing one a
    crash on every System nothing has touched."""
    assert set(empty_system_usage()) == {
        "surfaced_count", "pull_count", "last_surfaced_at", "last_pulled_at",
    }


@pytest.mark.asyncio
async def test_an_empty_id_list_reads_nothing():
    with patch("grimoire.services.system_usage.async_session") as session:
        assert await real_usage_for_systems([]) == {}
    session.assert_not_called()


# ── the aggregates, against Postgres ─────────────────────────────────────


@pytest_asyncio.fixture
async def ruled_areas():
    """An owner, a project and two Systems, with usage rows on both."""
    from grimoire.models.project import Project
    from grimoire.models.system import System
    from grimoire.models.system_usage import SystemUsageEvent

    async with async_session() as s:
        owner = await ensure_user(s, "system_usage_owner")
        other = await ensure_user(s, "system_usage_other")
        project = Project(user_id=owner.id, title="Ruled areas")
        s.add(project)
        await s.flush()
        billing = System(user_id=owner.id, project_id=project.id, name="Billing")
        storage = System(user_id=owner.id, project_id=project.id, name="Storage")
        s.add_all([billing, storage])
        await s.flush()

        def ev(uid, sid, event, source):
            return SystemUsageEvent(
                user_id=uid, system_id=sid, event=event, source=source,
                project_id=project.id,
            )

        s.add_all([
            ev(owner.id, billing.id, "surfaced", "rulings_pre_tool"),
            ev(owner.id, billing.id, "surfaced", "rulings_pre_tool"),
            ev(owner.id, billing.id, "surfaced", "rulings_write_path"),
            ev(owner.id, billing.id, "pulled", "mcp_get_system"),
            ev(owner.id, storage.id, "surfaced", "rulings_write_path"),
            # Another user's session: in the per-System count, which is about
            # the System, and NOT in the owner's telemetry, which is theirs.
            ev(other.id, storage.id, "surfaced", "rulings_pre_tool"),
        ])
        ids = {
            "owner": owner.id, "other": other.id, "pid": project.id,
            "billing": billing.id, "storage": storage.id,
        }
        await s.commit()
    yield ids

    from sqlalchemy import delete
    async with async_session() as s:
        await s.execute(delete(SystemUsageEvent).where(
            SystemUsageEvent.system_id.in_([ids["billing"], ids["storage"]])))
        await s.execute(delete(System).where(System.project_id == ids["pid"]))
        await s.execute(delete(Project).where(Project.id == ids["pid"]))
        await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_usage_for_systems_counts_each_system(ruled_areas):
    ids = ruled_areas
    out = await real_usage_for_systems([ids["billing"], ids["storage"], 987654321])
    billing, storage = out[ids["billing"]], out[ids["storage"]]
    assert billing["surfaced_count"] == 3 and billing["pull_count"] == 1
    assert billing["last_surfaced_at"] and billing["last_pulled_at"]
    assert storage["surfaced_count"] == 2 and storage["pull_count"] == 0
    assert storage["last_pulled_at"] is None
    assert out[987654321] == empty_system_usage()


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_the_telemetry_block_names_the_areas_and_offers_no_ratio(ruled_areas):
    from grimoire.services.retrieval_telemetry import retrieval_summary

    ids = ruled_areas
    block = (await retrieval_summary(ids["owner"], days=30))["system_usage"]
    assert "system_usage_failed" not in block, "the aggregate did not execute"
    assert block["surfaced"] == 4 and block["pulled"] == 1
    assert block["by_source"] == {
        "rulings_pre_tool": 2, "rulings_write_path": 2, "mcp_get_system": 1,
    }
    assert block["distinct_systems_surfaced"] == 2
    assert block["distinct_systems_pulled"] == 1
    assert [(r["name"], r["surfaced"], r["pulled"]) for r in block["by_system"]] == [
        ("Billing", 3, 1), ("Storage", 1, 0),
    ]
    # Rulings are delivered in full in the line; a ratio of opens would read
    # near zero on an arm that is working.
    assert "pull_through" not in block


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_a_fresh_install_reads_zero_not_failed():
    from grimoire.services.retrieval_telemetry import retrieval_summary

    block = (await retrieval_summary(990031, days=30))["system_usage"]
    assert "system_usage_failed" not in block
    assert block["surfaced"] == 0 and block["by_system"] == []
    assert block["covers_window"] is None