"""Org-scoped lesson audio paths and one-time SD sync.""" import asyncio from types import SimpleNamespace from core.app_runtime.lesson_audio import ( DEFAULT_ORG_ID, course_id_for, missing_lesson_names, resolve_disk_file, resolve_org_id, sanitize_relative_path, unique_lesson_names, ) from core.app_runtime.playback import TEMP_REPEATER_PLAYLIST, TEMP_SHADOWING_PLAYLIST def test_empty_org_is_platform_default(): assert resolve_org_id(None) == DEFAULT_ORG_ID assert resolve_org_id("") == DEFAULT_ORG_ID assert resolve_org_id(" ") == DEFAULT_ORG_ID assert resolve_org_id("school_a") == "school_a" assert resolve_org_id("../etc") == DEFAULT_ORG_ID def test_unique_repeater_names_drop_repeats(): names = unique_lesson_names(TEMP_REPEATER_PLAYLIST) assert names == [ "/english_learning/repeater/lessons/l1/001/cn/01.mp3", "/english_learning/repeater/lessons/l1/001/en/01.mp3", ] def test_unique_shadowing_names_keep_sd_relative_paths(): names = unique_lesson_names(TEMP_SHADOWING_PLAYLIST) assert len(names) == 58 assert names[0] == "/english_learning/shadowing/lessons/l1/001/en/01.mp3" assert names[-1].endswith("/58.mp3") assert all(not name.startswith("/default_org") for name in names) def test_missing_lesson_names_ignore_present_files(): needed = unique_lesson_names(TEMP_REPEATER_PLAYLIST) present = [needed[0], "english_learning/repeater/lessons/l1/001/en/01.mp3"] assert missing_lesson_names(needed, present) == [] assert missing_lesson_names(needed, [needed[0]]) == [needed[1]] def test_sanitize_rejects_traversal_and_non_lesson_paths(): assert sanitize_relative_path("english_learning/repeater/lessons/l1/001/cn/01.mp3") assert sanitize_relative_path("/english_learning/shadowing/lessons/l1/001/en/01.mp3") assert sanitize_relative_path("english_learning/../secret.mp3") is None assert sanitize_relative_path("other/lessons/a.mp3") is None assert sanitize_relative_path("english_learning/repeater/lessons/l1/001/cn.txt") is None def test_course_id_fits_firmware_limit(): assert course_id_for("repeater", "l1.001") == "l1-001-rpt" assert course_id_for("shadowing", "l1.001") == "l1-001-shd" assert len(course_id_for("shadowing", "l1.001")) < 32 def test_resolve_disk_file_falls_back_to_default_org(tmp_path): rel = "english_learning/repeater/lessons/l1/001/cn/01.mp3" platform = tmp_path / "lesson_audio" / DEFAULT_ORG_ID for part in rel.split("/"): platform = platform / part platform.parent.mkdir(parents=True) platform.write_bytes(b"ID3") found = resolve_disk_file({"log": {"data_dir": str(tmp_path)}}, "school_a", rel) assert found and found.replace("\\", "/").endswith("cn/01.mp3") assert DEFAULT_ORG_ID in found.replace("\\", "/") def test_ensure_lesson_mp3s_downloads_missing_once(monkeypatch): from core.app_runtime import runtime lists = [ [], [ "/english_learning/repeater/lessons/l1/001/cn/01.mp3", "/english_learning/repeater/lessons/l1/001/en/01.mp3", ], ] posted = [] async def fake_list(conn): return lists.pop(0) if lists else lists[-1:] or [] async def fake_sync(payload): posted.append(payload) return {"ok": True} monkeypatch.setattr(runtime, "_list_sd_mp3_names", fake_list) monkeypatch.setattr("config.manage_api_client.sync_lesson_audio", fake_sync) conn = SimpleNamespace( config={ "app_runtime": {"lesson_sync_timeout_sec": 0}, "server": { "http_port": 8003, "lesson_audio_public_url": "http://127.0.0.1:8003/lesson-audio", }, }, device_id="aa:bb:cc:dd:ee:ff", ) names = unique_lesson_names(TEMP_REPEATER_PLAYLIST) ok = asyncio.run(runtime._ensure_lesson_mp3s(conn, names, "repeater", {"lesson_id": "l1.001"})) assert ok is True assert len(posted) == 1 assert posted[0]["orgId"] == DEFAULT_ORG_ID assert posted[0]["files"] == names assert posted[0]["baseUrl"].endswith("/lesson-audio/" + DEFAULT_ORG_ID) assert "default_org" not in posted[0]["files"][0][1:] def test_ensure_skips_preload_when_cached(monkeypatch): from core.app_runtime import runtime posted = [] async def fake_list(conn): return unique_lesson_names(TEMP_REPEATER_PLAYLIST) async def fake_sync(payload): posted.append(payload) return {"ok": True} monkeypatch.setattr(runtime, "_list_sd_mp3_names", fake_list) monkeypatch.setattr("config.manage_api_client.sync_lesson_audio", fake_sync) conn = SimpleNamespace(config={"app_runtime": {"lesson_sync_timeout_sec": 0}}, device_id="mac") ok = asyncio.run( runtime._ensure_lesson_mp3s(conn, unique_lesson_names(TEMP_REPEATER_PLAYLIST), "repeater", {}) ) assert ok is True assert posted == [] def test_ensure_skips_sync_when_list_raises(monkeypatch): from core.app_runtime import runtime posted = [] spoken = [] async def fake_list(conn): raise RuntimeError("mcp timeout") async def fake_sync(payload): posted.append(payload) return {"ok": True} monkeypatch.setattr(runtime, "_list_sd_mp3_names", fake_list) monkeypatch.setattr("config.manage_api_client.sync_lesson_audio", fake_sync) monkeypatch.setattr(runtime, "_speak", lambda conn, text: spoken.append(text)) conn = SimpleNamespace(config={"app_runtime": {"lesson_sync_timeout_sec": 0}}, device_id="mac") ok = asyncio.run( runtime._ensure_lesson_mp3s(conn, unique_lesson_names(TEMP_REPEATER_PLAYLIST), "repeater", {}) ) assert ok is True assert posted == [] assert spoken == [] def test_list_lesson_clips_uses_files_on_disk(tmp_path, monkeypatch): from core.app_runtime.lesson_audio import list_lesson_clips from core.app_runtime.playback import expand_lesson_playlist monkeypatch.setattr( "core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: list_lesson_clips( {"log": {"data_dir": str(tmp_path)}}, None, _args[2], _args[3], _args[4] ), ) root = tmp_path / "lesson_audio" / "default_org" cn = root / "english_learning" / "shadowing" / "lessons" / "l1" / "001" / "cn" en = root / "english_learning" / "shadowing" / "lessons" / "l1" / "001" / "en" rpt_cn = root / "english_learning" / "repeater" / "lessons" / "l1" / "002" / "cn" rpt_en = root / "english_learning" / "repeater" / "lessons" / "l1" / "002" / "en" for folder, names in ( (en, ["01.mp3", "03.mp3", "02.mp3"]), (cn, ["01.mp3"]), (rpt_cn, ["01.mp3", "02.mp3"]), (rpt_en, ["01.mp3"]), ): folder.mkdir(parents=True) for name in names: (folder / name).write_bytes(b"ID3") listed = list_lesson_clips({"log": {"data_dir": str(tmp_path)}}, None, "shadowing", "l1.001", "en") assert listed == [ "/english_learning/shadowing/lessons/l1/001/en/01.mp3", "/english_learning/shadowing/lessons/l1/001/en/02.mp3", "/english_learning/shadowing/lessons/l1/001/en/03.mp3", ] shadowing = expand_lesson_playlist("shadowing", "l1.001", language="en", speed=0.8) assert [item["name"] for item in shadowing] == listed repeater = expand_lesson_playlist( "repeater", "l1.002", en_passes=[{"speed": 0.8, "repeat": 1}], ) assert [item["name"] for item in repeater] == [ "/english_learning/repeater/lessons/l1/002/cn/01.mp3", "/english_learning/repeater/lessons/l1/002/cn/02.mp3", "/english_learning/repeater/lessons/l1/002/en/01.mp3", ]