Spaces:
Sleeping
Sleeping
File size: 3,171 Bytes
b76f199 | 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 | from __future__ import annotations
import base64
import uuid
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Document, IngestStatus, Report, ReportStatus
def _tiny_png_bytes() -> bytes:
# 1x1 PNG
return base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO6q3d8AAAAASUVORK5CYII="
)
@pytest.mark.asyncio
async def test_photo_policy_endpoint_returns_sections(async_client: AsyncClient, test_db: AsyncSession) -> None:
tenant_id = "tenant_test"
doc = Document(
id=str(uuid.uuid4()),
tenant_id=tenant_id,
filename="x.pdf",
file_path="C:/tmp/x.pdf",
status=IngestStatus.complete,
survey_level=3,
)
rep = Report(
id=str(uuid.uuid4()),
tenant_id=tenant_id,
document_id=doc.id,
status=ReportStatus.pending,
survey_level=3,
)
test_db.add_all([doc, rep])
await test_db.commit()
r = await async_client.get(f"/reports/{rep.id}/photo-policy")
assert r.status_code == 200
data = r.json()
assert data["report_id"] == rep.id
assert data["survey_level"] == 3
assert isinstance(data["sections"], list)
assert any(x["code"] == "D" for x in data["sections"])
assert "policy_configuration_required" in data
assert isinstance(data["policy_configuration_required"], bool)
assert "indexed_upload_count" in data
assert isinstance(data["indexed_upload_count"], int)
for row in data["sections"]:
assert "source" in row
assert row["source"] in (
"override",
"tenant_library",
"tenant_insufficient",
"tenant_no_documents",
"data_driven_disabled",
)
@pytest.mark.asyncio
async def test_upload_list_and_get_section_photo_roundtrip(async_client: AsyncClient, test_db: AsyncSession, tmp_path) -> None:
tenant_id = "tenant_test"
doc = Document(
id=str(uuid.uuid4()),
tenant_id=tenant_id,
filename="x.pdf",
file_path="C:/tmp/x.pdf",
status=IngestStatus.complete,
survey_level=3,
)
rep = Report(
id=str(uuid.uuid4()),
tenant_id=tenant_id,
document_id=doc.id,
status=ReportStatus.pending,
survey_level=3,
)
test_db.add_all([doc, rep])
await test_db.commit()
from app.config import settings
settings.upload_dir = tmp_path / "uploads"
settings.upload_dir.mkdir(parents=True, exist_ok=True)
png = _tiny_png_bytes()
files = [("files", ("p.png", png, "image/png"))]
up = await async_client.post(f"/reports/{rep.id}/sections/D/photos", files=files)
assert up.status_code == 200
saved = up.json()["saved"]
assert saved and saved[0]["photo_id"]
lst = await async_client.get(f"/reports/{rep.id}/sections/D/photos")
assert lst.status_code == 200
photos = lst.json()["photos"]
assert len(photos) == 1
url = photos[0]["url"]
img = await async_client.get(url)
assert img.status_code == 200
assert img.headers["content-type"].startswith("image/")
|