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
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""Tests for the UK bus stop finder backend."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import responses
|
|
|
|
from uk_bus_stops import core
|
|
from uk_bus_stops.app import ATCO_PATTERN, create_app, parse_coordinates
|
|
|
|
|
|
@pytest.fixture()
|
|
def client() -> Any:
|
|
"""Return a Flask test client."""
|
|
return create_app({"TESTING": True, "STOP_SOURCE": "osm"}).test_client()
|
|
|
|
|
|
@responses.activate
|
|
def test_geocode_limits_search_to_great_britain_and_identifies_app() -> None:
|
|
"""Nominatim receives the UK restriction and identifying User-Agent."""
|
|
responses.get(
|
|
core.NOMINATIM_URL,
|
|
json=[{"lat": "51.45", "lon": "-2.59", "display_name": "Bristol, England"}],
|
|
)
|
|
result = core.geocode("Bristol")
|
|
assert result == [{
|
|
"lat": 51.45, "lon": -2.59, "label": "Bristol, England", "type": None,
|
|
}]
|
|
request = responses.calls[0].request
|
|
assert "countrycodes=gb" in request.url
|
|
assert "limit=20" in request.url
|
|
assert "bounded=0" in request.url
|
|
assert request.headers["User-Agent"].startswith("uk-bus-stops/")
|
|
assert "edward@4angle.com" in request.headers["User-Agent"]
|
|
|
|
|
|
@responses.activate
|
|
def test_nearby_stops_normalises_deduplicates_and_orders() -> None:
|
|
"""Nearby Overpass objects become nearest-first stop records without duplicates."""
|
|
responses.get(
|
|
core.OVERPASS_URL,
|
|
json={"elements": [
|
|
{"type": "node", "id": 2, "lat": 51.46, "lon": -2.58,
|
|
"tags": {"highway": "bus_stop", "name": "Far Stop"}},
|
|
{"type": "node", "id": 1, "lat": 51.4501, "lon": -2.5901,
|
|
"tags": {"naptan:CommonName": "Near Stop", "naptan:AtcoCode": "0100BRP90314",
|
|
"naptan:Bearing": "NW"}},
|
|
{"type": "node", "id": 1, "lat": 51.4501, "lon": -2.5901,
|
|
"tags": {"name": "Near Stop", "naptan:AtcoCode": "0100BRP90314",
|
|
"naptan:Bearing": "NW"}},
|
|
]},
|
|
)
|
|
result = core.nearby_stops(51.45, -2.59)
|
|
assert [stop["id"] for stop in result] == [1, 2]
|
|
assert result[0]["atco_code"] == "0100BRP90314"
|
|
assert result[0]["transport_type"] == "Transport stop"
|
|
assert result[0]["bearing"] == "NW"
|
|
assert result[1]["atco_code"] is None
|
|
assert "highway%22%3D%22bus_stop" in responses.calls[0].request.url
|
|
|
|
|
|
@responses.activate
|
|
def test_find_atco_code_is_case_insensitive() -> None:
|
|
"""Direct code lookup emits an anchored, case-insensitive Overpass query."""
|
|
responses.get(
|
|
core.OVERPASS_URL,
|
|
json={"elements": [{
|
|
"type": "node", "id": 123, "lat": 51.45, "lon": -2.59,
|
|
"tags": {"name": "Central", "naptan:AtcoCode": "0100BRP90314"},
|
|
}]},
|
|
)
|
|
result = core.find_atco_code("0100brp90314")
|
|
assert result[0]["name"] == "Central"
|
|
assert "%5E0100brp90314%24%22%2Ci" in responses.calls[0].request.url
|
|
|
|
|
|
@responses.activate
|
|
def test_fetch_stop_by_osm_reference() -> None:
|
|
"""A shared stop reference is resolved to normalised stop details."""
|
|
responses.get(core.OVERPASS_URL, json={"elements": [{
|
|
"type": "node", "id": 123, "lat": 51.45, "lon": -2.59,
|
|
"tags": {"name": "Central", "highway": "bus_stop"},
|
|
}]})
|
|
stop = core.fetch_stop("node", 123)
|
|
assert stop is not None
|
|
assert stop["id"] == 123
|
|
assert stop["transport_type"] == "Bus stop"
|
|
|
|
|
|
@responses.activate
|
|
def test_overpass_uses_fallback_instance() -> None:
|
|
"""An unavailable UK Overpass host is retried on the global fallback."""
|
|
responses.get(core.OVERPASS_URL, status=503)
|
|
responses.get(core.OVERPASS_FALLBACK_URLS[0], json={"elements": []})
|
|
assert core.find_atco_code("0100BRP90314") == []
|
|
assert len(responses.calls) == 2
|
|
assert responses.calls[1].request.url.startswith(core.OVERPASS_FALLBACK_URLS[0])
|
|
|
|
|
|
@responses.activate
|
|
def test_routes_for_stop_returns_useful_route_fields() -> None:
|
|
"""Route lookup uses stop-area members and returns human-readable metadata."""
|
|
responses.get(
|
|
core.OVERPASS_URL,
|
|
json={"elements": [
|
|
{"type": "relation", "id": 9, "tags": {
|
|
"ref": "A1", "name": "Airport bus", "operator": "Example Bus",
|
|
"from": "Airport", "to": "City Centre",
|
|
}},
|
|
]},
|
|
)
|
|
routes = core.routes_for_stop("node", 123)
|
|
assert routes == [{
|
|
"id": 9, "ref": "A1", "name": "Airport bus", "operator": "Example Bus",
|
|
"from": "Airport", "to": "City Centre",
|
|
}]
|
|
query_url = responses.calls[0].request.url
|
|
assert "stop_area" in query_url
|
|
assert "route" in query_url
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tags", "expected"),
|
|
[
|
|
({"highway": "bus_stop"}, "Bus stop"),
|
|
({"railway": "station"}, "Railway station"),
|
|
({"railway": "tram_stop"}, "Tram stop"),
|
|
({"station": "subway"}, "Underground station"),
|
|
({"amenity": "ferry_terminal"}, "Ferry terminal"),
|
|
],
|
|
)
|
|
def test_transport_type_classifies_osm_tags(tags: dict[str, str], expected: str) -> None:
|
|
"""Common OSM public transport tagging receives a clear display type."""
|
|
assert core.transport_type(tags) == expected
|
|
|
|
|
|
def test_search_rejects_empty_query(client: Any) -> None:
|
|
"""The search endpoint returns the standard JSON error shape."""
|
|
response = client.get("/api/search")
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"] == "missing_query"
|
|
|
|
|
|
def test_place_name_is_not_mistaken_for_atco_code() -> None:
|
|
"""Long alphabetic place names still go through Nominatim geocoding."""
|
|
assert ATCO_PATTERN.fullmatch("Manchester") is None
|
|
assert ATCO_PATTERN.fullmatch("0100BRP90314") is not None
|
|
assert ATCO_PATTERN.fullmatch("010000056") is not None
|
|
|
|
|
|
def test_parse_coordinates_accepts_valid_pair_and_rejects_invalid_pair() -> None:
|
|
"""Coordinate searches require a comma and valid latitude/longitude ranges."""
|
|
assert parse_coordinates(" 51.4545, -2.5879 ") == (51.4545, -2.5879)
|
|
assert parse_coordinates("51.4545 -2.5879") is None
|
|
assert parse_coordinates("91, -2") is None
|
|
|
|
|
|
@responses.activate
|
|
def test_search_coordinates_bypasses_nominatim(client: Any) -> None:
|
|
"""Latitude/longitude entered in the search endpoint goes straight to Overpass."""
|
|
responses.get(core.OVERPASS_URL, json={"elements": []})
|
|
response = client.get("/api/search?q=51.4545%2C+-2.5879")
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data["location"]["lat"] == 51.4545
|
|
assert data["location"]["lon"] == -2.5879
|
|
assert len(responses.calls) == 1
|
|
assert responses.calls[0].request.url.startswith(core.OVERPASS_URL)
|
|
|
|
|
|
@responses.activate
|
|
def test_geocode_uses_unbounded_map_bias() -> None:
|
|
"""The map view biases Nominatim ranking without restricting its results."""
|
|
responses.get(core.NOMINATIM_URL, json=[])
|
|
assert core.geocode("North Street", viewbox=(-2.7, 51.4, -2.5, 51.5)) == []
|
|
url = responses.calls[0].request.url
|
|
assert "viewbox=-2.7%2C51.5%2C-2.5%2C51.4" in url
|
|
assert "bounded=0" in url
|
|
|
|
|
|
@responses.activate
|
|
def test_place_search_returns_uk_choices_before_loading_stops(client: Any) -> None:
|
|
"""Ambiguous place searches return UK Nominatim choices without querying Overpass."""
|
|
responses.get(core.NOMINATIM_URL, json=[
|
|
{"lat": "51.1", "lon": "-1.3", "display_name": "North Street, Winchester",
|
|
"type": "residential"},
|
|
{"lat": "51.4", "lon": "-2.6", "display_name": "North Street, Bristol",
|
|
"type": "secondary"},
|
|
])
|
|
response = client.get(
|
|
"/api/search?q=North+Street&west=-2.7&south=51.4&east=-2.5&north=51.5"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data["kind"] == "geocode"
|
|
assert [item["label"] for item in data["locations"]] == [
|
|
"North Street, Winchester", "North Street, Bristol",
|
|
]
|
|
assert len(responses.calls) == 1
|
|
assert "viewbox=-2.7%2C51.5%2C-2.5%2C51.4" in responses.calls[0].request.url
|
|
|
|
|
|
def test_search_rejects_partial_viewbox(client: Any) -> None:
|
|
"""Search-bias bounds must contain all four valid coordinates."""
|
|
response = client.get("/api/search?q=North+Street&west=-2.7")
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"] == "invalid_viewbox"
|
|
|
|
|
|
def test_location_coordinates_are_validated(client: Any) -> None:
|
|
"""Latitude and longitude must be numeric and within geographic ranges."""
|
|
assert client.get("/api/stops?lat=hello&lon=1").status_code == 400
|
|
assert client.get("/api/stops?lat=91&lon=1").status_code == 400
|
|
|
|
|
|
def test_map_bounds_reject_large_overpass_area(client: Any) -> None:
|
|
"""The server refuses viewport queries that are too large for Overpass."""
|
|
response = client.get("/api/stops/in-bounds?south=50&west=-3&north=51&east=-2")
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"] == "area_too_large"
|
|
|
|
|
|
@responses.activate
|
|
def test_shared_stop_api(client: Any) -> None:
|
|
"""The stop detail endpoint supports refreshing a shared stop URL."""
|
|
responses.get(core.OVERPASS_URL, json={"elements": [{
|
|
"type": "node", "id": 123, "lat": 51.45, "lon": -2.59,
|
|
"tags": {"name": "Central", "highway": "bus_stop"},
|
|
}]})
|
|
response = client.get("/api/stop/node/123")
|
|
assert response.status_code == 200
|
|
assert response.get_json()["stop"]["name"] == "Central"
|
|
|
|
|
|
@responses.activate
|
|
def test_map_bounds_load_small_area(client: Any) -> None:
|
|
"""A small viewport returns classified stops from Overpass."""
|
|
responses.get(core.OVERPASS_URL, json={"elements": [{
|
|
"type": "node", "id": 4, "lat": 51.45, "lon": -2.59,
|
|
"tags": {"railway": "station", "name": "Temple Meads"},
|
|
}]})
|
|
response = client.get(
|
|
"/api/stops/in-bounds?south=51.44&west=-2.60&north=51.46&east=-2.58"
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.get_json()["stops"][0]["transport_type"] == "Railway station"
|
|
|
|
|
|
def test_index_contains_search_and_map(client: Any) -> None:
|
|
"""The landing page includes the primary search and map controls."""
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert b"Find a UK bus stop code" in response.data
|
|
assert b'placeholder="Postcode, place or ATCO code"' in response.data
|
|
assert b'id="search-input"' in response.data
|
|
assert b'id="map"' in response.data
|
|
assert b'href="/about"' in response.data
|
|
|
|
|
|
def test_about_documents_urls_and_services(client: Any) -> None:
|
|
"""The About page documents shareable links, APIs, and OSM data sources."""
|
|
response = client.get("/about")
|
|
assert response.status_code == 200
|
|
assert b"Shareable URL parameters" in response.data
|
|
assert b"?q=Bristol+Temple+Meads" in response.data
|
|
assert b"?node=485403178" in response.data
|
|
assert b"?lat=51.4545&lon=-2.5879" in response.data
|
|
assert b"overpass.atownsend.org.uk" in response.data
|
|
assert b"https://git.4angle.com/edward/openstreetmap-tools" in response.data
|
|
assert b"JSON endpoints" in response.data
|