Add UK bus stop finder

tests/test_uk_bus_stops.py

@@ -0,0 +1,213 @@
+"""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}).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"}
+    request = responses.calls[0].request
+    assert "countrycodes=gb" 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
+
+
+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)
+
+
+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'id="search-input"' in response.data
+    assert b'id="map"' in response.data

tests/test_uk_bus_stops_playwright.py

@@ -0,0 +1,194 @@
+"""Browser tests for the UK bus stop finder using Python Playwright."""
+
+from __future__ import annotations
+
+import re
+import threading
+from collections.abc import Iterator
+from typing import Any
+
+import pytest
+from werkzeug.serving import make_server
+
+from uk_bus_stops.app import create_app
+
+playwright_api = pytest.importorskip("playwright.sync_api")
+
+
+@pytest.fixture(scope="module")
+def flask_url() -> Iterator[str]:
+    """Run the Flask application on a temporary local port."""
+    server = make_server("127.0.0.1", 0, create_app({"TESTING": True}))
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    yield f"http://127.0.0.1:{server.server_port}"
+    server.shutdown()
+    thread.join()
+
+
+@pytest.fixture(scope="module")
+def chromium_browser() -> Iterator[Any]:
+    """Launch the installed Chromium browser through Python Playwright."""
+    with playwright_api.sync_playwright() as playwright:
+        instance = playwright.chromium.launch(headless=True)
+        yield instance
+        instance.close()
+
+
+def test_search_and_stop_details_in_browser(chromium_browser: Any, flask_url: str) -> None:
+    """A location search renders stops, an ATCO code, tags, and route metadata."""
+    page = chromium_browser.new_page(viewport={"width": 1280, "height": 800})
+    page_errors: list[str] = []
+    page.on("pageerror", lambda error: page_errors.append(str(error)))
+    stop = {
+        "type": "node", "id": 123, "lat": 51.451, "lon": -2.591,
+        "name": "Central Stop", "atco_code": "0100BRP90314", "indicator": "N",
+        "bearing": "NW",
+        "transport_type": "Bus stop",
+        "tags": {"highway": "bus_stop", "shelter": "yes"},
+    }
+    page.route("**/api/search?*", lambda route: route.fulfill(json={
+        "kind": "location",
+        "location": {"lat": 51.45, "lon": -2.59, "label": "Bristol"},
+        "stops": [{
+            "type": "node", "id": 125, "lat": 51.47, "lon": -2.61,
+            "name": "Far Stop", "atco_code": "0100BRP90316", "indicator": None,
+            "bearing": "E", "transport_type": "Bus stop",
+            "tags": {"highway": "bus_stop"},
+        }, stop],
+    }))
+    page.route("**/api/stops?*", lambda route: route.fulfill(json={
+        "kind": "location",
+        "location": {"lat": 51.451, "lon": -2.591, "label": "Central Stop"},
+        "stops": [stop, {
+            "type": "node", "id": 124, "lat": 51.452, "lon": -2.592,
+            "name": "Nearby Stop", "atco_code": "0100BRP90315", "indicator": "S",
+            "bearing": "SE", "transport_type": "Bus stop",
+            "tags": {"highway": "bus_stop"},
+        }],
+    }))
+    page.route("**/api/stops/in-bounds?*", lambda route: route.fulfill(json={
+        "kind": "map",
+        "stops": [stop, {
+            "type": "node", "id": 124, "lat": 51.452, "lon": -2.592,
+            "name": "Nearby Stop", "atco_code": "0100BRP90315", "indicator": "S",
+            "bearing": "SE", "transport_type": "Bus stop",
+            "tags": {"highway": "bus_stop"},
+        }],
+    }))
+    page.route("**/api/stop/node/123", lambda route: route.fulfill(json={"stop": stop}))
+    page.route("**/api/stop/node/123/routes", lambda route: route.fulfill(json={"routes": [{
+        "id": 9, "ref": "A1", "name": "Airport bus", "operator": "Example Bus",
+        "from": "Airport", "to": "City Centre",
+    }]}))
+    page.goto(flask_url, wait_until="networkidle")
+    page.get_by_label("Location or ATCO code").fill("Bristol")
+    page.get_by_role("button", name="Search").click()
+    playwright_api.expect(page).to_have_url(re.compile(r"\?q=Bristol$"))
+    playwright_api.expect(page.get_by_text("Central Stop", exact=True)).to_be_visible()
+    assert page.locator("#stop-list .stop-title").all_inner_texts() == ["Central Stop", "Far Stop"]
+    playwright_api.expect(page.locator("#stop-list .stop-meta").first).to_contain_text("m · Bus stop")
+    playwright_api.expect(page.locator("#stop-list .stop-meta").first).to_contain_text("NW-bound")
+    assert page.evaluate("map.getZoom()") > 6
+    page.evaluate("markers.getLayers()[0].openTooltip()")
+    playwright_api.expect(page.get_by_text("Central Stop · NW-bound · Bus stop", exact=True)).to_be_visible()
+    page.get_by_text("Central Stop", exact=True).click()
+    playwright_api.expect(page).to_have_url(re.compile(r"\?node=123$"))
+    playwright_api.expect(page.locator("#atco-code")).to_have_text("0100BRP90314")
+    playwright_api.expect(page.locator("#stop-type")).to_have_text("Bus stop")
+    playwright_api.expect(page.locator("#stop-bearing")).to_have_text("NW-bound")
+    playwright_api.expect(page.locator(".selected-stop-marker")).to_be_visible()
+    playwright_api.expect(page.get_by_text("Airport → City Centre · Example Bus", exact=True)).to_be_visible()
+    assert page_errors == []
+    page.reload(wait_until="networkidle")
+    playwright_api.expect(page.get_by_label("Location or ATCO code")).to_have_value("")
+    playwright_api.expect(page.locator("#stop-name")).to_have_text("Central Stop")
+    playwright_api.expect(page.locator("#atco-code")).to_have_text("0100BRP90314")
+    playwright_api.expect(page.locator("#stop-bearing")).to_have_text("NW-bound")
+    playwright_api.expect(page.get_by_role("button", name="All nearby stops")).to_be_visible()
+    page.get_by_role("button", name="All nearby stops").click()
+    playwright_api.expect(page.get_by_text("Nearby Stop", exact=True)).to_be_visible()
+    playwright_api.expect(page).to_have_url(re.compile(r"\?lat=.*&lon=.*$"))
+    page.close()
+
+
+def test_moving_map_loads_only_when_zoomed_in(chromium_browser: Any, flask_url: str) -> None:
+    """Panning a close map loads visible stops while a wide map makes no query."""
+    page = chromium_browser.new_page(viewport={"width": 1280, "height": 800})
+    requests: list[str] = []
+
+    def fulfil_bounds(route: Any) -> None:
+        """Record viewport calls and return a railway station."""
+        requests.append(route.request.url)
+        route.fulfill(json={"kind": "map", "stops": [{
+            "type": "node", "id": 456, "lat": 51.45, "lon": -2.59,
+            "name": "Temple Meads", "atco_code": "9100BRSTLTM", "indicator": None,
+            "transport_type": "Railway station", "tags": {"railway": "station"},
+        }]})
+
+    page.route("**/api/stops/in-bounds?*", fulfil_bounds)
+    page.route("**/api/stop/node/456/routes", lambda route: route.fulfill(json={"routes": []}))
+    page.goto(flask_url, wait_until="networkidle")
+    page.evaluate("userMapInteractionPending = true; map.setView([51.45, -2.59], 15)")
+    playwright_api.expect(page.get_by_text("Temple Meads", exact=True)).to_be_visible()
+    assert len(requests) == 1
+    page.evaluate("userMapInteractionPending = true; map.setZoom(10)")
+    playwright_api.expect(page.locator("#map-status")).to_have_text(
+        "Zoom in to load transport stops"
+    )
+    page.wait_for_timeout(500)
+    assert len(requests) == 1
+    page.evaluate("userMapInteractionPending = true; map.setZoom(15)")
+    page.wait_for_timeout(700)
+    assert len(requests) == 2
+    page.get_by_text("Temple Meads", exact=True).click()
+    playwright_api.expect(page.locator("#stop-name")).to_have_text("Temple Meads")
+    page.wait_for_timeout(700)
+    playwright_api.expect(page.locator("#stop-detail")).to_be_visible()
+    assert len(requests) == 3
+    page.evaluate("userMapInteractionPending = true; map.panBy([80, 0])")
+    page.wait_for_timeout(700)
+    playwright_api.expect(page.locator("#stop-name")).to_have_text("Temple Meads")
+    playwright_api.expect(page.locator("#stop-detail")).to_be_visible()
+    playwright_api.expect(page).to_have_url(re.compile(r"\?node=456$"))
+    assert len(requests) == 4
+    page.close()
+
+
+def test_coordinate_url_and_search_input(chromium_browser: Any, flask_url: str) -> None:
+    """lat/lon URLs load on refresh and coordinate input creates a shareable URL."""
+    page = chromium_browser.new_page(viewport={"width": 1280, "height": 800})
+    requested_urls: list[str] = []
+
+    def fulfil_stops(route: Any) -> None:
+        """Record coordinate API calls and return an empty nearby result."""
+        requested_urls.append(route.request.url)
+        route.fulfill(json={
+            "kind": "location",
+            "location": {"lat": 51.4545, "lon": -2.5879, "label": "Coordinates"},
+            "stops": [],
+        })
+
+    page.route("**/api/stops?*", fulfil_stops)
+    page.goto(f"{flask_url}?lat=51.4545&lon=-2.5879", wait_until="networkidle")
+    playwright_api.expect(page.get_by_label("Location or ATCO code")).to_have_value(
+        "51.4545, -2.5879"
+    )
+    playwright_api.expect(page.locator("#stop-count")).to_have_text("0 found")
+    assert page.evaluate("map.getZoom()") == 16
+    assert "lat=51.4545" in requested_urls[-1]
+
+    page.get_by_label("Location or ATCO code").fill("51.46, -2.58")
+    page.get_by_role("button", name="Search").click()
+    playwright_api.expect(page).to_have_url(re.compile(r"\?lat=51.46&lon=-2.58$"))
+    playwright_api.expect(page.locator("#stop-count")).to_have_text("0 found")
+    page.close()
+
+
+def test_mobile_layout_opens_results_sheet(chromium_browser: Any, flask_url: str) -> None:
+    """The finder exposes its search panel as an open mobile bottom sheet."""
+    page = chromium_browser.new_page(viewport={"width": 390, "height": 844})
+    page.goto(flask_url, wait_until="domcontentloaded")
+    playwright_api.expect(page.locator("#sidebar")).to_have_class(re.compile("panel-open"))
+    playwright_api.expect(page.get_by_label("Location or ATCO code")).to_be_visible()
+    page.close()