praxis/tests/test_p2_assist_integration.py

"""P2 integration test — assist aggregation → endpoint → cost → NFR (TASK-12-05).

Requires Postgres (skips if PRAXIS_PG_DSN not set). End-to-end P2 integration:
  1. Seed 12 mock assist shifts (12 distinct learners — above k-anon threshold).
  2. Run the aggregation hook for each → cohort_aggregates populated with assist metrics.
  3. GET /api/operator/cohort (with auth cookie) → returns assist volume (non-suppressed).
  4. GET /api/operator/failure-patterns → returns assist_guardrail_block_rate.
  5. Seed 5 more assist shifts from 5 NEW distinct learners for a different path →
     GET /api/operator/cohort for that path → suppressed cells (5 < 10).
  6. Verify assist_p95_latency_ms is in the aggregates.
  7. Verify assist_cost_cents is in the session_outcome.
  8. Verify the C-3 budget check runs at shift-end.
  9. Verify the tech-debt fixes: aggregation cache survives restart (mock),
     cookie-secret warning, credential status enum, argon2id offloaded.

G-038 differencing-attack e2e: k-anon threshold enforced (12 not suppressed,
5 suppressed). No per-learner data in any response.
"""

from __future__ import annotations

import asyncio
import datetime as _dt
import os
from unittest.mock import AsyncMock, MagicMock

import pytest

pytestmark = pytest.mark.skipif(
    not os.environ.get("PRAXIS_PG_DSN"),
    reason="PRAXIS_PG_DSN not set — P2 assist integration tests skipped.",
)


@pytest.fixture
async def pg_pool():
    import asyncpg

    pool = await asyncpg.create_pool(
        dsn=os.environ["PRAXIS_PG_DSN"], min_size=1, max_size=5, command_timeout=10,
    )
    try:
        yield pool
    finally:
        await pool.close()


@pytest.fixture
async def pg_store(pg_pool):
    from db.pg_migrate import apply_pg_migrations
    from db.pg_store import PgStore

    await apply_pg_migrations(pg_pool)
    # Clean cohort_aggregates + operators for an isolated run.
    async with pg_pool.acquire() as conn:
        await conn.execute("TRUNCATE cohort_aggregates, operators, issued_credentials")
    return PgStore(pg_pool)


@pytest.fixture
async def authed_client(pg_store):
    """A TestClient with auth + the operator routers wired to pg_store."""
    from fastapi import FastAPI
    from fastapi.testclient import TestClient
    from starlette.middleware.sessions import SessionMiddleware

    from server.auth.dependencies import current_operator
    from server.auth.models import Operator
    from server.operator.cohort import router as cohort_router
    from server.operator.failure_patterns import router as failure_router
    from server.operator.mastery import router as mastery_router

    app = FastAPI()
    app.state.pg_store = pg_store
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef1234567890")
    app.include_router(cohort_router)
    app.include_router(failure_router)
    app.include_router(mastery_router)

    # Stub auth — every request is operator "integration-tester".
    async def _stub_op():
        return Operator(id="op-1", username="tester", display_name="T", role="operator")
    app.dependency_overrides[current_operator] = _stub_op
    return TestClient(app)


def _assist_outcome(
    learner_ref: str,
    path: str = "customer_service",
    turn_count: int = 20,
    blocks: int = 2,
    p95_latency_ms: float = 580.0,
    cost_cents: int = 20,
) -> dict:
    return {
        "learner_ref": learner_ref,
        "path": path,
        "scenario_id": "assist:refund",
        "outcome": "completed",
        "session_type": "assist",
        "rubric_scores": [],
        "failure_mode": None,
        "branch_path": [],
        "assist_turn_count": turn_count,
        "guardrail_blocks": blocks,
        "assist_p95_latency_ms": p95_latency_ms,
        "assist_p50_latency_ms": 500.0,
        "assist_p99_latency_ms": 620.0,
        "assist_within_pilot": True,
        "assist_cost_cents": cost_cents,
        "timestamp": _dt.datetime.now(_dt.timezone.utc).isoformat(),
    }


