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]