File size: 9,271 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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
"""Legacy UI compatibility routes backed by v2 generation."""

from __future__ import annotations

import io
import zipfile
from pathlib import Path

import pytest
from docx import Document
from fastapi.testclient import TestClient

from backend.config import settings
from backend.core import ingest, template_discoverer
from backend.main import app


@pytest.fixture
def client():
    return TestClient(app)


def _seed_tenant(tmp_path: Path, monkeypatch, tenant: str = "legacy-ui") -> None:
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    monkeypatch.setattr(settings, "openai_api_key", "")

    tpl = tmp_path / "template.docx"
    doc = Document()
    doc.add_paragraph("D2 Roof coverings")
    doc.add_paragraph("D1 Chimney stacks")
    doc.save(str(tpl))

    past = tmp_path / "past.docx"
    doc2 = Document()
    doc2.add_paragraph("D2 Roof coverings")
    doc2.add_paragraph(
        "The roof is covered with concrete interlocking tiles laid to pitched timber rafters."
    )
    doc2.add_paragraph("D1 Chimney stacks")
    doc2.add_paragraph("Chimney pots appear serviceable from ground level.")
    doc2.save(str(past))

    ingest.ingest_report_template(tenant, tpl)
    ingest.ingest_reference(tenant, past)


def test_legacy_auth_passphrase_and_catalog(tmp_path, monkeypatch, client):
    tenant = "legacy-auth"
    _seed_tenant(tmp_path, monkeypatch, tenant)

    r = client.post("/auth/register", json={"tenant_id": tenant, "passphrase": "secret123"})
    assert r.status_code == 201
    body = r.json()
    assert body["access_token"]
    assert body["tenant_id"] == tenant
    token = body["access_token"]

    r = client.get("/auth/me", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200

    r = client.get("/templates/catalog?survey_level=3", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    cat = r.json()
    assert any(s["code"] == "D2" for s in cat["sections"])


def test_legacy_upload_to_export_flow(tmp_path, monkeypatch, client):
    tenant = "legacy-flow"
    _seed_tenant(tmp_path, monkeypatch, tenant)

    r = client.post("/auth/register", json={"tenant_id": tenant, "passphrase": "secret123"})
    token = r.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    past = tmp_path / "ref.docx"
    doc = Document()
    doc.add_paragraph("D2 Roof coverings")
    doc.add_paragraph("The roof comprises slate tiles to pitched rafters.")
    doc.save(str(past))

    with past.open("rb") as fh:
        r = client.post(
            "/upload/batch",
            headers=headers,
            files=[("files", ("ref.docx", fh, "application/vnd.openxmlformats-officedocument.wordprocessingml.document"))],
        )
    assert r.status_code == 201
    up = r.json()
    assert up["accepted"] == 1
    doc_id = up["items"][0]["document_id"]

    r = client.post("/documents/batch-status", headers=headers, json={"document_ids": [doc_id]})
    assert r.json()["complete"] == 1

    r = client.get("/documents/tenant-chunk-summary", headers=headers)
    assert r.json()["indexed_chunk_count"] > 0

    r = client.post(f"/reports?document_id={doc_id}&survey_level=3", headers=headers)
    assert r.status_code == 200
    report_id = r.json()["report_id"]

    r = client.get(f"/reports/{report_id}/photo-policy", headers=headers)
    assert r.status_code == 200

    r = client.post(
        f"/reports/{report_id}/generate",
        headers=headers,
        json={
            "template_id": "D2",
            "bullets": ["slate tile slipped south slope"],
            "mode": "generate",
            "interference_level": "minimum",
        },
    )
    assert r.status_code == 200

    for _ in range(60):
        st = client.get(f"/reports/{report_id}/status", headers=headers).json()
        if st["status"] in ("complete", "partial", "failed"):
            break
        import time
        time.sleep(0.1)
    assert st["status"] == "complete"

    r = client.get(f"/reports/{report_id}/sections", headers=headers)
    sections = r.json()["sections"]
    assert "D2" in sections
    assert sections["D2"]["text"]
    assert sections["D2"]["provenance"]

    r = client.get(f"/reports/{report_id}/export?format=docx", headers=headers)
    assert r.status_code == 200
    with zipfile.ZipFile(io.BytesIO(r.content)) as zf:
        xml = zf.read("word/document.xml").decode("utf-8")
    assert "slate" in xml.lower() or "roof" in xml.lower()
    prov = sections["D2"]["provenance"]
    source_name = prov[0].get("filename") or prov[0].get("doc_id") or ""
    if source_name.startswith("reference:"):
        source_name = source_name.split(":", 1)[-1]
    assert source_name in xml
    assert "Source:" in xml


def test_serves_legacy_index_html(client):
    r = client.get("/")
    assert r.status_code == 200
    assert "Report Genius" in r.text


def test_extract_notes_txt(client):
    r = client.post(
        "/extract-notes",
        files=[("file", ("notes.txt", b"E2: slate tile slipped\nD1: chimney pots ok", "text/plain"))],
    )
    assert r.status_code == 200
    body = r.json()
    assert body["line_count"] >= 2
    assert any("slate" in ln.lower() for ln in body["lines"])


def test_route_notes_screenshot_cases(client):
    r = client.post(
        "/route-notes",
        json={
            "lines": [
                "Main roof: Pitched valley-style roof with imitation slate covering. "
                "Some tiles have slipped. Valley gutters appear to be in good condition.",
                "Rainwater fittings: UPVC gutters and gullies present. Rainwater downpipe discharges below ground.",
            ]
        },
    )
    assert r.status_code == 200
    body = r.json()
    assert body["routed_line_count"] == 2
    assert any("Main roof" in ln for ln in body["routed"].get("D2", []))
    assert any("Rainwater fittings" in ln for ln in body["routed"].get("D3", []))
    assert "D4" not in body["routed"]


def test_document_delete_and_similarity(tmp_path, monkeypatch, client):
    tenant = "legacy-docs"
    _seed_tenant(tmp_path, monkeypatch, tenant)

    r = client.post("/auth/register", json={"tenant_id": tenant, "passphrase": "secret123"})
    token = r.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    past = tmp_path / "ref2.docx"
    doc = Document()
    doc.add_paragraph("D2 Roof coverings")
    doc.add_paragraph("Concrete tile roof in fair condition with minor moss.")
    doc.save(str(past))

    with past.open("rb") as fh:
        r = client.post(
            "/upload/batch",
            headers=headers,
            files=[("files", ("ref2.docx", fh, "application/vnd.openxmlformats-officedocument.wordprocessingml.document"))],
        )
    doc_id = r.json()["items"][0]["document_id"]

    r = client.post(
        "/content/similar",
        headers=headers,
        json={
            "text": "concrete tile roof moss growth south elevation",
            "section_code": "D2",
            "peer_sections": {},
            "limit": 5,
            "min_relevance_percent": 10,
        },
    )
    assert r.status_code == 200
    sim = r.json()
    assert "library_matches" in sim

    r = client.delete(f"/documents/{doc_id}", headers=headers)
    assert r.status_code == 200
    assert r.json()["deleted"] is True

    r = client.get("/documents", headers=headers)
    ids = [d["document_id"] for d in r.json()["documents"]]
    assert doc_id not in ids


def test_proofread_mode(tmp_path, monkeypatch, client):
    tenant = "legacy-proof"
    _seed_tenant(tmp_path, monkeypatch, tenant)

    r = client.post("/auth/register", json={"tenant_id": tenant, "passphrase": "secret123"})
    token = r.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    past = tmp_path / "past.docx"
    doc = Document()
    doc.add_paragraph("D2 Roof coverings")
    doc.add_paragraph("The roof comprises slate tiles to pitched rafters.")
    doc.save(str(past))

    with past.open("rb") as fh:
        up = client.post(
            "/upload/batch",
            headers=headers,
            files=[("files", ("past.docx", fh, "application/vnd.openxmlformats-officedocument.wordprocessingml.document"))],
        )
    doc_id = up.json()["items"][0]["document_id"]

    r = client.post(f"/reports?document_id={doc_id}&survey_level=3", headers=headers)
    report_id = r.json()["report_id"]

    draft = "The roof comprise slate tiles to pitched rafters."
    r = client.post(
        f"/reports/{report_id}/generate",
        headers=headers,
        json={
            "template_id": "D2",
            "bullets": ["slate tiles"],
            "mode": "proofread",
            "draft_paragraph": draft,
        },
    )
    assert r.status_code == 200

    import time
    for _ in range(60):
        st = client.get(f"/reports/{report_id}/status", headers=headers).json()
        if st["status"] in ("complete", "failed"):
            break
        time.sleep(0.1)
    assert st["status"] == "complete"

    sections = client.get(f"/reports/{report_id}/sections", headers=headers).json()["sections"]
    assert sections["D2"]["mode"] == "proofread"
    assert sections["D2"]["text"]
    assert sections["D2"].get("style_profile") is not None