Track four test files that were never committed
CLAUDE.md cites tests/test_model_config.py as the guard against reintroducing retired OpenRouter model ids, but the file existed only on disk — the guard was not in the repository. Same for the town geo, voice roster and voicemail transcript tests, and the Reaper bleep-selection test. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,219 @@
|
||||
-- Stub harness for reaper/bleep_selection.lua
|
||||
-- Models tracks/items in memory and asserts the split/delete/insert behaviour.
|
||||
|
||||
local SCRIPT = (debug.getinfo(1,"S").source:match("@?(.*[/\\])") or "") .. "bleep_selection.lua"
|
||||
|
||||
local function new_item(pos, len, tag)
|
||||
return { pos = pos, len = len, tag = tag or "audio", vol = 1, fin_ = 0, fout_ = 0 }
|
||||
end
|
||||
|
||||
local W -- world
|
||||
|
||||
local function make_reaper()
|
||||
return {
|
||||
CountTrackMediaItems = function(tr) return #tr.items end,
|
||||
GetTrackMediaItem = function(tr, i) return tr.items[i + 1] end,
|
||||
GetMediaItemInfo_Value = function(it, k)
|
||||
if k == "D_POSITION" then return it.pos end
|
||||
if k == "D_LENGTH" then return it.len end
|
||||
return 0
|
||||
end,
|
||||
SetMediaItemInfo_Value = function(it, k, v)
|
||||
if k == "D_POSITION" then it.pos = v
|
||||
elseif k == "D_LENGTH" then it.len = v
|
||||
elseif k == "D_VOL" then it.vol = v
|
||||
elseif k == "D_FADEINLEN" then it.fin_ = v
|
||||
elseif k == "D_FADEOUTLEN" then it.fout_ = v end
|
||||
end,
|
||||
SplitMediaItem = function(it, at)
|
||||
local tr
|
||||
for _, t in ipairs(W.tracks) do
|
||||
for idx, x in ipairs(t.items) do if x == it then tr = t; it_idx = idx end end
|
||||
end
|
||||
if not tr then return nil end
|
||||
if at <= it.pos or at >= it.pos + it.len then return nil end
|
||||
local right = new_item(at, it.pos + it.len - at, it.tag)
|
||||
it.len = at - it.pos
|
||||
local pos_in = 0
|
||||
for idx, x in ipairs(tr.items) do if x == it then pos_in = idx end end
|
||||
table.insert(tr.items, pos_in + 1, right)
|
||||
W.splits = W.splits + 1
|
||||
return right
|
||||
end,
|
||||
DeleteTrackMediaItem = function(tr, it)
|
||||
for idx, x in ipairs(tr.items) do
|
||||
if x == it then table.remove(tr.items, idx); W.deletes = W.deletes + 1; return true end
|
||||
end
|
||||
return false
|
||||
end,
|
||||
InsertMedia = function() W.forbidden["InsertMedia"] = true; return 1 end,
|
||||
GetSelectedMediaItem = function(_, _) return W.last_inserted end,
|
||||
SetOnlyTrackSelected = function(tr) W.forbidden["SetOnlyTrackSelected"] = true end,
|
||||
SetTrackSelected = function() W.forbidden["SetTrackSelected"] = true end,
|
||||
SetEditCurPos = function(p) W.forbidden["SetEditCurPos"] = true end,
|
||||
PCM_Source_CreateFromFile = function(path) W.src_path = path; return { src = path } end,
|
||||
AddMediaItemToTrack = function(tr)
|
||||
local it = new_item(0, 0, "tone"); table.insert(tr.items, it)
|
||||
W.last_inserted = it; it.owner = tr; return it
|
||||
end,
|
||||
AddTakeToMediaItem = function(it) it.take = { item = it }; return it.take end,
|
||||
SetMediaItemTake_Source = function(take, src) take.src = src end,
|
||||
GetToggleCommandStateEx = function(_, cmd) return W.ripple[cmd] and 1 or 0 end,
|
||||
Main_OnCommand = function(cmd) W.commands[#W.commands + 1] = cmd end,
|
||||
GetSet_LoopTimeRange = function() return W.sel_start, W.sel_end end,
|
||||
CountSelectedTracks = function() return #W.sel_tracks end,
|
||||
GetSelectedTrack = function(_, i) return W.sel_tracks[i + 1] end,
|
||||
Undo_BeginBlock = function() end,
|
||||
Undo_EndBlock = function(desc) W.undo = desc end,
|
||||
PreventUIRefresh = function() end,
|
||||
UpdateArrange = function() end,
|
||||
ShowMessageBox = function(msg) W.msg = msg end,
|
||||
}
|
||||
end
|
||||
|
||||
local function run(setup)
|
||||
W = { tracks = {}, sel_tracks = {}, splits = 0, deletes = 0,
|
||||
inserted_paths = {}, edit_cur = 0, msg = nil, undo = nil,
|
||||
forbidden = {}, ripple = {}, commands = {} }
|
||||
setup(W)
|
||||
reaper = make_reaper()
|
||||
local fn = assert(loadfile(SCRIPT))
|
||||
fn()
|
||||
return W
|
||||
end
|
||||
|
||||
local function track(items)
|
||||
local t = { items = {} }
|
||||
for _, it in ipairs(items) do t.items[#t.items + 1] = new_item(it[1], it[2]) end
|
||||
return t
|
||||
end
|
||||
|
||||
local pass, fail = 0, 0
|
||||
local function check(name, cond, detail)
|
||||
if cond then pass = pass + 1; print((" PASS %s"):format(name))
|
||||
else fail = fail + 1; print((" FAIL %s -- %s"):format(name, detail or "")) end
|
||||
end
|
||||
|
||||
local function layout(tr)
|
||||
local s = {}
|
||||
for _, it in ipairs(tr.items) do
|
||||
s[#s + 1] = ("%s[%.2f..%.2f]"):format(it.tag == "tone" and "T" or "A", it.pos, it.pos + it.len)
|
||||
end
|
||||
return table.concat(s, " ")
|
||||
end
|
||||
|
||||
print("\n1) Selection inside one long item -> split x2, middle deleted, tone inserted")
|
||||
local w = run(function(W)
|
||||
local t = track({ {0, 30} })
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 15
|
||||
end)
|
||||
local t1 = w.tracks[1]
|
||||
check("two splits", w.splits == 2, "splits=" .. w.splits)
|
||||
check("one delete", w.deletes == 1, "deletes=" .. w.deletes)
|
||||
check("tone inserted", w.last_inserted ~= nil)
|
||||
check("tone spans selection", math.abs(w.last_inserted.pos - 10) < 1e-6
|
||||
and math.abs(w.last_inserted.len - 5) < 1e-6,
|
||||
("pos=%.3f len=%.3f"):format(w.last_inserted.pos, w.last_inserted.len))
|
||||
check("fades applied", w.last_inserted.fin_ > 0 and w.last_inserted.fout_ > 0)
|
||||
check("no audio left inside range", (function()
|
||||
for _, it in ipairs(t1.items) do
|
||||
if it.tag == "audio" and it.pos >= 10 - 1e-6 and it.pos + it.len <= 15 + 1e-6 then return false end
|
||||
end
|
||||
return true
|
||||
end)())
|
||||
print(" layout: " .. layout(t1))
|
||||
|
||||
print("\n2) Item entirely inside selection -> deleted outright")
|
||||
w = run(function(W)
|
||||
local t = track({ {11, 2} })
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 15
|
||||
end)
|
||||
check("no splits needed", w.splits == 0, "splits=" .. w.splits)
|
||||
check("deleted", w.deletes == 1, "deletes=" .. w.deletes)
|
||||
print(" layout: " .. layout(w.tracks[1]))
|
||||
|
||||
print("\n3) Items straddling each edge only")
|
||||
w = run(function(W)
|
||||
local t = track({ {5, 7}, {13, 6} }) -- 5..12 and 13..19, selection 10..15
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 15
|
||||
end)
|
||||
local t3 = w.tracks[1]
|
||||
check("split each straddler once", w.splits == 2, "splits=" .. w.splits)
|
||||
check("both inner halves deleted", w.deletes == 2, "deletes=" .. w.deletes)
|
||||
check("audio before survives", (function()
|
||||
for _, it in ipairs(t3.items) do
|
||||
if it.tag == "audio" and math.abs(it.pos - 5) < 1e-6 and math.abs(it.len - 5) < 1e-6 then return true end
|
||||
end
|
||||
return false
|
||||
end)())
|
||||
print(" layout: " .. layout(t3))
|
||||
|
||||
print("\n4) Item entirely outside selection -> untouched")
|
||||
w = run(function(W)
|
||||
local t = track({ {20, 5} })
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 15
|
||||
end)
|
||||
check("no splits", w.splits == 0)
|
||||
check("no deletes", w.deletes == 0)
|
||||
check("still one audio item + tone", #w.tracks[1].items == 2, "n=" .. #w.tracks[1].items)
|
||||
|
||||
print("\n5) Only the selected track is touched")
|
||||
w = run(function(W)
|
||||
local a, b = track({ {0, 30} }), track({ {0, 30} })
|
||||
W.tracks = { a, b }; W.sel_tracks = { a }; W.sel_start, W.sel_end = 10, 15
|
||||
end)
|
||||
check("other track untouched", #w.tracks[2].items == 1, "n=" .. #w.tracks[2].items)
|
||||
check("selected track modified", #w.tracks[1].items > 1)
|
||||
|
||||
print("\n6) Guard rails")
|
||||
w = run(function(W)
|
||||
local t = track({ {0, 30} })
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 10 -- empty selection
|
||||
end)
|
||||
check("aborts with message on empty selection", w.msg ~= nil and w.splits == 0, tostring(w.msg))
|
||||
|
||||
w = run(function(W)
|
||||
local t = track({ {0, 30} })
|
||||
W.tracks = { t }; W.sel_tracks = {}; W.sel_start, W.sel_end = 10, 15 -- no track selected
|
||||
end)
|
||||
check("aborts with message on no track", w.msg ~= nil and w.splits == 0, tostring(w.msg))
|
||||
|
||||
print("\n7) Never uses project-wide / UI-level calls that move other tracks")
|
||||
w = run(function(W)
|
||||
local a, b = track({ {0, 30} }), track({ {0, 30} })
|
||||
W.tracks = { a, b }; W.sel_tracks = { a }; W.sel_start, W.sel_end = 10, 15
|
||||
end)
|
||||
check("no InsertMedia (ripples + can spawn tracks)", not w.forbidden["InsertMedia"])
|
||||
check("does not move edit cursor", not w.forbidden["SetEditCurPos"])
|
||||
check("does not change track selection", not w.forbidden["SetOnlyTrackSelected"]
|
||||
and not w.forbidden["SetTrackSelected"])
|
||||
check("built item from PCM source", w.src_path ~= nil and w.last_inserted.take ~= nil)
|
||||
check("tone landed on the SELECTED track", w.last_inserted.owner == w.tracks[1])
|
||||
|
||||
print("\n8) Ripple editing forced off, then restored")
|
||||
w = run(function(W)
|
||||
local t = track({ {0, 30} })
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 15
|
||||
W.ripple[40311] = true -- user had "ripple all tracks" on
|
||||
end)
|
||||
check("turned ripple off first", w.commands[1] == 40309, "cmds=" .. table.concat(w.commands, ","))
|
||||
check("restored ripple-all after", w.commands[#w.commands] == 40311,
|
||||
"cmds=" .. table.concat(w.commands, ","))
|
||||
|
||||
w = run(function(W)
|
||||
local t = track({ {0, 30} })
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 15
|
||||
W.ripple[40310] = true -- per-track ripple
|
||||
end)
|
||||
check("restored per-track ripple", w.commands[1] == 40309 and w.commands[#w.commands] == 40310,
|
||||
"cmds=" .. table.concat(w.commands, ","))
|
||||
|
||||
w = run(function(W)
|
||||
local t = track({ {0, 30} })
|
||||
W.tracks = { t }; W.sel_tracks = { t }; W.sel_start, W.sel_end = 10, 15
|
||||
end)
|
||||
check("ripple already off -> no mode changes at all", #w.commands == 0,
|
||||
"cmds=" .. table.concat(w.commands, ","))
|
||||
|
||||
print(("\n%d passed, %d failed\n"):format(pass, fail))
|
||||
os.exit(fail == 0 and 0 or 1)
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Guards against the two model-config failures that have bitten us:
|
||||
a routed model with no pricing entry (silently costs $0.00 on the dashboard),
|
||||
and a stale model id that 404s at the OpenRouter API."""
|
||||
|
||||
from backend.config import settings
|
||||
from backend.services.cost_tracker import OPENROUTER_PRICING
|
||||
from backend.services.llm import OPENROUTER_MODELS, LLMService
|
||||
|
||||
_CALLER_DIALOG_MODEL_PARAMS = LLMService._CALLER_DIALOG_MODEL_PARAMS
|
||||
|
||||
# Retired on OpenRouter — kept here so a reintroduction fails loudly.
|
||||
RETIRED_MODELS = {
|
||||
"anthropic/claude-3.5-haiku",
|
||||
"anthropic/claude-3.5-sonnet",
|
||||
"anthropic/claude-sonnet-4-5", # note the dash; the live id uses a dot
|
||||
"google/gemini-flash-1.5",
|
||||
"mistralai/mistral-small-creative",
|
||||
"x-ai/grok-4",
|
||||
"x-ai/grok-4-fast",
|
||||
"x-ai/grok-4.1-fast",
|
||||
}
|
||||
|
||||
|
||||
def test_every_routed_model_has_pricing():
|
||||
"""A routed model missing from OPENROUTER_PRICING records cost as $0.00."""
|
||||
unpriced = {
|
||||
cat: model
|
||||
for cat, model in settings.category_models.items()
|
||||
if model not in OPENROUTER_PRICING
|
||||
}
|
||||
assert not unpriced, f"routed models with no pricing entry: {unpriced}"
|
||||
|
||||
|
||||
def test_no_retired_models_are_routed():
|
||||
routed = {
|
||||
cat: model
|
||||
for cat, model in settings.category_models.items()
|
||||
if model in RETIRED_MODELS
|
||||
}
|
||||
assert not routed, f"category routed to a retired model: {routed}"
|
||||
|
||||
|
||||
def test_no_retired_models_in_the_pool():
|
||||
stale = sorted(set(OPENROUTER_MODELS) & RETIRED_MODELS)
|
||||
assert not stale, f"retired models still listed in OPENROUTER_MODELS: {stale}"
|
||||
|
||||
|
||||
def test_no_retired_models_in_caller_dialog_params():
|
||||
stale = sorted(set(_CALLER_DIALOG_MODEL_PARAMS) & RETIRED_MODELS)
|
||||
assert not stale, f"per-model tuning keyed to retired models: {stale}"
|
||||
|
||||
|
||||
def test_pool_models_are_priced():
|
||||
"""Anything selectable should be costable."""
|
||||
unpriced = [m for m in OPENROUTER_MODELS if m not in OPENROUTER_PRICING]
|
||||
assert not unpriced, f"pool models with no pricing entry: {unpriced}"
|
||||
|
||||
|
||||
def test_no_retired_model_ids_anywhere_in_the_codebase():
|
||||
"""publish_episode.py, make_clips.py and relabel_transcripts.py each shipped
|
||||
a retired model id and only failed when someone ran them. Scan every source
|
||||
file so the next one fails here instead.
|
||||
|
||||
cost_tracker.py is exempt: it intentionally keeps retired ids as pricing
|
||||
keys so historical cost records stay costable.
|
||||
"""
|
||||
import pathlib
|
||||
|
||||
root = pathlib.Path(__file__).resolve().parent.parent
|
||||
exempt = {root / "backend" / "services" / "cost_tracker.py",
|
||||
pathlib.Path(__file__).resolve()}
|
||||
|
||||
offenders = {}
|
||||
for path in root.rglob("*.py"):
|
||||
if path in exempt:
|
||||
continue
|
||||
# Skip venvs, vendored models, and git worktrees (separate checkouts
|
||||
# on other branches — not this tree's code).
|
||||
if any(part in {"venv", "mlx_models", ".git", ".claude", ".worktrees",
|
||||
"node_modules", "remotion-demo"} for part in path.parts):
|
||||
continue
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8", errors="ignore")
|
||||
except OSError:
|
||||
continue
|
||||
hits = sorted(m for m in RETIRED_MODELS if f'"{m}"' in text or f"'{m}'" in text)
|
||||
if hits:
|
||||
offenders[str(path.relative_to(root))] = hits
|
||||
|
||||
assert not offenders, f"retired model ids still referenced: {offenders}"
|
||||
@@ -0,0 +1,29 @@
|
||||
from backend.main import _get_town_from_location, BIG_BEND_TOWNS
|
||||
|
||||
|
||||
def test_matches_simple_town():
|
||||
assert _get_town_from_location("Alpine, Texas") == "alpine"
|
||||
assert _get_town_from_location("Marfa") == "marfa"
|
||||
assert _get_town_from_location("outside Terlingua") == "terlingua"
|
||||
|
||||
|
||||
def test_matches_two_word_town():
|
||||
assert _get_town_from_location("Fort Stockton, TX") == "fort stockton"
|
||||
assert _get_town_from_location("ft stockton") == "fort stockton"
|
||||
|
||||
|
||||
def test_case_insensitive():
|
||||
assert _get_town_from_location("MARATHON, texas") == "marathon"
|
||||
|
||||
|
||||
def test_no_match_returns_none():
|
||||
assert _get_town_from_location("Albuquerque, New Mexico") is None
|
||||
assert _get_town_from_location("") is None
|
||||
|
||||
|
||||
def test_every_town_has_coords():
|
||||
for town, coords in BIG_BEND_TOWNS.items():
|
||||
assert town == town.lower()
|
||||
lat, lon = coords
|
||||
assert 28.0 < lat < 32.0
|
||||
assert -105.0 < lon < -102.0
|
||||
@@ -0,0 +1,47 @@
|
||||
from backend.main import (
|
||||
INWORLD_MALE_VOICES,
|
||||
INWORLD_FEMALE_VOICES,
|
||||
BLACKLISTED_VOICES,
|
||||
)
|
||||
|
||||
NEWLY_ADDED = ["Arthur", "Daniel", "Brooke", "Joy", "Selene", "Zadie"]
|
||||
|
||||
# Child / adolescent voices — deliberately kept off the roster for an
|
||||
# adult late-night call-in show.
|
||||
EXCLUDED_CHILD_VOICES = ["Abby", "Mia", "Riley"]
|
||||
|
||||
|
||||
def test_rosters_are_sorted_and_deduped():
|
||||
for roster in (INWORLD_MALE_VOICES, INWORLD_FEMALE_VOICES):
|
||||
assert roster == sorted(roster), "roster must stay alphabetical"
|
||||
assert len(roster) == len(set(roster)), "duplicate voice in roster"
|
||||
|
||||
|
||||
def test_no_voice_appears_in_both_genders():
|
||||
overlap = set(INWORLD_MALE_VOICES) & set(INWORLD_FEMALE_VOICES)
|
||||
assert not overlap, f"voice in both rosters: {overlap}"
|
||||
|
||||
|
||||
def test_newly_added_voices_are_present():
|
||||
roster = set(INWORLD_MALE_VOICES) | set(INWORLD_FEMALE_VOICES)
|
||||
missing = [v for v in NEWLY_ADDED if v not in roster]
|
||||
assert not missing, f"missing: {missing}"
|
||||
|
||||
|
||||
def test_child_voices_stay_off_the_roster():
|
||||
roster = set(INWORLD_MALE_VOICES) | set(INWORLD_FEMALE_VOICES)
|
||||
present = [v for v in EXCLUDED_CHILD_VOICES if v in roster]
|
||||
assert not present, f"child/adolescent voice on roster: {present}"
|
||||
|
||||
|
||||
def test_blacklist_only_references_real_roster_voices():
|
||||
roster = set(INWORLD_MALE_VOICES) | set(INWORLD_FEMALE_VOICES)
|
||||
orphans = sorted(set(BLACKLISTED_VOICES) - roster)
|
||||
assert not orphans, f"blacklist names voices not in the roster: {orphans}"
|
||||
|
||||
|
||||
def test_effective_pool_is_large_enough_for_session_sampling():
|
||||
"""main.py samples 25 voices per session from the non-blacklisted pool."""
|
||||
roster = set(INWORLD_MALE_VOICES) | set(INWORLD_FEMALE_VOICES)
|
||||
effective = roster - set(BLACKLISTED_VOICES)
|
||||
assert len(effective) >= 25
|
||||
@@ -0,0 +1,138 @@
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.main import (
|
||||
Voicemail,
|
||||
_load_voicemails,
|
||||
_save_voicemails,
|
||||
play_voicemail_on_air,
|
||||
)
|
||||
import backend.main as main
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_store(tmp_path, monkeypatch):
|
||||
"""Point the voicemail store at a temp dir and start from a clean slate."""
|
||||
meta = tmp_path / "voicemails.json"
|
||||
monkeypatch.setattr(main, "VOICEMAILS_META", meta)
|
||||
monkeypatch.setattr(main, "_voicemails", [])
|
||||
monkeypatch.setattr(main, "_deleted_vm_timestamps", set())
|
||||
return meta
|
||||
|
||||
|
||||
def test_voicemail_defaults_to_empty_transcript():
|
||||
vm = Voicemail(id="a1", phone="+15551234567", timestamp=1.0, duration=30, file_path="/x.wav")
|
||||
assert vm.transcript == ""
|
||||
|
||||
|
||||
def test_transcript_round_trips_through_save_and_load(isolated_store):
|
||||
main._voicemails.append(
|
||||
Voicemail(
|
||||
id="a1", phone="+15551234567", timestamp=1.0, duration=30,
|
||||
file_path="/x.wav", transcript="Hey Luke, it's Susan out in Deming.",
|
||||
)
|
||||
)
|
||||
_save_voicemails()
|
||||
|
||||
on_disk = json.loads(isolated_store.read_text())
|
||||
assert on_disk["voicemails"][0]["transcript"] == "Hey Luke, it's Susan out in Deming."
|
||||
|
||||
main._voicemails.clear()
|
||||
_load_voicemails()
|
||||
assert main._voicemails[0].transcript == "Hey Luke, it's Susan out in Deming."
|
||||
|
||||
|
||||
def test_load_tolerates_legacy_entries_without_transcript(isolated_store):
|
||||
isolated_store.write_text(json.dumps({
|
||||
"voicemails": [{
|
||||
"id": "old", "phone": "+15551234567", "timestamp": 1.0,
|
||||
"duration": 30, "file_path": "/x.wav", "listened": True,
|
||||
}],
|
||||
"deleted_timestamps": [],
|
||||
}))
|
||||
_load_voicemails()
|
||||
assert main._voicemails[0].transcript == ""
|
||||
|
||||
|
||||
def test_play_on_air_puts_transcript_into_conversation(isolated_store, tmp_path, monkeypatch):
|
||||
"""The actual fix: Devon reads session.conversation, so playing a
|
||||
voicemail on air must write its transcript there."""
|
||||
wav = tmp_path / "vm.wav"
|
||||
wav.write_bytes(b"fake wav bytes")
|
||||
|
||||
main._voicemails.append(
|
||||
Voicemail(
|
||||
id="vm1", phone="+15753424105", timestamp=1.0, duration=77,
|
||||
file_path=str(wav), transcript="My neighbor keeps moving my fence posts.",
|
||||
)
|
||||
)
|
||||
|
||||
monkeypatch.setattr(main.session, "conversation", [])
|
||||
monkeypatch.setattr(main, "_save_voicemails", lambda: None)
|
||||
|
||||
played = {}
|
||||
monkeypatch.setattr(
|
||||
main.audio_service, "play_caller_audio",
|
||||
lambda data, sr: played.update(bytes=len(data), rate=sr),
|
||||
)
|
||||
|
||||
class FakeLibrosa:
|
||||
@staticmethod
|
||||
def load(path, sr=None, mono=True):
|
||||
import numpy as np
|
||||
return np.zeros(sr or 24000, dtype="float32"), sr or 24000
|
||||
|
||||
monkeypatch.setitem(__import__("sys").modules, "librosa", FakeLibrosa)
|
||||
|
||||
result = asyncio.run(play_voicemail_on_air("vm1"))
|
||||
assert result["status"] == "playing"
|
||||
|
||||
roles = [m["role"] for m in main.session.conversation]
|
||||
contents = [m["content"] for m in main.session.conversation]
|
||||
assert any("voicemail" in r for r in roles), f"no voicemail role in {roles}"
|
||||
assert any("fence posts" in c for c in contents), f"transcript missing from {contents}"
|
||||
|
||||
|
||||
def test_voicemail_role_is_normalized_to_a_valid_llm_role():
|
||||
"""A raw 'voicemail:+1555...' role would reach the LLM API as an invalid
|
||||
role; llm.py swallows the resulting error and returns empty text, so
|
||||
callers would silently go mute."""
|
||||
out = main._normalize_messages_for_llm([
|
||||
{"role": "voicemail:+15753424105", "content": "My name is Sandra."},
|
||||
])
|
||||
assert out[0]["role"] in ("user", "assistant", "system"), out[0]["role"]
|
||||
assert "Sandra" in out[0]["content"]
|
||||
assert "+15753424105" in out[0]["content"], "caller identity should survive"
|
||||
|
||||
|
||||
def test_play_on_air_without_transcript_still_announces_the_voicemail(isolated_store, tmp_path, monkeypatch):
|
||||
"""An untranscribed voicemail should still tell Devon something happened,
|
||||
rather than silently playing audio he can't perceive."""
|
||||
wav = tmp_path / "vm.wav"
|
||||
wav.write_bytes(b"fake wav bytes")
|
||||
|
||||
main._voicemails.append(
|
||||
Voicemail(
|
||||
id="vm2", phone="+15753424105", timestamp=1.0, duration=77,
|
||||
file_path=str(wav), transcript="",
|
||||
)
|
||||
)
|
||||
|
||||
monkeypatch.setattr(main.session, "conversation", [])
|
||||
monkeypatch.setattr(main, "_save_voicemails", lambda: None)
|
||||
monkeypatch.setattr(main.audio_service, "play_caller_audio", lambda data, sr: None)
|
||||
|
||||
class FakeLibrosa:
|
||||
@staticmethod
|
||||
def load(path, sr=None, mono=True):
|
||||
import numpy as np
|
||||
return np.zeros(sr or 24000, dtype="float32"), sr or 24000
|
||||
|
||||
monkeypatch.setitem(__import__("sys").modules, "librosa", FakeLibrosa)
|
||||
|
||||
asyncio.run(play_voicemail_on_air("vm2"))
|
||||
|
||||
assert main.session.conversation, "nothing was added to the conversation"
|
||||
assert any("voicemail" in m["role"] for m in main.session.conversation)
|
||||
Reference in New Issue
Block a user