83 lines
3.4 KiB
Python
83 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.schemas.demo import DemoWorkflowResponse
|
|
from app.services.demo_workflow_service import DemoWorkflowService
|
|
|
|
|
|
def test_demo_workflow_endpoint_returns_canonical_envelope(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
area_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
candidate_dataset_id = uuid4()
|
|
quality_check_id = uuid4()
|
|
|
|
monkeypatch.setattr(
|
|
DemoWorkflowService,
|
|
"seed",
|
|
lambda _db: DemoWorkflowResponse(
|
|
project_id=project_id,
|
|
area_id=area_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
candidate_dataset_id=candidate_dataset_id,
|
|
quality_check_id=quality_check_id,
|
|
metric_count=6,
|
|
status="ready",
|
|
message="Demo workflow seeded from explicit local fixtures.",
|
|
created=True,
|
|
),
|
|
)
|
|
|
|
response = TestClient(app).post("/api/v1/demo/workflow")
|
|
|
|
assert response.status_code == 201
|
|
payload = response.json()
|
|
assert set(payload) == {"data"}
|
|
assert payload["data"]["project_id"] == str(project_id)
|
|
assert payload["data"]["reference_dataset_id"] == str(reference_dataset_id)
|
|
assert payload["data"]["candidate_dataset_id"] == str(candidate_dataset_id)
|
|
assert payload["data"]["quality_check_id"] == str(quality_check_id)
|
|
assert payload["data"]["metric_count"] == 6
|
|
assert payload["data"]["status"] == "ready"
|
|
assert payload["data"]["created"] is True
|
|
|
|
|
|
def test_demo_workflow_service_uses_explicit_golden_fixtures() -> None:
|
|
reference_path = DemoWorkflowService._fixture_path("reference_buildings.geojson")
|
|
candidate_path = DemoWorkflowService._fixture_path("predicted_buildings.geojson")
|
|
|
|
assert reference_path.exists()
|
|
assert candidate_path.exists()
|
|
assert DemoWorkflowService.PROJECT_NAME == "GeoIntel Demo - Building QA"
|
|
assert DemoWorkflowService.REFERENCE_FILENAME == "demo_reference_buildings.geojson"
|
|
assert DemoWorkflowService.CANDIDATE_FILENAME == "demo_predicted_buildings.geojson"
|
|
assert DemoWorkflowService.EXPECTED_METRICS_FILENAME == "expected_qa_metrics.json"
|
|
assert DemoWorkflowService._demo_area_geometry()["coordinates"][0][0][0][0] < 4.99
|
|
assert DemoWorkflowService._demo_area_geometry()["coordinates"][0][0][2][0] > 4.992
|
|
|
|
|
|
def test_demo_workflow_service_supports_container_fixture_mount() -> None:
|
|
service = (DemoWorkflowService._repo_root() / "backend" / "app" / "services" / "demo_workflow_service.py").read_text(encoding="utf-8")
|
|
|
|
assert "GEOINTEL_FIXTURES_ROOT" in service
|
|
assert 'Path("/app/fixtures/golden")' in service
|
|
assert "reference_payload, reference_raw = DemoWorkflowService._load_fixture" in service
|
|
assert "project = existing" in service
|
|
assert "if not reference:" in service
|
|
assert "if not candidate:" in service
|
|
assert "_sync_demo_area" in service
|
|
assert "_quality_check_matches_expected" in service
|
|
|
|
|
|
def test_demo_workflow_prefers_complete_existing_demo_project() -> None:
|
|
service = (DemoWorkflowService._repo_root() / "backend" / "app" / "services" / "demo_workflow_service.py").read_text(encoding="utf-8")
|
|
|
|
assert "_has_complete_demo_state" in service
|
|
assert "order_by(Project.created_at.asc())" in service
|
|
assert "if DemoWorkflowService._has_complete_demo_state(db, project.id):" in service
|
|
assert "return projects[0] if projects else None" in service
|