549 lines
22 KiB
Python
549 lines
22 KiB
Python
import asyncio
|
|
import threading
|
|
|
|
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_default_control_commands_target_manual_import_service(monkeypatch):
|
|
monkeypatch.delenv("IMPORTARR_START_COMMAND", raising=False)
|
|
monkeypatch.delenv("IMPORTARR_STOP_COMMAND", raising=False)
|
|
monkeypatch.delenv("IMPORTARR_RESTART_COMMAND", raising=False)
|
|
|
|
settings = Settings.from_env()
|
|
|
|
assert settings.start_command == ["systemctl", "start", "manual-media-import.service"]
|
|
assert settings.stop_command == ["systemctl", "stop", "manual-media-import.service"]
|
|
assert settings.restart_command == ["systemctl", "restart", "manual-media-import.service"]
|
|
|
|
|
|
def test_start_control_starts_manual_import_service(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
main.settings.start_command = ["systemctl", "start", "manual-media-import.service"]
|
|
|
|
calls = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
calls.append(command)
|
|
return main.subprocess.CompletedProcess(command, 0, stdout="started", stderr="")
|
|
|
|
monkeypatch.setattr(main.subprocess, "run", fake_run)
|
|
|
|
result = main.start_queue()
|
|
|
|
assert result["control"]["queue_mode"] == "running"
|
|
assert result["command_result"]["command"] == ["systemctl", "start", "manual-media-import.service"]
|
|
assert calls == [["systemctl", "start", "manual-media-import.service"]]
|
|
|
|
|
|
def test_stop_control_stops_manual_import_service(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
main.settings.stop_command = ["systemctl", "stop", "manual-media-import.service"]
|
|
|
|
calls = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
calls.append(command)
|
|
return main.subprocess.CompletedProcess(command, 0, stdout="stopped", stderr="")
|
|
|
|
monkeypatch.setattr(main.subprocess, "run", fake_run)
|
|
|
|
result = main.stop_queue()
|
|
|
|
assert result["control"]["queue_mode"] == "stopped"
|
|
assert result["control"]["cancel_requested"] is True
|
|
assert result["command_result"]["command"] == ["systemctl", "stop", "manual-media-import.service"]
|
|
assert calls == [["systemctl", "stop", "manual-media-import.service"]]
|
|
|
|
|
|
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
|
|
assert jobs[0]["state"] == "ready"
|
|
|
|
|
|
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="ready")
|
|
main.state.upsert_queue_item(source_type="manual", source_id=str(waiting), source_path=waiting, name=waiting.name, state="ready")
|
|
|
|
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"] == "ready"
|
|
|
|
|
|
def test_current_job_status_includes_progress(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
monkeypatch.setattr(main.time, "time", lambda: 110.0)
|
|
|
|
main.set_current_job("Movie.mkv", bytes_copied=50, total_bytes=200, started_at=100.0)
|
|
|
|
current = main.control_status()["current"]
|
|
assert current["file"] == "Movie.mkv"
|
|
assert current["bytes_copied"] == 50
|
|
assert current["total_bytes"] == 200
|
|
assert current["percent"] == 25
|
|
assert current["elapsed_seconds"] == 10
|
|
|
|
|
|
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_status_includes_queue_counts(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
main.state.upsert_queue_item(source_type="manual", source_id="b", name="B.mkv", state="failed")
|
|
|
|
status = main.status()
|
|
|
|
assert status["queue_total"] == 2
|
|
assert status["queue_counts"]["ready"] == 1
|
|
assert status["queue_counts"]["failed"] == 1
|
|
|
|
|
|
def test_worker_claimed_failure_retries_item(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")
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id=str(source), source_path=source, name=source.name, state="ready")
|
|
claimed = main.state.claim_next_queue_item(main.WORKER_ID)
|
|
|
|
class BrokenImporter:
|
|
def import_file(self, *args, **kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
imported = main._import_queue_item(claimed, BrokenImporter(), from_worker=True)
|
|
updated = main.state.get_queue_item(row["id"])
|
|
|
|
assert imported == 0
|
|
assert updated["state"] == "retrying"
|
|
assert updated["attempt_count"] == 1
|
|
assert updated["next_retry_at"] is not None
|
|
|
|
|
|
def test_startup_releases_stale_claims_from_previous_worker(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
main.state.claim_queue_item(row["id"], "old-worker", {"ready"})
|
|
|
|
main.ensure_worker_running()
|
|
updated = main.state.get_queue_item(row["id"])
|
|
main.stop_worker()
|
|
|
|
assert updated["state"] == "retrying"
|
|
assert updated["claimed_by"] is None
|
|
|
|
|
|
def test_run_now_conflicts_when_item_is_claimed(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
main.state.claim_queue_item(row["id"], "other-worker", {"ready"})
|
|
|
|
try:
|
|
main.queue_item_action(row["id"], main.QueueItemActionRequest(action="run-now"))
|
|
except main.HTTPException as exc:
|
|
assert exc.status_code == 409
|
|
else:
|
|
raise AssertionError("expected HTTPException")
|
|
|
|
|
|
def test_bulk_run_now_skips_item_claimed_by_worker(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")
|
|
created_batch = main.state.add_manual_batch(batch)
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id=str(source), source_path=source, name=source.name, state="ready", batch_id=created_batch["id"])
|
|
main.state.claim_queue_item(row["id"], "worker", {"ready"})
|
|
|
|
assert main._import_manual_batches(main.Importer(movies, tv)) == 0
|
|
assert source.exists()
|
|
assert main.state.get_queue_item(row["id"])["claimed_by"] == "worker"
|
|
|
|
|
|
def test_retry_loses_atomic_race_with_worker_claim(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="failed")
|
|
original = main.state.transition_queue_item_if_unclaimed
|
|
|
|
def claim_then_transition(*args, **kwargs):
|
|
main.state.claim_queue_item(row["id"], "worker", {"failed"})
|
|
return original(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(main.state, "transition_queue_item_if_unclaimed", claim_then_transition)
|
|
|
|
try:
|
|
main.queue_item_action(row["id"], main.QueueItemActionRequest(action="retry"))
|
|
except main.HTTPException as exc:
|
|
assert exc.status_code == 409
|
|
else:
|
|
raise AssertionError("expected HTTPException")
|
|
assert main.state.get_queue_item(row["id"])["claimed_by"] == "worker"
|
|
|
|
|
|
def test_ignore_loses_atomic_race_with_worker_claim(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
original = main.state.transition_queue_item_if_unclaimed
|
|
|
|
def claim_then_transition(*args, **kwargs):
|
|
main.state.claim_queue_item(row["id"], "worker", {"ready"})
|
|
return original(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(main.state, "transition_queue_item_if_unclaimed", claim_then_transition)
|
|
|
|
try:
|
|
main.queue_item_action(row["id"], main.QueueItemActionRequest(action="ignore"))
|
|
except main.HTTPException as exc:
|
|
assert exc.status_code == 409
|
|
else:
|
|
raise AssertionError("expected HTTPException")
|
|
assert main.state.get_queue_item(row["id"])["claimed_by"] == "worker"
|
|
|
|
|
|
def test_shutdown_timeout_does_not_release_live_worker_claim(tmp_path, monkeypatch):
|
|
main, download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
source = download / "A.mkv"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_bytes(b"a")
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id=str(source), source_path=source, name=source.name, state="ready")
|
|
started = threading.Event()
|
|
finish = threading.Event()
|
|
|
|
class BlockingImporter:
|
|
def import_file(self, *args, **kwargs):
|
|
started.set()
|
|
finish.wait()
|
|
raise RuntimeError("stopped")
|
|
|
|
def active_worker():
|
|
claimed = main.state.claim_queue_item(row["id"], main.WORKER_ID, {"ready"})
|
|
main._import_queue_item(claimed, BlockingImporter(), from_worker=True)
|
|
|
|
worker = threading.Thread(target=active_worker)
|
|
monkeypatch.setattr(main, "_worker_thread", worker)
|
|
monkeypatch.setattr(main, "WORKER_SHUTDOWN_TIMEOUT_SECONDS", 0.01)
|
|
worker.start()
|
|
assert started.wait(1)
|
|
|
|
main.shutdown_queue_worker()
|
|
|
|
assert worker.is_alive()
|
|
assert main.state.get_queue_item(row["id"])["claimed_by"] == main.WORKER_ID
|
|
|
|
finish.set()
|
|
worker.join(1)
|
|
assert not worker.is_alive()
|
|
assert main.state.get_queue_item(row["id"])["claimed_by"] is None
|
|
|
|
|
|
def test_manual_sync_preserves_claim_after_source_is_unlinked(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")
|
|
created_batch = main.state.add_manual_batch(batch)
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id=str(source), source_path=source, name=source.name, state="ready", batch_id=created_batch["id"])
|
|
main.state.claim_queue_item(row["id"], main.WORKER_ID, {"ready"})
|
|
|
|
source.unlink()
|
|
main.sync_manual_queue()
|
|
|
|
preserved = main.state.get_queue_item(row["id"])
|
|
assert preserved["state"] == "importing"
|
|
assert preserved["claimed_by"] == main.WORKER_ID
|
|
|
|
|
|
def test_bulk_sab_import_preserves_job_metadata(tmp_path, monkeypatch):
|
|
main, download, movies, tv = configure_main(tmp_path, monkeypatch)
|
|
release = download / "Release"
|
|
release.mkdir(parents=True)
|
|
source = release / "Movie.mkv"
|
|
source.write_bytes(b"movie")
|
|
|
|
class FakeSabnzbdClient:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
async def active_nzo_ids(self):
|
|
return set()
|
|
|
|
async def history(self):
|
|
return {"history": {"slots": [{"nzo_id": "SAB-123", "name": "Release", "category": "manual", "status": "Completed", "storage": str(release)}]}}
|
|
|
|
monkeypatch.setattr(main, "SabnzbdClient", FakeSabnzbdClient)
|
|
|
|
assert asyncio.run(main._import_ready_sab_jobs(main.Importer(movies, tv))) == 1
|
|
row = main.state.list_queue_items(active_only=False)[0]
|
|
assert row["job_id"] == "SAB-123"
|
|
assert row["sab_category"] == "manual"
|
|
|
|
|
|
def test_remove_conflicts_when_item_is_claimed(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
main.state.claim_queue_item(row["id"], "other-worker", {"ready"})
|
|
|
|
try:
|
|
main.queue_item_action(row["id"], main.QueueItemActionRequest(action="remove"))
|
|
except main.HTTPException as exc:
|
|
assert exc.status_code == 409
|
|
else:
|
|
raise AssertionError("expected HTTPException")
|
|
|
|
|
|
def test_manual_import_completion_persists_completed_row_and_batch_completion(tmp_path, monkeypatch):
|
|
main, download, movies, tv = configure_main(tmp_path, monkeypatch)
|
|
batch = download / "Release"
|
|
batch.mkdir(parents=True)
|
|
source = batch / "Done.mkv"
|
|
source.write_bytes(b"done")
|
|
created_batch = main.state.add_manual_batch(batch)
|
|
|
|
imported = main._import_manual_batches(main.Importer(movies, tv))
|
|
main.sync_manual_queue()
|
|
|
|
assert imported == 1
|
|
rows = {row["name"]: row for row in main.state.list_queue_items(active_only=False)}
|
|
assert rows["Done.mkv"]["state"] == "imported"
|
|
assert main.state.get_queue_item(rows["Done.mkv"]["id"])["state"] == "imported"
|
|
batch_row = next(batch for batch in main.state.list_manual_batches() if batch["id"] == created_batch["id"])
|
|
assert batch_row["status"] == "completed"
|
|
|
|
|
|
def test_worker_failure_stops_retrying_after_limit(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")
|
|
row = main.state.upsert_queue_item(source_type="manual", source_id=str(source), source_path=source, name=source.name, state="ready")
|
|
|
|
class BrokenImporter:
|
|
def import_file(self, *args, **kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
for _ in range(main.MAX_RETRY_ATTEMPTS):
|
|
claimed = main.state.claim_queue_item(row["id"], main.WORKER_ID, {"ready", "retrying"})
|
|
assert claimed is not None
|
|
main._import_queue_item(claimed, BrokenImporter(), from_worker=True)
|
|
current = main.state.get_queue_item(row["id"])
|
|
if current["state"] == "retrying":
|
|
main.state.mark_queue_item("manual", current["source_id"], "ready", "retry window elapsed")
|
|
|
|
updated = main.state.get_queue_item(row["id"])
|
|
assert updated["state"] == "failed"
|
|
assert updated["attempt_count"] == main.MAX_RETRY_ATTEMPTS
|
|
|
|
|
|
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"]
|
|
monkeypatch.setattr(main, "check_update_available", lambda: {"status": "update_available", "current_version": "0.1.0", "latest_version": "0.2.0", "update_available": True, "release_url": None})
|
|
|
|
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"] == "update_available"
|
|
assert result["command_result"]["status"] == "ok"
|
|
assert result["command_result"]["command"] == ["upgrade", "now"]
|
|
assert result["command_result"]["stdout"] == "updated"
|
|
assert calls[0][0] == ["upgrade", "now"]
|
|
assert calls[0][1].get("shell") is not True
|
|
|
|
|
|
def test_control_update_skips_command_when_current(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
main.settings.update_command = ["upgrade", "now"]
|
|
monkeypatch.setattr(main, "check_update_available", lambda: {"status": "current", "current_version": "0.2.0", "latest_version": "v0.2.0", "update_available": False, "release_url": None})
|
|
|
|
def fake_run(command, **kwargs):
|
|
raise AssertionError("update command should not run without a newer release")
|
|
|
|
monkeypatch.setattr(main.subprocess, "run", fake_run)
|
|
|
|
result = main.update_service()
|
|
|
|
assert result["status"] == "current"
|
|
assert result["command"] == ["upgrade", "now"]
|
|
assert result["update_available"] is False
|
|
|
|
|
|
def test_update_check_compares_latest_release(tmp_path, monkeypatch):
|
|
main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
|
|
monkeypatch.setenv("IMPORTARR_VERSION", "0.1.0")
|
|
|
|
class FakeResponse:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"tag_name": "v0.2.0", "html_url": "https://example.test/releases/v0.2.0"}
|
|
|
|
class FakeClient:
|
|
def __init__(self, timeout):
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def get(self, url, headers):
|
|
assert url == main.settings.update_release_url
|
|
assert headers["Accept"] == "application/json"
|
|
return FakeResponse()
|
|
|
|
monkeypatch.setattr(main.httpx, "Client", FakeClient)
|
|
|
|
result = main.check_update_available()
|
|
|
|
assert result["status"] == "update_available"
|
|
assert result["current_version"] == "0.1.0"
|
|
assert result["latest_version"] == "v0.2.0"
|
|
assert result["update_available"] is 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")
|