"""Tests for space/proxy.py turn logic (offline; mock mode + monkeypatch).""" import os import sys _HERE = os.path.dirname(os.path.abspath(__file__)) _SPACE_DIR = os.path.dirname(_HERE) if _SPACE_DIR not in sys.path: sys.path.insert(0, _SPACE_DIR) import arc # noqa: E402 import content # noqa: E402 import proxy # noqa: E402 def test_templated_rejoin_names_the_dj(): assert arc.DJ_NAME in proxy._templated_text("rejoin") def test_templated_caller_intro_nonempty(): assert proxy._templated_text("caller_intro") def test_song_intro_template_includes_handle(): t = proxy._templated_text( "song_intro", {"title": "Neon Rain", "artist": "The Tuesday Ghosts", "recommended_by": "@mara"}, ) assert "@mara" in t def test_song_intro_template_omits_handle_when_absent(): t = proxy._templated_text( "song_intro", {"title": "Neon Rain", "artist": "The Tuesday Ghosts"} ) assert "@" not in t def test_wmo_phrase_mapping(): assert proxy._wmo_phrase(0) == "clear skies" assert proxy._wmo_phrase(61) == "light rain" assert proxy._wmo_phrase(999) == "quiet skies" # unknown -> safe default def test_fmt_clock(): assert proxy._fmt_clock("2026-06-15T02:14") == "2:14 AM" assert proxy._fmt_clock("2026-06-15T13:05") == "1:05 PM" assert proxy._fmt_clock(None) is None def test_resolve_locale_mock(monkeypatch): monkeypatch.setenv("NIGHTWAVE_MOCK", "1") f = proxy.resolve_locale(30.27, -97.74) assert f["resolved"] is True assert f["city"] and f["temp_f"] is not None and f["sky"] def test_local_weather_falls_back_to_fictional_on_junk(monkeypatch): monkeypatch.setattr(proxy, "call_brain", lambda system, messages: {"text": "", "mood": "warm", "arc_cue": "none"}) monkeypatch.setattr(proxy, "call_speak", lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.segment_turn("local_weather", {"city": "Austin", "temp_f": 58}) assert out["text"] in content.WEATHER # junk brain text -> fictional fallback assert out["kind"] == "local_weather" # --------------------------------------------------------------------------- # Regression coverage for the segment_turn junk-fallback path and the two # segment helpers server.py calls (segment_fallback, get_stalls). `_clean_dj_text` # is the junk-detector the Task-5 local_weather fallback test relies on; cover it # directly so that load-bearing behaviour can't silently regress. # --------------------------------------------------------------------------- def test_clean_dj_text_rejects_degenerate_text_as_none(): # Empty / placeholder / angle-bracket template / too-few-letters -> None, # so segment_turn falls back to a clean templated line instead of junk. This # is the exact mechanism the Task-5 local_weather fallback test depends on. assert proxy._clean_dj_text("") is None assert proxy._clean_dj_text(None) is None assert proxy._clean_dj_text("") is None assert proxy._clean_dj_text("a b c") is None assert proxy._clean_dj_text("Sit tight, night owls.") == "Sit tight, night owls." def test_get_stalls_returns_cached_clip_list(monkeypatch): # server.py /api/stalls calls this; reset the module cache for order-independence. monkeypatch.setattr(proxy, "_stall_cache", None) clips = proxy.get_stalls() assert isinstance(clips, list) and len(clips) == len(proxy._STALL_LINES) assert all(isinstance(c, str) and c for c in clips) # each is a base64 WAV assert proxy.get_stalls() is clips # second call is cached def test_segment_fallback_is_never_failing_and_well_shaped(): # server.py /api/segment uses this on any failure so the show never breaks. fb = proxy.segment_fallback("thought") assert fb["kind"] == "thought" assert fb["text"] in content.THOUGHTS # templated, stage-appropriate assert fb["audio_b64"] # a real (silent) WAV bed for key in ("mood", "arc_cue", "words", "wtimes", "wdurations"): assert key in fb # --------------------------------------------------------------------------- # Regression (bug 2): the all-caps station name must be normalized before TTS so # Kokoro speaks "Nightwave" instead of spelling N-I-G-H-T-W-A-V-E. _speakable was # dropped during a proxy rebuild and call_speak sent raw text. # --------------------------------------------------------------------------- def test_speakable_titlecases_nightwave(): assert proxy._speakable("You're on NIGHTWAVE 98.6") == "You're on Nightwave 98.6" assert proxy._speakable("night wave") == "Nightwave" assert proxy._speakable(None) == "" def test_call_speak_normalizes_nightwave_before_tts(monkeypatch): sent = {} class _Resp: def raise_for_status(self): pass def json(self): return {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []} class _Client: def post(self, url, headers=None, json=None): sent.update(json or {}) return _Resp() monkeypatch.setattr(proxy, "is_mock", lambda: False) monkeypatch.setattr(proxy, "_modal_url", lambda: "http://modal.test") monkeypatch.setattr(proxy, "_modal_headers", lambda: {}) monkeypatch.setattr(proxy, "_get_client", lambda: _Client()) proxy.call_speak("This is NIGHTWAVE, the last station on the dial.") assert "NIGHTWAVE" not in sent["text"] assert "Nightwave" in sent["text"] # --------------------------------------------------------------------------- # Regression (bug 4): the caller turn must ANSWER the caller, not echo. The # straight station uses build_host_prompt("caller") + an answer-directive user # turn so a 1B model doesn't parrot a question that appears verbatim in context. # call_turn had reverted to the arc builder + raw caller_text as the user message. # --------------------------------------------------------------------------- def test_call_turn_answers_without_echoing_the_question(monkeypatch): captured = {} monkeypatch.setattr(proxy, "call_asr", lambda a: {"text": "are you real?"}) def _brain(system, messages): captured["system"] = system captured["messages"] = messages return {"text": "Real as the rain on the glass, friend.", "mood": "warm", "arc_cue": "none"} monkeypatch.setattr(proxy, "call_brain", _brain) monkeypatch.setattr(proxy, "call_speak", lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.call_turn("oblivious", 0, "") # straight-station host prompt (names the DJ), NOT the arc builder assert arc.DJ_NAME in captured["system"] # the user turn carries an answer-directive, not the bare echoed question user = captured["messages"][-1]["content"] assert "do not repeat" in user.lower() assert user.strip() != "are you real?" assert out["caller_text"] == "are you real?" assert out["meter_delta"] >= 14 # an identity/reality question pushes the meter # --------------------------------------------------------------------------- # Prompt-audit fixes: structured field labels (title:/artist:/temp:/...) must # never reach TTS; degenerate/echo output is rejected -> templated fallback. # --------------------------------------------------------------------------- def test_clean_dj_text_rejects_label_leakage(): assert proxy._clean_dj_text("title: Neon Rain, artist: The Tuesday Ghosts") is None assert proxy._clean_dj_text('{"text": "hi", "mood": "warm"}') is None assert proxy._clean_dj_text("[song]") is None assert proxy._clean_dj_text("temp: 58, sky: clear skies") is None assert proxy._clean_dj_text("Coming up next, a quiet one for the night owls.") == \ "Coming up next, a quiet one for the night owls." def test_call_turn_degenerate_uses_caller_fallback_not_crash(monkeypatch): # The model leaks labels -> _clean_dj_text returns None -> must use a clean # caller fallback line (the old arc.sample_line was undefined and crashed here). monkeypatch.setattr(proxy, "call_asr", lambda a: {"text": "you there?"}) monkeypatch.setattr(proxy, "call_brain", lambda s, m: {"text": "title: X, artist: Y", "mood": "warm", "arc_cue": "none"}) monkeypatch.setattr(proxy, "call_speak", lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.call_turn("oblivious", 0, "") assert out["text"] in content.CALLER_FALLBACKS def test_broadcast_turn_uses_host_path_and_sanitizes(monkeypatch): captured = {} def _brain(system, messages): captured["system"] = system return {"text": "artist: X", "mood": "warm", "arc_cue": "none"} # degenerate monkeypatch.setattr(proxy, "call_brain", _brain) monkeypatch.setattr(proxy, "call_speak", lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.broadcast_turn("oblivious", 0, None) assert arc.DJ_NAME in captured["system"] # straight-station host, not arc builder assert out["text"] in content.THOUGHTS # degenerate -> templated thought assert out["arc_cue"] == "none" def test_extract_name_place(): assert proxy._extract_name("Hi, I'm Mira and I'm calling in") == "Mira" assert proxy._extract_name("just wanted to say hi") is None assert proxy._extract_place("I'm driving in Austin tonight") == "Austin" assert proxy._extract_place("no place here") is None def test_caller_gist(): assert proxy._caller_gist(" I'm driving home after a double shift ") == \ "I'm driving home after a double shift" assert proxy._caller_gist("hi") is None assert proxy._caller_gist("") is None def test_build_memory_patch(): p = proxy._build_memory_patch("I'm Mira, driving home after a double shift", "tender") assert p["caller_name"] == "Mira" and p["topic"] and p["mood"] == "tender" assert proxy._build_memory_patch("hi", "warm") is None # too short -> None assert proxy._build_memory_patch("I'm driving all night", "bogus")["mood"] == "warm" # bad mood -> warm def test_call_turn_returns_memory_patch(monkeypatch): monkeypatch.setattr(proxy, "call_asr", lambda a: {"text": "I'm Mira, driving home after a double shift"}) monkeypatch.setattr(proxy, "call_brain", lambda s, m: {"text": "Drive safe, friend.", "mood": "tender", "arc_cue": "none"}) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.call_turn("oblivious", 0, "") assert out["queue_dedication"] is True assert out["memory_patch"]["caller_name"] == "Mira" assert out["memory_patch"]["mood"] == "tender" def test_call_turn_no_memory_on_short_caller(monkeypatch): monkeypatch.setattr(proxy, "call_asr", lambda a: {"text": "hi"}) monkeypatch.setattr(proxy, "call_brain", lambda s, m: {"text": "Hey there, friend.", "mood": "warm", "arc_cue": "none"}) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.call_turn("oblivious", 0, "") assert out["queue_dedication"] is False and out["memory_patch"] is None def test_segment_dedication_memory_aware(monkeypatch): captured = {} def _brain(system, messages): captured["user"] = messages[-1]["content"] return {"text": "This one's for the night-shift drivers, headlights home.", "mood": "tender", "arc_cue": "none"} monkeypatch.setattr(proxy, "call_brain", _brain) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.segment_turn("dedication", {"topic": "driving home after a double shift", "caller_name": "Mira", "mood": "tender"}) assert "Mira" in captured["user"] and "double shift" in captured["user"] assert out["text"].startswith("This one's for") def test_segment_dedication_no_memory_is_templated(monkeypatch): monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.segment_turn("dedication", None) # no model call on this path assert out["text"].startswith("This next one's for") # from content.DEDICATIONS def test_segment_dedication_junk_falls_back_to_name(monkeypatch): monkeypatch.setattr(proxy, "call_brain", lambda s, m: {"text": "title: X", "mood": "warm", "arc_cue": "none"}) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.segment_turn("dedication", {"topic": "waiting up", "caller_name": "Sam"}) assert "Sam" in out["text"] # _dedication_fallback weaves the name def test_song_intro_weaves_segue(monkeypatch): captured = {} def _brain(system, messages): captured["user"] = messages[-1]["content"] return {"text": "After that caller, a soft one -- 'Neon Rain'.", "mood": "tender", "arc_cue": "none"} monkeypatch.setattr(proxy, "call_brain", _brain) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) proxy.segment_turn("song_intro", {"title": "Neon Rain", "artist": "The Tuesday Ghosts", "vibe": "melancholy", "segue": "right after that caller"}) assert "right after that caller" in captured["user"] def test_song_intro_no_segue_when_absent(monkeypatch): captured = {} def _brain(system, messages): captured["user"] = messages[-1]["content"] return {"text": "Coming up, 'Neon Rain'.", "mood": "warm", "arc_cue": "none"} monkeypatch.setattr(proxy, "call_brain", _brain) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) proxy.segment_turn("song_intro", {"title": "Neon Rain", "artist": "The Tuesday Ghosts", "vibe": "melancholy"}) assert "through-line" not in captured["user"] def test_fragment_user_mentions_another_station(): u = proxy._fragment_user(None) assert "another station" in u and len(u) > 20 def test_segment_fragment_is_persona_free(monkeypatch): captured = {} def _brain(system, messages): captured["system"] = system return {"text": "calling all night nurses near mile marker nine", "mood": "warm", "arc_cue": "none"} monkeypatch.setattr(proxy, "call_brain", _brain) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.segment_turn("fragment", None) assert arc.HOST_PERSONA not in captured["system"] # not the Sam Dusk host persona assert out["text"].startswith("calling all night nurses") def test_segment_fragment_junk_falls_back(monkeypatch): monkeypatch.setattr(proxy, "call_brain", lambda s, m: {"text": "title: X", "mood": "warm", "arc_cue": "none"}) monkeypatch.setattr(proxy, "call_speak", lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}) out = proxy.segment_turn("fragment", None) assert out["text"] in content.FRAGMENTS # junk -> templated bank