File size: 2,387 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
"""Active-section short-circuit and anchor cache tests."""

from __future__ import annotations

from unittest.mock import MagicMock, patch

from backend.config import settings
from backend.core import notes_parser, section_mapper, template_discoverer


def _schema(master_docx):
    schema, _ = template_discoverer.discover_schema(master_docx)
    return schema


def test_initialize_section_anchors_embeds_once(monkeypatch):
    calls = {"n": 0}

    class FakeEmbedder:
        def embed_documents(self, texts: list[str]) -> list[list[float]]:
            calls["n"] += 1
            return [[0.1] * 8 for _ in texts]

        def embed_query(self, text: str) -> list[float]:
            return [0.1] * 8

    notes_parser.reset_section_anchors()
    monkeypatch.setattr(
        notes_parser._embeddings,
        "get_embedder",
        lambda: FakeEmbedder(),
    )
    first = notes_parser.initialize_section_anchors()
    second = notes_parser.initialize_section_anchors()
    assert first == second
    assert calls["n"] == 1
    assert notes_parser.anchors_ready()


def test_generate_report_short_circuits_empty_sections(master_docx, tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data"))
    tenant = "short-circuit-tenant"
    schema = _schema(master_docx)
    template_discoverer.ensure_canonical_schema(tenant)

    all_leaf_ids = [sec.id for sec in schema.ordered_sections()]
    only_ids = all_leaf_ids[:20]

    result = section_mapper.generate_report(
        tenant,
        "significant cracking noted to the chimney stack",
        only_section_ids=only_ids,
    )
    assert result.active_section_count <= 2
    assert result.processed_section_count == result.active_section_count
    assert result.processed_section_count < len(only_ids)
    section_ids = {s.section_id for s in result.sections if s.section_id != "UNASSIGNED"}
    assert section_ids <= {"D1"}


def test_estimate_active_sections_from_generate_body():
    body = MagicMock()
    body.bullets_by_section = {
        "D1": ["chimney cracked"],
        "E3": [""],
        "F2": ["   "],
    }
    body.template_id = "D1"
    body.bullets = []
    body.template_ids = ["D1", "E3", "F2", "D4"]

    active = section_mapper.estimate_active_sections_from_generate_body(
        body,
        tenant_id="t1",
        draft_id=None,
    )
    assert active == ["D1"]