134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import date
|
|
|
|
from monthly_reports.contracts import ErrorCode
|
|
from monthly_reports.repository import DerivedMonthlyRequest
|
|
from monthly_reports.service import RunResult
|
|
from monthly_reports.worker import (
|
|
CLAIM_EVENT_SQL,
|
|
FAIL_EVENT_SQL,
|
|
PUBLISH_EVENT_SQL,
|
|
MonthlyOutboxWorker,
|
|
OutboxEvent,
|
|
)
|
|
|
|
|
|
class FakeOutbox:
|
|
def __init__(self, event: OutboxEvent | None) -> None:
|
|
self.event = event
|
|
self.published: list[tuple[int, int]] = []
|
|
self.failed: list[tuple[OutboxEvent, str, bool]] = []
|
|
|
|
def claim_next(self):
|
|
value, self.event = self.event, None
|
|
return value
|
|
|
|
def mark_published(self, event_id: int, report_id: int) -> None:
|
|
self.published.append((event_id, report_id))
|
|
|
|
def mark_failed(self, event: OutboxEvent, code: str, *, retryable: bool) -> str:
|
|
self.failed.append((event, code, retryable))
|
|
return "pending" if retryable else "dead"
|
|
|
|
|
|
class FakeRequests:
|
|
def __init__(self, derived: DerivedMonthlyRequest | None) -> None:
|
|
self.derived = derived
|
|
self.calls: list[int] = []
|
|
|
|
def derive_monthly_request(self, daily_version_id: int):
|
|
self.calls.append(daily_version_id)
|
|
return self.derived
|
|
|
|
|
|
class FakeService:
|
|
def __init__(self, result_factory) -> None:
|
|
self.result_factory = result_factory
|
|
self.requests = []
|
|
|
|
def run(self, request):
|
|
self.requests.append(request)
|
|
return self.result_factory(request)
|
|
|
|
|
|
def event(payload=None, attempts: int = 1) -> OutboxEvent:
|
|
return OutboxEvent(
|
|
event_id=9,
|
|
event_key="processing-run:27:accepted",
|
|
payload=payload or {"daily_version_id": 4, "business_date": "1999-01-01"},
|
|
publish_attempts=attempts,
|
|
)
|
|
|
|
|
|
class MonthlyReportsWorkerTests(unittest.TestCase):
|
|
def test_worker_uses_derived_arrival_scope_and_publishes_only_real_report(self):
|
|
outbox = FakeOutbox(event())
|
|
requests = FakeRequests(DerivedMonthlyRequest(2026, 7, date(2026, 7, 27)))
|
|
service = FakeService(
|
|
lambda request: RunResult(
|
|
request=request,
|
|
status="success",
|
|
report_version_id=44,
|
|
version_no=1,
|
|
artifact={"sha256": "a" * 64},
|
|
)
|
|
)
|
|
|
|
outcome = MonthlyOutboxWorker(outbox, requests, service).process_next()
|
|
|
|
self.assertEqual(outcome.status, "published")
|
|
self.assertEqual(outcome.report_id, 44)
|
|
self.assertEqual(requests.calls, [4])
|
|
self.assertEqual(service.requests[0].as_of_date, date(2026, 7, 27))
|
|
self.assertEqual(outbox.published, [(9, 44)])
|
|
self.assertEqual(outbox.failed, [])
|
|
|
|
def test_empty_arrival_scope_is_dead_lettered_without_generation(self):
|
|
outbox = FakeOutbox(event())
|
|
service = FakeService(lambda request: None)
|
|
|
|
outcome = MonthlyOutboxWorker(
|
|
outbox,
|
|
FakeRequests(None),
|
|
service,
|
|
).process_next()
|
|
|
|
self.assertEqual(outcome.status, "dead")
|
|
self.assertEqual(outcome.error_code, "MONTHLY_WORKER_ARRIVAL_SCOPE_EMPTY")
|
|
self.assertEqual(service.requests, [])
|
|
self.assertFalse(outbox.failed[0][2])
|
|
|
|
def test_retryable_report_failure_returns_event_to_pending(self):
|
|
outbox = FakeOutbox(event())
|
|
requests = FakeRequests(DerivedMonthlyRequest(2026, 7, date(2026, 7, 27)))
|
|
service = FakeService(
|
|
lambda request: RunResult(
|
|
request=request,
|
|
status="failed",
|
|
error_code=ErrorCode.PUBLISH_FAILED,
|
|
error_stage="publish",
|
|
)
|
|
)
|
|
|
|
outcome = MonthlyOutboxWorker(outbox, requests, service).process_next()
|
|
|
|
self.assertEqual(outcome.status, "pending")
|
|
self.assertEqual(outcome.error_code, ErrorCode.PUBLISH_FAILED)
|
|
self.assertTrue(outbox.failed[0][2])
|
|
self.assertEqual(outbox.published, [])
|
|
|
|
def test_claim_and_ack_sql_are_leased_locked_and_downloadable(self):
|
|
self.assertIn("FOR UPDATE SKIP LOCKED", CLAIM_EVENT_SQL)
|
|
self.assertIn("publish_status IN ('pending', 'publishing')", CLAIM_EVENT_SQL)
|
|
self.assertIn("available_at <= now()", CLAIM_EVENT_SQL)
|
|
self.assertIn("reporting.monthly_runs", PUBLISH_EVENT_SQL)
|
|
self.assertIn("monthly_xlsx", PUBLISH_EVENT_SQL)
|
|
self.assertIn("result_json", PUBLISH_EVENT_SQL)
|
|
self.assertIn("publish_status = %s", FAIL_EVENT_SQL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|