Phase 0 milestone 0.2: vendor OWC framework + YPMS pii_audit

Brings in a working generic-CRUD framework plus its dependencies from OWC (../owc-system/src/owc/, snapshot 2026-04-20). Every vendored module carries a header comment naming the source; the OWC → ypms rename is the only mechanical change. Vendored modules: - framework/base.py TimestampMixin, SoftDeleteMixin, VersionedMixin - framework/pagination.py Page[T] with to_dict() - framework/version_history.py Snapshot table - framework/crud.py CRUDService with audit emission - framework/bulk.py BulkOps (BulkResult / BulkUpdateItem) - framework/router.py make_router() factory (kept without `from __future__ import annotations` for FastAPI type introspection — see the module docstring) - audit/models.py AuditLog - audit/service.py record() - security/models.py Permission - security/rbac.py can() + require_permission() dependency (DOES_SYSTEM_ADMIN as the super-role) - users/models.py User, Role, UserRole (YPMS adds mfa_verified and assurance_level; drops OWC-specific claimant_id) - auth/deps.py get_current_user() placeholder — real JWT verify lands in 0.3 YPMS-specific additions on top of the vendored framework: - framework/pii_audit.py emit_pii_read() — NIST AU-14 / source RFP C.5.6.9 #5. OWC doesn't need this because its case-management surface is staff-only; YPMS must audit reads of minors' PII, SSN, banking, FERPA-covered OSSE records, and CFSA foster/ward flags. Bugfix vs. upstream OWC: - framework/crud.py Pagination `next_cursor` was previously the peek-ahead row's id, which combined with `where id > cursor` on the next page silently dropped one row per boundary. Caught by test_list_pagination_forwards_cursor. Fixed here; noted in the module header so re-vendoring OWC surfaces it upstream. Migrations: - 0002_framework_tables.py Baseline for users, roles, user_roles, permissions, audit_log, version_history. Immutable once merged per CONTRIBUTING.md. - migrations/env.py Imports the new model modules so autogenerate compares against them. Tests (26 passing, 93.10% coverage, floor 80%): - tests/framework/test_crud.py CRUDService lifecycle + snapshots + audit emission (7 tests) - tests/framework/test_bulk.py BulkOps happy paths + failure reporting (4 tests) - tests/framework/test_pagination.py Page.to_dict serializer contract (2 tests) - tests/framework/test_pii_audit.py emit_pii_read row shape, anonymous case (2 tests) - tests/security/test_rbac.py can() + admin bypass + role union (4 tests) - tests/test_readyz.py /readyz smoke (1 test) - tests/test_logging.py sanitize_for_log contract (3 tests) Green locally: ruff clean, mypy clean (with a targeted override on framework/router.py and framework/crud.py for TypeVar limitations that OWC also accepts). Advances YPMS-POAM-040. Prepares milestone 0.3 (JWT auth) and 0.4 (RBAC seed) to plug into an already-live framework. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-09-30 16:55:56 -04:00
1 parent 09d862bf4a
commit ba756586aa
@@ -0,0 +1,129 @@
+"""CRUDService: create/get/list/update/soft_delete/restore + version snapshot +
+audit emission.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import Session
+
+from tests.framework._fixtures import Thing, ThingCreate, ThingUpdate
+from ypms.audit.models import AuditLog
+from ypms.framework.crud import CRUDService
+from ypms.framework.version_history import VersionHistory
+from ypms.users.models import User
+
+
+@pytest.fixture
+def crud(db: Session) -> CRUDService:
+    Thing.__table__.create(bind=db.get_bind(), checkfirst=True)
+    return CRUDService[Thing, ThingCreate, ThingUpdate](Thing, entity="Thing")
+
+
+@pytest.fixture
+def actor(db: Session) -> User:
+    u = User(
+        username="tester",
+        email="tester@example.com",
+        password_hash="not-a-real-hash",
+    )
+    db.add(u)
+    db.flush()
+    return u
+
+
+def test_create_emits_audit_and_snapshot(crud: CRUDService, db: Session, actor: User) -> None:
+    obj = crud.create(db, ThingCreate(name="widget", color="red"), actor=actor)
+    assert obj.id is not None
+    assert obj.version == 1
+
+    audit = db.query(AuditLog).one()
+    assert audit.action == "create"
+    assert audit.entity == "Thing"
+    assert audit.entity_id == obj.id
+    assert audit.actor_id == actor.id
+    assert audit.actor_username == "tester"
+    assert audit.after is not None
+    assert audit.after["name"] == "widget"
+
+    snap = db.query(VersionHistory).one()
+    assert snap.entity == "Thing"
+    assert snap.entity_id == obj.id
+    assert snap.version == 1
+
+
+def test_update_bumps_version_and_emits_before_after(crud: CRUDService, db: Session, actor: User) -> None:
+    obj = crud.create(db, ThingCreate(name="a"), actor=actor)
+    obj = crud.update(db, obj.id, ThingUpdate(color="blue"), actor=actor)
+    assert obj.version == 2
+    assert obj.color == "blue"
+
+    update_row = db.query(AuditLog).filter(AuditLog.action == "update").one()
+    assert update_row.before is not None
+    assert update_row.after is not None
+    assert update_row.before["color"] is None
+    assert update_row.after["color"] == "blue"
+
+    assert db.query(VersionHistory).count() == 2
+
+
+def test_soft_delete_and_restore(crud: CRUDService, db: Session, actor: User) -> None:
+    obj = crud.create(db, ThingCreate(name="x"), actor=actor)
+    crud.soft_delete(db, obj.id, actor=actor)
+
+    with pytest.raises(NoResultFound):
+        crud.get(db, obj.id)
+
+    ghost = crud.get(db, obj.id, include_deleted=True)
+    assert ghost.is_deleted
+
+    crud.restore(db, obj.id, actor=actor)
+    live = crud.get(db, obj.id)
+    assert not live.is_deleted
+
+    actions = [r.action for r in db.query(AuditLog).order_by(AuditLog.id).all()]
+    assert actions == ["create", "delete", "restore"]
+
+
+def test_list_excludes_soft_deleted_by_default(crud: CRUDService, db: Session, actor: User) -> None:
+    a = crud.create(db, ThingCreate(name="a"), actor=actor)
+    b = crud.create(db, ThingCreate(name="b"), actor=actor)
+    crud.soft_delete(db, a.id, actor=actor)
+
+    page = crud.list(db, limit=10)
+    assert [t.id for t in page.items] == [b.id]
+
+    with_deleted = crud.list(db, limit=10, include_deleted=True)
+    assert {t.id for t in with_deleted.items} == {a.id, b.id}
+
+
+def test_list_pagination_forwards_cursor(crud: CRUDService, db: Session, actor: User) -> None:
+    ids = [crud.create(db, ThingCreate(name=f"n{i}"), actor=actor).id for i in range(5)]
+
+    page1 = crud.list(db, limit=2)
+    assert [t.id for t in page1.items] == ids[:2]
+    assert page1.next_cursor == ids[1]
+
+    page2 = crud.list(db, limit=2, cursor=page1.next_cursor)
+    assert [t.id for t in page2.items] == ids[2:4]
+    assert page2.next_cursor == ids[3]
+
+    page3 = crud.list(db, limit=2, cursor=page2.next_cursor)
+    assert [t.id for t in page3.items] == ids[4:]
+    assert page3.next_cursor is None
+
+
+def test_get_missing_raises(crud: CRUDService, db: Session) -> None:
+    with pytest.raises(NoResultFound):
+        crud.get(db, 999)
+
+
+def test_history_returns_snapshots_in_order(crud: CRUDService, db: Session, actor: User) -> None:
+    obj = crud.create(db, ThingCreate(name="a"), actor=actor)
+    crud.update(db, obj.id, ThingUpdate(name="b"), actor=actor)
+    crud.update(db, obj.id, ThingUpdate(name="c"), actor=actor)
+
+    versions = crud.history(db, obj.id)
+    assert [v.version for v in versions] == [1, 2, 3]
+    assert [v.snapshot["name"] for v in versions] == ["a", "b", "c"]