Edward Betts
29afb4f115
Import active NaPTAN stops into a local SQLite spatial index, including conversion of British National Grid coordinates. Add local searches, shareable stop links, source details and transport mode colours while retaining OSM route lookup by ATCO code. Closes #1
147 lines
6.8 KiB
Python
147 lines
6.8 KiB
Python
"""Offline NaPTAN import, spatial lookup and API regression tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import responses
|
|
|
|
from uk_bus_stops import core
|
|
from uk_bus_stops.app import create_app
|
|
from uk_bus_stops.naptan import NaptanError, NaptanStore, import_csv
|
|
|
|
|
|
@pytest.fixture()
|
|
def csv_file(tmp_path: Path) -> Path:
|
|
"""Write representative stop records, including grid-only and invalid data."""
|
|
path = tmp_path / "Stops.csv"
|
|
fields = ["ATCOCode", "CommonName", "StopType", "Status", "Latitude", "Longitude", "Easting", "Northing", "GridType", "Bearing", "Indicator"]
|
|
records = [
|
|
["0100BRP90317", "Temple Meads Stn", "BCT", "active", "51.44827", "-2.58302", "359581", "172304", "UKOS", "SE", "T1"],
|
|
["0100GRID001", "Gare café", "RSE", "active", "", "", "359581", "172304", "", "", ""],
|
|
["0100INACTIVE", "Closed", "BCT", "inactive", "51.448", "-2.583", "", "", "", "", ""],
|
|
["0100INVALID", "Invalid", "BCT", "active", "nan", "0", "", "", "", "", ""],
|
|
["0100CORNER", "Outside radius", "FER", "active", "51.456", "-2.571", "", "", "", "", ""],
|
|
["0100EAST001", "East", "BCT", "active", "51.44827", "-2.57302", "", "", "", "", ""],
|
|
["0100NORTH01", "North", "BCT", "active", "51.45627", "-2.58302", "", "", "", "", ""],
|
|
]
|
|
with path.open("w", encoding="utf-8-sig", newline="") as stream:
|
|
writer = csv.writer(stream)
|
|
writer.writerow(fields)
|
|
writer.writerows(records)
|
|
return path
|
|
|
|
|
|
@pytest.fixture()
|
|
def database(csv_file: Path, tmp_path: Path) -> Path:
|
|
"""Import an isolated database shared by query and API tests."""
|
|
path = tmp_path / "naptan.sqlite3"
|
|
assert import_csv(csv_file, path) == {"imported": 5, "inactive": 1, "invalid": 1, "converted": 1}
|
|
return path
|
|
|
|
|
|
def test_import_and_exact_lookup(database: Path) -> None:
|
|
"""Preserve codes, accents, transport metadata and converted grid positions."""
|
|
store = NaptanStore(database)
|
|
stop = store.find_atco_code("0100brp90317")[0]
|
|
assert stop["id"] == "0100BRP90317"
|
|
assert stop["source"] == "NaPTAN"
|
|
assert stop["bearing"] == "SE"
|
|
assert stop["indicator"] == "T1"
|
|
grid = store.find_atco_code("0100GRID001")[0]
|
|
assert grid["name"] == "Gare café"
|
|
assert grid["transport_type"] == "Railway station entrance"
|
|
assert grid["lat"] == pytest.approx(stop["lat"], abs=0.0001)
|
|
assert grid["lon"] == pytest.approx(stop["lon"], abs=0.0001)
|
|
assert store.find_atco_code("0100INACTIVE") == []
|
|
assert store.find_atco_code("' OR 1=1 --") == []
|
|
|
|
|
|
def test_spatial_search(database: Path) -> None:
|
|
"""Radius excludes box corners and orders longitude distances correctly."""
|
|
store = NaptanStore(database)
|
|
nearby = store.nearby_stops(51.44827, -2.58302)
|
|
assert [stop["id"] for stop in nearby][-2:] == ["0100EAST001", "0100NORTH01"]
|
|
assert "0100CORNER" not in {stop["id"] for stop in nearby}
|
|
assert len(store.stops_in_bounds(51.448, -2.584, 51.449, -2.582)) == 2
|
|
assert store.stops_in_bounds(51.44828, -2.583021, 51.44829, -2.583019) == []
|
|
assert store.nearby_stops(0, 0) == []
|
|
|
|
|
|
def test_atomic_refresh_and_failed_import(database: Path, csv_file: Path) -> None:
|
|
"""Invalid refreshes preserve the old database; valid refreshes replace it."""
|
|
original = csv_file.read_text(encoding="utf-8-sig")
|
|
csv_file.write_text("invalid\n")
|
|
with pytest.raises(NaptanError):
|
|
import_csv(csv_file, database)
|
|
store = NaptanStore(database)
|
|
assert store.find_atco_code("0100BRP90317")[0]["name"] == "Temple Meads Stn"
|
|
csv_file.write_text(original.replace("Temple Meads Stn", "Updated name"))
|
|
import_csv(csv_file, database)
|
|
assert store.find_atco_code("0100BRP90317")[0]["name"] == "Updated name"
|
|
assert not list(database.parent.glob(".naptan-*"))
|
|
|
|
|
|
@responses.activate
|
|
def test_naptan_api_needs_no_network(database: Path) -> None:
|
|
"""Coordinate, viewport, code and shared stop requests all use local data."""
|
|
client = create_app({"TESTING": True, "STOP_SOURCE": "naptan", "NAPTAN_DATABASE": str(database)}).test_client()
|
|
for url in (
|
|
"/api/search?q=0100brp90317",
|
|
"/api/search?q=51.44827,-2.58302",
|
|
"/api/stops?lat=51.44827&lon=-2.58302",
|
|
"/api/stops/in-bounds?south=51.448&west=-2.584&north=51.449&east=-2.582",
|
|
):
|
|
response = client.get(url)
|
|
assert response.status_code == 200
|
|
assert response.json["stops"][0]["source"] == "NaPTAN"
|
|
response = client.get("/api/stop/naptan/0100brp90317")
|
|
assert response.status_code == 200
|
|
assert response.json["stop"]["id"] == "0100BRP90317"
|
|
assert client.get("/api/stop/naptan/missing").status_code == 404
|
|
assert len(responses.calls) == 0
|
|
|
|
|
|
def test_missing_database(tmp_path: Path) -> None:
|
|
"""An unconfigured installation explains how to import instead of creating a DB."""
|
|
database = tmp_path / "missing.sqlite3"
|
|
client = create_app({"TESTING": True, "NAPTAN_DATABASE": str(database), "STOP_SOURCE": "naptan"}).test_client()
|
|
response = client.get("/api/stops?lat=51.45&lon=-2.58")
|
|
assert response.status_code == 503
|
|
assert response.json["error"] == "naptan_unavailable"
|
|
assert not database.exists()
|
|
|
|
|
|
@responses.activate
|
|
def test_naptan_routes_match_atco_and_deduplicate(database: Path) -> None:
|
|
"""Multiple matching OSM objects contribute one combined route list."""
|
|
responses.get(core.OVERPASS_URL, json={"elements": [
|
|
{"type": "node", "id": 1, "lat": 51.448, "lon": -2.583},
|
|
{"type": "way", "id": 2, "center": {"lat": 51.448, "lon": -2.583}},
|
|
]})
|
|
for _ in range(2):
|
|
responses.get(core.OVERPASS_URL, json={"elements": [
|
|
{"id": 12, "tags": {"ref": "A1", "route": "bus"}},
|
|
]})
|
|
client = create_app({"TESTING": True, "NAPTAN_DATABASE": str(database)}).test_client()
|
|
response = client.get("/api/stop/naptan/0100BRP90317/routes")
|
|
assert response.status_code == 200
|
|
assert response.json["matched_stops"] == 2
|
|
assert [route["ref"] for route in response.json["routes"]] == ["A1"]
|
|
assert "0100BRP90317" in responses.calls[0].request.url
|
|
|
|
|
|
@responses.activate
|
|
def test_naptan_routes_without_osm_match(database: Path) -> None:
|
|
"""A missing OSM match has a distinct result from a service failure."""
|
|
responses.get(core.OVERPASS_URL, json={"elements": []})
|
|
client = create_app({"TESTING": True, "NAPTAN_DATABASE": str(database)}).test_client()
|
|
assert client.get("/api/stop/naptan/0100BRP90317/routes").json == {
|
|
"routes": [], "source": "OpenStreetMap", "matched_stops": 0,
|
|
}
|
|
responses.replace(responses.GET, core.OVERPASS_URL, status=503)
|
|
responses.get(core.OVERPASS_FALLBACK_URLS[0], status=503)
|
|
assert client.get("/api/stop/naptan/0100BRP90317/routes").status_code == 502
|
|
assert client.get("/api/stop/naptan/0100BRP90317").status_code == 200
|