Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
122 lines
3.3 KiB
Python
122 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.models import Metric, QualityCheck
|
|
from app.schemas.qa import QualityCheckRead
|
|
from app.services.quality_check_service import QualityCheckService
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, rows):
|
|
self.rows = rows
|
|
|
|
def filter(self, *_args):
|
|
return self
|
|
|
|
def order_by(self, *_args):
|
|
return self
|
|
|
|
def count(self):
|
|
return len(self.rows)
|
|
|
|
def offset(self, _offset):
|
|
return self
|
|
|
|
def limit(self, _limit):
|
|
return self
|
|
|
|
def all(self):
|
|
return self.rows
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, quality_checks, metrics):
|
|
self.quality_checks = quality_checks
|
|
self.metrics = metrics
|
|
|
|
def query(self, model):
|
|
if model is QualityCheck:
|
|
return FakeQuery(self.quality_checks)
|
|
if model is Metric:
|
|
return FakeQuery(self.metrics)
|
|
return FakeQuery([])
|
|
|
|
|
|
def test_quality_check_service_lists_checks_with_metrics() -> None:
|
|
project_id = uuid4()
|
|
quality_check_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
candidate_dataset_id = uuid4()
|
|
created_at = datetime.now(timezone.utc)
|
|
quality_check = QualityCheck(
|
|
id=quality_check_id,
|
|
project_id=project_id,
|
|
candidate_dataset_id=candidate_dataset_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
check_type="demo_candidate_vs_reference",
|
|
status="ok",
|
|
score=0.5,
|
|
parameters_json={"iou_threshold": 0.5},
|
|
findings_json={"matches": 1},
|
|
created_at=created_at,
|
|
completed_at=created_at,
|
|
)
|
|
metric = Metric(
|
|
id=uuid4(),
|
|
quality_check_id=quality_check_id,
|
|
metric_key="precision",
|
|
metric_value=0.5,
|
|
metadata_json={},
|
|
created_at=created_at,
|
|
)
|
|
|
|
items, total = QualityCheckService.list_quality_checks(
|
|
FakeSession([quality_check], [metric]),
|
|
project_id=project_id,
|
|
)
|
|
|
|
assert total == 1
|
|
assert len(items) == 1
|
|
assert items[0].id == quality_check_id
|
|
assert items[0].metrics[0].metric_key == "precision"
|
|
assert items[0].metrics[0].metric_value == 0.5
|
|
|
|
|
|
def test_quality_checks_endpoint_returns_canonical_envelope(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
quality_check_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
|
|
monkeypatch.setattr(
|
|
QualityCheckService,
|
|
"list_quality_checks",
|
|
lambda *_args, **_kwargs: (
|
|
[
|
|
QualityCheckRead(
|
|
id=quality_check_id,
|
|
project_id=project_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
check_type="demo_candidate_vs_reference",
|
|
status="ok",
|
|
score=0.5,
|
|
metrics=[],
|
|
)
|
|
],
|
|
1,
|
|
),
|
|
)
|
|
|
|
response = TestClient(app).get(f"/api/v1/projects/{project_id}/quality-checks")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert set(payload) == {"data"}
|
|
assert payload["data"]["total"] == 1
|
|
assert payload["data"]["items"][0]["id"] == str(quality_check_id)
|
|
assert payload["data"]["items"][0]["check_type"] == "demo_candidate_vs_reference"
|