174 lines
6.5 KiB
Python
174 lines
6.5 KiB
Python
from importarr.config import Settings
|
|
from importarr.state import State
|
|
|
|
|
|
def configure_main(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("IMPORTARR_STATE_PATH", str(tmp_path / "initial.db"))
|
|
import importarr.main as main
|
|
|
|
download = tmp_path / "downloads"
|
|
movies = tmp_path / "movies"
|
|
tv = tmp_path / "tv"
|
|
monkeypatch.setattr(main, "settings", Settings(download_root=download, movies_root=movies, tv_root=tv, state_path=tmp_path / "state.db"))
|
|
monkeypatch.setattr(main, "state", State(tmp_path / "state.db"))
|
|
return main, download, movies, tv
|
|
|
|
|
|
def test_pause_prevents_manual_queue_sync(tmp_path, monkeypatch):
|
|
main, download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
batch = download / "Release"
|
|
batch.mkdir(parents=True)
|
|
(batch / "Movie.mkv").write_bytes(b"movie")
|
|
main.state.add_manual_batch(batch)
|
|
|
|
main.state.set_app_state("queue_mode", "paused")
|
|
main.sync_manual_queue()
|
|
|
|
assert main.state.list_queue_items() == []
|
|
|
|
|
|
def test_start_reenables_manual_queue_sync(tmp_path, monkeypatch):
|
|
main, download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
batch = download / "Release"
|
|
batch.mkdir(parents=True)
|
|
(batch / "Movie.mkv").write_bytes(b"movie")
|
|
main.state.add_manual_batch(batch)
|
|
|
|
main.state.set_app_state("queue_mode", "paused")
|
|
main.sync_manual_queue()
|
|
main.state.set_app_state("queue_mode", "running")
|
|
main.sync_manual_queue()
|
|
|
|
assert len(main.state.list_queue_items()) == 1
|
|
|
|
|
|
def test_queue_jobs_include_groups_and_manual_context(tmp_path, monkeypatch):
|
|
main, download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
batch = download / "Release" / "Season 1"
|
|
batch.mkdir(parents=True)
|
|
(batch / "Episode.mkv").write_bytes(b"episode")
|
|
main.state.add_manual_batch(batch.parent)
|
|
|
|
main.sync_manual_queue()
|
|
jobs = main.queue_jobs()
|
|
|
|
assert jobs[0]["group"] == "manual_batch"
|
|
assert jobs[0]["relative_path"] == "Release/Season 1/Episode.mkv"
|
|
assert jobs[0]["can_run_now"] is True
|
|
|
|
|
|
def test_queue_item_retry_ignore_and_remove_actions(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
row = main.state.upsert_queue_item(source_type="sab", source_id="job-1", name="Release", state="failed", reason="ImportError")
|
|
|
|
retried = main.queue_item_action(row["id"], main.QueueItemActionRequest(action="retry"))
|
|
assert retried["item"]["state"] == "ready"
|
|
assert retried["item"]["reason"] == "retry requested"
|
|
|
|
ignored = main.queue_item_action(row["id"], main.QueueItemActionRequest(action="ignore"))
|
|
assert ignored["item"]["state"] == "skipped"
|
|
|
|
removed = main.queue_item_action(row["id"], main.QueueItemActionRequest(action="remove"))
|
|
assert removed == {"status": "removed", "id": row["id"]}
|
|
assert main.state.get_queue_item(row["id"]) is None
|
|
|
|
|
|
def test_queue_item_run_now_imports_only_selected_item(tmp_path, monkeypatch):
|
|
main, download, movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
batch = download / "Release"
|
|
batch.mkdir(parents=True)
|
|
selected = batch / "Selected.mkv"
|
|
waiting = batch / "Waiting.mkv"
|
|
selected.write_bytes(b"selected")
|
|
waiting.write_bytes(b"waiting")
|
|
selected_row = main.state.upsert_queue_item(source_type="manual", source_id=str(selected), source_path=selected, name=selected.name, state="manual_batch")
|
|
main.state.upsert_queue_item(source_type="manual", source_id=str(waiting), source_path=waiting, name=waiting.name, state="manual_batch")
|
|
|
|
result = main.queue_item_action(selected_row["id"], main.QueueItemActionRequest(action="run-now"))
|
|
|
|
assert result["status"] == "imported"
|
|
assert result["imported"] == 1
|
|
assert (movies / "Selected.mkv").read_bytes() == b"selected"
|
|
assert waiting.exists()
|
|
rows = {row["name"]: row for row in main.state.list_queue_items(active_only=False)}
|
|
assert rows["Selected.mkv"]["state"] == "imported"
|
|
assert rows["Waiting.mkv"]["state"] == "manual_batch"
|
|
|
|
|
|
def test_cancel_current_stops_before_next_manual_item(tmp_path, monkeypatch):
|
|
main, download, movies, tv = configure_main(tmp_path, monkeypatch)
|
|
batch = download / "Release"
|
|
batch.mkdir(parents=True)
|
|
(batch / "A.mkv").write_bytes(b"a")
|
|
(batch / "B.mkv").write_bytes(b"b")
|
|
main.state.add_manual_batch(batch)
|
|
main.sync_manual_queue()
|
|
main.state.set_app_state("cancel_requested", "true")
|
|
|
|
assert main._import_manual_batches(main.Importer(movies, tv)) == 0
|
|
assert len(main.state.list_queue_items()) == 2
|
|
|
|
|
|
def test_cancel_current_stops_active_copy(tmp_path, monkeypatch):
|
|
main, download, movies, tv = configure_main(tmp_path, monkeypatch)
|
|
batch = download / "Release"
|
|
batch.mkdir(parents=True)
|
|
source = batch / "A.mkv"
|
|
source.write_bytes(b"a" * (1024 * 1024 + 1))
|
|
main.state.add_manual_batch(batch)
|
|
main.sync_manual_queue()
|
|
|
|
calls = 0
|
|
|
|
def cancel_during_copy() -> bool:
|
|
nonlocal calls
|
|
calls += 1
|
|
return calls > 1
|
|
|
|
monkeypatch.setattr(main, "consume_cancel_request", cancel_during_copy)
|
|
|
|
assert main._import_manual_batches(main.Importer(movies, tv)) == 0
|
|
assert source.exists()
|
|
assert not any(movies.glob("*.partial"))
|
|
assert main.state.list_queue_items(active_only=False)[0]["state"] == "skipped"
|
|
|
|
|
|
def test_control_update_runs_configured_command(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
main.settings.update_command = ["upgrade", "now"]
|
|
|
|
calls = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
calls.append((command, kwargs))
|
|
return main.subprocess.CompletedProcess(command, 0, stdout="updated", stderr="")
|
|
|
|
monkeypatch.setattr(main.subprocess, "run", fake_run)
|
|
|
|
result = main.update_service()
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["command"] == ["upgrade", "now"]
|
|
assert result["stdout"] == "updated"
|
|
assert calls[0][0] == ["upgrade", "now"]
|
|
assert calls[0][1].get("shell") is not True
|
|
|
|
|
|
def test_control_restart_reports_command_failure(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
main.settings.restart_command = ["restart"]
|
|
|
|
def fake_run(command, **kwargs):
|
|
return main.subprocess.CompletedProcess(command, 1, stdout="", stderr="failed")
|
|
|
|
monkeypatch.setattr(main.subprocess, "run", fake_run)
|
|
|
|
try:
|
|
main.restart_service()
|
|
except main.HTTPException as exc:
|
|
assert exc.status_code == 500
|
|
assert exc.detail["status"] == "failed"
|
|
assert exc.detail["stderr"] == "failed"
|
|
else:
|
|
raise AssertionError("expected HTTPException")
|