FastAPI + SQLite backend with vanilla-JS web UI. Requests via Open Library / Audible / AniList metadata search (or manual entry) with per-request target library selection. Manual import flow: scan the download dir, fuzzy-match files against missing requests, review, then move+rename into the library per configurable naming templates. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
109 lines
3.8 KiB
Python
109 lines
3.8 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
|