DanielandClaude Sonnet 5
60d5bd7cb5
70 neue Testfälle für Datenabruf.py, Datenschreiber.py, LayerLoader.py und excel_importer.py - die einzigen Module ohne jede Testdatei laut Testkatalog-Abschnitt "Ungetestete Funktionen". Schwerpunkt: Provider-URL- Bau (WFS/REST/OGR), Raumfilter, CRS-Transformation, GPKG-Schreib-/Ladepfad und Excel-Import-Fehlerzweige. qgiscore_wrapper._MockQgsProject um transformContext() und addMapLayer() ergänzt - beide fehlten im Mock und ließen die betroffenen Funktionen im Mock-Modus für jeden gültigen Layer scheitern (siehe fallstricke.md). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
330 lines
14 KiB
Python
330 lines
14 KiB
Python
"""Tests für sn_basis/modules/Datenabruf.py (Datenabruf-Kernpipeline, Issue #63)."""
|
|
|
|
import sys
|
|
import unittest
|
|
from typing import Any, cast
|
|
from unittest.mock import patch
|
|
|
|
from sn_basis.modules.Datenabruf import Datenabruf
|
|
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
|
|
|
|
|
|
class _DummyPruefmanager:
|
|
def verarbeite(self, ergebnis: Any):
|
|
return ergebnis
|
|
|
|
|
|
class _FakeCrs:
|
|
def __init__(self, authid: str) -> None:
|
|
self._authid = authid
|
|
|
|
def authid(self) -> str:
|
|
return self._authid
|
|
|
|
|
|
class _FakeLayer:
|
|
def __init__(self, authid: str = "EPSG:25833") -> None:
|
|
self._crs = _FakeCrs(authid)
|
|
|
|
def crs(self) -> _FakeCrs:
|
|
return self._crs
|
|
|
|
|
|
def _make_abruf() -> Datenabruf:
|
|
return Datenabruf(pruefmanager=cast(Any, _DummyPruefmanager()))
|
|
|
|
|
|
class DetermineSpatialFilterTests(unittest.TestCase):
|
|
def test_raumfilter_ohne_liefert_none(self):
|
|
"""Prüft, dass Raumfilter 'ohne' unabhängig vom Layer immer None liefert."""
|
|
abruf = _make_abruf()
|
|
self.assertIsNone(abruf._determine_spatial_filter("ohne", _FakeLayer()))
|
|
|
|
def test_fehlender_layer_liefert_none(self):
|
|
"""Prüft, dass ohne Verfahrensgebiet-Layer kein Raumfilter gebaut wird."""
|
|
abruf = _make_abruf()
|
|
self.assertIsNone(abruf._determine_spatial_filter("Verfahrensgebiet", None))
|
|
|
|
def test_unbekannter_raumfilter_liefert_none(self):
|
|
"""Prüft, dass ein unbekannter Raumfilter-Wert None liefert, statt einen Fehler zu werfen."""
|
|
abruf = _make_abruf()
|
|
self.assertIsNone(abruf._determine_spatial_filter("Sonstwas", _FakeLayer()))
|
|
|
|
@patch("sn_basis.modules.Datenabruf.qgiscore.get_layer_extent")
|
|
def test_verfahrensgebiet_liefert_extent_und_crs(self, mock_extent):
|
|
"""Prüft, dass Raumfilter 'Verfahrensgebiet' Extent und CRS des Verfahrensgebiet-Layers liefert."""
|
|
mock_extent.return_value = "FAKE_EXTENT"
|
|
abruf = _make_abruf()
|
|
|
|
ergebnis = abruf._determine_spatial_filter("Verfahrensgebiet", _FakeLayer("EPSG:25833"))
|
|
|
|
self.assertEqual(ergebnis, {"extent": "FAKE_EXTENT", "crs_authid": "EPSG:25833"})
|
|
|
|
@patch("sn_basis.modules.Datenabruf.qgiscore.get_layer_extent")
|
|
def test_verfahrensgebiet_ohne_extent_liefert_none(self, mock_extent):
|
|
"""Prüft, dass ein Verfahrensgebiet-Layer ohne ermittelbaren Extent None liefert."""
|
|
mock_extent.return_value = None
|
|
abruf = _make_abruf()
|
|
|
|
self.assertIsNone(abruf._determine_spatial_filter("Verfahrensgebiet", _FakeLayer()))
|
|
|
|
@patch("sn_basis.modules.Datenabruf.qgiscore.get_layer_extent")
|
|
@patch("sn_basis.modules.Datenabruf.qgiscore.create_buffer_layer")
|
|
def test_pufferlayer_liefert_extent_des_pufferlayers(self, mock_buffer, mock_extent):
|
|
"""Prüft, dass Raumfilter 'Pufferlayer' Extent und CRS des erzeugten Pufferlayers liefert."""
|
|
mock_buffer.return_value = _FakeLayer("EPSG:25833")
|
|
mock_extent.return_value = "FAKE_PUFFER_EXTENT"
|
|
abruf = _make_abruf()
|
|
|
|
ergebnis = abruf._determine_spatial_filter("Pufferlayer", _FakeLayer("EPSG:4326"))
|
|
|
|
mock_buffer.assert_called_once()
|
|
self.assertEqual(ergebnis, {"extent": "FAKE_PUFFER_EXTENT", "crs_authid": "EPSG:25833"})
|
|
|
|
@patch("sn_basis.modules.Datenabruf.qgiscore.create_buffer_layer")
|
|
def test_pufferlayer_ohne_puffer_liefert_none(self, mock_buffer):
|
|
"""Prüft, dass 'Pufferlayer' None liefert, wenn kein Pufferlayer erzeugt werden konnte."""
|
|
mock_buffer.return_value = None
|
|
abruf = _make_abruf()
|
|
|
|
self.assertIsNone(abruf._determine_spatial_filter("Pufferlayer", _FakeLayer()))
|
|
|
|
|
|
class BuildProviderUrlTests(unittest.TestCase):
|
|
def test_wms_liefert_link_unveraendert(self):
|
|
"""Prüft, dass bei Provider WMS der Link unverändert durchgereicht wird, auch mit Raumfilter."""
|
|
abruf = _make_abruf()
|
|
bbox_geom = {"extent": (1.0, 2.0, 3.0, 4.0), "crs_authid": "EPSG:25833"}
|
|
|
|
url = abruf._build_provider_url(link="http://server/wms?SERVICE=WMS", provider="WMS", bbox_geom=bbox_geom)
|
|
|
|
self.assertEqual(url, "http://server/wms?SERVICE=WMS")
|
|
|
|
def test_url_praefix_wird_entfernt(self):
|
|
"""Prüft, dass ein vorangestelltes 'url=' aus dem Link entfernt wird."""
|
|
abruf = _make_abruf()
|
|
|
|
url = abruf._build_provider_url(link="url=http://server/dienst", provider="OGR", bbox_geom=None)
|
|
|
|
self.assertTrue(url.startswith("http://server/dienst"))
|
|
|
|
def test_wfs_baut_standardparameter_und_bbox_mit_crs(self):
|
|
"""Prüft, dass WFS-Abrufe Standardparameter, umbenanntes TYPENAMES und BBOX mit CRS-URN erhalten."""
|
|
abruf = _make_abruf()
|
|
bbox_geom = {"extent": (1.0, 2.0, 3.0, 4.0), "crs_authid": "EPSG:25833"}
|
|
|
|
url = abruf._build_provider_url(
|
|
link="http://server/wfs?typename=Flurstueck", provider="WFS", bbox_geom=bbox_geom
|
|
)
|
|
|
|
self.assertIn("service=WFS", url)
|
|
self.assertIn("request=GetFeature", url)
|
|
self.assertIn("VERSION=2.0.0", url)
|
|
self.assertIn("resultType=hits", url)
|
|
self.assertIn("TYPENAMES=Flurstueck", url)
|
|
self.assertNotIn("typename=Flurstueck", url)
|
|
self.assertIn("1.0%2C2.0%2C3.0%2C4.0", url)
|
|
self.assertIn("urn:ogc:def:crs:EPSG::25833", url)
|
|
|
|
def test_wfs_repariert_doppeltes_fragezeichen(self):
|
|
"""Prüft, dass ein zweites '?' im WFS-Link zu '&' korrigiert wird, statt eine ungültige URL zu erzeugen."""
|
|
abruf = _make_abruf()
|
|
|
|
url = abruf._build_provider_url(
|
|
link="http://server/wfs?service=WFS?typename=Flurstueck", provider="WFS", bbox_geom=None
|
|
)
|
|
|
|
self.assertEqual(url.count("?"), 1)
|
|
self.assertIn("typename=Flurstueck", url.replace("TYPENAMES", "typename"))
|
|
|
|
def test_wfs_ohne_raumfilter_setzt_keine_bbox(self):
|
|
"""Prüft, dass WFS ohne Raumfilter (bbox_geom=None) keinen BBOX-Parameter setzt."""
|
|
abruf = _make_abruf()
|
|
|
|
url = abruf._build_provider_url(link="http://server/wfs?typename=Flurstueck", provider="WFS", bbox_geom=None)
|
|
|
|
self.assertNotIn("BBOX", url)
|
|
|
|
def test_rest_provider_haengt_query_endpunkt_an(self):
|
|
"""Prüft, dass REST/ArcGIS-Links ohne '/query'-Endpunkt automatisch ergänzt werden."""
|
|
abruf = _make_abruf()
|
|
|
|
url = abruf._build_provider_url(link="http://server/FeatureServer/0", provider="REST", bbox_geom=None)
|
|
|
|
self.assertTrue(url.startswith("http://server/FeatureServer/0/query?"))
|
|
self.assertIn("where=1%3D1", url)
|
|
self.assertIn("outFields=%2A", url)
|
|
self.assertIn("returnGeometry=true", url)
|
|
self.assertIn("f=json", url)
|
|
|
|
def test_rest_provider_mit_bbox_baut_geometry_envelope_und_srid(self):
|
|
"""Prüft, dass REST-Abrufe mit Raumfilter eine Geometry-Envelope und inSR/outSR aus dem CRS setzen."""
|
|
abruf = _make_abruf()
|
|
bbox_geom = {"extent": (1.0, 2.0, 3.0, 4.0), "crs_authid": "EPSG:25833"}
|
|
|
|
url = abruf._build_provider_url(link="http://server/FeatureServer/0/query", provider="REST", bbox_geom=bbox_geom)
|
|
|
|
self.assertIn("geometryType=esriGeometryEnvelope", url)
|
|
self.assertIn("spatialRel=esriSpatialRelIntersects", url)
|
|
self.assertIn("inSR=25833", url)
|
|
self.assertIn("outSR=25833", url)
|
|
|
|
def test_default_provider_haengt_generischen_bbox_parameter_an(self):
|
|
"""Prüft, dass ein unbekannter Provider mit Raumfilter einen generischen bbox-Parameter erhält."""
|
|
abruf = _make_abruf()
|
|
bbox_geom = {"extent": (1.0, 2.0, 3.0, 4.0), "crs_authid": None}
|
|
|
|
url = abruf._build_provider_url(link="http://server/ogr-quelle", provider="OGR", bbox_geom=bbox_geom)
|
|
|
|
self.assertIn("bbox=", url)
|
|
|
|
def test_default_provider_ohne_raumfilter_laesst_query_unveraendert(self):
|
|
"""Prüft, dass ein unbekannter Provider ohne Raumfilter keinen bbox-Parameter ergänzt."""
|
|
abruf = _make_abruf()
|
|
|
|
url = abruf._build_provider_url(link="http://server/ogr-quelle?feld=wert", provider="OGR", bbox_geom=None)
|
|
|
|
self.assertNotIn("bbox=", url)
|
|
self.assertIn("feld=wert", url)
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, text: str) -> None:
|
|
self.text = text
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
|
|
class FetchFeaturesTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self._alte_module = sys.modules.get("requests")
|
|
self.addCleanup(self._requests_wiederherstellen)
|
|
|
|
def _requests_wiederherstellen(self):
|
|
if self._alte_module is None:
|
|
sys.modules.pop("requests", None)
|
|
else:
|
|
sys.modules["requests"] = self._alte_module
|
|
|
|
def _fake_requests(self, get):
|
|
fake = type(sys)("requests")
|
|
fake.get = get
|
|
sys.modules["requests"] = fake
|
|
|
|
def test_wms_liefert_keine_features_ohne_fehler(self):
|
|
"""Prüft, dass bei Provider WMS kein Featureabruf stattfindet und kein Fehler gemeldet wird."""
|
|
abruf = _make_abruf()
|
|
|
|
features, fehler = abruf._fetch_features(url="http://server/wms", provider="WMS")
|
|
|
|
self.assertEqual(features, [])
|
|
self.assertIsNone(fehler)
|
|
|
|
def test_abbruch_vor_abruf_liefert_fehlermeldung(self):
|
|
"""Prüft, dass ein bereits gesetzter Abbruch vor dem eigentlichen Abruf erkannt wird."""
|
|
abruf = _make_abruf()
|
|
|
|
features, fehler = abruf._fetch_features(url="http://server/wfs", provider="WFS", cancel_callback=lambda: True)
|
|
|
|
self.assertEqual(features, [])
|
|
self.assertEqual(fehler, "Abbruch durch Benutzer")
|
|
|
|
def test_json_featurecollection_wird_geparst(self):
|
|
"""Prüft, dass eine JSON-FeatureCollection-Antwort in eine Feature-Liste umgewandelt wird."""
|
|
self._fake_requests(lambda url, timeout: _FakeResponse(
|
|
'{"type": "FeatureCollection", "features": [{"id": 1}, {"id": 2}]}'
|
|
))
|
|
abruf = _make_abruf()
|
|
|
|
features, fehler = abruf._fetch_features(url="http://server/rest/query", provider="REST")
|
|
|
|
self.assertIsNone(fehler)
|
|
self.assertEqual(features, [{"id": 1}, {"id": 2}])
|
|
|
|
def test_netzwerkfehler_wird_als_fehlermeldung_gemeldet(self):
|
|
"""Prüft, dass ein fehlgeschlagener HTTP-Abruf eine Fehlermeldung statt einer Ausnahme liefert."""
|
|
def _get(url, timeout):
|
|
raise ConnectionError("Server nicht erreichbar")
|
|
|
|
self._fake_requests(_get)
|
|
abruf = _make_abruf()
|
|
|
|
features, fehler = abruf._fetch_features(url="http://server/rest/query", provider="REST")
|
|
|
|
self.assertEqual(features, [])
|
|
self.assertIsNotNone(fehler)
|
|
self.assertIn("Server nicht erreichbar", fehler)
|
|
|
|
def test_wfs_xml_liefert_anzahl_aus_numberofmatched(self):
|
|
"""Prüft, dass eine WFS-XML-Antwort ohne JSON die Trefferzahl aus numberOfMatched übernimmt."""
|
|
self._fake_requests(lambda url, timeout: _FakeResponse(
|
|
'<wfs:FeatureCollection numberOfMatched="3" xmlns:wfs="http://www.opengis.net/wfs/2.0"/>'
|
|
))
|
|
abruf = _make_abruf()
|
|
|
|
features, fehler = abruf._fetch_features(url="http://server/wfs", provider="WFS")
|
|
|
|
self.assertIsNone(fehler)
|
|
self.assertEqual(len(features), 3)
|
|
|
|
|
|
class DatenabrufOrchestrierungTests(unittest.TestCase):
|
|
def test_fehlende_pflichtfelder_werden_als_ungueltig_gemeldet(self):
|
|
"""Prüft, dass eine Zeile ohne Link/Provider mit Aktion 'pflichtfelder_fehlen' gemeldet und übersprungen wird."""
|
|
abruf = _make_abruf()
|
|
result_dict = {"rows": [{"ident": "1", "Inhalt": "Ohne Link"}]}
|
|
|
|
_daten, ergebnisse = abruf.datenabruf(
|
|
result_dict=result_dict, raumfilter="ohne", verfahrensgebiet_layer=None, speicherort="C:/tmp/ziel.gpkg"
|
|
)
|
|
|
|
aktionen = [e.aktion for e in cast(list, ergebnisse) if isinstance(e, pruef_ergebnis)]
|
|
self.assertIn("pflichtfelder_fehlen", aktionen)
|
|
|
|
def test_wms_zeile_ohne_features_landet_in_ausserhalb(self):
|
|
"""Prüft, dass eine erfolgreich abgerufene WMS-Zeile mit 0 Features als 'ausserhalb' gezählt wird."""
|
|
abruf = _make_abruf()
|
|
result_dict = {"rows": [{"ident": "1", "Link": "http://server/wms", "Provider": "WMS", "Inhalt": "Kartendienst"}]}
|
|
|
|
daten_dict, ergebnisse = abruf.datenabruf(
|
|
result_dict=result_dict, raumfilter="ohne", verfahrensgebiet_layer=None, speicherort="C:/tmp/ziel.gpkg"
|
|
)
|
|
|
|
self.assertEqual(daten_dict["daten"]["1"], [])
|
|
summary = next(e for e in cast(list, ergebnisse) if getattr(e, "aktion", None) == "datenabruf" and "Dienste" in e.meldung)
|
|
self.assertEqual(summary.kontext["ausserhalb"], {"Kartendienst": 0})
|
|
|
|
def test_abbruch_ueber_progress_stoppt_weitere_zeilen(self):
|
|
"""Prüft, dass ein Abbruch über das Progress-Objekt die Verarbeitung weiterer Zeilen stoppt."""
|
|
abruf = _make_abruf()
|
|
result_dict = {
|
|
"rows": [
|
|
{"ident": "1", "Link": "http://server/wms", "Provider": "WMS"},
|
|
{"ident": "2", "Link": "http://server/wms", "Provider": "WMS"},
|
|
]
|
|
}
|
|
|
|
class _AbbruchProgress:
|
|
def set_label(self, _text):
|
|
return None
|
|
|
|
def is_canceled(self):
|
|
return True
|
|
|
|
daten_dict, ergebnisse = abruf.datenabruf(
|
|
result_dict=result_dict,
|
|
raumfilter="ohne",
|
|
verfahrensgebiet_layer=None,
|
|
speicherort="C:/tmp/ziel.gpkg",
|
|
progress=_AbbruchProgress(),
|
|
)
|
|
|
|
self.assertEqual(daten_dict["daten"], {})
|
|
aktionen = [getattr(e, "aktion", None) for e in cast(list, ergebnisse)]
|
|
self.assertIn("abbruch", aktionen)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|