daniels
f9eb633e19
Only treat configured manual storage as Importarr-owned and map SAB container paths to local paths.
88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
from pathlib import Path
|
|
|
|
from importarr.readiness import classify_history_item
|
|
|
|
|
|
ROOT = Path("/tmp/downloads/manual")
|
|
|
|
|
|
def item(**kwargs):
|
|
data = {"nzo_id": "1", "category": "manual", "status": "Completed", "storage": str(ROOT / "Movie")}
|
|
data.update(kwargs)
|
|
return data
|
|
|
|
|
|
def test_completed_manual_is_ready():
|
|
result = classify_history_item(item(), set(), "manual", ROOT)
|
|
assert result.ready
|
|
|
|
|
|
def test_wrong_category_ignored():
|
|
result = classify_history_item(item(category="*"), set(), "manual", ROOT)
|
|
assert result.ready
|
|
assert result.reason == "SAB completed inside Importarr download root"
|
|
|
|
|
|
def test_wrong_category_outside_root_ignored():
|
|
result = classify_history_item(item(category="*", storage="/tmp/other/Movie"), set(), "manual", ROOT)
|
|
assert result.state == "ignored"
|
|
|
|
|
|
def test_sab_storage_root_maps_to_local_download_root():
|
|
result = classify_history_item(
|
|
item(category="*", storage="/data/downloads/manual/Movie"),
|
|
set(),
|
|
"manual",
|
|
ROOT,
|
|
sab_storage_root=Path("/data/downloads/manual"),
|
|
)
|
|
assert result.ready
|
|
assert result.storage == ROOT / "Movie"
|
|
|
|
|
|
def test_radarr_sonarr_storage_roots_are_not_importarr_owned():
|
|
for storage in ("/data/downloads/movies/Movie", "/data/downloads/tv/Show"):
|
|
result = classify_history_item(
|
|
item(category="*", storage=storage),
|
|
set(),
|
|
"manual",
|
|
ROOT,
|
|
sab_storage_root=Path("/data/downloads/manual"),
|
|
)
|
|
assert result.state == "ignored"
|
|
|
|
|
|
def test_sab_cat_field_is_treated_as_category():
|
|
result = classify_history_item(item(category=None, cat="manual"), set(), "manual", ROOT)
|
|
assert result.ready
|
|
|
|
|
|
def test_queue_item_not_ready():
|
|
result = classify_history_item(item(), {"1"}, "manual", ROOT)
|
|
assert result.state == "processing"
|
|
|
|
|
|
def test_force_status_overrides_queue_and_processing_status():
|
|
result = classify_history_item(item(status="Extracting"), {"1"}, "manual", ROOT, force_status=True)
|
|
assert result.ready
|
|
assert result.storage == ROOT / "Movie"
|
|
|
|
|
|
def test_post_processing_not_ready():
|
|
for status in ["Queued", "Repairing", "Extracting", "Moving"]:
|
|
assert classify_history_item(item(status=status), set(), "manual", ROOT).state == "processing"
|
|
|
|
|
|
def test_empty_storage_unknown():
|
|
assert classify_history_item(item(storage=""), set(), "manual", ROOT).state == "unknown"
|
|
|
|
|
|
def test_unpack_path_not_ready():
|
|
result = classify_history_item(item(storage=str(ROOT / "_UNPACK_Movie")), set(), "manual", ROOT)
|
|
assert result.state == "processing"
|
|
|
|
|
|
def test_force_status_keeps_transient_path_safety():
|
|
result = classify_history_item(item(storage=str(ROOT / "_UNPACK_Movie"), status="Extracting"), {"1"}, "manual", ROOT, force_status=True)
|
|
assert result.state == "processing"
|