263 lines
9.2 KiB
Python
263 lines
9.2 KiB
Python
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"
|