247 lines
11 KiB
Python
247 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from arr_web.app import PortalApplication, RuntimeHealth
|
|
from arr_web.arr_downloads import DownloadOutcome, PersistentARRDownloads, default_report_date, validate_report_date
|
|
from arr_web.contracts import PortalError
|
|
from tests.test_arr_web import TEST_CREDENTIALS, FakeRepository, login
|
|
|
|
|
|
class Executor:
|
|
def __init__(self, outcome=None):
|
|
self.calls = []
|
|
self.release = threading.Event()
|
|
self.entered = threading.Event()
|
|
self.outcome = outcome or DownloadOutcome("succeeded", "arrjob-download-test")
|
|
self.raise_once = False
|
|
|
|
def execute(self, **kwargs):
|
|
self.calls.append(kwargs)
|
|
self.entered.set()
|
|
if not self.release.wait(3):
|
|
raise RuntimeError("test timed out")
|
|
kwargs["report_stage"]("processing")
|
|
if self.raise_once:
|
|
self.raise_once = False
|
|
raise RuntimeError("private credential or guest must never be exposed")
|
|
return self.outcome
|
|
|
|
|
|
class DownloadTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
self.executor = Executor()
|
|
self.service = PersistentARRDownloads(self.root, self.executor)
|
|
|
|
def tearDown(self):
|
|
self.executor.release.set()
|
|
self.service.close()
|
|
self.temp.cleanup()
|
|
|
|
def finished(self, key="a" * 32):
|
|
deadline = time.monotonic() + 4
|
|
while time.monotonic() < deadline:
|
|
result = self.service.get(key)
|
|
if result["status"] not in {"queued", "downloading", "processing"}:
|
|
return result
|
|
time.sleep(.01)
|
|
self.fail("worker did not settle")
|
|
|
|
def test_one_date_passed_as_both_bounds_and_repeat_is_same_job(self):
|
|
task = self.service.create("2026-09-15", "a" * 32)
|
|
self.assertTrue(self.executor.entered.wait(1))
|
|
repeat = self.service.create("2026-09-15", "a" * 32)
|
|
self.assertEqual(task["request_id"], repeat["request_id"])
|
|
self.assertEqual(self.executor.calls[0]["from_date"], date(2026, 9, 15))
|
|
self.assertEqual(self.executor.calls[0]["to_date"], date(2026, 9, 15))
|
|
self.executor.release.set()
|
|
final = self.finished()
|
|
self.assertEqual(final["status"], "succeeded")
|
|
self.assertEqual(self.service.create("2026-09-15", "a" * 32), final)
|
|
self.assertEqual(len(self.executor.calls), 1)
|
|
|
|
def test_same_day_double_click_and_tabs_coalesce(self):
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
ids = list(pool.map(lambda n: self.service.create("2026-09-15", f"{n:032x}")["request_id"], range(8)))
|
|
self.assertEqual(len(set(ids)), 1)
|
|
self.assertTrue(self.executor.entered.wait(1))
|
|
self.assertEqual(len(self.executor.calls), 1)
|
|
|
|
def test_same_id_cannot_change_date(self):
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
with self.assertRaises(PortalError) as raised:
|
|
self.service.create("2026-09-16", "a" * 32)
|
|
self.assertEqual(raised.exception.status, 409)
|
|
|
|
def test_unknown_outcome_retries_exact_identity_without_leaking_exception(self):
|
|
self.executor.raise_once = True
|
|
self.executor.release.set()
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
failed = self.finished()
|
|
self.assertEqual(failed["status"], "interrupted")
|
|
self.assertNotIn("private", json.dumps(failed))
|
|
self.assertEqual(self.service.create("2026-09-15", "b" * 32)["request_id"], "a" * 32)
|
|
self.service.retry("a" * 32)
|
|
self.assertEqual(self.finished()["status"], "succeeded")
|
|
self.assertEqual([c["request_id"] for c in self.executor.calls], ["a" * 32] * 2)
|
|
self.assertEqual([c["from_date"] for c in self.executor.calls], [date(2026, 9, 15)] * 2)
|
|
|
|
def test_review_is_not_success_and_does_not_rerun_on_retry(self):
|
|
self.executor.outcome = DownloadOutcome("needs_review", "arrjob-price-review")
|
|
self.executor.release.set()
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
result = self.finished()
|
|
self.assertEqual(result["status"], "needs_review")
|
|
self.assertFalse(result["can_retry"])
|
|
self.assertEqual(self.service.retry("a" * 32), result)
|
|
self.assertEqual(len(self.executor.calls), 1)
|
|
|
|
def test_capture_only_success_without_processing_job_is_not_accepted(self):
|
|
self.executor.outcome = DownloadOutcome("succeeded")
|
|
self.executor.release.set()
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
self.assertEqual(self.finished()["status"], "interrupted")
|
|
|
|
def test_restart_preserves_success_without_reexecuting(self):
|
|
self.executor.release.set()
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
final = self.finished()
|
|
self.service.close()
|
|
self.service = PersistentARRDownloads(self.root, self.executor)
|
|
self.assertEqual(self.service.latest(), final)
|
|
self.assertEqual(self.service.create("2026-09-15", "a" * 32), final)
|
|
self.assertEqual(len(self.executor.calls), 1)
|
|
|
|
def test_restart_marks_unknown_commit_interrupted_instead_of_failed(self):
|
|
self.executor.release.set()
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
self.finished()
|
|
self.service.close()
|
|
with sqlite3.connect(self.root / "arr-downloads.sqlite3") as db:
|
|
db.execute("UPDATE downloads SET status='processing'")
|
|
self.service = PersistentARRDownloads(self.root, self.executor)
|
|
self.assertEqual(self.service.get("a" * 32)["status"], "interrupted")
|
|
self.assertTrue(self.service.get("a" * 32)["can_retry"])
|
|
self.assertEqual(len(self.executor.calls), 1)
|
|
|
|
def test_second_owner_cannot_start_same_queue(self):
|
|
with self.assertRaises(BlockingIOError):
|
|
PersistentARRDownloads(self.root, self.executor)
|
|
|
|
def test_another_process_cannot_dispatch_from_owned_queue(self):
|
|
script = """from pathlib import Path
|
|
import sys
|
|
from arr_web.arr_downloads import PersistentARRDownloads
|
|
try:
|
|
PersistentARRDownloads(Path(sys.argv[1]), object())
|
|
except BlockingIOError:
|
|
sys.exit(0)
|
|
sys.exit(1)
|
|
"""
|
|
result = subprocess.run([sys.executable, "-c", script, str(self.root)], capture_output=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode())
|
|
|
|
def test_retry_cannot_jump_over_another_unresolved_task_for_same_day(self):
|
|
self.executor.outcome = DownloadOutcome("failed", retryable=True)
|
|
self.executor.release.set()
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
self.finished()
|
|
self.executor.release.clear()
|
|
self.service.create("2026-09-15", "b" * 32)
|
|
with self.assertRaises(PortalError) as raised:
|
|
self.service.retry("a" * 32)
|
|
self.assertEqual(raised.exception.code, "ARR_DOWNLOAD_OTHER_ACTIVE")
|
|
|
|
def test_explicit_new_request_after_success_can_refresh_same_day(self):
|
|
self.executor.release.set()
|
|
self.service.create("2026-09-15", "a" * 32)
|
|
self.finished()
|
|
self.service.create("2026-09-15", "b" * 32)
|
|
self.finished("b" * 32)
|
|
self.assertEqual(len(self.executor.calls), 2)
|
|
|
|
|
|
class DownloadRoutesTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.executor = Executor()
|
|
self.service = PersistentARRDownloads(Path(self.temp.name), self.executor)
|
|
self.repo = FakeRepository()
|
|
self.app = PortalApplication(repository=self.repo, arr_downloads=self.service,
|
|
credentials=TEST_CREDENTIALS, health=RuntimeHealth(True, True, False))
|
|
_, self.headers = login(self.app)
|
|
|
|
def tearDown(self):
|
|
self.executor.release.set()
|
|
self.service.close()
|
|
self.temp.cleanup()
|
|
|
|
def post(self, payload, headers=None):
|
|
return self.app.handle("POST", "/api/arr-downloads", self.headers if headers is None else headers, json.dumps(payload).encode())
|
|
|
|
def test_auth_and_csrf_before_mutation(self):
|
|
self.assertEqual(self.app.handle("GET", "/api/arr-downloads", {}).status, 401)
|
|
request = {"report_date": "2026-09-15", "request_id": "a" * 32}
|
|
self.assertEqual(self.post(request, {}).status, 401)
|
|
self.assertEqual(self.post(request, {"Cookie": self.headers["Cookie"]}).status, 403)
|
|
self.assertIsNone(self.service.latest())
|
|
|
|
def test_reject_range_timestamps_invalid_calendar_days_and_extra_fields(self):
|
|
for value in ("2026-9-15", "2026-02-29", "2026-09-15T00:00:00Z", "0000-01-01", "2026-09-15,2026-09-16", None, 42):
|
|
with self.subTest(value=value):
|
|
self.assertEqual(self.post({"report_date": value, "request_id": "a" * 32}).status, 400)
|
|
self.assertEqual(self.post({"report_date": "2026-09-15", "request_id": "a" * 32, "to_date": "2026-09-16"}).status, 400)
|
|
self.assertIsNone(self.service.latest())
|
|
self.assertEqual(validate_report_date("2024-02-29"), "2024-02-29")
|
|
|
|
def test_create_poll_and_retry_have_stable_public_contract(self):
|
|
response = self.post({"report_date": "2026-09-15", "request_id": "a" * 32})
|
|
self.assertEqual(response.status, 202)
|
|
result = json.loads(response.body)["data"]
|
|
self.assertEqual(result["from_date"], result["to_date"])
|
|
get = self.app.handle("GET", "/api/arr-downloads/" + "a" * 32, self.headers)
|
|
self.assertEqual(get.status, 200)
|
|
retry_path = "/api/arr-downloads/" + "a" * 32 + "/retry"
|
|
self.assertEqual(self.app.handle("POST", retry_path, {}, b"{}").status, 401)
|
|
self.assertEqual(self.app.handle("POST", retry_path, self.headers, b'{"report_date":"2026-09-16"}').status, 400)
|
|
self.assertEqual(self.app.handle("POST", retry_path, self.headers, b"{}").status, 202)
|
|
|
|
def test_default_runtime_is_explicitly_unavailable(self):
|
|
app = PortalApplication(credentials=TEST_CREDENTIALS)
|
|
_, headers = login(app)
|
|
config = json.loads(app.handle("GET", "/api/arr-downloads", headers).body)["data"]
|
|
self.assertFalse(config["ready"])
|
|
self.assertIsNone(config["latest_task"])
|
|
response = app.handle("POST", "/api/arr-downloads", headers,
|
|
json.dumps({"report_date": "2026-09-15", "request_id": "a" * 32}).encode())
|
|
self.assertEqual(response.status, 503)
|
|
|
|
def test_default_day_uses_bangkok_calendar_at_year_boundary(self):
|
|
with patch("arr_web.arr_downloads.datetime") as clock:
|
|
clock.now.return_value = datetime(2026, 1, 1, 0, 1, tzinfo=ZoneInfo("Asia/Bangkok"))
|
|
self.assertEqual(default_report_date(), "2025-12-31")
|
|
clock.now.assert_called_once_with(ZoneInfo("Asia/Bangkok"))
|
|
|
|
def test_review_snapshot_follows_authoritative_finance_status(self):
|
|
task = {"status": "needs_review", "job_id": "arrjob-reviewed", "can_retry": False}
|
|
with patch.object(self.repo, "get_job_trace", return_value={"job": {"status": "accepted"}}):
|
|
self.assertEqual(self.app._arr_download_snapshot(task)["status"], "succeeded")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|