"""The brain over HTTP: auth on every route, visibility, tools, memory, voice, push, cron, data. Real access-token verification (tokens minted by the real issuer), the in-memory store and the fake provider. Nothing here runs in dev-auth mode, so every 401/403 below is the production path. """ import time from datetime import UTC, datetime, timedelta import pytest from fastapi.testclient import TestClient from brain import config from brain.app import create_app from brain.auth import AuthConfig, User, Verifier, issue_access_token from brain.deps import Deps from brain.llm.base import ToolCall from brain.llm.fake import FakeLlm from brain.persona.types import Scope from brain.push.sender import RecordingSender, VapidConfig from brain.services.chat import ChatService from brain.store.memory import MemoryStore from brain.tools.catalog import default_registry SECRET = "api-test-secret-long-enough-for-signing-tokens-32b+" OWNER = "karnak@example.com" USER_ID = "11111111-1111-1111-1111-111111111111" CLIENT_ID = "test-web-client.apps.googleusercontent.com" ORIGIN = "https://prova.web.app" CRON = {"x-prova-cron-secret": "cron-secret"} def auth_config() -> AuthConfig: return AuthConfig( secret=SECRET, google_client_ids=(CLIENT_ID,), allowed_emails=(OWNER,), cron_secret="cron-secret", ) def token(email: str = OWNER, sub: str = USER_ID) -> str: user = User(id=sub, email=email) return issue_access_token(auth_config(), user, session_id="test-session", now=time.time()) AUTH = {"Authorization": f"Bearer {token()}"} class Clock: """A settable clock, so gates that depend on the hour can be tested at any hour.""" def __init__(self, at: datetime) -> None: self.at = at def __call__(self) -> datetime: return self.at @pytest.fixture def env(): store = MemoryStore() llm = FakeLlm() sender = RecordingSender() # Monday 21 Sep 2026, 11:00 IST clock = Clock(datetime(2026, 9, 21, 5, 30, tzinfo=UTC)) auth = auth_config() settings = config.Settings() deps = Deps( settings=settings, store=store, llm=llm, chat=ChatService(store, llm, default_registry(), settings, clock=clock), auth=auth, verifier=Verifier(auth), sender=sender, vapid=VapidConfig("pub", "priv", "mailto:x@example.com"), allowed_origins=(ORIGIN,), ) with TestClient(create_app(deps)) as client: yield client, store, llm, sender, clock def send(client, text: str, mode: str = "normal", **extra): return client.post("/chat", json={"text": text, "mode": mode, **extra}, headers=AUTH) # --- Auth -------------------------------------------------------------------- def test_health_needs_no_token(env): client, *_ = env assert client.get("/health").json()["ok"] is True def test_the_root_page_shows_the_brain_is_up_without_a_token(env): client, *_ = env response = client.get("/") assert response.status_code == 200 assert response.headers["content-type"].startswith("text/html") assert "Backend running" in response.text assert "= 1 # --- Memory ------------------------------------------------------------------ def test_memory_extraction_runs_on_every_third_turn(env): client, store, llm, *_ = env llm.queue("one", "two", "three", "shared|His sister's name is Ria.") for text in ("first message here", "second message here", "my sister is called Ria"): send(client, text) assert [m["content"] for m in store.memories] == ["His sister's name is Ria."] assert store.memories[0]["scope"] == "shared" def test_normal_mode_memory_screen_lists_shared_facts_only(env): client, *_ = env client.post( "/memory", json={"content": "His sister's name is Ria.", "scope": "shared", "mode": "gf"}, headers=AUTH, ) client.post( "/memory", json={"content": "They watched the rain together.", "scope": "private", "mode": "gf"}, headers=AUTH, ) normal = client.get("/memory?mode=normal", headers=AUTH).json() gf = client.get("/memory?mode=gf", headers=AUTH).json() assert [m["content"] for m in normal] == ["His sister's name is Ria."] assert len(gf) == 2 def test_a_fact_added_from_normal_mode_is_always_shared(env): client, store, *_ = env client.post( "/memory", json={"content": "His dentist is on Friday.", "scope": "private", "mode": "normal"}, headers=AUTH, ) assert store.memories[0]["scope"] == "shared" def test_normal_mode_cannot_make_a_fact_shared(env): client, store, *_ = env made = client.post( "/memory", json={"content": "A private thing here.", "scope": "private", "mode": "gf"}, headers=AUTH, ).json() memory_id = made["memory"]["id"] blocked = client.patch( f"/memory/{memory_id}", json={"scope": "shared", "mode": "normal"}, headers=AUTH ) assert blocked.status_code == 403 assert ( client.patch( f"/memory/{memory_id}", json={"scope": "shared", "mode": "gf"}, headers=AUTH ).status_code == 200 ) def test_clearing_from_normal_mode_leaves_private_facts_alone(env): client, store, *_ = env client.post( "/memory", json={"content": "A shared fact about work.", "scope": "shared", "mode": "gf"}, headers=AUTH, ) client.post( "/memory", json={"content": "A private fact about us.", "scope": "private", "mode": "gf"}, headers=AUTH, ) assert client.delete("/memory?mode=normal", headers=AUTH).json() == {"deleted": 1} assert [m["scope"] for m in store.memories] == ["private"] # --- Settings and character -------------------------------------------------- def test_settings_merge_over_the_defaults(env): client, *_ = env defaults = client.get("/settings", headers=AUTH).json() assert defaults["gfRequiresLock"] is True # master prompt 8: default on assert defaults["privacy"]["normalSeesPrivate"] is False # master prompt 4.5 patched = client.patch( "/settings", json={"privacy": {"blurOnHidden": False}}, headers=AUTH ).json() assert patched["privacy"] == { "normalSeesPrivate": False, "hideNotifContent": True, "blurOnHidden": False, } def test_an_unknown_mood_is_refused(env): client, *_ = env assert client.patch("/settings", json={"mood": "furious"}, headers=AUTH).status_code == 422 def test_gf_persona_text_is_only_returned_in_gf_mode(env): client, *_ = env normal = client.get("/character?mode=normal", headers=AUTH).json() gf = client.get("/character?mode=gf", headers=AUTH).json() assert "relationship_layer" not in normal and "system_base" not in normal assert "been together a long time" in gf["relationship_layer"] # --- Voice ------------------------------------------------------------------- def test_tts_returns_audio(env): client, *_ = env # Gemini speech is opt-in (its free tier is 10 requests a day); auto has no Groq here. client.patch("/settings", json={"voice": {"engine": "gemini"}}, headers=AUTH) response = client.post( "/tts", json={"text": "The report is ready.", "mode": "normal"}, headers=AUTH ) assert response.status_code == 200 assert response.headers["content-type"] == "audio/wav" def test_auto_voice_has_no_gemini_speech_so_the_client_uses_the_device_voice(env): client, *_ = env response = client.post( "/tts", json={"text": "The report is ready.", "mode": "normal"}, headers=AUTH ) assert response.status_code == 503 assert response.json()["detail"]["code"] == "tts_unavailable" def test_gf_tts_transliterates_once_and_caches_it(env): client, store, llm, *_ = env client.patch("/settings", json={"voice": {"engine": "gemini"}}, headers=AUTH) llm.queue("Kemon acho?", "কেমন আছো?") send(client, "hi there", mode="gf") message_id = store.messages[-1]["id"] client.post("/tts", json={"messageId": message_id, "mode": "gf"}, headers=AUTH) assert store.messages[-1]["bengali_text"] == "কেমন আছো?" def test_a_normal_device_cannot_speak_a_gf_message_by_id(env): client, store, llm, *_ = env llm.queue("private reply") send(client, "gf line here", mode="gf") gf_id = store.messages[-1]["id"] assert ( client.post("/tts", json={"messageId": gf_id, "mode": "normal"}, headers=AUTH).status_code == 404 ) def test_stt_fix_romanises_bengali_script_only(env): client, _, llm, *_ = env llm.queue("kothay chiley etokkhon") bengali = client.post("/stt-fix", json={"text": "কোথায় ছিলে এতক্ষণ"}, headers=AUTH).json() assert bengali == {"latin": "kothay chiley etokkhon", "bengali": "কোথায় ছিলে এতক্ষণ"} latin = client.post("/stt-fix", json={"text": "ki korcho"}, headers=AUTH).json() assert latin == {"latin": "ki korcho", "bengali": None} # --- Push, proactive, reminders ---------------------------------------------- SUBSCRIPTION = { "endpoint": "https://push.example.com/abc", "keys": {"p256dh": "p", "auth": "a"}, "deviceLabel": "Phone", } def test_subscribing_twice_does_not_duplicate_the_device(env): client, store, *_ = env client.post("/push/subscribe", json=SUBSCRIPTION, headers=AUTH) client.post("/push/subscribe", json=SUBSCRIPTION, headers=AUTH) listed = client.get("/push/subscriptions", headers=AUTH).json() assert len(listed) == 1 assert "p256dh" not in listed[0] # keys never leave the brain def test_the_test_notification_is_sent(env): client, _, _, sender, _ = env client.post("/push/subscribe", json=SUBSCRIPTION, headers=AUTH) assert client.post("/push/test", headers=AUTH).json()["sent"] == 1 assert sender.sent[0][1]["tag"] == "test" def test_cron_routes_need_the_shared_secret(env): client, *_ = env assert client.post("/proactive").status_code == 401 assert client.post("/reminders", headers={"x-prova-cron-secret": "wrong"}).status_code == 401 assert client.post("/reminders", headers=CRON).status_code == 200 def test_a_due_reminder_is_pushed_once_in_normal_wording(env): client, store, llm, sender, clock = env client.post("/push/subscribe", json=SUBSCRIPTION, headers=AUTH) llm.queue( ToolCall("set_reminder", {"text": "Client call", "when_iso": "2026-09-21T14:45:00"}), "Done.", ) send(client, "remind me about the client call at 2:45") clock.at = datetime(2026, 9, 21, 9, 16, tzinfo=UTC) # 14:46 IST assert client.post("/reminders", headers=CRON).json()["sent"] == 1 assert client.post("/reminders", headers=CRON).json()["sent"] == 0 # never twice body = sender.sent[-1][1]["body"] assert body.startswith("Reminder: Client call") def test_a_reminder_set_in_gf_mode_has_a_neutral_lock_screen_body(env): client, store, llm, sender, clock = env client.post("/push/subscribe", json=SUBSCRIPTION, headers=AUTH) llm.queue( ToolCall("set_reminder", {"text": "anniversary gift", "when_iso": "2026-09-21T12:00:00"}), "Mone rakhbo 😌", ) send(client, "remind me about the gift at noon", mode="gf") clock.at = datetime(2026, 9, 21, 6, 31, tzinfo=UTC) # 12:01 IST client.post("/reminders", headers=CRON) assert "anniversary" not in sender.sent[-1][1]["body"] def test_an_expired_push_subscription_is_removed(env): client, store, llm, sender, clock = env client.post("/push/subscribe", json=SUBSCRIPTION, headers=AUTH) sender.status = 410 client.post("/push/test", headers=AUTH) assert store.subscriptions == [] def seed_history( store, hours_ago: float, clock: Clock, role: str = "user", proactive: bool = False ): store.settings.setdefault(USER_ID, {}) store.messages.append( { "id": f"h{len(store.messages)}", "user_id": USER_ID, "mode": "gf", "role": role, "content": "earlier", "is_proactive": proactive, "created_at": clock.at - timedelta(hours=hours_ago), } ) def test_proactive_sends_a_neutral_push_after_a_long_silence(env): client, store, llm, sender, clock = env client.post("/push/subscribe", json=SUBSCRIPTION, headers=AUTH) seed_history(store, hours_ago=6, clock=clock) llm.queue("Ki korcho? Lunch hoyeche?") report = client.post("/proactive", headers=CRON).json() assert report["sent"] == 1 payload = sender.sent[-1][1] assert payload["body"] == "Prova: new message" # never the GF text on a lock screen assert payload["url"] == "/?badge=gf" assert store.messages[-1]["is_proactive"] is True def test_proactive_is_silent_in_quiet_hours(env): client, store, llm, sender, clock = env clock.at = datetime(2026, 9, 20, 22, 30, tzinfo=UTC) # 04:00 IST Monday seed_history(store, hours_ago=8, clock=clock) report = client.post("/proactive", headers=CRON).json() assert report["sent"] == 0 assert report["skipped"][USER_ID] == "quiet_hours" def test_an_unanswered_check_in_is_never_stacked(env): client, store, llm, sender, clock = env seed_history(store, hours_ago=5.5, clock=clock, role="assistant", proactive=True) report = client.post("/proactive", headers=CRON).json() assert report["skipped"][USER_ID] == "unanswered_backoff" # --- Data (P8) --------------------------------------------------------------- def test_export_then_import_round_trips_memories(env): client, store, *_ = env client.post( "/memory", json={"content": "His sister's name is Ria.", "mode": "gf"}, headers=AUTH ) exported = client.get("/data/export", headers=AUTH).json() assert exported["format"] == "prova-export" assert "embedding" not in str(exported) store.memories.clear() result = client.post("/data/import", json=exported, headers=AUTH).json() assert result["memories"]["added"] == 1 assert store.memories[0]["content"] == "His sister's name is Ria." def test_delete_everything_needs_the_confirmation_word(env): client, store, llm, *_ = env llm.queue("ok") send(client, "hello there Prova") assert client.post("/data/delete", json={"confirm": "yes"}, headers=AUTH).status_code == 422 assert client.post("/data/delete", json={"confirm": "DELETE"}, headers=AUTH).json() == { "ok": True } assert store.messages == [] # --- Positive controls -------------------------------------------------------- # The two privacy tests above assert that something is ABSENT. Absence is also what a broken # retrieval or a broken history would produce, so each gets a twin proving the same data does # reach the model in GF mode. Without these, the privacy tests could pass for the wrong reason. def test_control_the_same_private_memory_does_reach_a_gf_prompt(env): client, store, llm, *_ = env seed_private_memory(store, llm) llm.queue("Hmm 😌") send(client, PRIVATE_QUERY, mode="gf") assert "babu when he is tired" in llm.prompts[-1] def test_control_normal_sees_private_lets_it_through_when_he_turns_it_on(env): client, store, llm, *_ = env seed_private_memory(store, llm) client.patch("/settings", json={"privacy": {"normalSeesPrivate": True}}, headers=AUTH) llm.queue("Noted.") send(client, PRIVATE_QUERY, mode="normal") assert "babu when he is tired" in llm.prompts[-1] def test_control_a_gf_message_does_reach_the_model_on_a_gf_turn(env): client, _, llm, *_ = env llm.queue("Tomake chara ghum ashe na", "Hmm") send(client, "secret gf line about the balcony", mode="gf") send(client, "another gf line", mode="gf") assert "balcony" in str(llm.requests[-1]["contents"]) def test_fixed_lines_come_from_the_brain_in_the_right_voice(env): client, *_ = env normal = client.get("/character?mode=normal", headers=AUTH).json()["lines"] gf = client.get("/character?mode=gf", headers=AUTH).json()["lines"] assert ( normal["videoCallUnavailable"] == "Video calling isn't available yet. Voice call instead?" ) assert "🙈" in gf["videoCallUnavailable"] # --- The Gemini / Groq switch over HTTP --------------------------------------- def test_the_chat_brain_setting_defaults_to_auto_and_rejects_junk(env): client, *_ = env assert client.get("/settings", headers=AUTH).json()["chatProvider"] == "groq" assert ( client.patch("/settings", json={"chatProvider": "groq"}, headers=AUTH).json()[ "chatProvider" ] == "groq" ) assert client.patch("/settings", json={"chatProvider": "nope"}, headers=AUTH).status_code == 422 def test_the_setting_actually_chooses_the_provider(env): from brain.llm.router import Router client, store, llm, *_ = env groq = FakeLlm(["answered by groq"]) router = Router(llm, groq) client.app.state.deps.chat.llm = router llm.queue("answered by gemini") client.patch("/settings", json={"chatProvider": "groq"}, headers=AUTH) assert send(client, "hello there Prova").json()["message"]["content"] == "answered by groq" client.patch("/settings", json={"chatProvider": "gemini"}, headers=AUTH) assert send(client, "hello again Prova").json()["message"]["content"] == "answered by gemini" def test_a_photo_always_goes_to_gemini_because_the_backup_cannot_see(env): from brain.llm.router import Router client, store, llm, *_ = env groq = FakeLlm(["groq must not see photos"]) client.app.state.deps.chat.llm = Router(llm, groq) llm.queue("gemini saw it") client.patch("/settings", json={"chatProvider": "groq"}, headers=AUTH) user = USER_ID store.images[f"{user}/pic.jpg"] = (b"\xff\xd8", "image/jpeg") body = send(client, "look at this", imagePath=f"{user}/pic.jpg").json() assert body["message"]["content"] == "gemini saw it" def test_a_greeting_is_hers_alone_and_stores_no_user_turn(env): client, store, llm, *_ = env llm.queue("Good evening, Rudro. Nothing is outstanding today.") before = len(store.messages) body = client.post( "/chat", json={"text": "", "mode": "normal", "greeting": True}, headers=AUTH ).json() assert body["message"]["content"].startswith("Good evening") assert body["userMessage"] is None # Exactly one row: hers. An invented user line would be two. assert len(store.messages) == before + 1 # Answering a call is not a check-in: marking it proactive would make the phone notify him # about a message he is already looking at. assert store.messages[-1]["is_proactive"] is False def test_an_ordinary_empty_message_is_still_refused(env): client, *_ = env assert ( client.post("/chat", json={"text": " ", "mode": "normal"}, headers=AUTH).status_code == 422 ) def test_a_greeting_forbids_inventing_his_day_and_does_not_use_the_checkin_prompt(env): client, _, llm, *_ = env llm.queue("Good evening, Rudro. I'm here.") client.post("/chat", json={"text": "", "mode": "normal", "greeting": True}, headers=AUTH) prompt = llm.prompts[-1] if hasattr(llm, "prompts") else "" if prompt: assert "ANSWERING A CALL" in prompt assert "MESSAGE HIM FIRST" not in prompt assert "invented" in prompt