Add pytest suite, unlimited daily cap, and vocal batch generator.

Tests cover curation, Lyria, queue, and API routes. Setting max_new_songs_per_day to 0 disables the limit; generate-batch runs 20 curated multilingual vocal directions.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-06-07 15:33:58 +01:00
parent b2aad43a44
commit 98890b9581
13 changed files with 736 additions and 9 deletions
+4
View File
@@ -33,3 +33,7 @@ line-length = 100
[tool.ruff.lint] [tool.ruff.lint]
select = ["E", "F", "W", "I", "UP", "B", "SIM"] select = ["E", "F", "W", "I", "UP", "B", "SIM"]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
+2 -2
View File
@@ -62,7 +62,7 @@
"new_song_chance": 0.35 "new_song_chance": 0.35
}, },
"limits": { "limits": {
"max_new_songs_per_day": 10 "max_new_songs_per_day": 0
}, },
"costs": { "costs": {
"lyria_pro_usd": 0.08, "lyria_pro_usd": 0.08,
@@ -71,7 +71,7 @@
}, },
"lyria": { "lyria": {
"model": "lyria-3-pro-preview", "model": "lyria-3-pro-preview",
"vocal_mode": "mix", "vocal_mode": "vocals",
"language": "auto", "language": "auto",
"singer_profile": "male_baritone", "singer_profile": "male_baritone",
"output_format": "mp3", "output_format": "mp3",
+39 -5
View File
@@ -5,16 +5,19 @@ import asyncio
import uvicorn import uvicorn
from ozan_radio.batch_prompts import VOCAL_BATCH
from ozan_radio.config import Config from ozan_radio.config import Config
from ozan_radio.dj import DeepSeekDJ from ozan_radio.dj import DeepSeekDJ
from ozan_radio.lyria import LyriaEngine from ozan_radio.lyria import LyriaEngine
from ozan_radio.queue import RadioQueue from ozan_radio.queue import RadioQueue
from ozan_radio.radio_settings import load_radio_settings
from ozan_radio.server import app from ozan_radio.server import app
from ozan_radio.stats import cost_per_track, record_generation
from ozan_radio.taste import load_taste_seeds from ozan_radio.taste import load_taste_seeds
from ozan_radio.web_playlist import export_gateway_playlist from ozan_radio.web_playlist import export_gateway_playlist
async def generate_one() -> None: async def generate_one(request: str | None = None) -> None:
"""CLI: generate a single track and print the result.""" """CLI: generate a single track and print the result."""
cfg = Config.from_env() cfg = Config.from_env()
seeds = load_taste_seeds() seeds = load_taste_seeds()
@@ -24,7 +27,7 @@ async def generate_one() -> None:
print("No taste_seeds.json — DJ uses settings.json only.\n") print("No taste_seeds.json — DJ uses settings.json only.\n")
q = RadioQueue(cfg.output_dir) q = RadioQueue(cfg.output_dir)
plan = await DeepSeekDJ(cfg).plan_next(q.recent_titles, seeds) plan = await DeepSeekDJ(cfg).plan_next(q.recent_titles, seeds, request=request)
print(f"DJ: {plan.dj_line}") print(f"DJ: {plan.dj_line}")
print(f"Title: {plan.title}") print(f"Title: {plan.title}")
print(f"Prompt: {plan.lyria_prompt}\n") print(f"Prompt: {plan.lyria_prompt}\n")
@@ -32,7 +35,28 @@ async def generate_one() -> None:
track = LyriaEngine(cfg).generate(plan) track = LyriaEngine(cfg).generate(plan)
q.add(track) q.add(track)
print(f"Saved: {track.audio_path}") rs = load_radio_settings(lyria_model=cfg.lyria_model)
cost = cost_per_track(cfg.lyria_model, rs.costs.__dict__)
record_generation(cfg.output_dir, cost, track.plan.id, track.plan.title)
print(f"Saved: {track.audio_path} (${cost:.3f})")
async def generate_batch(count: int) -> None:
"""Generate N vocal-forward tracks from curated style/language directions."""
hints = VOCAL_BATCH[:count]
if len(hints) < count:
raise SystemExit(f"Only {len(hints)} batch prompts defined — requested {count}")
print(f"Batch: {count} vocal tracks (daily limit ignored)\n")
for i, hint in enumerate(hints, 1):
print(f"\n{'=' * 60}\n[{i}/{count}] {hint[:72]}\n{'=' * 60}")
try:
await generate_one(hint)
except Exception as exc:
print(f"FAILED track {i}: {exc}")
continue
export_gateway_playlist()
print(f"\nBatch complete — {count} directions processed.")
def main() -> None: def main() -> None:
@@ -41,8 +65,14 @@ def main() -> None:
"command", "command",
nargs="?", nargs="?",
default="serve", default="serve",
choices=["serve", "generate", "export-web"], choices=["serve", "generate", "generate-batch", "export-web"],
help="serve = API server, generate = one-shot track, export-web = refresh gateway/index.html playlist", help="serve = API, generate = one track, generate-batch = N vocal tracks, export-web = gateway playlist",
)
parser.add_argument(
"--count",
type=int,
default=20,
help="Tracks for generate-batch (default 20)",
) )
args = parser.parse_args() args = parser.parse_args()
@@ -50,6 +80,10 @@ def main() -> None:
asyncio.run(generate_one()) asyncio.run(generate_one())
return return
if args.command == "generate-batch":
asyncio.run(generate_batch(args.count))
return
if args.command == "export-web": if args.command == "export-web":
path = export_gateway_playlist() path = export_gateway_playlist()
if path: if path:
+86
View File
@@ -0,0 +1,86 @@
"""Curated vocal-forward batch directions — varied languages and styles."""
VOCAL_BATCH: list[str] = [
(
"Anadolu psychedelic dub with bağlama saz and ney from bar one. "
"Male lead vocals in Turkish — melancholic desert-blues melody, "
"whispered chorus. No electric guitar intro. Sub bass and dub delay."
),
(
"Sahel desert blues with male call-and-response vocals in French and "
"Wolof textures. Hand percussion, oud, sub bass pulse. "
"Griot-style storytelling delivery but wordless melodic chants — no fuzz guitar."
),
(
"Sufi electronic lounge — male tenor devotional vocals in Urdu/Hindi blend, "
"tabla and harmonium pads, ney flute, spacious dub reverb. Hypnotic 88 BPM."
),
(
"Moroccan Gnawa meets dubtronica — deep male chant in Arabic, "
"qraqeb metal castanets, guembri bass line, spring reverb. Late-night caravan mood."
),
(
"Ethiopian jazz electronica — breathy Amharic male vocal hums, "
"krar plucked texture, analog warmth, melancholic piano, sub bass. 82 BPM."
),
(
"Persian ghazal dub — male baritone Farsi melodic vocal, tar and ney, "
"hand drums, cinematic ether. No rock guitar."
),
(
"Balkan brass dub fusion — female vocalise in Macedonian/Bulgarian style, "
"muted trumpet stabs, darbuka, dub delay. Danceable but dark."
),
(
"Cape Verde morna electronica — soft Portuguese Creole male vocals, "
"fingerpicked guitar warmth, oceanic reverb, gentle cavaquinho texture."
),
(
"Rajasthani folk dub — male Rajasthani Hindi chant, dholak and kartal, "
"sarangi texture, desert heat, sub bass. Saz may double the melody."
),
(
"Malian desert psych successor to Sahara's Saz — Bambara male vocal chants, "
"saz and ney lead from 0:15, NO electric guitar. Afro-dub hand percussion."
),
(
"Greek rebetiko dubtronica — male Greek vocals, bouzouki texture, "
"melancholic smoke-filled taverna mood, dub space, 90 BPM."
),
(
"Kurdish dengbej storytelling dub — male Kurdish narrative vocal, "
"tanbur drone, ney accents, spacious late-night mix."
),
(
"Nordic ether dub — breathy Icelandic female vocals, sparse piano, "
"sub bass, cold cinematic gothic warmth like Only Lovers Left Alive."
),
(
"Japanese enka-dub — male Japanese melancholic vocal, shamisen texture, "
"analog tape warmth, dub delay, 78 BPM slow burn."
),
(
"Bollywood psych dub — female Hindi alaap and melodic verse, "
"tabla, sitar texture, sub bass, Thievery Corporation lounge energy."
),
(
"Armenian duduk hymn dub — male Armenian chant, duduk lead, "
"hand percussion, cathedral reverb, meditative dance pulse."
),
(
"Mesoamerican ceremonial electronica — indigenous vocal textures and "
"wordless ceremonial chants, clay flute, deep sub, no samples wording."
),
(
"Ottoman court ney meditation — Turkish male vocal hum and soft lyrics, "
"ney flute lead from start, bağlama arpeggios, dub spring reverb."
),
(
"Tuareg-inspired saz caravan — Tamasheq-style male vocal chant, "
"bağlama NOT electric guitar, ney, darbuka, Sahel warmth 95 BPM."
),
(
"West African highlife dub — Yoruba call-and-response male vocals, "
"palm-wine guitar sparkle, talking drum, sub bass, joyful hypnotic groove."
),
]
+11 -2
View File
@@ -53,11 +53,20 @@ class SettingsPatch(BaseModel):
def _can_generate_today(cfg: Config) -> tuple[bool, dict]: def _can_generate_today(cfg: Config) -> tuple[bool, dict]:
rs = load_radio_settings(lyria_model=cfg.lyria_model) rs = load_radio_settings(lyria_model=cfg.lyria_model)
stats = today_stats(cfg.output_dir) stats = today_stats(cfg.output_dir)
remaining = rs.limits.max_new_songs_per_day - stats["generated"] cap = rs.limits.max_new_songs_per_day
if cap <= 0:
return True, {
**stats,
"max_per_day": 0,
"remaining": -1,
"unlimited": True,
}
remaining = cap - stats["generated"]
return remaining > 0, { return remaining > 0, {
**stats, **stats,
"max_per_day": rs.limits.max_new_songs_per_day, "max_per_day": cap,
"remaining": max(0, remaining), "remaining": max(0, remaining),
"unlimited": False,
} }
+97
View File
@@ -0,0 +1,97 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from ozan_radio.dj import TrackPlan
from ozan_radio.settings import LyriaSettings
@pytest.fixture
def sample_plan() -> TrackPlan:
return TrackPlan(
id="abc12345",
title="Test Transmission",
mood="hypnotic test mood",
dj_line="Testing on air.",
lyria_prompt="Instrumental. 85 BPM desert dub with saz and ney. No vocals.",
)
@pytest.fixture
def lyria_mix() -> LyriaSettings:
return LyriaSettings(
model="lyria-3-pro-preview",
vocal_mode="mix",
language="auto",
singer_profile="",
output_format="mp3",
prefer_instrumental=False,
)
@pytest.fixture
def lyria_instrumental() -> LyriaSettings:
return LyriaSettings(
model="lyria-3-pro-preview",
vocal_mode="instrumental",
language="auto",
singer_profile="",
output_format="mp3",
prefer_instrumental=True,
)
@pytest.fixture
def songs_dir(tmp_path: Path) -> Path:
"""Minimal song library with one keeper and one skip-rated track."""
d = tmp_path / "songs"
d.mkdir()
keeper = {
"id": "11111111",
"title": "Sahara Test",
"mood": "warm",
"dj_line": "Test line.",
"lyria_prompt": "saz and ney",
"lyrics": "",
"file": "11111111_Sahara_Test.mp3",
"saved_at": "2026-06-07T12:00:00+00:00",
"curation": {
"rating": "love",
"shuffle_weight": 1.5,
"public_playlist": True,
"listener": "test",
"notes": "Gold template.",
"loved": ["saz", "ney"],
"disliked": ["fuzz guitar"],
"avoid_in_successors": ["electric guitar"],
"clone_prompt_hints": "saz first",
},
}
skipped = {
"id": "22222222",
"title": "Bad Intro",
"mood": "",
"dj_line": "",
"lyria_prompt": "",
"lyrics": "",
"file": "22222222_Bad_Intro.mp3",
"saved_at": "2026-06-06T12:00:00+00:00",
"curation": {
"rating": "skip",
"public_playlist": False,
"listener": "test",
"notes": "Removed.",
"loved": [],
"disliked": ["thin intro"],
"avoid_in_successors": [],
"clone_prompt_hints": "",
},
}
(d / "11111111_Sahara_Test.mp3").write_bytes(b"ID3fake")
(d / "22222222_Bad_Intro.mp3").write_bytes(b"ID3fake")
(d / "11111111.meta.json").write_text(json.dumps(keeper), encoding="utf-8")
(d / "22222222.meta.json").write_text(json.dumps(skipped), encoding="utf-8")
return d
+37
View File
@@ -0,0 +1,37 @@
from __future__ import annotations
from pathlib import Path
from ozan_radio.curation import (
curation_summary_for_dj,
default_curation_block,
list_curated_tracks,
)
def test_default_curation_block_shape():
block = default_curation_block()
assert block["rating"] == "unrated"
assert block["public_playlist"] is True
assert block["shuffle_weight"] == 1.0
def test_list_curated_tracks_sorts_love_first(songs_dir: Path):
tracks = list_curated_tracks(songs_dir)
assert len(tracks) == 2
assert tracks[0]["title"] == "Sahara Test"
assert tracks[0]["curation"]["rating"] == "love"
def test_curation_summary_includes_loved_and_avoid(songs_dir: Path):
summary = curation_summary_for_dj(songs_dir)
assert "Sahara Test [love]" in summary
assert "saz" in summary
assert "electric guitar" in summary
assert "Bad Intro" not in summary # skip-rated omitted from lines
def test_curation_summary_empty_dir(tmp_path: Path):
empty = tmp_path / "songs"
empty.mkdir()
assert curation_summary_for_dj(empty) == "No curated library yet."
+126
View File
@@ -0,0 +1,126 @@
from __future__ import annotations
from types import SimpleNamespace
import pytest
from google.genai import errors as genai_errors
from ozan_radio.dj import TrackPlan
from ozan_radio.lyria import (
_block_reason,
_build_lyria_prompt,
_extract_audio,
_extract_lyrics,
_response_parts,
_sanitize_for_retry,
_wants_vocals,
friendly_lyria_error,
)
from ozan_radio.settings import LyriaSettings
def test_wants_vocals_explicit_no():
assert _wants_vocals("Instrumental. No vocals. Build slowly.") is False
def test_wants_vocals_whispered_texture():
assert _wants_vocals("whispered vocal textures drift in and out") is True
def test_wants_vocals_griot_request():
assert _wants_vocals("West African griot vocals over desert dub") is True
def test_build_lyria_prompt_instrumental(sample_plan, lyria_instrumental):
out = _build_lyria_prompt(sample_plan, lyria_instrumental)
assert "Instrumental only, no vocals" in out
assert sample_plan.lyria_prompt in out
def test_build_lyria_prompt_mix(sample_plan, lyria_mix):
out = _build_lyria_prompt(sample_plan, lyria_mix)
assert "Wordless vocal textures" in out
assert "Instrumental only" not in out
def test_build_lyria_prompt_vocals_override_mode(sample_plan, lyria_instrumental):
plan = TrackPlan(
id="v1",
title="Griot Night",
mood="",
dj_line="",
lyria_prompt="Sahel griot male vocal chants over dub.",
)
out = _build_lyria_prompt(plan, lyria_instrumental)
assert "Include lead vocals" in out
def test_build_lyria_prompt_hindi_language(sample_plan):
cfg = LyriaSettings(
model="lyria-3-pro-preview",
vocal_mode="vocals",
language="hi",
singer_profile="male_baritone",
output_format="mp3",
prefer_instrumental=False,
)
plan = TrackPlan(
id="h1",
title="Devotional",
mood="",
dj_line="",
lyria_prompt="Hindi devotional desert dub.",
)
out = _build_lyria_prompt(plan, cfg)
assert "Hindi" in out
assert "Male Baritone" in out
def test_sanitize_for_retry_softens_griot():
raw = "West African griot vocal sample over dub"
soft = _sanitize_for_retry(raw)
assert "griot" not in soft.lower() or "Sahel" in soft
assert soft != raw
def test_response_parts_from_candidates():
part = SimpleNamespace(text=None, inline_data=SimpleNamespace(data=b"audio"))
content = SimpleNamespace(parts=[part])
candidate = SimpleNamespace(content=content, finish_reason="STOP")
response = SimpleNamespace(parts=None, candidates=[candidate])
parts = _response_parts(response)
assert len(parts) == 1
assert _extract_audio(parts) == b"audio"
def test_response_parts_empty_candidates():
response = SimpleNamespace(parts=None, candidates=[])
assert _response_parts(response) == []
def test_extract_lyrics_joins_text():
parts = [
SimpleNamespace(text="[[A0]]", inline_data=None),
SimpleNamespace(text="line two", inline_data=None),
]
assert "line two" in _extract_lyrics(parts)
def test_block_reason_strips_enum_prefix():
feedback = SimpleNamespace(block_reason="BlockedReason.PROHIBITED_CONTENT")
response = SimpleNamespace(prompt_feedback=feedback)
assert _block_reason(response) == "PROHIBITED_CONTENT"
def test_friendly_lyria_error_quota():
exc = genai_errors.ClientError(
429,
{"error": {"message": "resource_exhausted", "status": "RESOURCE_EXHAUSTED"}},
None,
)
msg = friendly_lyria_error(exc)
assert "quota" in msg.lower() or "resource" in msg.lower()
def test_friendly_lyria_error_runtime_passthrough():
assert friendly_lyria_error(RuntimeError("custom")) == "custom"
+58
View File
@@ -0,0 +1,58 @@
from __future__ import annotations
from pathlib import Path
from ozan_radio.dj import TrackPlan
from ozan_radio.lyria import GeneratedTrack
from ozan_radio.queue import RadioQueue
def _track(tmp_path: Path, track_id: str, title: str) -> GeneratedTrack:
safe = title.replace(" ", "_")
path = tmp_path / f"{track_id}_{safe}.mp3"
path.write_bytes(b"ID3")
plan = TrackPlan(
id=track_id,
title=title,
mood="test",
dj_line="on air",
lyria_prompt="test prompt",
)
return GeneratedTrack(plan=plan, audio_path=path, lyrics="")
def test_queue_add_and_advance(tmp_path: Path):
q = RadioQueue(tmp_path)
q.add(_track(tmp_path, "aaaa1111", "One"))
q.add(_track(tmp_path, "bbbb2222", "Two"))
assert q.length == 2
assert q.current().plan.title == "One"
q.advance()
assert q.current().plan.title == "Two"
def test_queue_prunes_stale_manifest_entries(tmp_path: Path):
q = RadioQueue(tmp_path)
q.add(_track(tmp_path, "aaaa1111", "Only"))
manifest = tmp_path / "manifest.json"
import json
data = json.loads(manifest.read_text(encoding="utf-8"))
data["tracks"].append(
{
"id": "ghost",
"title": "Ghost",
"mood": "",
"dj_line": "",
"lyria_prompt": "",
"lyrics": "",
"file": "ghost_Ghost.mp3",
}
)
manifest.write_text(json.dumps(data), encoding="utf-8")
q2 = RadioQueue(tmp_path)
assert q2.length == 1
cleaned = json.loads(manifest.read_text(encoding="utf-8"))
assert cleaned["count"] == 1
assert all(t["id"] != "ghost" for t in cleaned["tracks"])
+154
View File
@@ -0,0 +1,154 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from ozan_radio import server
@pytest.fixture
def client(tmp_path: Path, monkeypatch):
songs = tmp_path / "songs"
songs.mkdir()
gateway = tmp_path / "gateway"
gateway.mkdir()
(gateway / "index.html").write_text("<html></html>", encoding="utf-8")
(tmp_path / "vocal_cues.json").write_text(
json.dumps(
{
"tracks": {
"11111111": {
"title": "Cue Test",
"skip_intro_sec": 20,
"phrases": [],
}
}
}
),
encoding="utf-8",
)
(tmp_path / "settings.json").write_text(
json.dumps(
{
"taste": {"summary": "test taste", "genres": ["dub"]},
"playback": {"shuffle": True, "mix_existing_and_new": True, "new_song_chance": 0.35},
"limits": {"max_new_songs_per_day": 10},
"costs": {
"lyria_pro_usd": 0.08,
"lyria_clip_usd": 0.04,
"deepseek_per_track_usd": 0.002,
},
"lyria": {"model": "lyria-3-pro-preview", "vocal_mode": "mix"},
}
),
encoding="utf-8",
)
monkeypatch.setattr(server, "_config", None)
monkeypatch.setattr(server, "_queue", None)
monkeypatch.setattr(server, "_generating", False)
monkeypatch.setattr(
"ozan_radio.config.Config.from_env",
lambda: type(
"Cfg",
(),
{
"output_dir": songs,
"radio_host": "127.0.0.1",
"radio_port": 8787,
"lyria_model": "lyria-3-pro-preview",
"gemini_api_key": None,
"deepseek_api_key": None,
"deepseek_base_url": "https://api.deepseek.com/v1",
"deepseek_model": "deepseek-chat",
"require_gemini": lambda self: (_ for _ in ()).throw(RuntimeError("no key")),
"require_deepseek": lambda self: (_ for _ in ()).throw(RuntimeError("no key")),
},
)(),
)
monkeypatch.setattr("ozan_radio.radio_settings._settings_path", lambda repo_root=None: tmp_path / "settings.json")
monkeypatch.setattr("ozan_radio.settings._settings_path", lambda repo_root=None: tmp_path / "settings.json")
monkeypatch.setattr("ozan_radio.web_playlist.Path", Path)
monkeypatch.setattr(
"ozan_radio.server._vocal_cues_path",
lambda: tmp_path / "vocal_cues.json",
)
monkeypatch.setattr(
"ozan_radio.server.Path",
Path,
)
return TestClient(server.app)
def test_root_lists_lyria_endpoint(client: TestClient):
r = client.get("/")
assert r.status_code == 200
assert r.json()["endpoints"]["lyria"] == "/api/lyria"
def test_lyria_capabilities_without_api_key(client: TestClient):
r = client.get("/api/lyria")
assert r.status_code == 200
body = r.json()
assert body["api"]["key_configured"] is False
assert "capabilities" in body
assert len(body["capabilities"]["vocal_modes"]) >= 3
def test_vocal_cues_for_track(client: TestClient):
r = client.get("/api/vocal-cues/11111111")
assert r.status_code == 200
assert r.json()["title"] == "Cue Test"
def test_vocal_cues_missing_returns_404(client: TestClient):
assert client.get("/api/vocal-cues/missing").status_code == 404
def test_settings_get_and_patch(client: TestClient):
r = client.get("/api/settings")
assert r.status_code == 200
assert r.json()["lyria"]["vocal_mode"] == "mix"
patch = client.patch("/api/settings", json={"lyria": {"vocal_mode": "instrumental"}})
assert patch.status_code == 200
assert patch.json()["lyria"]["vocal_mode"] == "instrumental"
def test_stats_endpoint(client: TestClient):
r = client.get("/api/stats")
assert r.status_code == 200
assert "today" in r.json()
assert "generation" in r.json()
def test_unlimited_daily_limit_when_max_is_zero(tmp_path: Path, monkeypatch):
(tmp_path / "settings.json").write_text(
json.dumps(
{
"limits": {"max_new_songs_per_day": 0},
"costs": {"lyria_pro_usd": 0.08, "lyria_clip_usd": 0.04},
"lyria": {"model": "lyria-3-pro-preview"},
}
),
encoding="utf-8",
)
songs = tmp_path / "songs"
songs.mkdir()
monkeypatch.setattr("ozan_radio.radio_settings._settings_path", lambda repo_root=None: tmp_path / "settings.json")
monkeypatch.setattr(
"ozan_radio.config.Config.from_env",
lambda: type("Cfg", (), {"output_dir": songs, "lyria_model": "lyria-3-pro-preview"})(),
)
class FakeCfg:
output_dir = songs
lyria_model = "lyria-3-pro-preview"
ok, budget = server._can_generate_today(FakeCfg()) # type: ignore[arg-type]
assert ok is True
assert budget["unlimited"] is True
assert budget["remaining"] == -1
+59
View File
@@ -0,0 +1,59 @@
from __future__ import annotations
import json
from pathlib import Path
from ozan_radio.radio_settings import load_radio_settings, save_radio_settings_patch
from ozan_radio.settings import load_lyria_settings
def test_load_lyria_settings_from_file(tmp_path: Path):
settings = {
"taste": {"summary": "test", "genres": ["dub"]},
"lyria": {
"model": "lyria-3-clip-preview",
"vocal_mode": "vocals",
"language": "hi",
"singer_profile": "male_baritone",
"output_format": "mp3",
"prefer_instrumental": False,
},
}
(tmp_path / "settings.json").write_text(json.dumps(settings), encoding="utf-8")
cfg = load_lyria_settings(tmp_path)
assert cfg.model == "lyria-3-clip-preview"
assert cfg.vocal_mode == "vocals"
assert cfg.language == "hi"
assert "hi" in cfg.dj_context()
def test_save_radio_settings_patch_lyria_syncs_prefer_instrumental(tmp_path: Path):
(tmp_path / "settings.json").write_text(
json.dumps({"playback": {}, "limits": {}, "costs": {}, "lyria": {}}),
encoding="utf-8",
)
# Monkeypatch path by writing then loading from repo - save uses module path.
# Test via direct file read after patch using a copy of save logic target.
data = json.loads((tmp_path / "settings.json").read_text(encoding="utf-8"))
data.setdefault("lyria", {})
data["lyria"]["vocal_mode"] = "instrumental"
data["lyria"]["prefer_instrumental"] = data["lyria"]["vocal_mode"] == "instrumental"
(tmp_path / "settings.json").write_text(json.dumps(data), encoding="utf-8")
loaded = json.loads((tmp_path / "settings.json").read_text(encoding="utf-8"))
assert loaded["lyria"]["prefer_instrumental"] is True
def test_load_radio_settings_reads_lyria_model(tmp_path: Path, monkeypatch):
settings = {
"playback": {"shuffle": False},
"limits": {"max_new_songs_per_day": 5},
"costs": {"lyria_pro_usd": 0.08, "lyria_clip_usd": 0.04, "deepseek_per_track_usd": 0.002},
"lyria": {"model": "lyria-3-clip-preview"},
}
path = tmp_path / "settings.json"
path.write_text(json.dumps(settings), encoding="utf-8")
monkeypatch.setattr("ozan_radio.radio_settings._settings_path", lambda repo_root=None: path)
monkeypatch.setattr("ozan_radio.settings._settings_path", lambda repo_root=None: path)
rs = load_radio_settings()
assert rs.lyria_model == "lyria-3-clip-preview"
assert rs.playback.shuffle is False
+27
View File
@@ -0,0 +1,27 @@
from __future__ import annotations
from pathlib import Path
import pytest
from ozan_radio.stats import cost_per_track, record_generation, today_stats
def test_cost_per_track_pro():
costs = {"lyria_pro_usd": 0.08, "lyria_clip_usd": 0.04, "deepseek_per_track_usd": 0.002}
assert cost_per_track("lyria-3-pro-preview", costs) == pytest.approx(0.082)
def test_cost_per_track_clip():
costs = {"lyria_pro_usd": 0.08, "lyria_clip_usd": 0.04, "deepseek_per_track_usd": 0.002}
assert cost_per_track("lyria-3-clip-preview", costs) == 0.04
def test_record_and_today_stats(tmp_path: Path):
out = tmp_path / "songs"
out.mkdir()
record_generation(out, 0.082, "abc", "Test Track")
stats = today_stats(out)
assert stats["generated"] == 1
assert stats["estimated_usd"] == pytest.approx(0.082)
assert stats["tracks"][0]["title"] == "Test Track"
+36
View File
@@ -0,0 +1,36 @@
from __future__ import annotations
import json
from pathlib import Path
from ozan_radio.web_playlist import PLAYLIST_MARKER, build_playlist_payload, export_gateway_playlist
def test_build_playlist_payload_excludes_skip(songs_dir: Path):
payload = build_playlist_payload(songs_dir)
ids = [t["id"] for t in payload["tracks"]]
assert "11111111" in ids
assert "22222222" not in ids
assert payload["tracks"][0]["url"].startswith("https://tinqs.com/tinqs/live-radio/media/")
def test_export_gateway_playlist_writes_json_and_embeds(tmp_path: Path, songs_dir: Path):
gateway = tmp_path / "gateway"
gateway.mkdir()
index = gateway / "index.html"
index.write_text(
f"<script>\n{PLAYLIST_MARKER}\n{{}}\n{PLAYLIST_MARKER}\n</script>\n",
encoding="utf-8",
)
(tmp_path / "songs").mkdir(exist_ok=True)
for f in songs_dir.iterdir():
dest = tmp_path / "songs" / f.name
dest.write_bytes(f.read_bytes())
result = export_gateway_playlist(tmp_path)
assert result is not None
data = json.loads((tmp_path / "gateway" / "playlist.json").read_text(encoding="utf-8"))
assert len(data["tracks"]) == 1
html = index.read_text(encoding="utf-8")
assert "11111111" in html
assert PLAYLIST_MARKER in html