80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
from importarr.state import State
|
|
|
|
|
|
def test_delete_queue_items_by_state_can_target_reason(tmp_path):
|
|
state = State(tmp_path / "state.db")
|
|
state.upsert_queue_item(source_type="sab", source_id="stale", name="Stale", state="ignored", reason="SAB category is not owned by Importarr")
|
|
state.upsert_queue_item(source_type="sab", source_id="other", name="Other", state="ignored", reason="other reason")
|
|
|
|
assert state.delete_queue_items_by_state("sab", "ignored", "SAB category is not owned by Importarr") == 1
|
|
|
|
rows = state.list_queue_items()
|
|
assert len(rows) == 1
|
|
assert rows[0]["source_id"] == "other"
|
|
|
|
|
|
def test_upsert_preserves_terminal_state_by_default(tmp_path):
|
|
state = State(tmp_path / "state.db")
|
|
state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="failed", reason="boom")
|
|
|
|
updated = state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready", reason="rescanned")
|
|
|
|
assert updated["state"] == "failed"
|
|
assert updated["reason"] == "boom"
|
|
|
|
|
|
def test_claim_next_queue_item_marks_importing(tmp_path):
|
|
state = State(tmp_path / "state.db")
|
|
state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
|
|
row = state.claim_next_queue_item("worker-1")
|
|
|
|
assert row is not None
|
|
assert row["state"] == "importing"
|
|
assert row["claimed_by"] == "worker-1"
|
|
|
|
|
|
def test_release_stale_claims_requeues_importing_items(tmp_path):
|
|
state = State(tmp_path / "state.db")
|
|
state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
state.claim_next_queue_item("worker-1")
|
|
|
|
released = state.release_stale_claims()
|
|
row = state.get_queue_item(1)
|
|
|
|
assert released == 1
|
|
assert row["state"] == "retrying"
|
|
assert row["claimed_by"] is None
|
|
|
|
|
|
def test_claim_queue_item_requires_unclaimed_allowed_state(tmp_path):
|
|
state = State(tmp_path / "state.db")
|
|
row = state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
|
|
claimed = state.claim_queue_item(row["id"], "worker-1", {"ready"})
|
|
blocked = state.claim_queue_item(row["id"], "worker-2", {"ready", "importing"})
|
|
|
|
assert claimed is not None
|
|
assert claimed["claimed_by"] == "worker-1"
|
|
assert blocked is None
|
|
|
|
|
|
def test_remove_missing_manual_items_keeps_terminal_rows(tmp_path):
|
|
state = State(tmp_path / "state.db")
|
|
state.upsert_queue_item(source_type="manual", source_id="done", name="Done.mkv", state="imported", batch_id=1)
|
|
state.upsert_queue_item(source_type="manual", source_id="pending", name="Pending.mkv", state="ready", batch_id=1)
|
|
|
|
state.remove_missing_manual_items(1, set())
|
|
|
|
rows = {row["source_id"]: row for row in state.list_queue_items(active_only=False)}
|
|
assert "done" in rows
|
|
assert "pending" not in rows
|
|
|
|
|
|
def test_retry_item_respects_next_retry_at(tmp_path):
|
|
state = State(tmp_path / "state.db")
|
|
state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
|
|
state.mark_queue_item_result("manual", "a", "retrying", "boom", increment_attempts=True, next_retry_seconds=60)
|
|
|
|
assert state.claim_next_queue_item("worker-1") is None
|