200 lines
7.3 KiB
Python
200 lines
7.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from dataclasses import dataclass
|
|
from typing import Any, cast
|
|
|
|
from services.runtime.ingestion_auth import ActorContext
|
|
from services.runtime.ingestion_hub_client import HubClient
|
|
from services.runtime.ingestion_workflow import (
|
|
IngestionWorkflow,
|
|
IngestionWorkflowError,
|
|
safe_error_response,
|
|
)
|
|
|
|
PROJECT_ID = "11111111-1111-4111-8111-111111111111"
|
|
PROPOSAL_ID = "22222222-2222-4222-8222-222222222222"
|
|
|
|
|
|
@dataclass
|
|
class _Reply:
|
|
data: dict[str, Any]
|
|
etag: str | None = None
|
|
status_code: int = 200
|
|
correlation_id: str = "trace-1"
|
|
|
|
@property
|
|
def id(self) -> str | None:
|
|
value = self.data.get("id")
|
|
return str(value) if value is not None else None
|
|
|
|
|
|
class _Hub:
|
|
def __init__(self, status: str) -> None:
|
|
self.status = status
|
|
self.calls: list[tuple[str, str, dict[str, Any]]] = []
|
|
self.detail = {
|
|
"id": PROPOSAL_ID,
|
|
"correlationId": "trace-hub",
|
|
"status": status,
|
|
"manifest": f'{{"projectId":"{PROJECT_ID}"}}',
|
|
"reviewerComment": "Please verify the target page.",
|
|
"retryCount": 1,
|
|
"maxPublishRetries": 3,
|
|
"isRetryEligible": True,
|
|
"publishedPageId": "55555555-5555-4555-8555-555555555555",
|
|
"publishedPageRevisionNumber": 4,
|
|
"sources": [],
|
|
"citations": [
|
|
{"label": "Verified source", "pageId": "55555555-5555-4555-8555-555555555555"}
|
|
],
|
|
}
|
|
|
|
async def request_json(self, method: str, path: str, **kwargs: Any) -> _Reply:
|
|
self.calls.append((method, path, kwargs))
|
|
if method == "GET" and path == f"/api/v1/ingestion-proposals/{PROPOSAL_ID}":
|
|
return _Reply(dict(self.detail), etag='"v7"')
|
|
if method == "POST" and path == f"/api/v1/ingestion-proposals/{PROPOSAL_ID}/retry":
|
|
self.status = "Analyzing"
|
|
self.detail["status"] = self.status
|
|
return _Reply(dict(self.detail), etag='"v8"')
|
|
raise AssertionError(f"unexpected request: {method} {path}")
|
|
|
|
|
|
class _Authorizer:
|
|
def __call__(self, actor: ActorContext, *, capability: str, project: str | None = None) -> None:
|
|
if capability not in actor.scopes or project not in actor.projects:
|
|
raise PermissionError("not authorized")
|
|
|
|
|
|
class IngestionStatusTests(unittest.IsolatedAsyncioTestCase):
|
|
def actor(self) -> ActorContext:
|
|
return ActorContext(
|
|
subject="user-1",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"ingestion:write"}),
|
|
correlation_id="trace-1",
|
|
)
|
|
|
|
async def test_status_only_projects_confirmed_fields_and_gates_unknown_status(self) -> None:
|
|
hub = _Hub("FutureStatus")
|
|
workflow = IngestionWorkflow(cast(HubClient, hub), actor_authorizer=_Authorizer())
|
|
|
|
result = await workflow.get_ingestion_status(
|
|
project_id=PROJECT_ID,
|
|
proposal_id=PROPOSAL_ID,
|
|
actor=self.actor(),
|
|
)
|
|
|
|
self.assertEqual(result["status"], "Unknown")
|
|
self.assertFalse(result["retryEligible"])
|
|
self.assertFalse(result["terminal"])
|
|
self.assertIsNone(result["finalCitation"])
|
|
self.assertEqual(result["reviewerFeedback"], "Please verify the target page.")
|
|
self.assertEqual(hub.calls[0][1], f"/api/v1/ingestion-proposals/{PROPOSAL_ID}")
|
|
|
|
async def test_known_published_status_returns_only_hub_confirmed_final_citation(self) -> None:
|
|
hub = _Hub("Published")
|
|
workflow = IngestionWorkflow(cast(HubClient, hub), actor_authorizer=_Authorizer())
|
|
|
|
result = await workflow.get_ingestion_status(
|
|
project_id=PROJECT_ID,
|
|
proposal_id=PROPOSAL_ID,
|
|
actor=self.actor(),
|
|
)
|
|
|
|
self.assertEqual(result["status"], "Published")
|
|
self.assertTrue(result["terminal"])
|
|
self.assertFalse(result["retryEligible"])
|
|
self.assertEqual(
|
|
result["finalCitation"],
|
|
{"pageId": "55555555-5555-4555-8555-555555555555", "revisionNumber": 4},
|
|
)
|
|
self.assertEqual(
|
|
result["citations"],
|
|
[{"pageId": "55555555-5555-4555-8555-555555555555"}],
|
|
)
|
|
self.assertNotIn("label", result["citations"][0])
|
|
|
|
async def test_retry_requires_failed_eligible_proposal_and_current_etag(self) -> None:
|
|
hub = _Hub("ReadyForReview")
|
|
workflow = IngestionWorkflow(cast(HubClient, hub), actor_authorizer=_Authorizer())
|
|
with self.assertRaises(ValueError):
|
|
await workflow.retry_failed_proposal(
|
|
project_id=PROJECT_ID,
|
|
proposal_id=PROPOSAL_ID,
|
|
etag='"v7"',
|
|
idempotency_key="retry-operation-001",
|
|
actor=self.actor(),
|
|
)
|
|
self.assertEqual([call[0] for call in hub.calls], ["GET"])
|
|
|
|
failed_hub = _Hub("Failed")
|
|
failed_workflow = IngestionWorkflow(
|
|
cast(HubClient, failed_hub),
|
|
actor_authorizer=_Authorizer(),
|
|
)
|
|
with self.assertRaises(IngestionWorkflowError):
|
|
await failed_workflow.retry_failed_proposal(
|
|
project_id=PROJECT_ID,
|
|
proposal_id=PROPOSAL_ID,
|
|
etag='"stale"',
|
|
idempotency_key="retry-operation-002",
|
|
actor=self.actor(),
|
|
)
|
|
self.assertEqual([call[0] for call in failed_hub.calls], ["GET"])
|
|
|
|
retried = await failed_workflow.retry_failed_proposal(
|
|
project_id=PROJECT_ID,
|
|
proposal_id=PROPOSAL_ID,
|
|
etag='"v7"',
|
|
idempotency_key="retry-operation-003",
|
|
actor=self.actor(),
|
|
)
|
|
self.assertEqual(retried["status"], "Analyzing")
|
|
retry_call = failed_hub.calls[-1]
|
|
self.assertEqual(retry_call[0], "POST")
|
|
self.assertEqual(
|
|
retry_call[1],
|
|
f"/api/v1/ingestion-proposals/{PROPOSAL_ID}/retry",
|
|
)
|
|
self.assertEqual(retry_call[2]["etag"], '"v7"')
|
|
self.assertEqual(
|
|
retry_call[2]["idempotency_key"],
|
|
"retry-operation-003",
|
|
)
|
|
|
|
def test_safe_error_maps_conflict_and_excludes_raw_sensitive_exception_text(self) -> None:
|
|
class UpstreamConflict(Exception):
|
|
status_code = 412
|
|
correlation_id = "trace-412"
|
|
|
|
raw_details = "file-download-host signed-query local-path credential-value"
|
|
response = safe_error_response(UpstreamConflict(raw_details), "trace-412")
|
|
encoded = str(response)
|
|
|
|
self.assertEqual(response["error"]["code"], "stale_version")
|
|
self.assertEqual(response["error"]["action"], "refresh")
|
|
self.assertEqual(response["correlationId"], "trace-412")
|
|
for marker in (
|
|
"file-download-host",
|
|
"signed-query",
|
|
"local-path",
|
|
"credential-value",
|
|
):
|
|
self.assertNotIn(marker, encoded)
|
|
|
|
def test_rate_limit_retry_after_is_bounded(self) -> None:
|
|
class RateLimited(Exception):
|
|
status_code = 429
|
|
retry_after_seconds = 900
|
|
|
|
response = safe_error_response(RateLimited("raw response"), "trace-429")
|
|
self.assertEqual(response["error"]["code"], "rate_limited")
|
|
self.assertEqual(response["error"]["retryAfterSeconds"], 60)
|
|
self.assertNotIn("raw response", str(response))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|