Phase 0 milestone 0.6: 501-stub routers for remaining entities

26 stub entities (every matrix entity except Registrant, User, Role,
AuditLog) get a router from framework/stub_router.py: one route per
matrix action (List, Search, Create, View, Edit, Delete, Restore,
VersionHistory). Each route runs the real RBAC check, writes an audit
row, then returns 501. Source RFP C.5.7.27.

- Registry in src/ypms/stubs.py; test fails if a matrix entity is
  neither real nor stubbed.
- tests/test_stubs.py sweeps entity x role x route against the matrix
  JSON: 501 + exactly one audit row when granted, 403 + none when not.
  FERPA/CFSA stubs pinned closed to non-examiner roles.
- Registrant restore now gated on the matrix's Restore action (was
  Edit, the vendored-factory convention); only the super-role holds it.
- Test fixtures (roles, make_user, api) lifted to conftest; one app per
  session (route construction dominated runtime) and a shared password
  hash for minted-token users.

138 tests passing, ruff + mypy clean.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chris TonjesandClaude Opus 5.5 committed 2026-09-30 20:54:12 -04:00
1 parent e5720440fb
commit 98b783b6b5
8 files changed
+381 -58

No files matched your search

+80 -1
View File
@@ -12,7 +12,8 @@ imports `ypms.main` (which instantiates the FastAPI app at import time).
from __future__ import annotations
import os
from collections.abc import Iterator
from collections.abc import Callable, Iterator
from typing import Any
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("JWT_SECRET", "test-secret")
@@ -70,3 +71,81 @@ def client() -> Iterator[TestClient]:
app = create_app()
with TestClient(app) as c:
yield c
# ---------------- seeded RBAC + authenticated API client ----------------
Headers = dict[str, str]
@pytest.fixture
def roles(db: Session) -> dict[str, Any]:
"""All seed roles with the real permissions matrix loaded."""
from ypms.seed.permissions import seed_permissions
from ypms.seed.run import seed_roles
r = seed_roles(db)
seed_permissions(db, r)
db.flush()
return r
_HASH_CACHE: list[str] = []
def _unused_password_hash(hasher: Callable[[str], str]) -> str:
# PBKDF2 is deliberately slow; these users authenticate by minted token,
# so one hash shared across the session is enough.
if not _HASH_CACHE:
_HASH_CACHE.append(hasher("unused-in-these-tests"))
return _HASH_CACHE[0]
@pytest.fixture
def make_user(db: Session, roles: dict[str, Any]) -> Callable[..., tuple[Any, Headers]]:
"""Factory: make_user("name", "ROLE_A", ...) -> (User, auth headers)."""
from ypms.auth.security import create_access_token, hash_password
from ypms.users.models import ASSURANCE_STANDARD, User, UserRole
def _make(username: str, *role_codes: str) -> tuple[User, Headers]:
user = User(
username=username,
password_hash=_unused_password_hash(hash_password),
is_active=True,
assurance_level=ASSURANCE_STANDARD,
)
db.add(user)
db.flush()
for code in role_codes:
db.add(UserRole(user_id=user.id, role_id=roles[code].id))
db.flush()
db.refresh(user)
token = create_access_token(
sub=username, uid=user.id, roles=list(role_codes),
assurance_level=ASSURANCE_STANDARD, mfa_verified=False,
)
return user, {"Authorization": f"Bearer {token}"}
return _make
@pytest.fixture(scope="session")
def _shared_app() -> Any:
"""One app per session: building ~240 routes costs ~0.1s per create_app()."""
from ypms.main import create_app
return create_app()
@pytest.fixture
def api(db: Session, _shared_app: Any) -> Iterator[TestClient]:
"""TestClient on the shared app, wired to the per-test SQLite session."""
from ypms.db import get_db
def _override() -> Iterator[Session]:
yield db
_shared_app.dependency_overrides[get_db] = _override
with TestClient(_shared_app) as c:
yield c
_shared_app.dependency_overrides.clear()
+104
View File
@@ -0,0 +1,104 @@
"""Milestone 0.6 — every stub route enforces RBAC and audits before its 501.
The permissions matrix JSON is the oracle: for every stub entity, every seeded
role, and every stub route, the expected status is 501 if the role holds the
action (or is the super-role) and 403 otherwise. Each 501 must leave exactly
one audit row; each 403 must leave none.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.orm import Session
from ypms.audit.models import AuditLog
from ypms.framework.stub_router import STUB_ROUTES, STUB_STATUS
from ypms.security.rbac import ADMIN_ROLE
from ypms.seed.permissions import load_matrix
from ypms.seed.roles import ROLES
from ypms.stubs import STUB_ENTITIES
MATRIX = load_matrix()
REAL_ENTITIES = {"Registrant", "User", "Role", "AuditLog"}
STUB_ID = 7
def _url(prefix: str, path: str) -> str:
return prefix + path.replace("{id}", str(STUB_ID))
def test_every_matrix_entity_is_real_or_stubbed() -> None:
assert set(STUB_ENTITIES) == set(MATRIX["entities"]) - REAL_ENTITIES
def test_stub_routes_cover_every_matrix_action() -> None:
assert {action for _, _, action in STUB_ROUTES} == set(MATRIX["actions"])
def test_stub_paths_are_unique() -> None:
assert len(set(STUB_ENTITIES.values())) == len(STUB_ENTITIES)
def test_stub_requires_authentication(api: TestClient) -> None:
assert api.get("/guardians").status_code == 401
@pytest.mark.parametrize("entity", sorted(STUB_ENTITIES))
def test_stub_rbac_and_audit_follow_matrix(
entity: str,
api: TestClient,
db: Session,
make_user: Callable[..., tuple[Any, dict[str, str]]],
) -> None:
prefix = STUB_ENTITIES[entity]
for role in (r["code"] for r in ROLES):
_, headers = make_user(f"u.{role.lower()}", role)
granted = set(MATRIX["grants"].get(role, {}).get(entity, []))
for method, path, action in STUB_ROUTES:
allowed = role == ADMIN_ROLE or action in granted
before = db.execute(select(AuditLog.id)).all()
r = api.request(method, _url(prefix, path), headers=headers)
assert r.status_code == (501 if allowed else 403), (role, method, path)
new = list(
db.execute(select(AuditLog).where(AuditLog.id.notin_([b[0] for b in before]))).scalars()
)
if not allowed:
assert new == [], (role, action)
continue
body = r.json()
assert body["status"] == "not_implemented"
assert (body["entity"], body["action"]) == (entity, action)
(row,) = new
assert (row.entity, row.action) == (entity, action)
assert row.actor_username == f"u.{role.lower()}"
assert row.after == {"status": STUB_STATUS}
assert row.entity_id == (STUB_ID if "{id}" in path else None)
def test_search_not_captured_by_id_route(
api: TestClient, make_user: Callable[..., tuple[Any, dict[str, str]]]
) -> None:
_, admin = make_user("admin", ADMIN_ROLE)
assert api.get("/programs/search", headers=admin).json()["action"] == "Search"
@pytest.mark.parametrize("entity", ["EducationRecord", "FosterIndicator"])
@pytest.mark.parametrize("role", ["OYP_STAFF", "PAYROLL_ADMIN", "REPORTING_ANALYST", "HOST_USER"])
def test_ferpa_and_cfsa_stubs_closed_to_non_examiners(
entity: str,
role: str,
api: TestClient,
make_user: Callable[..., tuple[Any, dict[str, str]]],
) -> None:
"""Pinned separately from the matrix sweep: ARCH §6.2 guardrail."""
_, headers = make_user("u", role)
prefix = STUB_ENTITIES[entity]
for method, path, _ in STUB_ROUTES:
assert api.request(method, _url(prefix, path), headers=headers).status_code == 403
+23 -53
View File
@@ -6,7 +6,7 @@ grants), so these tests also pin the matrix's Registrant rows.
from __future__ import annotations
from collections.abc import Callable, Iterator
from collections.abc import Callable
from datetime import date
from typing import Any
@@ -16,15 +16,10 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from ypms.audit.models import AuditLog
from ypms.auth.security import create_access_token, hash_password
from ypms.db import get_db
from ypms.framework.crud import REDACTED
from ypms.framework.pii_audit import PII_READ_ACTION
from ypms.framework.version_history import VersionHistory
from ypms.main import create_app
from ypms.seed.permissions import seed_permissions
from ypms.seed.run import seed_roles
from ypms.users.models import ASSURANCE_STANDARD, Role, User, UserRole
from ypms.users.models import User
from ypms.youth.models import Registrant
SSN = "123-45-6789"
@@ -42,56 +37,11 @@ PAYLOAD: dict[str, Any] = {
Headers = dict[str, str]
@pytest.fixture
def roles(db: Session) -> dict[str, Role]:
r = seed_roles(db)
seed_permissions(db, r)
db.flush()
return r
@pytest.fixture
def make_user(db: Session, roles: dict[str, Role]) -> Callable[..., tuple[User, Headers]]:
def _make(username: str, *role_codes: str) -> tuple[User, Headers]:
user = User(
username=username,
password_hash=hash_password("unused-in-these-tests"),
is_active=True,
assurance_level=ASSURANCE_STANDARD,
)
db.add(user)
db.flush()
for code in role_codes:
db.add(UserRole(user_id=user.id, role_id=roles[code].id))
db.flush()
db.refresh(user)
token = create_access_token(
sub=username, uid=user.id, roles=list(role_codes),
assurance_level=ASSURANCE_STANDARD, mfa_verified=False,
)
return user, {"Authorization": f"Bearer {token}"}
return _make
@pytest.fixture
def staff(make_user: Callable[..., tuple[User, Headers]]) -> tuple[User, Headers]:
return make_user("oyp.staff", "OYP_STAFF")
@pytest.fixture
def api(db: Session) -> Iterator[TestClient]:
app = create_app()
def _override() -> Iterator[Session]:
yield db
app.dependency_overrides[get_db] = _override
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
def _create(api: TestClient, headers: Headers, **overrides: Any) -> dict[str, Any]:
r = api.post("/registrants", json={**PAYLOAD, **overrides}, headers=headers)
assert r.status_code == 201, r.text
@@ -256,7 +206,9 @@ def test_delete_requires_delete_permission_then_restore(
assert r.status_code == 200 and r.json()["deleted_at"] is not None
assert api.get(f"/registrants/{rid}", headers=staff[1]).status_code == 404
r = api.post(f"/registrants/{rid}/restore", headers=staff[1])
# Restore is its own matrix action; only the super-role holds it today.
assert api.post(f"/registrants/{rid}/restore", headers=staff[1]).status_code == 403
r = api.post(f"/registrants/{rid}/restore", headers=admin)
assert r.status_code == 200 and r.json()["deleted_at"] is None
assert [a.action for a in db.execute(select(AuditLog).order_by(AuditLog.id)).scalars()][-2:] == [
"delete",
@@ -365,3 +317,21 @@ def test_host_user_denied(
_, hh = make_user("host", "HOST_USER")
rid = _create(api, staff[1])["id"]
assert api.get(f"/registrants/{rid}", headers=hh).status_code == 403
def test_staff_guard_holds_even_if_matrix_grants_portal_role_list(
api: TestClient,
db: Session,
roles: dict[str, Any],
make_user: Callable[..., tuple[User, Headers]],
) -> None:
"""Defense in depth: a future matrix edit granting YOUTH_USER List must not
expose every registrant."""
from ypms.security.models import Permission
db.add(Permission(role_id=roles["YOUTH_USER"].id, entity="Registrant", action="List"))
db.flush()
_, yh = make_user("youth", "YOUTH_USER")
r = api.get("/registrants", headers=yh)
assert r.status_code == 403
assert r.json()["detail"]["reason"] == "staff-only route"