Spaces:
Runtime error
Runtime error
File size: 4,555 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """Tests for upload-time redaction strategy wiring."""
from __future__ import annotations
import io
import json
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Document, IngestStatus
from app.services.document_sanitiser import RedactionStrategy
from app.services.redaction_upload import (
InvalidRedactionStrategyError,
build_upload_db_context,
deserialize_db_context,
parse_redaction_strategy,
serialize_db_context,
)
def test_parse_redaction_strategy_defaults_and_aliases() -> None:
assert parse_redaction_strategy(None) is RedactionStrategy.AI_HYBRID
assert parse_redaction_strategy("") is RedactionStrategy.AI_HYBRID
assert parse_redaction_strategy("AI_HYBRID") is RedactionStrategy.AI_HYBRID
assert parse_redaction_strategy("deterministic") is RedactionStrategy.DETERMINISTIC_CODE
assert parse_redaction_strategy("DETERMINISTIC_CODE") is RedactionStrategy.DETERMINISTIC_CODE
def test_parse_redaction_strategy_rejects_unknown() -> None:
with pytest.raises(InvalidRedactionStrategyError):
parse_redaction_strategy("magic_redact")
def test_serialize_deserialize_db_context_roundtrip() -> None:
ctx = {"client_name": "Jane Doe", "property_address": "10 High Street"}
raw = serialize_db_context(ctx)
assert raw is not None
assert deserialize_db_context(raw) == ctx
@pytest.mark.asyncio
async def test_build_upload_db_context_prefers_form_over_tenant_backfill(
test_db: AsyncSession,
) -> None:
ctx = await build_upload_db_context(
test_db,
"tenant_test",
client_name="Explicit Client",
property_address="99 Form Street",
surveyor_name="Form Surveyor",
)
assert ctx["client_name"] == "Explicit Client"
assert ctx["property_address"] == "99 Form Street"
assert ctx["surveyor_name"] == "Form Surveyor"
def _minimal_docx_bytes() -> bytes:
from docx import Document
buf = io.BytesIO()
doc = Document()
doc.add_paragraph("Upload redaction strategy fixture.")
doc.save(buf)
return buf.getvalue()
@pytest.fixture
def noop_ingest(monkeypatch: pytest.MonkeyPatch) -> None:
async def _noop(doc_id: str, file_path) -> None: # noqa: ANN001, ARG001
return None
monkeypatch.setattr("app.ingest.pipeline.ingest_document", _noop)
@pytest.mark.asyncio
async def test_upload_accepts_deterministic_strategy(
async_client: AsyncClient,
test_db: AsyncSession,
noop_ingest: None,
) -> None:
files = [
(
"file",
(
"report.docx",
_minimal_docx_bytes(),
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
),
)
]
data = {
"redaction_strategy": "deterministic",
"client_name": "Jane Client",
"property_address": "1 Test Lane",
}
res = await async_client.post("/upload", files=files, data=data)
assert res.status_code == 202
body = res.json()
assert body["redaction_strategy"] == "deterministic"
doc = await test_db.get(Document, body["document_id"])
assert doc is not None
assert doc.redaction_strategy == "deterministic"
assert json.loads(doc.redaction_context_json or "{}")["client_name"] == "Jane Client"
@pytest.mark.asyncio
async def test_upload_rejects_invalid_redaction_strategy(
async_client: AsyncClient,
test_db: AsyncSession,
noop_ingest: None,
) -> None:
files = [
(
"file",
(
"report.docx",
_minimal_docx_bytes(),
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
),
)
]
res = await async_client.post(
"/upload",
files=files,
data={"redaction_strategy": "not_a_real_strategy"},
)
assert res.status_code == 400
assert "Invalid redaction_strategy" in res.json()["detail"]
def test_ingest_sanitisation_gate_for_deterministic_strategy() -> None:
"""Explicit deterministic uploads must always enter the sanitiser path."""
from app.db.models import DocumentPurpose
from app.services.document_sanitiser import RedactionStrategy
strategy = RedactionStrategy.DETERMINISTIC_CODE
doc_purpose = DocumentPurpose.report_source
needs_sanitise = (
doc_purpose == DocumentPurpose.style_corpus
or strategy is RedactionStrategy.DETERMINISTIC_CODE
)
assert needs_sanitise is True
|