File size: 5,492 Bytes
c7d5985
 
 
 
 
 
 
 
 
bbbfba8
c7d5985
 
 
 
 
 
 
 
 
 
 
bbbfba8
 
 
c7d5985
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Integration test: PaddleOCR β†’ CanonicalDocument β†’ ALTO + PAGE (dual export).

Validates that the same canonical document can be serialized to both
ALTO XML and PAGE XML, and that both outputs are structurally correct.
"""

from __future__ import annotations

import json
from typing import TYPE_CHECKING

from lxml import etree
from src.app.domain.models import RawProviderPayload
from src.app.domain.models.geometry import GeometryContext
from src.app.normalization.pipeline import normalize
from src.app.policies.document_policy import DocumentPolicy
from src.app.policies.export_policy import check_alto_export, check_page_export
from src.app.serializers.alto_xml import ALTO_NS, serialize_alto
from src.app.serializers.page_xml import PAGE_NS, serialize_page_xml
from src.app.validators.export_eligibility_validator import compute_export_eligibility

if TYPE_CHECKING:
    from pathlib import Path


class TestDualExport:
    """Full pipeline: raw β†’ canon β†’ validate β†’ ALTO + PAGE."""

    def test_both_exports_from_same_document(self, fixtures_dir: Path) -> None:
        # 1. Load and normalize
        with open(fixtures_dir / "paddle_ocr_sample.json") as f:
            paddle_output = json.load(f)

        raw = RawProviderPayload(
            provider_id="paddleocr",
            adapter_id="adapter.word_box_json.v1",
            runtime_type="local",
            payload=paddle_output,
            image_width=2480, image_height=3508,
        )
        geo_ctx = GeometryContext(source_width=2480, source_height=3508)

        doc = normalize(
            raw, family="word_box_json", geometry_context=geo_ctx,
            document_id="dual_export_test", source_filename="page.png",
        )

        # 2. Check export eligibility
        policy = DocumentPolicy()
        eligibility = compute_export_eligibility(doc, policy)

        alto_decision = check_alto_export(eligibility, policy)
        page_decision = check_page_export(eligibility, policy)

        assert alto_decision.allowed
        assert page_decision.allowed

        # 3. Serialize both
        alto_bytes = serialize_alto(doc)
        page_bytes = serialize_page_xml(doc)

        assert alto_bytes
        assert page_bytes

        # 4. Parse both
        alto_root = etree.fromstring(alto_bytes)
        page_root = etree.fromstring(page_bytes)

        # 5. Validate ALTO structure
        alto_strings = alto_root.findall(f".//{{{ALTO_NS}}}String")
        assert len(alto_strings) == 5
        assert alto_strings[0].get("CONTENT") == "Bonjour"

        # 6. Validate PAGE structure
        page_words = page_root.findall(f".//{{{PAGE_NS}}}Word")
        assert len(page_words) == 5
        w1_te = page_words[0].find(f".//{{{PAGE_NS}}}Unicode")
        assert w1_te.text == "Bonjour"

        # 7. Verify same word count in both
        assert len(alto_strings) == len(page_words)

        # 8. Verify same text content in both
        alto_texts = [s.get("CONTENT") for s in alto_strings]
        page_texts = [
            w.find(f".//{{{PAGE_NS}}}Unicode").text for w in page_words
        ]
        assert alto_texts == page_texts

    def test_page_has_coords_alto_has_hpos(self, fixtures_dir: Path) -> None:
        """PAGE uses Coords/points, ALTO uses HPOS/VPOS/WIDTH/HEIGHT."""
        with open(fixtures_dir / "paddle_ocr_sample.json") as f:
            paddle_output = json.load(f)

        raw = RawProviderPayload(
            provider_id="paddleocr", adapter_id="v1", runtime_type="local",
            payload=paddle_output, image_width=2480, image_height=3508,
        )
        geo_ctx = GeometryContext(source_width=2480, source_height=3508)

        doc = normalize(
            raw, family="word_box_json", geometry_context=geo_ctx,
            document_id="coord_test",
        )

        alto_root = etree.fromstring(serialize_alto(doc))
        page_root = etree.fromstring(serialize_page_xml(doc))

        # ALTO: String has HPOS, VPOS, WIDTH, HEIGHT
        alto_s = alto_root.find(f".//{{{ALTO_NS}}}String")
        assert alto_s.get("HPOS") is not None
        assert alto_s.get("VPOS") is not None
        assert alto_s.get("WIDTH") is not None
        assert alto_s.get("HEIGHT") is not None

        # PAGE: Word has Coords with points
        page_w = page_root.find(f".//{{{PAGE_NS}}}Word")
        coords = page_w.find(f"{{{PAGE_NS}}}Coords")
        assert coords is not None
        points = coords.get("points")
        assert points is not None
        # Points should be polygon (4 vertices from PaddleOCR)
        parts = points.split()
        assert len(parts) == 4

    def test_page_has_reading_order(self, fixtures_dir: Path) -> None:
        with open(fixtures_dir / "paddle_ocr_sample.json") as f:
            paddle_output = json.load(f)

        raw = RawProviderPayload(
            provider_id="paddleocr", adapter_id="v1", runtime_type="local",
            payload=paddle_output, image_width=2480, image_height=3508,
        )
        geo_ctx = GeometryContext(source_width=2480, source_height=3508)

        doc = normalize(
            raw, family="word_box_json", geometry_context=geo_ctx,
            document_id="ro_test",
        )

        page_root = etree.fromstring(serialize_page_xml(doc))
        ro = page_root.find(f".//{{{PAGE_NS}}}ReadingOrder")
        assert ro is not None
        refs = ro.findall(f".//{{{PAGE_NS}}}RegionRefIndexed")
        assert len(refs) >= 1
        assert refs[0].get("regionRef") == "tb1"