Edward Betts
7790d10f08
Tests replay pre-recorded WDQS and Wikidata API responses so they run offline against only the local PostGIS database. capture_fixtures.py records live responses into tests/fixtures/ for later replay.
101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Download and save fixture data for the sample locations.
|
|
|
|
Run from the project root directory:
|
|
|
|
python tests/capture_fixtures.py
|
|
|
|
This makes real HTTP requests to WDQS and the Wikidata API and uses the
|
|
local PostGIS database. The output is saved to tests/fixtures/ and is
|
|
used by test_examples.py so those tests run without any network access.
|
|
|
|
Re-run this script whenever the expected results change (e.g. after an OSM
|
|
or Wikidata edit that affects a sample location).
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
import geocode.wikidata as wikidata_module
|
|
from geocode import samples
|
|
from lookup import app, lat_lon_to_wikidata
|
|
|
|
FIXTURES_DIR = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def capture_sample(i: int, lat: float, lon: float, name: str) -> None:
|
|
"""Capture and save fixture data for one sample location."""
|
|
wdqs_calls: dict[str, list] = {}
|
|
api_calls: dict[str, dict] = {}
|
|
|
|
# Hold references to the real functions before patching.
|
|
real_wdqs = wikidata_module.wdqs
|
|
real_api_call = wikidata_module.api_call
|
|
|
|
def recording_wdqs(query: str) -> list:
|
|
result = real_wdqs(query)
|
|
wdqs_calls[query] = result
|
|
return result
|
|
|
|
def recording_api_call(params: dict) -> dict:
|
|
key = json.dumps(params, sort_keys=True)
|
|
result = real_api_call(params)
|
|
api_calls[key] = result
|
|
return result
|
|
|
|
with patch("geocode.wikidata.wdqs", side_effect=recording_wdqs), patch(
|
|
"geocode.wikidata.api_call", side_effect=recording_api_call
|
|
):
|
|
reply = lat_lon_to_wikidata(lat, lon)
|
|
|
|
result = reply["result"]
|
|
commons_cat = result.get("commons_cat")
|
|
|
|
fixture: dict = {
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"name": name,
|
|
"wdqs": wdqs_calls,
|
|
"api": api_calls,
|
|
"expected_wikidata": result.get("wikidata"),
|
|
"expected_commons_cat": commons_cat["title"]
|
|
if isinstance(commons_cat, dict)
|
|
else None,
|
|
}
|
|
|
|
fixture_path = FIXTURES_DIR / f"sample_{i:02d}.json"
|
|
fixture_path.write_text(json.dumps(fixture, indent=2, ensure_ascii=False))
|
|
|
|
qid = fixture["expected_wikidata"] or "none"
|
|
cat = fixture["expected_commons_cat"] or "no commons cat"
|
|
print(f" [{i:02d}] {name}: {qid} / {cat}")
|
|
|
|
|
|
def main() -> None:
|
|
"""Capture fixtures for all samples."""
|
|
FIXTURES_DIR.mkdir(exist_ok=True)
|
|
|
|
indices = None
|
|
if len(sys.argv) > 1:
|
|
# Allow passing specific indices on the command line, e.g.:
|
|
# python tests/capture_fixtures.py 0 5 12
|
|
indices = {int(a) for a in sys.argv[1:]}
|
|
|
|
print(f"Capturing fixtures for {len(samples)} samples...")
|
|
with app.app_context():
|
|
for i, (lat, lon, name) in enumerate(samples):
|
|
if indices and i not in indices:
|
|
continue
|
|
print(f"[{i:02d}/{len(samples) - 1}] {name} ({lat}, {lon})")
|
|
capture_sample(i, lat, lon, name)
|
|
|
|
print(f"\nDone. Fixtures saved to {FIXTURES_DIR}/")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|