d27dd0ffbb
Close 12 audit-driven platform-compliance gaps on a single branch. - D4 dispatch: app/integrations/dispatch_client.py participant `legacyhub`, emits LegacyhubDocumentIndexed + AssetDerivativeReady after the indexing commit (idempotent uuid5), http_inbox route (reindex/tombstone) with audit-based dedupe; docs/dispatch-contract.md. Celery+Redis stays intra-module. - D2 SSO: app/integrations/identity.py validates X-TeamHub-* + role/scope mapper; security.py adds trusted-header enforcement (AUTH_REQUIRE_IDENTITY) and a scope check on /search; docker-compose.teamhub.yml (external teamhub_net + internal db net, api not host-published); RUNBOOK network/firewall section. - Asset standard: SearchHit/Citation carry asset_id/owner_module; buckets renamed teamhub-legacyhub-* (+quarantine/tmp/exports); purge-by-asset_id with legal-hold guard (app/indexing/projection.py); OCR-markdown derivative event. - audit_log model + Alembic 0003 + record_audit on writes (same transaction). - Secret masking: app/common/json_logger.py recursive mask wired into structlog (+ensure_ascii=False); event payloads redacted before persistence. - Service X-API-Key mandatory on ingest endpoints (defence-in-depth). - Port: host API 8000->8050 (collision with SalesHUB/MailHUB resolved), container still listens on 8000. - Config: no plaintext secret defaults; fail-loud in non-dev (no value leak). - Docs drift: README PG 5440, layered-auth note, 5173 removed from CORS; ingest/folder gated by ENABLE_FOLDER_INGEST (410 by default). - ADRs: layers mapping, shared-core extraction, UI locale (RU-first). Tests: 78 passing (ruff, compileall, pytest, tsc, vite build, compose config). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
"""Tests for asset projection purge (legal-hold aware)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.indexing.projection import AssetPurgeTarget, purge_projection_by_asset_id
|
|
|
|
|
|
class _FakeRepo:
|
|
def __init__(self, target: AssetPurgeTarget) -> None:
|
|
self.target = target
|
|
self.purged: tuple[str, list[str]] | None = None
|
|
|
|
def find_asset_documents(self, asset_id: str) -> AssetPurgeTarget:
|
|
return self.target
|
|
|
|
def purge_chunks(self, asset_id: str, document_ids: list[str]) -> None:
|
|
self.purged = (asset_id, document_ids)
|
|
|
|
|
|
def test_unknown_asset_is_noop():
|
|
repo = _FakeRepo(AssetPurgeTarget(found=False))
|
|
calls: list[list[str]] = []
|
|
result = purge_projection_by_asset_id("a1", repository=repo, search_purge=calls.append)
|
|
assert result["purged"] is False
|
|
assert result["reason"] == "unknown_asset"
|
|
assert calls == []
|
|
assert repo.purged is None
|
|
|
|
|
|
def test_legal_hold_blocks_purge():
|
|
repo = _FakeRepo(AssetPurgeTarget(found=True, document_ids=["d1"], legal_hold=True))
|
|
calls: list[list[str]] = []
|
|
result = purge_projection_by_asset_id("a1", repository=repo, search_purge=calls.append)
|
|
assert result["purged"] is False
|
|
assert result["reason"] == "legal_hold"
|
|
assert calls == [] # search index NOT touched under legal hold
|
|
assert repo.purged is None # chunks NOT deleted
|
|
|
|
|
|
def test_purge_removes_chunks_and_search_entries():
|
|
repo = _FakeRepo(AssetPurgeTarget(found=True, document_ids=["d1", "d2"], legal_hold=False))
|
|
calls: list[list[str]] = []
|
|
result = purge_projection_by_asset_id("a1", repository=repo, search_purge=calls.append)
|
|
assert result["purged"] is True
|
|
assert result["documents"] == ["d1", "d2"]
|
|
assert repo.purged == ("a1", ["d1", "d2"])
|
|
assert calls == [["d1", "d2"]]
|