Chris CoutinhoandClaude Opus 4.8
44f72839ed
- tests: make the transient-backoff progression assertion load-independent by bracketing the get_retry_decision call with before/after timestamps instead of measuring against a second datetime.now() (no freezegun dependency). - tests: use pytest.approx for the ingest-queue-depth gauge assertions — SonarCloud python:S1244 (float == ) was a MAJOR reliability finding that tripped the new_reliability_rating quality gate. - processor: tighten the EscalateError lazy-bind comment (file processing already imports the document stack via get_registry; the gating only spares the delete / text-doc paths and module-load time). Deck #323. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
123 lines
4.9 KiB
Python
123 lines
4.9 KiB
Python
"""Unit tests for the per-tier escalation primitives (Deck #323).
|
|
|
|
Covers the tier-ladder helpers + EscalateError (document_processors.escalation)
|
|
and the procrastinate TieredEscalationStrategy that turns a raised exception
|
|
into a queue-hop / same-tier retry / give-up decision.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
import pytest
|
|
from procrastinate.jobs import Job
|
|
|
|
import nextcloud_mcp_server.vector.queue.procrastinate as pq
|
|
from nextcloud_mcp_server.document_processors.escalation import (
|
|
TIER_LADDER,
|
|
EscalateError,
|
|
next_tier,
|
|
)
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
def _job(queue: str = pq.INGEST_QUEUE_FAST, attempts: int = 1) -> Job:
|
|
return Job(
|
|
id=1,
|
|
queue=queue,
|
|
task_name=pq.INGEST_TASK_NAME,
|
|
lock=None,
|
|
queueing_lock=None,
|
|
attempts=attempts,
|
|
)
|
|
|
|
|
|
class TestLadder:
|
|
def test_next_tier_ordering(self):
|
|
assert next_tier("fast") == "structured"
|
|
assert next_tier("structured") == "ocr"
|
|
assert next_tier("ocr") is None # terminal
|
|
assert next_tier("unknown") is None
|
|
|
|
def test_ladder_is_cheapest_first(self):
|
|
assert TIER_LADDER == ("fast", "structured", "ocr")
|
|
|
|
def test_tier_for_queue(self):
|
|
assert pq.tier_for_queue(pq.INGEST_QUEUE_OCR) == "ocr"
|
|
assert pq.tier_for_queue(pq.INGEST_QUEUE_STRUCTURED) == "structured"
|
|
# Legacy / unknown / None all fall back to the cheapest tier.
|
|
assert pq.tier_for_queue(pq.LEGACY_INGEST_QUEUE) == "fast"
|
|
assert pq.tier_for_queue(None) == "fast"
|
|
|
|
|
|
class TestTieredEscalationStrategy:
|
|
def _strategy(self, max_transient: int = 5):
|
|
return pq.TieredEscalationStrategy(max_transient_attempts=max_transient)
|
|
|
|
def test_escalate_hops_to_target_queue(self):
|
|
exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text")
|
|
decision = self._strategy().get_retry_decision(exception=exc, job=_job())
|
|
assert decision is not None
|
|
assert decision.queue == pq.INGEST_QUEUE_OCR
|
|
|
|
def test_escalate_to_structured(self):
|
|
exc = EscalateError(
|
|
from_tier="fast", to_tier="structured", reason="low_confidence"
|
|
)
|
|
decision = self._strategy().get_retry_decision(exception=exc, job=_job())
|
|
assert decision is not None
|
|
assert decision.queue == pq.INGEST_QUEUE_STRUCTURED
|
|
|
|
def test_escalate_unknown_tier_gives_up(self):
|
|
exc = EscalateError(from_tier="ocr", to_tier="bogus", reason="low_confidence")
|
|
decision = self._strategy().get_retry_decision(exception=exc, job=_job())
|
|
assert decision is None
|
|
|
|
def test_escalate_unwraps_exception_group(self):
|
|
exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text")
|
|
group = ExceptionGroup("wrapped", [exc])
|
|
decision = self._strategy().get_retry_decision(exception=group, job=_job())
|
|
assert decision is not None
|
|
assert decision.queue == pq.INGEST_QUEUE_OCR
|
|
|
|
def test_transient_retries_same_queue_under_cap(self):
|
|
decision = self._strategy(max_transient=5).get_retry_decision(
|
|
exception=httpx.ConnectError("refused"), job=_job(attempts=1)
|
|
)
|
|
assert decision is not None
|
|
# Same-tier retry: no queue override (stays on its current queue).
|
|
assert decision.queue is None
|
|
assert decision.retry_at is not None
|
|
|
|
def test_transient_backoff_progression(self):
|
|
# min(4 * 2**(attempts-1), 300): 4, 8, 16, ... capped at 300s.
|
|
# procrastinate sets retry_at = utcnow() + wait at call time. Bracketing
|
|
# the call with before/after makes the assertion exact and independent of
|
|
# runner load: with before <= call_now <= after, we have
|
|
# (retry_at - after) <= wait <= (retry_at - before).
|
|
strat = self._strategy(max_transient=100)
|
|
for attempts, expected in [(1, 4), (2, 8), (3, 16), (4, 32), (20, 300)]:
|
|
before = datetime.now(timezone.utc)
|
|
decision = strat.get_retry_decision(
|
|
exception=httpx.ConnectError("x"), job=_job(attempts=attempts)
|
|
)
|
|
after = datetime.now(timezone.utc)
|
|
assert decision is not None and decision.retry_at is not None
|
|
lo = (decision.retry_at - after).total_seconds()
|
|
hi = (decision.retry_at - before).total_seconds()
|
|
assert lo <= expected <= hi, (
|
|
f"attempts={attempts}: expected={expected}s not in [{lo:.3f}, {hi:.3f}]"
|
|
)
|
|
|
|
def test_transient_gives_up_over_cap(self):
|
|
decision = self._strategy(max_transient=5).get_retry_decision(
|
|
exception=httpx.ConnectError("refused"), job=_job(attempts=5)
|
|
)
|
|
assert decision is None
|
|
|
|
def test_non_transient_error_gives_up(self):
|
|
decision = self._strategy().get_retry_decision(
|
|
exception=ValueError("permanent"), job=_job(attempts=1)
|
|
)
|
|
assert decision is None
|