diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a7ea19fc..29694daa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,4 +1,4 @@ -# This workflow runs the repo's CI tests. +# This workflow runs the repo's (local/offline) CI tests. name: CI on: @@ -7,7 +7,7 @@ on: pull_request: branches: [main] schedule: - - cron: 5 4 * * 1 # every monday at 04:05 UTC + - cron: 30 4 * * 1 # every monday at 04:30 UTC workflow_dispatch: concurrency: diff --git a/.github/workflows/test-online-apis.yml b/.github/workflows/test-online-apis.yml new file mode 100644 index 00000000..21518162 --- /dev/null +++ b/.github/workflows/test-online-apis.yml @@ -0,0 +1,46 @@ +# This workflow runs live online public web API compatibility tests. +name: Test online public web APIs + +on: + schedule: + - cron: 5 4 * * 1 # every monday at 04:05 UTC + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + test_online_apis: + if: ${{ github.repository == 'gboeing/osmnx' }} + name: Public web APIs + runs-on: ubuntu-latest + timeout-minutes: 30 + + defaults: + run: + shell: bash -elo pipefail {0} + + steps: + - name: Checkout repo + uses: actions/checkout@v6 + + - name: Install uv + uses: astral-sh/setup-uv@v7 + with: + cache-dependency-glob: pyproject.toml + enable-cache: true + python-version: '3.14' + + - name: Install OSMnx + run: | + uv python pin 3.14 + uv sync --all-extras --group test + + - name: Test public web APIs + run: uv run pytest -m online --numprocesses=3 + env: + OSMNX_RUN_ONLINE_TESTS: '1' diff --git a/osmnx/_overpass.py b/osmnx/_overpass.py index d71fb5e5..26d13d4c 100644 --- a/osmnx/_overpass.py +++ b/osmnx/_overpass.py @@ -446,9 +446,6 @@ def _overpass_request(data: OrderedDict[str, Any]) -> dict[str, Any]: response_json The Overpass API's response. """ - # resolve url to same IP even if there is server round-robin redirecting - _http._config_dns(settings.overpass_url) - # prepare the Overpass API URL and see if request already exists in cache url = settings.overpass_url.rstrip("/") + "/interpreter" prepared_url = str(requests.Request("GET", url, params=data).prepare().url) @@ -456,6 +453,9 @@ def _overpass_request(data: OrderedDict[str, Any]) -> dict[str, Any]: if isinstance(cached_response_json, dict): return cached_response_json + # resolve url to same IP even if there is server round-robin redirecting + _http._config_dns(settings.overpass_url) + # pause then request this URL pause = _get_overpass_pause(settings.overpass_url) hostname = _http._hostname_from_url(url) diff --git a/osmnx/stats.py b/osmnx/stats.py index 62cd42ca..44a42b17 100644 --- a/osmnx/stats.py +++ b/osmnx/stats.py @@ -264,7 +264,7 @@ def circuity_avg(Gu: nx.MultiGraph) -> float | None: sl_dists_total = sl_dists[~np.isnan(sl_dists)].sum() try: return float(edge_length_total(Gu) / sl_dists_total) - except ZeroDivisionError: + except ZeroDivisionError: # pragma: no cover return None diff --git a/osmnx/utils.py b/osmnx/utils.py index 5d393549..9a2c1ab2 100644 --- a/osmnx/utils.py +++ b/osmnx/utils.py @@ -144,7 +144,7 @@ def log( message = f"{ts()} {message}" message = ud.normalize("NFKD", message).encode("ascii", errors="replace").decode() - try: + try: # pragma: no cover # print explicitly to terminal in case Jupyter has captured stdout if getattr(sys.stdout, "_original_stdstream_copy", None) is not None: # redirect the Jupyter-captured pipe back to original @@ -152,7 +152,7 @@ def log( sys.stdout._original_stdstream_copy = None # type: ignore[union-attr] with redirect_stdout(sys.__stdout__): print(message, file=sys.__stdout__, flush=True) - except OSError: + except OSError: # pragma: no cover # handle pytest on Windows raising OSError from sys.__stdout__ print(message, flush=True) # noqa: T201 diff --git a/pyproject.toml b/pyproject.toml index 27aec777..6e9bc7bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [build-system] build-backend = "uv_build" -requires = ["uv_build==0.10.*"] +requires = ["uv_build==0.11.*"] [project] authors = [{ name = "Geoff Boeing", email = "boeing@usc.edu" }] @@ -61,6 +61,7 @@ Documentation = "https://osmnx.readthedocs.io" [tool.coverage.report] exclude_also = ["@overload", "if TYPE_CHECKING:"] +fail_under = 99 [tool.mypy] cache_dir = "~/.cache/prek/cache/mypy" @@ -79,6 +80,10 @@ addopts = ["-ra", "--verbose", "--maxfail=1", "--numprocesses=3", "--dist=loadgr cache_dir = "~/.cache/pytest" filterwarnings = ["error", "ignore::UserWarning"] log_level = "INFO" +markers = [ + "offline: deterministic tests that do not make online public web API calls", + "online: tests that make live online public web API calls", +] minversion = 9 strict = true testpaths = ["tests"] @@ -108,7 +113,7 @@ convention = "numpy" max-args = 8 [tool.uv] -required-version = "==0.10.*" # match version above and in Dockerfile +required-version = "==0.11.*" # match version above and in Dockerfile [tool.uv.build-backend] module-root = "" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100755 index 00000000..90fc4b5f --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python +"""Shared pytest fixtures for deterministic OSMnx tests.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import TYPE_CHECKING +from typing import Any +from typing import TypeAlias + +import matplotlib as mpl +import networkx as nx +import pytest +import requests +from shapely import LineString + +mpl.use("Agg") + +if TYPE_CHECKING: + from collections.abc import Iterator + +LOCATION_POINT = (37.791427, -122.410018) +ADDRESS = "Transamerica Pyramid, 600 Montgomery Street, San Francisco, California, USA" +PLACE = {"city": "Piedmont", "state": "California", "country": "USA"} +TAGS: dict[str, bool | str | list[str]] = { + "landuse": True, + "building": True, + "highway": True, + "amenity": True, +} + +_ResponseJson: TypeAlias = dict[str, object] | list[dict[str, object]] +HTTP_OK = 200 +HTTP_ERROR = 500 + + +def _drive_graph() -> nx.MultiDiGraph: + import osmnx as ox # noqa: PLC0415 + + return ox.graph_from_point( + LOCATION_POINT, + dist=500, + network_type="drive", + simplify=False, + retain_all=True, + ) + + +def _toy_graph(*, crs: str = "epsg:4326") -> nx.MultiDiGraph: + G = nx.MultiDiGraph(crs=crs) + G.add_node(1, x=0.0, y=0.0, street_count=1, elevation=0.0) + G.add_node(2, x=1.0, y=0.0, street_count=2, elevation=10.0) + G.add_node(3, x=2.0, y=0.0, street_count=1, elevation=20.0) + G.add_edge( + 1, + 2, + osmid=10, + length=1.0, + highway="residential", + maxspeed="25 mph", + geometry=LineString([(0, 0), (1, 0)]), + ) + G.add_edge( + 2, + 3, + osmid=11, + length=1.0, + highway=["primary", "secondary"], + maxspeed=["30 mph", "50"], + geometry=LineString([(1, 0), (2, 0)]), + ) + return G + + +class _Response(requests.Response): + def __init__(self, payload: _ResponseJson, *, ok: bool = True, status_code: int = 200) -> None: + super().__init__() + self._payload = payload + self.status_code = status_code if ok or status_code != HTTP_OK else HTTP_ERROR + self.reason = "OK" if ok else "Error" + self._content = json.dumps(payload).encode() + self.url = "https://example.com/api" + + def json(self, **kwargs: object) -> _ResponseJson: + del kwargs + return self._payload + + +drive_graph = _drive_graph +toy_graph = _toy_graph +Response = _Response + +HTTP_CACHE_DIR = Path(__file__).parent / "input_data" / "http_cache" + + +def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: + """ + Skip live online tests unless explicitly requested. + + Parameters + ---------- + config + Pytest configuration object. + items + Collected test items. + """ + del config + + if os.environ.get("OSMNX_RUN_ONLINE_TESTS"): + return + + skip_online = pytest.mark.skip(reason="set OSMNX_RUN_ONLINE_TESTS=1 to run online tests") + for item in items: + if item.get_closest_marker("online"): + item.add_marker(skip_online) + + +@pytest.fixture(autouse=True) +def _isolate_settings(tmp_path: Path) -> Iterator[None]: + """ + Restore global settings after each test and isolate generated files. + + Parameters + ---------- + tmp_path + Temporary directory unique to the test. + + Yields + ------ + None + Control returns to pytest after each test. + """ + import osmnx as ox # noqa: PLC0415 + + original_settings = { + name: getattr(ox.settings, name) + for name in dir(ox.settings) + if not name.startswith("_") and name.islower() + } + + ox.settings.data_folder = tmp_path / "data" + ox.settings.logs_folder = tmp_path / "logs" + ox.settings.imgs_folder = tmp_path / "imgs" + ox.settings.cache_folder = tmp_path / "cache" + ox.settings.log_console = False + ox.settings.log_file = False + ox.settings.use_cache = True + + yield + + for name, value in original_settings.items(): + setattr(ox.settings, name, value) + + +@pytest.fixture(autouse=True) +def _block_network( + monkeypatch: pytest.MonkeyPatch, + request: pytest.FixtureRequest, +) -> None: + """ + Prevent accidental live HTTP calls in the default offline suite. + + Parameters + ---------- + monkeypatch + Pytest fixture for temporary object replacement. + request + Pytest request object for the active test. + """ + if request.node.get_closest_marker("online") and os.environ.get("OSMNX_RUN_ONLINE_TESTS"): + return + + def _blocked_request(*args: Any, **kwargs: Any) -> None: # noqa: ANN401, ARG001 + msg = ( + "Network access is blocked in offline tests. Mark the test with " + "`@pytest.mark.online` and set OSMNX_RUN_ONLINE_TESTS=1 to allow it." + ) + raise AssertionError(msg) + + monkeypatch.setattr(requests, "get", _blocked_request) + monkeypatch.setattr(requests, "post", _blocked_request) + + +@pytest.fixture +def http_cache() -> Path: + """ + Point OSMnx cache lookups at committed raw HTTP response fixtures. + + Returns + ------- + pathlib.Path + Directory containing committed raw HTTP cache files. + """ + import osmnx as ox # noqa: PLC0415 + + ox.settings.cache_folder = HTTP_CACHE_DIR + ox.settings.use_cache = True + ox.settings.overpass_rate_limit = False + return HTTP_CACHE_DIR diff --git a/tests/input_data/http_cache/0be0d6b34a9700ade60bc9f39b36639c0c37ca41.json b/tests/input_data/http_cache/0be0d6b34a9700ade60bc9f39b36639c0c37ca41.json new file mode 100644 index 00000000..fe51488c --- /dev/null +++ b/tests/input_data/http_cache/0be0d6b34a9700ade60bc9f39b36639c0c37ca41.json @@ -0,0 +1 @@ +[] diff --git a/tests/input_data/http_cache/4fc0414e3f5d633d1740c3063d92207ce10bfaf4.json b/tests/input_data/http_cache/4fc0414e3f5d633d1740c3063d92207ce10bfaf4.json new file mode 100644 index 00000000..c34606bc --- /dev/null +++ b/tests/input_data/http_cache/4fc0414e3f5d633d1740c3063d92207ce10bfaf4.json @@ -0,0 +1,25 @@ +{ + "results": [ + { + "elevation": 10.0, + "location": {} + }, + { + "elevation": 11.0, + "location": {} + }, + { + "elevation": 12.0, + "location": {} + }, + { + "elevation": 13.0, + "location": {} + }, + { + "elevation": 14.0, + "location": {} + } + ], + "status": "OK" +} diff --git a/tests/input_data/http_cache/51a8a60e27ec040a17a10a2659d7946d80607df7.json b/tests/input_data/http_cache/51a8a60e27ec040a17a10a2659d7946d80607df7.json new file mode 100644 index 00000000..b1a290f0 --- /dev/null +++ b/tests/input_data/http_cache/51a8a60e27ec040a17a10a2659d7946d80607df7.json @@ -0,0 +1,49 @@ +[ + { + "addresstype": "city", + "boundingbox": [ + "37.7860", + "37.7970", + "-122.4160", + "-122.4040" + ], + "class": "boundary", + "display_name": "Piedmont Fixture, California, USA", + "geojson": { + "coordinates": [ + [ + [ + -122.416, + 37.786 + ], + [ + -122.404, + 37.786 + ], + [ + -122.404, + 37.797 + ], + [ + -122.416, + 37.797 + ], + [ + -122.416, + 37.786 + ] + ] + ], + "type": "Polygon" + }, + "importance": 0.8, + "lat": "37.791427", + "lon": "-122.410018", + "name": "Piedmont Fixture", + "osm_id": 2999176, + "osm_type": "relation", + "place_id": 9002, + "place_rank": 16, + "type": "administrative" + } +] diff --git a/tests/input_data/http_cache/80be009ca881764187f03bf8c6aca285ef962424.json b/tests/input_data/http_cache/80be009ca881764187f03bf8c6aca285ef962424.json new file mode 100644 index 00000000..50a5b5a1 --- /dev/null +++ b/tests/input_data/http_cache/80be009ca881764187f03bf8c6aca285ef962424.json @@ -0,0 +1,29 @@ +[ + { + "addresstype": "neighbourhood", + "boundingbox": [ + "37.7900", + "37.7920", + "-122.4110", + "-122.4090" + ], + "class": "place", + "display_name": "Bunker Hill Fixture", + "geojson": { + "coordinates": [ + -122.410018, + 37.791427 + ], + "type": "Point" + }, + "importance": 0.1, + "lat": "37.791427", + "lon": "-122.410018", + "name": "Bunker Hill Fixture", + "osm_id": 9003, + "osm_type": "node", + "place_id": 9003, + "place_rank": 20, + "type": "neighbourhood" + } +] diff --git a/tests/input_data/http_cache/8882c7a2619f4a40f1bc962a03d395703a4113d1.json b/tests/input_data/http_cache/8882c7a2619f4a40f1bc962a03d395703a4113d1.json new file mode 100644 index 00000000..c7ad5e82 --- /dev/null +++ b/tests/input_data/http_cache/8882c7a2619f4a40f1bc962a03d395703a4113d1.json @@ -0,0 +1,101 @@ +{ + "elements": [ + { + "id": 101, + "lat": 37.7905, + "lon": -122.411, + "tags": { + "highway": "traffic_signals" + }, + "type": "node" + }, + { + "id": 102, + "lat": 37.7905, + "lon": -122.409, + "type": "node" + }, + { + "id": 103, + "lat": 37.792, + "lon": -122.409, + "type": "node" + }, + { + "id": 104, + "lat": 37.792, + "lon": -122.411, + "type": "node" + }, + { + "id": 105, + "lat": 37.79125, + "lon": -122.41, + "type": "node" + }, + { + "id": 201, + "nodes": [ + 101, + 102, + 103 + ], + "tags": { + "highway": "residential", + "lanes": "1", + "maxspeed": "25 mph", + "name": "Fixture Street" + }, + "type": "way" + }, + { + "id": 202, + "nodes": [ + 103, + 104, + 101 + ], + "tags": { + "highway": "secondary", + "lanes": "2", + "name": "Loop Avenue", + "oneway": "yes" + }, + "type": "way" + }, + { + "id": 203, + "nodes": [ + 102, + 105, + 104 + ], + "tags": { + "highway": "service", + "name": "Service Lane", + "service": "alley" + }, + "type": "way" + }, + { + "id": 204, + "nodes": [ + 104, + 105, + 102 + ], + "tags": { + "bicycle": "designated", + "highway": "cycleway", + "name": "Cycle Link" + }, + "type": "way" + } + ], + "generator": "OSMnx offline test fixture", + "osm3s": { + "copyright": "fixture", + "timestamp_osm_base": "2026-01-01T00:00:00Z" + }, + "version": 0.6 +} diff --git a/tests/input_data/http_cache/be7271acafb60f59ad3647ef253f21fbed720a0a.json b/tests/input_data/http_cache/be7271acafb60f59ad3647ef253f21fbed720a0a.json new file mode 100644 index 00000000..59a8e2bb --- /dev/null +++ b/tests/input_data/http_cache/be7271acafb60f59ad3647ef253f21fbed720a0a.json @@ -0,0 +1,29 @@ +[ + { + "addresstype": "office", + "boundingbox": [ + "37.7909", + "37.7920", + "-122.4107", + "-122.4094" + ], + "class": "office", + "display_name": "Transamerica Pyramid, 600 Montgomery Street, San Francisco, California, USA", + "geojson": { + "coordinates": [ + -122.410018, + 37.791427 + ], + "type": "Point" + }, + "importance": 0.9, + "lat": "37.791427", + "lon": "-122.410018", + "name": "Transamerica Pyramid Fixture", + "osm_id": 9001, + "osm_type": "node", + "place_id": 9001, + "place_rank": 30, + "type": "yes" + } +] diff --git a/tests/input_data/http_cache/c41588e004167fdff982a269a2fcc476a61076de.json b/tests/input_data/http_cache/c41588e004167fdff982a269a2fcc476a61076de.json new file mode 100644 index 00000000..c7ad5e82 --- /dev/null +++ b/tests/input_data/http_cache/c41588e004167fdff982a269a2fcc476a61076de.json @@ -0,0 +1,101 @@ +{ + "elements": [ + { + "id": 101, + "lat": 37.7905, + "lon": -122.411, + "tags": { + "highway": "traffic_signals" + }, + "type": "node" + }, + { + "id": 102, + "lat": 37.7905, + "lon": -122.409, + "type": "node" + }, + { + "id": 103, + "lat": 37.792, + "lon": -122.409, + "type": "node" + }, + { + "id": 104, + "lat": 37.792, + "lon": -122.411, + "type": "node" + }, + { + "id": 105, + "lat": 37.79125, + "lon": -122.41, + "type": "node" + }, + { + "id": 201, + "nodes": [ + 101, + 102, + 103 + ], + "tags": { + "highway": "residential", + "lanes": "1", + "maxspeed": "25 mph", + "name": "Fixture Street" + }, + "type": "way" + }, + { + "id": 202, + "nodes": [ + 103, + 104, + 101 + ], + "tags": { + "highway": "secondary", + "lanes": "2", + "name": "Loop Avenue", + "oneway": "yes" + }, + "type": "way" + }, + { + "id": 203, + "nodes": [ + 102, + 105, + 104 + ], + "tags": { + "highway": "service", + "name": "Service Lane", + "service": "alley" + }, + "type": "way" + }, + { + "id": 204, + "nodes": [ + 104, + 105, + 102 + ], + "tags": { + "bicycle": "designated", + "highway": "cycleway", + "name": "Cycle Link" + }, + "type": "way" + } + ], + "generator": "OSMnx offline test fixture", + "osm3s": { + "copyright": "fixture", + "timestamp_osm_base": "2026-01-01T00:00:00Z" + }, + "version": 0.6 +} diff --git a/tests/input_data/http_cache/ce6b1b54fd55e2c991a46e037d9038e4bc4fd308.json b/tests/input_data/http_cache/ce6b1b54fd55e2c991a46e037d9038e4bc4fd308.json new file mode 100644 index 00000000..fe51488c --- /dev/null +++ b/tests/input_data/http_cache/ce6b1b54fd55e2c991a46e037d9038e4bc4fd308.json @@ -0,0 +1 @@ +[] diff --git a/tests/input_data/http_cache/d206c7ec7b8dd802e70c1fabbce4cc8dd5a23054.json b/tests/input_data/http_cache/d206c7ec7b8dd802e70c1fabbce4cc8dd5a23054.json new file mode 100644 index 00000000..c7ad5e82 --- /dev/null +++ b/tests/input_data/http_cache/d206c7ec7b8dd802e70c1fabbce4cc8dd5a23054.json @@ -0,0 +1,101 @@ +{ + "elements": [ + { + "id": 101, + "lat": 37.7905, + "lon": -122.411, + "tags": { + "highway": "traffic_signals" + }, + "type": "node" + }, + { + "id": 102, + "lat": 37.7905, + "lon": -122.409, + "type": "node" + }, + { + "id": 103, + "lat": 37.792, + "lon": -122.409, + "type": "node" + }, + { + "id": 104, + "lat": 37.792, + "lon": -122.411, + "type": "node" + }, + { + "id": 105, + "lat": 37.79125, + "lon": -122.41, + "type": "node" + }, + { + "id": 201, + "nodes": [ + 101, + 102, + 103 + ], + "tags": { + "highway": "residential", + "lanes": "1", + "maxspeed": "25 mph", + "name": "Fixture Street" + }, + "type": "way" + }, + { + "id": 202, + "nodes": [ + 103, + 104, + 101 + ], + "tags": { + "highway": "secondary", + "lanes": "2", + "name": "Loop Avenue", + "oneway": "yes" + }, + "type": "way" + }, + { + "id": 203, + "nodes": [ + 102, + 105, + 104 + ], + "tags": { + "highway": "service", + "name": "Service Lane", + "service": "alley" + }, + "type": "way" + }, + { + "id": 204, + "nodes": [ + 104, + 105, + 102 + ], + "tags": { + "bicycle": "designated", + "highway": "cycleway", + "name": "Cycle Link" + }, + "type": "way" + } + ], + "generator": "OSMnx offline test fixture", + "osm3s": { + "copyright": "fixture", + "timestamp_osm_base": "2026-01-01T00:00:00Z" + }, + "version": 0.6 +} diff --git a/tests/input_data/http_cache/d37e6539fd80563459bfecfa63e7a4654eb9aa3a.json b/tests/input_data/http_cache/d37e6539fd80563459bfecfa63e7a4654eb9aa3a.json new file mode 100644 index 00000000..5c4e477f --- /dev/null +++ b/tests/input_data/http_cache/d37e6539fd80563459bfecfa63e7a4654eb9aa3a.json @@ -0,0 +1,87 @@ +{ + "elements": [ + { + "id": 301, + "lat": 37.791, + "lon": -122.41, + "tags": { + "amenity": "cafe", + "name": "Fixture Cafe" + }, + "type": "node" + }, + { + "id": 401, + "lat": 37.791, + "lon": -122.4108, + "type": "node" + }, + { + "id": 402, + "lat": 37.791, + "lon": -122.4098, + "type": "node" + }, + { + "id": 403, + "lat": 37.7918, + "lon": -122.4098, + "type": "node" + }, + { + "id": 404, + "lat": 37.7918, + "lon": -122.4108, + "type": "node" + }, + { + "id": 302, + "nodes": [ + 401, + 402, + 403, + 404, + 401 + ], + "tags": { + "building": "yes", + "name": "Fixture Building" + }, + "type": "way" + }, + { + "id": 303, + "nodes": [ + 401, + 402 + ], + "tags": { + "highway": "bus_stop", + "name": "Fixture Stop" + }, + "type": "way" + }, + { + "id": 304, + "members": [ + { + "ref": 302, + "role": "outer", + "type": "way" + } + ], + "tags": { + "landuse": "retail", + "name": "Fixture Retail", + "type": "multipolygon" + }, + "type": "relation" + } + ], + "generator": "OSMnx offline feature fixture", + "osm3s": { + "copyright": "fixture", + "timestamp_osm_base": "2026-01-01T00:00:00Z" + }, + "version": 0.6 +} diff --git a/tests/input_data/http_cache/deb786fd5711db743020090a375ad75807dede6c.json b/tests/input_data/http_cache/deb786fd5711db743020090a375ad75807dede6c.json new file mode 100644 index 00000000..5c4e477f --- /dev/null +++ b/tests/input_data/http_cache/deb786fd5711db743020090a375ad75807dede6c.json @@ -0,0 +1,87 @@ +{ + "elements": [ + { + "id": 301, + "lat": 37.791, + "lon": -122.41, + "tags": { + "amenity": "cafe", + "name": "Fixture Cafe" + }, + "type": "node" + }, + { + "id": 401, + "lat": 37.791, + "lon": -122.4108, + "type": "node" + }, + { + "id": 402, + "lat": 37.791, + "lon": -122.4098, + "type": "node" + }, + { + "id": 403, + "lat": 37.7918, + "lon": -122.4098, + "type": "node" + }, + { + "id": 404, + "lat": 37.7918, + "lon": -122.4108, + "type": "node" + }, + { + "id": 302, + "nodes": [ + 401, + 402, + 403, + 404, + 401 + ], + "tags": { + "building": "yes", + "name": "Fixture Building" + }, + "type": "way" + }, + { + "id": 303, + "nodes": [ + 401, + 402 + ], + "tags": { + "highway": "bus_stop", + "name": "Fixture Stop" + }, + "type": "way" + }, + { + "id": 304, + "members": [ + { + "ref": 302, + "role": "outer", + "type": "way" + } + ], + "tags": { + "landuse": "retail", + "name": "Fixture Retail", + "type": "multipolygon" + }, + "type": "relation" + } + ], + "generator": "OSMnx offline feature fixture", + "osm3s": { + "copyright": "fixture", + "timestamp_osm_base": "2026-01-01T00:00:00Z" + }, + "version": 0.6 +} diff --git a/tests/input_data/http_cache/f1f8610e6bb112def0a968fc6625bc5c51ca795b.json b/tests/input_data/http_cache/f1f8610e6bb112def0a968fc6625bc5c51ca795b.json new file mode 100644 index 00000000..b1a290f0 --- /dev/null +++ b/tests/input_data/http_cache/f1f8610e6bb112def0a968fc6625bc5c51ca795b.json @@ -0,0 +1,49 @@ +[ + { + "addresstype": "city", + "boundingbox": [ + "37.7860", + "37.7970", + "-122.4160", + "-122.4040" + ], + "class": "boundary", + "display_name": "Piedmont Fixture, California, USA", + "geojson": { + "coordinates": [ + [ + [ + -122.416, + 37.786 + ], + [ + -122.404, + 37.786 + ], + [ + -122.404, + 37.797 + ], + [ + -122.416, + 37.797 + ], + [ + -122.416, + 37.786 + ] + ] + ], + "type": "Polygon" + }, + "importance": 0.8, + "lat": "37.791427", + "lon": "-122.410018", + "name": "Piedmont Fixture", + "osm_id": 2999176, + "osm_type": "relation", + "place_id": 9002, + "place_rank": 16, + "type": "administrative" + } +] diff --git a/tests/input_data/http_cache/manifest.json b/tests/input_data/http_cache/manifest.json new file mode 100644 index 00000000..1642fc9d --- /dev/null +++ b/tests/input_data/http_cache/manifest.json @@ -0,0 +1,132 @@ +{ + "description": "Raw OSMnx HTTP cache fixtures for deterministic offline tests.", + "expected": { + "elevation_results": 5, + "feature_elements": 8, + "network_nodes": 5, + "network_ways": 4 + }, + "generated_by": "synthetic fixture generator; no public API calls", + "generated_on": "2026-05-09", + "records": [ + { + "cache_file": "be7271acafb60f59ad3647ef253f21fbed720a0a.json", + "endpoint": "https://nominatim.openstreetmap.org/search", + "label": "nominatim_geocode_address", + "query": "q=Transamerica Pyramid, 600 Montgomery Street, San Francisco, California, USA&format=json&limit=1", + "url_host": "nominatim.openstreetmap.org" + }, + { + "cache_file": "f1f8610e6bb112def0a968fc6625bc5c51ca795b.json", + "endpoint": "https://nominatim.openstreetmap.org/search", + "label": "nominatim_place_polygon", + "query": "city=Piedmont&state=California&country=USA&format=json&polygon_geojson=1&dedupe=0&limit=50", + "url_host": "nominatim.openstreetmap.org" + }, + { + "cache_file": "51a8a60e27ec040a17a10a2659d7946d80607df7.json", + "endpoint": "https://nominatim.openstreetmap.org/lookup", + "label": "nominatim_lookup_relation", + "query": "osm_ids=R2999176&format=json&polygon_geojson=1", + "url_host": "nominatim.openstreetmap.org" + }, + { + "cache_file": "ce6b1b54fd55e2c991a46e037d9038e4bc4fd308.json", + "endpoint": "https://nominatim.openstreetmap.org/search", + "label": "nominatim_geocode_empty", + "query": "q=!@#$%^&*&format=json&limit=1", + "url_host": "nominatim.openstreetmap.org" + }, + { + "cache_file": "0be0d6b34a9700ade60bc9f39b36639c0c37ca41.json", + "endpoint": "https://nominatim.openstreetmap.org/search", + "label": "nominatim_place_empty", + "query": "q=AAAZZZ&format=json&polygon_geojson=1&dedupe=0&limit=50", + "url_host": "nominatim.openstreetmap.org" + }, + { + "cache_file": "80be009ca881764187f03bf8c6aca285ef962424.json", + "endpoint": "https://nominatim.openstreetmap.org/search", + "label": "nominatim_nonpolygon_result", + "query": "q=Bunker Hill, Los Angeles, California, USA&format=json&polygon_geojson=1&dedupe=0&limit=50", + "url_host": "nominatim.openstreetmap.org" + }, + { + "cache_file": "d206c7ec7b8dd802e70c1fabbce4cc8dd5a23054.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_network_1", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "d206c7ec7b8dd802e70c1fabbce4cc8dd5a23054.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_network_2", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "d206c7ec7b8dd802e70c1fabbce4cc8dd5a23054.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_network_4", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "c41588e004167fdff982a269a2fcc476a61076de.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_network_5", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "8882c7a2619f4a40f1bc962a03d395703a4113d1.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_network_7", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "deb786fd5711db743020090a375ad75807dede6c.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_features_1", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "deb786fd5711db743020090a375ad75807dede6c.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_features_2", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "deb786fd5711db743020090a375ad75807dede6c.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_features_3", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "d37e6539fd80563459bfecfa63e7a4654eb9aa3a.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_features_4", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "d37e6539fd80563459bfecfa63e7a4654eb9aa3a.json", + "endpoint": "https://overpass-api.de/api/interpreter", + "label": "overpass_features_5", + "query": "data=", + "url_host": "overpass-api.de" + }, + { + "cache_file": "4fc0414e3f5d633d1740c3063d92207ce10bfaf4.json", + "endpoint": "https://api.opentopodata.org/v1/aster30m", + "label": "elevation_nodes", + "query": "locations=<5 committed graph fixture node coordinates>&key=None", + "url_host": "api.opentopodata.org" + } + ] +} diff --git a/tests/latest_lint_test.sh b/tests/latest_lint_test.sh index 23b57ccb..8d765b6e 100755 --- a/tests/latest_lint_test.sh +++ b/tests/latest_lint_test.sh @@ -4,7 +4,7 @@ set -euo pipefail # delete temp files and folders rm -r -f ./.coverage* ./.pytest_cache ./.temp ./dist ./docs/build ./*/__pycache__ -# activate the virtual environment with pre-releases +# activate the virtual environment with pre-release versions uv python pin 3.14 uv sync --all-extras --all-groups --upgrade --prerelease=allow source .venv/bin/activate @@ -13,7 +13,7 @@ source .venv/bin/activate SKIP=no-commit-to-branch prek run --all-files pytest --typeguard-packages=osmnx --cov=osmnx --cov-report=term-missing:skip-covered -# restore the environment without pre-releases +# restore the environment with current release versions uv python pin --rm uv sync --all-extras --all-groups --upgrade diff --git a/tests/minimal_lint_test.sh b/tests/minimal_lint_test.sh index 43a1241a..009c40e6 100755 --- a/tests/minimal_lint_test.sh +++ b/tests/minimal_lint_test.sh @@ -4,7 +4,7 @@ set -euo pipefail # delete temp files and folders rm -r -f ./.coverage* ./.pytest_cache ./.temp ./dist ./docs/build ./*/__pycache__ -# activate the virtual environment with pre-releases +# activate the virtual environment with minimal versions uv python pin 3.11 uv sync --all-extras --group test --upgrade --resolution=lowest-direct source .venv/bin/activate @@ -13,7 +13,7 @@ source .venv/bin/activate python ./tests/verify_min_deps.py pytest -W ignore -# restore the environment without pre-releases +# restore the environment with current release versions uv python pin --rm uv sync --all-extras --all-groups --upgrade diff --git a/tests/test_local_offline.py b/tests/test_local_offline.py new file mode 100755 index 00000000..8aee0fea --- /dev/null +++ b/tests/test_local_offline.py @@ -0,0 +1,1319 @@ +#!/usr/bin/env python +"""Offline CI tests for OSMnx.""" + +# ruff: noqa: D103, PLR2004, S101 + +from __future__ import annotations + +import bz2 +import gzip +import json +import logging as lg +import socket +import time as stdlib_time +from collections import OrderedDict +from pathlib import Path +from typing import cast + +import geopandas as gpd +import networkx as nx +import numpy as np +import pandas as pd +import pytest +import requests +from conftest import ADDRESS +from conftest import LOCATION_POINT +from conftest import PLACE +from conftest import TAGS +from conftest import Response +from conftest import drive_graph +from conftest import toy_graph +from lxml import etree +from shapely import LineString +from shapely import Point +from shapely import Polygon +from shapely import wkt +from typeguard import suppress_type_checks + +import osmnx as ox +from osmnx import _nominatim + +pytestmark = pytest.mark.offline + +# Sections: +# - Cached API workflows +# - HTTP/cache/request behavior +# - Graph creation, conversion, and file IO +# - Analysis, routing, distance, and plotting +# - Projection, geometry, validation, and simplification + +# Cached API workflows + + +def test_cache_fixture_manifest(http_cache: Path) -> None: + manifest = json.loads((http_cache / "manifest.json").read_text(encoding="utf-8")) + + assert manifest["expected"]["network_nodes"] == 5 + assert manifest["expected"]["network_ways"] == 4 + assert manifest["expected"]["feature_elements"] == 8 + assert len(manifest["records"]) >= 10 + assert all((http_cache / record["cache_file"]).is_file() for record in manifest["records"]) + assert all(record["endpoint"].startswith("https://") for record in manifest["records"]) + assert all(record["query"] for record in manifest["records"]) + + +def test_geocoder_uses_committed_cache( + http_cache: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ox.settings.cache_folder = http_cache + + city_by_id = ox.geocode_to_gdf("R2999176", by_osmid=True) + city_by_query = ox.geocode_to_gdf(PLACE, which_result=1) + city_list = ox.geocode_to_gdf([PLACE, PLACE], which_result=[1, 1]) + first_polygon = {"geojson": {"type": "Polygon"}} + + nonpolygon = { + "boundingbox": ["0", "1", "2", "3"], + "geojson": {"type": "Point", "coordinates": [2, 0]}, + "importance": 1, + "lat": "0", + "lon": "2", + "name": "Fixture Point", + } + + def _fake_download_nominatim_element( + *_args: object, + **_kwargs: object, + ) -> list[dict[str, object]]: + return [nonpolygon] + + with monkeypatch.context() as m: + m.setattr( + _nominatim, + "_download_nominatim_element", + _fake_download_nominatim_element, + ) + point_result = ox.geocoder._geocode_query_to_gdf( + "Fixture Point", + which_result=1, + by_osmid=False, + ) + city_projected = ox.projection.project_gdf(city_by_query, to_crs="epsg:3395") + + assert city_by_id.geometry.iloc[0].equals_exact(city_by_query.geometry.iloc[0], tolerance=0) + assert len(city_list) == 2 + assert point_result.geometry.iloc[0].geom_type == "Point" + assert ( + ox.geocoder._get_first_polygon([{"geojson": {"type": "Point"}}, first_polygon]) + == first_polygon + ) + assert city_by_query.loc[0, "name"] == "Piedmont Fixture" + assert city_projected.crs == "epsg:3395" + assert ox.geocode(ADDRESS) == LOCATION_POINT + + with pytest.raises(ox._errors.InsufficientResponseError): + ox.geocode("!@#$%^&*") + with pytest.raises(ox._errors.InsufficientResponseError): + ox.geocode_to_gdf(query="AAAZZZ") + with pytest.raises(TypeError, match="did not geocode"): + ox.geocode_to_gdf("Bunker Hill, Los Angeles, California, USA") + + +def test_graph_downloaders_use_committed_cache(http_cache: Path) -> None: + ox.settings.cache_folder = http_cache + + bbox = ox.utils_geo.bbox_from_point(LOCATION_POINT, dist=500) + polygon = ox.geocode_to_gdf(PLACE, which_result=1).geometry.iloc[0] + + graphs = [ + ox.graph_from_bbox(bbox, network_type="drive", simplify=False, retain_all=True), + ox.graph_from_point(LOCATION_POINT, dist=500, network_type="drive", retain_all=True), + ox.graph_from_address(ADDRESS, dist=500, network_type="drive", retain_all=True), + ox.graph_from_place(PLACE, network_type="all", which_result=1, retain_all=True), + ox.graph_from_polygon( + polygon, + network_type="walk", + simplify=False, + retain_all=True, + truncate_by_edge=True, + ), + ] + + for G in graphs: + ox.convert.validate_graph(G) + assert set(G.nodes) == {101, 102, 103, 104, 105} + assert G.graph["crs"] == ox.settings.default_crs + assert "Fixture Street" in { + data.get("name") for _, _, _, data in G.edges(keys=True, data=True) + } + + G_drive = graphs[0] + assert len(G_drive.edges) == 14 + assert dict(G_drive.nodes(data="street_count")) == {101: 2, 102: 4, 103: 2, 104: 4, 105: 4} + + +def test_features_downloaders_use_committed_cache(http_cache: Path) -> None: + ox.settings.cache_folder = http_cache + + bbox = ox.utils_geo.bbox_from_point(LOCATION_POINT, dist=500) + polygon = ox.geocode_to_gdf(PLACE, which_result=1).geometry.iloc[0] + gdfs = [ + ox.features_from_bbox(bbox, tags=TAGS), + ox.features_from_point(LOCATION_POINT, tags=TAGS, dist=500), + ox.features_from_address(ADDRESS, tags=TAGS, dist=500), + ox.features_from_place(PLACE, tags=TAGS, which_result=1), + ox.features_from_polygon(polygon, tags=TAGS), + ] + + expected_index = [("node", 301), ("relation", 304), ("way", 302), ("way", 303)] + for gdf in gdfs: + ox.convert.validate_features_gdf(gdf) + assert gdf.index.to_list() == expected_index + assert gdf.crs == ox.settings.default_crs + assert gdf.geometry.geom_type.to_list() == ["Point", "Polygon", "Polygon", "LineString"] + assert set(gdf["name"]) == { + "Fixture Cafe", + "Fixture Retail", + "Fixture Building", + "Fixture Stop", + } + + with pytest.raises(ValueError, match="The geometry of `polygon` is invalid"): + ox.features.features_from_polygon(Polygon(((0, 0), (0, 0), (0, 0), (0, 0))), tags={}) + with suppress_type_checks(), pytest.raises(TypeError): + ox.features.features_from_polygon(Point(0, 0), tags={}) + + +def test_elevation_from_cache_and_raster( + http_cache: Path, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + ox.settings.cache_folder = http_cache + G = drive_graph() + + with monkeypatch.context() as m: + + def _empty_elevation_response(_url: str, _pause: float) -> dict[str, list[object]]: + return {"results": []} + + m.setattr(ox.elevation, "_elevation_request", _empty_elevation_response) + with pytest.raises(ox._errors.InsufficientResponseError): + ox.elevation.add_node_elevations_google(G.copy(), api_key="", batch_size=350) + + ox.settings.elevation_url_template = ( + "https://api.opentopodata.org/v1/aster30m?locations={locations}&key={key}" + ) + G = ox.elevation.add_node_elevations_google(G, batch_size=100, pause=0) + assert dict(G.nodes(data="elevation")) == { + 101: 10.0, + 102: 11.0, + 103: 12.0, + 104: 13.0, + 105: 14.0, + } + G_grades = ox.add_edge_grades(G.copy(), add_absolute=True) + assert all("grade_abs" in data for _, _, data in G_grades.edges(data=True)) + + ox.settings.cache_folder = tmp_path / "raster-cache" + rasters = sorted(Path("tests/input_data").glob("elevation*.tif")) + G = ox.elevation.add_node_elevations_raster(G, rasters[0], cpus=1) + assert pd.notna(pd.Series(dict(G.nodes(data="elevation")))).any() + G = ox.elevation.add_node_elevations_raster(G, rasters[0], cpus=2) + assert pd.notna(pd.Series(dict(G.nodes(data="elevation")))).any() + G = ox.elevation.add_node_elevations_raster(G, rasters, cpus=1) + assert pd.notna(pd.Series(dict(G.nodes(data="elevation")))).sum() >= 2 + + +# HTTP/cache/request behavior + + +def test_request_validation_errors() -> None: + params: OrderedDict[str, int | str] = OrderedDict() + params["format"] = "json" + params["address_details"] = 0 + with pytest.raises(ValueError, match="Nominatim `request_type` must be"): + ox._nominatim._nominatim_request(params=params, request_type="xyz") + with pytest.raises(TypeError, match="`query` must be a string"): + ox.geocode_to_gdf(query={"City": "Boston"}, by_osmid=True) + + with pytest.raises(ValueError, match="Unrecognized network_type"): + ox._overpass._get_network_filter("not-real") + + +def test_http_cache_and_dns_helpers_are_deterministic( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + ox.settings.cache_folder = tmp_path + + assert ox._http._retrieve_from_cache("https://not-in-cache.example") is None + assert ox._http._parse_response(Response({"status": "error"}, ok=False, status_code=500)) == { + "status": "error", + } + + real_config_dns = ox._http._config_dns + original_getaddrinfo = socket.getaddrinfo + + def _fake_gethostbyname(_hostname: str) -> str: + return "127.0.0.1" + + def _fake_original_getaddrinfo(*_args: object, **_kwargs: object) -> list[object]: + return [] + + monkeypatch.setattr(socket, "gethostbyname", _fake_gethostbyname) + monkeypatch.setattr(ox._http, "_original_getaddrinfo", _fake_original_getaddrinfo) + try: + real_config_dns("https://example.com/api") + assert socket.getaddrinfo("example.com", 443) == [] + assert socket.getaddrinfo("other.example", 443) == [] + finally: + monkeypatch.setattr(socket, "getaddrinfo", original_getaddrinfo) + + +def test_overpass_status_and_query_helpers_parse_responses(monkeypatch: pytest.MonkeyPatch) -> None: + + def _sleep(_seconds: float) -> None: + return None + + class StatusResponse: + def __init__(self, text: str) -> None: + self.text = text + + status_responses = iter( + [ + StatusResponse("bad"), + StatusResponse("\n\n\n\nMysterious server status\n"), + StatusResponse("\n\n\n\nSlot available after: 2000-01-01T00:00:00Z,\n"), + StatusResponse("\n\n\n\nCurrently running a query\n"), + StatusResponse("\n\n\n\n2 slots available now.\n"), + ], + ) + + def _fake_status_get(*_args: object, **_kwargs: object) -> StatusResponse: + return next(status_responses) + + ox.settings.overpass_rate_limit = True + monkeypatch.setattr(stdlib_time, "sleep", _sleep) + monkeypatch.setattr(requests, "get", _fake_status_get) + assert ox._overpass._get_overpass_pause("https://overpass.example/api", default_pause=7) == 7 + assert ox._overpass._get_overpass_pause("https://overpass.example/api", default_pause=8) == 8 + assert ox._overpass._get_overpass_pause("https://overpass.example/api") == 1 + assert ox._overpass._get_overpass_pause("https://overpass.example/api") == 0 + + query = ox._overpass._create_overpass_features_query( + "0 0 0 1 1 1 0 0", + {"amenity": "cafe", "shop": ["books", "toys"]}, + ) + assert "amenity" in query + assert "books" in query + bad_tags = cast("dict[str, bool | str | list[str]]", {"amenity": [1]}) + with suppress_type_checks(), pytest.raises(TypeError, match="`tags` must be a dict"): + ox._overpass._create_overpass_features_query("0 0", bad_tags) + + payloads: list[str] = [] + + def _fake_overpass_request(data: OrderedDict[str, object]) -> dict[str, list[object]]: + payloads.append(str(data["data"])) + return {"elements": []} + + polygon = Polygon([(-1, -1), (1, -1), (1, 1), (-1, 1)]) + monkeypatch.setattr(ox._overpass, "_overpass_request", _fake_overpass_request) + responses_str = list(ox._overpass._download_overpass_network(polygon, "all", '["highway"]')) + responses_list = list( + ox._overpass._download_overpass_network( + polygon, + "all", + ['["railway"]', '["power"]'], + ), + ) + assert all(response == {"elements": []} for response in responses_str) + assert all(response == {"elements": []} for response in responses_list) + assert len(responses_list) == 2 * len(responses_str) + assert any("railway" in payload for payload in payloads) + + +def test_uncached_nominatim_request_paths( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + ox.settings.use_cache = False + ox.settings.cache_folder = tmp_path + + def _sleep(_seconds: float) -> None: + return None + + monkeypatch.setattr(stdlib_time, "sleep", _sleep) + + nominatim_responses = iter( + [ + Response([], ok=False, status_code=429), + Response([{"place_id": 1}]), + ], + ) + + def _fake_nominatim_get(*_args: object, **_kwargs: object) -> Response: + return next(nominatim_responses) + + ox.settings.nominatim_key = "fixture-key" + params: OrderedDict[str, int | str] = OrderedDict() + params["format"] = "json" + params["q"] = "fixture" + monkeypatch.setattr(requests, "get", _fake_nominatim_get) + assert ox._nominatim._nominatim_request(params=params) == [{"place_id": 1}] + assert params["key"] == "fixture-key" + + def _fake_nominatim_dict(*_args: object, **_kwargs: object) -> Response: + return Response({"not": "a-list"}) + + monkeypatch.setattr(requests, "get", _fake_nominatim_dict) + with pytest.raises(ox._errors.InsufficientResponseError, match="did not return a list"): + ox._nominatim._nominatim_request(params=OrderedDict(format="json", q="fixture")) + + +def test_uncached_overpass_and_elevation_request_paths( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + ox.settings.use_cache = False + ox.settings.cache_folder = tmp_path + + def _sleep(_seconds: float) -> None: + return None + + monkeypatch.setattr(stdlib_time, "sleep", _sleep) + + config_dns_calls: list[str] = [] + + def _fake_config_dns(url: str) -> None: + config_dns_calls.append(url) + + overpass_responses = iter( + [ + Response({"remark": "retry"}, ok=False, status_code=429), + Response({"elements": []}), + ], + ) + + def _fake_overpass_post(*_args: object, **_kwargs: object) -> Response: + return next(overpass_responses) + + ox.settings.overpass_rate_limit = False + monkeypatch.setattr(ox._http, "_config_dns", _fake_config_dns) + monkeypatch.setattr(requests, "post", _fake_overpass_post) + assert ox._overpass._overpass_request(OrderedDict(data="node(1);out;")) == {"elements": []} + assert config_dns_calls == [ox.settings.overpass_url, ox.settings.overpass_url] + + def _fake_overpass_list(*_args: object, **_kwargs: object) -> Response: + return Response([{"not": "a-dict"}]) + + monkeypatch.setattr(requests, "post", _fake_overpass_list) + with pytest.raises(ox._errors.InsufficientResponseError, match="did not return a dict"): + ox._overpass._overpass_request(OrderedDict(data="node(2);out;")) + + elevation_responses = iter( + [ + Response({"results": [{"elevation": 10.0}]}), + Response([{"not": "a-dict"}]), + ], + ) + + def _fake_elevation_get(*_args: object, **_kwargs: object) -> Response: + return next(elevation_responses) + + monkeypatch.setattr(requests, "get", _fake_elevation_get) + assert ox.elevation._elevation_request("https://example.com/elevation", pause=0) == { + "results": [{"elevation": 10.0}], + } + with pytest.raises(ox._errors.InsufficientResponseError, match="did not return a dict"): + ox.elevation._elevation_request("https://example.com/elevation?second", pause=0) + + +# Graph, GeoDataFrame, and file IO workflows + + +def test_stats_simplification_and_conversion(http_cache: Path) -> None: + ox.settings.cache_folder = http_cache + G = drive_graph() + G_proj = ox.project_graph(G) + G_proj = ox.project_graph(G_proj) + + stats = ox.basic_stats(G) + assert stats["n"] == 5 + assert stats["m"] == 14 + assert stats["streets_per_node_counts"] == {0: 0, 1: 0, 2: 2, 3: 0, 4: 3} + assert stats["edge_length_total"] == pytest.approx(1996.7794675) + + G_clean = ox.consolidate_intersections(G_proj, tolerance=10, rebuild_graph=True) + G_reconnected = ox.consolidate_intersections( + G_proj, + tolerance=10, + rebuild_graph=True, + reconnect_edges=False, + ) + centroids = ox.consolidate_intersections(G_proj, tolerance=10, rebuild_graph=False) + + assert len(G_clean) <= len(G_proj) + assert len(G_reconnected) <= len(G_proj) + assert len(centroids) <= len(G_proj) + + gdf_nodes, gdf_edges = ox.graph_to_gdfs(G) + G2 = ox.graph_from_gdfs(gdf_nodes, gdf_edges, graph_attrs=G.graph) + ox.convert.validate_graph(G2) + assert set(G2.nodes) == set(G.nodes) + assert set(G2.edges) == set(G.edges) + + D = ox.convert.to_digraph(G) + Gu = ox.convert.to_undirected(G) + assert isinstance(D, nx.DiGraph) + assert isinstance(Gu, nx.MultiGraph) + + +def test_save_load_graph_files(http_cache: Path) -> None: + ox.settings.cache_folder = http_cache + G = drive_graph() + ox.convert.validate_graph(G) + + ox.save_graph_geopackage(G, directed=False) + fp = Path(ox.settings.data_folder) / "graph-dir.gpkg" + ox.save_graph_geopackage(G, filepath=fp, directed=True) + gdf_nodes1 = gpd.read_file(fp, layer="nodes").set_index("osmid") + gdf_edges1 = gpd.read_file(fp, layer="edges").set_index(["u", "v", "key"]) + G2 = ox.graph_from_gdfs(gdf_nodes1, gdf_edges1, graph_attrs=G.graph) + gdf_nodes2, gdf_edges2 = ox.graph_to_gdfs(G2) + + assert set(gdf_nodes1.index) == set(gdf_nodes2.index) == set(G.nodes) + assert set(gdf_edges1.index) == set(gdf_edges2.index) == set(G.edges) + + with pytest.raises(ValueError, match="You must request nodes or edges or both"): + ox.graph_to_gdfs(G2, nodes=False, edges=False) + with pytest.raises(ValueError, match="Invalid literal for boolean"): + ox.io._convert_bool_string("T") + + attr_name = "test_bool" + G.graph[attr_name] = False + nx.set_node_attributes(G, {n: bool(i % 2) for i, n in enumerate(G.nodes)}, attr_name) + nx.set_edge_attributes(G, {e: bool(i % 2) for i, e in enumerate(G.edges)}, attr_name) + + graphml_fp = Path(ox.settings.data_folder) / "graph.graphml" + ox.save_graphml(G, gephi=True) + ox.save_graphml(G, gephi=False, filepath=graphml_fp) + G2 = ox.load_graphml( + graphml_fp, + graph_dtypes={attr_name: ox.io._convert_bool_string}, + node_dtypes={attr_name: ox.io._convert_bool_string}, + edge_dtypes={attr_name: ox.io._convert_bool_string}, + ) + ox.convert.validate_graph(G2) + assert set(G2.nodes) == set(G.nodes) + assert set(G2.edges) == set(G.edges) + + graphml = Path("tests/input_data/short.graphml").read_text(encoding="utf-8") + G3 = ox.load_graphml(graphml_str=graphml, node_dtypes={"osmid": str}) + assert len(G3) > 0 + + +def test_osm_xml_read_write(http_cache: Path, tmp_path: Path) -> None: + ox.settings.cache_folder = http_cache + node_id = 53098262 + neighbor_ids = 53092170, 53060438, 53027353, 667744075 + + path_bz2 = Path("tests/input_data/West-Oakland.osm.bz2") + file_contents = bz2.decompress(path_bz2.read_bytes()) + path_osm_temp = tmp_path / "West-Oakland.osm" + path_gz_temp = tmp_path / "West-Oakland.osm.gz" + path_osm_temp.write_bytes(file_contents) + path_gz_temp.write_bytes(gzip.compress(file_contents)) + + for filepath in (path_bz2, path_gz_temp, path_osm_temp): + G = ox.graph_from_xml(filepath) + ox.convert.validate_graph(G, strict=False) + assert node_id in G.nodes + for neighbor_id in neighbor_ids: + edge_key = (node_id, neighbor_id, 0) + assert neighbor_id in G.nodes + assert edge_key in G.edges + assert G.edges[edge_key]["name"] in {"8th Street", "Willow Street"} + + default_all_oneway = ox.settings.all_oneway + ox.settings.all_oneway = True + G = drive_graph() + fp = Path(ox.settings.data_folder) / "graph.osm" + ox.io.save_graph_xml(G, filepath=fp, way_tag_aggs={"lanes": "sum"}) + + parser = etree.XMLParser(schema=etree.XMLSchema(file="./tests/input_data/osm_schema.xsd")) + _ = etree.parse(fp, parser=parser) + + with pytest.raises(ox._errors.GraphSimplificationError): + ox.io.save_graph_xml(ox.simplification.simplify_graph(G)) + ox.settings.all_oneway = default_all_oneway + + +def test_graph_creation_validates_inputs(http_cache: Path) -> None: + ox.settings.cache_folder = http_cache + G_network = ox.graph_from_point( + LOCATION_POINT, + dist=500, + dist_type="network", + network_type="drive", + retain_all=True, + ) + assert len(G_network) > 0 + + bbox = ox.utils_geo.bbox_from_point(LOCATION_POINT, dist=500) + G_largest = ox.graph_from_bbox(bbox, network_type="drive", simplify=False, retain_all=False) + ox.convert.validate_graph(G_largest) + + with pytest.raises(ValueError, match="`dist_type` must be"): + ox.graph_from_point(LOCATION_POINT, dist=500, dist_type="bad") + with suppress_type_checks(), pytest.raises(TypeError, match="Geometry must be"): + ox.graph_from_polygon(Point(0, 0)) + with pytest.raises(ValueError, match="geometry of `polygon` is invalid"): + ox.graph_from_polygon(Polygon(((0, 0), (0, 0), (0, 0), (0, 0)))) + + ox.settings.cache_only_mode = True + with pytest.raises(ox._errors.CacheOnlyInterruptError, match="Interrupted because"): + ox.graph._create_graph([{"elements": []}], bidirectional=False) + ox.settings.cache_only_mode = False + with pytest.raises(ox._errors.InsufficientResponseError, match="No data elements"): + ox.graph._create_graph([{"elements": []}], bidirectional=False) + + assert ox.graph._is_path_one_way( + {"junction": "roundabout"}, + bidirectional=False, + oneway_values=set(), + ) + + G = nx.MultiDiGraph() + G.add_nodes_from([1, 2]) + path = {"osmid": 1, "nodes": [1, 2], "oneway": "-1", "highway": "residential"} + ox.graph._add_paths(G, [path], bidirectional=False) + assert (2, 1, 0) in G.edges + + +def test_conversion_and_graphml_validate_inputs(tmp_path: Path) -> None: + empty = nx.MultiDiGraph(crs="epsg:4326") + with pytest.raises(ValueError, match="contains no nodes"): + ox.graph_to_gdfs(empty, nodes=True, edges=False) + with pytest.raises(ValueError, match="contains no edges"): + ox.graph_to_gdfs(empty, nodes=False, edges=True) + + gdf_nodes = gpd.GeoDataFrame({"x": [0.0, 1.0], "y": [0.0, 1.0]}, index=[1, 2]) + gdf_edges = gpd.GeoDataFrame( + {"osmid": [1], "length": [1.0], "geometry": [LineString([(0, 0), (1, 1)])]}, + index=pd.MultiIndex.from_tuples([(1, 2, 0)], names=["u", "v", "key"]), + crs="epsg:4326", + ) + G = ox.graph_from_gdfs(gdf_nodes, gdf_edges) + assert G.graph["crs"] == "epsg:4326" + + with pytest.raises(ValueError, match="one and only one"): + ox.load_graphml() + with pytest.raises(ValueError, match="one and only one"): + ox.load_graphml(tmp_path / "missing.graphml", graphml_str="") + assert ox.io._convert_bool_string(value=True) + + G_types = nx.MultiDiGraph(consolidated="False", simplified="True") + G_types.add_node(1, x="0", y="0", osmid="1", street_count="1", tags="[1, 2]") + G_types.add_node(2, x="1", y="1", osmid="2", street_count="1") + G_types.add_edge( + 1, + 2, + osmid="[10, 11]", + length="1.5", + oneway="False", + reversed="True", + geometry="LINESTRING (0 0, 1 1)", + ) + ox.io._convert_graph_attr_types(G_types, {"consolidated": ox.io._convert_bool_string}) + ox.io._convert_node_attr_types(G_types, {"osmid": int, "street_count": int, "x": float}) + ox.io._convert_edge_attr_types( + G_types, + {"osmid": int, "length": float, "oneway": ox.io._convert_bool_string}, + ) + assert G_types.edges[1, 2, 0]["osmid"] == [10, 11] + + G_parallel = toy_graph() + G_parallel.add_edge( + 2, + 1, + osmid=10, + length=1.0, + highway="residential", + geometry=LineString([(1, 0), (0, 0)]), + ) + G_parallel.add_edge( + 1, + 2, + key=1, + osmid=12, + length=1.0, + highway="service", + geometry=LineString([(0, 0), (0.5, 0.2), (1, 0)]), + ) + G_parallel.add_edge( + 2, + 1, + key=1, + osmid=12, + length=1.0, + highway="service", + geometry=LineString([(1, 0), (0.5, -0.2), (0, 0)]), + ) + Gu = ox.convert.to_undirected(G_parallel) + assert isinstance(Gu, nx.MultiGraph) + + G_duplicate = nx.MultiDiGraph(crs="epsg:4326") + G_duplicate.add_node(1, x=0, y=0) + G_duplicate.add_node(2, x=1, y=1) + G_duplicate.add_edge(1, 2, key=0, osmid=1) + G_duplicate.add_edge(2, 1, key=1, osmid=1) + Gu_duplicate = ox.convert.to_undirected(G_duplicate) + assert len(Gu_duplicate.edges) == 1 + + +def test_osm_xml_roundtrip_warns_for_projected_graphs(tmp_path: Path) -> None: + G = nx.MultiDiGraph(crs="epsg:3857") + G.add_node(1, x=0.0, y=0.0, street_count=1) + G.add_node(2, x=1.0, y=0.0, street_count=2) + G.add_node(3, x=2.0, y=0.0, street_count=1) + G.add_node(1, x=0.0, y=0.0, street_count=1, uid=None) + G.add_edge(1, 2, osmid=10, length=1.0, highway="residential", uid=None) + G.add_edge(2, 3, osmid=11, length=1.0, highway="primary") + fp = tmp_path / "projected.osm" + with pytest.warns(UserWarning, match="all_oneway=True|unprojected"): + ox.io.save_graph_xml(G, filepath=fp) + with pytest.warns(UserWarning, match="generated by OSMnx"): + G_loaded = ox.graph_from_xml(fp, simplify=False, retain_all=True) + assert len(G_loaded) > 0 + + G_source_cycle = nx.MultiDiGraph([(1, 2), (2, 3), (3, 1), (1, 4)]) + assert set(ox._osm_xml._sort_nodes(G_source_cycle, osmid=1)) >= {1, 2, 3, 4} + + G_target_cycle = nx.MultiDiGraph([(1, 2), (2, 3), (3, 1), (4, 1)]) + assert set(ox._osm_xml._sort_nodes(G_target_cycle, osmid=2)) >= {1, 2, 3, 4} + + G_multi_cycle = nx.MultiDiGraph([(1, 2), (2, 3), (3, 1), (3, 4), (4, 2)]) + assert len(ox._osm_xml._sort_nodes(G_multi_cycle, osmid=3)) > 0 + G_simple_cycle = nx.MultiDiGraph([(1, 2), (2, 3), (3, 1)]) + assert len(ox._osm_xml._sort_nodes(G_simple_cycle, osmid=4)) > 0 + + +# Analysis, routing, distance, and plotting workflows + + +def test_logging_and_utils(tmp_path: Path) -> None: + ox.settings.log_console = True + ox.settings.log_file = True + ox.settings.logs_folder = tmp_path + + ox.utils.log("test a fake default message") + ox.utils.log("test a fake debug", level=lg.DEBUG) + ox.utils.log("test a fake info", level=lg.INFO) + ox.utils.log("test a fake warning", level=lg.WARNING) + ox.utils.log("test a fake error", level=lg.ERROR) + + ox.utils.citation(style="apa") + ox.utils.citation(style="bibtex") + ox.utils.citation(style="ieee") + + assert "T" in ox.utils.ts(style="iso8601") + assert len(ox.utils.ts(style="date")) == 10 + assert len(ox.utils.ts(style="time")) == 8 + assert ox.settings.logs_folder == tmp_path + + +def test_bearings_routing_and_nearest(http_cache: Path) -> None: + ox.settings.cache_folder = http_cache + G = drive_graph() + + assert ox.bearing.calculate_bearing(0, 0, 1, 1) == pytest.approx(44.99563646) + + G = ox.add_edge_bearings(G) + with pytest.warns(UserWarning, match="directional"): + bearings, weights = ox.bearing._extract_edge_bearings(G, min_length=0, weight=None) + assert len(bearings) == len(weights) == len(G.edges) + Gu = ox.convert.to_undirected(G) + entropy = ox.bearing.orientation_entropy(Gu, weight="length") + assert entropy == pytest.approx(1.777310166) + + _, ax = ox.plot.plot_orientation(Gu, area=True, title="Title") + _, _ = ox.plot.plot_orientation(Gu, ax=ax, area=False, title="Title") + + G = ox.add_edge_speeds(G) + G = ox.add_edge_speeds(G, hwy_speeds={"motorway": 100}) + G = ox.add_edge_travel_times(G) + route = ox.shortest_path(G, 101, 103, weight="travel_time") + assert route == [101, 102, 103] + assert ox.routing.route_to_gdf(G, route, weight="travel_time").index.to_list() == [ + (101, 102, 0), + (102, 103, 0), + ] + + assert ox.routing._clean_maxspeed("100,2") == 100.2 + assert ox.routing._clean_maxspeed("100 mph") == pytest.approx(160.934) + assert ox.routing._clean_maxspeed("signal") is None + assert ox.routing._collapse_multiple_maxspeed_values(["25 mph", "30 mph"], np.mean) == 44.25685 + paths1 = ox.shortest_path(G, [101, 102], [103, 104], weight="length", cpus=1) + paths2 = ox.shortest_path(G, [101, 102], [103, 104], weight="length", cpus=2) + assert paths1 == paths2 == [[101, 102, 103], [102, 105, 104]] + + routes = list(ox.routing.k_shortest_paths(G, 101, 103, k=2, weight="travel_time")) + assert routes[0] == [101, 102, 103] + + assert ox.distance.great_circle(0, 0, 1, 1) == pytest.approx(157249.6034105) + assert ox.distance.euclidean(0, 0, 1, 1) == pytest.approx(1.4142135) + assert ox.distance.nearest_nodes(G, -122.4100, 37.79125) == 105 + + Gp = ox.project_graph(G) + points = ox.utils_geo.sample_points(ox.convert.to_undirected(Gp), 5) + X = points.x.to_numpy() + Y = points.y.to_numpy() + assert len(ox.distance.nearest_nodes(Gp, X, Y, return_dist=True)[0]) == 5 + nearest_edges = ox.distance.nearest_edges(Gp, X, Y, return_dist=False) + assert len(nearest_edges) == 5 + assert len(nearest_edges[0]) == 3 + + +def test_plots_and_colors(http_cache: Path) -> None: + ox.settings.cache_folder = http_cache + G = drive_graph() + Gp = ox.project_graph(G) + + colors = ox.plot.get_colors(n=5, cmap="plasma", start=0.1, stop=0.9, alpha=0.5) + node_colors = ox.plot.get_node_colors_by_attr(G, "x") + edge_colors = ox.plot.get_edge_colors_by_attr(G, "length", num_bins=5) + + assert len(colors) == 5 + assert len(node_colors) == len(G) + assert len(edge_colors) == len(G.edges) + + filepath = Path(ox.settings.data_folder) / "test.svg" + _, ax = ox.plot_graph(G, show=False, save=True, close=True, filepath=filepath) + _, ax = ox.plot_graph(Gp, edge_linewidth=0, figsize=(5, 5), bgcolor="y") + _, _ = ox.plot_graph( + Gp, + ax=ax, + dpi=180, + node_color="k", + node_size=5, + node_alpha=0.1, + node_edgecolor="b", + node_zorder=5, + edge_color="r", + edge_linewidth=2, + edge_alpha=0.1, + show=False, + save=True, + close=True, + ) + _, _ = ox.plot_figure_ground(G=G) + _, _ = ox.plot_graph_route(G, [101, 102, 103], save=True) + _, _ = ox.plot_graph_routes(G, [[101, 102, 103], [101, 102, 105, 104]]) + + assert filepath.is_file() + + +def test_utils_geo_projection_and_http_helpers(tmp_path: Path) -> None: + geom = ox.utils_geo.buffer_geometry(Point(-122.41, 37.79), dist=100) + bbox = ox.utils_geo.bbox_from_point(LOCATION_POINT, dist=500, project_utm=True) + bbox_with_crs = ox.utils_geo.bbox_from_point( + LOCATION_POINT, + dist=500, + project_utm=True, + return_crs=True, + ) + poly = ox.utils_geo.bbox_to_poly(ox.utils_geo.bbox_from_point(LOCATION_POINT, dist=500)) + subdivided = ox.utils_geo._consolidate_subdivide_geometry(poly) + + assert geom.area > 0 + assert len(bbox) == 4 + assert len(bbox_with_crs) == 2 + assert len(subdivided.geoms) == 1 + assert list(ox.utils_geo.interpolate_points(LineString([(0, 0), (1, 0)]), 0.25)) == [ + (0.0, 0.0), + (0.25, 0.0), + (0.5, 0.0), + (0.75, 0.0), + (1.0, 0.0), + ] + + assert ox.projection.is_projected("epsg:3857") + assert not ox.projection.is_projected("epsg:4326") + geom_proj, crs_proj = ox.projection.project_geometry(poly) + assert ox.projection.is_projected(crs_proj) + geom_latlon, crs_latlon = ox.projection.project_geometry( + geom_proj, + crs=crs_proj, + to_latlong=True, + ) + assert crs_latlon == ox.settings.default_crs + assert geom_latlon.area == pytest.approx(poly.area, rel=0.01) + + ox.settings.cache_folder = tmp_path + ox._http._save_to_cache("https://example.com/test", {"ok": True}, ok=True) + assert ox._http._retrieve_from_cache("https://example.com/test") == {"ok": True} + assert ox._http._hostname_from_url("https://example.com:443/api") == "example.com" + assert "User-Agent" in ox._http._get_http_headers(user_agent="test-agent") + + +def test_routing_and_http_helpers_validate_inputs_and_statuses( + http_cache: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ox.settings.cache_folder = http_cache + G = drive_graph() + + G_dist = ox.truncate.truncate_graph_dist(G, 101, 150) + assert set(G_dist.nodes).issubset(G.nodes) + assert len(G_dist) < len(G) + + bbox = (-122.412, 37.790, -122.410, 37.793) + G_bbox = ox.truncate.truncate_graph_bbox(G, bbox, truncate_by_edge=True) + assert 101 in G_bbox.nodes + with pytest.raises(ValueError, match="Found no graph nodes"): + ox.truncate.truncate_graph_bbox(G, (0, 0, 1, 1)) + + G_disconnected = G.copy() + G_disconnected.add_node(999, x=0, y=0, street_count=0) + G_largest = ox.truncate.largest_component(G_disconnected) + assert 999 not in G_largest.nodes + + with pytest.raises(TypeError, match="must either both be iterable"): + ox.shortest_path(G, 101, [103]) # type: ignore[call-overload] + with pytest.raises(ValueError, match="must be of equal length"): + ox.shortest_path(G, [101], [102, 103]) + + G_no_speed = G.copy() + for _, _, _, data in G_no_speed.edges(keys=True, data=True): + data.pop("maxspeed", None) + with pytest.raises(ValueError, match="must pass `hwy_speeds` or `fallback`"): + ox.routing.add_edge_speeds(G_no_speed) + + ox.settings.doh_url_template = None + assert ox._http._resolve_host_via_doh("example.com") == "example.com" + ox.settings.doh_url_template = "https://dns.example.test/{hostname}" + + doh_response = Response({"Status": 0, "Answer": [{"data": "192.0.2.1"}]}) + + def _fake_doh_get(*_args: object, **_kwargs: object) -> Response: + return doh_response + + monkeypatch.setattr(requests, "get", _fake_doh_get) + assert ox._http._resolve_host_via_doh("example.com") == "192.0.2.1" + + doh_response = Response({"Status": 1}) + assert ox._http._resolve_host_via_doh("example.com") == "example.com" + + class _StatusResponse: + text = "\n\n\n\n2 slots available now.\n" + + def _fake_status_get(*_args: object, **_kwargs: object) -> _StatusResponse: + return _StatusResponse() + + ox.settings.overpass_rate_limit = True + monkeypatch.setattr(requests, "get", _fake_status_get) + assert ox._overpass._get_overpass_pause("https://overpass.example.test/api") == 0 + + +def test_projection_stats_distance_and_geometry_behaviors() -> None: + gdf_north = gpd.GeoDataFrame(geometry=[Point(0, 85.1)], crs="epsg:4326") + gdf_south = gpd.GeoDataFrame(geometry=[Point(0, -84.1)], crs="epsg:4326") + assert ox.projection.project_gdf(gdf_north).crs == "epsg:32661" + assert ox.projection.project_gdf(gdf_south).crs == "epsg:32761" + with pytest.raises(ValueError, match="valid CRS"): + ox.projection.project_gdf(gpd.GeoDataFrame(geometry=[])) + + G_simple = nx.MultiDiGraph(crs="epsg:4326") + G_simple.add_node(1, x=0.0, y=0.0, street_count=1) + G_simple.add_node(2, x=0.01, y=0.0, street_count=2) + G_simple.add_node(3, x=0.02, y=0.0, street_count=1) + G_simple.add_edge(1, 2, osmid=1, length=1.0) + G_simple.add_edge(2, 3, osmid=2, length=1.0) + G_simple = ox.simplification.simplify_graph(G_simple) + G_projected = ox.project_graph(G_simple) + assert ox.project_graph(G_projected, to_latlong=True).graph["crs"] == ox.settings.default_crs + + G_stats = toy_graph(crs="epsg:3857") + G_missing_count = G_stats.copy() + del G_missing_count.nodes[1]["street_count"] + assert set(ox.stats.streets_per_node(G_missing_count)) != set(G_missing_count.nodes) + Gu = ox.convert.to_undirected(G_stats) + assert ox.stats.circuity_avg(Gu) == pytest.approx(1.0) + assert ox.stats.count_streets_per_node(G_stats) == {1: 1, 2: 2, 3: 1} + stats = ox.basic_stats(G_stats, area=1000, clean_int_tol=1) + assert "clean_intersection_density_km" in stats + + G_bad = nx.MultiDiGraph(crs="epsg:4326") + G_bad.add_nodes_from([(1, {"x": np.nan, "y": 0}), (2, {"x": 1, "y": 1})]) + G_bad.add_edge(1, 2) + with pytest.raises(ValueError, match="possibly due to input data clipping"): + ox.distance.add_edge_lengths(G_bad) + with pytest.raises(ValueError, match="cannot contain nulls"): + ox.distance.nearest_nodes(G_stats, [np.nan], [0]) + with pytest.raises(ValueError, match="cannot contain nulls"): + ox.distance.nearest_edges(G_stats, [0], [np.nan]) + assert ox.distance.nearest_nodes(G_stats, 0, 0, return_dist=True)[0] == 1 + assert ox.distance.nearest_nodes(G_stats, [0], [0], return_dist=False)[0] == 1 + assert ox.distance.nearest_edges(G_stats, 0, 0, return_dist=True)[0] == (1, 2, 0) + assert ox.distance.nearest_edges(G_stats, 0, 0, return_dist=False) == (1, 2, 0) + assert len(ox.distance.nearest_edges(G_stats, [0], [0], return_dist=True)[0]) == 1 + + with pytest.warns(UserWarning, match="undirected"): + assert len(ox.utils_geo.sample_points(G_stats, 1)) == 1 + with suppress_type_checks(), pytest.raises(TypeError, match="LineString"): + list(ox.utils_geo.interpolate_points(Point(0, 0), 1)) + with suppress_type_checks(), pytest.raises(TypeError, match="Geometry must be"): + ox.utils_geo._consolidate_subdivide_geometry(Point(0, 0)) + + assert ox.simplification._build_path(nx.MultiDiGraph([(1, 2), (2, 1)]), 1, 2, {1}) == [ + 1, + 2, + ] + empty_graph = nx.MultiDiGraph(crs="epsg:3857") + assert len(ox.consolidate_intersections(empty_graph, rebuild_graph=True)) == 0 + assert len(ox.consolidate_intersections(empty_graph, rebuild_graph=False)) == 0 + assert ( + len( + ox.consolidate_intersections( + G_stats, + tolerance={1: 0.5, 2: 0.5}, + rebuild_graph=False, + dead_ends=True, + ), + ) + > 0 + ) + + +def test_truncation_plotting_and_routing_errors(tmp_path: Path) -> None: + G_trunc = nx.MultiDiGraph(crs="epsg:4326") + G_trunc.add_node(1, x=0.0, y=0.0) + G_trunc.add_node(2, x=2.0, y=0.0) + G_trunc.add_node(3, x=4.0, y=0.0) + G_trunc.add_edges_from([(1, 2), (2, 3)]) + polygon = Polygon([(-1, -1), (1, -1), (1, 1), (-1, 1)]) + G_truncated = ox.truncate.truncate_graph_polygon(G_trunc, polygon, truncate_by_edge=True) + assert set(G_truncated.nodes) == {1, 2} + + G_strong = nx.MultiDiGraph(crs="epsg:4326") + G_strong.add_edges_from([(1, 2), (2, 1), (3, 4)]) + assert set(ox.truncate.largest_component(G_strong, strongly=True).nodes) == {1, 2} + + G_plot = toy_graph() + G_plot.add_node(99, x=0.0, y=1.0, street_count=0) + _, _ = ox.plot_graph_route(G_plot, [1, 2, 3], show=False, close=True) + _, _ = ox.plot_figure_ground(G_plot, show=False, close=True) + footprints = gpd.GeoDataFrame( + geometry=[Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])], + crs=ox.settings.default_crs, + ) + _, _ = ox.plot_footprints( + footprints, + show=False, + close=True, + save=True, + filepath=tmp_path / "footprints.png", + ) + color_values = pd.Series([1.0, 2.0, 3.0, 4.0]) + assert ( + len( + ox.plot._get_colors_by_value( + color_values, + num_bins=2, + cmap="viridis", + start=0, + stop=1, + na_color="none", + equal_size=True, + ), + ) + == 4 + ) + with pytest.raises(ValueError, match="no attribute values"): + ox.plot._get_colors_by_value( + pd.Series(dtype=float), + num_bins=None, + cmap="viridis", + start=0, + stop=1, + na_color="none", + equal_size=False, + ) + + G_route = toy_graph() + G_route.add_node(4, x=10.0, y=10.0, street_count=0) + assert ox.shortest_path(G_route, 1, 4) is None + assert ox.shortest_path(G_route, [1], [3], cpus=None) == [[1, 2, 3]] + assert ox.routing._collapse_multiple_maxspeed_values(["10"], lambda _values: "bad") is None + assert ox.routing._collapse_multiple_maxspeed_values(["mph", "kph"], np.mean) is None + assert ox.routing._collapse_multiple_maxspeed_values(["signal"], np.mean) is None + + with pytest.raises(ValueError, match="Invalid citation style"): + ox.utils.citation(style="bad") + with pytest.raises(ValueError, match="Invalid timestamp style"): + ox.utils.ts(style="bad") + + +# Geometry, validation, and simplification workflows + + +def test_validation_errors() -> None: + G = nx.MultiDiGraph() + G.add_edge(0, 1) + with pytest.raises(ox._errors.ValidationError): + ox._validate._verify_numeric_edge_attribute(G, "length", strict=True) + + with pytest.raises(ox._errors.ValidationError): + ox.convert.validate_features_gdf(gpd.GeoDataFrame(index=[0, 0])) + + gdf_nodes = pd.DataFrame(index=[0, 0]) + gdf_edges = pd.DataFrame() + with suppress_type_checks(), pytest.raises(ox._errors.ValidationError): + ox.convert.validate_node_edge_gdfs(gdf_nodes, gdf_edges) + + gdf_nodes = gpd.GeoDataFrame(geometry=[Polygon(), Polygon()]) + gdf_edges = gpd.GeoDataFrame() + with pytest.raises(ox._errors.ValidationError): + ox.convert.validate_node_edge_gdfs(gdf_nodes, gdf_edges) + + data = {"x": [0, 1], "y": [2, 3]} + gdf_nodes = gpd.GeoDataFrame(data=data, geometry=[Point((6, 7)), Point((8, 9))]) + with pytest.raises(ox._errors.ValidationError): + ox.convert.validate_node_edge_gdfs(gdf_nodes, gdf_edges) + + G = nx.Graph() + with suppress_type_checks(), pytest.raises(ox._errors.ValidationError): + ox.convert.validate_graph(G) + + G = nx.MultiDiGraph() + del G.graph + G.add_edge("0", "1") + with pytest.raises(ox._errors.ValidationError): + ox.convert.validate_graph(G) + + G = nx.MultiDiGraph() + G.graph["crs"] = "epsg:4326" + G.add_edge(0, 1) + with pytest.raises(ox._errors.ValidationError): + ox.convert.validate_graph(G) + + nx.set_node_attributes(G, values=0, name="x") + nx.set_node_attributes(G, values=0, name="y") + nx.set_node_attributes(G, values=0, name="street_count") + nx.set_edge_attributes(G, values=[0], name="osmid") + nx.set_edge_attributes(G, values=1.5, name="length") + ox.convert.validate_graph(G) + + +def test_validation_reports_bad_graph_attributes() -> None: + G = nx.MultiDiGraph(crs="epsg:4326") + G.add_node("a", x="bad", y=0) + G.add_node("b", x=1, y="bad") + G.add_edge("a", "b", osmid="bad", length="bad") + + with pytest.raises(ox._errors.ValidationError, match="contains non-numeric"): + ox._validate._verify_numeric_edge_attribute(G, "length", strict=True) + + with pytest.raises(ox._errors.ValidationError, match="Node 'x' and 'y'"): + ox.convert.validate_graph(G) + + with pytest.warns(UserWarning, match="Node 'x' and 'y'"): + ox.convert.validate_graph(G, strict=False) + + G_nan = nx.MultiDiGraph(crs="epsg:4326") + G_nan.add_node(1, x=0, y=0, street_count=1) + G_nan.add_node(2, x=1, y=1, street_count=1) + G_nan.add_edge(1, 2, osmid=1, length=np.nan) + with pytest.raises(ox._errors.ValidationError, match="missing or null"): + ox._validate._verify_numeric_edge_attribute(G_nan, "length", strict=True) + with pytest.warns(UserWarning, match="missing or null"): + ox._validate._verify_numeric_edge_attribute(G_nan, "length", strict=False) + + G_bad_crs = nx.MultiDiGraph(crs="epsg:999999") + G_bad_crs.add_node(1, x=0, y=0, street_count=1) + G_bad_crs.add_node(2, x=1, y=1, street_count=1) + G_bad_crs.add_edge(1, 2, osmid=1, length=1.0) + with pytest.raises(ox._errors.ValidationError, match="valid CRS"): + ox.convert.validate_graph(G_bad_crs) + + gdf_nodes = gpd.GeoDataFrame( + {"x": [0.0, 1.0], "y": [0.0, 1.0]}, + geometry=[Point(0, 0), Point(1, 1)], + index=[1, 2], + crs="epsg:4326", + ) + gdf_edges = gpd.GeoDataFrame( + {"osmid": [1], "length": [1.0], "geometry": [LineString([(0, 0), (1, 1)])]}, + index=pd.MultiIndex.from_tuples([(1, 2, 0)], names=["u", "v", "key"]), + crs="epsg:4326", + ) + ox.convert.validate_node_edge_gdfs(gdf_nodes, gdf_edges) + + +def test_features_from_xml_and_polygon_holes() -> None: + gdf = ox.features_from_xml("tests/input_data/planet_10.068,48.135_10.071,48.137.osm") + assert len(gdf) > 0 + + gdf = ox.features_from_xml("tests/input_data/West-Oakland.osm.bz2") + assert "Willow Street" in gdf["name"].to_numpy() + + outer1 = Polygon(((0, 0), (4, 0), (4, 4), (0, 4))) + inner1 = Polygon(((1, 1), (2, 1), (2, 3), (1, 3))) + inner2 = Polygon(((2, 1), (3, 1), (3, 3), (2, 3))) + outer2 = Polygon(((1.5, 1.5), (2.5, 1.5), (2.5, 2.5), (1.5, 2.5))) + result = ox.features._remove_polygon_holes([outer1, outer2], [inner1, inner2]) + geom_wkt = ( + "MULTIPOLYGON (((4 4, 4 0, 0 0, 0 4, 4 4), " + "(3 1, 3 3, 2 3, 1 3, 1 1, 2 1, 3 1)), " + "((2.5 2.5, 2.5 1.5, 1.5 1.5, 1.5 2.5, 2.5 2.5)))" + ) + assert result.equals(wkt.loads(geom_wkt)) + + +def test_simplification_preserves_merged_edge_attrs(monkeypatch: pytest.MonkeyPatch) -> None: + G = nx.MultiDiGraph(crs="epsg:4326") + G.add_node(1, x=0.0, y=0.0, street_count=1) + G.add_node(2, x=1.0, y=0.0, street_count=2) + G.add_node(3, x=2.0, y=0.0, street_count=1) + G.add_edge(1, 2, osmid=1, length=1.0, travel_time=1.0, name="a") + G.add_edge(2, 3, osmid=2, length=1.0, travel_time=1.0, name="b") + G.add_edge(2, 3, osmid=3, length=2.0, travel_time=2.0, name="c") + + def _fake_paths( + _G: nx.MultiDiGraph, + _node_attrs_include: object, + _edge_attrs_differ: object, + ) -> list[list[int]]: + return [[1, 2, 3]] + + monkeypatch.setattr(ox.simplification, "_get_paths_to_simplify", _fake_paths) + Gs = ox.simplification.simplify_graph(G, track_merged=True) + edge_data = next(iter(Gs.edges(data=True)))[2] + assert edge_data["length"] == 2.0 + assert edge_data["travel_time"] == 2.0 + assert edge_data["merged_edges"] == [(1, 2), (2, 3)] + + Gs.graph["simplified"] = True + with pytest.raises(ox._errors.GraphSimplificationError, match="already been simplified"): + ox.simplification.simplify_graph(Gs) + + H = nx.MultiDiGraph() + H.add_node(1) + H.add_node(2, highway="traffic_signals") + H.add_node(3) + H.add_edges_from([(1, 2, {"osmid": 1}), (2, 1, {"osmid": 1})]) + H.add_edges_from([(2, 3, {"osmid": 2}), (3, 2, {"osmid": 3})]) + assert ox.simplification._is_endpoint(H, 2, ["highway"], None) + assert ox.simplification._is_endpoint(H, 2, None, ["osmid"]) + + B = nx.MultiDiGraph([(1, 2), (2, 3), (3, 1)]) + assert ox.simplification._build_path(B, 1, 2, {1}) == [1, 2, 3, 1] + C = nx.MultiDiGraph([(1, 2), (2, 3)]) + assert ox.simplification._build_path(C, 1, 2, {1}) == [1, 2, 3] + D = nx.MultiDiGraph([(1, 2), (2, 3), (3, 4), (3, 5)]) + with pytest.raises(ox._errors.GraphSimplificationError, match="Impossible"): + ox.simplification._build_path(D, 1, 2, {1, 4, 5}) + + R = nx.MultiDiGraph(crs="epsg:4326") + R.add_edges_from([(1, 2), (2, 3), (3, 1)]) + assert len(ox.simplification._remove_rings(R, None, None)) == 0 + + +def test_consolidation_merges_nearby_intersections() -> None: + Gm = nx.MultiDiGraph(crs="epsg:3857") + Gm.add_node(1, x=0.0, y=0.0, street_count=2, elevation=1.0, color="red") + Gm.add_node(2, x=1.0, y=0.0, street_count=2, elevation=3.0, color="blue") + Gm.add_node(3, x=5.0, y=0.0, street_count=1, elevation=5.0, color="green") + Gm.add_edge(1, 2, osmid=1, length=1.0, geometry=LineString([(0, 0), (1, 0)])) + Gm.add_edge(2, 3, osmid=2, length=4.0, geometry=LineString([(1, 0), (5, 0)])) + Gc = ox.consolidate_intersections( + Gm, + tolerance=2, + dead_ends=True, + node_attr_aggs={"elevation": "mean"}, + ) + merged_nodes = [ + data for _, data in Gc.nodes(data=True) if isinstance(data["osmid_original"], list) + ] + assert merged_nodes[0]["elevation"] == 2.0 + assert set(merged_nodes[0]["color"]) == {"red", "blue"} + assert any(data["length"] > 4.0 for _, _, data in Gc.edges(data=True)) + + Gm.graph["consolidated"] = True + with pytest.raises(ox._errors.GraphSimplificationError, match="already been consolidated"): + ox.consolidate_intersections(Gm, tolerance=2, dead_ends=True) + + Gsplit = nx.MultiDiGraph(crs="epsg:3857") + Gsplit.add_node(1, x=0.0, y=0.0, street_count=2) + Gsplit.add_node(2, x=1.0, y=0.0, street_count=2) + Gsplit.add_node(3, x=10.0, y=0.0, street_count=1) + Gsplit.add_node(4, x=11.0, y=0.0, street_count=1) + Gsplit.add_edge(1, 3, osmid=13, length=10.0, geometry=LineString([(0, 0), (10, 0)])) + Gsplit.add_edge(2, 4, osmid=24, length=10.0, geometry=LineString([(1, 0), (11, 0)])) + Gc_split = ox.consolidate_intersections(Gsplit, tolerance=2, dead_ends=True) + assert len(Gc_split) == 4 + + +def test_feature_processing_filters_tags_and_geometry() -> None: + outer_line = LineString([(0, 0), (4, 0), (4, 4), (0, 4), (0, 0)]) + inner_line = LineString([(1, 1), (2, 1), (2, 2), (1, 2), (1, 1)]) + inner_poly = Polygon([(2.5, 2.5), (3, 2.5), (3, 3), (2.5, 3)]) + relation_geom = ox.features._build_relation_geometry( + [ + {"type": "way", "ref": 1, "role": "outer"}, + {"type": "way", "ref": 2, "role": "inner"}, + {"type": "way", "ref": 3, "role": "inner"}, + ], + {1: outer_line, 2: inner_line, 3: inner_poly}, + ) + assert relation_geom.area > 0 + assert ox.features._build_relation_geometry( + [{"type": "way", "ref": 999, "role": "outer"}], + {}, + ).is_empty + assert ( + ox.features._remove_polygon_holes([Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])], []).area == 1 + ) + assert ox.features._build_way_geometry(1, [1, 2], {}, {}).is_empty + + idx = pd.MultiIndex.from_tuples( + [("node", 1), ("node", 2), ("node", 3)], + names=["element", "id"], + ) + gdf = gpd.GeoDataFrame( + { + "amenity": ["cafe", "school", None], + "landuse": [None, "retail", "industrial"], + "geometry": [Point(0.5, 0.5), Point(2, 2), Point(5, 5)], + }, + index=idx, + crs=ox.settings.default_crs, + ) + polygon = Polygon([(0, 0), (3, 0), (3, 3), (0, 3)]) + filtered = ox.features._filter_features( + gdf, + polygon, + {"amenity": "cafe", "landuse": ["retail"]}, + ) + assert filtered.index.to_list() == [("node", 1), ("node", 2)] + with ( + suppress_type_checks(), + pytest.raises( + ox._errors.InsufficientResponseError, + match="No matching features", + ), + ): + ox.features._filter_features(gdf, polygon, {"amenity": "library"}) + + ox.settings.cache_only_mode = True + with pytest.raises(ox._errors.CacheOnlyInterruptError, match="Interrupted because"): + ox.features._create_gdf([{"elements": []}], Polygon(), {}) + ox.settings.cache_only_mode = False + with pytest.raises(ox._errors.InsufficientResponseError, match="No matching features"): + ox.features._create_gdf( + [{"elements": [{"type": "node", "id": 1, "lat": 0, "lon": 0, "tags": {}}]}], + Polygon(), + {"amenity": True}, + ) diff --git a/tests/test_online_apis.py b/tests/test_online_apis.py new file mode 100755 index 00000000..26b62500 --- /dev/null +++ b/tests/test_online_apis.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python +"""Live public web API compatibility tests.""" + +# ruff: noqa: PLR2004, S101 + +from __future__ import annotations + +import networkx as nx +import pytest + +import osmnx as ox + +pytestmark = pytest.mark.online + +LOCATION_POINT = (37.791427, -122.410018) +ADDRESS = "Transamerica Pyramid, 600 Montgomery Street, San Francisco, California, USA" +PLACE = {"city": "Piedmont", "state": "California", "country": "USA"} +TAGS: dict[str, bool | str | list[str]] = {"building": True, "amenity": True, "highway": "bus_stop"} + + +@pytest.fixture(autouse=True) +def _configure_live_api_tests() -> None: + """Disable cache so these tests exercise current public API behavior.""" + ox.settings.use_cache = False + ox.settings.overpass_rate_limit = True + ox.settings.requests_timeout = 180 + + +def _assert_valid_graph(G: nx.MultiDiGraph) -> None: + ox.convert.validate_graph(G) + assert len(G) > 0 + assert len(G.edges) > 0 + assert G.graph["crs"] == ox.settings.default_crs + + +@pytest.mark.xdist_group(name="group1") +def test_live_overpass_graph_downloaders() -> None: + """Smoke-test public Overpass graph downloader entry points.""" + bbox = ox.utils_geo.bbox_from_point(LOCATION_POINT, dist=250) + polygon = ox.geocode_to_gdf(PLACE, which_result=1).geometry.iloc[0] + + graphs = [ + ox.graph_from_bbox(bbox, network_type="drive"), + ox.graph_from_point(LOCATION_POINT, dist=250, network_type="drive"), + ox.graph_from_address(ADDRESS, dist=250, network_type="drive"), + ox.graph_from_place(PLACE, network_type="drive", which_result=1), + ox.graph_from_polygon(polygon, network_type="drive"), + ] + + for G in graphs: + _assert_valid_graph(G) + + +@pytest.mark.xdist_group(name="group2") +def test_live_overpass_feature_downloaders() -> None: + """Smoke-test public Overpass feature downloader entry points.""" + bbox = ox.utils_geo.bbox_from_point(LOCATION_POINT, dist=250) + polygon = ox.geocode_to_gdf(PLACE, which_result=1).geometry.iloc[0] + + gdfs = [ + ox.features_from_bbox(bbox, tags=TAGS), + ox.features_from_point(LOCATION_POINT, tags=TAGS, dist=250), + ox.features_from_address(ADDRESS, tags=TAGS, dist=250), + ox.features_from_place(PLACE, tags=TAGS, which_result=1), + ox.features_from_polygon(polygon, tags=TAGS), + ] + + for gdf in gdfs: + ox.convert.validate_features_gdf(gdf) + assert len(gdf) > 0 + assert gdf.crs == ox.settings.default_crs + + +@pytest.mark.xdist_group(name="group3") +def test_live_nominatim_downloaders() -> None: + """Smoke-test public Nominatim geocoding endpoints.""" + point = ox.geocode(ADDRESS) + gdf_place = ox.geocode_to_gdf(PLACE, which_result=1) + gdf_osmid = ox.geocode_to_gdf("R2999176", by_osmid=True) + + assert len(point) == 2 + assert len(gdf_place) == 1 + assert len(gdf_osmid) == 1 + assert gdf_place.crs == ox.settings.default_crs + assert not gdf_place.geometry.is_empty.any() + + +@pytest.mark.xdist_group(name="group3") +def test_live_elevation_downloader() -> None: + """Smoke-test a public Google-compatible elevation endpoint.""" + G = nx.MultiDiGraph(crs=ox.settings.default_crs) + G.add_node(1, x=-122.410018, y=37.791427, street_count=1) + G.add_node(2, x=-122.409018, y=37.792427, street_count=1) + G.add_edge(1, 2, osmid=1, length=100.0) + + ox.settings.elevation_url_template = ( + "https://api.opentopodata.org/v1/aster30m?locations={locations}&key={key}" + ) + G = ox.elevation.add_node_elevations_google(G, batch_size=100, pause=1) + + elevations = dict(G.nodes(data="elevation")) + assert set(elevations) == {1, 2} + assert all(isinstance(value, float) for value in elevations.values()) diff --git a/tests/test_osmnx.py b/tests/test_osmnx.py deleted file mode 100755 index 48a45557..00000000 --- a/tests/test_osmnx.py +++ /dev/null @@ -1,872 +0,0 @@ -#!/usr/bin/env python -# ruff: noqa: F841, PLR2004, S101 -"""Test suite for the package.""" - -from __future__ import annotations - -# use agg backend so you don't need a display on CI -# do this first before pyplot is imported by anything -import matplotlib as mpl - -mpl.use("Agg") - -import bz2 -import gzip -import logging as lg -import os -import tempfile -from collections import OrderedDict -from pathlib import Path - -import geopandas as gpd -import networkx as nx -import numpy as np -import pandas as pd -import pytest -from lxml import etree -from requests.exceptions import ConnectionError as RequestsConnectionError -from shapely import Point -from shapely import Polygon -from shapely import wkt -from typeguard import suppress_type_checks - -import osmnx as ox - -ox.settings.log_console = True -ox.settings.log_file = True -ox.settings.use_cache = True -ox.settings.data_folder = ".temp/data" -ox.settings.logs_folder = ".temp/logs" -ox.settings.imgs_folder = ".temp/imgs" -ox.settings.cache_folder = ".temp/cache" - -# define queries to use throughout tests -location_point = (37.791427, -122.410018) -polar_point_south = (-84.5501149, -64.1500283) -polar_point_north = (85.0511092, -30.4142117) - -address = "Transamerica Pyramid, 600 Montgomery Street, San Francisco, California, USA" -place1 = {"city": "Piedmont", "state": "California", "country": "USA"} -polygon_wkt = ( - "POLYGON ((-122.262 37.869, -122.255 37.869, -122.255 37.874, " - "-122.262 37.874, -122.262 37.869))" -) -polygon = ox.utils_geo.buffer_geometry(geom=wkt.loads(polygon_wkt), dist=1) - - -@pytest.mark.xdist_group(name="group1") -def test_logging() -> None: - """Test the logger.""" - ox.utils.log("test a fake default message") - ox.utils.log("test a fake debug", level=lg.DEBUG) - ox.utils.log("test a fake info", level=lg.INFO) - ox.utils.log("test a fake warning", level=lg.WARNING) - ox.utils.log("test a fake error", level=lg.ERROR) - - ox.utils.citation(style="apa") - ox.utils.citation(style="bibtex") - ox.utils.citation(style="ieee") - ox.utils.ts(style="iso8601") - ox.utils.ts(style="date") - ox.utils.ts(style="time") - - -@pytest.mark.xdist_group(name="group1") -def test_exceptions() -> None: - """Test the custom errors.""" - message = "testing exception" - - with pytest.raises(ox._errors.CacheOnlyInterruptError): - raise ox._errors.CacheOnlyInterruptError(message) - - with pytest.raises(ox._errors.GraphSimplificationError): - raise ox._errors.GraphSimplificationError(message) - - with pytest.raises(ox._errors.ValidationError): - raise ox._errors.ValidationError(message) - - with pytest.raises(ox._errors.InsufficientResponseError): - raise ox._errors.InsufficientResponseError(message) - - with pytest.raises(ox._errors.ResponseStatusCodeError): - raise ox._errors.ResponseStatusCodeError(message) - - -@pytest.mark.xdist_group(name="group1") -def test_validating() -> None: # noqa: PLR0915 - """Test validating graph inputs and objects.""" - # validate graph edge attribute is numeric and non-null - G = nx.MultiDiGraph() - G.add_edge(0, 1) - with pytest.raises(ox._errors.ValidationError): - ox._validate._verify_numeric_edge_attribute(G, "length", strict=True) - - # features GeoDataFrame validation - # pass in gdf with missing geometries and non-unique, non-multi index - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_features_gdf(gpd.GeoDataFrame(index=[0, 0])) - - # node/edge GeoDataFrame validation - # pass in wrong types, bad indexes, and missing x/y columns - gdf_nodes = pd.DataFrame(index=[0, 0]) - gdf_edges = pd.DataFrame() - with suppress_type_checks(), pytest.raises(ox._errors.ValidationError): - ox.convert.validate_node_edge_gdfs(gdf_nodes, gdf_edges) - - # pass in non-Point node geometries - gdf_nodes = gpd.GeoDataFrame(geometry=[Polygon(), Polygon()]) - gdf_edges = gpd.GeoDataFrame() - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_node_edge_gdfs(gdf_nodes, gdf_edges) - - # pass in x/y not matching geometries - data = {"x": [0, 1], "y": [2, 3]} - gdf_nodes = gpd.GeoDataFrame(data=data, geometry=[Point((6, 7)), Point((8, 9))]) - gdf_edges = gpd.GeoDataFrame() - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_node_edge_gdfs(gdf_nodes, gdf_edges) - - # graph validation - # pass an empty non-MultiDiGraph - G = nx.Graph() - with suppress_type_checks(), pytest.raises(ox._errors.ValidationError): - ox.convert.validate_graph(G) - - # test missing top-level graph attribute and non-int node IDs - G = nx.MultiDiGraph() - del G.graph - G.add_edge("0", "1") - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_graph(G) - - # pass an empty MultiDiGraph with an invalid CRS - G = nx.MultiDiGraph() - G.graph["crs"] = "epsg:999999" - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_graph(G) - - # fix the CRS and add an edge - G.graph["crs"] = "epsg:4326" - G.add_edge(0, 1) - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_graph(G) - - # add required node attributes, but with invalid types - nx.set_node_attributes(G, values=None, name="x") - nx.set_node_attributes(G, values=None, name="y") - nx.set_node_attributes(G, values=None, name="street_count") - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_graph(G) - - # fix the invalid node attribute types - nx.set_node_attributes(G, values=0, name="x") - nx.set_node_attributes(G, values=0, name="y") - nx.set_node_attributes(G, values=None, name="street_count") - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_graph(G) - - # add required edge attributes, but with invalid types - nx.set_edge_attributes(G, values=None, name="osmid") - nx.set_edge_attributes(G, values=None, name="length") - with pytest.raises(ox._errors.ValidationError): - ox.convert.validate_graph(G) - - # fix the invalid node attribute types: should finally pass validation - nx.set_edge_attributes(G, values=[0], name="osmid") - nx.set_edge_attributes(G, values=1.5, name="length") - ox.convert.validate_graph(G) - - -@pytest.mark.xdist_group(name="group1") -def test_geocoder() -> None: - """Test retrieving elements by place name and OSM ID.""" - city = ox.geocode_to_gdf("R2999176", by_osmid=True) - city = ox.geocode_to_gdf(place1, which_result=1) - city_projected = ox.projection.project_gdf(city, to_crs="epsg:3395") - - # test geocoding a bad query: should raise exception - with pytest.raises(ox._errors.InsufficientResponseError): - _ = ox.geocode("!@#$%^&*") - - with pytest.raises(ox._errors.InsufficientResponseError): - _ = ox.geocode_to_gdf(query="AAAZZZ") - - # fails to geocode to a (Multi)Polygon - with pytest.raises(TypeError): - _ = ox.geocode_to_gdf("Bunker Hill, Los Angeles, California, USA") - - -@pytest.mark.xdist_group(name="group1") -def test_stats() -> None: - """Test generating graph stats.""" - # create graph, add a new node, add bearings, project it - G = ox.graph_from_place(place1, network_type="all") - G.add_node(0, x=location_point[1], y=location_point[0], street_count=0) - G_proj = ox.project_graph(G) - G_proj = ox.project_graph(G_proj) # test double-projection - G_proj = ox.distance.add_edge_lengths(G_proj, edges=tuple(G_proj.edges)[0:3]) - - # calculate stats - cspn = ox.stats.count_streets_per_node(G) - stats = ox.basic_stats(G) - stats = ox.basic_stats(G, area=1000) - stats = ox.basic_stats(G_proj, area=1000, clean_int_tol=15) - - # test cleaning and rebuilding graph - G_clean = ox.consolidate_intersections(G_proj, tolerance=10, rebuild_graph=True, dead_ends=True) - G_clean = ox.consolidate_intersections( - G_proj, - tolerance=10, - rebuild_graph=True, - reconnect_edges=False, - ) - G_clean = ox.consolidate_intersections(G_proj, tolerance=10, rebuild_graph=False) - G_clean = ox.consolidate_intersections(G_proj, tolerance=50000, rebuild_graph=True) - - # try consolidating an empty graph - G = nx.MultiDiGraph(crs="epsg:4326") - G_clean = ox.consolidate_intersections(G, rebuild_graph=True) - G_clean = ox.consolidate_intersections(G, rebuild_graph=False) - - # test passing dict of tolerances to consolidate_intersections - tols: dict[int, float] - # every node present - tols = dict.fromkeys(G_proj.nodes, 5) - G_clean = ox.consolidate_intersections(G_proj, tolerance=tols, rebuild_graph=True) - # one node missing - tols.popitem() - G_clean = ox.consolidate_intersections(G_proj, tolerance=tols, rebuild_graph=True) - # one node 0 - tols[next(iter(tols))] = 0 - G_clean = ox.consolidate_intersections(G_proj, tolerance=tols, rebuild_graph=True) - - -@pytest.mark.xdist_group(name="group1") -def test_bearings() -> None: - """Test bearings and orientation entropy.""" - G = ox.graph_from_place(place1, network_type="all") - G.add_node(0, x=location_point[1], y=location_point[0], street_count=0) - _ = ox.bearing.calculate_bearing(0, 0, 1, 1) - G = ox.add_edge_bearings(G) - G_proj = ox.project_graph(G) - - # calculate entropy - Gu = ox.convert.to_undirected(G) - entropy = ox.bearing.orientation_entropy(Gu, weight="length") - _, ax = ox.plot.plot_orientation(Gu, area=True, title="Title") - _, _ = ox.plot.plot_orientation(Gu, ax=ax, area=False, title="Title") - - # test support of edge bearings for directed and undirected graphs - G = nx.MultiDiGraph(crs="epsg:4326") - G.add_node("point_1", x=0.0, y=0.0) - G.add_node("point_2", x=0.0, y=1.0) # latitude increases northward - G.add_edge("point_1", "point_2", weight=2.0) - G = ox.distance.add_edge_lengths(G) - G = ox.add_edge_bearings(G) - with pytest.warns(UserWarning, match="edge bearings will be directional"): - bearings, weights = ox.bearing._extract_edge_bearings(G, min_length=0, weight=None) - assert list(bearings) == [0.0] # north - assert list(weights) == [1.0] - bearings, weights = ox.bearing._extract_edge_bearings( - ox.convert.to_undirected(G), - min_length=0, - weight="weight", - ) - assert list(bearings) == [0.0, 180.0] # north and south - assert list(weights) == [2.0, 2.0] - - # test _bearings_distribution split bin implementation - bin_counts, bin_centers = ox.bearing._bearings_distribution( - G, - num_bins=1, - min_length=0, - weight=None, - ) - assert list(bin_counts) == [1.0] - assert list(bin_centers) == [0.0] - bin_counts, bin_centers = ox.bearing._bearings_distribution( - G, - num_bins=2, - min_length=0, - weight=None, - ) - assert list(bin_counts) == [1.0, 0.0] - assert list(bin_centers) == [0.0, 180.0] - - -@pytest.mark.xdist_group(name="group1") -def test_osm_xml() -> None: - """Test working with .osm XML data.""" - # test loading a graph from a local .osm xml (and bz2 and gzip) file - node_id = 53098262 - neighbor_ids = 53092170, 53060438, 53027353, 667744075 - - # read the contents of the bzip2 file - path_bz2 = "tests/input_data/West-Oakland.osm.bz2" - with bz2.open(path_bz2, mode="rb") as f: - file_contents = f.read() - - # write the contents to a .osm file - path_osm_temp = path_bz2.strip(".bz2") - with Path(path_osm_temp).open("wb") as f: - f.write(file_contents) - - # write the contents to a gzip file - path_gz_temp = path_osm_temp + ".gz" - with gzip.open(path_gz_temp, mode="wb") as f: - f.write(file_contents) - - # load and test graph_from_xml across the .osm, .bz2, and .gz files - for filepath in (path_bz2, path_gz_temp, path_osm_temp): - G = ox.graph_from_xml(filepath) - ox.convert.validate_graph(G, strict=False) # non-strict because nodes lack street_count - assert node_id in G.nodes - - for neighbor_id in neighbor_ids: - edge_key = (node_id, neighbor_id, 0) - assert neighbor_id in G.nodes - assert edge_key in G.edges - assert G.edges[edge_key]["name"] in {"8th Street", "Willow Street"} - - # delete the temporary .osm and .gz files - Path.unlink(Path(path_osm_temp)) - Path.unlink(Path(path_gz_temp)) - - # test OSM xml saving - G = ox.graph_from_point(location_point, dist=500, network_type="drive", simplify=False) - fp = Path(ox.settings.data_folder) / "graph.osm" - ox.io.save_graph_xml(G, filepath=fp, way_tag_aggs={"lanes": "sum"}) - - # validate saved XML against XSD schema - xsd_filepath = "./tests/input_data/osm_schema.xsd" - parser = etree.XMLParser(schema=etree.XMLSchema(file=xsd_filepath)) - _ = etree.parse(fp, parser=parser) - - # test roundabout handling - default_all_oneway = ox.settings.all_oneway - ox.settings.all_oneway = True - default_overpass_settings = ox.settings.overpass_settings - ox.settings.overpass_settings += '[date:"2023-04-01T00:00:00Z"]' - point = (39.0290346, -84.4696884) - G = ox.graph_from_point(point, dist=500, dist_type="bbox", network_type="drive", simplify=False) - ox.io.save_graph_xml(G) - _ = etree.parse(fp, parser=parser) - - # raise error if trying to save a simplified graph - with pytest.raises(ox._errors.GraphSimplificationError): - ox.io.save_graph_xml(ox.simplification.simplify_graph(G)) - - # save a projected/consolidated graph as OSM XML - Gc = ox.simplification.consolidate_intersections(ox.projection.project_graph(G)) - ox.convert.validate_graph(Gc) - nx.set_node_attributes(Gc, 0, name="uid") - ox.io.save_graph_xml(Gc, fp) # issues UserWarning - Gc = ox.graph.graph_from_xml(fp) # issues UserWarning - ox.convert.validate_graph(Gc, strict=False) # non-strict because nodes lack street_count - _ = etree.parse(fp, parser=parser) - - # restore settings - ox.settings.overpass_settings = default_overpass_settings - ox.settings.all_oneway = default_all_oneway - - -@pytest.mark.xdist_group(name="group1") -def test_elevation() -> None: - """Test working with elevation data.""" - G = ox.graph_from_address(address=address, dist=500, dist_type="bbox", network_type="bike") - - # add node elevations from Google (fails without API key) - with pytest.raises(ox._errors.InsufficientResponseError): - _ = ox.elevation.add_node_elevations_google(G, api_key="", batch_size=350) - - # add node elevations from Open Topo Data (works without API key) - ox.settings.elevation_url_template = ( - "https://api.opentopodata.org/v1/aster30m?locations={locations}&key={key}" - ) - _ = ox.elevation.add_node_elevations_google(G, batch_size=100, pause=1) - - # same thing again, to hit the cache - _ = ox.elevation.add_node_elevations_google(G, batch_size=100, pause=0) - - # add node elevations from a single raster file (some nodes will be null) - rasters = list(Path("tests/input_data").glob("elevation*.tif")) - G = ox.elevation.add_node_elevations_raster(G, rasters[0], cpus=1) - assert pd.notna(pd.Series(dict(G.nodes(data="elevation")))).any() - - # add node elevations from multiple raster files (no nodes should be null) - G = ox.elevation.add_node_elevations_raster(G, rasters) - assert pd.notna(pd.Series(dict(G.nodes(data="elevation")))).all() - - # consolidate nodes with elevation (by default will aggregate via mean) - G = ox.simplification.consolidate_intersections(G) - - # add edge grades and their absolute values - G = ox.add_edge_grades(G, add_absolute=True) - - -@pytest.mark.xdist_group(name="group1") -def test_routing() -> None: - """Test working with speed, travel time, and routing.""" - G = ox.graph_from_address(address=address, dist=500, dist_type="bbox", network_type="bike") - - # give each edge speed and travel time attributes - G = ox.add_edge_speeds(G) - G = ox.add_edge_speeds(G, hwy_speeds={"motorway": 100}) - G = ox.add_edge_travel_times(G) - - # test value cleaning - assert ox.routing._clean_maxspeed("100,2") == 100.2 - assert ox.routing._clean_maxspeed("100.2") == 100.2 - assert ox.routing._clean_maxspeed("100 km/h") == 100.0 - assert ox.routing._clean_maxspeed("100 mph") == pytest.approx(160.934) - assert ox.routing._clean_maxspeed("60|100") == 80 - assert ox.routing._clean_maxspeed("60|100 mph") == pytest.approx(128.7472) - assert ox.routing._clean_maxspeed("signal") is None - assert ox.routing._clean_maxspeed("100;70") is None - assert ox.routing._clean_maxspeed("FR:urban") == 50.0 - - # test collapsing multiple mph values to single kph value - assert ox.routing._collapse_multiple_maxspeed_values(["25 mph", "30 mph"], np.mean) == 44.25685 - - # test collapsing invalid values: should return None - assert ox.routing._collapse_multiple_maxspeed_values(["mph", "kph"], np.mean) is None - - orig_x = np.array([-122.404771]) - dest_x = np.array([-122.401429]) - orig_y = np.array([37.794302]) - dest_y = np.array([37.794987]) - orig_node = int(ox.distance.nearest_nodes(G, orig_x, orig_y)[0]) - dest_node = int(ox.distance.nearest_nodes(G, dest_x, dest_y)[0]) - - # test non-numeric weight, should raise ValueError - with pytest.raises(ValueError, match="contains non-numeric values"): - route1 = ox.shortest_path(G, orig_node, dest_node, weight="highway") - - # mismatch iterable and non-iterable orig/dest, should raise TypeError - msg = "must either both be iterable or neither must be iterable" - with pytest.raises(TypeError, match=msg): - route2 = ox.shortest_path(G, orig_node, [dest_node]) # type: ignore[call-overload] - - # mismatch lengths of orig/dest, should raise ValueError - msg = "must be of equal length" - with pytest.raises(ValueError, match=msg): - route2 = ox.shortest_path(G, [orig_node] * 2, [dest_node] * 3) - - # test missing weight (should raise warning) - route3 = ox.shortest_path(G, orig_node, dest_node, weight="time") - # test good weight - route4 = ox.routing.shortest_path(G, orig_node, dest_node, weight="travel_time") - route5 = ox.shortest_path(G, orig_node, dest_node, weight="travel_time") - assert route5 is not None - - route_edges = ox.routing.route_to_gdf(G, route5, weight="travel_time") - - _, _ = ox.plot_graph_route(G, route5, save=True) - - # test multiple origins-destinations - n = 5 - nodes = np.array(G.nodes) - origs = [int(x) for x in np.random.default_rng().choice(nodes, size=n, replace=True)] - dests = [int(x) for x in np.random.default_rng().choice(nodes, size=n, replace=True)] - paths1 = ox.shortest_path(G, origs, dests, weight="length", cpus=1) - paths2 = ox.shortest_path(G, origs, dests, weight="length", cpus=2) - paths3 = ox.shortest_path(G, origs, dests, weight="length", cpus=None) - assert paths1 == paths2 == paths3 - - # test k shortest paths - routes = ox.routing.k_shortest_paths(G, orig_node, dest_node, k=2, weight="travel_time") - _, _ = ox.plot_graph_routes(G, list(routes)) - - # test great circle and euclidean distance calculators - assert ox.distance.great_circle(0, 0, 1, 1) == pytest.approx(157249.6034105) - assert ox.distance.euclidean(0, 0, 1, 1) == pytest.approx(1.4142135) - - -@pytest.mark.xdist_group(name="group1") -def test_plots() -> None: - """Test visualization methods.""" - G = ox.graph_from_point(location_point, dist=500, network_type="drive") - Gp = ox.project_graph(G) - G = ox.project_graph(G, to_latlong=True) - - # test getting colors - co1 = ox.plot.get_colors(n=5, cmap="plasma", start=0.1, stop=0.9, alpha=0.5) - co2 = ox.plot.get_colors(n=5, cmap="plasma", start=0.1, stop=0.9, alpha=None) - nc = ox.plot.get_node_colors_by_attr(G, "x") - ec = ox.plot.get_edge_colors_by_attr(G, "length", num_bins=5) - - # plot and save to disk - filepath = Path(ox.settings.data_folder) / "test.svg" - _, ax = ox.plot_graph(G, show=False, save=True, close=True, filepath=filepath) - _, ax = ox.plot_graph(Gp, edge_linewidth=0, figsize=(5, 5), bgcolor="y") - _, ax = ox.plot_graph( - Gp, - ax=ax, - dpi=180, - node_color="k", - node_size=5, - node_alpha=0.1, - node_edgecolor="b", - node_zorder=5, - edge_color="r", - edge_linewidth=2, - edge_alpha=0.1, - show=False, - save=True, - close=True, - ) - - # figure-ground plots - _, _ = ox.plot_figure_ground(G=G) - - -@pytest.mark.xdist_group(name="group1") -def test_nearest() -> None: - """Test nearest node/edge searching.""" - # get graph and x/y coords to search - G = ox.graph_from_point(location_point, dist=500, network_type="drive", simplify=False) - Gp = ox.project_graph(G) - points = ox.utils_geo.sample_points(ox.convert.to_undirected(Gp), 5) - X = points.x.to_numpy() - Y = points.y.to_numpy() - - # get nearest nodes - _ = ox.distance.nearest_nodes(G, X, Y, return_dist=True) - _ = ox.distance.nearest_nodes(G, X, Y, return_dist=False) - _, _ = ox.distance.nearest_nodes(G, X[0], Y[0], return_dist=True) - nn1 = ox.distance.nearest_nodes(Gp, X[0], Y[0], return_dist=False) - - # get nearest edge - _ = ox.distance.nearest_edges(Gp, X, Y, return_dist=False) - _ = ox.distance.nearest_edges(Gp, X, Y, return_dist=True) - _ = ox.distance.nearest_edges(Gp, X[0], Y[0], return_dist=False) - _ = ox.distance.nearest_edges(Gp, X[0], Y[0], return_dist=True) - - -@pytest.mark.xdist_group(name="group1") -def test_endpoints() -> None: - """Test different API endpoints.""" - default_requests_timeout = ox.settings.requests_timeout - default_key = ox.settings.nominatim_key - default_nominatim_url = ox.settings.nominatim_url - default_overpass_url = ox.settings.overpass_url - default_overpass_rate_limit = ox.settings.overpass_rate_limit - - # test good and bad DNS resolution - ox.settings.requests_timeout = 1 - ip = ox._http._resolve_host_via_doh("overpass-api.de") - ip = ox._http._resolve_host_via_doh("AAAAAAAAAAA") - _doh_url_template_default = ox.settings.doh_url_template - ox.settings.doh_url_template = "http://aaaaaa.hostdoesntexist.org/nothinguseful" - ip = ox._http._resolve_host_via_doh("overpass-api.de") - ox.settings.doh_url_template = None - ip = ox._http._resolve_host_via_doh("overpass-api.de") - ox.settings.doh_url_template = _doh_url_template_default - - # Test changing the Overpass endpoint. - # This should fail because we didn't provide a valid endpoint - ox.settings.overpass_rate_limit = False - ox.settings.overpass_url = "http://NOT_A_VALID_ENDPOINT/api/" - with pytest.raises(RequestsConnectionError, match="Max retries exceeded with url"): - G = ox.graph_from_place(place1, network_type="all") - - ox.settings.overpass_rate_limit = default_overpass_rate_limit - ox.settings.requests_timeout = default_requests_timeout - - params: OrderedDict[str, int | str] = OrderedDict() - params["format"] = "json" - params["address_details"] = 0 - - # Bad Address - should return an empty response - params["q"] = "AAAAAAAAAAA" - response_json = ox._nominatim._nominatim_request(params=params, request_type="search") - - # Good Address - should return a valid response with a valid osm_id - params["q"] = "Newcastle A186 Westgate Rd" - response_json = ox._nominatim._nominatim_request(params=params, request_type="search") - - # Lookup - params = OrderedDict() - params["format"] = "json" - params["address_details"] = 0 - params["osm_ids"] = "W68876073" - - # good call - response_json = ox._nominatim._nominatim_request(params=params, request_type="lookup") - - # bad call - with pytest.raises( - ox._errors.InsufficientResponseError, - match="Nominatim API did not return a list of results", - ): - response_json = ox._nominatim._nominatim_request(params=params, request_type="search") - - # query must be a str if by_osmid=True - with pytest.raises(TypeError, match="`query` must be a string if `by_osmid` is True"): - ox.geocode_to_gdf(query={"City": "Boston"}, by_osmid=True) - - # Invalid nominatim query type - with pytest.raises(ValueError, match="Nominatim `request_type` must be"): - response_json = ox._nominatim._nominatim_request(params=params, request_type="xyz") - - # Searching on public nominatim should work even if a (bad) key was provided - ox.settings.nominatim_key = "NOT_A_KEY" - response_json = ox._nominatim._nominatim_request(params=params, request_type="lookup") - - ox.settings.nominatim_key = default_key - ox.settings.nominatim_url = default_nominatim_url - ox.settings.overpass_url = default_overpass_url - - -@pytest.mark.xdist_group(name="group1") -def test_save_load() -> None: # noqa: PLR0915 - """Test saving/loading graphs to/from disk.""" - G = ox.graph_from_point(location_point, dist=500, network_type="drive") - ox.convert.validate_graph(G) - - # save/load geopackage and convert graph to/from node/edge GeoDataFrames - ox.save_graph_geopackage(G, directed=False) - fp = ".temp/data/graph-dir.gpkg" - ox.save_graph_geopackage(G, filepath=fp, directed=True) - gdf_nodes1 = gpd.read_file(fp, layer="nodes").set_index("osmid") - gdf_edges1 = gpd.read_file(fp, layer="edges").set_index(["u", "v", "key"]) - G2 = ox.convert.graph_from_gdfs(gdf_nodes1, gdf_edges1) - ox.convert.validate_graph(G2, strict=False) # non-strict because osmid wasn't loaded as int - G2 = ox.graph_from_gdfs(gdf_nodes1, gdf_edges1, graph_attrs=G.graph) - ox.convert.validate_graph(G2, strict=False) # non-strict because osmid wasn't loaded as int - gdf_nodes2, gdf_edges2 = ox.convert.graph_to_gdfs(G2) - _ = list(ox.utils_geo.interpolate_points(gdf_edges2["geometry"].iloc[0], 0.001)) - assert set(gdf_nodes1.index) == set(gdf_nodes2.index) == set(G.nodes) == set(G2.nodes) - assert set(gdf_edges1.index) == set(gdf_edges2.index) == set(G.edges) == set(G2.edges) - - # test code branches that should raise exceptions - with pytest.raises(ValueError, match="You must request nodes or edges or both"): - ox.graph_to_gdfs(G2, nodes=False, edges=False) - with pytest.raises(ValueError, match="Invalid literal for boolean"): - ox.io._convert_bool_string("T") - - # create random boolean graph/node/edge attributes - attr_name = "test_bool" - G.graph[attr_name] = False - bools = np.random.default_rng().integers(low=0, high=2, size=len(G.nodes)) - node_attrs = {n: bool(b) for n, b in zip(G.nodes, bools, strict=True)} - nx.set_node_attributes(G, node_attrs, attr_name) - bools = np.random.default_rng().integers(low=0, high=2, size=len(G.edges)) - edge_attrs = {n: bool(b) for n, b in zip(G.edges, bools, strict=True)} - nx.set_edge_attributes(G, edge_attrs, attr_name) - - # create list, set, and dict attributes for nodes and edges - rand_ints_nodes = np.random.default_rng().integers(low=0, high=10, size=len(G.nodes)) - rand_ints_edges = np.random.default_rng().integers(low=0, high=10, size=len(G.edges)) - list_node_attrs = {n: [n, int(r)] for n, r in zip(G.nodes, rand_ints_nodes, strict=True)} - nx.set_node_attributes(G, list_node_attrs, "test_list") - list_edge_attrs = {e: [e, int(r)] for e, r in zip(G.edges, rand_ints_edges, strict=True)} - nx.set_edge_attributes(G, list_edge_attrs, "test_list") - set_node_attrs = {n: {n, int(r)} for n, r in zip(G.nodes, rand_ints_nodes, strict=True)} - nx.set_node_attributes(G, set_node_attrs, "test_set") - set_edge_attrs = {e: {e, int(r)} for e, r in zip(G.edges, rand_ints_edges, strict=True)} - nx.set_edge_attributes(G, set_edge_attrs, "test_set") - dict_node_attrs = {n: {n: int(r)} for n, r in zip(G.nodes, rand_ints_nodes, strict=True)} - nx.set_node_attributes(G, dict_node_attrs, "test_dict") - dict_edge_attrs = {e: {e: int(r)} for e, r in zip(G.edges, rand_ints_edges, strict=True)} - nx.set_edge_attributes(G, dict_edge_attrs, "test_dict") - - # save/load graph as graphml file - ox.save_graphml(G, gephi=True) - ox.save_graphml(G, gephi=False) - ox.save_graphml(G, gephi=False, filepath=fp) - G2 = ox.load_graphml( - fp, - graph_dtypes={attr_name: ox.io._convert_bool_string}, - node_dtypes={attr_name: ox.io._convert_bool_string}, - edge_dtypes={attr_name: ox.io._convert_bool_string}, - ) - ox.convert.validate_graph(G2) - - # verify everything in G is equivalent in G2 - assert tuple(G.graph.keys()) == tuple(G2.graph.keys()) - assert tuple(G.graph.values()) == tuple(G2.graph.values()) - z = zip(G.nodes(data=True), G2.nodes(data=True), strict=True) - for (n1, d1), (n2, d2) in z: - assert n1 == n2 - assert tuple(d1.keys()) == tuple(d2.keys()) - assert tuple(d1.values()) == tuple(d2.values()) - z = zip(G.edges(keys=True, data=True), G2.edges(keys=True, data=True), strict=True) - for (u1, v1, k1, d1), (u2, v2, k2, d2) in z: - assert u1 == u2 - assert v1 == v2 - assert k1 == k2 - assert tuple(d1.keys()) == tuple(d2.keys()) - assert tuple(d1.values()) == tuple(d2.values()) - - # test custom data types - nd = {"osmid": str} - ed = {"length": str, "osmid": float} - G2 = ox.load_graphml(fp, node_dtypes=nd, edge_dtypes=ed) - ox.convert.validate_graph(G2, strict=False) # non-strict because of non-standard types - - # test loading graphml from a file stream - graphml = Path("tests/input_data/short.graphml").read_text(encoding="utf-8") - G = ox.load_graphml(graphml_str=graphml, node_dtypes=nd, edge_dtypes=ed) - - -@pytest.mark.xdist_group(name="group2") -def test_graph_from() -> None: - """Test downloading graphs from Overpass.""" - # test subdividing a large geometry (raises a UserWarning) - bbox = ox.utils_geo.bbox_from_point((0, 0), dist=1e5, project_utm=True) - poly = ox.utils_geo.bbox_to_poly(bbox) - _ = ox.utils_geo._consolidate_subdivide_geometry(poly) - - # graph from bounding box - _ = ox.utils_geo.bbox_from_point(location_point, dist=1000, project_utm=True, return_crs=True) - bbox = ox.utils_geo.bbox_from_point(location_point, dist=500) - G = ox.graph_from_bbox(bbox, network_type="drive") - ox.convert.validate_graph(G) - G = ox.graph_from_bbox(bbox, network_type="drive_service", truncate_by_edge=True) - ox.convert.validate_graph(G) - - # truncate graph by bounding box - bbox = ox.utils_geo.bbox_from_point(location_point, dist=400) - G = ox.truncate.truncate_graph_bbox(G, bbox) - ox.convert.validate_graph(G) - G = ox.truncate.largest_component(G, strongly=True) - ox.convert.validate_graph(G) - - # graph from address - G = ox.graph_from_address(address=address, dist=500, dist_type="bbox", network_type="bike") - ox.convert.validate_graph(G) - - # graph from list of places - G = ox.graph_from_place([place1], which_result=[None], network_type="all") - ox.convert.validate_graph(G) - - # graph from polygon - G = ox.graph_from_polygon(polygon, network_type="walk", truncate_by_edge=True, simplify=False) - ox.convert.validate_graph(G) - G = ox.simplify_graph( - G, - node_attrs_include=["junction", "ref"], - edge_attrs_differ=["osmid"], - remove_rings=False, - track_merged=True, - ) - ox.convert.validate_graph(G) - - # test custom query filter - cf = ( - '["highway"]' - '["area"!~"yes"]' - '["highway"!~"motor|proposed|construction|abandoned|platform|raceway"]' - '["foot"!~"no"]' - '["service"!~"private"]' - '["access"!~"private"]' - ) - G = ox.graph_from_point( - location_point, - dist=500, - custom_filter=cf, - dist_type="bbox", - network_type="all_public", - ) - ox.convert.validate_graph(G) - - # test union of multiple custom filters - cf_union = ['["highway"~"tertiary"]', '["railway"~"tram"]'] - G = ox.graph_from_point(location_point, dist=500, custom_filter=cf_union, retain_all=True) - ox.convert.validate_graph(G) - - ox.settings.overpass_memory = 1073741824 - G = ox.graph_from_point( - location_point, - dist=500, - dist_type="network", - network_type="all", - ) - ox.convert.validate_graph(G) - - -@pytest.mark.xdist_group(name="group3") -def test_features() -> None: - """Test downloading features from Overpass.""" - bbox = ox.utils_geo.bbox_from_point(location_point, dist=500) - tags1: dict[str, bool | str | list[str]] = {"landuse": True, "building": True, "highway": True} - - with pytest.raises(ValueError, match="The geometry of `polygon` is invalid"): - ox.features.features_from_polygon(Polygon(((0, 0), (0, 0), (0, 0), (0, 0))), tags={}) - with suppress_type_checks(), pytest.raises(TypeError): - ox.features.features_from_polygon(Point(0, 0), tags={}) - - # test cache_only_mode - ox.settings.cache_only_mode = True - with pytest.raises(ox._errors.CacheOnlyInterruptError, match="Interrupted because"): - _ = ox.features_from_bbox(bbox, tags=tags1) - ox.settings.cache_only_mode = False - - # features_from_bbox - bounding box query to return no data - with pytest.raises(ox._errors.InsufficientResponseError): - gdf = ox.features_from_bbox(bbox=(-2.001, -2.001, -2.000, -2.000), tags={"building": True}) - - # features_from_bbox - successful - gdf = ox.features_from_bbox(bbox, tags=tags1) - _, ax = ox.plot_footprints(gdf) - _, ax = ox.plot_footprints(gdf, ax=ax, bbox=(0, 0, 10, 10)) - - # features_from_bbox - test < -80 deg latitude - tags2: dict[str, bool | str | list[str]] = {"natural": True, "amenity": True} - bbox = ox.utils_geo.bbox_from_point(polar_point_south, dist=500) - gdf = ox.features_from_bbox(bbox, tags=tags2) - - # features_from_bbox - test > 84 deg latitude - bbox = ox.utils_geo.bbox_from_point(polar_point_north, dist=500) - gdf = ox.features_from_bbox(bbox, tags=tags2) - - # features_from_point - tests multipolygon creation - gdf = ox.utils_geo.bbox_from_point(location_point, dist=500) - - # features_from_place - includes test of list of places - tags3: dict[str, bool | str | list[str]] = { - "amenity": True, - "landuse": ["retail", "commercial"], - "highway": "bus_stop", - } - gdf = ox.features_from_place(place1, tags=tags3) - gdf = ox.features_from_place([place1], which_result=[None], tags=tags3) - - # features_from_polygon - polygon = ox.geocode_to_gdf(place1).geometry.iloc[0] - ox.features_from_polygon(polygon, tags3) - - # features_from_address - includes testing overpass settings and snapshot from 2019 - ox.settings.overpass_settings = '[out:json][timeout:200][date:"2019-10-28T19:20:00Z"]' - gdf = ox.features_from_address(address, tags=tags3, dist=1000) - - # features_from_xml - tests error handling of clipped XMLs with incomplete geometry - gdf = ox.features_from_xml("tests/input_data/planet_10.068,48.135_10.071,48.137.osm") - - # test loading a geodataframe from a local .osm xml file - with bz2.BZ2File("tests/input_data/West-Oakland.osm.bz2") as f: - handle, temp_filename = tempfile.mkstemp(suffix=".osm") - os.write(handle, f.read()) - os.close(handle) - for filename in ("tests/input_data/West-Oakland.osm.bz2", temp_filename): - gdf = ox.features_from_xml(filename) - assert "Willow Street" in gdf["name"].to_numpy() - Path.unlink(Path(temp_filename)) - - # test the "island within a hole" and "touching inner rings" use cases - # https://wiki.openstreetmap.org/wiki/Relation:multipolygon#Island_within_a_hole - # https://wiki.openstreetmap.org/wiki/Relation:multipolygon#Touching_inner_rings - outer1 = Polygon(((0, 0), (4, 0), (4, 4), (0, 4))) - inner1 = Polygon(((1, 1), (2, 1), (2, 3), (1, 3))) - inner2 = Polygon(((2, 1), (3, 1), (3, 3), (2, 3))) - outer2 = Polygon(((1.5, 1.5), (2.5, 1.5), (2.5, 2.5), (1.5, 2.5))) - outer_polygons = [outer1, outer2] - inner_polygons = [inner1, inner2] - result = ox.features._remove_polygon_holes(outer_polygons, inner_polygons) - geom_wkt = ( - "MULTIPOLYGON (((4 4, 4 0, 0 0, 0 4, 4 4), " - "(3 1, 3 3, 2 3, 1 3, 1 1, 2 1, 3 1)), " - "((2.5 2.5, 2.5 1.5, 1.5 1.5, 1.5 2.5, 2.5 2.5)))" - ) - assert result.equals(wkt.loads(geom_wkt))