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" # "?" is stripped from paths 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"