"""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",
|
]
|
