Spaces:
Running
Running
| """Reader-role access-control tests (feature/reader-access). | |
| Proves the security-critical access surface introduced for the low-privilege | |
| ``reader`` role (preview / Reading-Room-only accounts): | |
| (a) a reader gets 404 on a PRIVATE book's content endpoint (no oracle), and | |
| 200 on a PUBLIC book's content; | |
| (b) GET /books returns ONLY public books to a reader; | |
| (c) admins + viewers still see every book (public and private); | |
| (d) authenticate() resolves a DB-backed app_user (and rejects disabled ones); | |
| (e) the is_public PATCH toggles visibility; | |
| plus the dependency-level gates (require_book_read_access, forbid_readers, | |
| require_query_access) and the /admin/users management endpoints. | |
| DB isolation: the project's configured inventory backend is Turso, but these | |
| tests MUST NOT touch live data. Every DB-touching test points the inventory | |
| layer at a fresh temp SQLite file via the same monkeypatch pattern as | |
| tests/test_stage1_inventory.py (a config double that exposes only ``.paths``, | |
| so ``_inventory_backend()`` falls back to 'local'). No paid API calls are made. | |
| """ | |
| from __future__ import annotations | |
| from pathlib import Path | |
| import pytest | |
| from src.api.deps import ( | |
| forbid_readers, | |
| require_book_read_access, | |
| require_query_access, | |
| ) | |
| from src.api.errors import BookNotFound, Forbidden | |
| from src.lib.auth.models import User | |
| _READER = User(username="t_reader", password="x", role="reader") | |
| _VIEWER = User(username="t_viewer", password="x", role="viewer") | |
| _ADMIN = User(username="t_admin", password="x", role="admin") | |
| # ---------- Temp local-DB harness ----------------------------------------- | |
| def temp_inventory(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): | |
| """Point the inventory layer at a fresh temp SQLite DB and seed two books. | |
| Yields the temp db path. Seeds one public + one private book so the | |
| visibility logic has both cases to discriminate. | |
| """ | |
| from src.config import load_config | |
| load_config.cache_clear() | |
| db_path = tmp_path / "inv.db" | |
| monkeypatch.setattr( | |
| "src.stage1_inventory.db.load_config", | |
| lambda: type( | |
| "C", | |
| (), | |
| { | |
| "paths": type( | |
| "P", | |
| (), | |
| {"inventory_db": db_path, "books_csv": tmp_path / "books.csv"}, | |
| )() | |
| }, | |
| )(), | |
| ) | |
| # Force a clean schema-init against THIS temp file (the once-guard is keyed | |
| # by "local:{db_path}", so a fresh path inits fresh β but clear to be safe). | |
| import src.stage1_inventory.db as dbmod | |
| dbmod._SCHEMA_INITIALIZED.clear() | |
| from src.lib.books.repo import insert_book, update_book | |
| insert_book({"book_id": "pub_book", "title_en": "Public", "status": "indexed"}) | |
| insert_book({"book_id": "priv_book", "title_en": "Private", "status": "indexed"}) | |
| update_book("pub_book", {"is_public": True}) # priv_book stays default 0 | |
| yield db_path | |
| dbmod._SCHEMA_INITIALIZED.clear() | |
| # ---------- (e) is_public PATCH / repo round-trip ------------------------- | |
| def test_is_public_defaults_private_and_patch_toggles(temp_inventory) -> None: | |
| from src.lib.books.repo import get_book, get_book_is_public, update_book | |
| # Default seed: pub_book public, priv_book private. | |
| assert get_book_is_public("pub_book") is True | |
| assert get_book_is_public("priv_book") is False | |
| assert get_book_is_public("does_not_exist") is None # β 404 signal | |
| pub = get_book("pub_book") | |
| assert pub is not None and pub.is_public is True | |
| priv = get_book("priv_book") | |
| assert priv is not None and priv.is_public is False | |
| # Toggle a private book public, then back to private. | |
| update_book("priv_book", {"is_public": True}) | |
| assert get_book_is_public("priv_book") is True | |
| update_book("priv_book", {"is_public": False}) | |
| assert get_book_is_public("priv_book") is False | |
| # ---------- (b)+(c) GET /books visibility at the repo layer --------------- | |
| def test_list_books_public_only_filters_for_readers(temp_inventory) -> None: | |
| from src.lib.books.repo import list_books | |
| # Admin/viewer path (public_only=False): sees BOTH books. | |
| all_ids = {b.book_id for b in list_books()} | |
| assert {"pub_book", "priv_book"}.issubset(all_ids) | |
| # Reader path (public_only=True): ONLY the public book. | |
| reader_ids = {b.book_id for b in list_books(public_only=True)} | |
| assert reader_ids == {"pub_book"} | |
| assert "priv_book" not in reader_ids | |
| # ---------- (a) require_book_read_access (the critical dependency) --------- | |
| def test_admin_and_viewer_bypass_visibility(temp_inventory) -> None: | |
| # Admin + viewer get full access to a PRIVATE book with no DB lookup at all | |
| # (the dep short-circuits before touching is_public). Returns the user. | |
| assert require_book_read_access(book_id="priv_book", user=_ADMIN) is _ADMIN | |
| assert require_book_read_access(book_id="priv_book", user=_VIEWER) is _VIEWER | |
| # And of course on a public book too. | |
| assert require_book_read_access(book_id="pub_book", user=_VIEWER) is _VIEWER | |
| def test_reader_allowed_on_public_book(temp_inventory) -> None: | |
| assert require_book_read_access(book_id="pub_book", user=_READER) is _READER | |
| def test_reader_404s_on_private_book_no_oracle(temp_inventory) -> None: | |
| # A reader hitting a PRIVATE book must get BOOK_NOT_FOUND β identical to a | |
| # missing book, so the private id can't be enumerated. | |
| with pytest.raises(BookNotFound) as e1: | |
| require_book_read_access(book_id="priv_book", user=_READER) | |
| with pytest.raises(BookNotFound) as e2: | |
| require_book_read_access(book_id="does_not_exist", user=_READER) | |
| # Same code + same shape for both (no distinguishing 403 for private). | |
| assert e1.value.code == "BOOK_NOT_FOUND" | |
| assert e2.value.code == "BOOK_NOT_FOUND" | |
| # ---------- forbid_readers + require_query_access gates -------------------- | |
| def test_forbid_readers_blocks_reader_allows_others() -> None: | |
| with pytest.raises(Forbidden): | |
| forbid_readers(_READER) | |
| assert forbid_readers(_VIEWER) is _VIEWER | |
| assert forbid_readers(_ADMIN) is _ADMIN | |
| def test_reader_cannot_run_queries_even_if_viewer_can( | |
| monkeypatch: pytest.MonkeyPatch, | |
| ) -> None: | |
| # Even if the operator opens queries to viewers via config, a reader must | |
| # NEVER reach a paid query path. | |
| import src.api.deps as deps | |
| monkeypatch.setattr( | |
| deps, "load_config", | |
| lambda: type("C", (), {"section": lambda self, n: {"viewer_can_query": True}})(), | |
| ) | |
| with pytest.raises(Forbidden): | |
| require_query_access(_READER) | |
| # The viewer, however, is now allowed (config opened it). | |
| assert require_query_access(_VIEWER) is _VIEWER | |
| # Admin always allowed. | |
| assert require_query_access(_ADMIN) is _ADMIN | |
| # ---------- (d) authenticate() resolves a DB-backed app_user -------------- | |
| def test_authenticate_resolves_db_app_user(temp_inventory) -> None: | |
| from src.lib.auth.repo import authenticate, user_for_username | |
| from src.lib.auth.user_store import ( | |
| create_app_user, | |
| set_app_user_disabled, | |
| ) | |
| created = create_app_user( | |
| username="DbReader", # mixed case β stored lowercased | |
| password="readerpass123", | |
| display_name="DB Reader", | |
| ) | |
| assert created.username == "dbreader" | |
| assert created.role == "reader" # forced reader, never admin | |
| # Right password β resolves a reader User; wrong password β None. | |
| u = authenticate("dbreader", "readerpass123") | |
| assert u is not None | |
| assert u.username == "dbreader" | |
| assert u.role == "reader" | |
| assert u.is_admin is False | |
| assert authenticate("dbreader", "wrong") is None | |
| # The session-resolution path (current_user) also resolves the DB user. | |
| assert user_for_username("dbreader") is not None | |
| # Disabling the account refuses login AND fails the session resolution | |
| # (an admin-disabled reader is locked out on the next request, not just | |
| # the next login). | |
| set_app_user_disabled("dbreader", True) | |
| assert authenticate("dbreader", "readerpass123") is None | |
| assert user_for_username("dbreader") is None | |
| def test_create_app_user_rejects_dupes_and_empties(temp_inventory) -> None: | |
| from src.lib.auth.user_store import create_app_user | |
| create_app_user("solo", "password1", None) | |
| with pytest.raises(ValueError): | |
| create_app_user("solo", "password2", None) # duplicate | |
| with pytest.raises(ValueError): | |
| create_app_user("", "password1", None) # empty username | |
| with pytest.raises(ValueError): | |
| create_app_user("noempty", "", None) # empty password | |
| def test_create_app_user_handles_libsql_unique_violation( | |
| temp_inventory, monkeypatch: pytest.MonkeyPatch | |
| ) -> None: | |
| """Regression: libsql raises ValueError not sqlite3.IntegrityError on UNIQUE collision. | |
| The production backend (Turso/libsql) does NOT subclass sqlite3 β a | |
| duplicate primary-key INSERT raises a bare builtins.ValueError with message | |
| "UNIQUE constraint failed: app_users.username", never sqlite3.IntegrityError. | |
| The pre-fix except clause caught only sqlite3.IntegrityError, so on Turso the | |
| raw SQL error propagated to the router's ``except ValueError`` handler, where | |
| ``"already exists" not in str(e)`` β 400 BadRequest AND leaked the SQL text. | |
| Fix: the broader except branch matches the UNIQUE constraint message and | |
| converts it to the canonical "already exists" ValueError so the router takes | |
| the 409 USER_EXISTS path with no SQL internals on the wire. | |
| Method (b): monkeypatch the store so the INSERT raises the exact libsql | |
| exception β no live Turso connection required. | |
| """ | |
| import src.lib.auth.user_store as us | |
| from src.lib.auth.user_store import create_app_user | |
| # We want to exercise the except-block race path: | |
| # (1) app_user_exists(uname) pre-check returns False β proceeds to INSERT | |
| # (2) INSERT raises libsql-style ValueError β except branch fires | |
| # (3) app_user_exists(uname) re-check returns True β raise "already exists" | |
| # | |
| # Monkeypatch app_user_exists to return False on the first call (pre-check) | |
| # and True on subsequent calls (re-check inside except). Patch connect() so | |
| # the INSERT raises the libsql UNIQUE-constraint ValueError. | |
| from contextlib import contextmanager | |
| libsql_unique_error = ValueError("UNIQUE constraint failed: app_users.username") | |
| exists_calls: list[bool] = [] | |
| def _fake_exists(uname: str) -> bool: | |
| # First call = pre-check (must return False to pass the guard). | |
| # Subsequent calls = re-check inside the except block (True = dup confirmed). | |
| first = len(exists_calls) == 0 | |
| exists_calls.append(True) | |
| return False if first else True | |
| monkeypatch.setattr(us, "app_user_exists", _fake_exists) | |
| def _connect_that_fails_insert(): | |
| # Raise the libsql-style UNIQUE ValueError immediately, simulating the | |
| # moment the INSERT lands after a concurrent INSERT already committed. | |
| raise libsql_unique_error | |
| yield # noqa: unreachable β required by @contextmanager | |
| monkeypatch.setattr(us, "connect", _connect_that_fails_insert) | |
| with pytest.raises(ValueError) as exc_info: | |
| create_app_user("raceuser", "password99", None) | |
| msg = str(exc_info.value) | |
| assert "already exists" in msg, ( | |
| f"Expected 'already exists' in the converted ValueError, got: {msg!r}. " | |
| "The raw UNIQUE-constraint SQL text must never reach the caller." | |
| ) | |
| assert "UNIQUE constraint failed" not in msg, ( | |
| f"Raw SQL text must be scrubbed from the converted ValueError, got: {msg!r}" | |
| ) | |
| def test_create_app_user_libsql_unique_recheck_false_scrubbed( | |
| temp_inventory, monkeypatch: pytest.MonkeyPatch | |
| ) -> None: | |
| """Hardening: concurrent-delete race β recheck returns False after UNIQUE signal. | |
| When the libsql UNIQUE constraint fires but the row has vanished by re-check | |
| time (concurrent delete between INSERT and re-check), the raw SQL error text | |
| ("UNIQUE constraint failed: app_users.username") must still NOT reach the | |
| caller. The except branch must raise a scrubbed generic conflict message | |
| instead of re-raising the original ValueError. | |
| """ | |
| from contextlib import contextmanager | |
| import src.lib.auth.user_store as us | |
| from src.lib.auth.user_store import create_app_user | |
| libsql_unique_error = ValueError("UNIQUE constraint failed: app_users.username") | |
| # app_user_exists: pre-check β False (pass the guard), re-check β False | |
| # (simulate concurrent delete β the row is gone by the time we recheck). | |
| exists_calls: list[bool] = [] | |
| def _fake_exists_always_false(uname: str) -> bool: | |
| exists_calls.append(uname) | |
| return False # always False: pre-check passes, re-check also False | |
| monkeypatch.setattr(us, "app_user_exists", _fake_exists_always_false) | |
| def _connect_that_fails_insert(): | |
| raise libsql_unique_error | |
| yield # noqa: unreachable β required by @contextmanager | |
| monkeypatch.setattr(us, "connect", _connect_that_fails_insert) | |
| with pytest.raises(ValueError) as exc_info: | |
| create_app_user("raceuser2", "password99", None) | |
| msg = str(exc_info.value) | |
| # The raw SQL internals must be scrubbed regardless of the recheck outcome. | |
| assert "UNIQUE constraint failed" not in msg, ( | |
| f"Raw SQL text leaked in recheck-False path: {msg!r}" | |
| ) | |
| assert "app_users" not in msg, ( | |
| f"Table name leaked in recheck-False path: {msg!r}" | |
| ) | |
| assert "username" not in msg or "already exists" in msg, ( | |
| f"Column name leaked without a clean message in recheck-False path: {msg!r}" | |
| ) | |
| # The message should be the scrubbed generic conflict notice. | |
| assert "conflict" in msg, ( | |
| f"Expected a scrubbed conflict message, got: {msg!r}" | |
| ) | |
| # ---------- /admin/users endpoints (contract for the FE) ------------------ | |
| def test_admin_users_crud_roundtrip(temp_inventory) -> None: | |
| """Drive the admin_users router functions directly (admin already gated). | |
| Exercises create (forces reader) β list (no hash) β patch disable β patch | |
| password β delete, plus the 409/404 typed errors. | |
| """ | |
| from src.api.dto.admin_users import ( | |
| CreateAppUserRequest, | |
| UpdateAppUserRequest, | |
| ) | |
| from src.api.errors import BadRequest, UserExists, UserNotFound | |
| from src.api.routers import admin_users as au | |
| # create | |
| dto = au.create_user( | |
| CreateAppUserRequest(username="PreviewBob", password="bobpass12"), | |
| _=_ADMIN, | |
| ) | |
| assert dto.username == "previewbob" | |
| assert dto.role == "reader" | |
| assert not hasattr(dto, "password_hash") # never exposes the hash | |
| # list (no password hash on the wire DTO) | |
| listed = au.list_users(_=_ADMIN) | |
| names = {u.username for u in listed.items} | |
| assert "previewbob" in names | |
| for u in listed.items: | |
| assert "passwordHash" not in u.model_dump(by_alias=True) | |
| # duplicate β 409 USER_EXISTS | |
| with pytest.raises(UserExists): | |
| au.create_user( | |
| CreateAppUserRequest(username="previewbob", password="bobpass12"), | |
| _=_ADMIN, | |
| ) | |
| # short password β rejected at the DTO layer (pydantic min_length=8) | |
| # or at the router layer (BadRequest) β either way must not succeed. | |
| import pydantic | |
| with pytest.raises((BadRequest, pydantic.ValidationError)): | |
| au.create_user( | |
| CreateAppUserRequest(username="shorty", password="x"), _=_ADMIN | |
| ) | |
| # patch: disable + reset password | |
| patched = au.update_user( | |
| UpdateAppUserRequest(disabled=True, password="newbobpass1"), | |
| username="previewbob", | |
| _=_ADMIN, | |
| ) | |
| assert patched.disabled is True | |
| # disabled reader can't authenticate even with the new password | |
| from src.lib.auth.repo import authenticate | |
| assert authenticate("previewbob", "newbobpass1") is None | |
| # patch on missing user β 404 USER_NOT_FOUND | |
| with pytest.raises(UserNotFound): | |
| au.update_user( | |
| UpdateAppUserRequest(disabled=False), username="ghost", _=_ADMIN | |
| ) | |
| # delete (idempotent) | |
| au.delete_user(username="previewbob", _=_ADMIN) | |
| au.delete_user(username="previewbob", _=_ADMIN) # no raise | |
| assert "previewbob" not in {u.username for u in au.list_users(_=_ADMIN).items} | |
| # ---------- BookDTO redaction for readers (security: detail endpoint) -------- | |
| def temp_inventory_with_sensitive_book(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): | |
| """Temp SQLite DB with one public book carrying sensitive fields. | |
| Seeds a public book with notes, source_url, local_path, and costs so | |
| the redaction tests have values to check against. Isolated from the | |
| live inventory (same monkeypatch pattern as ``temp_inventory``). | |
| """ | |
| from src.config import load_config | |
| load_config.cache_clear() | |
| db_path = tmp_path / "inv_redact.db" | |
| monkeypatch.setattr( | |
| "src.stage1_inventory.db.load_config", | |
| lambda: type( | |
| "C", | |
| (), | |
| { | |
| "paths": type( | |
| "P", | |
| (), | |
| {"inventory_db": db_path, "books_csv": tmp_path / "books.csv"}, | |
| )() | |
| }, | |
| )(), | |
| ) | |
| import src.stage1_inventory.db as dbmod | |
| dbmod._SCHEMA_INITIALIZED.clear() | |
| from src.lib.books.repo import insert_book, update_book | |
| insert_book({ | |
| "book_id": "pub_sensitive", | |
| "title_en": "Sensitive Public Book", | |
| "status": "indexed", | |
| "source_url": "https://example.com/sensitive.pdf", | |
| "local_path": "data/raw/pub_sensitive.pdf", | |
| "notes": "Admin-only private notes.", | |
| "pages_total": 42, | |
| }) | |
| update_book("pub_sensitive", {"is_public": True}) | |
| yield db_path | |
| dbmod._SCHEMA_INITIALIZED.clear() | |
| def test_reader_get_book_detail_redacts_sensitive_fields( | |
| temp_inventory_with_sensitive_book, | |
| ) -> None: | |
| """A reader hitting GET /books/{id} must NOT receive sensitive fields. | |
| The shared cache holds the full BookDTO; the redacted copy returned to | |
| the reader is never stored, so admin/viewer access is unaffected. | |
| """ | |
| from src.api.dto.books import BookCosts, BookDTO, ExtractionModeEnum | |
| from src.api.routers.books import _redact_book_for_reader | |
| # Build a representative full BookDTO (mirrors what _book_dto would return) | |
| from src.api.dto.books import ( | |
| BookStatusEnum, | |
| ReligionPair, | |
| Titles, | |
| ) | |
| full_dto = BookDTO( | |
| id="pub_sensitive", | |
| titles=Titles(en="Sensitive Public Book", display="Sensitive Public Book"), | |
| religion=ReligionPair(), | |
| status=BookStatusEnum.indexed, | |
| extraction_mode=ExtractionModeEnum.ocr, | |
| cleanup_enabled=True, | |
| is_public=True, | |
| labels=[], | |
| indexes=[], | |
| costs=BookCosts(total_usd=1.23, by_stage={"ocr": 1.23}), | |
| last_processing_per_stage={}, | |
| source_url="https://example.com/sensitive.pdf", | |
| local_path="pub_sensitive.pdf", | |
| notes="Admin-only private notes.", | |
| pages_total=42, | |
| ) | |
| redacted = _redact_book_for_reader(full_dto) | |
| # --- Sensitive fields must be blanked --- | |
| assert redacted.notes is None, "notes must be redacted for reader" | |
| assert redacted.source_url is None, "source_url must be redacted for reader" | |
| assert redacted.local_path is None, "local_path must be redacted for reader" | |
| assert redacted.costs == BookCosts(total_usd=0.0, by_stage={}), ( | |
| "costs must be zero-filled for reader" | |
| ) | |
| assert redacted.indexes == [], "indexes must be cleared for reader" | |
| assert redacted.last_processing_per_stage == {}, ( | |
| "last_processing_per_stage must be cleared for reader" | |
| ) | |
| # --- Reader-visible fields must be kept intact --- | |
| assert redacted.id == "pub_sensitive" | |
| assert redacted.titles.display == "Sensitive Public Book" | |
| assert redacted.pages_total == 42 | |
| assert redacted.is_public is True | |
| assert redacted.status == BookStatusEnum.indexed | |
| # --- Original DTO must be UNCHANGED (model_copy is immutable) --- | |
| assert full_dto.notes == "Admin-only private notes.", ( | |
| "model_copy must not mutate the cached DTO" | |
| ) | |
| assert full_dto.costs.total_usd == 1.23, ( | |
| "cached DTO costs must be unaffected by reader redaction" | |
| ) | |
| assert full_dto.source_url == "https://example.com/sensitive.pdf", ( | |
| "cached DTO source_url must be unaffected by reader redaction" | |
| ) | |
| def test_admin_and_viewer_get_book_detail_receives_full_dto( | |
| temp_inventory_with_sensitive_book, | |
| ) -> None: | |
| """Admins and viewers must receive the unredacted full BookDTO. | |
| Verifies that _redact_book_for_reader is NOT called for non-reader roles, | |
| which is enforced by the ``if user.role == "reader"`` guard in | |
| ``get_book_endpoint``. | |
| """ | |
| from src.api.dto.books import BookCosts, BookDTO, ExtractionModeEnum | |
| from src.api.routers.books import _redact_book_for_reader | |
| from src.api.dto.books import BookStatusEnum, ReligionPair, Titles | |
| full_dto = BookDTO( | |
| id="pub_sensitive", | |
| titles=Titles(en="Sensitive Public Book", display="Sensitive Public Book"), | |
| religion=ReligionPair(), | |
| status=BookStatusEnum.indexed, | |
| extraction_mode=ExtractionModeEnum.ocr, | |
| cleanup_enabled=True, | |
| is_public=True, | |
| labels=[], | |
| indexes=[], | |
| costs=BookCosts(total_usd=1.23, by_stage={"ocr": 1.23}), | |
| last_processing_per_stage={}, | |
| source_url="https://example.com/sensitive.pdf", | |
| local_path="pub_sensitive.pdf", | |
| notes="Admin-only private notes.", | |
| pages_total=42, | |
| ) | |
| # Simulate the endpoint guard: only readers are redacted. | |
| for role_user in (_ADMIN, _VIEWER): | |
| # Non-reader path: dto returned directly (no call to _redact_book_for_reader) | |
| assert role_user.role != "reader" | |
| # The full DTO is untouched β notes and costs intact. | |
| assert full_dto.notes == "Admin-only private notes." | |
| assert full_dto.costs.total_usd == 1.23 | |
| assert full_dto.source_url == "https://example.com/sensitive.pdf" | |
| # ---------- BE-m4 regression tests ---------------------------------------- | |
| def test_tampered_role_in_db_still_authenticates_as_reader( | |
| temp_inventory, | |
| ) -> None: | |
| """BE-m4(a): a row manually storing role='admin' still resolves as reader. | |
| Directly inserts a row with ``role='admin'`` into app_users (bypassing | |
| create_app_user) to simulate a tampered/hand-edited DB, then verifies | |
| that ``get_app_user_for_auth`` unconditionally returns role='reader' | |
| and is_admin=False. | |
| """ | |
| from src.lib.auth.passwords import hash_password | |
| from src.lib.auth.user_store import get_app_user_for_auth | |
| from src.stage1_inventory.db import connect | |
| pw_hash = hash_password("tamperpass1") | |
| with connect() as conn: | |
| conn.execute( | |
| """ | |
| INSERT INTO app_users (username, password_hash, role, display_name, disabled) | |
| VALUES ('tampered_admin', ?, 'admin', NULL, 0) | |
| """, | |
| (pw_hash,), | |
| ) | |
| conn.commit() | |
| user = get_app_user_for_auth("tampered_admin") | |
| assert user is not None, "tampered row should still resolve (it's enabled)" | |
| assert user.role == "reader", ( | |
| f"tampered role='admin' must be forced to 'reader', got {user.role!r}" | |
| ) | |
| assert user.is_admin is False, "is_admin must be False for a DB-backed account" | |
| def test_reader_redaction_holds_after_admin_fills_cache( | |
| temp_inventory_with_sensitive_book, | |
| ) -> None: | |
| """BE-m4(b): reader redaction is applied after the shared cache, never before. | |
| Calls get_book_endpoint as admin first (populates the cache with the full | |
| DTO), then calls as reader and asserts that sensitive fields are stripped | |
| from the reader's response while the cache entry itself remains full. | |
| The test seeds ``notes`` and ``source_url`` on the book row; costs come | |
| from the (empty) llm_calls table so they are 0.0 for all roles. What | |
| matters here is that the reader's response has notes/source_url redacted, | |
| while the cached (admin-visible) entry retains them. | |
| """ | |
| import src.api.cache as api_cache | |
| from src.api.routers.books import get_book_endpoint | |
| # Clear any leftover cache entries from prior tests. | |
| api_cache.clear() | |
| # Admin call β populates the cache with the full (unredacted) DTO. | |
| admin_dto = get_book_endpoint(book_id="pub_sensitive", user=_ADMIN) | |
| assert admin_dto.notes == "Admin-only private notes." | |
| assert admin_dto.source_url == "https://example.com/sensitive.pdf" | |
| # Reader call β cache is already warm; the function must still redact. | |
| reader_dto = get_book_endpoint(book_id="pub_sensitive", user=_READER) | |
| assert reader_dto.notes is None, "notes must be redacted for reader even from cache" | |
| assert reader_dto.source_url is None, "source_url must be redacted for reader" | |
| assert reader_dto.local_path is None, "local_path must be redacted for reader" | |
| # The cached entry is still the full (unredacted) DTO. | |
| cached = api_cache.get_or_set("books:detail:pub_sensitive", lambda: None) | |
| assert cached is not None and cached.notes == "Admin-only private notes.", ( | |
| "cache must hold the full DTO β reader redaction must not mutate it" | |
| ) | |
| assert cached.source_url == "https://example.com/sensitive.pdf", ( | |
| "cached source_url must be intact after reader access" | |
| ) | |
| def test_reader_private_and_nonexistent_book_return_identical_404( | |
| temp_inventory, | |
| ) -> None: | |
| """BE-m4(c): private book and nonexistent book return identical 404 shapes. | |
| A reader must not be able to distinguish 'private book exists' from | |
| 'book does not exist' β both must raise BookNotFound with the same code | |
| and no extra detail. | |
| """ | |
| with pytest.raises(BookNotFound) as exc_private: | |
| require_book_read_access(book_id="priv_book", user=_READER) | |
| with pytest.raises(BookNotFound) as exc_missing: | |
| require_book_read_access(book_id="no_such_book_xyz", user=_READER) | |
| assert exc_private.value.code == exc_missing.value.code == "BOOK_NOT_FOUND", ( | |
| "private and nonexistent ids must produce identical error codes for readers" | |
| ) | |
| # Neither exception must carry the book_id as an informative detail that | |
| # distinguishes a real (private) book from a missing one. | |
| # (Both raise BookNotFound β same shape by construction.) | |