Clément Saillant
ec2dd3f24f
The additive refresh accumulated episodes without bound (Oli reached 13). Tag each manifest episode with a source (auto/pick/upload); the refresh now prunes "auto" episodes that aged out of the feed's latest N while keeping hand-picked and uploaded ones. Feeds default to N=5.
68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
import os, tempfile, json
|
|
os.environ["LISAEL_BASE"] = tempfile.mkdtemp(prefix="merlin_test_")
|
|
os.environ["LISAEL_STAGING"] = os.path.join(os.environ["LISAEL_BASE"], "podcasts")
|
|
import lisael_content as C
|
|
|
|
def test_episode_fname_stable():
|
|
a = C.episode_fname("oli", "http://x/ep1.m4a")
|
|
assert a == C.episode_fname("oli", "http://x/ep1.m4a")
|
|
assert a.endswith("_a48.mp3") and a.startswith("oli_")
|
|
assert a != C.episode_fname("oli", "http://x/ep2.m4a")
|
|
|
|
def test_feeds_default_seed():
|
|
feeds = C.load_feeds()
|
|
assert any(f["key"] == "oli" for f in feeds)
|
|
assert os.path.exists(C.FEEDS_JSON)
|
|
C.add_feed("zz", "Test", "http://r/ss", 1)
|
|
assert any(f["key"] == "zz" for f in C.load_feeds())
|
|
C.remove_feed("zz")
|
|
assert not any(f["key"] == "zz" for f in C.load_feeds())
|
|
|
|
def test_manifest_upsert_remove():
|
|
os.makedirs(C.STAGING, exist_ok=True)
|
|
m = []
|
|
C._manifest_upsert(m, "Oli", "oli.bin", "oli_a_a48.mp3", "Ep A")
|
|
C._manifest_upsert(m, "Oli", "", "oli_b_a48.mp3", "Ep B")
|
|
assert len(m) == 1 and len(m[0]["episodes"]) == 2 and m[0]["cover"] == "oli.bin"
|
|
C.manifest_save(m)
|
|
open(os.path.join(C.STAGING, "oli_a_a48.mp3"), "wb").close()
|
|
C.remove_file("oli_a_a48.mp3")
|
|
m2 = C.manifest_load()
|
|
assert [e["file"] for e in m2[0]["episodes"]] == ["oli_b_a48.mp3"]
|
|
|
|
def test_compute_delta():
|
|
assert C.compute_delta([], []) == []
|
|
assert set(C.compute_delta(["a.mp3", "manifest.json"], [])) == {"a.mp3", "manifest.json"}
|
|
assert C.compute_delta(["a.mp3", "manifest.json"], ["a.mp3"]) == []
|
|
|
|
def test_bad_name_rejected():
|
|
try:
|
|
C.remove_file("../x"); assert False
|
|
except ValueError:
|
|
pass
|
|
|
|
def test_prune_show_caps_auto_keeps_manual():
|
|
os.makedirs(C.STAGING, exist_ok=True)
|
|
m = []
|
|
# 2 auto (one stays, one ages out) + 1 hand-picked old + 1 upload
|
|
C._manifest_upsert(m, "Sh", "sh.bin", "sh_new_a48.mp3", "new", "auto")
|
|
C._manifest_upsert(m, "Sh", "", "sh_old_a48.mp3", "old", "auto")
|
|
C._manifest_upsert(m, "Sh", "", "sh_pick_a48.mp3", "picked", "pick")
|
|
C._manifest_upsert(m, "Sh", "", "sh_up_a48.mp3", "uploaded", "upload")
|
|
C.manifest_save(m)
|
|
for f in ("sh_new_a48.mp3", "sh_old_a48.mp3", "sh_pick_a48.mp3", "sh_up_a48.mp3"):
|
|
open(os.path.join(C.STAGING, f), "wb").close()
|
|
# keep only the new auto episode; old auto should be pruned, manual kept
|
|
removed = C.prune_show("Sh", {"sh_new_a48.mp3"})
|
|
assert removed == 1
|
|
files = {e["file"] for e in C.manifest_load()[0]["episodes"]}
|
|
assert files == {"sh_new_a48.mp3", "sh_pick_a48.mp3", "sh_up_a48.mp3"}
|
|
assert not os.path.exists(os.path.join(C.STAGING, "sh_old_a48.mp3"))
|
|
assert os.path.exists(os.path.join(C.STAGING, "sh_pick_a48.mp3"))
|
|
|
|
if __name__ == "__main__":
|
|
for name, fn in sorted(globals().items()):
|
|
if name.startswith("test_"):
|
|
fn(); print("ok", name)
|
|
print("ALL OK")
|