fix(import): close the append path check and the relocate overwrite

This commit is contained in:
Steppenstreuner
2026-08-29 16:38:25 +02:00
parent 551352e838
commit e6b6eba138
6 changed files with 181 additions and 19 deletions
+110
View File
@@ -983,3 +983,113 @@ def test_a_single_folder_with_many_files_is_still_one_audiobook(client):
(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
def test_append_rejects_files_outside_download_dir(client):
"""The append branch used to skip the download-dir check entirely, so it
would happily pull in - and clean up - folders anywhere on disk."""
root = client.tmp_path / "library" / "k"
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"
part_a.mkdir()
for i in (1, 2):
(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,
}]})
outside = client.tmp_path / "elsewhere"
outside.mkdir()
stranger = outside / "private.mp3"
stranger.write_bytes(b"not yours")
res = client.post("/api/import", json={"items": [{
"path": str(outside), "is_dir": True,
"files": [str(stranger)], "request_id": req_id, "append": True,
}]}).json()["results"][0]
assert not res["ok"]
assert res["error"] == "path outside download dir"
assert stranger.exists() and outside.is_dir()
def test_import_survives_a_broken_base64_path(client):
"""A malformed b64: payload is one bad item, not a 500 for the whole batch."""
lib_id = client.post("/api/libraries", json={
"name": "K", "media_type": "ebook", "root_path": str(client.tmp_path / "k"),
}).json()["id"]
req_id = client.post("/api/requests",
json={"library_id": lib_id, "title": "X"}).json()["id"]
resp = client.post("/api/import", json={"items": [{
"path": "b64:!!!not base64!!!", "is_dir": False,
"files": [], "request_id": req_id,
}]})
assert resp.status_code == 200
assert not resp.json()["results"][0]["ok"]
def test_relocate_does_not_overwrite_within_the_same_folder(client):
"""Renaming in place: a target name may already be held by another file of
the same set. Sorting decides who moves first, so the slot can still be
occupied when the move happens - it must not be overwritten."""
root = client.tmp_path / "library" / "k"
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": "Insel"}).json()["id"]
src = client.downloads / "insel"
src.mkdir()
for i in (1, 2):
(src / f"{i:02d}.mp3").write_bytes(f"track {i}".encode())
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,
}]})
dest = root / "Insel"
# "Aaa" sorts first, so it claims "Part 01" while part 1 still sits there
(dest / "Insel - Part 02.mp3").rename(dest / "Aaa.mp3")
res = client.post(f"/api/requests/{req_id}/relocate").json()
assert res["ok"], res
assert sorted(f.name for f in dest.iterdir()) == [
"Insel - Part 01.mp3", "Insel - Part 02.mp3",
]
blobs = [f.read_bytes() for f in dest.iterdir()]
assert all(any(f"track {i}".encode() in b for b in blobs) for i in (1, 2))
def test_single_file_folder_import_removes_the_empty_source(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": "Solo"}).json()["id"]
folder = client.downloads / "solo release"
folder.mkdir()
(folder / "book.m4b").write_bytes(b"")
res = client.post("/api/import", json={"items": [{
"path": str(folder), "is_dir": True,
"files": [str(folder / "book.m4b")], "request_id": req_id,
}]}).json()["results"][0]
assert res["ok"], res
assert not folder.exists()
def test_scan_does_not_loop_on_a_symlink_cycle(client):
from wordarr.importer import scanner
d = client.downloads / "series"
d.mkdir()
(d / "book.m4b").write_bytes(b"")
(d / "back").symlink_to(client.downloads, target_is_directory=True)
items = scanner.scan(client.downloads)
assert [i["name"] for i in items] == ["book"]