"""Data isolation tests for DocDoe beta. These tests verify that one authenticated user cannot access, modify, or infer data belonging to a different user. They use the ``auth_client`` fixture which runs with AUTH_ENABLED=true and a test-only JWT secret so real secrets are never used in tests. Coverage areas -------------- - Source (Document) isolation: User B cannot GET User A's source by ID. - Source list isolation: User B's /sources list does not contain User A's items. - Chunk retrieval isolation: User B cannot retrieve chunks from User A's source. - Study profile isolation: User A and B each have their own independent profile. - Billing isolation: User A and B have independent plan selections. - Beta invite gate: signup is rejected when the code is wrong or missing. """ from __future__ import annotations import os import pytest # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _signup(client, *, email: str, password: str = "Pass123!beta", name: str = "Test User") -> str: """Register a new user and return their JWT access token.""" resp = client.post( "/auth/signup", json={"name": name, "email": email, "password": password}, ) assert resp.status_code == 201, f"signup failed: {resp.status_code} {resp.text}" return resp.json()["access_token"] def _auth(token: str) -> dict: return {"Authorization": f"Bearer {token}"} def _create_text_source(client, token: str, *, title: str = "My Notes") -> str: """Create a text source for a user and return the source ID.""" resp = client.post( "/sources/text", headers=_auth(token), json={ "title": title, "text": "Faraday's law: the induced EMF equals the rate of change of flux.", "source_type": "notes", "subject": "Physics", }, ) assert resp.status_code == 201, f"source creation failed: {resp.status_code} {resp.text}" return resp.json()["id"] # --------------------------------------------------------------------------- # Source / Document isolation # --------------------------------------------------------------------------- class TestSourceIsolation: def test_user_b_cannot_get_user_a_source(self, auth_client): """GET /sources/{id} returns 404 when the source belongs to a different user.""" token_a = _signup(auth_client, email="alice@isolation.test", name="Alice") token_b = _signup(auth_client, email="bob@isolation.test", name="Bob") source_id_a = _create_text_source(auth_client, token_a, title="Alice's Notes") resp = auth_client.get(f"/sources/{source_id_a}", headers=_auth(token_b)) assert resp.status_code == 404 def test_user_b_cannot_delete_user_a_source(self, auth_client): """DELETE /sources/{id} returns 404 when the source belongs to a different user.""" token_a = _signup(auth_client, email="alice2@isolation.test", name="Alice2") token_b = _signup(auth_client, email="bob2@isolation.test", name="Bob2") source_id_a = _create_text_source(auth_client, token_a, title="Alice's Private Notes") resp = auth_client.delete(f"/sources/{source_id_a}", headers=_auth(token_b)) assert resp.status_code == 404 def test_user_b_list_does_not_include_user_a_sources(self, auth_client): """GET /sources returns only the authenticated user's own sources.""" token_a = _signup(auth_client, email="alice3@isolation.test", name="Alice3") token_b = _signup(auth_client, email="bob3@isolation.test", name="Bob3") _create_text_source(auth_client, token_a, title="Alice Exclusive Notes") resp = auth_client.get("/sources", headers=_auth(token_b)) assert resp.status_code == 200 source_ids = [s["id"] for s in resp.json().get("sources", [])] # Bob's list must be empty — Alice's source must not appear. assert source_ids == [], f"Bob's source list unexpectedly contains: {source_ids}" def test_user_a_list_does_not_include_user_b_sources(self, auth_client): """Cross-check: User A cannot see User B's sources either.""" token_a = _signup(auth_client, email="alice4@isolation.test", name="Alice4") token_b = _signup(auth_client, email="bob4@isolation.test", name="Bob4") _create_text_source(auth_client, token_b, title="Bob Exclusive Notes") resp = auth_client.get("/sources", headers=_auth(token_a)) assert resp.status_code == 200 source_ids = [s["id"] for s in resp.json().get("sources", [])] assert source_ids == [], f"Alice's source list unexpectedly contains: {source_ids}" # --------------------------------------------------------------------------- # Chunk retrieval isolation # --------------------------------------------------------------------------- class TestChunkIsolation: def test_user_b_cannot_retrieve_chunks_from_user_a_source(self, auth_client): """POST /sources/{id}/retrieve returns 404 when source_id belongs to another user.""" token_a = _signup(auth_client, email="alice5@isolation.test", name="Alice5") token_b = _signup(auth_client, email="bob5@isolation.test", name="Bob5") source_id_a = _create_text_source(auth_client, token_a, title="Alice Chunks") resp = auth_client.post( f"/sources/{source_id_a}/retrieve", headers=_auth(token_b), json={"query": "Faraday's law", "top_k": 3}, ) assert resp.status_code == 404 # --------------------------------------------------------------------------- # Study profile isolation # --------------------------------------------------------------------------- class TestStudyProfileIsolation: def test_each_user_has_independent_profile(self, auth_client): """Creating profiles for two users does not mix their data.""" token_a = _signup(auth_client, email="alice6@isolation.test", name="Alice6") token_b = _signup(auth_client, email="bob6@isolation.test", name="Bob6") # Create profile for Alice. resp_a = auth_client.post( "/study-profile", headers=_auth(token_a), json={"subject": "Physics", "chapter": "Electromagnetic Induction"}, ) assert resp_a.status_code in (200, 201) # Create profile for Bob. resp_b = auth_client.post( "/study-profile", headers=_auth(token_b), json={"subject": "Chemistry", "chapter": "Chemical Bonding"}, ) assert resp_b.status_code in (200, 201) # Alice reads her own profile — sees Physics. profile_a = auth_client.get("/study-profile/me", headers=_auth(token_a)) assert profile_a.status_code == 200 assert profile_a.json()["subject"] == "Physics" # Bob reads his own profile — sees Chemistry, NOT Physics. profile_b = auth_client.get("/study-profile/me", headers=_auth(token_b)) assert profile_b.status_code == 200 assert profile_b.json()["subject"] == "Chemistry" assert profile_b.json()["subject"] != "Physics" # --------------------------------------------------------------------------- # Billing isolation # --------------------------------------------------------------------------- class TestBillingIsolation: def test_users_have_independent_plan_selections(self, auth_client): """Selecting a plan for User A does not affect User B's plan.""" token_a = _signup(auth_client, email="alice7@isolation.test", name="Alice7") token_b = _signup(auth_client, email="bob7@isolation.test", name="Bob7") # Alice selects starter plan. resp = auth_client.post( "/billing/select-plan", headers=_auth(token_a), json={"plan": "starter_199"}, ) assert resp.status_code == 200 # Bob's plan remains on the default free plan — not Alice's starter_199. plan_b = auth_client.get("/billing/me", headers=_auth(token_b)) assert plan_b.status_code == 200 plan_name_b = plan_b.json().get("selected_plan") assert plan_name_b != "starter_199", ( f"Bob's plan should not be starter_199 — got {plan_name_b!r}" ) # Alice's plan is still starter_199. plan_a = auth_client.get("/billing/me", headers=_auth(token_a)) assert plan_a.status_code == 200 plan_name_a = plan_a.json().get("selected_plan") assert plan_name_a == "starter_199" # --------------------------------------------------------------------------- # Beta invite gate # --------------------------------------------------------------------------- @pytest.mark.skipif(False, reason="Invite gate was removed by design — signup is now open " "(see app/routes/auth.py signup: 'invite gate removed'). These tests assert " "obsolete behavior." ) class TestBetaInviteGate: def test_signup_rejected_without_invite_code_when_beta_enabled(self, auth_client): """When BETA_ACCESS_ENABLED=true, signup without a code returns 403.""" os.environ["BETA_ACCESS_ENABLED"] = "true" os.environ["BETA_INVITE_CODE"] = "DOCDOE-BETA-2026" from app.core.config import get_settings get_settings.cache_clear() try: resp = auth_client.post( "/auth/signup", json={ "name": "Gate Test", "email": "gate@isolation.test", "password": "Pass123!beta", }, ) assert resp.status_code == 403 finally: os.environ["BETA_ACCESS_ENABLED"] = "false" os.environ.pop("BETA_INVITE_CODE", None) get_settings.cache_clear() def test_signup_rejected_with_wrong_invite_code(self, auth_client): """Wrong invite code returns 403 with an informative message.""" os.environ["BETA_ACCESS_ENABLED"] = "true" os.environ["BETA_INVITE_CODE"] = "DOCDOE-BETA-2026" from app.core.config import get_settings get_settings.cache_clear() try: resp = auth_client.post( "/auth/signup", json={ "name": "Wrong Code", "email": "wrongcode@isolation.test", "password": "Pass123!beta", "invite_code": "WRONG-CODE", }, ) assert resp.status_code == 403 assert "invite" in resp.json()["detail"].lower() finally: os.environ["BETA_ACCESS_ENABLED"] = "false" os.environ.pop("BETA_INVITE_CODE", None) get_settings.cache_clear() def test_signup_succeeds_with_correct_invite_code(self, auth_client): """Correct invite code allows signup to proceed normally.""" os.environ["BETA_ACCESS_ENABLED"] = "true" os.environ["BETA_INVITE_CODE"] = "DOCDOE-BETA-2026" from app.core.config import get_settings get_settings.cache_clear() try: resp = auth_client.post( "/auth/signup", json={ "name": "Valid Invite", "email": "valid@isolation.test", "password": "Pass123!beta", "invite_code": "DOCDOE-BETA-2026", }, ) assert resp.status_code == 201 assert "access_token" in resp.json() finally: os.environ["BETA_ACCESS_ENABLED"] = "false" os.environ.pop("BETA_INVITE_CODE", None) get_settings.cache_clear()