From 2558429d83260a4a648722307b9c6e0ee245a379 Mon Sep 17 00:00:00 2001 From: Whiznificent Date: Wed, 24 Jun 2026 17:00:05 +0100 Subject: [PATCH] feat(ai-service): standardize result envelope across AI endpoints (#609) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add ResultEnvelope base class with result, confidence, reasons, anchor_metadata and trace_id fields. Compose into all v1 response schemas (OCR, fraud, anonymize, humanitarian, proof-of-life) and populate the fields in every endpoint handler. All envelope fields are Optional so existing callers see no breaking changes. trace_id is auto-generated (UUID4) per request. - schemas/envelope.py – new ResultEnvelope Pydantic mixin - schemas/ocr.py – OCRResponse extends ResultEnvelope - schemas/fraud.py – FraudDetectionResponse extends ResultEnvelope - schemas/anonymization.py – AnonymizeResponse extends ResultEnvelope - schemas/humanitarian.py – HumanitarianVerificationResponse extends ResultEnvelope - api/v1/ocr.py – populate envelope (avg confidence, filename anchor) - api/v1/fraud.py – populate envelope (risk-derived confidence, flagged IDs) - api/v1/anonymize.py – populate envelope (rule-based, PII count anchor) - api/v1/humanitarian.py – populate envelope (verdict confidence, provider anchor) - api/v1/proof_of_life.py – populate envelope (confidence passthrough, checks anchor) - tests/test_result_envelope.py – schema + endpoint envelope validation tests --- app/ai-service/api/v1/anonymize.py | 34 +- app/ai-service/api/v1/fraud.py | 34 +- app/ai-service/api/v1/humanitarian.py | 46 ++- app/ai-service/api/v1/ocr.py | 26 ++ app/ai-service/api/v1/proof_of_life.py | 49 ++- app/ai-service/schemas/anonymization.py | 6 +- app/ai-service/schemas/envelope.py | 73 ++++ app/ai-service/schemas/fraud.py | 6 +- app/ai-service/schemas/humanitarian.py | 6 +- app/ai-service/schemas/ocr.py | 6 +- app/ai-service/tests/test_result_envelope.py | 372 +++++++++++++++++++ 11 files changed, 644 insertions(+), 14 deletions(-) create mode 100644 app/ai-service/schemas/envelope.py create mode 100644 app/ai-service/tests/test_result_envelope.py diff --git a/app/ai-service/api/v1/anonymize.py b/app/ai-service/api/v1/anonymize.py index 3ed1e779..261f05fb 100644 --- a/app/ai-service/api/v1/anonymize.py +++ b/app/ai-service/api/v1/anonymize.py @@ -3,6 +3,7 @@ """ import logging +import uuid from fastapi import APIRouter, HTTPException @@ -18,11 +19,42 @@ async def anonymize_text(request: AnonymizeRequest): """Anonymize names, locations, and dates before text is sent to external LLMs.""" import main as _main + trace_id = str(uuid.uuid4()) logger.info("Processing privacy-preserving anonymization request") try: result = _main.pii_scrubber_service.anonymize(request.text) - return AnonymizeResponse(success=True, **result) + + pii_summary = result.get("pii_summary", {}) + total_redacted = ( + pii_summary.get("total", 0) + if isinstance(pii_summary, dict) + else getattr(pii_summary, "total", 0) + ) + + reasons = ["PII scrubbing completed successfully"] + if total_redacted: + reasons.append(f"{total_redacted} PII token(s) redacted") + else: + reasons.append("No PII detected in input") + + # Strip envelope keys so our explicit values take precedence. + _ENVELOPE_KEYS = {"result", "confidence", "reasons", "anchor_metadata", "trace_id"} + safe_result = {k: v for k, v in result.items() if k not in _ENVELOPE_KEYS} + + return AnonymizeResponse( + success=True, + **safe_result, + # Standard result envelope fields (Issue #609) + result="anonymization_complete", + confidence=None, # rule-based; no probabilistic confidence + reasons=reasons, + anchor_metadata={ + "original_length": result.get("original_length"), + "pii_total": total_redacted, + }, + trace_id=trace_id, + ) except Exception as e: logger.error(f"Anonymization failed: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail="Failed to anonymize text") diff --git a/app/ai-service/api/v1/fraud.py b/app/ai-service/api/v1/fraud.py index b5fee119..d173bbcf 100644 --- a/app/ai-service/api/v1/fraud.py +++ b/app/ai-service/api/v1/fraud.py @@ -3,6 +3,7 @@ """ import logging +import uuid from fastapi import APIRouter, HTTPException @@ -23,11 +24,42 @@ async def detect_fraud_endpoint(request: FraudDetectionRequest) -> FraudDetectio statistical outliers relative to the batch are flagged with ``is_flagged=true``. """ + trace_id = str(uuid.uuid4()) + try: results = detect_fraud(request.claims) + flagged_count = sum(r.is_flagged for r in results) + + # Derive top-level confidence as 1 - mean fraud risk score (ensemble certainty). + if results: + mean_risk = sum(r.fraud_risk_score for r in results) / len(results) + confidence = round(1.0 - mean_risk, 4) + else: + confidence = None + + result_label = "fraud_detected" if flagged_count > 0 else "no_fraud_detected" + + reasons: list[str] = [f"Analysed {len(results)} claim(s)"] + if flagged_count: + reasons.append(f"{flagged_count} claim(s) flagged as suspicious") + flagged_ids = [r.claim_id for r in results if r.is_flagged] + if flagged_ids: + reasons.append(f"Flagged claim IDs: {', '.join(flagged_ids)}") + else: + reasons.append("No suspicious patterns detected") + return FraudDetectionResponse( results=results, - flagged_count=sum(r.is_flagged for r in results), + flagged_count=flagged_count, + # Standard result envelope fields (Issue #609) + result=result_label, + confidence=confidence, + reasons=reasons, + anchor_metadata={ + "total_claims": len(results), + "flagged_count": flagged_count, + }, + trace_id=trace_id, ) except Exception as exc: logger.error("Fraud detection failed: %s", exc) diff --git a/app/ai-service/api/v1/humanitarian.py b/app/ai-service/api/v1/humanitarian.py index 8d4dac2c..6e3f83da 100644 --- a/app/ai-service/api/v1/humanitarian.py +++ b/app/ai-service/api/v1/humanitarian.py @@ -3,6 +3,7 @@ """ import logging +import uuid from fastapi import APIRouter @@ -23,6 +24,7 @@ async def verify_humanitarian_claim(request: HumanitarianVerificationRequest): # tests (and any future dependency-injection wiring) works transparently. import main as _main + trace_id = str(uuid.uuid4()) logger.info("Processing humanitarian verification request") try: @@ -44,7 +46,47 @@ async def verify_humanitarian_claim(request: HumanitarianVerificationRequest): ) else: raise exc - return HumanitarianVerificationResponse(success=True, **result) + + # Extract envelope fields from the nested verification dict. + verification = result.get("verification") or {} + confidence_raw = verification.get("confidence") + confidence = float(confidence_raw) if confidence_raw is not None else None + verdict = verification.get("verdict") + summary = verification.get("summary") + + result_label = f"humanitarian_{verdict}" if verdict else "humanitarian_verified" + + reasons: list[str] = [] + if verdict: + reasons.append(f"Verdict: {verdict}") + if summary: + reasons.append(summary) + + # Strip any envelope-conflicting keys from the service result dict before + # spreading, so our explicit envelope values always take precedence. + _ENVELOPE_KEYS = {"result", "confidence", "reasons", "anchor_metadata", "trace_id"} + safe_result = {k: v for k, v in result.items() if k not in _ENVELOPE_KEYS} + + return HumanitarianVerificationResponse( + success=True, + **safe_result, + # Standard result envelope fields (Issue #609) + result=result_label, + confidence=confidence, + reasons=reasons or None, + anchor_metadata={ + "provider": result.get("provider"), + "model": result.get("model"), + "prompt_variant": result.get("prompt_variant"), + }, + trace_id=trace_id, + ) except Exception as e: logger.error("Humanitarian verification failed: %s", str(e), exc_info=True) - return HumanitarianVerificationResponse(success=False, error=str(e)) + return HumanitarianVerificationResponse( + success=False, + error=str(e), + result="humanitarian_error", + reasons=[str(e)], + trace_id=trace_id, + ) diff --git a/app/ai-service/api/v1/ocr.py b/app/ai-service/api/v1/ocr.py index 7afa6026..9b0e6e2a 100644 --- a/app/ai-service/api/v1/ocr.py +++ b/app/ai-service/api/v1/ocr.py @@ -7,6 +7,7 @@ import io import time +import uuid from typing import Annotated from fastapi import APIRouter, File, HTTPException, Request, UploadFile @@ -40,6 +41,7 @@ async def process_ocr( ) -> OCRResponse: """Extract text fields from an uploaded document image.""" start_time = time.time() + trace_id = str(uuid.uuid4()) if image.content_type not in ALLOWED_CONTENT_TYPES: raise HTTPException( @@ -87,6 +89,16 @@ async def process_ocr( processing_time_ms = int((time.time() - start_time) * 1000) + # Derive top-level confidence as the mean of all extracted field confidences. + field_confidences = [f.confidence for f in result.fields.values()] + avg_confidence = ( + sum(field_confidences) / len(field_confidences) if field_confidences else None + ) + + reasons = ["OCR extraction completed successfully"] + if avg_confidence is not None: + reasons.append(f"Average field confidence: {avg_confidence:.2f}") + return OCRResponse( success=True, data=OCRData( @@ -98,6 +110,15 @@ async def process_ocr( processing_time_ms=processing_time_ms, ), processing_time_ms=processing_time_ms, + # Standard result envelope fields (Issue #609) + result="ocr_complete", + confidence=avg_confidence, + reasons=reasons, + anchor_metadata={ + "filename": image.filename, + "content_type": image.content_type, + }, + trace_id=trace_id, ) except HTTPException: @@ -111,4 +132,9 @@ async def process_ocr( "message": str(e), }, processing_time_ms=processing_time_ms, + # Standard result envelope fields (Issue #609) + result="ocr_error", + reasons=[str(e)], + anchor_metadata={"filename": image.filename}, + trace_id=trace_id, ) diff --git a/app/ai-service/api/v1/proof_of_life.py b/app/ai-service/api/v1/proof_of_life.py index 4b1d0853..5acda23a 100644 --- a/app/ai-service/api/v1/proof_of_life.py +++ b/app/ai-service/api/v1/proof_of_life.py @@ -3,11 +3,14 @@ """ import logging +import uuid from typing import Any, Dict, List, Optional from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field +from schemas.envelope import ResultEnvelope + logger = logging.getLogger(__name__) router = APIRouter(tags=["proof-of-life"]) @@ -21,14 +24,19 @@ class ProofOfLifeRequest(BaseModel): confidence_threshold: Optional[float] = Field(default=None, ge=0.0, le=1.0) -class ProofOfLifeResponse(BaseModel): - """Response model for proof-of-life analysis.""" +class ProofOfLifeResponse(ResultEnvelope): + """ + Proof-of-life analysis response – includes the standardised result envelope (Issue #609). + + Backward-compatible: existing fields (is_real_person, confidence, threshold, + checks, reason) are preserved alongside the new envelope fields. + """ is_real_person: bool - confidence: float + confidence: float # overrides Optional[float] from envelope – always present here threshold: float checks: Dict[str, Any] - reason: str + reason: str # kept for backward compat; also surfaced as reasons[0] @router.post("/ai/proof-of-life", response_model=ProofOfLifeResponse) @@ -43,15 +51,44 @@ async def analyze_proof_of_life(request: ProofOfLifeRequest): import main as _main + trace_id = str(uuid.uuid4()) logger.info("Processing proof-of-life verification request") try: - result = _main.proof_of_life_analyzer.analyze( + raw = _main.proof_of_life_analyzer.analyze( selfie_image_base64=request.selfie_image_base64, burst_images_base64=request.burst_images_base64, confidence_threshold=request.confidence_threshold, ) - return result + + # raw may be a dict or a Pydantic model; normalise to dict. + if isinstance(raw, dict): + data = raw + else: + data = raw.model_dump() if hasattr(raw, "model_dump") else dict(raw) + + is_real = data.get("is_real_person", False) + reason_str = data.get("reason", "") + + result_label = "real_person" if is_real else "not_real_person" + reasons = [reason_str] if reason_str else [] + + # Strip envelope keys from data so our explicit values take precedence. + _ENVELOPE_KEYS = {"result", "confidence", "reasons", "anchor_metadata", "trace_id"} + safe_data = {k: v for k, v in data.items() if k not in _ENVELOPE_KEYS} + + return ProofOfLifeResponse( + **safe_data, + # Standard result envelope fields (Issue #609) + result=result_label, + # confidence is a required field in ProofOfLifeResponse, already in safe_data + reasons=reasons or None, + anchor_metadata={ + "threshold": data.get("threshold"), + "checks": data.get("checks"), + }, + trace_id=trace_id, + ) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) except Exception as e: diff --git a/app/ai-service/schemas/anonymization.py b/app/ai-service/schemas/anonymization.py index a9bb9c19..47941841 100644 --- a/app/ai-service/schemas/anonymization.py +++ b/app/ai-service/schemas/anonymization.py @@ -2,6 +2,8 @@ from pydantic import BaseModel, Field +from schemas.envelope import ResultEnvelope + class AnonymizeRequest(BaseModel): text: str = Field(min_length=1, description="Input text to anonymize before LLM processing") @@ -14,7 +16,9 @@ class PIISummary(BaseModel): total: int -class AnonymizeResponse(BaseModel): +class AnonymizeResponse(ResultEnvelope): + """Anonymization endpoint response – includes the standardised result envelope (Issue #609).""" + success: bool anonymized_text: str original_length: int diff --git a/app/ai-service/schemas/envelope.py b/app/ai-service/schemas/envelope.py new file mode 100644 index 00000000..f0d75be3 --- /dev/null +++ b/app/ai-service/schemas/envelope.py @@ -0,0 +1,73 @@ +""" +Standardised result envelope – Issue #609. + +Every AI endpoint response must carry: + + result – a concise, domain-specific summary of the AI decision + (e.g. "ocr_complete", "real_person", "fraud_detected"). + confidence – float in [0, 1] representing model certainty. + reasons – ordered list of human-readable explanation strings. + anchor_metadata – arbitrary key/value pairs that let callers correlate the + response to their own context (claim_id, image_hash, …). + trace_id – UUID-style request identifier for end-to-end tracing. + +All fields are Optional so that callers that have not yet migrated continue to +work (no breaking change), while new consumers can rely on the full contract. + +Usage +----- +Inherit or compose with ``ResultEnvelope`` and populate the fields inside +each endpoint handler:: + + from schemas.envelope import ResultEnvelope + + class OCRResponse(OCRData, ResultEnvelope): + ... + + # in the route handler: + return OCRResponse( + ... + result="ocr_complete", + confidence=avg_confidence, + reasons=["all fields extracted", "high confidence on name"], + anchor_metadata={"filename": image.filename}, + trace_id=str(uuid.uuid4()), + ) +""" + +import uuid +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + + +def _new_trace_id() -> str: + """Generate a fresh UUID4 trace identifier.""" + return str(uuid.uuid4()) + + +class ResultEnvelope(BaseModel): + """Mixin that adds the standard result envelope fields to any response model.""" + + result: Optional[str] = Field( + default=None, + description="Concise AI decision label (e.g. 'ocr_complete', 'real_person').", + ) + confidence: Optional[float] = Field( + default=None, + ge=0.0, + le=1.0, + description="Model certainty in [0, 1].", + ) + reasons: Optional[List[str]] = Field( + default=None, + description="Ordered list of human-readable explanation strings.", + ) + anchor_metadata: Optional[Dict[str, Any]] = Field( + default=None, + description="Caller-supplied or service-derived key/value context.", + ) + trace_id: Optional[str] = Field( + default_factory=_new_trace_id, + description="UUID-style request identifier for end-to-end tracing.", + ) diff --git a/app/ai-service/schemas/fraud.py b/app/ai-service/schemas/fraud.py index 8fa5c2a1..b49d9dbe 100644 --- a/app/ai-service/schemas/fraud.py +++ b/app/ai-service/schemas/fraud.py @@ -1,6 +1,8 @@ from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field +from schemas.envelope import ResultEnvelope + class ClaimMetadata(BaseModel): claim_id: str @@ -22,6 +24,8 @@ class ClaimFraudResult(BaseModel): reason: Optional[str] = None -class FraudDetectionResponse(BaseModel): +class FraudDetectionResponse(ResultEnvelope): + """Fraud detection endpoint response – includes the standardised result envelope (Issue #609).""" + results: List[ClaimFraudResult] flagged_count: int diff --git a/app/ai-service/schemas/humanitarian.py b/app/ai-service/schemas/humanitarian.py index 660ed87b..937267a0 100644 --- a/app/ai-service/schemas/humanitarian.py +++ b/app/ai-service/schemas/humanitarian.py @@ -2,6 +2,8 @@ from pydantic import BaseModel, Field +from schemas.envelope import ResultEnvelope + class HumanitarianVerificationRequest(BaseModel): aid_claim: str = Field(min_length=10, description="Aid claim to verify") @@ -11,7 +13,9 @@ class HumanitarianVerificationRequest(BaseModel): timeout: Optional[float] = Field(default=None, description="Request-level timeout in seconds for provider call") -class HumanitarianVerificationResponse(BaseModel): +class HumanitarianVerificationResponse(ResultEnvelope): + """Humanitarian verification endpoint response – includes the standardised result envelope (Issue #609).""" + success: bool provider: Optional[str] = None model: Optional[str] = None diff --git a/app/ai-service/schemas/ocr.py b/app/ai-service/schemas/ocr.py index 5432d868..f1f005d0 100644 --- a/app/ai-service/schemas/ocr.py +++ b/app/ai-service/schemas/ocr.py @@ -1,5 +1,7 @@ from pydantic import BaseModel, Field +from schemas.envelope import ResultEnvelope + class OCRFieldResult(BaseModel): value: str @@ -12,7 +14,9 @@ class OCRData(BaseModel): processing_time_ms: int -class OCRResponse(BaseModel): +class OCRResponse(ResultEnvelope): + """OCR endpoint response – includes the standardised result envelope (Issue #609).""" + success: bool data: OCRData | None = None error: dict[str, str] | None = None diff --git a/app/ai-service/tests/test_result_envelope.py b/app/ai-service/tests/test_result_envelope.py new file mode 100644 index 00000000..5c04f17d --- /dev/null +++ b/app/ai-service/tests/test_result_envelope.py @@ -0,0 +1,372 @@ +""" +Tests for the standardised result envelope across AI endpoints – Issue #609. + +Every AI endpoint response must include: + result – str label for the AI decision + confidence – float in [0, 1] OR None (for deterministic/rule-based endpoints) + reasons – list of explanation strings OR None + anchor_metadata – dict of correlating context OR None + trace_id – UUID-style string + +Acceptance criteria: + 1. No breaking changes – all previously tested fields still present. + 2. Envelope fields present in every successful response. + 3. trace_id is a non-empty string that looks like a UUID. + 4. Schema validation: ResultEnvelope model enforces confidence bounds. +""" + +import io +import re +import uuid +import pytest +from unittest.mock import patch +from fastapi.testclient import TestClient + +import main +import metrics +from main import app +from schemas.envelope import ResultEnvelope + +client = TestClient(app, follow_redirects=True) + +UUID_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE +) + + +# --------------------------------------------------------------------------- +# Fixture: always report healthy resources so throttle never fires +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _healthy_resources(): + with patch.object(metrics, "check_system_resources", return_value=True): + yield + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + +def assert_envelope(data: dict, *, has_confidence: bool = True) -> None: + """Assert all standard envelope keys are present and well-formed.""" + assert "result" in data, f"Missing 'result' in response: {list(data.keys())}" + assert "reasons" in data, f"Missing 'reasons' in response: {list(data.keys())}" + assert "anchor_metadata" in data, f"Missing 'anchor_metadata' in response: {list(data.keys())}" + assert "trace_id" in data, f"Missing 'trace_id' in response: {list(data.keys())}" + assert "confidence" in data, f"Missing 'confidence' in response: {list(data.keys())}" + + assert data["trace_id"] is not None, "trace_id must not be None" + assert UUID_RE.match(str(data["trace_id"])), ( + f"trace_id does not look like a UUID: {data['trace_id']}" + ) + + if has_confidence and data["confidence"] is not None: + assert 0.0 <= data["confidence"] <= 1.0, ( + f"confidence out of [0, 1] range: {data['confidence']}" + ) + + if data["reasons"] is not None: + assert isinstance(data["reasons"], list), "'reasons' must be a list or None" + + if data["anchor_metadata"] is not None: + assert isinstance(data["anchor_metadata"], dict), ( + "'anchor_metadata' must be a dict or None" + ) + + +# --------------------------------------------------------------------------- +# 1. OCR endpoint +# --------------------------------------------------------------------------- + +class TestOCREnvelope: + def _post_image(self, color="white", size=(100, 100)): + from PIL import Image as PILImage + img = PILImage.new("RGB", size, color=color) + buf = io.BytesIO() + img.save(buf, format="PNG") + return buf.getvalue() + + def test_ocr_envelope_fields_present(self): + response = client.post( + "/v1/ai/ocr", + files={"image": ("test.png", self._post_image(), "image/png")}, + ) + assert response.status_code == 200 + data = response.json() + assert_envelope(data) + + def test_ocr_result_label(self): + response = client.post( + "/v1/ai/ocr", + files={"image": ("test.png", self._post_image(), "image/png")}, + ) + data = response.json() + assert data["result"] in ("ocr_complete", "ocr_error") + + def test_ocr_trace_id_unique_per_request(self): + img = self._post_image() + r1 = client.post("/v1/ai/ocr", files={"image": ("t.png", img, "image/png")}) + r2 = client.post("/v1/ai/ocr", files={"image": ("t.png", img, "image/png")}) + assert r1.json()["trace_id"] != r2.json()["trace_id"] + + def test_ocr_backward_compat_fields_still_present(self): + """Existing fields must not disappear.""" + response = client.post( + "/v1/ai/ocr", + files={"image": ("test.png", self._post_image(), "image/png")}, + ) + data = response.json() + assert "success" in data + assert "processing_time_ms" in data + + def test_legacy_ocr_envelope_fields_present(self): + """Legacy /ai/ocr path must also carry the envelope.""" + response = client.post( + "/ai/ocr", + files={"image": ("test.png", self._post_image(), "image/png")}, + ) + assert response.status_code == 200 + assert_envelope(response.json()) + + +# --------------------------------------------------------------------------- +# 2. Anonymize endpoint +# --------------------------------------------------------------------------- + +class TestAnonymizeEnvelope: + PAYLOAD = {"text": "On 1 Jan 2025, Jane Doe received aid in Lagos."} + + def test_anonymize_envelope_fields_present(self): + response = client.post("/v1/ai/anonymize", json=self.PAYLOAD) + assert response.status_code == 200 + assert_envelope(response.json(), has_confidence=False) + + def test_anonymize_result_label(self): + data = client.post("/v1/ai/anonymize", json=self.PAYLOAD).json() + assert data["result"] == "anonymization_complete" + + def test_anonymize_reasons_is_list(self): + data = client.post("/v1/ai/anonymize", json=self.PAYLOAD).json() + assert isinstance(data["reasons"], list) + assert len(data["reasons"]) >= 1 + + def test_anonymize_trace_id_unique(self): + r1 = client.post("/v1/ai/anonymize", json=self.PAYLOAD).json() + r2 = client.post("/v1/ai/anonymize", json=self.PAYLOAD).json() + assert r1["trace_id"] != r2["trace_id"] + + def test_anonymize_backward_compat(self): + data = client.post("/v1/ai/anonymize", json=self.PAYLOAD).json() + for key in ("success", "anonymized_text", "original_length", "pii_summary"): + assert key in data, f"Legacy field '{key}' missing" + + +# --------------------------------------------------------------------------- +# 3. Humanitarian endpoint +# --------------------------------------------------------------------------- + +class TestHumanitarianEnvelope: + PAYLOAD = { + "aid_claim": "Teams distributed emergency kits to all registered households.", + "supporting_evidence": ["Distribution list #B-17"], + "context_factors": {}, + "provider_preference": "auto", + } + + @pytest.fixture + def fake_verify(self, monkeypatch): + def _verify(aid_claim, supporting_evidence=None, context_factors=None, + provider_preference="auto", **_): + return { + "provider": "openai", + "model": "gpt-4o-mini", + "prompt_variant": "primary", + "verification": { + "verdict": "credible", + "confidence": 0.87, + "summary": "Claim is well-supported by the evidence.", + }, + "raw_response": "{}", + } + monkeypatch.setattr(main.humanitarian_verification_service, "verify_claim", _verify) + + def test_humanitarian_envelope_fields_present(self, fake_verify): + response = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD) + assert response.status_code == 200 + assert_envelope(response.json()) + + def test_humanitarian_result_label(self, fake_verify): + data = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD).json() + assert data["result"].startswith("humanitarian_") + + def test_humanitarian_confidence_from_verification(self, fake_verify): + data = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD).json() + assert data["confidence"] == pytest.approx(0.87) + + def test_humanitarian_reasons_contain_verdict(self, fake_verify): + data = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD).json() + assert any("credible" in r.lower() for r in data["reasons"]) + + def test_humanitarian_trace_id_unique(self, fake_verify): + r1 = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD).json() + r2 = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD).json() + assert r1["trace_id"] != r2["trace_id"] + + def test_humanitarian_backward_compat(self, fake_verify): + data = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD).json() + for key in ("success", "provider", "model", "verification"): + assert key in data, f"Legacy field '{key}' missing" + + def test_humanitarian_error_path_has_envelope(self, monkeypatch): + monkeypatch.setattr( + main.humanitarian_verification_service, + "verify_claim", + lambda **_: (_ for _ in ()).throw(RuntimeError("provider unavailable")), + ) + data = client.post("/v1/ai/humanitarian/verify", json=self.PAYLOAD).json() + assert data["success"] is False + assert "trace_id" in data + assert "result" in data + + +# --------------------------------------------------------------------------- +# 4. Fraud detection endpoint +# --------------------------------------------------------------------------- + +class TestFraudEnvelope: + PAYLOAD = { + "claims": [ + {"claim_id": "C001", "amount": 100.0, "location": "Lagos"}, + {"claim_id": "C002", "amount": 9999.0, "location": "Lagos"}, + ] + } + + def test_fraud_envelope_fields_present(self): + response = client.post("/v1/fraud/detect", json=self.PAYLOAD) + assert response.status_code == 200 + assert_envelope(response.json()) + + def test_fraud_result_label_values(self): + data = client.post("/v1/fraud/detect", json=self.PAYLOAD).json() + assert data["result"] in ("fraud_detected", "no_fraud_detected") + + def test_fraud_confidence_in_range(self): + data = client.post("/v1/fraud/detect", json=self.PAYLOAD).json() + if data["confidence"] is not None: + assert 0.0 <= data["confidence"] <= 1.0 + + def test_fraud_reasons_mention_claim_count(self): + data = client.post("/v1/fraud/detect", json=self.PAYLOAD).json() + assert any("2" in r or "claim" in r.lower() for r in data["reasons"]) + + def test_fraud_trace_id_unique(self): + r1 = client.post("/v1/fraud/detect", json=self.PAYLOAD).json() + r2 = client.post("/v1/fraud/detect", json=self.PAYLOAD).json() + assert r1["trace_id"] != r2["trace_id"] + + def test_fraud_backward_compat(self): + data = client.post("/v1/fraud/detect", json=self.PAYLOAD).json() + for key in ("results", "flagged_count"): + assert key in data, f"Legacy field '{key}' missing" + + def test_fraud_anchor_metadata_has_counts(self): + data = client.post("/v1/fraud/detect", json=self.PAYLOAD).json() + meta = data.get("anchor_metadata") or {} + assert "total_claims" in meta + assert "flagged_count" in meta + + +# --------------------------------------------------------------------------- +# 5. Proof-of-life endpoint +# --------------------------------------------------------------------------- + +class TestProofOfLifeEnvelope: + PAYLOAD = {"selfie_image_base64": "dGVzdA=="} + + @pytest.fixture + def fake_analyze(self, monkeypatch): + def _analyze(selfie_image_base64, burst_images_base64=None, + confidence_threshold=None): + return { + "is_real_person": True, + "confidence": 0.93, + "threshold": confidence_threshold or 0.65, + "checks": {"face_detected": True, "blink_detected": False}, + "reason": "Face detected with high confidence", + } + monkeypatch.setattr(main.proof_of_life_analyzer, "analyze", _analyze) + + def test_proof_of_life_envelope_fields_present(self, fake_analyze): + response = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD) + assert response.status_code == 200 + assert_envelope(response.json()) + + def test_proof_of_life_result_real_person(self, fake_analyze): + data = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD).json() + assert data["result"] == "real_person" + + def test_proof_of_life_confidence_value(self, fake_analyze): + data = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD).json() + assert data["confidence"] == pytest.approx(0.93) + + def test_proof_of_life_reasons_from_reason(self, fake_analyze): + data = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD).json() + assert isinstance(data["reasons"], list) + assert any("face" in r.lower() for r in data["reasons"]) + + def test_proof_of_life_trace_id_unique(self, fake_analyze): + r1 = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD).json() + r2 = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD).json() + assert r1["trace_id"] != r2["trace_id"] + + def test_proof_of_life_backward_compat(self, fake_analyze): + data = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD).json() + for key in ("is_real_person", "confidence", "threshold", "checks", "reason"): + assert key in data, f"Legacy field '{key}' missing" + + def test_proof_of_life_not_real_person_result_label(self, monkeypatch): + def _not_real(selfie_image_base64, burst_images_base64=None, + confidence_threshold=None): + return { + "is_real_person": False, + "confidence": 0.22, + "threshold": 0.65, + "checks": {"face_detected": False}, + "reason": "No face detected", + } + monkeypatch.setattr(main.proof_of_life_analyzer, "analyze", _not_real) + data = client.post("/v1/ai/proof-of-life", json=self.PAYLOAD).json() + assert data["result"] == "not_real_person" + + +# --------------------------------------------------------------------------- +# 6. Schema-level validation – ResultEnvelope +# --------------------------------------------------------------------------- + +class TestResultEnvelopeSchema: + def test_confidence_below_zero_rejected(self): + with pytest.raises(Exception): + ResultEnvelope(confidence=-0.1) + + def test_confidence_above_one_rejected(self): + with pytest.raises(Exception): + ResultEnvelope(confidence=1.1) + + def test_trace_id_auto_generated(self): + envelope = ResultEnvelope() + assert envelope.trace_id is not None + assert UUID_RE.match(envelope.trace_id) + + def test_all_fields_optional(self): + """ResultEnvelope must be constructable with no arguments.""" + envelope = ResultEnvelope() + assert envelope.result is None + assert envelope.confidence is None + assert envelope.reasons is None + assert envelope.anchor_metadata is None + + def test_explicit_trace_id_accepted(self): + tid = str(uuid.uuid4()) + envelope = ResultEnvelope(trace_id=tid) + assert envelope.trace_id == tid