feat(T03): OCR-Erfassung via OpenRouter Qwen2.5-VL + OCR UI with drag-and-drop
This commit is contained in:
@@ -0,0 +1,624 @@
|
||||
"""Tests for OCR module: upload, results, apply, and processing with mocked OpenRouter."""
|
||||
|
||||
import io
|
||||
import uuid
|
||||
from datetime import date
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.database import Base, get_db
|
||||
from app.main import app
|
||||
from app.models.ocr_result import OCRResult, OCRStatus
|
||||
from app.models.vehicle import Vehicle
|
||||
from app.services import ocr_service
|
||||
from app.services.ocr_service import CONFIDENCE_THRESHOLD
|
||||
|
||||
# Ensure all models are registered with Base.metadata
|
||||
from app.models import user, vehicle, ocr_result # noqa: F401
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_vehicle(db_session: AsyncSession) -> Vehicle:
|
||||
"""Create a test vehicle for OCR linking."""
|
||||
vehicle = Vehicle(
|
||||
make="Mercedes",
|
||||
model="Actros",
|
||||
fin="WDB9066351L123456",
|
||||
year=2020,
|
||||
power_kw=350,
|
||||
fuel_type="Diesel",
|
||||
condition="used",
|
||||
availability="available",
|
||||
price=45000,
|
||||
vehicle_type="lkw",
|
||||
)
|
||||
db_session.add(vehicle)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(vehicle)
|
||||
return vehicle
|
||||
|
||||
|
||||
def _make_mock_openrouter_response(
|
||||
confidence: float = 0.85,
|
||||
brand: str = "Mercedes",
|
||||
model: str = "Actros",
|
||||
vin: str = "WDB9066351L123456",
|
||||
first_registration: str = "15.03.2020",
|
||||
mileage: int = 120000,
|
||||
power_kw: int = 350,
|
||||
fuel_type: str = "Diesel",
|
||||
) -> dict:
|
||||
"""Build a mock OpenRouter response dict."""
|
||||
return {
|
||||
"structured_data": {
|
||||
"brand": brand,
|
||||
"model": model,
|
||||
"vin": vin,
|
||||
"first_registration": first_registration,
|
||||
"mileage": mileage,
|
||||
"power_kw": power_kw,
|
||||
"fuel_type": fuel_type,
|
||||
},
|
||||
"confidence_score": confidence,
|
||||
"raw_text": f'{{"brand": "{brand}", "model": "{model}", "vin": "{vin}", "confidence_score": {confidence}}}',
|
||||
}
|
||||
|
||||
|
||||
class TestOCRService:
|
||||
"""Unit tests for ocr_service functions."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_mime_type_valid(self):
|
||||
assert ocr_service.validate_mime_type("image/png") is True
|
||||
assert ocr_service.validate_mime_type("image/jpeg") is True
|
||||
assert ocr_service.validate_mime_type("image/webp") is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_mime_type_invalid(self):
|
||||
assert ocr_service.validate_mime_type("application/pdf") is False
|
||||
assert ocr_service.validate_mime_type("text/plain") is False
|
||||
assert ocr_service.validate_mime_type("") is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_file_size_valid(self):
|
||||
# 1 MB should be valid (limit is 50 MB)
|
||||
assert ocr_service.validate_file_size(1024 * 1024) is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_file_size_too_large(self):
|
||||
# 51 MB should be invalid
|
||||
assert ocr_service.validate_file_size(51 * 1024 * 1024) is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_success(self, db_session: AsyncSession, tmp_path):
|
||||
"""Test uploading a valid image file creates an OCRResult."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
file_bytes = b"fake-image-data"
|
||||
result = await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=file_bytes,
|
||||
file_name="scan.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
assert result.id is not None
|
||||
assert result.status == OCRStatus.pending
|
||||
assert result.file_name == "scan.png"
|
||||
assert result.mime_type == "image/png"
|
||||
assert result.file_path.endswith(".png")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_invalid_mime(self, db_session: AsyncSession):
|
||||
"""Test uploading with invalid MIME type raises ValueError."""
|
||||
with pytest.raises(ValueError, match="Invalid MIME type"):
|
||||
await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"data",
|
||||
file_name="doc.pdf",
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_too_large(self, db_session: AsyncSession):
|
||||
"""Test uploading a file that exceeds size limit raises ValueError."""
|
||||
large_bytes = b"x" * (51 * 1024 * 1024)
|
||||
with pytest.raises(ValueError, match="File size exceeds"):
|
||||
await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=large_bytes,
|
||||
file_name="big.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_result_not_found(self, db_session: AsyncSession):
|
||||
"""Test getting a non-existent OCR result returns None."""
|
||||
result = await ocr_service.get_result(db_session, uuid.uuid4())
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_results_empty(self, db_session: AsyncSession):
|
||||
"""Test listing OCR results when none exist."""
|
||||
items, total = await ocr_service.list_results(db_session)
|
||||
assert total == 0
|
||||
assert items == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_results_with_data(self, db_session: AsyncSession, tmp_path):
|
||||
"""Test listing OCR results with data."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"img1",
|
||||
file_name="scan1.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"img2",
|
||||
file_name="scan2.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
items, total = await ocr_service.list_results(db_session)
|
||||
assert total == 2
|
||||
assert len(items) == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_results_filter_by_vehicle(
|
||||
self, db_session: AsyncSession, test_vehicle: Vehicle, tmp_path
|
||||
):
|
||||
"""Test listing OCR results filtered by vehicle_id."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"img1",
|
||||
file_name="scan1.png",
|
||||
mime_type="image/png",
|
||||
vehicle_id=test_vehicle.id,
|
||||
)
|
||||
await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"img2",
|
||||
file_name="scan2.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
items, total = await ocr_service.list_results(
|
||||
db_session, vehicle_id=test_vehicle.id
|
||||
)
|
||||
assert total == 1
|
||||
assert len(items) == 1
|
||||
assert items[0].vehicle_id == test_vehicle.id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_ocr_high_confidence(
|
||||
self, db_session: AsyncSession, tmp_path
|
||||
):
|
||||
"""Test OCR processing with high confidence sets status to completed."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
ocr_result = await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"fake-image",
|
||||
file_name="scan.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
mock_response = _make_mock_openrouter_response(confidence=0.92)
|
||||
with patch(
|
||||
"app.services.ocr_service.perform_ocr",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_response,
|
||||
):
|
||||
result = await ocr_service.process_ocr(db_session, ocr_result.id)
|
||||
|
||||
assert result.status == OCRStatus.completed
|
||||
assert result.confidence_score == 0.92
|
||||
assert result.structured_data is not None
|
||||
assert result.structured_data["brand"] == "Mercedes"
|
||||
assert result.structured_data["model"] == "Actros"
|
||||
assert result.structured_data["vin"] == "WDB9066351L123456"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_ocr_low_confidence(
|
||||
self, db_session: AsyncSession, tmp_path
|
||||
):
|
||||
"""Test OCR processing with low confidence sets status to manual_review."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
ocr_result = await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"fake-image",
|
||||
file_name="scan.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
mock_response = _make_mock_openrouter_response(confidence=0.45)
|
||||
with patch(
|
||||
"app.services.ocr_service.perform_ocr",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_response,
|
||||
):
|
||||
result = await ocr_service.process_ocr(db_session, ocr_result.id)
|
||||
|
||||
assert result.status == OCRStatus.manual_review
|
||||
assert result.confidence_score == 0.45
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_ocr_openrouter_failure(
|
||||
self, db_session: AsyncSession, tmp_path
|
||||
):
|
||||
"""Test OCR processing when OpenRouter fails sets status to failed."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
ocr_result = await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"fake-image",
|
||||
file_name="scan.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
with patch(
|
||||
"app.services.ocr_service.perform_ocr",
|
||||
new_callable=AsyncMock,
|
||||
side_effect=Exception("OpenRouter API unavailable"),
|
||||
):
|
||||
result = await ocr_service.process_ocr(db_session, ocr_result.id)
|
||||
|
||||
assert result.status == OCRStatus.failed
|
||||
assert result.error_message is not None
|
||||
assert "OpenRouter API unavailable" in result.error_message
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_to_vehicle_success(
|
||||
self, db_session: AsyncSession, test_vehicle: Vehicle, tmp_path
|
||||
):
|
||||
"""Test applying OCR data to a vehicle."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
ocr_result = await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"fake-image",
|
||||
file_name="scan.png",
|
||||
mime_type="image/png",
|
||||
vehicle_id=test_vehicle.id,
|
||||
)
|
||||
# Set structured data manually
|
||||
ocr_result.structured_data = {
|
||||
"brand": "MAN",
|
||||
"model": "TGX",
|
||||
"vin": "WDB9066351L123456",
|
||||
"first_registration": "15.03.2020",
|
||||
"mileage": 85000,
|
||||
"power_kw": 400,
|
||||
"fuel_type": "Diesel",
|
||||
}
|
||||
ocr_result.confidence_score = 0.88
|
||||
ocr_result.status = OCRStatus.completed
|
||||
await db_session.commit()
|
||||
|
||||
ocr, vehicle, updated_fields = await ocr_service.apply_to_vehicle(
|
||||
db_session, ocr_result.id
|
||||
)
|
||||
|
||||
assert vehicle.make == "MAN"
|
||||
assert vehicle.model == "TGX"
|
||||
assert vehicle.mileage_km == 85000
|
||||
assert vehicle.power_kw == 400
|
||||
assert vehicle.fuel_type == "Diesel"
|
||||
assert "make" in updated_fields
|
||||
assert "model" in updated_fields
|
||||
assert "mileage_km" in updated_fields
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_to_vehicle_no_vehicle(
|
||||
self, db_session: AsyncSession, tmp_path
|
||||
):
|
||||
"""Test applying OCR data when no vehicle is linked."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
ocr_result = await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"fake-image",
|
||||
file_name="scan.png",
|
||||
mime_type="image/png",
|
||||
)
|
||||
ocr_result.structured_data = {"brand": "MAN"}
|
||||
await db_session.commit()
|
||||
|
||||
with pytest.raises(ValueError, match="No vehicle linked"):
|
||||
await ocr_service.apply_to_vehicle(db_session, ocr_result.id)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_to_vehicle_no_data(
|
||||
self, db_session: AsyncSession, test_vehicle: Vehicle, tmp_path
|
||||
):
|
||||
"""Test applying OCR data when no structured data exists."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
ocr_result = await ocr_service.upload_file(
|
||||
db=db_session,
|
||||
file_bytes=b"fake-image",
|
||||
file_name="scan.png",
|
||||
mime_type="image/png",
|
||||
vehicle_id=test_vehicle.id,
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
with pytest.raises(ValueError, match="No structured data"):
|
||||
await ocr_service.apply_to_vehicle(db_session, ocr_result.id)
|
||||
|
||||
|
||||
class TestOCRRouter:
|
||||
"""Integration tests for OCR API endpoints."""
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def ocr_client(self, test_session_factory, admin_token):
|
||||
"""HTTP client with DB override and admin auth."""
|
||||
async def _override_get_db():
|
||||
async with test_session_factory() as session:
|
||||
try:
|
||||
yield session
|
||||
await session.commit()
|
||||
except Exception:
|
||||
await session.rollback()
|
||||
raise
|
||||
finally:
|
||||
await session.close()
|
||||
|
||||
app.dependency_overrides[get_db] = _override_get_db
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||||
ac.headers.update({"Authorization": f"Bearer {admin_token}"})
|
||||
yield ac
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_success(self, ocr_client: AsyncClient, tmp_path):
|
||||
"""POST /api/v1/ocr/upload with valid image returns 202."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
# Patch background task to avoid actual processing
|
||||
with patch("app.routers.ocr.run_ocr_processing") as mock_task:
|
||||
response = await ocr_client.post(
|
||||
"/api/v1/ocr/upload",
|
||||
files={"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")},
|
||||
)
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert "ocr_result_id" in data
|
||||
assert data["status"] == "pending"
|
||||
assert data["message"] == "OCR processing queued"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_no_file(self, ocr_client: AsyncClient):
|
||||
"""POST /api/v1/ocr/upload without file returns 422."""
|
||||
response = await ocr_client.post("/api/v1/ocr/upload")
|
||||
assert response.status_code == 422
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_invalid_mime(self, ocr_client: AsyncClient, tmp_path):
|
||||
"""POST /api/v1/ocr/upload with invalid MIME type returns 422."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
response = await ocr_client.post(
|
||||
"/api/v1/ocr/upload",
|
||||
files={"file": ("doc.pdf", io.BytesIO(b"fake-pdf"), "application/pdf")},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
data = response.json()
|
||||
assert data["detail"]["error"]["code"] == "INVALID_MIME_TYPE"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_result_success(self, ocr_client: AsyncClient, tmp_path):
|
||||
"""GET /api/v1/ocr/results/:id returns 200 with result data."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
with patch("app.routers.ocr.run_ocr_processing"):
|
||||
upload_resp = await ocr_client.post(
|
||||
"/api/v1/ocr/upload",
|
||||
files={"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")},
|
||||
)
|
||||
result_id = upload_resp.json()["ocr_result_id"]
|
||||
|
||||
response = await ocr_client.get(f"/api/v1/ocr/results/{result_id}")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == result_id
|
||||
assert data["status"] == "pending"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_result_not_found(self, ocr_client: AsyncClient):
|
||||
"""GET /api/v1/ocr/results/:nonexistent returns 404."""
|
||||
fake_id = uuid.uuid4()
|
||||
response = await ocr_client.get(f"/api/v1/ocr/results/{fake_id}")
|
||||
assert response.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_results(self, ocr_client: AsyncClient, tmp_path):
|
||||
"""GET /api/v1/ocr/results returns 200 with list."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
with patch("app.routers.ocr.run_ocr_processing"):
|
||||
for i in range(3):
|
||||
await ocr_client.post(
|
||||
"/api/v1/ocr/upload",
|
||||
files={"file": (f"scan{i}.png", io.BytesIO(b"fake-image"), "image/png")},
|
||||
)
|
||||
|
||||
response = await ocr_client.get("/api/v1/ocr/results")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["total"] >= 3
|
||||
assert len(data["items"]) >= 3
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_results_filter_vehicle(
|
||||
self, ocr_client: AsyncClient, test_vehicle: Vehicle, tmp_path
|
||||
):
|
||||
"""GET /api/v1/ocr/results?vehicle_id=X returns filtered list."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
with patch("app.routers.ocr.run_ocr_processing"):
|
||||
await ocr_client.post(
|
||||
"/api/v1/ocr/upload",
|
||||
files={"file": ("scan1.png", io.BytesIO(b"img1"), "image/png")},
|
||||
data={"vehicle_id": str(test_vehicle.id)},
|
||||
)
|
||||
await ocr_client.post(
|
||||
"/api/v1/ocr/upload",
|
||||
files={"file": ("scan2.png", io.BytesIO(b"img2"), "image/png")},
|
||||
)
|
||||
|
||||
response = await ocr_client.get(
|
||||
f"/api/v1/ocr/results?vehicle_id={test_vehicle.id}"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["total"] == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_to_vehicle_endpoint(
|
||||
self, ocr_client: AsyncClient, test_vehicle: Vehicle, tmp_path
|
||||
):
|
||||
"""POST /api/v1/ocr/results/:id/apply returns 200 and updates vehicle."""
|
||||
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
with patch("app.routers.ocr.run_ocr_processing"):
|
||||
upload_resp = await ocr_client.post(
|
||||
"/api/v1/ocr/upload",
|
||||
files={"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")},
|
||||
data={"vehicle_id": str(test_vehicle.id)},
|
||||
)
|
||||
result_id = upload_resp.json()["ocr_result_id"]
|
||||
|
||||
# Manually set structured data via direct DB session
|
||||
from tests.conftest import TEST_DATABASE_URL
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
||||
engine = create_async_engine(TEST_DATABASE_URL)
|
||||
factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
async with factory() as session:
|
||||
stmt = select(OCRResult).where(OCRResult.id == uuid.UUID(result_id))
|
||||
res = await session.execute(stmt)
|
||||
ocr = res.scalar_one()
|
||||
ocr.structured_data = {
|
||||
"brand": "Volvo",
|
||||
"model": "FH16",
|
||||
"vin": "WDB9066351L123456",
|
||||
"first_registration": "20.01.2021",
|
||||
"mileage": 200000,
|
||||
"power_kw": 500,
|
||||
"fuel_type": "Diesel",
|
||||
}
|
||||
ocr.confidence_score = 0.9
|
||||
ocr.status = OCRStatus.completed
|
||||
await session.commit()
|
||||
await engine.dispose()
|
||||
|
||||
response = await ocr_client.post(f"/api/v1/ocr/results/{result_id}/apply")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["vehicle_id"] == str(test_vehicle.id)
|
||||
assert "make" in data["updated_fields"]
|
||||
|
||||
|
||||
class TestOpenRouterClient:
|
||||
"""Tests for the OpenRouter API client utility."""
|
||||
|
||||
def test_parse_response_valid_json(self):
|
||||
"""Test parsing a valid JSON response."""
|
||||
from app.utils.openrouter import _parse_response
|
||||
|
||||
raw = '{"brand": "BMW", "model": "X5", "vin": "ABC123", "confidence_score": 0.9}'
|
||||
result = _parse_response(raw)
|
||||
assert result["structured_data"]["brand"] == "BMW"
|
||||
assert result["confidence_score"] == 0.9
|
||||
|
||||
def test_parse_response_markdown_fenced(self):
|
||||
"""Test parsing a markdown-fenced JSON response."""
|
||||
from app.utils.openrouter import _parse_response
|
||||
|
||||
raw = '```json\n{"brand": "Audi", "model": "A4", "confidence_score": 0.85}\n```'
|
||||
result = _parse_response(raw)
|
||||
assert result["structured_data"]["brand"] == "Audi"
|
||||
assert result["confidence_score"] == 0.85
|
||||
|
||||
def test_parse_response_with_text_around(self):
|
||||
"""Test parsing JSON embedded in text."""
|
||||
from app.utils.openrouter import _parse_response
|
||||
|
||||
raw = 'Here is the result: {"brand": "VW", "model": "Golf", "confidence_score": 0.7} done.'
|
||||
result = _parse_response(raw)
|
||||
assert result["structured_data"]["brand"] == "VW"
|
||||
assert result["confidence_score"] == 0.7
|
||||
|
||||
def test_parse_response_invalid(self):
|
||||
"""Test parsing an invalid response returns defaults."""
|
||||
from app.utils.openrouter import _parse_response
|
||||
|
||||
result = _parse_response("not json at all")
|
||||
assert result["structured_data"] == {}
|
||||
assert result["confidence_score"] == 0.0
|
||||
|
||||
def test_parse_response_clamps_confidence(self):
|
||||
"""Test that confidence score is clamped to 0.0-1.0."""
|
||||
from app.utils.openrouter import _parse_response
|
||||
|
||||
result = _parse_response('{"brand": "X", "confidence_score": 1.5}')
|
||||
assert result["confidence_score"] == 1.0
|
||||
|
||||
result = _parse_response('{"brand": "X", "confidence_score": -0.5}')
|
||||
assert result["confidence_score"] == 0.0
|
||||
|
||||
def test_parse_response_all_expected_fields(self):
|
||||
"""Test that all expected fields are present in structured_data."""
|
||||
from app.utils.openrouter import _parse_response, EXPECTED_FIELDS
|
||||
|
||||
raw = '{"brand": "M", "model": "A", "vin": "V", "first_registration": "01.01.2020", "mileage": 100, "power_kw": 200, "fuel_type": "D", "confidence_score": 0.8}'
|
||||
result = _parse_response(raw)
|
||||
for field in EXPECTED_FIELDS:
|
||||
assert field in result["structured_data"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_perform_ocr_no_api_key(self):
|
||||
"""Test perform_ocr raises ValueError when no API key is configured."""
|
||||
from app.utils.openrouter import perform_ocr
|
||||
|
||||
with patch("app.utils.openrouter.settings") as mock_settings:
|
||||
mock_settings.OPENROUTER_API_KEY = ""
|
||||
mock_settings.OPENROUTER_OCR_MODEL = "test-model"
|
||||
mock_settings.OPENROUTER_BASE_URL = "https://test.example.com"
|
||||
with pytest.raises(ValueError, match="OPENROUTER_API_KEY"):
|
||||
await perform_ocr(b"image", "image/png")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_perform_ocr_mocked_httpx(self):
|
||||
"""Test perform_ocr with mocked httpx client."""
|
||||
from app.utils.openrouter import perform_ocr
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
"choices": [
|
||||
{
|
||||
"message": {
|
||||
"content": '{"brand": "Test", "model": "Model", "vin": "VIN123", "confidence_score": 0.95}'
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch("app.utils.openrouter.settings") as mock_settings:
|
||||
mock_settings.OPENROUTER_API_KEY = "test-key"
|
||||
mock_settings.OPENROUTER_OCR_MODEL = "test-model"
|
||||
mock_settings.OPENROUTER_BASE_URL = "https://test.example.com"
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await perform_ocr(b"image", "image/png")
|
||||
|
||||
assert result["structured_data"]["brand"] == "Test"
|
||||
assert result["confidence_score"] == 0.95
|
||||
Reference in New Issue
Block a user