667 lines
30 KiB
Python
667 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
import app.jobs as jobs_module
|
|
import app.main as main_module
|
|
from app.config import settings
|
|
from app.db import init_db, session_scope
|
|
from app.db_lock import DatabaseWriteBusy, database_write_lock
|
|
from app.jobs import run_worker_once
|
|
from app.main import app
|
|
from app.models import Dataset, GtfsRoute, Job, Source
|
|
from app.source_catalog import import_ingestable_sources
|
|
|
|
|
|
def test_api_sample_and_geojson():
|
|
client = TestClient(app)
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert "Mobility Workbench" in response.text
|
|
assert "GTFS Harmonization" in response.text
|
|
assert "Mapping Data" in response.text
|
|
assert "journeyTransitSnapshot" in response.text
|
|
assert "journeySource" not in response.text
|
|
|
|
response = client.post("/api/sample/reset")
|
|
assert response.status_code == 200
|
|
stats = client.get("/api/stats").json()
|
|
assert stats["gtfs_routes"] == 6
|
|
assert stats["osm_routes"] == 6
|
|
geojson = client.get("/api/map/gtfs_routes.geojson").json()
|
|
assert geojson["type"] == "FeatureCollection"
|
|
assert len(geojson["features"]) == 6
|
|
matched_geojson = client.get("/api/map/matched_gtfs_routes.geojson?status=matched").json()
|
|
assert matched_geojson["features"]
|
|
assert {feature["properties"]["visual_source"] for feature in matched_geojson["features"]} == {"osm"}
|
|
filtered = client.get("/api/map/osm_features.geojson?kind=route&mode=tram&bbox=13.3,52.4,13.5,52.6").json()
|
|
assert filtered["type"] == "FeatureCollection"
|
|
assert {feature["properties"]["ref"] for feature in filtered["features"]} == {"M5", "M10"}
|
|
source_filtered_gtfs = client.get("/api/map/gtfs_routes.geojson?source_id=1").json()
|
|
assert len(source_filtered_gtfs["features"]) == 6
|
|
source_filtered_osm = client.get("/api/map/osm_features.geojson?source_id=2&kind=route&mode=tram").json()
|
|
assert {feature["properties"]["ref"] for feature in source_filtered_osm["features"]} == {"M5", "M10"}
|
|
route_layer = client.post("/api/route-layer/build").json()
|
|
assert route_layer["route_patterns"] > 0
|
|
assert client.get("/api/stats").json()["route_patterns"] == route_layer["route_patterns"]
|
|
regional_osm = client.get("/api/map/osm_features.geojson?kind=route&mode=train&route_scope=regional").json()
|
|
assert {feature["properties"]["ref"] for feature in regional_osm["features"]} == {"RE1"}
|
|
regional_patterns = client.get("/api/map/route_patterns.geojson?mode=train&source_kind=osm&route_scope=regional").json()
|
|
assert {feature["properties"]["ref"] for feature in regional_patterns["features"]} == {"RE1"}
|
|
local_patterns = client.get("/api/map/route_patterns.geojson?mode=subway&source_kind=osm&route_scope=local").json()
|
|
assert {feature["properties"]["ref"] for feature in local_patterns["features"]} == {"U2"}
|
|
local_bus_patterns = client.get("/api/map/route_patterns.geojson?mode=bus&source_kind=osm&route_scope=local").json()
|
|
assert {feature["properties"]["ref"] for feature in local_bus_patterns["features"]} == {"100"}
|
|
|
|
|
|
def test_journey_demo_direct_and_one_transfer():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
hbf = _first_stop(client, "Hauptbahnhof")
|
|
alex = _first_stop(client, "Alexanderplatz")
|
|
direct = client.get(f"/api/journey/search?from_stop_id={hbf['id']}&to_stop_id={alex['id']}&departure=08:00&max_transfers=0").json()
|
|
assert direct["journeys"]
|
|
assert direct["journeys"][0]["transfers"] == 0
|
|
assert direct["journeys"][0]["legs"][0]["route_ref"] in {"RE1", "M5"}
|
|
coords = direct["journeys"][0]["features"]["features"][0]["geometry"]["coordinates"]
|
|
assert coords[-1] == [13.4132, 52.5219]
|
|
assert [13.4344, 52.51] not in coords
|
|
stop_roles = {
|
|
feature["properties"]["role"]
|
|
for feature in direct["journeys"][0]["features"]["features"]
|
|
if feature["geometry"]["type"] == "Point"
|
|
}
|
|
assert {"start", "end", "passed"} <= stop_roles
|
|
|
|
zoo = _first_stop(client, "Zoologischer")
|
|
ost = _first_stop(client, "Ostbahnhof")
|
|
transfer = client.get(
|
|
f"/api/journey/search?from_stop_id={zoo['id']}&to_stop_id={ost['id']}&departure=08:00&max_transfers=1&transfer_seconds=0"
|
|
).json()
|
|
assert transfer["journeys"]
|
|
assert transfer["journeys"][0]["transfers"] == 1
|
|
assert [leg["route_ref"] for leg in transfer["journeys"][0]["legs"]] == ["100", "RE1"]
|
|
|
|
|
|
def test_route_layer_job_endpoint_completes():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
queued = client.post("/api/jobs/route-layer-build").json()
|
|
assert queued["kind"] == "route_layer_rebuild"
|
|
assert queued["status"] == "queued"
|
|
assert queued["priority"] == 0
|
|
|
|
worker = run_worker_once(worker_id="test-worker")
|
|
assert worker["processed"] == 1
|
|
job = client.get(f"/api/jobs/{queued['id']}").json()
|
|
|
|
assert job["status"] == "completed"
|
|
assert job["result"]["route_patterns"] > 0
|
|
events = client.get(f"/api/jobs/{queued['id']}/events").json()
|
|
assert [event["event_type"] for event in events["events"]][-1] == "completed"
|
|
|
|
|
|
def test_route_matching_job_endpoint_completes():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
queued = client.post("/api/jobs/match-run").json()
|
|
assert queued["kind"] == "route_matching"
|
|
assert queued["status"] == "queued"
|
|
|
|
worker = run_worker_once(worker_id="test-worker")
|
|
assert worker["processed"] == 1
|
|
job = client.get(f"/api/jobs/{queued['id']}").json()
|
|
|
|
assert job["status"] == "completed"
|
|
assert job["result"]["routes"] == 6
|
|
assert job["result"]["matches"] > 0
|
|
events = client.get(f"/api/jobs/{queued['id']}/events").json()
|
|
event_types = [event["event_type"] for event in events["events"]]
|
|
assert "route_matching_batch" in event_types
|
|
assert event_types[-1] == "completed"
|
|
|
|
|
|
def test_qa_summary_endpoint_exposes_harmonization_sections():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
summary = client.get("/api/qa/summary").json()
|
|
|
|
assert summary["decision"]["deployment"] == "same_workbench_for_now"
|
|
section_ids = {section["id"] for section in summary["sections"]}
|
|
assert {
|
|
"source_discovery",
|
|
"import_health",
|
|
"gtfs_validation",
|
|
"deduplication",
|
|
"route_quality",
|
|
"publication_readiness",
|
|
} <= section_ids
|
|
gtfs_section = next(section for section in summary["sections"] if section["id"] == "gtfs_validation")
|
|
assert any(item["label"] == "Routes" for item in gtfs_section["items"])
|
|
|
|
|
|
def test_gtfs_harmonization_inventory_and_detail():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
inventory = client.get("/api/harmonization/gtfs/inventory").json()
|
|
assert inventory["summary"]["sources"] == 1
|
|
assert inventory["summary"]["active_sources"] == 1
|
|
feed = inventory["feeds"][0]
|
|
assert feed["source"]["name"] == "Sample Berlin GTFS"
|
|
assert feed["active_dataset"]["counts"]["routes"] == 6
|
|
assert feed["validation"]["items"]
|
|
assert feed["service"]["items"]
|
|
|
|
detail = client.get(f"/api/harmonization/gtfs/sources/{feed['source']['id']}").json()
|
|
assert detail["source"]["id"] == feed["source"]["id"]
|
|
assert {section["id"] for section in detail["sections"]} == {"validation", "service", "overlap", "license"}
|
|
assert all({"id", "severity", "title", "detail"} <= set(issue) for issue in detail["issues"])
|
|
assert detail["qa_status"] in {"ready", "needs_review", "blocked"}
|
|
|
|
reviewed = client.patch(
|
|
f"/api/harmonization/gtfs/sources/{feed['source']['id']}/review",
|
|
json={"license": "CC-BY-4.0", "review_status": "approved", "review_note": "Operator publication allowed.", "enabled": True},
|
|
).json()
|
|
assert reviewed["source"]["license"] == "CC-BY-4.0"
|
|
assert reviewed["source"]["qa_review"]["status"] == "approved"
|
|
assert reviewed["source"]["qa_review"]["note"] == "Operator publication allowed."
|
|
assert reviewed["source"]["enabled"] is True
|
|
|
|
|
|
def test_terminal_jobs_can_be_dismissed_from_default_view():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
queued = client.post("/api/jobs/route-layer-build").json()
|
|
assert run_worker_once(worker_id="test-worker")["processed"] == 1
|
|
|
|
listed = client.get("/api/jobs").json()
|
|
assert any(job["id"] == queued["id"] for job in listed["jobs"])
|
|
|
|
dismissed = client.post(f"/api/jobs/{queued['id']}/dismiss").json()
|
|
assert dismissed["dismissed_at"]
|
|
|
|
hidden = client.get("/api/jobs").json()
|
|
assert all(job["id"] != queued["id"] for job in hidden["jobs"])
|
|
|
|
visible = client.get("/api/jobs?include_dismissed=true").json()
|
|
assert any(job["id"] == queued["id"] for job in visible["jobs"])
|
|
|
|
|
|
def test_jobs_revision_endpoint_reports_changes():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
initial = client.get("/api/jobs/revision").json()
|
|
assert initial["changed"] is True
|
|
assert initial["revision"]
|
|
assert initial["job_revision"]
|
|
assert "workers" in initial
|
|
|
|
queued = client.post("/api/jobs/route-layer-build").json()
|
|
changed = client.get("/api/jobs/revision", params={"since": initial["revision"]}).json()
|
|
assert changed["changed"] is True
|
|
assert changed["latest_job_id"] >= queued["id"]
|
|
assert changed["job_count"] >= 1
|
|
|
|
unchanged = client.get("/api/jobs/revision", params={"since": changed["revision"]}).json()
|
|
assert unchanged["changed"] is False
|
|
|
|
listed = client.get("/api/jobs").json()
|
|
assert listed["revision"] == unchanged["revision"]
|
|
assert listed["jobs"]
|
|
|
|
|
|
def test_nearest_location_skips_address_lookup_while_address_index_rebuilds(monkeypatch):
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
with session_scope() as session:
|
|
session.add(Job(kind="address_index_rebuild", status="running", description="test address rebuild"))
|
|
session.commit()
|
|
|
|
def fail_address_lookup(**_kwargs):
|
|
raise AssertionError("address lookup should be skipped while address index rebuilds")
|
|
|
|
monkeypatch.setattr(main_module, "address_at_point", fail_address_lookup)
|
|
response = client.get("/api/journey/nearest-location?lat=0&lon=0")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["selection_kind"] == "coordinate"
|
|
assert data["address_lookup_skipped"] is True
|
|
assert "Address index rebuild" in data["message"]
|
|
|
|
|
|
def test_job_queue_controls_for_queued_job():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
queued = client.post("/api/jobs/route-layer-build?priority=5").json()
|
|
assert queued["status"] == "queued"
|
|
assert queued["priority"] == 5
|
|
|
|
priority = client.post(f"/api/jobs/{queued['id']}/priority", json={"priority": 20}).json()
|
|
assert priority["priority"] == 20
|
|
|
|
paused = client.post(f"/api/jobs/{queued['id']}/pause").json()
|
|
assert paused["status"] == "paused"
|
|
|
|
idle_worker = run_worker_once(worker_id="test-worker")
|
|
assert idle_worker["processed"] == 0
|
|
|
|
resumed = client.post(f"/api/jobs/{queued['id']}/resume").json()
|
|
assert resumed["status"] == "queued"
|
|
|
|
stopped = client.post(f"/api/jobs/{queued['id']}/stop").json()
|
|
assert stopped["status"] == "cancelled"
|
|
|
|
retried = client.post(f"/api/jobs/{queued['id']}/retry").json()
|
|
assert retried["status"] == "queued"
|
|
assert retried["error"] is None
|
|
|
|
|
|
def test_worker_once_returns_idle_when_claim_is_busy(monkeypatch):
|
|
def busy_claim(_worker_id):
|
|
raise DatabaseWriteBusy("job:claim", {"operation": "update source"})
|
|
|
|
monkeypatch.setattr(jobs_module, "claim_next_job", busy_claim)
|
|
|
|
assert jobs_module.run_worker_once(worker_id="test-worker") == {"worker_id": "test-worker", "processed": 0}
|
|
|
|
|
|
def test_running_job_can_be_stopped_while_write_lock_is_held():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
queued = client.post("/api/jobs/route-layer-build").json()
|
|
|
|
with session_scope() as session:
|
|
job = session.get(Job, queued["id"])
|
|
job.status = "running"
|
|
job.lease_owner = "test-worker"
|
|
|
|
with database_write_lock("job:route_layer_rebuild:test"):
|
|
response = client.post(f"/api/jobs/{queued['id']}/stop")
|
|
|
|
assert response.status_code == 200
|
|
stopped = response.json()
|
|
assert stopped["id"] == queued["id"]
|
|
assert stopped["requested_action"] == "cancel"
|
|
|
|
|
|
def test_itinerary_generation_and_leg_locking():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
hbf = _first_stop(client, "Hauptbahnhof")
|
|
alex = _first_stop(client, "Alexanderplatz")
|
|
|
|
generated = client.post(
|
|
"/api/itineraries/generate",
|
|
json={
|
|
"from_stop_id": hbf["id"],
|
|
"to_stop_id": alex["id"],
|
|
"departure": "08:00",
|
|
"service_date": "2026-06-27",
|
|
"max_transfers": 1,
|
|
"transfer_seconds": 120,
|
|
"limit": 2,
|
|
},
|
|
).json()
|
|
|
|
assert generated["request"]["service_date"] == "2026-06-27"
|
|
assert any(item["family"] == "public_transport" for item in generated["itineraries"])
|
|
assert any(item["family"] == "flight_access" for item in generated["itineraries"])
|
|
public = next(item for item in generated["itineraries"] if item["family"] == "public_transport")
|
|
saved = client.post(f"/api/itineraries/{public['id']}/save", json={"saved": True}).json()
|
|
assert saved["saved"] is True
|
|
leg_id = saved["legs"][0]["id"]
|
|
locked = client.post(f"/api/itinerary-legs/{leg_id}/lock", json={"locked": True}).json()
|
|
assert locked["locked"] is True
|
|
recent = client.get("/api/itineraries?saved_only=true").json()
|
|
assert any(item["id"] == public["id"] for item in recent["itineraries"])
|
|
|
|
|
|
def test_geofabrik_catalog_source_creation(monkeypatch):
|
|
from app import main
|
|
from app.geofabrik import create_geofabrik_source
|
|
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
fake_entry = {
|
|
"id": "berlin",
|
|
"name": "Berlin",
|
|
"parent": "germany",
|
|
"country_codes": ["DE"],
|
|
"pbf_url": "https://download.geofabrik.de/europe/germany/berlin-latest.osm.pbf",
|
|
"updates_url": "https://download.geofabrik.de/europe/germany/berlin-updates",
|
|
"taginfo_url": "https://taginfo.geofabrik.de/europe:germany:berlin",
|
|
"urls": {},
|
|
}
|
|
|
|
monkeypatch.setattr(main, "geofabrik_catalog", lambda q=None, limit=80: [fake_entry])
|
|
monkeypatch.setattr("app.geofabrik.geofabrik_entry", lambda geofabrik_id: fake_entry if geofabrik_id == "berlin" else None)
|
|
|
|
catalog = client.get("/api/geofabrik/catalog?q=berlin").json()
|
|
assert catalog["entries"][0]["id"] == "berlin"
|
|
created = client.post(
|
|
"/api/geofabrik/sources",
|
|
json={"geofabrik_id": "berlin", "import_updates": True, "run_import": False},
|
|
).json()
|
|
assert created["source"]["kind"] == "osm_pbf"
|
|
assert "berlin-latest.osm.pbf" in created["source"]["url"]
|
|
|
|
|
|
def test_source_management_and_match_candidates():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
stats = client.get("/api/stats").json()
|
|
assert stats["match_summary"]["missing"] + stats["match_summary"]["weak"] >= 1
|
|
sources = client.get("/api/sources").json()
|
|
gtfs_source = next(source for source in sources if source["kind"] == "gtfs")
|
|
assert gtfs_source["stats"]["routes"] == 6
|
|
assert gtfs_source["datasets"][0]["stats"]["stop_times"] == 20
|
|
|
|
match = client.get("/api/matches?limit=1").json()[0]
|
|
candidates = client.get(f"/api/matches/{match['id']}/candidates").json()
|
|
assert candidates["route"]["id"] == match["gtfs"]["id"]
|
|
assert candidates["route"]["geometry"]["present"] is True
|
|
assert candidates["candidates"]
|
|
assert "score" in candidates["candidates"][0]
|
|
assert candidates["candidates"][0]["osm"]["geometry"]["present"] is True
|
|
assert candidates["preview"]["type"] == "FeatureCollection"
|
|
preview_roles = {feature["properties"]["preview_role"] for feature in candidates["preview"]["features"]}
|
|
assert {"gtfs_route", "candidate"} <= preview_roles
|
|
candidate_preview = next(feature for feature in candidates["preview"]["features"] if feature["properties"]["preview_role"] == "candidate")
|
|
assert "candidate_score" in candidate_preview["properties"]
|
|
picked = candidates["candidates"][0]
|
|
accepted = client.post(f"/api/matches/{match['id']}/candidates/{picked['osm']['id']}/accept").json()
|
|
assert accepted["status"] == "accepted"
|
|
assert accepted["match"]["osm"]["osm_type"] == picked["osm"]["osm_type"]
|
|
assert accepted["match"]["osm"]["osm_id"] == picked["osm"]["osm_id"]
|
|
|
|
search = client.get("/api/datasets/search?q=M5&active_only=true").json()
|
|
assert search["gtfs_routes"]
|
|
assert search["osm_routes"]
|
|
m5_route = next(item for item in search["gtfs_routes"] if item["route"]["ref"] == "M5")
|
|
assert m5_route["timetable"]["stop_times"] > 0
|
|
assert m5_route["geometry"]["present"] is True
|
|
feature = client.get(f"/api/datasets/search/feature.geojson?type=gtfs_route&id={m5_route['route']['id']}").json()
|
|
assert feature["features"]
|
|
assert feature["features"][0]["properties"]["search_result_type"] == "gtfs_route"
|
|
|
|
update_check = client.post(f"/api/sources/{gtfs_source['id']}/check-update").json()
|
|
assert update_check["status"] == "checked"
|
|
assert update_check["update_available"] is False
|
|
update_result = client.post(f"/api/sources/{gtfs_source['id']}/update").json()
|
|
assert update_result["status"] == "skipped"
|
|
history = client.get(f"/api/sources/{gtfs_source['id']}/update-checks").json()
|
|
assert history["checks"]
|
|
|
|
response = client.delete(f"/api/sources/{gtfs_source['id']}")
|
|
assert response.status_code == 200
|
|
delete_job = response.json()
|
|
assert delete_job["kind"] == "source_delete"
|
|
assert delete_job["status"] == "queued"
|
|
duplicate = client.delete(f"/api/sources/{gtfs_source['id']}").json()
|
|
assert duplicate["id"] == delete_job["id"]
|
|
|
|
worker = run_worker_once(worker_id="test-worker")
|
|
assert worker["processed"] == 1
|
|
completed = client.get(f"/api/jobs/{delete_job['id']}").json()
|
|
assert completed["status"] == "completed"
|
|
assert completed["result"]["delete_result"]["deleted"] is True
|
|
stats_after_delete = client.get("/api/stats").json()
|
|
assert stats_after_delete["gtfs_routes"] == 0
|
|
assert stats_after_delete["osm_routes"] == 6
|
|
|
|
osm_source = next(source for source in client.get("/api/sources").json() if source["kind"] == "osm_geojson")
|
|
dataset_id = osm_source["datasets"][0]["id"]
|
|
dataset_delete_job = client.delete(f"/api/datasets/{dataset_id}").json()
|
|
assert dataset_delete_job["kind"] == "dataset_delete"
|
|
assert dataset_delete_job["status"] == "queued"
|
|
queued_source = next(source for source in client.get("/api/sources").json() if source["id"] == osm_source["id"])
|
|
assert queued_source["datasets"][0]["active_job"]["id"] == dataset_delete_job["id"]
|
|
assert queued_source["active_job"]["id"] == dataset_delete_job["id"]
|
|
|
|
assert run_worker_once(worker_id="test-worker")["processed"] == 1
|
|
completed_dataset_delete = client.get(f"/api/jobs/{dataset_delete_job['id']}").json()
|
|
assert completed_dataset_delete["status"] == "completed"
|
|
assert completed_dataset_delete["result"]["delete_result"]["deleted"] is True
|
|
assert client.get("/api/stats").json()["osm_routes"] == 0
|
|
|
|
|
|
def test_missing_gtfs_sidecar_queues_recovery_without_breaking_sources():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
with session_scope() as session:
|
|
dataset = session.scalar(select(Dataset).where(Dataset.kind == "gtfs", Dataset.is_active.is_(True)))
|
|
assert dataset is not None
|
|
source_id = dataset.source_id
|
|
metadata = json.loads(dataset.metadata_json or "{}")
|
|
metadata["gtfs_storage"]["sidecar_path"] = str(settings.data_dir / "sidecars" / f"missing_gtfs_dataset_{dataset.id}.sqlite")
|
|
dataset.metadata_json = json.dumps(metadata)
|
|
dataset_id = dataset.id
|
|
|
|
response = client.get("/api/sources")
|
|
|
|
assert response.status_code == 200
|
|
source = next(item for item in response.json() if item["id"] == source_id)
|
|
assert source["active_job"]["kind"] == "source_import"
|
|
assert "GTFS sidecar missing" in source["active_job"]["result"]["recovery_reason"]
|
|
recovered_dataset = next(item for item in source["datasets"] if item["id"] == dataset_id)
|
|
assert recovered_dataset["status"] == "missing_files"
|
|
assert recovered_dataset["stats"]["missing_sidecar"] is True
|
|
assert recovered_dataset["stats"]["stop_times"] == 0
|
|
|
|
second_response = client.get("/api/sources")
|
|
assert second_response.status_code == 200
|
|
with session_scope() as session:
|
|
recovery_jobs = session.scalars(select(Job).where(Job.kind == "source_import", Job.status == "queued")).all()
|
|
assert len(recovery_jobs) == 1
|
|
|
|
|
|
def test_admin_maintenance_endpoints_are_guarded_and_callable():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
init_job = client.post("/api/admin/init-db").json()
|
|
assert init_job["kind"] == "maintenance"
|
|
assert init_job["result"]["action"] == "init-db"
|
|
assert run_worker_once(worker_id="test-worker")["processed"] == 1
|
|
init_completed = client.get(f"/api/jobs/{init_job['id']}").json()
|
|
assert init_completed["status"] == "completed"
|
|
assert init_completed["result"]["result"]["status"] == "initialized"
|
|
|
|
backfill_job = client.post("/api/admin/backfill-gtfs-shapes", json={}).json()
|
|
assert backfill_job["kind"] == "maintenance"
|
|
assert run_worker_once(worker_id="test-worker")["processed"] == 1
|
|
backfill = client.get(f"/api/jobs/{backfill_job['id']}").json()
|
|
assert "datasets" in backfill["result"]["result"]
|
|
|
|
prune_cache_job = client.post("/api/admin/prune-cache", json={}).json()
|
|
assert prune_cache_job["kind"] == "maintenance"
|
|
assert run_worker_once(worker_id="test-worker")["processed"] == 1
|
|
prune_cache = client.get(f"/api/jobs/{prune_cache_job['id']}").json()["result"]["result"]
|
|
assert prune_cache["dry_run"] is True
|
|
assert "files" in prune_cache
|
|
assert "bytes" in prune_cache
|
|
|
|
prune_inactive_job = client.post("/api/admin/prune-inactive-datasets", json={}).json()
|
|
assert prune_inactive_job["kind"] == "maintenance"
|
|
assert run_worker_once(worker_id="test-worker")["processed"] == 1
|
|
prune_inactive = client.get(f"/api/jobs/{prune_inactive_job['id']}").json()["result"]["result"]
|
|
assert prune_inactive["dry_run"] is True
|
|
assert "would_delete" in prune_inactive
|
|
|
|
sample_job = client.post("/api/jobs/sample-reset").json()
|
|
assert sample_job["kind"] == "maintenance"
|
|
assert sample_job["result"]["action"] == "sample-reset"
|
|
assert run_worker_once(worker_id="test-worker")["processed"] == 1
|
|
sample_completed = client.get(f"/api/jobs/{sample_job['id']}").json()
|
|
assert sample_completed["status"] == "completed"
|
|
assert sample_completed["result"]["result"]["status"] == "ok"
|
|
assert client.get("/api/stats").json()["gtfs_routes"] == 6
|
|
|
|
assert client.post("/api/admin/prune-cache", json={"dry_run": False}).status_code == 400
|
|
assert client.post("/api/admin/prune-inactive-datasets", json={"dry_run": False}).status_code == 400
|
|
assert client.post("/api/admin/vacuum-db", json={}).status_code == 400
|
|
assert client.post("/api/admin/reset-db", json={}).status_code == 400
|
|
|
|
|
|
def test_source_catalog_import_and_ingestable_seed_metadata():
|
|
init_db()
|
|
client = TestClient(app)
|
|
|
|
catalog_import = client.post("/api/source-catalog/import").json()
|
|
assert catalog_import["summary"]["catalog_entries"] >= 50
|
|
|
|
catalog = client.get("/api/source-catalog?country=DE&priority=P0&limit=10").json()
|
|
assert catalog["entries"]
|
|
assert any("DELFI" in entry["source_name"] for entry in catalog["entries"])
|
|
assert "geometry_notes" in catalog["entries"][0]
|
|
|
|
osm_catalog = client.get("/api/source-catalog?q=Geofabrik&limit=5").json()
|
|
osm_entry = next(entry for entry in osm_catalog["entries"] if "Geofabrik" in entry["source_name"])
|
|
created_source = client.post(
|
|
"/api/sources",
|
|
json={
|
|
"catalog_entry_id": osm_entry["id"],
|
|
"name": "Berlin Geofabrik OSM PBF",
|
|
"kind": "osm_pbf",
|
|
"url": "https://download.geofabrik.de/europe/germany/berlin-latest.osm.pbf",
|
|
"country": "DE",
|
|
},
|
|
).json()
|
|
sources = client.get("/api/sources").json()
|
|
linked_source = next(source for source in sources if source["id"] == created_source["id"])
|
|
assert linked_source["catalog_entry_id"] == osm_entry["id"]
|
|
assert linked_source["priority"] == osm_entry["priority"]
|
|
linked_catalog = client.get("/api/source-catalog?q=Geofabrik&limit=5").json()
|
|
linked_entry = next(entry for entry in linked_catalog["entries"] if entry["id"] == osm_entry["id"])
|
|
assert linked_entry["linked_source_count"] == 1
|
|
|
|
seed_import = client.post("/api/source-catalog/import-ingestable").json()
|
|
assert seed_import["created"] + seed_import["updated"] >= 10
|
|
|
|
sources = client.get("/api/sources").json()
|
|
swiss = next(source for source in sources if source["name"] == "CH Swiss national GTFS")
|
|
assert swiss["kind"] == "gtfs"
|
|
assert swiss["priority"] == "P0"
|
|
assert "rail" in swiss["mode_scope"]
|
|
assert swiss["notes"]
|
|
vbb = next(source for source in sources if source["name"] == "VBB Berlin-Brandenburg GTFS")
|
|
vbb_catalog = next(entry for entry in client.get("/api/source-catalog?q=VBB&limit=5").json()["entries"] if entry["source_name"] == "VBB Berlin-Brandenburg GTFS")
|
|
assert vbb["kind"] == "gtfs"
|
|
assert vbb["priority"] == "P5"
|
|
assert vbb["catalog_entry_id"] == vbb_catalog["id"]
|
|
|
|
|
|
def test_ingestable_source_import_deduplicates_by_kind_and_url(tmp_path):
|
|
init_db()
|
|
first = tmp_path / "first.csv"
|
|
first.write_text(
|
|
"name,kind,url,country,license,mode_scope,source_basis,priority,notes\n"
|
|
"Original GTFS,gtfs,https://example.test/feed.zip,DE,CC0,bus,test,P1,first\n",
|
|
encoding="utf-8",
|
|
)
|
|
second = tmp_path / "second.csv"
|
|
second.write_text(
|
|
"name,kind,url,country,license,mode_scope,source_basis,priority,notes\n"
|
|
"Renamed GTFS,gtfs,https://example.test/feed.zip,DE,CC0,bus,test,P0,second\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with session_scope() as session:
|
|
assert import_ingestable_sources(session, first)["created"] == 1
|
|
with session_scope() as session:
|
|
result = import_ingestable_sources(session, second)
|
|
assert result["created"] == 0
|
|
assert result["updated"] == 1
|
|
sources = session.scalars(select(Source).where(Source.url == "https://example.test/feed.zip")).all()
|
|
assert len(sources) == 1
|
|
assert sources[0].name == "Renamed GTFS"
|
|
assert sources[0].priority == "P0"
|
|
|
|
|
|
def test_write_endpoint_returns_busy_when_another_write_is_active():
|
|
init_db()
|
|
client = TestClient(app)
|
|
previous_timeout = settings.database_write_lock_timeout_seconds
|
|
settings.database_write_lock_timeout_seconds = 0.05
|
|
try:
|
|
with database_write_lock("test long write", timeout=0.1):
|
|
response = client.post(
|
|
"/api/sources",
|
|
json={"name": "Busy test source", "kind": "gtfs", "url": "https://example.invalid/feed.zip"},
|
|
)
|
|
finally:
|
|
settings.database_write_lock_timeout_seconds = previous_timeout
|
|
|
|
assert response.status_code == 409
|
|
assert "Database is busy" in response.json()["detail"]
|
|
|
|
|
|
def test_manual_match_rule_survives_new_gtfs_dataset_row():
|
|
client = TestClient(app)
|
|
assert client.post("/api/sample/reset").status_code == 200
|
|
|
|
match = next(item for item in client.get("/api/matches?status=matched").json() if item["osm"])
|
|
accepted = client.post(f"/api/matches/{match['id']}/accept").json()
|
|
assert accepted["status"] == "accepted"
|
|
|
|
with session_scope() as session:
|
|
old_route = session.get(GtfsRoute, match["gtfs"]["id"])
|
|
assert old_route is not None
|
|
old_dataset = session.get(Dataset, old_route.dataset_id)
|
|
assert old_dataset is not None
|
|
old_dataset.is_active = False
|
|
replacement_dataset = Dataset(
|
|
source_id=old_dataset.source_id,
|
|
kind="gtfs",
|
|
local_path="./data/replacement.gtfs.zip",
|
|
sha256="replacement",
|
|
is_active=True,
|
|
status="imported",
|
|
)
|
|
session.add(replacement_dataset)
|
|
session.flush()
|
|
session.add(
|
|
GtfsRoute(
|
|
dataset_id=replacement_dataset.id,
|
|
route_id=old_route.route_id,
|
|
agency_id=old_route.agency_id,
|
|
short_name=old_route.short_name,
|
|
long_name=old_route.long_name,
|
|
route_type=old_route.route_type,
|
|
mode=old_route.mode,
|
|
operator_name=old_route.operator_name,
|
|
min_lon=old_route.min_lon,
|
|
min_lat=old_route.min_lat,
|
|
max_lon=old_route.max_lon,
|
|
max_lat=old_route.max_lat,
|
|
route_key=old_route.route_key,
|
|
operator_key=old_route.operator_key,
|
|
)
|
|
)
|
|
|
|
rerun = client.post("/api/match/run").json()
|
|
assert rerun["manual"] >= 1
|
|
matches = client.get("/api/matches?status=accepted").json()
|
|
assert any(item["gtfs"]["route_id"] == match["gtfs"]["route_id"] for item in matches)
|
|
|
|
|
|
def _first_stop(client: TestClient, query: str) -> dict:
|
|
response = client.get(f"/api/journey/stops?q={query}")
|
|
assert response.status_code == 200
|
|
stops = response.json()["stops"]
|
|
assert stops
|
|
return stops[0]
|