267 lines
10 KiB
Python
267 lines
10 KiB
Python
import importlib
|
|
|
|
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" # "?" becomes fullwidth
|
|
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]
|
|
assert item["name"] == "CD1"
|
|
assert item["rel_dir"] == "Sammlung/Sonderfolgen"
|
|
|
|
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"
|