@pytest.mark.asyncio
async def test_p2_assist_aggregation_k_anon_threshold(pg_store, authed_client):
    """1-4: 12 assist shifts (12 learners) → non-suppressed; 5 → suppressed."""
    from server.cohort.aggregator import aggregate_session

    # 1. Seed 12 assist shifts for 'customer_service' (12 distinct learners).
    for i in range(12):
        await aggregate_session(pg_store, _assist_outcome(f"learner-{i}"))

    # 2. Verify cohort_aggregates has assist metrics.
    async with pg_store.pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT metric, value, cell_count, cell_suppressed "
            "FROM cohort_aggregates WHERE path = 'customer_service' "
            "AND metric LIKE 'assist_%'"
        )
    metrics = {r["metric"]: r for r in rows}
    assert "assist_shifts_count" in metrics
    assert "assist_turns_count" in metrics
    assert "assist_active_learners_count" in metrics
    assert "assist_guardrail_block_rate" in metrics
    # 12 learners → not suppressed.
    assert metrics["assist_active_learners_count"]["cell_suppressed"] is False
    assert metrics["assist_active_learners_count"]["value"] == 12.0

    # 3. GET /api/operator/cohort → returns assist volume (non-suppressed).
    r = authed_client.get("/api/operator/cohort")
    assert r.status_code == 200
    cohort_metrics = {
        c["metric"]: c for v in r.json()["views"] if v["path"] == "customer_service"
        for c in v["metrics"]
    }
    assert "assist_shifts_count" in cohort_metrics
    assert cohort_metrics["assist_shifts_count"]["cell_suppressed"] is False

    # 4. GET /api/operator/failure-patterns → returns assist_guardrail_block_rate.
    r = authed_client.get("/api/operator/failure-patterns")
    assert r.status_code == 200
    fp_metrics = {
        c["metric"]: c for v in r.json()["views"] if v["path"] == "customer_service"
        for c in v["metrics"]
    }
    assert "assist_guardrail_block_rate" in fp_metrics

    # 5. Seed 5 assist shifts for a DIFFERENT path (5 NEW learners) → suppressed.
    for i in range(5):
        await aggregate_session(pg_store, _assist_outcome(f"new-learner-{i}", path="retail_sales"))

    r = authed_client.get("/api/operator/cohort")
    retail_metrics = {
        c["metric"]: c for v in r.json()["views"] if v["path"] == "retail_sales"
        for c in v["metrics"]
    }
    assert "assist_shifts_count" in retail_metrics
    # 5 < 10 → suppressed.
    assert retail_metrics["assist_shifts_count"]["cell_suppressed"] is True
    assert retail_metrics["assist_shifts_count"]["value"] is None


@pytest.mark.asyncio
async def test_p2_assist_p95_latency_in_aggregates(pg_store):
    """6: assist_p95_latency_ms is in the aggregates (D-072)."""
    from server.cohort.aggregator import aggregate_session

    for i in range(12):
        await aggregate_session(pg_store, _assist_outcome(f"learner-{i}", p95_latency_ms=580.0))

    async with pg_store.pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT value, cell_suppressed FROM cohort_aggregates "
            "WHERE path = 'customer_service' AND metric = 'assist_p95_latency_ms'"
        )
    assert row is not None
    assert row["cell_suppressed"] is False
    assert row["value"] is not None
    # The running mean of per-shift p95 (580.0) → ~580.
    assert 570.0 <= float(row["value"]) <= 590.0


@pytest.mark.asyncio
async def test_p2_assist_cost_cents_in_session_outcome():
    """7: assist_cost_cents is in the session_outcome (TASK-11-01)."""
    from server.assist.session import AssistSession
    from server.assist.context import AssistContext

    ctx = AssistContext(
        system_prompt="", current_week=1, scenario_tag="refund",
        theta=0.0, coaching_focus="empathy", path_slug="customer_service",
    )
    session = AssistSession.__new__(AssistSession)
    session.assist_cost_cents = 0
    session.turn_count = 3
    session.guardrail_block_count = 0
    session.latency_metrics = MagicMock()
    session.latency_metrics.summary = MagicMock(return_value={
        "p50": 500.0, "p95": 580.0, "p99": 620.0, "count": 3,
        "target_ms": 600, "pilot_tolerance_ms": 650,
        "within_target": True, "within_pilot": True,
    })
    session.context = ctx
    session.learner_id = "learner-1"
    session.session_id = "test-session"

    # Add 3 turns of cost.
    session.add_assist_turn_cost(5)
    session.add_assist_turn_cost(10)
    session.add_assist_turn_cost(3)

    outcome = session._build_session_outcome("completed")
    assert outcome["assist_cost_cents"] == 18  # 5 + 10 + 3
    assert outcome["session_type"] == "assist"
    assert outcome["assist_p95_latency_ms"] == 580.0


