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
137 lines
5.3 KiB
Python
137 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.providers.registry import (
|
|
get_provider_dataset_mapping,
|
|
get_provider,
|
|
import_provider_dataset,
|
|
list_provider_capabilities,
|
|
)
|
|
|
|
|
|
def test_provider_registry_lists_sprint7b_providers() -> None:
|
|
providers = {provider.provider_name: provider for provider in list_provider_capabilities()}
|
|
|
|
assert set(providers) == {"grb", "osm", "manual", "fixture"}
|
|
assert providers["grb"].authority_level == "authoritative"
|
|
assert providers["grb"].configured is True
|
|
assert providers["grb"].status == "configured"
|
|
assert providers["osm"].authority_level == "contextual"
|
|
assert providers["osm"].configured is False
|
|
assert providers["osm"].status == "not_configured"
|
|
assert providers["manual"].authority_level == "manual"
|
|
assert providers["manual"].configured is True
|
|
assert providers["manual"].status == "configured"
|
|
assert providers["fixture"].authority_level == "fixture"
|
|
assert providers["fixture"].configured is True
|
|
assert providers["fixture"].status == "configured"
|
|
|
|
|
|
def test_provider_capabilities_include_required_metadata() -> None:
|
|
grb = get_provider("grb").capability.to_dict()
|
|
|
|
assert grb["provider_name"] == "grb"
|
|
assert grb["display_name"] == "GRB"
|
|
assert grb["supported_layers"] == ["buildings", "roads", "water", "parcels"]
|
|
assert grb["supported_geometry_types"] == ["Polygon", "MultiPolygon", "LineString", "MultiLineString"]
|
|
assert grb["supported_query_modes"] == ["bbox", "persisted_area"]
|
|
assert grb["limitation_message"]
|
|
assert grb["attribution"]
|
|
assert grb["license_note"]
|
|
|
|
|
|
def test_provider_to_dataset_mapping_is_enforced() -> None:
|
|
assert get_provider_dataset_mapping("grb").model_dump() == {
|
|
"provider_name": "grb",
|
|
"dataset_role": "reference",
|
|
"source_name": "grb",
|
|
"reference_required": True,
|
|
"write_path": "DatasetService",
|
|
}
|
|
assert get_provider_dataset_mapping("manual").dataset_role == "reference"
|
|
assert get_provider_dataset_mapping("fixture").source_name == "fixture"
|
|
assert get_provider_dataset_mapping("osm").dataset_role == "source"
|
|
assert get_provider_dataset_mapping("osm", requested_dataset_role="reference").dataset_role == "reference"
|
|
|
|
|
|
def test_grb_import_contract_requires_bounded_request_and_osm_remains_not_configured() -> None:
|
|
grb = import_provider_dataset("grb", project_id="project", area_id="area", layers=["buildings"])
|
|
osm = import_provider_dataset("osm", project_id="project", area_id="area", layers=["buildings"])
|
|
|
|
assert grb.status == "bounded_request_required"
|
|
assert grb.dataset_id is None
|
|
assert "bounding box" in grb.message
|
|
assert osm.status == "not_configured"
|
|
assert osm.dataset_id is None
|
|
|
|
|
|
def test_manual_fixture_import_contract_points_to_existing_flows() -> None:
|
|
manual = import_provider_dataset("manual", project_id="project", area_id=None, layers=["buildings"])
|
|
fixture = import_provider_dataset("fixture", project_id="project", area_id=None, layers=["buildings"])
|
|
|
|
assert manual.status == "upload_flow_required"
|
|
assert "upload" in manual.message.lower()
|
|
assert fixture.status == "fixture_flow_required"
|
|
assert "fixture" in fixture.message.lower()
|
|
|
|
|
|
def test_provider_api_envelopes_and_invalid_provider() -> None:
|
|
client = TestClient(app)
|
|
|
|
list_response = client.get("/api/v1/external/providers")
|
|
assert list_response.status_code == 200
|
|
assert {provider["provider_name"] for provider in list_response.json()["data"]["providers"]} == {
|
|
"grb",
|
|
"osm",
|
|
"manual",
|
|
"fixture",
|
|
}
|
|
|
|
detail_response = client.get("/api/v1/external/providers/grb")
|
|
assert detail_response.status_code == 200
|
|
assert detail_response.json()["data"]["provider_name"] == "grb"
|
|
|
|
layers_response = client.get("/api/v1/external/providers/osm/layers")
|
|
assert layers_response.status_code == 200
|
|
assert layers_response.json()["data"]["layers"] == ["buildings", "roads", "water", "landuse"]
|
|
|
|
status_response = client.get("/api/v1/external/providers/manual/status")
|
|
assert status_response.status_code == 200
|
|
assert status_response.json()["data"]["configured"] is True
|
|
|
|
invalid_response = client.get("/api/v1/external/providers/unknown")
|
|
assert invalid_response.status_code == 404
|
|
assert invalid_response.json()["error"] == "PROVIDER_NOT_FOUND"
|
|
assert invalid_response.json()["message"] == "Provider not found"
|
|
|
|
|
|
def test_provider_import_api_points_grb_to_governed_bounded_endpoint() -> None:
|
|
client = TestClient(app)
|
|
|
|
response = client.post(
|
|
"/api/v1/external/providers/grb/import",
|
|
json={
|
|
"project_id": "project",
|
|
"area_id": "area",
|
|
"layers": ["buildings"],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["provider_name"] == "grb"
|
|
assert response.json()["data"]["status"] == "bounded_request_required"
|
|
assert response.json()["data"]["dataset_id"] is None
|
|
|
|
|
|
def test_live_migration_smoke_script_exists() -> None:
|
|
from pathlib import Path
|
|
|
|
script = Path(__file__).parents[2] / "scripts" / "live_migration_smoke.sh"
|
|
text = script.read_text(encoding="utf-8")
|
|
|
|
assert "alembic upgrade head" in text
|
|
assert "SELECT PostGIS_Version()" in text
|
|
assert "alembic heads" in text
|