221 lines
8.3 KiB
Python
221 lines
8.3 KiB
Python
from unittest.mock import patch, MagicMock
|
||
from overpass.models import POI, PoiType
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hilfsfunktion – vermeidet Wiederholung in jedem Test
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def make_poi(**kwargs) -> POI:
|
||
defaults = dict(id="123", poi_type="bergbahn", type="node",
|
||
lat=47.1, lon=8.5, tags={"name": "Rössli"})
|
||
return POI(**{**defaults, **kwargs})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Einfacher UnitTest
|
||
# Eine Test-Klasse ohne Mocks, keine DB, keine HTTP -> Unittest
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestPoiFromRow:
|
||
"""POI.from_row: DB-Zeile wird korrekt in ein POI-Objekt umgewandelt."""
|
||
|
||
def test_is_a_poi_instance(self):
|
||
row = {
|
||
"id": "123",
|
||
"type": "node",
|
||
"poi_type": "bergbahn",
|
||
"lat": 46.72,
|
||
"lon": 9.70,
|
||
"tags": {"name": "Davos"},
|
||
}
|
||
poi = POI.from_row(row)
|
||
assert isinstance(poi, POI)
|
||
|
||
def test_all_attributes_set(self):
|
||
row = {
|
||
"id": "123",
|
||
"type": "node",
|
||
"poi_type": "bergbahn",
|
||
"lat": 46.72,
|
||
"lon": 9.70,
|
||
"tags": {"name": "Davos"},
|
||
}
|
||
poi = POI.from_row(row)
|
||
|
||
assert poi.id == "123"
|
||
assert poi.type == "node"
|
||
assert poi.poi_type == "bergbahn"
|
||
assert poi.lat == 46.72
|
||
assert poi.lon == 9.70
|
||
assert poi.tags == {"name": "Davos"}
|
||
|
||
def test_none_tag_will_become_empty_dict(self):
|
||
row = {"id": "1", "type": "node", "poi_type": "bergbahn",
|
||
"lat": 0.0, "lon": 0.0, "tags": None}
|
||
poi = POI.from_row(row)
|
||
assert poi.tags == {}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Unser Projekt hat aber einige externe Abhängigkeiten, wie post-requests auf OverpassAPI oder das Schreiben in eine
|
||
# Postgres-Datenbank. Um diese Problematiken im Testing in den Griff zu kriegen, gibt es mehrere Möglichkeiten, die wir
|
||
# hier exemplarisch Anschauen wollen.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
# STUB → "Was kommt rein?" → gibt immer dieselbe Antwort zurück
|
||
# FAKE → "Funktioniert es?" → echte Logik, keine Infrastruktur
|
||
# MOCK → "Wurde es benutzt?" → prüft Aufruf, Argumente, Häufigkeit
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# STUB – liefert vorbereitete Antworten zurück
|
||
# Kein echtes Verhalten, nur feste Rückgabewerte
|
||
# ──────────────────────────────────────────────
|
||
|
||
from overpass.fetcher import load_pois
|
||
|
||
def get_stub_response() -> MagicMock:
|
||
stub_response = MagicMock()
|
||
stub_response.json.return_value = {
|
||
"elements": [
|
||
{"id": 1, "type": "node", "lat": 47.1, "lon": 8.5,
|
||
"tags": {"name": "Davos"}}
|
||
]
|
||
}
|
||
stub_response.raise_for_status = MagicMock() # tut nichts
|
||
return stub_response
|
||
|
||
|
||
def test_stub_load_pois():
|
||
"""Stub ersetzt nur requests.post – der Rest von requests bleibt echt."""
|
||
|
||
stub_response = get_stub_response()
|
||
|
||
with patch("overpass.fetcher.requests.post", return_value=stub_response):
|
||
pois = load_pois(query="[out:json];", poi_type=PoiType.BERGBAHN)
|
||
|
||
assert len(pois) == 1
|
||
assert pois[0].tags["name"] == "Davos"
|
||
|
||
|
||
# Code-Fluss:
|
||
#
|
||
# test_stub_load_pois()
|
||
# │
|
||
# ├── load_pois(query, poi_type) ← echter Code
|
||
# │ │
|
||
# │ ├── _fetch_overpass(query) ← echter Code
|
||
# │ │ │
|
||
# │ │ ├── requests.post(...) ← ❌ ERSETZT durch Stub
|
||
# │ │ │ └── gibt stub_response zurück
|
||
# │ │ │
|
||
# │ │ ├── response.raise_for_status() ← läuft (tut nichts)
|
||
# │ │ └── response.json() ← gibt {"elements": [...]} zurück
|
||
# │ │
|
||
# │ ├── _parse_pois(data, poi_type) ← echter Code
|
||
# │ │ └── _parse_poi(element) ← echter Code
|
||
# │ │ └── POI(id=1, ...) ← echter Code
|
||
# │ │
|
||
# │ └── gibt [POI(...)] zurück
|
||
# │
|
||
# └── assert len(pois) == 1 ✅
|
||
|
||
# Vorteil:
|
||
# - externe API wird nicht gebraucht!
|
||
# - Antwort ist vorhersagbar!
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# FAKE – vereinfachte, aber funktionierende Implementierung
|
||
# Hat echte Logik – nur ohne Datenbank/Datei
|
||
# ──────────────────────────────────────────────
|
||
|
||
# ZIEL: Funktion 'fetch_and_store' soll getestet werden, aber diese hat sowohl einen HTTP-Request als auch das Speichern
|
||
# in die PostGreSQL-DB drin...
|
||
|
||
# Wir 'stuben' den HTTP-Request (bereits erledigt) und 'faken' den Datenbankzugriff
|
||
|
||
from overpass.storage import Storage
|
||
from overpass.pipeline import fetch_and_store, FetchMode
|
||
|
||
class FakeStorage(Storage):
|
||
"""Speichert POIs im Arbeitsspeicher statt in DB oder Datei."""
|
||
def __init__(self):
|
||
self.saved: list[POI] = []
|
||
|
||
def store(self, pois: list[POI]) -> str:
|
||
self.saved.extend(pois) # echte Logik, aber kein I/O
|
||
return f"memory ({len(self.saved)} POIs)"
|
||
|
||
|
||
def test_fetch_and_store_speichert_alle_bboxen():
|
||
"""
|
||
Prüft: Sammelt fetch_and_store POIs aus ALLEN Bboxen
|
||
und übergibt sie gesammelt an storage.store()
|
||
"""
|
||
storage = FakeStorage() # kein JSON, kein Postgres, kein I/O
|
||
|
||
bboxen = {
|
||
"davos": [46.72, 9.70, 46.92, 10.00],
|
||
"zuerich": [47.30, 8.40, 47.50, 8.70],
|
||
}
|
||
|
||
stub_response = get_stub_response()
|
||
|
||
with patch("overpass.fetcher.requests.post", return_value=stub_response):
|
||
fetch_and_store(
|
||
poi_type = PoiType.BERGBAHN,
|
||
bboxen = bboxen,
|
||
timeout = 25,
|
||
maxsize = 500000,
|
||
storage = storage, # FakeStorage statt JsonStorage/PostgresStorage
|
||
fetch_mode = FetchMode.SERIAL,
|
||
)
|
||
|
||
# Wurden POIs aus beiden Bboxen gesammelt?
|
||
assert len(storage.saved) > 0
|
||
assert len(storage.saved) == 2
|
||
|
||
|
||
# Was wird getestet?
|
||
# Nicht FakeStorage selbst – die ist trivial.
|
||
# Getestet wird der Code der Storage verwendet, also 'fetch_and_store' in pipeline.py:
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# MOCK – prüft wie ein Objekt verwendet wurde
|
||
# Interessiert sich nicht für den Rückgabewert,
|
||
# sondern ob store() mit den richtigen Argumenten aufgerufen wurde
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_mock_fetch_and_store_calls_store_exactly_once():
|
||
"""
|
||
fetch_and_store soll store() genau einmal aufrufen – egal wie viele Bboxen es gibt.
|
||
"""
|
||
storage = MagicMock(spec=Storage) # spec=Storage bedeutet: nur Methoden die in der echten Storage-Klasse
|
||
# existieren sind erlaubt – storage.speichern() würde einen Fehler werfen,
|
||
# storage.store() nicht
|
||
stub_response = get_stub_response()
|
||
|
||
bboxen = {
|
||
"davos": [46.72, 9.70, 46.92, 10.00],
|
||
"zuerich": [47.30, 8.40, 47.50, 8.70],
|
||
"bern": [46.90, 7.30, 47.10, 7.60],
|
||
}
|
||
|
||
|
||
with patch("overpass.fetcher.requests.post", return_value=stub_response):
|
||
fetch_and_store(
|
||
poi_type = PoiType.BERGBAHN,
|
||
bboxen = bboxen,
|
||
timeout = 25,
|
||
maxsize = 500000,
|
||
storage = storage,
|
||
fetch_mode = FetchMode.SERIAL,
|
||
)
|
||
|
||
# 3 Bboxen – aber store() darf nur EINMAL aufgerufen werden
|
||
storage.store.assert_called_once()
|
||
|
||
# Der Test prüft nicht ob POIs korrekt gespeichert werden – er prüft ob store()
|
||
# überhaupt und mit den richtigen Argumenten aufgerufen wurde. |