File size: 5,083 Bytes
e4e6ceb
 
 
4124165
e4e6ceb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
497c3ef
e4e6ceb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
497c3ef
 
e4e6ceb
497c3ef
 
 
 
 
 
 
 
 
 
 
 
e4e6ceb
 
 
 
 
 
 
497c3ef
e4e6ceb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
497c3ef
 
e4e6ceb
497c3ef
 
 
 
 
 
 
 
 
e4e6ceb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
from __future__ import annotations

from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from bl_ocr.api.main import create_app
from bl_ocr.application.service import InferenceRequest, InferenceResult, InferenceMetadata
from bl_ocr.core.config import Settings


def get_test_settings() -> Settings:
    return Settings(
        service_name="bl-ocr-backend-test",
        api_version="v1",
        mistral_api_key="fake-key",
        mistral_ocr_url="https://example.test/ocr",
        mistral_model="test-model",
        mistral_strict_schema=True,
        request_timeout_seconds=5.0,
        max_retries=1,
        max_upload_mb=1,
        schema_path=Path("schema.json"),
        example_path=Path("example.json"),
        cors_allow_origins=("*",),
    )


class MockInferenceService:
    def __init__(self, settings: Settings) -> None:
        self.settings = settings
        self.last_request = None

    def run(self, request: InferenceRequest) -> InferenceResult:
        self.last_request = request
        return InferenceResult(
            filename=request.filename,
            content_type=request.content_type,
            checksum_md5="d41d8cd98f00b204e9800998ecf8427e",
            annotation={"extracted": "yes"},
            schema_valid=True,
            schema_errors=[],
            validation_warnings=["warning1"] if request.validate else [],
            repairs=(
                [
                    {
                        "path": "products[0].pharmacist_price_ttc",
                        "original": 10.0,
                        "value": 6.0,
                        "reason": "derived from line_total_ttc / quantity",
                    }
                ]
                if request.repair
                else []
            ),
            raw_provider_response={"document_annotation": {"extracted": "yes"}},
            parse_error=None,
            metadata=InferenceMetadata(
                provider="mistral",
                model=self.settings.mistral_model,
                retry_count=0,
                critical_retry_reasons=[],
                retry_reasons=[],
                duration_ms=10,
            ),
        )


@pytest.fixture
def client_and_service() -> tuple[TestClient, MockInferenceService]:
    settings = get_test_settings()
    service = MockInferenceService(settings)
    app = create_app(settings=settings, inference_service=service)
    client = TestClient(app)
    return client, service


def test_health_endpoint(client_and_service):
    client, _ = client_and_service
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "ok"
    assert data["service"] == "bl-ocr-backend-test"
    assert data["provider_configured"] is True


def test_schema_endpoint(client_and_service):
    client, _ = client_and_service
    response = client.get("/v1/schema")
    assert response.status_code == 200
    data = response.json()
    # Should load the schema from schema.json
    assert "$schema" in data or "type" in data


def test_ocr_upload_endpoint(client_and_service):
    client, service = client_and_service
    files = {"file": ("test.jpg", b"fake-image-bytes", "image/jpeg")}
    response = client.post("/v1/ocr?repair=true&validate=true", files=files)
    
    assert response.status_code == 200
    data = response.json()
    assert data["filename"] == "test.jpg"
    assert data["content_type"] == "image/jpeg"
    assert data["annotation"] == {"extracted": "yes"}
    assert data["schema_valid"] is True
    assert data["schema_errors"] == []
    assert data["validation_warnings"] == ["warning1"]
    assert data["repairs"] == [
        {
            "path": "products[0].pharmacist_price_ttc",
            "original": 10.0,
            "value": 6.0,
            "reason": "derived from line_total_ttc / quantity",
        }
    ]

    assert service.last_request is not None
    assert service.last_request.filename == "test.jpg"
    assert service.last_request.content_type == "image/jpeg"
    assert service.last_request.repair is True
    assert service.last_request.validate is True


def test_ocr_base64_endpoint(client_and_service):
    client, service = client_and_service
    # base64 representation of b"fake-image-bytes" is ZmFrZS1pbWFnZS1ieXRlcw==
    payload = {
        "image_base64": "ZmFrZS1pbWFnZS1ieXRlcw==",
        "filename": "test_b64.jpg",
        "content_type": "image/jpeg",
        "repair": True,
        "validate": False
    }
    response = client.post("/v1/ocr/base64", json=payload)
    
    assert response.status_code == 200
    data = response.json()
    assert data["filename"] == "test_b64.jpg"
    assert data["content_type"] == "image/jpeg"
    assert data["validation_warnings"] == []
    
    assert service.last_request is not None
    assert service.last_request.filename == "test_b64.jpg"
    assert service.last_request.image_bytes == b"fake-image-bytes"
    assert service.last_request.repair is True
    assert service.last_request.validate is False