from __future__ import annotations import hashlib import uuid from pathlib import Path from typing import Any from fastapi.testclient import TestClient from app.api import routes_ingestion from app.api.schemas import AssetManifestEnvelope, KnowledgeIngestResponse from app.config import settings from app.ingestion.knowledge_ingest import ( ExistingKnowledgeIngestJob, KnowledgeIngestRegistration, accept_knowledge_ingest, ) from app.main import app def _manifest_payload(*, payload: bytes | None = None, gate_status: str = "approved") -> dict[str, Any]: body = payload or b"%PDF-1.4\nTeamHUB asset manifest\n" sha = hashlib.sha256(body).hexdigest() return { "manifest": { "manifest_version": "1.0", "asset": { "asset_id": "asset_01JZ8W0MAILHUBDOC0001", "owner_module": "mailhub", "owner_record_type": "document", "owner_record_id": "doc_01JZ8W0MAILHUBDOC0001", "asset_kind": "document", "title": "incoming-letter.pdf", "original_filename": "incoming-letter.pdf", "content_type": "application/pdf", "size_bytes": len(body), "sha256": sha, "created_at": "2026-06-14T12:00:00Z", "created_by": "staff:123", }, "storage": { "provider": "s3", "bucket": "teamhub-mailhub-originals", "object_key": "mailhub/2026/06/14/asset_01JZ8W0/original/incoming-letter.pdf", "version_id": None, }, "security": { "gate_status": gate_status, "classification": "internal", "av_status": "clean", "content_scan_status": "clean", "pii_status": "reviewed", "approved_at": "2026-06-14T12:03:00Z", "approved_by": "staff:123", "quarantine_reason": None, }, "retention": { "policy_id": "mailhub-default", "retain_until": None, "legal_hold": False, "legal_hold_reason": None, }, "derivatives": [], "links": [], "source": { "correlation_id": "corr_01JZ8W0", "source_system": "mailhub", "source_event_id": "evt_01JZ8W0", }, }, "ingest_options": { "generate_derivatives": True, "index_full_text": True, "index_vectors": True, "emit_events": True, }, } def test_knowledge_ingest_route_accepts_asset_manifest(monkeypatch): job_id = uuid.uuid4() seen: dict[str, Any] = {} def fake_accept(req: AssetManifestEnvelope, *, idempotency_key: str | None = None): seen["asset_id"] = req.manifest.asset.asset_id seen["idempotency_key"] = idempotency_key return KnowledgeIngestResponse( status="accepted", ingest_job_id=job_id, asset_id=req.manifest.asset.asset_id, idempotency_key=idempotency_key, ) monkeypatch.setattr(routes_ingestion, "accept_knowledge_ingest", fake_accept) client = TestClient(app) res = client.post( f"{settings.app_api_prefix}/knowledge-ingest", json=_manifest_payload(), headers={"Idempotency-Key": "evt_01JZ8W0"}, ) assert res.status_code == 202, res.text assert res.json() == { "status": "accepted", "ingest_job_id": str(job_id), "asset_id": "asset_01JZ8W0MAILHUBDOC0001", "idempotency_key": "evt_01JZ8W0", "reason_code": None, } assert seen == { "asset_id": "asset_01JZ8W0MAILHUBDOC0001", "idempotency_key": "evt_01JZ8W0", } def test_knowledge_ingest_rejects_local_filesystem_paths(monkeypatch): def must_not_run(*_args, **_kwargs): raise AssertionError("service must not run for invalid local paths") monkeypatch.setattr(routes_ingestion, "accept_knowledge_ingest", must_not_run) body = _manifest_payload() body["manifest"]["storage"]["object_key"] = "/data/input/incoming-letter.pdf" client = TestClient(app) res = client.post(f"{settings.app_api_prefix}/knowledge-ingest", json=body) assert res.status_code == 422 assert "local filesystem paths" in res.text def test_knowledge_ingest_route_returns_rejected_for_unapproved_gate(): client = TestClient(app) res = client.post( f"{settings.app_api_prefix}/knowledge-ingest", json=_manifest_payload(gate_status="quarantined"), ) assert res.status_code == 200, res.text assert res.json()["status"] == "rejected" assert res.json()["reason_code"] == "security_gate_not_approved" def test_knowledge_ingest_rejects_unapproved_security_gate(): class MustNotTouchStorage: def get_to_path(self, *_args, **_kwargs): raise AssertionError("storage must not be read before security approval") class MustNotTouchRepository: def find_existing_job(self, *_args, **_kwargs): raise AssertionError("repository must not be read before security approval") req = AssetManifestEnvelope.model_validate(_manifest_payload(gate_status="quarantined")) res = accept_knowledge_ingest( req, storage=MustNotTouchStorage(), repository=MustNotTouchRepository(), enqueue_document=lambda _document_id, _job_id: None, cache_original=lambda _document_id, _sha, _path: None, ) assert res.status == "rejected" assert res.reason_code == "security_gate_not_approved" assert res.ingest_job_id is None def test_knowledge_ingest_verifies_object_and_enqueues_once(tmp_path: Path): payload = b"%PDF-1.4\nverified asset bytes\n" req = AssetManifestEnvelope.model_validate(_manifest_payload(payload=payload)) document_id = uuid.uuid4() job_id = uuid.uuid4() class FakeStorage: def __init__(self) -> None: self.calls: list[tuple[str, str, str | None]] = [] def get_to_path(self, bucket: str, key: str, dest: Path, version_id: str | None = None): self.calls.append((bucket, key, version_id)) dest.write_bytes(payload) return dest class FakeRepository: def __init__(self) -> None: self.verified_size: int | None = None self.verified_sha256: str | None = None def find_existing_job(self, *_args, **_kwargs): return None def register_verified_manifest(self, manifest, ingest_options, verified, idempotency_key): self.verified_size = verified.size_bytes self.verified_sha256 = verified.sha256 return KnowledgeIngestRegistration( job_id=job_id, document_id=document_id, duplicate=False, ) storage = FakeStorage() repository = FakeRepository() enqueued: list[tuple[uuid.UUID, uuid.UUID]] = [] cached: list[tuple[uuid.UUID, str, Path]] = [] res = accept_knowledge_ingest( req, idempotency_key="evt_01JZ8W0", storage=storage, repository=repository, enqueue_document=lambda doc_id, run_id: enqueued.append((doc_id, run_id)), cache_original=lambda doc_id, sha, path: cached.append((doc_id, sha, path)), temp_dir=tmp_path, ) assert res.status == "accepted" assert res.ingest_job_id == job_id assert res.idempotency_key == "evt_01JZ8W0" assert storage.calls == [ ( "teamhub-mailhub-originals", "mailhub/2026/06/14/asset_01JZ8W0/original/incoming-letter.pdf", None, ) ] assert repository.verified_size == len(payload) assert repository.verified_sha256 == hashlib.sha256(payload).hexdigest() assert enqueued == [(document_id, job_id)] assert len(cached) == 1 assert cached[0][0] == document_id assert cached[0][1] == hashlib.sha256(payload).hexdigest() assert cached[0][2].read_bytes() == payload def test_knowledge_ingest_duplicate_does_not_read_or_enqueue(): existing_job_id = uuid.uuid4() req = AssetManifestEnvelope.model_validate(_manifest_payload()) class MustNotTouchStorage: def get_to_path(self, *_args, **_kwargs): raise AssertionError("duplicate manifest must not be downloaded again") class FakeRepository: def find_existing_job(self, manifest): assert manifest.asset.asset_id == "asset_01JZ8W0MAILHUBDOC0001" return ExistingKnowledgeIngestJob(job_id=existing_job_id) def register_verified_manifest(self, *_args, **_kwargs): raise AssertionError("duplicate manifest must not be registered again") res = accept_knowledge_ingest( req, storage=MustNotTouchStorage(), repository=FakeRepository(), enqueue_document=lambda _document_id, _job_id: (_ for _ in ()).throw( AssertionError("duplicate manifest must not enqueue a job") ), cache_original=lambda _document_id, _sha, _path: None, ) assert res.status == "duplicate" assert res.ingest_job_id == existing_job_id assert res.asset_id == "asset_01JZ8W0MAILHUBDOC0001"