@pytest.mark.asyncio
async def test_p2_c3_budget_check_runs():
    """8: the C-3 budget check runs + reports within_budget (TASK-11-02)."""
    from server.assist.budget_check import C3_TARGET_USD, check_c3_budget

    # 20 turns/shift × 20 shifts/month at 0.05 cents/turn → $0.20/month.
    result = check_c3_budget(
        assist_turns_per_shift=20,
        shifts_per_month=20,
        cost_per_turn_cents=0.05,
    )
    assert result["within_budget"] is True
    assert result["total_with_practice"] <= C3_TARGET_USD
    assert result["c3_target"] == 3.0


@pytest.mark.asyncio
async def test_p2_techdebt_aggregation_cache_survives_restart(pg_store, tmp_path):
    """9a: aggregation cache survives a restart (TASK-12-01, P1+ #7)."""
    from server.cohort.aggregator import aggregate_session
    from server.cohort.learner_cache import (
        _clear_learner_cache,
        _count_distinct_learners,
        _load_learner_cache,
    )

    # Point the cache to a temp file.
    pg_store.cohort_cache_db_path = str(tmp_path / "cache.db")

    # Seed 10 learners.
    for i in range(10):
        await aggregate_session(pg_store, _assist_outcome(f"learner-{i}"))

    # The persisted cache should have 10 distinct learners for this path.
    window_start = (_dt.datetime.now(_dt.timezone.utc).date() - _dt.timedelta(days=6))
    count = await _count_distinct_learners(pg_store, "customer_service", window_start)
    assert count == 10

    # Simulate a restart: clear the in-memory cache + reload from SQLite.
    if hasattr(pg_store, "_agg_cache"):
        del pg_store._agg_cache
    loaded = await _load_learner_cache(pg_store)
    key = ("customer_service", "__learners__", window_start)
    assert key in loaded
    assert len(loaded[key]) == 10  # survived the "restart"

    # Clear the cache (nightly reconciliation).
    await _clear_learner_cache(pg_store)
    count_after_clear = await _count_distinct_learners(pg_store, "customer_service", window_start)
    assert count_after_clear == 0


@pytest.mark.asyncio
async def test_p2_techdebt_cookie_secret_warning(monkeypatch):
    """9b: cookie-secret <32 bytes logs a WARNING (TASK-12-02, P1+ #3)."""
    from loguru import logger as _logger
    from server.auth.cookies import get_session_middleware_kwargs

    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "short")  # 5 bytes < 32
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "true")
    msgs: list[str] = []
    sink_id = _logger.add(lambda m: msgs.append(str(m)), level="WARNING")
    try:
        kw = get_session_middleware_kwargs()
    finally:
        _logger.remove(sink_id)
    assert kw["secret_key"] == "short"  # accepted (backward compat)
    assert any("<32 bytes" in m for m in msgs)


@pytest.mark.asyncio
async def test_p2_techdebt_credential_status_enum():
    """9c: set_credential_status enum validation (TASK-12-03, P1+ #4)."""
    from db.pg_store import PgStore

    pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    pool.acquire = MagicMock(return_value=cm)

    store = PgStore(pool)
    # Invalid status → ValueError.
    with pytest.raises(ValueError, match="Invalid credential status"):
        await store.set_credential_status("cred-1", "deleted")
    # Valid statuses work.
    await store.set_credential_status("cred-1", "revoked")
    await store.set_credential_status("cred-1", "active")


@pytest.mark.asyncio
async def test_p2_techdebt_argon2id_offloaded():
    """9d: argon2id verify_password offloaded to asyncio.to_thread (P1+ #1)."""
    import asyncio as _asyncio
    import server.auth.routes as _routes_mod
    from server.auth.passwords import hash_password

    # The login handler should use asyncio.to_thread for verify_password.
    # Verify the module imports asyncio + the handler references to_thread.
    assert hasattr(_routes_mod, "asyncio")
    assert _asyncio.to_thread is _routes_mod.asyncio.to_thread

    # Functional check: verify_password is callable via to_thread.
    h = hash_password("pw")
    result = await _asyncio.to_thread(_routes_mod.verify_password, h, "pw")
    assert result is True


@pytest.mark.asyncio
async def test_p2_no_per_learner_data_in_responses(pg_store, authed_client):
    """No per-learner data in any dashboard response (D-031, G-038)."""
    from server.cohort.aggregator import aggregate_session

    for i in range(12):
        await aggregate_session(pg_store, _assist_outcome(f"learner-sensitive-{i}"))

    for endpoint in ("/api/operator/cohort", "/api/operator/failure-patterns", "/api/operator/mastery"):
        r = authed_client.get(endpoint)
        assert r.status_code == 200
        # No learner ref should appear in the response.
        text = r.text
        assert "learner-sensitive-" not in text, \
            f"per-learner data leaked in {endpoint} response"