Finalizing mini-project

.env.example

@@ -0,0 +1 @@
+DB_CONNECTION_STRING="postgresql://user:password@localhost:5432/DB_name"

TASK.md

@@ -1,52 +1,34 @@
-# Task 18 — Testing
+# Finalizing
 
-## Rückblick Task 17: PostgresStorage
+## Rückblick Task 18: Testing
 
-Falls alles geklappt hat, habt ihr nun eure POI's in einer lokalen PostGreSQL-Datenbank abgespeichert. Erreicht haben wir
+In der Musterlösung wurde zusätzlich ein `pyproject.toml`-File eingefügt. Dieses erlaubt es euch u.a. die erstellten Tests
-diese in einer Implementation von `PostgresStorage(Storage)` ohne bestehende Businesslogik in `main.py` etc. anzufassen.
+direkt aus der Konsole im Root-Verzeichnis mit `pytest tests` auszuführen.
-Das garantiert maximale Flexibilität und Wiederverwendbarkeit des Codes und ist deutlich wartungsfreundlicher und damit 
-auch übersichtlicher.
 
-## Aufgabe
+Zusätzlich findet ihr eine `.env.example`-Datei. Diese ist selbst nicht funktional und soll dem User anzeigen, was die
+funktionale `.env` für eine Struktur hat. So weiss in unserem Falle jeder, er/sie muss eine `.env` mit `DB_CONNECTION_STRING` 
+anlegen.
 
-Bisher haben wir keinen einzigen Test für unser Miniprojekt geschrieben. Wir werden das hier auch nicht im grossen Stil
+Im Testing (`test_overpass.py`) wurde lediglich ein einziger Unittest und Demonstration für `stub`, `mock` und `fake` gemacht.
-und vollumfänglich machen, sondern nur exemplarisch. 
+In der Realität wäre das zu wenig, für Demonstrationszwecke sollte es genügen ... 
 
-UND: Wir haben noch eine 'sicherheitsrelevante' Einschränkung im Code. Im jetzigen Stand werden persönliche/private 
-Informationen wie der `username` und vor allem das `password` im `connection_string` in `config.yaml` geschrieben. Das 
-ist nicht ideal, weil die config ins `git` geschrieben wird und somit für alle (berechtigten) einsehbar ist. Das wollen 
-und müssen wir ändern! 
 
-**Konkret:**
+**Stärken**
+* Schichtenarchitektur ist (hoffentlich) klar: fetcher → pipeline → storage, keine zirkulären Abhängigkeiten
+* Fehlerbehandlung konsistent mit eigenen Exception-Typen
+* Konfiguration sauber von Code getrennt
+* Threading von API-Requests (concurrent)
 
-1. Speichert den privaten `connection_string` in einer neue angelegten `.env` und verwendet diesen in euerem Code unter
+**Schwächen/Entwicklungspotenzial**
-   Nutzung von `load_dotenv` aus (aus dem dotenv-Modul). Schaut zudem, dass euer `.env` in `.gitignore` aufgeführt ist 
+* Keine Retry-Logik für Overpass: Die API ist öffentlich und manchmal überlastet. Eine einfache Retry-Logik mit tenacity wäre produktionsreif.
-   und somit nicht von git 'getracked' wird!
+* ROOT-Konstante ist fragil. Drei .parent-Aufrufe funktionieren nur bei genau dieser Verzeichnistiefe. Wir haben nun am Schluss noch ein 
-2. Erstellt einen neuen Ordner `tests` in eurem root-Projektverzeichnis (also neben `src`)
+  `pyproject.toml` gebildet und könnten deshalb das ROOT daraus extrahieren.
-3. Erstellt darin ein neues Modul `tests_overpass.py`
-4. Wie erwähnt wollen wir nur exemplarisch wenige Tests machen:
-   
-    * in der POI-Klasse haben wir eine Klassenmethode `from_row`: Schreibt einen ganz einfachen Unittest, welcher überprüft,
-      ob diese tatsächlich ein `POI`-Objekt zurück gibt.
 
   ```python
-        @classmethod
+  def find_root() -> Path:
-        def from_row(cls, row: dict) -> POI:
+    for parent in Path(__file__).parents:
-            """DB-Zeile → POI-Objekt (Lesen)"""
+        if (parent / "pyproject.toml").exists():
-            return cls(
+            return parent
-                id       = row["id"],
+    raise RuntimeError("Projektroot nicht gefunden")
-                type     = row["type"],
-                poi_type = row["poi_type"],
-                lat      = row["lat"],
-                lon      = row["lon"],
-                tags     = row["tags"] or {},
-            )
   ```
-    
+* ...
-   * im Unterricht wurden kurz Inhalte aus dem Testing mit `mocks`, `stub` und `fake` angesprochen. Lest das nochmals und 
-     versucht zu verstehen wie und wo man diese in unserem Projekt gebrauchen könnte und warum. Wir werden gemeinsam je ein
-     Beispiel aus jeder dieser 3 Kategorien zusammen anschauen.
-
-**Fragen zum Nachdenken:**
-- Zum Projektabschluss: Bitte überlegt euch Stärken aber sicher auch Einschränkungen/Schwächen unseres Mini-Projekts. 
-  Was könnte man besser/einfacher/anders machen?

pyproject.toml

@@ -0,0 +1,37 @@
+[build-system]
+requires      = ["setuptools>=72", "wheel"]
+build-backend = "setuptools.backends.legacy:build"
+
+[project]
+name        = "overpass"
+version     = "0.1.0"
+description = "Overpass API POI Fetcher – CDS Unterrichtsprojekt"
+requires-python = ">=3.12"
+
+dependencies = [
+    "requests",
+    "pyyaml",
+    "psycopg2-binary",
+    "dotenv"
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest",
+    "pytest-cov",
+]
+
+[project.scripts]
+overpass = "overpass.main:main"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.pytest.ini_options]
+pythonpath  = ["src"]
+testpaths   = ["tests"]
+addopts     = "-v --tb=short"
+
+[tool.coverage.run]
+source = ["src/overpass"]
+omit   = ["*/__init__.py"]

requirements.txt

@@ -1 +1,2 @@
 requests
+dotenv

src/overpass/config.yaml

@@ -22,4 +22,4 @@ storage:
   type: postgres # json | postgres
   params:
     # output_dir: data/results
-    connection_string: "postgresql://postgres:marco1234@localhost:5432/overpass"
+    connection_string: Null

src/overpass/main.py

@@ -1,6 +1,8 @@
+import os
 import yaml
 import logging
 from pathlib import Path
+from dotenv import load_dotenv
 
 from .models import PoiType
 from .pipeline import fetch_and_store, FetchMode
@@ -14,11 +16,14 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
+load_dotenv()
 ROOT = Path(__file__).parent.parent.parent
 
 @timer
 def main() -> None:
     config    = yaml.safe_load((Path(__file__).parent / "config.yaml").read_text())
+    config["storage"]["params"]["connection_string"] = os.environ["DB_CONNECTION_STRING"] # Credentials direkt aus .env
+
     timeout   = config["overpass"]["timeout"]
     maxsize   = config["overpass"]["maxsize"]
     bboxen    = config["bboxen"]

tests/test_overpass.py

@@ -0,0 +1,221 @@
+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.