"""SLICE-08 TASK-08-02 — mastery gate audit log queryability test. Verifies the mastery_gate_events audit log (REQ-NFR-MAST-02) is queryable by learner, by path, and by date range, and that the evidence (scenarios_passed, rubric_scores) is persisted and reconstructable as structured JSON. Three events are inserted across two learners and two paths; queries verify: - list_gate_events(learner_id) returns all rows for that learner - list_gate_events(learner_id, path) filters by path - raw SQL date-range query filters by recorded_at - JSON fields parse back to the original structured evidence """ from __future__ import annotations import asyncio import json import sqlite3 from pathlib import Path import aiosqlite import pytest from db.store import PraxisStore _LEARNER_A = "learner-audit-A" _LEARNER_B = "learner-audit-B" _PATH_CS = "customer_service" _PATH_OTHER = "health_electrical" def _rubric_scores_a1() -> list[dict]: return [ {"criterion_id": "empathy", "level": 4, "weight": 0.35, "evidence_quote": "I hear you.", "matched_signals": ["named_emotion_in_own_words"]}, {"criterion_id": "resolution", "level": 3, "weight": 0.30, "evidence_quote": "Refund issued.", "matched_signals": ["concrete_method", "concrete_next_step"]}, {"criterion_id": "de_escalation", "level": 3, "weight": 0.20, "evidence_quote": "I hear you.", "matched_signals": ["explicit_acknowledge_reframe_offer"]}, {"criterion_id": "professionalism", "level": 3, "weight": 0.15, "evidence_quote": "Anything else?", "matched_signals": ["plain_language"]}, ] def _rubric_scores_a2() -> list[dict]: return [ {"criterion_id": "empathy", "level": 5, "weight": 0.35, "evidence_quote": "That's frustrating.", "matched_signals": ["tone_pace_adjusted"]}, {"criterion_id": "resolution", "level": 4, "weight": 0.30, "evidence_quote": "70% credit today.", "matched_signals": ["decision_tree_of_options"]}, {"criterion_id": "de_escalation", "level": 4, "weight": 0.20, "evidence_quote": "Let me reframe.", "matched_signals": ["cycles_acknowledge_reframe"]}, {"criterion_id": "professionalism", "level": 4, "weight": 0.15, "evidence_quote": "Confirmed.", "matched_signals": ["adapts_register"]}, ] def _rubric_scores_b1() -> list[dict]: return [ {"criterion_id": "safety", "level": 3, "weight": 0.6, "evidence_quote": "Isolated the circuit.", "matched_signals": ["lockout_tagout"]}, {"criterion_id": "communication", "level": 3, "weight": 0.4, "evidence_quote": "Told the customer to stand back.", "matched_signals": ["plain_language"]}, ] @pytest.fixture def tmp_db(tmp_path: Path) -> Path: return tmp_path / "test_gate_audit.db" @pytest.mark.asyncio async def test_insert_three_events_and_query_by_learner(tmp_db: Path): store = PraxisStore(tmp_db) await store.init() e1 = await store.record_gate_event( _LEARNER_A, _PATH_CS, week=1, scenarios_passed=["cs_refund_ca_v01"], rubric_scores=_rubric_scores_a1(), mastery_score=3.4, gate_open=False, ) e2 = await store.record_gate_event( _LEARNER_A, _PATH_CS, week=1, scenarios_passed=["cs_refund_ca_v01", "cs_escalation_ca_v02"], rubric_scores=_rubric_scores_a2(), mastery_score=4.1, gate_open=True, ) e3 = await store.record_gate_event( _LEARNER_B, _PATH_OTHER, week=3, scenarios_passed=["he_lockout_v01"], rubric_scores=_rubric_scores_b1(), mastery_score=3.0, gate_open=False, ) events_a = await store.list_gate_events(_LEARNER_A) assert len(events_a) == 2 assert {ev["id"] for ev in events_a} == {e1, e2} events_b = await store.list_gate_events(_LEARNER_B) assert len(events_b) == 1 assert events_b[0]["id"] == e3 @pytest.mark.asyncio async def test_query_by_path_filters_correctly(tmp_db: Path): store = PraxisStore(tmp_db) await store.init() await store.record_gate_event( _LEARNER_A, _PATH_CS, week=1, scenarios_passed=["cs_refund_ca_v01"], rubric_scores=_rubric_scores_a1(), mastery_score=3.4, gate_open=False, ) await store.record_gate_event( _LEARNER_A, _PATH_OTHER, week=2, scenarios_passed=["he_lockout_v01"], rubric_scores=_rubric_scores_b1(), mastery_score=3.0, gate_open=False, ) cs_only = await store.list_gate_events(_LEARNER_A, _PATH_CS) assert len(cs_only) == 1 assert cs_only[0]["path"] == _PATH_CS other_only = await store.list_gate_events(_LEARNER_A, _PATH_OTHER) assert len(other_only) == 1 assert other_only[0]["path"] == _PATH_OTHER no_match = await store.list_gate_events(_LEARNER_A, "nonexistent_path") assert no_match == [] @pytest.mark.asyncio async def test_date_range_query_via_raw_sql(tmp_db: Path): """list_gate_events does not take a date range; verify via a direct query that recorded_at is queryable and that a date-range filter works.""" store = PraxisStore(tmp_db) await store.init() await store.record_gate_event( _LEARNER_A, _PATH_CS, week=1, scenarios_passed=["cs_refund_ca_v01"], rubric_scores=_rubric_scores_a1(), mastery_score=3.4, gate_open=False, ) async with aiosqlite.connect(str(tmp_db)) as db: db.row_factory = aiosqlite.Row cur = await db.execute( "SELECT * FROM mastery_gate_events " "WHERE learner_id = ? AND recorded_at >= datetime('now', '-1 day') " "ORDER BY recorded_at", (_LEARNER_A,), ) rows = [dict(r) for r in await cur.fetchall()] assert len(rows) == 1 assert rows[0]["learner_id"] == _LEARNER_A async with aiosqlite.connect(str(tmp_db)) as db: db.row_factory = aiosqlite.Row cur = await db.execute( "SELECT * FROM mastery_gate_events " "WHERE learner_id = ? AND recorded_at < datetime('now', '-10 year')", (_LEARNER_A,), ) rows_old = [dict(r) for r in await cur.fetchall()] assert rows_old == [] @pytest.mark.asyncio async def test_evidence_json_parses_back_reconstructable(tmp_db: Path): store = PraxisStore(tmp_db) await store.init() scenarios = ["cs_refund_ca_v01", "cs_escalation_ca_v02", "cs_policy_exception_ca_v03"] scores = _rubric_scores_a1() + _rubric_scores_a2() await store.record_gate_event( _LEARNER_A, _PATH_CS, week=2, scenarios_passed=scenarios, rubric_scores=scores, mastery_score=4.0, gate_open=True, ) events = await store.list_gate_events(_LEARNER_A, _PATH_CS) assert len(events) == 1 ev = events[0] sp = json.loads(ev["scenarios_passed_json"]) assert sp == scenarios rs = json.loads(ev["rubric_scores_json"]) assert len(rs) == len(_rubric_scores_a1()) + len(_rubric_scores_a2()) for item in rs: assert "criterion_id" in item assert "level" in item assert isinstance(item["level"], int) and 1 <= item["level"] <= 5 assert "weight" in item assert isinstance(item["matched_signals"], list) assert ev["mastery_score"] == 4.0 assert ev["gate_open"] == 1 assert ev["week"] == 2 @pytest.mark.asyncio async def test_three_events_all_queryable_distinct_ids(tmp_db: Path): store = PraxisStore(tmp_db) await store.init() ids: list[str] = [] ids.append(await store.record_gate_event( _LEARNER_A, _PATH_CS, week=1, scenarios_passed=["s1"], rubric_scores=_rubric_scores_a1(), mastery_score=3.0, gate_open=False, )) ids.append(await store.record_gate_event( _LEARNER_A, _PATH_CS, week=2, scenarios_passed=["s1", "s2"], rubric_scores=_rubric_scores_a2(), mastery_score=3.6, gate_open=False, )) ids.append(await store.record_gate_event( _LEARNER_A, _PATH_CS, week=3, scenarios_passed=["s1", "s2", "s3"], rubric_scores=_rubric_scores_a1(), mastery_score=4.0, gate_open=True, )) assert len(set(ids)) == 3 events = await store.list_gate_events(_LEARNER_A, _PATH_CS) assert len(events) == 3 assert {ev["id"] for ev in events} == set(ids) weeks = sorted(ev["week"] for ev in events) assert weeks == [1, 2, 3]