997 lines
41 KiB
Python
997 lines
41 KiB
Python
import importlib
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
downloads = tmp_path / "downloads"
|
|
downloads.mkdir()
|
|
monkeypatch.setenv("WORDARR_DOWNLOAD_DIR", str(downloads))
|
|
monkeypatch.setenv("WORDARR_DB_PATH", str(tmp_path / "test.db"))
|
|
|
|
from wordarr import config
|
|
importlib.reload(config)
|
|
from wordarr import db
|
|
db.init_db(tmp_path / "test.db")
|
|
from wordarr import main
|
|
importlib.reload(main)
|
|
|
|
with TestClient(main.app) as c:
|
|
c.tmp_path = tmp_path
|
|
c.downloads = downloads
|
|
yield c
|
|
|
|
|
|
def test_full_import_flow(client):
|
|
library_root = client.tmp_path / "library" / "kids"
|
|
resp = client.post("/api/libraries", json={
|
|
"name": "Kids", "media_type": "ebook", "root_path": str(library_root),
|
|
})
|
|
assert resp.status_code == 200
|
|
lib_id = resp.json()["id"]
|
|
|
|
resp = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "The Hobbit",
|
|
"authors": "J.R.R. Tolkien", "year": 1937, "external_id": "9780261103573",
|
|
})
|
|
assert resp.status_code == 200
|
|
req_id = resp.json()["id"]
|
|
assert resp.json()["status"] == "missing"
|
|
|
|
(client.downloads / "J.R.R. Tolkien - The Hobbit [retail].epub").write_bytes(b"fake epub")
|
|
|
|
resp = client.get("/api/import/scan")
|
|
items = resp.json()["items"]
|
|
assert len(items) == 1
|
|
assert items[0]["suggested_request_id"] == req_id
|
|
|
|
resp = client.post("/api/import", json={"items": [{
|
|
"path": items[0]["path"], "is_dir": False,
|
|
"files": items[0]["files"], "request_id": req_id,
|
|
}]})
|
|
assert resp.status_code == 200
|
|
result = resp.json()["results"][0]
|
|
assert result["ok"], result
|
|
|
|
dest = library_root / "J.R.R. Tolkien" / "The Hobbit (1937)" / "J.R.R. Tolkien - The Hobbit.epub"
|
|
assert dest.is_file()
|
|
assert not (client.downloads / "J.R.R. Tolkien - The Hobbit [retail].epub").exists()
|
|
|
|
resp = client.get("/api/requests", params={"status": "imported"})
|
|
assert [r["id"] for r in resp.json()] == [req_id]
|
|
|
|
|
|
def test_audiobook_folder_import(client):
|
|
root = client.tmp_path / "library" / "audio"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "English", "media_type": "audiobook", "root_path": str(root),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Dune", "authors": "Frank Herbert",
|
|
}).json()["id"]
|
|
|
|
book_dir = client.downloads / "Frank Herbert - Dune (Unabridged)"
|
|
book_dir.mkdir()
|
|
for i in range(3):
|
|
(book_dir / f"track{i}.mp3").write_bytes(b"audio")
|
|
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert len(items) == 1
|
|
assert items[0]["is_dir"] is True
|
|
assert items[0]["media_type"] == "audiobook"
|
|
|
|
resp = client.post("/api/import", json={"items": [{
|
|
"path": items[0]["path"], "is_dir": True,
|
|
"files": items[0]["files"], "request_id": req_id,
|
|
}]})
|
|
assert resp.json()["results"][0]["ok"]
|
|
|
|
dest = root / "Frank Herbert" / "Dune"
|
|
assert sorted(p.name for p in dest.iterdir()) == [
|
|
"Dune - Part 01.mp3", "Dune - Part 02.mp3", "Dune - Part 03.mp3",
|
|
]
|
|
assert not book_dir.exists()
|
|
|
|
|
|
def test_import_rejects_path_outside_download_dir(client):
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "E", "media_type": "ebook", "root_path": str(client.tmp_path / "lib"),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={"library_id": lib_id, "title": "X"}).json()["id"]
|
|
outside = client.tmp_path / "outside.epub"
|
|
outside.write_bytes(b"x")
|
|
resp = client.post("/api/import", json={"items": [{
|
|
"path": str(outside), "is_dir": False, "files": [str(outside)], "request_id": req_id,
|
|
}]})
|
|
assert resp.json()["results"][0]["ok"] is False
|
|
|
|
|
|
def test_bulk_requests_and_split_scan(client):
|
|
root = client.tmp_path / "library" / "audio"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "Hoerspiele", "media_type": "audiobook", "root_path": str(root),
|
|
}).json()["id"]
|
|
resp = client.post("/api/requests/bulk", json={
|
|
"library_id": lib_id, "series": "Die drei ???", "volume_from": 1, "volume_to": 3,
|
|
})
|
|
assert resp.status_code == 200
|
|
created = resp.json()
|
|
assert [r["title"] for r in created] == [
|
|
"Die drei ??? Folge 01", "Die drei ??? Folge 02", "Die drei ??? Folge 03",
|
|
]
|
|
|
|
folder = client.downloads / "Die drei Fragezeichen Sammlung"
|
|
folder.mkdir()
|
|
for n in (1, 2, 3):
|
|
(folder / f"Die drei Fragezeichen - Folge {n:03d}.mp3").write_bytes(b"a")
|
|
|
|
# default scan: one audiobook unit; split scan: three individual files
|
|
assert len(client.get("/api/import/scan").json()["items"]) == 1
|
|
items = client.get("/api/import/scan", params={"split_dirs": "true"}).json()["items"]
|
|
assert len(items) == 3
|
|
by_name = {i["name"]: i["suggested_request_id"] for i in items}
|
|
expected = {f"Die drei Fragezeichen - Folge {n:03d}": created[n - 1]["id"] for n in (1, 2, 3)}
|
|
assert by_name == expected
|
|
|
|
|
|
def test_audio_tags_written_on_import(client):
|
|
from mutagen.id3 import ID3
|
|
|
|
root = client.tmp_path / "library" / "kids_audio"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "KidsAudio", "media_type": "audiobook", "root_path": str(root),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Die drei ??? Kids Folge 085",
|
|
"authors": "Ulf Blanck", "series": "Die drei ??? Kids", "volume": 85,
|
|
}).json()["id"]
|
|
|
|
book_dir = client.downloads / "Kids 85 Schnueffler"
|
|
book_dir.mkdir()
|
|
for i in range(2):
|
|
(book_dir / f"track{i}.mp3").write_bytes(b"")
|
|
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
resp = client.post("/api/import", json={"items": [{
|
|
"path": items[0]["path"], "is_dir": True,
|
|
"files": items[0]["files"], "request_id": req_id,
|
|
}]})
|
|
assert resp.json()["results"][0]["ok"]
|
|
|
|
dest = root / "Ulf Blanck" / "Die drei ??? Kids Folge 085"
|
|
files = sorted(dest.iterdir())
|
|
tags = ID3(files[0])
|
|
assert str(tags["TALB"]) == "Die drei ??? Kids Folge 085"
|
|
assert str(tags["TPE1"]) == "Ulf Blanck"
|
|
assert str(tags["TXXX:SERIES"]) == "Die drei ??? Kids"
|
|
assert str(tags["TXXX:SERIES-PART"]) == "85"
|
|
assert str(tags["TRCK"]) == "1/2"
|
|
|
|
|
|
def test_narrator_in_path_and_tags(client):
|
|
from mutagen.id3 import ID3
|
|
|
|
root = client.tmp_path / "library" / "hp"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "HP", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Author}/{Title} ({Narrator})", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
beck = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Harry Potter und der Stein der Weisen",
|
|
"authors": "J.K. Rowling", "narrator": "Rufus Beck",
|
|
}).json()
|
|
client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Harry Potter und der Stein der Weisen",
|
|
"authors": "J.K. Rowling", "narrator": "Felix von Manteuffel",
|
|
})
|
|
|
|
f = client.downloads / "Harry Potter Stein der Weisen (Rufus Beck).mp3"
|
|
f.write_bytes(b"")
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert items[0]["suggested_request_id"] == beck["id"]
|
|
|
|
resp = client.post("/api/import", json={"items": [{
|
|
"path": items[0]["path"], "is_dir": False,
|
|
"files": items[0]["files"], "request_id": beck["id"],
|
|
}]})
|
|
assert resp.json()["results"][0]["ok"]
|
|
dest = (root / "J.K. Rowling"
|
|
/ "Harry Potter und der Stein der Weisen (Rufus Beck)"
|
|
/ "Harry Potter und der Stein der Weisen.mp3")
|
|
assert dest.is_file()
|
|
assert str(ID3(dest)["TCOM"]) == "Rufus Beck"
|
|
|
|
|
|
def test_scan_rel_dir_and_request_update(client):
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "A", "media_type": "audiobook", "root_path": str(client.tmp_path / "a"),
|
|
}).json()["id"]
|
|
|
|
nested = client.downloads / "Sammlung" / "Sonderfolgen" / "CD1"
|
|
nested.mkdir(parents=True)
|
|
for i in range(2):
|
|
(nested / f"t{i}.mp3").write_bytes(b"")
|
|
item = client.get("/api/import/scan").json()["items"][0]
|
|
# a disc subfolder is folded into its parent, which carries the title
|
|
assert item["name"] == "Sonderfolgen"
|
|
assert item["rel_dir"] == "Sammlung"
|
|
|
|
req = client.post("/api/requests", json={"library_id": lib_id, "title": "X"}).json()
|
|
updated = client.put(f"/api/requests/{req['id']}", json={
|
|
"library_id": lib_id, "title": "X", "narrator": "Rufus Beck",
|
|
"series": "HP", "volume": 1,
|
|
}).json()
|
|
assert updated["narrator"] == "Rufus Beck"
|
|
assert updated["series"] == "HP"
|
|
assert client.get("/api/requests", params={"status": "missing"}).json()[0]["narrator"] == "Rufus Beck"
|
|
|
|
|
|
def test_retag_after_update(client):
|
|
from mutagen.id3 import ID3
|
|
|
|
root = client.tmp_path / "library" / "retag"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "Retag", "media_type": "audiobook", "root_path": str(root),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Der Ickabog", "authors": "J.K. Rowling",
|
|
}).json()["id"]
|
|
|
|
book_dir = client.downloads / "Der Ickabog"
|
|
book_dir.mkdir()
|
|
for i in range(2):
|
|
(book_dir / f"t{i}.mp3").write_bytes(b"")
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
client.post("/api/import", json={"items": [{
|
|
"path": items[0]["path"], "is_dir": True,
|
|
"files": items[0]["files"], "request_id": req_id,
|
|
}]})
|
|
|
|
# no series yet -> tag absent
|
|
dest = root / "J.K. Rowling" / "Der Ickabog"
|
|
first = sorted(dest.iterdir())[0]
|
|
assert "TXXX:SERIES" not in ID3(first)
|
|
|
|
# add series + narrator, then retag
|
|
client.put(f"/api/requests/{req_id}", json={
|
|
"library_id": lib_id, "title": "Der Ickabog", "authors": "J.K. Rowling",
|
|
"narrator": "Ben Becker", "series": "Rowling Kinderbücher", "volume": 2,
|
|
})
|
|
res = client.post(f"/api/requests/{req_id}/retag").json()
|
|
assert res == {"ok": True, "files": 2}
|
|
tags = ID3(first)
|
|
assert str(tags["TXXX:SERIES"]) == "Rowling Kinderbücher"
|
|
assert str(tags["TXXX:SERIES-PART"]) == "2"
|
|
assert str(tags["TCOM"]) == "Ben Becker"
|
|
|
|
|
|
def test_multi_disc_folder_is_one_audiobook(client):
|
|
root = client.tmp_path / "library" / "dr3i"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "DR3i", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Die dr3i - Böses Erwachen",
|
|
}).json()["id"]
|
|
|
|
folder = client.downloads / "DiE DR3i - Boeses Erwachen"
|
|
for disc, tracks in (("CD 1", 3), ("CD2", 2)):
|
|
d = folder / disc
|
|
d.mkdir(parents=True)
|
|
for i in range(1, tracks + 1):
|
|
(d / f"Track {i}.mp3").write_bytes(b"")
|
|
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert len(items) == 1
|
|
item = items[0]
|
|
assert item["name"] == "DiE DR3i - Boeses Erwachen"
|
|
assert item["is_dir"] and len(item["files"]) == 5
|
|
# discs in order, tracks naturally sorted within a disc
|
|
assert [Path(f).parent.name for f in item["files"]] == ["CD 1"] * 3 + ["CD2"] * 2
|
|
assert item["suggested_request_id"] == req_id
|
|
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": item["path"], "is_dir": True, "files": item["files"],
|
|
"request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert res["ok"], res
|
|
dest = sorted(p.name for p in Path(res["dest"]).iterdir())
|
|
assert dest == [f"Die dr3i - Böses Erwachen - Part 0{i}.mp3" for i in range(1, 6)]
|
|
assert not folder.exists()
|
|
|
|
|
|
def test_natural_track_order_in_flat_folder(client):
|
|
folder = client.downloads / "Folge 12"
|
|
folder.mkdir(parents=True)
|
|
for i in (1, 2, 10):
|
|
(folder / f"track{i}.mp3").write_bytes(b"")
|
|
item = client.get("/api/import/scan").json()["items"][0]
|
|
assert [Path(f).name for f in item["files"]] == ["track1.mp3", "track2.mp3", "track10.mp3"]
|
|
|
|
|
|
def test_real_world_series_tree(client):
|
|
"""Downloads laid out as Serie/Folgen/<Sammelordner>/<NNN - Titel>/CD/*.mp3,
|
|
with a Cover/ folder next to the CD folder."""
|
|
base = client.downloads / "Die Drei Fragezeichen" / "Folgen" / "3478632869 001-010"
|
|
episodes = ["001 - Der Super - Papagei", "002 - Der Phantomsee", "003 - Der Karpatenhund"]
|
|
for ep in episodes:
|
|
cd = base / ep / "CD"
|
|
cd.mkdir(parents=True)
|
|
for i in (1, 2, 10):
|
|
(cd / f"{i:02d} Track.mp3").write_bytes(b"")
|
|
cover = base / ep / "Cover"
|
|
cover.mkdir()
|
|
(cover / "front.jpg").write_bytes(b"")
|
|
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert [i["name"] for i in items] == episodes
|
|
for item in items:
|
|
assert item["is_dir"] and len(item["files"]) == 3
|
|
assert [Path(f).name for f in item["files"]] == ["01 Track.mp3", "02 Track.mp3", "10 Track.mp3"]
|
|
assert item["rel_dir"] == "Die Drei Fragezeichen/Folgen/3478632869 001-010"
|
|
|
|
|
|
def test_two_discs_next_to_cover(client):
|
|
folder = client.downloads / "099 - Die Villa der Toten"
|
|
for disc in ("CD 1", "CD 2"):
|
|
(folder / disc).mkdir(parents=True)
|
|
(folder / disc / "track.mp3").write_bytes(b"")
|
|
(folder / "Cover").mkdir()
|
|
(folder / "Cover" / "back.png").write_bytes(b"")
|
|
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert len(items) == 1
|
|
assert items[0]["name"] == "099 - Die Villa der Toten"
|
|
assert [Path(f).parent.name for f in items[0]["files"]] == ["CD 1", "CD 2"]
|
|
|
|
|
|
def test_subfolder_with_own_content_is_not_a_disc(client):
|
|
"""A real subfolder (an episode) must keep its own identity."""
|
|
folder = client.downloads / "Sammlung"
|
|
for ep in ("Folge 1", "Folge 2"):
|
|
(folder / ep).mkdir(parents=True)
|
|
for i in (1, 2):
|
|
(folder / ep / f"t{i}.mp3").write_bytes(b"")
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert sorted(i["name"] for i in items) == ["Folge 1", "Folge 2"]
|
|
|
|
|
|
def test_bulk_items_creates_series_and_skips_duplicates(client):
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "DDF", "media_type": "audiobook",
|
|
"root_path": str(client.tmp_path / "ddf"),
|
|
}).json()["id"]
|
|
items = [{"title": f"Die drei ??? Folge {n}", "series": "Die drei ???",
|
|
"volume": n, "external_id": f"ASIN{n}"} for n in range(1, 6)]
|
|
|
|
res = client.post("/api/requests/bulk-items",
|
|
json={"library_id": lib_id, "items": items}).json()
|
|
assert len(res["created"]) == 5 and res["skipped"] == 0
|
|
assert res["created"][0]["media_type"] == "audiobook"
|
|
assert res["created"][0]["volume"] == 1
|
|
|
|
# same call again: known external_ids are skipped, new ones still land
|
|
items.append({"title": "Die drei ??? Folge 6", "series": "Die drei ???",
|
|
"volume": 6, "external_id": "ASIN6"})
|
|
res2 = client.post("/api/requests/bulk-items",
|
|
json={"library_id": lib_id, "items": items}).json()
|
|
assert res2["skipped"] == 5 and len(res2["created"]) == 1
|
|
assert len(client.get("/api/requests", params={"status": "missing"}).json()) == 6
|
|
|
|
|
|
def test_bulk_items_series_then_scan_matches_by_episode_number(client):
|
|
"""The whole point: bulk-request a series, then let the scan assign folders."""
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "DDF", "media_type": "audiobook",
|
|
"root_path": str(client.tmp_path / "ddf"),
|
|
}).json()["id"]
|
|
titles = {1: "Die drei ??? und der Super-Papagei",
|
|
4: "Die drei ??? und die schwarze Katze",
|
|
8: "Die drei ??? und der grüne Geist"}
|
|
created = client.post("/api/requests/bulk-items", json={
|
|
"library_id": lib_id,
|
|
"items": [{"title": t, "series": "Die drei ???", "volume": n,
|
|
"external_id": f"ASIN{n}"} for n, t in titles.items()],
|
|
}).json()["created"]
|
|
by_volume = {r["volume"]: r["id"] for r in created}
|
|
|
|
base = client.downloads / "Die Drei Fragezeichen" / "Folgen" / "3478632869 001-010"
|
|
folders = {1: "001 - Der Super - Papagei", 4: "004 - Die schwarzn Katze",
|
|
8: "008 - Der gruene Geist"}
|
|
for n, folder in folders.items():
|
|
cd = base / folder / "CD"
|
|
cd.mkdir(parents=True)
|
|
(cd / "01.mp3").write_bytes(b"")
|
|
(cd / "02.mp3").write_bytes(b"")
|
|
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert len(items) == 3
|
|
for item in items:
|
|
n = int(item["name"][:3])
|
|
assert item["suggested_request_id"] == by_volume[n], item
|
|
assert item["score"] >= 90
|
|
|
|
|
|
def test_merged_multipart_import(client):
|
|
"""A multi-part episode ("Teil 1/2/3") imported as one audiobook: the client
|
|
merges the scanned entries and posts their files as a single item."""
|
|
root = client.tmp_path / "library" / "ddf"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "DDF", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Series}/{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Die drei ??? und die Toteninsel",
|
|
"series": "Die drei ???", "volume": 100,
|
|
}).json()["id"]
|
|
|
|
base = client.downloads / "Folgen" / "100 - Toteninsel"
|
|
for part in ("Teil 1", "Teil 2", "Teil 3"):
|
|
d = base / part
|
|
d.mkdir(parents=True)
|
|
for i in (1, 2):
|
|
(d / f"{i:02d}.mp3").write_bytes(b"")
|
|
(base / "Cover").mkdir()
|
|
(base / "Cover" / "front.jpg").write_bytes(b"")
|
|
|
|
# "Teil N" folders are disc folders, so the scan already yields one entry
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert len(items) == 1 and len(items[0]["files"]) == 6
|
|
|
|
# a split scan yields the parts separately - that is what the merge button
|
|
# in the UI stitches back together
|
|
parts = client.get("/api/import/scan", params={"split_dirs": True}).json()["items"]
|
|
assert len(parts) == 6
|
|
merged_files = [f for p in parts for f in p["files"]]
|
|
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(base), "is_dir": True, "files": merged_files, "request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert res["ok"], res
|
|
|
|
dest = sorted(p.name for p in Path(res["dest"]).iterdir())
|
|
assert dest == [f"Die drei ??? und die Toteninsel - Part 0{i}.mp3" for i in range(1, 7)]
|
|
# emptied part folders are gone, the cover folder survives
|
|
assert not (base / "Teil 1").exists()
|
|
assert (base / "Cover" / "front.jpg").exists()
|
|
|
|
|
|
def test_import_rejects_files_outside_download_dir(client):
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "L", "media_type": "audiobook", "root_path": str(client.tmp_path / "l"),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "X"}).json()["id"]
|
|
outside = client.tmp_path / "elsewhere"
|
|
outside.mkdir()
|
|
(outside / "a.mp3").write_bytes(b"")
|
|
folder = client.downloads / "ok"
|
|
folder.mkdir()
|
|
(folder / "b.mp3").write_bytes(b"")
|
|
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(folder), "is_dir": True,
|
|
"files": [str(folder / "b.mp3"), str(outside / "a.mp3")], "request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert not res["ok"] and "outside" in res["error"]
|
|
assert (outside / "a.mp3").exists()
|
|
|
|
|
|
def test_merge_at_download_root_keeps_the_download_dir(client):
|
|
"""Merging entries that sit directly in the download dir makes the item path
|
|
the download dir itself - which must never be removed."""
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "L", "media_type": "audiobook", "root_path": str(client.tmp_path / "l"),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "Toteninsel"}).json()["id"]
|
|
files = []
|
|
for part in ("Teil 1", "Teil 2"):
|
|
d = client.downloads / part
|
|
d.mkdir()
|
|
(d / "a.mp3").write_bytes(b"")
|
|
(d / "b.mp3").write_bytes(b"")
|
|
files += [str(d / "a.mp3"), str(d / "b.mp3")]
|
|
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(client.downloads), "is_dir": True, "files": files,
|
|
"request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert res["ok"], res
|
|
assert client.downloads.is_dir()
|
|
assert not (client.downloads / "Teil 1").exists()
|
|
|
|
|
|
def test_merged_parts_clean_up_nested_source_folders(client):
|
|
""""100 - Toteninsel/A - Sphinx/CD/*.mp3" - every emptied level below the
|
|
merged item goes away, the cover folder keeps its parent alive."""
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "L", "media_type": "audiobook", "root_path": str(client.tmp_path / "l"),
|
|
"folder_template": "{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "Toteninsel"}).json()["id"]
|
|
|
|
base = client.downloads / "100 - Toteninsel"
|
|
files = []
|
|
for part in ("A - Sphinx", "B - Volk", "C - Graeber"):
|
|
cd = base / part / "CD"
|
|
cd.mkdir(parents=True)
|
|
for i in (1, 2):
|
|
f = cd / f"{i:02d}.mp3"
|
|
f.write_bytes(b"")
|
|
files.append(str(f))
|
|
(base / "Cover").mkdir()
|
|
(base / "Cover" / "front.jpg").write_bytes(b"")
|
|
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(base), "is_dir": True, "files": files, "request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert res["ok"], res
|
|
assert sorted(p.name for p in Path(res["dest"]).iterdir()) == [
|
|
f"Toteninsel - Part 0{i}.mp3" for i in range(1, 7)
|
|
]
|
|
assert not (base / "A - Sphinx").exists()
|
|
assert not (base / "B - Volk").exists()
|
|
assert (base / "Cover" / "front.jpg").exists()
|
|
|
|
|
|
def test_letter_indexed_parts_are_one_audiobook(client):
|
|
""""Teil A/B/C" and "A - Titel/B - Titel" both describe one story."""
|
|
schattenwelt = client.downloads / "175 Schattenwelt-3CD-DE-2015-VOiCE"
|
|
for part in ("Teil A", "Teil B", "Teil C"):
|
|
d = schattenwelt / part
|
|
d.mkdir(parents=True)
|
|
(d / "01.mp3").write_bytes(b"")
|
|
|
|
toteninsel = client.downloads / "100 - Toteninsel"
|
|
for part in ("A - Das.Raetsel.der.Sphinx", "B - Das.vergessene.Volk",
|
|
"C - Der Fluch der Graeber"):
|
|
d = toteninsel / part
|
|
d.mkdir(parents=True)
|
|
(d / "01.mp3").write_bytes(b"")
|
|
(toteninsel / "Cover").mkdir()
|
|
(toteninsel / "Cover" / "f.jpg").write_bytes(b"")
|
|
|
|
items = {i["name"]: i for i in client.get("/api/import/scan").json()["items"]}
|
|
assert sorted(items) == ["100 - Toteninsel", "175 Schattenwelt-3CD-DE-2015-VOiCE"]
|
|
for item in items.values():
|
|
assert item["is_dir"] and len(item["files"]) == 3
|
|
# parts stay in A, B, C order
|
|
assert [Path(f).parent.name[0] for f in item["files"]] in (
|
|
["T", "T", "T"], ["A", "B", "C"],
|
|
)
|
|
order = [Path(f).parent.name for f in items["175 Schattenwelt-3CD-DE-2015-VOiCE"]["files"]]
|
|
assert order == ["Teil A", "Teil B", "Teil C"]
|
|
|
|
|
|
def test_letter_folders_that_are_not_parts_stay_separate(client):
|
|
"""Two episodes that happen to start with a letter must not be merged."""
|
|
base = client.downloads / "Sammlung"
|
|
for name in ("A - Erste Folge", "Zweite Folge"):
|
|
d = base / name
|
|
d.mkdir(parents=True)
|
|
(d / "01.mp3").write_bytes(b"")
|
|
(d / "02.mp3").write_bytes(b"")
|
|
names = sorted(i["name"] for i in client.get("/api/import/scan").json()["items"])
|
|
assert names == ["A - Erste Folge", "Zweite Folge"]
|
|
|
|
|
|
def test_append_to_an_already_imported_audiobook(client):
|
|
"""The Toteninsel case: part A got imported, B and C failed. They can be
|
|
added afterwards without moving anything back by hand."""
|
|
root = client.tmp_path / "library" / "kids"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "K", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "Toteninsel"}).json()["id"]
|
|
|
|
part_a = client.downloads / "A - Sphinx"
|
|
part_a.mkdir()
|
|
for i in (1, 2, 3):
|
|
(part_a / f"{i:02d}.mp3").write_bytes(b"")
|
|
client.post("/api/import", json={"items": [{
|
|
"path": str(part_a), "is_dir": True,
|
|
"files": [str(f) for f in sorted(part_a.iterdir())], "request_id": req_id,
|
|
}]})
|
|
imported = client.get("/api/requests", params={"status": "imported"}).json()[0]
|
|
assert len(list(Path(imported["imported_path"]).iterdir())) == 3
|
|
|
|
# B and C arrive later
|
|
later = []
|
|
for part in ("B - Volk", "C - Graeber"):
|
|
d = client.downloads / part
|
|
d.mkdir()
|
|
for i in (1, 2):
|
|
f = d / f"{i:02d}.mp3"
|
|
f.write_bytes(b"")
|
|
later.append(str(f))
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(client.downloads / "B - Volk"), "is_dir": True,
|
|
"files": later, "request_id": req_id, "append": True,
|
|
}]}).json()["results"][0]
|
|
assert res["ok"] and res["appended"]
|
|
|
|
files = sorted(p.name for p in Path(res["dest"]).iterdir())
|
|
assert files == [f"Toteninsel - Part 0{i}.mp3" for i in range(1, 8)]
|
|
|
|
|
|
def test_append_without_the_flag_is_still_refused(client):
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "K", "media_type": "audiobook", "root_path": str(client.tmp_path / "k"),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "X"}).json()["id"]
|
|
folder = client.downloads / "first"
|
|
folder.mkdir()
|
|
(folder / "a.mp3").write_bytes(b"")
|
|
(folder / "b.mp3").write_bytes(b"")
|
|
client.post("/api/import", json={"items": [{
|
|
"path": str(folder), "is_dir": True,
|
|
"files": [str(folder / "a.mp3"), str(folder / "b.mp3")], "request_id": req_id,
|
|
}]})
|
|
|
|
second = client.downloads / "second"
|
|
second.mkdir()
|
|
(second / "c.mp3").write_bytes(b"")
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(second), "is_dir": False,
|
|
"files": [str(second / "c.mp3")], "request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert not res["ok"]
|
|
assert "bereits importiert" in res["error"]
|
|
assert (second / "c.mp3").exists()
|
|
|
|
|
|
def test_library_language_round_trip(client):
|
|
lib = client.post("/api/libraries", json={
|
|
"name": "english", "media_type": "audiobook",
|
|
"root_path": str(client.tmp_path / "en"), "language": "english",
|
|
}).json()
|
|
assert lib["language"] == "english"
|
|
assert client.get("/api/libraries").json()[0]["language"] == "english"
|
|
|
|
# libraries without a language keep working and stay unrestricted
|
|
other = client.post("/api/libraries", json={
|
|
"name": "adults", "media_type": "audiobook",
|
|
"root_path": str(client.tmp_path / "de"),
|
|
}).json()
|
|
assert other["language"] == ""
|
|
|
|
updated = client.put(f"/api/libraries/{other['id']}", json={
|
|
"name": "adults", "media_type": "audiobook",
|
|
"root_path": str(client.tmp_path / "de"), "language": "german",
|
|
}).json()
|
|
assert updated["language"] == "german"
|
|
|
|
|
|
def test_search_passes_language_to_the_provider(client, monkeypatch):
|
|
from wordarr.metadata.base import MetadataResult
|
|
seen = {}
|
|
|
|
async def fake(query, language="", fallback=True):
|
|
seen["query"], seen["language"] = query, language
|
|
return [MetadataResult(media_type="audiobook", title="A Game of Thrones",
|
|
language="english")]
|
|
|
|
monkeypatch.setitem(
|
|
__import__("wordarr.metadata", fromlist=["PROVIDERS"]).PROVIDERS, "audiobook", fake)
|
|
res = client.get("/api/search", params={
|
|
"media_type": "audiobook", "q": "Game of Thrones", "language": "english",
|
|
}).json()
|
|
assert seen == {"query": "Game of Thrones", "language": "english"}
|
|
assert res[0]["language"] == "english"
|
|
|
|
|
|
def test_folder_of_separate_books_is_flagged(client):
|
|
"""One book per file (a series bought as single titles) still scans as one
|
|
entry - but gets flagged so the UI can offer to split it."""
|
|
got = client.downloads / "George R. R. Martin" / "A Game of Thrones"
|
|
got.mkdir(parents=True)
|
|
for name in ("1 - A Game of Thrones- A Song of Ice and Fire, Book 1 (Unabridged)",
|
|
"2 - A Clash of Kings- A Song of Ice and Fire, Book 2 (Unabridged)",
|
|
"A - The World of Ice & Fire - The Untold History of Westeros"):
|
|
with open(got / f"{name}.mp3", "wb") as f:
|
|
f.truncate(700 * 1024 * 1024) # sparse, costs no disk space
|
|
|
|
chapters = client.downloads / "Ein Hoerbuch"
|
|
chapters.mkdir()
|
|
for i, title in enumerate(["Prolog", "Die Ankunft", "Das Ende"], 1):
|
|
with open(chapters / f"{i:02d} - {title}.mp3", "wb") as f:
|
|
f.truncate(25 * 1024 * 1024)
|
|
|
|
tracks = client.downloads / "Folge 100"
|
|
tracks.mkdir()
|
|
for i in (1, 2, 3):
|
|
with open(tracks / f"{i:02d} Track.mp3", "wb") as f:
|
|
f.truncate(200 * 1024 * 1024)
|
|
|
|
flags = {i["name"]: i["maybe_separate"]
|
|
for i in client.get("/api/import/scan").json()["items"]}
|
|
assert flags == {
|
|
"A Game of Thrones": True, # different titles, each big enough
|
|
"Ein Hoerbuch": False, # different titles, but chapter sized
|
|
"Folge 100": False, # same name beyond the index
|
|
}
|
|
|
|
|
|
def test_split_dirs_still_lists_every_file(client):
|
|
got = client.downloads / "A Game of Thrones"
|
|
got.mkdir()
|
|
for n in (1, 2, 3):
|
|
(got / f"{n} - Book {n}.mp3").write_bytes(b"")
|
|
items = client.get("/api/import/scan", params={"split_dirs": True}).json()["items"]
|
|
assert len(items) == 3
|
|
assert all(not i["is_dir"] for i in items)
|
|
|
|
|
|
def test_relocate_after_fixing_the_series(client):
|
|
"""Audible files Harry Potter under "Wizarding World"; after correcting the
|
|
series the folder on disk should follow."""
|
|
root = client.tmp_path / "library" / "english"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "english", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Series}/{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Philosopher's Stone",
|
|
"series": "Wizarding World", "volume": 1,
|
|
}).json()["id"]
|
|
|
|
src = client.downloads / "hp1"
|
|
src.mkdir()
|
|
for i in (1, 2):
|
|
(src / f"{i:02d}.mp3").write_bytes(b"")
|
|
dest = client.post("/api/import", json={"items": [{
|
|
"path": str(src), "is_dir": True,
|
|
"files": [str(f) for f in sorted(src.iterdir())], "request_id": req_id,
|
|
}]}).json()["results"][0]["dest"]
|
|
assert Path(dest) == root / "Wizarding World" / "Philosopher's Stone"
|
|
# a cover dropped next to the audio must survive the move
|
|
(Path(dest) / "cover.jpg").write_bytes(b"")
|
|
|
|
client.put(f"/api/requests/{req_id}", json={
|
|
"library_id": lib_id, "title": "Philosopher's Stone",
|
|
"series": "Harry Potter", "volume": 1,
|
|
})
|
|
res = client.post(f"/api/requests/{req_id}/relocate").json()
|
|
assert res["ok"] and res["moved"]
|
|
new = root / "Harry Potter" / "Philosopher's Stone"
|
|
assert Path(res["dest"]) == new
|
|
assert sorted(p.name for p in new.iterdir()) == [
|
|
"Philosopher's Stone - Part 01.mp3", "Philosopher's Stone - Part 02.mp3", "cover.jpg",
|
|
]
|
|
assert not (root / "Wizarding World").exists() # emptied parent is gone
|
|
assert client.get("/api/requests", params={"status": "imported"}).json()[0][
|
|
"imported_path"] == str(new)
|
|
|
|
|
|
def test_relocate_is_a_no_op_when_nothing_changed(client):
|
|
root = client.tmp_path / "library" / "l"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "L", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "Ein Buch"}).json()["id"]
|
|
src = client.downloads / "buch"
|
|
src.mkdir()
|
|
for i in (1, 2):
|
|
(src / f"{i:02d}.mp3").write_bytes(b"")
|
|
client.post("/api/import", json={"items": [{
|
|
"path": str(src), "is_dir": True,
|
|
"files": [str(f) for f in sorted(src.iterdir())], "request_id": req_id,
|
|
}]})
|
|
res = client.post(f"/api/requests/{req_id}/relocate").json()
|
|
assert res["ok"] and not res["moved"]
|
|
assert sorted(p.name for p in (root / "Ein Buch").iterdir()) == [
|
|
"Ein Buch - Part 01.mp3", "Ein Buch - Part 02.mp3",
|
|
]
|
|
|
|
|
|
def test_relocate_refuses_to_overwrite(client):
|
|
root = client.tmp_path / "library" / "l"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "L", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "Erstes"}).json()["id"]
|
|
src = client.downloads / "b"
|
|
src.mkdir()
|
|
(src / "01.mp3").write_bytes(b"")
|
|
(src / "02.mp3").write_bytes(b"")
|
|
client.post("/api/import", json={"items": [{
|
|
"path": str(src), "is_dir": True,
|
|
"files": [str(f) for f in sorted(src.iterdir())], "request_id": req_id,
|
|
}]})
|
|
occupied = root / "Zweites"
|
|
occupied.mkdir(parents=True)
|
|
(occupied / "fremd.mp3").write_bytes(b"")
|
|
|
|
client.put(f"/api/requests/{req_id}", json={"library_id": lib_id, "title": "Zweites"})
|
|
res = client.post(f"/api/requests/{req_id}/relocate")
|
|
assert res.status_code == 409
|
|
assert (occupied / "fremd.mp3").exists()
|
|
assert (root / "Erstes" / "Erstes - Part 01.mp3").exists()
|
|
|
|
|
|
def test_files_moved_back_are_matched_against_imported_requests(client):
|
|
"""Moving a library folder back into the download dir: the request still says
|
|
"imported", but its files are gone - the scan should suggest it anyway."""
|
|
root = client.tmp_path / "library" / "kids"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "kids", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Series}/{Volume} - {Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Die Zeitreisende",
|
|
"series": "Die drei ???", "volume": 194,
|
|
}).json()["id"]
|
|
|
|
src = client.downloads / "194 - Die Zeitreisende"
|
|
src.mkdir()
|
|
for i in (1, 2):
|
|
(src / f"{i:02d}.mp3").write_bytes(b"")
|
|
dest = Path(client.post("/api/import", json={"items": [{
|
|
"path": str(src), "is_dir": True,
|
|
"files": [str(f) for f in sorted(src.iterdir())], "request_id": req_id,
|
|
}]}).json()["results"][0]["dest"])
|
|
|
|
# while the files are in place the request is not offered again
|
|
scan = client.get("/api/import/scan").json()
|
|
assert scan["orphaned_request_ids"] == []
|
|
|
|
# now move the folder back into the download dir, as the user did
|
|
shutil.move(str(dest), client.downloads / dest.name)
|
|
scan = client.get("/api/import/scan").json()
|
|
assert scan["orphaned_request_ids"] == [req_id]
|
|
item = next(i for i in scan["items"] if i["name"] == dest.name)
|
|
assert item["suggested_request_id"] == req_id
|
|
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": item["path"], "is_dir": True, "files": item["files"],
|
|
"request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert res["ok"], res
|
|
assert Path(res["dest"]) == root / "Die drei ???" / "194 - Die Zeitreisende"
|
|
assert sorted(p.name for p in Path(res["dest"]).iterdir()) == [
|
|
"Die Zeitreisende - Part 01.mp3", "Die Zeitreisende - Part 02.mp3",
|
|
]
|
|
|
|
|
|
def test_imported_request_with_files_in_place_is_still_refused(client):
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "L", "media_type": "audiobook", "root_path": str(client.tmp_path / "l"),
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests",
|
|
json={"library_id": lib_id, "title": "X"}).json()["id"]
|
|
src = client.downloads / "a"
|
|
src.mkdir()
|
|
(src / "1.mp3").write_bytes(b"")
|
|
(src / "2.mp3").write_bytes(b"")
|
|
client.post("/api/import", json={"items": [{
|
|
"path": str(src), "is_dir": True,
|
|
"files": [str(src / "1.mp3"), str(src / "2.mp3")], "request_id": req_id,
|
|
}]})
|
|
again = client.downloads / "b"
|
|
again.mkdir()
|
|
(again / "1.mp3").write_bytes(b"")
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(again), "is_dir": False, "files": [str(again / "1.mp3")],
|
|
"request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert not res["ok"] and "bereits importiert" in res["error"]
|
|
|
|
|
|
def test_two_titles_rendering_to_the_same_name_do_not_overwrite(client):
|
|
"""Two books whose templates produce the same folder and file names must not
|
|
silently replace each other - that loses files and confuses ABS."""
|
|
root = client.tmp_path / "library" / "dcc"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "DCC", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Series}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
ids = [client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Dungeon Crawler Carl",
|
|
"series": "Dungeon Crawler Carl", "volume": v,
|
|
}, params={"allow_duplicate": True}).json()["id"] for v in (1, 7)]
|
|
|
|
dests = []
|
|
for n, req_id in enumerate(ids, 1):
|
|
src = client.downloads / f"teil{n}"
|
|
src.mkdir()
|
|
for i in (1, 2):
|
|
(src / f"{i:02d}.mp3").write_bytes(f"Teil {n} Datei {i}".encode())
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": str(src), "is_dir": True,
|
|
"files": [str(f) for f in sorted(src.iterdir())], "request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
dests.append(res)
|
|
|
|
assert dests[0]["ok"]
|
|
assert not dests[1]["ok"]
|
|
assert "already exists" in dests[1]["error"]
|
|
# the first title is untouched and the second one still has its files
|
|
folder = root / "Dungeon Crawler Carl"
|
|
assert sorted(p.name for p in folder.iterdir()) == [
|
|
"Dungeon Crawler Carl - Part 01.mp3", "Dungeon Crawler Carl - Part 02.mp3",
|
|
]
|
|
content = (folder / "Dungeon Crawler Carl - Part 01.mp3").read_bytes()
|
|
assert b"Teil 1 Datei 1" in content and b"Teil 2" not in content
|
|
assert sorted(p.name for p in (client.downloads / "teil2").iterdir()) == \
|
|
["01.mp3", "02.mp3"]
|
|
|
|
|
|
def test_scan_and_import_survive_a_name_that_is_not_utf8(client):
|
|
"""A single Latin-1 byte in a folder name used to fail the whole scan with a
|
|
500 (UnicodeEncodeError: surrogates not allowed)."""
|
|
broken = os.fsdecode(b"Die drei ??? - Die Fu\xdfball-Falle")
|
|
folder = client.downloads / broken
|
|
folder.mkdir()
|
|
for i in (1, 2):
|
|
(folder / f"{i:02d}.mp3").write_bytes(b"")
|
|
|
|
root = client.tmp_path / "library" / "kids"
|
|
lib_id = client.post("/api/libraries", json={
|
|
"name": "kids", "media_type": "audiobook", "root_path": str(root),
|
|
"folder_template": "{Title}", "file_template": "{Title}",
|
|
}).json()["id"]
|
|
req_id = client.post("/api/requests", json={
|
|
"library_id": lib_id, "title": "Die drei ??? und die Fußball-Falle",
|
|
}).json()["id"]
|
|
|
|
resp = client.get("/api/import/scan")
|
|
assert resp.status_code == 200
|
|
item = resp.json()["items"][0]
|
|
assert "�" in item["name"] # shown with a replacement character
|
|
assert item["path"].startswith("b64:") # but transported losslessly
|
|
|
|
res = client.post("/api/import", json={"items": [{
|
|
"path": item["path"], "is_dir": True, "files": item["files"],
|
|
"request_id": req_id,
|
|
}]}).json()["results"][0]
|
|
assert res["ok"], res
|
|
assert sorted(p.name for p in Path(res["dest"]).iterdir()) == [
|
|
"Die drei ??? und die Fußball-Falle - Part 01.mp3",
|
|
"Die drei ??? und die Fußball-Falle - Part 02.mp3",
|
|
]
|
|
assert not folder.exists()
|
|
|
|
|
|
def test_loose_files_in_the_download_root_do_not_swallow_everything(client):
|
|
"""Two loose audio files directly in the download dir used to make the whole
|
|
root look like one audiobook - and its early return hid every subfolder."""
|
|
(client.downloads / "Teil 01.mp3").write_bytes(b"")
|
|
(client.downloads / "Teil 02.mp3").write_bytes(b"")
|
|
for folge in ("Folge 1", "Folge 2"):
|
|
d = client.downloads / "Serie" / folge
|
|
d.mkdir(parents=True)
|
|
for i in (1, 2):
|
|
(d / f"{i:02d}.mp3").write_bytes(b"")
|
|
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert sorted(i["name"] for i in items) == ["Folge 1", "Folge 2", "Teil 01", "Teil 02"]
|
|
# the folders stay grouped, the loose files are listed on their own
|
|
by_name = {i["name"]: i for i in items}
|
|
assert by_name["Folge 1"]["is_dir"] and len(by_name["Folge 1"]["files"]) == 2
|
|
assert not by_name["Teil 01"]["is_dir"] and len(by_name["Teil 01"]["files"]) == 1
|
|
|
|
|
|
def test_a_single_folder_with_many_files_is_still_one_audiobook(client):
|
|
"""The fix must not break the normal case one level down."""
|
|
d = client.downloads / "Ein Hoerbuch"
|
|
d.mkdir()
|
|
for i in (1, 2, 3):
|
|
(d / f"{i:02d}.mp3").write_bytes(b"")
|
|
items = client.get("/api/import/scan").json()["items"]
|
|
assert len(items) == 1 and items[0]["is_dir"] and len(items[0]["files"]) == 3
|