diff --git a/db/pg_migrate.py b/db/pg_migrate.py new file mode 100644 index 0000000..421ec5f --- /dev/null +++ b/db/pg_migrate.py @@ -0,0 +1,71 @@ +"""Postgres migration runner — applies db/pg_migrations/*.sql in order. + +Mirrors db/migrate.py: ordered .sql files tracked in a `_pg_migrations` +table so re-running is idempotent. Uses an asyncpg pool. Retries on +connection failure (3 attempts, 2s backoff — R-MT-02 mitigation). +""" + +from __future__ import annotations + +import asyncio +import datetime as _dt +from pathlib import Path + +import asyncpg + +_DEFAULT_MIGRATIONS_DIR = Path(__file__).resolve().parent / "pg_migrations" +_RETRY_ATTEMPTS = 3 +_RETRY_BACKOFF_S = 2.0 + + +async def apply_pg_migrations( + pool: asyncpg.Pool, + migrations_dir: Path | None = None, +) -> list[str]: + """Apply all pending Postgres migrations in order. Returns applied names. + + Idempotent — no-op if all migrations are already applied. Each migration + runs within a transaction; the `_pg_migrations` tracking row is inserted + in the same transaction so a failure rolls back cleanly. + """ + mdir = migrations_dir or _DEFAULT_MIGRATIONS_DIR + if not mdir.exists(): + return [] + + async def _run() -> list[str]: + async with pool.acquire() as conn: + await conn.execute( + "CREATE TABLE IF NOT EXISTS _pg_migrations (" + "id TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now()" + ")" + ) + rows = await conn.fetch("SELECT id FROM _pg_migrations") + applied_ids = {r["id"] for r in rows} + applied: list[str] = [] + for sql_path in sorted(mdir.glob("*.sql")): + mid = sql_path.stem + if mid in applied_ids: + continue + sql = sql_path.read_text(encoding="utf-8") + async with conn.transaction(): + await conn.execute(sql) + await conn.execute( + "INSERT INTO _pg_migrations (id) VALUES ($1)", mid + ) + applied.append(mid) + return applied + + last_exc: Exception | None = None + for attempt in range(1, _RETRY_ATTEMPTS + 1): + try: + return await _run() + except (asyncpg.PostgresConnectionError, ConnectionError, OSError) as exc: + last_exc = exc + if attempt < _RETRY_ATTEMPTS: + await asyncio.sleep(_RETRY_BACKOFF_S) + continue + assert last_exc is not None + raise last_exc + + +__all__ = ["apply_pg_migrations"] \ No newline at end of file diff --git a/db/pg_migrations/0001_operator_tier.sql b/db/pg_migrations/0001_operator_tier.sql new file mode 100644 index 0000000..6c42dcc --- /dev/null +++ b/db/pg_migrations/0001_operator_tier.sql @@ -0,0 +1,59 @@ +-- Praxis v0.4 operator-tier schema migration 0001. +-- Creates the 5 operator-tier tables. Uses gen_random_uuid() (PG16 core). +-- Idempotent via IF NOT EXISTS (also safe through pg_migrate tracking). + +CREATE TABLE IF NOT EXISTS operators ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + display_name TEXT, + role TEXT NOT NULL DEFAULT 'operator', + is_active BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_login_at TIMESTAMPTZ +); + +CREATE TABLE IF NOT EXISTS issued_credentials ( + id UUID PRIMARY KEY, + operator_id UUID REFERENCES operators(id), + learner_ref TEXT NOT NULL, + vc_type TEXT, + payload_jsonb JSONB NOT NULL, + signature_b64 TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + issued_at TIMESTAMPTZ NOT NULL DEFAULT now(), + revoked_at TIMESTAMPTZ +); + +CREATE TABLE IF NOT EXISTS mastery_gate_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + learner_ref TEXT NOT NULL, + scenario_id TEXT, + path_id TEXT NOT NULL, + gate_outcome TEXT, + rubric_scores_jsonb JSONB, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(), + source TEXT NOT NULL DEFAULT 'sync' +); + +CREATE TABLE IF NOT EXISTS cohort_aggregates ( + path TEXT NOT NULL, + metric TEXT NOT NULL, + window_start DATE NOT NULL, + window_end DATE NOT NULL, + value NUMERIC, + cell_count INTEGER NOT NULL DEFAULT 0, + cell_suppressed BOOLEAN NOT NULL DEFAULT FALSE, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (path, metric, window_start) +); +CREATE INDEX IF NOT EXISTS cohort_aggregates_path_window_idx + ON cohort_aggregates (path, window_start); + +CREATE TABLE IF NOT EXISTS issuer_keys ( + id TEXT PRIMARY KEY, + public_key TEXT NOT NULL, + private_key_enc BYTEA, + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); \ No newline at end of file diff --git a/db/pg_schema.sql b/db/pg_schema.sql new file mode 100644 index 0000000..ab8f441 --- /dev/null +++ b/db/pg_schema.sql @@ -0,0 +1,71 @@ +-- Praxis v0.4 operator-tier Postgres schema (reference). +-- Applied in order by db/pg_migrate.py via db/pg_migrations/*.sql. +-- The canonical migration is 0001_operator_tier.sql; this file is the +-- human-readable reference (kept in sync). Uses gen_random_uuid() which +-- is in PG16 core (no extension needed — R-MT-05 verified). +-- +-- Tables: +-- operators — operator accounts (argon2id password hash) +-- issued_credentials — VC issuance log (learner_ref is opaque, no FK) +-- mastery_gate_events — mastery gate audit log (REQ-NFR-MAST-02) +-- cohort_aggregates — k-anonymized cohort metrics (plain table, D-050) +-- issuer_keys — Ed25519 issuer key lifecycle (active/superseded) +-- +-- No cross-DB FKs (D-031). learner_ref is an opaque string in Postgres. + +CREATE TABLE IF NOT EXISTS operators ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + display_name TEXT, + role TEXT NOT NULL DEFAULT 'operator', + is_active BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_login_at TIMESTAMPTZ +); + +CREATE TABLE IF NOT EXISTS issued_credentials ( + id UUID PRIMARY KEY, + operator_id UUID REFERENCES operators(id), + learner_ref TEXT NOT NULL, + vc_type TEXT, + payload_jsonb JSONB NOT NULL, + signature_b64 TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + issued_at TIMESTAMPTZ NOT NULL DEFAULT now(), + revoked_at TIMESTAMPTZ +); + +CREATE TABLE IF NOT EXISTS mastery_gate_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + learner_ref TEXT NOT NULL, + scenario_id TEXT, + path_id TEXT NOT NULL, + gate_outcome TEXT, + rubric_scores_jsonb JSONB, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(), + source TEXT NOT NULL DEFAULT 'sync' +); + +CREATE TABLE IF NOT EXISTS cohort_aggregates ( + path TEXT NOT NULL, + metric TEXT NOT NULL, + window_start DATE NOT NULL, + window_end DATE NOT NULL, + value NUMERIC, + cell_count INTEGER NOT NULL DEFAULT 0, + cell_suppressed BOOLEAN NOT NULL DEFAULT FALSE, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (path, metric, window_start) +); +-- Plain table, NOT partitioned (D-050..D-053; add partitioning post-pilot). +CREATE INDEX IF NOT EXISTS cohort_aggregates_path_window_idx + ON cohort_aggregates (path, window_start); + +CREATE TABLE IF NOT EXISTS issuer_keys ( + id TEXT PRIMARY KEY, + public_key TEXT NOT NULL, + private_key_enc BYTEA, + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); \ No newline at end of file diff --git a/db/pg_store.py b/db/pg_store.py new file mode 100644 index 0000000..18655e6 --- /dev/null +++ b/db/pg_store.py @@ -0,0 +1,280 @@ +"""Postgres store — operator-tier access layer (D-040, D-050, TASK-01-06). + +Async access via an asyncpg.Pool. Implements the IssuerKeyStore protocol +(server/vc/issuer_keys.py) so VC verification can use either PraxisStore +(SQLite, v0.3) or PgStore (Postgres, v0.4). No cross-DB joins (D-031); +`learner_ref` is an opaque string in Postgres (not a FK to SQLite). +""" + +from __future__ import annotations + +import json +import uuid +from typing import Any + +import asyncpg + + +class PgStore: + """Async Postgres store for the v0.4 operator tier.""" + + def __init__(self, pool: asyncpg.Pool) -> None: + self.pool = pool + + # ── Operator CRUD ──────────────────────────────────────────────────── + + async def get_operator_by_username(self, username: str) -> dict | None: + async with self.pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT id, username, password_hash, display_name, role, " + "is_active, created_at, last_login_at " + "FROM operators WHERE username = $1", + username, + ) + return dict(row) if row else None + + async def get_operator_by_id(self, operator_id: str) -> dict | None: + async with self.pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT id, username, password_hash, display_name, role, " + "is_active, created_at, last_login_at " + "FROM operators WHERE id = $1", + operator_id, + ) + return dict(row) if row else None + + async def update_last_login(self, operator_id: str) -> None: + async with self.pool.acquire() as conn: + await conn.execute( + "UPDATE operators SET last_login_at = now() WHERE id = $1", + operator_id, + ) + + async def insert_operator( + self, + username: str, + password_hash: str, + display_name: str | None = None, + *, + on_conflict_update: bool = False, + ) -> str | None: + """Insert an operator (idempotent on username). Returns the id, or + None if the row already existed and on_conflict_update is False.""" + async with self.pool.acquire() as conn: + if on_conflict_update: + row = await conn.fetchrow( + "INSERT INTO operators (username, password_hash, display_name) " + "VALUES ($1, $2, $3) " + "ON CONFLICT (username) DO UPDATE SET " + "password_hash = excluded.password_hash, " + "display_name = excluded.display_name " + "RETURNING id", + username, + password_hash, + display_name, + ) + return str(row["id"]) if row else None + row = await conn.fetchrow( + "INSERT INTO operators (username, password_hash, display_name) " + "VALUES ($1, $2, $3) " + "ON CONFLICT (username) DO NOTHING " + "RETURNING id", + username, + password_hash, + display_name, + ) + return str(row["id"]) if row else None + + # ── Cohort aggregate read/write ────────────────────────────────────── + + async def get_cohort_aggregates( + self, + path: str, + metric: str, + since_date: Any, + ) -> list[dict]: + async with self.pool.acquire() as conn: + rows = await conn.fetch( + "SELECT path, metric, window_start, window_end, value, " + "cell_count, cell_suppressed, updated_at " + "FROM cohort_aggregates " + "WHERE path = $1 AND metric = $2 AND window_start >= $3 " + "ORDER BY window_start", + path, + metric, + since_date, + ) + return [dict(r) for r in rows] + + async def upsert_cohort_aggregate( + self, + path: str, + metric: str, + window_start: Any, + window_end: Any, + value: float | None, + cell_count: int, + cell_suppressed: bool, + ) -> None: + async with self.pool.acquire() as conn: + await conn.execute( + "INSERT INTO cohort_aggregates " + "(path, metric, window_start, window_end, value, cell_count, " + "cell_suppressed, updated_at) " + "VALUES ($1, $2, $3, $4, $5, $6, $7, now()) " + "ON CONFLICT (path, metric, window_start) DO UPDATE SET " + "window_end = excluded.window_end, value = excluded.value, " + "cell_count = excluded.cell_count, " + "cell_suppressed = excluded.cell_suppressed, " + "updated_at = now()", + path, + metric, + window_start, + window_end, + value, + cell_count, + cell_suppressed, + ) + + # ── IssuerKeyStore protocol (D-051, TASK-04-02) ────────────────────── + + async def init_issuer_key( + self, + key_id: str, + public_key: str, + private_key_enc: bytes | None, + ) -> None: + async with self.pool.acquire() as conn: + await conn.execute( + "INSERT INTO issuer_keys (id, public_key, private_key_enc, status) " + "VALUES ($1, $2, $3, 'active') " + "ON CONFLICT (id) DO NOTHING", + key_id, + public_key, + private_key_enc if private_key_enc is not None else b"", + ) + + async def get_active_signing_key_row(self) -> dict | None: + async with self.pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT id, public_key, private_key_enc, status, created_at " + "FROM issuer_keys WHERE status = 'active' " + "ORDER BY created_at DESC LIMIT 1" + ) + return dict(row) if row else None + + async def get_public_key_row(self, key_id: str) -> dict | None: + # Queries by id (NOT status) so superseded keys are found too — + # this is the R-VC-MIG-01 verification fallback (D-051). + async with self.pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT id, public_key, private_key_enc, status, created_at " + "FROM issuer_keys WHERE id = $1", + key_id, + ) + return dict(row) if row else None + + async def set_issuer_key_superseded(self, key_id: str) -> None: + async with self.pool.acquire() as conn: + await conn.execute( + "UPDATE issuer_keys SET status = 'superseded' WHERE id = $1", + key_id, + ) + + # ── Credential methods ─────────────────────────────────────────────── + + async def insert_credential( + self, + cred_id: str, + learner_ref: str, + payload_json: str, + signature_b64: str, + *, + operator_id: str | None = None, + vc_type: str = "MasteryCredential", + ) -> None: + async with self.pool.acquire() as conn: + await conn.execute( + "INSERT INTO issued_credentials " + "(id, operator_id, learner_ref, vc_type, payload_jsonb, " + "signature_b64, status) " + "VALUES ($1, $2, $3, $4, $5::jsonb, $6, 'active')", + cred_id, + operator_id, + learner_ref, + vc_type, + payload_json, + signature_b64, + ) + + async def get_credential(self, cred_id: str) -> dict | None: + # Returns a row shaped like PraxisStore.get_credential so the + # verification code can use either store interchangeably. + async with self.pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT id, learner_ref, " + "payload_jsonb::text AS vc_payload_json, signature_b64, " + "status, issued_at " + "FROM issued_credentials WHERE id = $1", + cred_id, + ) + return dict(row) if row else None + + async def set_credential_status(self, cred_id: str, status: str) -> None: + extra = ", revoked_at = now()" if status == "revoked" else "" + async with self.pool.acquire() as conn: + await conn.execute( + f"UPDATE issued_credentials SET status = $1{extra} WHERE id = $2", + status, + cred_id, + ) + + async def list_credentials(self, operator_id: str | None = None) -> list[dict]: + async with self.pool.acquire() as conn: + if operator_id is None: + rows = await conn.fetch( + "SELECT id, learner_ref, vc_type, status, issued_at, " + "revoked_at FROM issued_credentials ORDER BY issued_at DESC" + ) + else: + rows = await conn.fetch( + "SELECT id, learner_ref, vc_type, status, issued_at, " + "revoked_at FROM issued_credentials " + "WHERE operator_id = $1 ORDER BY issued_at DESC", + operator_id, + ) + return [dict(r) for r in rows] + + # ── Mastery gate event ─────────────────────────────────────────────── + + async def record_gate_event( + self, + learner_ref: str, + path_id: str, + scenario_id: str | None = None, + gate_outcome: str | None = None, + rubric_scores_jsonb: Any | None = None, + ) -> str: + event_id = str(uuid.uuid4()) + scores_json = ( + rubric_scores_jsonb + if isinstance(rubric_scores_jsonb, str) + else (json.dumps(rubric_scores_jsonb) if rubric_scores_jsonb is not None else None) + ) + async with self.pool.acquire() as conn: + await conn.execute( + "INSERT INTO mastery_gate_events " + "(id, learner_ref, scenario_id, path_id, gate_outcome, " + "rubric_scores_jsonb, source) " + "VALUES ($1, $2, $3, $4, $5, $6::jsonb, 'sync')", + event_id, + learner_ref, + scenario_id, + path_id, + gate_outcome, + scores_json, + ) + return event_id + + +__all__ = ["PgStore"] \ No newline at end of file diff --git a/tests/test_pg_store.py b/tests/test_pg_store.py new file mode 100644 index 0000000..8db6183 --- /dev/null +++ b/tests/test_pg_store.py @@ -0,0 +1,221 @@ +"""PgStore + asyncpg pool integration test (TASK-01-07). + +Requires a live Postgres instance. Skips gracefully when PRAXIS_PG_DSN is +unset so the test suite has no hard CI dependency on Postgres. +""" + +from __future__ import annotations + +import os +import uuid +from datetime import date + +import asyncpg +import pytest + +from db.pg_migrate import apply_pg_migrations +from db.pg_store import PgStore + +pytestmark = pytest.mark.skipif( + "PRAXIS_PG_DSN" not in os.environ, + reason="PRAXIS_PG_DSN not set — Postgres integration tests skipped (dev mode).", +) + + +@pytest.fixture(scope="module") +async def pool() -> asyncpg.Pool: + p = await asyncpg.create_pool( + dsn=os.environ["PRAXIS_PG_DSN"], + min_size=1, + max_size=5, + command_timeout=10, + ) + try: + await apply_pg_migrations(p) + yield p + finally: + await p.close() + + +@pytest.fixture(autouse=True) +async def _clean_tables(pool: asyncpg.Pool): + async with pool.acquire() as conn: + await conn.execute( + "TRUNCATE operators, issued_credentials, mastery_gate_events, " + "cohort_aggregates, issuer_keys RESTART IDENTITY CASCADE" + ) + yield + + +@pytest.mark.asyncio +async def test_migration_creates_tables(pool: asyncpg.Pool): + async with pool.acquire() as conn: + tables = await conn.fetch( + "SELECT tablename FROM pg_tables WHERE schemaname = 'public' " + "ORDER BY tablename" + ) + names = {r["tablename"] for r in tables} + assert {"operators", "issued_credentials", "mastery_gate_events", + "cohort_aggregates", "issuer_keys"}.issubset(names) + + +@pytest.mark.asyncio +async def test_migration_idempotent(pool: asyncpg.Pool): + applied = await apply_pg_migrations(pool) + assert applied == [] + + +@pytest.mark.asyncio +async def test_operator_insert_and_lookup(pool: asyncpg.Pool): + store = PgStore(pool) + oid = await store.insert_operator( + "alice", "$argon2id$fakehash", "Alice" + ) + assert oid is not None + op = await store.get_operator_by_username("alice") + assert op is not None + assert op["username"] == "alice" + assert op["display_name"] == "Alice" + assert op["is_active"] is True + by_id = await store.get_operator_by_id(oid) + assert by_id is not None + assert by_id["id"] == op["id"] + + +@pytest.mark.asyncio +async def test_operator_insert_idempotent(pool: asyncpg.Pool): + store = PgStore(pool) + first = await store.insert_operator("bob", "$argon2id$h1", "Bob") + assert first is not None + second = await store.insert_operator("bob", "$argon2id$h2", "Bob") + assert second is None + + +@pytest.mark.asyncio +async def test_operator_on_conflict_update(pool: asyncpg.Pool): + store = PgStore(pool) + await store.insert_operator("carol", "$argon2id$old", "Carol") + updated = await store.insert_operator( + "carol", "$argon2id$new", "Carol", on_conflict_update=True + ) + assert updated is not None + op = await store.get_operator_by_username("carol") + assert op["password_hash"] == "$argon2id$new" + + +@pytest.mark.asyncio +async def test_update_last_login(pool: asyncpg.Pool): + store = PgStore(pool) + oid = await store.insert_operator("dave", "$argon2id$h", "Dave") + assert oid is not None + assert (await store.get_operator_by_id(oid))["last_login_at"] is None + await store.update_last_login(oid) + assert (await store.get_operator_by_id(oid))["last_login_at"] is not None + + +@pytest.mark.asyncio +async def test_cohort_aggregate_upsert_idempotent(pool: asyncpg.Pool): + store = PgStore(pool) + ws, we = date(2026, 8, 1), date(2026, 8, 7) + await store.upsert_cohort_aggregate( + "cs-refund", "sessions_count", ws, we, 42.0, 15, False + ) + await store.upsert_cohort_aggregate( + "cs-refund", "sessions_count", ws, we, 42.0, 15, False + ) + rows = await store.get_cohort_aggregates( + "cs-refund", "sessions_count", date(2026, 7, 1) + ) + assert len(rows) == 1 + assert rows[0]["value"] == 42.0 + assert rows[0]["cell_count"] == 15 + + +@pytest.mark.asyncio +async def test_cohort_aggregate_suppressed_cell(pool: asyncpg.Pool): + store = PgStore(pool) + ws, we = date(2026, 8, 1), date(2026, 8, 7) + await store.upsert_cohort_aggregate( + "cs-refund", "active_learners", ws, we, None, 9, True + ) + rows = await store.get_cohort_aggregates( + "cs-refund", "active_learners", date(2026, 7, 1) + ) + assert len(rows) == 1 + assert rows[0]["cell_suppressed"] is True + assert rows[0]["value"] is None + + +@pytest.mark.asyncio +async def test_issuer_key_init_active_then_superseded(pool: asyncpg.Pool): + store = PgStore(pool) + kid = f"key-{uuid.uuid4().hex[:12]}" + await store.init_issuer_key(kid, "pub-b64-aaa", b"\x01\x02\x03") + active = await store.get_active_signing_key_row() + assert active is not None + assert active["id"] == kid + assert active["status"] == "active" + await store.set_issuer_key_superseded(kid) + assert await store.get_active_signing_key_row() is None + archived = await store.get_public_key_row(kid) + assert archived is not None + assert archived["status"] == "superseded" + assert archived["public_key"] == "pub-b64-aaa" + + +@pytest.mark.asyncio +async def test_get_public_key_row_finds_superseded(pool: asyncpg.Pool): + store = PgStore(pool) + kid = f"key-{uuid.uuid4().hex[:12]}" + await store.init_issuer_key(kid, "pub-b64-bbb", b"\x04\x05") + await store.set_issuer_key_superseded(kid) + row = await store.get_public_key_row(kid) + assert row is not None + assert row["status"] == "superseded" + + +@pytest.mark.asyncio +async def test_credential_insert_and_get(pool: asyncpg.Pool): + store = PgStore(pool) + oid = await store.insert_operator("ed", "$argon2id$h", "Ed") + cid = f"vc-{uuid.uuid4().hex[:16]}" + await store.insert_credential( + cid, "learner-1", '{"id":"vc-x"}', "sig-b64", + operator_id=oid, + ) + row = await store.get_credential(cid) + assert row is not None + assert row["id"] == cid + assert row["learner_ref"] == "learner-1" + assert row["signature_b64"] == "sig-b64" + assert row["status"] == "active" + assert row["vc_payload_json"] == '{"id":"vc-x"}' + + +@pytest.mark.asyncio +async def test_credential_status_revoke(pool: asyncpg.Pool): + store = PgStore(pool) + cid = f"vc-{uuid.uuid4().hex[:16]}" + await store.insert_credential(cid, "learner-2", "{}", "sig") + await store.set_credential_status(cid, "revoked") + row = await store.get_credential(cid) + assert row["status"] == "revoked" + assert row["revoked_at"] is not None + + +@pytest.mark.asyncio +async def test_record_gate_event(pool: asyncpg.Pool): + store = PgStore(pool) + eid = await store.record_gate_event( + "learner-3", "cs-refund", scenario_id="sc-1", + gate_outcome="open", rubric_scores_jsonb=[{"c": "x", "l": 4}], + ) + assert eid is not None + async with pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT * FROM mastery_gate_events WHERE id = $1", eid + ) + assert row is not None + assert row["learner_ref"] == "learner-3" + assert row["gate_outcome"] == "open" + assert row["source"] == "sync" \ No newline at end of file