417 lines
16 KiB
Python
417 lines
16 KiB
Python
import importlib
|
|
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
|