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()