Merge pull request 'Tests für die vier Kernmodule der Datenpipeline ergänzt (#63)' (#64) from feature/63-tests-datenpipeline into unstable
Reviewed-on: #64
This commit was merged in pull request #64.
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
- Tests für die Kernpipeline ergänzt: Datenabruf (WFS/REST/OGR-URL-Bau, Raumfilter), Datenschreiber (GPKG-Schreiben, Laden, Log), LayerLoader (CRS-Transformation, Raum-/Geometriefilter) und excel_importer (Linklisten-Import) waren zuvor ohne Testdatei (#63)
|
||||
- Bugfix: `get_plugin_root()` löst keine Symlinks/Junctions mehr auf – im per `link_qgis_profile.py` verknüpften Dev-Setup zeigte die Funktion sonst auf den Workspace-Root statt auf den QGIS-Plugin-Ordner, wodurch Assets von Fachplugins (z. B. QML-Stile in sn_verfahrensgebiet/sn_plan41) nicht gefunden wurden (Verfahrensgebiet #50)
|
||||
- Bezeichner-Bereinigung (Layernamen) um Bindestrich ergänzt und in eine zentrale Funktion `sanitize_bezeichner` (`functions/text_wrapper.py`) ausgelagert, damit Umlaute/Sonderzeichen nicht mehr pro Aufrufstelle vergessen werden (#42)
|
||||
- Landkreis-Dropdown (statt Freitext) in den Projekteigenschaften ergänzt, inkl. Landkreiskennzahl als eigene Projektvariable
|
||||
|
||||
@@ -277,6 +277,12 @@ except Exception:
|
||||
def layoutManager(self):
|
||||
return self._layout_manager
|
||||
|
||||
def transformContext(self):
|
||||
return None
|
||||
|
||||
def addMapLayer(self, layer: Any) -> Any:
|
||||
return layer
|
||||
|
||||
QgsProject = _MockQgsProject
|
||||
|
||||
class _MockQgsVectorLayer:
|
||||
|
||||
@@ -0,0 +1,329 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,197 @@
|
||||
"""Tests für sn_basis/modules/Datenschreiber.py (Schreib-/Ladepipeline, Issue #63)."""
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from typing import Any, List, cast
|
||||
from unittest.mock import patch
|
||||
|
||||
from sn_basis.modules.Datenschreiber import Datenschreiber
|
||||
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
|
||||
|
||||
|
||||
class _AufzeichnenderPruefmanager:
|
||||
def __init__(self) -> None:
|
||||
self.ergebnisse: List[Any] = []
|
||||
|
||||
def verarbeite(self, ergebnis: Any):
|
||||
self.ergebnisse.append(ergebnis)
|
||||
return ergebnis
|
||||
|
||||
|
||||
class _FakeLayer:
|
||||
def __init__(self, valid: bool = True) -> None:
|
||||
self._valid = valid
|
||||
|
||||
def isValid(self) -> bool:
|
||||
return self._valid
|
||||
|
||||
def loadNamedStyle(self, path: str):
|
||||
return True, ""
|
||||
|
||||
def triggerRepaint(self) -> None:
|
||||
return None
|
||||
|
||||
def dataProvider(self):
|
||||
return None
|
||||
|
||||
|
||||
def _make_schreiber(gpkg_path: str = "C:/tmp/ziel.gpkg") -> tuple[Datenschreiber, _AufzeichnenderPruefmanager]:
|
||||
pruefmanager = _AufzeichnenderPruefmanager()
|
||||
return Datenschreiber(pruefmanager=cast(Any, pruefmanager), gpkg_path=gpkg_path), pruefmanager
|
||||
|
||||
|
||||
class WriteLayerToGpkgTests(unittest.TestCase):
|
||||
def test_ungueltiger_layer_liefert_fehlermeldung(self):
|
||||
"""Prüft, dass ein ungültiger Layer eine Fehlermeldung statt eines Absturzes liefert."""
|
||||
schreiber, _pm = _make_schreiber()
|
||||
|
||||
fehler = schreiber._write_layer_to_gpkg(layer_name="test", layer=_FakeLayer(valid=False))
|
||||
|
||||
self.assertIsNotNone(fehler)
|
||||
|
||||
def test_gueltiger_layer_wird_ohne_fehler_geschrieben(self):
|
||||
"""Prüft, dass ein gültiger Layer im Mock-Modus ohne Fehlermeldung geschrieben wird."""
|
||||
schreiber, _pm = _make_schreiber()
|
||||
|
||||
fehler = schreiber._write_layer_to_gpkg(layer_name="test", layer=_FakeLayer(valid=True))
|
||||
|
||||
self.assertIsNone(fehler)
|
||||
|
||||
|
||||
class ResolveStylePathTests(unittest.TestCase):
|
||||
def test_leerer_stilpfad_liefert_none(self):
|
||||
"""Prüft, dass ein leerer oder fehlender Stilpfad None liefert."""
|
||||
schreiber, _pm = _make_schreiber()
|
||||
|
||||
self.assertIsNone(schreiber._resolve_style_path(None))
|
||||
self.assertIsNone(schreiber._resolve_style_path(" "))
|
||||
|
||||
def test_absoluter_vorhandener_stilpfad_wird_aufgeloest(self):
|
||||
"""Prüft, dass ein absoluter, vorhandener Stilpfad normalisiert und unverändert erkannt wird."""
|
||||
schreiber, _pm = _make_schreiber()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
stil_datei = Path(tmp) / "stil.qml"
|
||||
stil_datei.write_text("<qgis/>", encoding="utf-8")
|
||||
|
||||
ergebnis = schreiber._resolve_style_path(str(stil_datei))
|
||||
|
||||
self.assertIsNotNone(ergebnis)
|
||||
self.assertTrue(str(ergebnis).endswith("stil.qml"))
|
||||
|
||||
def test_nicht_vorhandener_stilpfad_liefert_none(self):
|
||||
"""Prüft, dass ein nicht vorhandener Stilpfad None liefert, statt einen ungültigen Pfad zurückzugeben."""
|
||||
schreiber, _pm = _make_schreiber()
|
||||
|
||||
self.assertIsNone(schreiber._resolve_style_path("C:/nicht/vorhanden/stil.qml"))
|
||||
|
||||
|
||||
class SchreibeDatenTests(unittest.TestCase):
|
||||
def test_ohne_speicherort_wirft_valueerror(self):
|
||||
"""Prüft, dass schreibe_Daten ohne Speicherort einen ValueError wirft."""
|
||||
schreiber, _pm = _make_schreiber(gpkg_path="")
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
schreiber.schreibe_Daten(daten_dict={"daten": {}}, processed_results=[], speicherort="")
|
||||
|
||||
def test_eintrag_ohne_gueltigen_layer_wird_uebersprungen(self):
|
||||
"""Prüft, dass ein Eintrag ohne gültigen Layer übersprungen wird und ein Fehlerergebnis meldet."""
|
||||
schreiber, pruefmanager = _make_schreiber()
|
||||
daten_dict = {"daten": {"1": {"layer": None}}}
|
||||
|
||||
ergebnisse = schreiber.schreibe_Daten(daten_dict=daten_dict, processed_results=[], speicherort="C:/tmp/ziel.gpkg")
|
||||
|
||||
self.assertEqual(ergebnisse, [])
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("save_exception", aktionen)
|
||||
|
||||
def test_erfolgreicher_eintrag_liefert_layer_path_und_thema(self):
|
||||
"""Prüft, dass ein gültiger Layer einen Layer-Path mit GPKG-Pfad und Layername liefert."""
|
||||
schreiber, _pm = _make_schreiber(gpkg_path="C:/tmp/ziel.gpkg")
|
||||
daten_dict = {"daten": {"1": {"layer": _FakeLayer(valid=True)}}}
|
||||
processed_results = [
|
||||
pruef_ergebnis(ok=True, meldung="ok", aktion="datenabruf", kontext={"ident": "1", "thema": "Testthema"})
|
||||
]
|
||||
|
||||
ergebnisse = schreiber.schreibe_Daten(
|
||||
daten_dict=daten_dict, processed_results=processed_results, speicherort="C:/tmp/ziel.gpkg"
|
||||
)
|
||||
|
||||
self.assertEqual(len(ergebnisse), 1)
|
||||
eintrag = ergebnisse[0]
|
||||
self.assertEqual(eintrag["thema"], "Testthema")
|
||||
self.assertEqual(eintrag["ident"], "1")
|
||||
self.assertEqual(eintrag["layer_path"], "C:/tmp/ziel.gpkg|layername=Testthema")
|
||||
|
||||
def test_ohne_thema_wird_ident_als_layername_verwendet(self):
|
||||
"""Prüft, dass ohne passendes Thema in processed_results der ident als Layername verwendet wird."""
|
||||
schreiber, _pm = _make_schreiber(gpkg_path="C:/tmp/ziel.gpkg")
|
||||
daten_dict = {"daten": {"42": {"layer": _FakeLayer(valid=True)}}}
|
||||
|
||||
ergebnisse = schreiber.schreibe_Daten(daten_dict=daten_dict, processed_results=[], speicherort="C:/tmp/ziel.gpkg")
|
||||
|
||||
self.assertEqual(ergebnisse[0]["thema"], "42")
|
||||
|
||||
|
||||
class LadeLayerTests(unittest.TestCase):
|
||||
def test_ohne_layer_path_wird_uebersprungen(self):
|
||||
"""Prüft, dass ein Eintrag ohne layer_path übersprungen wird, ohne einen Fehler zu melden."""
|
||||
schreiber, pruefmanager = _make_schreiber()
|
||||
|
||||
schreiber.lade_Layer([{"thema": "ohne_pfad"}])
|
||||
|
||||
self.assertEqual(pruefmanager.ergebnisse, [])
|
||||
|
||||
def test_gueltiger_layer_wird_dem_projekt_hinzugefuegt(self):
|
||||
"""Prüft, dass ein gültig ladbarer Layer ohne Fehlermeldung verarbeitet wird."""
|
||||
schreiber, pruefmanager = _make_schreiber()
|
||||
|
||||
schreiber.lade_Layer([{"layer_path": "C:/tmp/ziel.gpkg|layername=Testthema", "thema": "Testthema"}])
|
||||
|
||||
fehler = [e for e in pruefmanager.ergebnisse if not getattr(e, "ok", True)]
|
||||
self.assertEqual(fehler, [])
|
||||
|
||||
@patch("sn_basis.modules.Datenschreiber.qgiscore.QgsVectorLayer")
|
||||
def test_ungueltiger_layer_meldet_layer_nicht_gefunden(self, mock_layer_cls):
|
||||
"""Prüft, dass ein ungültig ladbarer Layer als 'layer_nicht_gefunden' gemeldet wird."""
|
||||
mock_layer_cls.return_value = _FakeLayer(valid=False)
|
||||
schreiber, pruefmanager = _make_schreiber()
|
||||
|
||||
schreiber.lade_Layer([{"layer_path": "C:/tmp/ziel.gpkg|layername=Testthema", "thema": "Testthema"}])
|
||||
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("layer_nicht_gefunden", aktionen)
|
||||
|
||||
|
||||
class SchreibeLogTests(unittest.TestCase):
|
||||
def test_ohne_speicherort_wirft_valueerror(self):
|
||||
"""Prüft, dass schreibe_log ohne Speicherort einen ValueError wirft."""
|
||||
schreiber, _pm = _make_schreiber()
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
schreiber.schreibe_log(processed_results=[], speicherort="")
|
||||
|
||||
def test_schreibt_json_datei_mit_ergebnissen_und_gibt_pfad_zurueck(self):
|
||||
"""Prüft, dass schreibe_log eine lesbare JSON-Log-Datei erzeugt und deren Pfad zurückgibt."""
|
||||
schreiber, pruefmanager = _make_schreiber()
|
||||
processed_results = [
|
||||
pruef_ergebnis(ok=True, meldung="alles ok", aktion="datenabruf", kontext={"ident": "1"}),
|
||||
pruef_ergebnis(ok=False, meldung="Fehler", aktion="url_nicht_erreichbar", kontext={"ident": "2"}),
|
||||
]
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
log_path = schreiber.schreibe_log(processed_results=processed_results, speicherort=tmp)
|
||||
|
||||
self.assertTrue(Path(log_path).is_file())
|
||||
inhalt = json.loads(Path(log_path).read_text(encoding="utf-8"))
|
||||
|
||||
self.assertEqual(len(inhalt), 2)
|
||||
self.assertEqual(inhalt[0]["meldung"], "alles ok")
|
||||
self.assertEqual(inhalt[1]["kontext"], {"ident": "2"})
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("standarddatei_vorschlagen", aktionen)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Tests für sn_basis/modules/excel_importer.py (Excel-Import für Linklisten, Issue #63)."""
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from typing import Any, List, cast
|
||||
|
||||
from openpyxl import Workbook
|
||||
|
||||
from sn_basis.modules.excel_importer import ExcelImporter
|
||||
|
||||
|
||||
class _AufzeichnenderPruefmanager:
|
||||
def __init__(self) -> None:
|
||||
self.ergebnisse: List[Any] = []
|
||||
|
||||
def verarbeite(self, ergebnis: Any):
|
||||
self.ergebnisse.append(ergebnis)
|
||||
return ergebnis
|
||||
|
||||
|
||||
class ImportXlsxTests(unittest.TestCase):
|
||||
def test_erfolgreicher_import_liefert_zeilen_als_dicts(self):
|
||||
"""Prüft, dass eine gültige Excel-Datei als Liste von Dicts mit den Header-Spalten geliefert wird."""
|
||||
pruefmanager = _AufzeichnenderPruefmanager()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
pfad = Path(tmp) / "linkliste.xlsx"
|
||||
wb = Workbook()
|
||||
ws = wb.active
|
||||
ws.append(["ident", "Link", "Provider"])
|
||||
ws.append(["1", "http://server/wfs", "WFS"])
|
||||
ws.append(["2", "http://server/rest", "REST"])
|
||||
wb.save(pfad)
|
||||
|
||||
importer = ExcelImporter(filepath=str(pfad), pruefmanager=cast(Any, pruefmanager))
|
||||
zeilen = importer.import_xlsx()
|
||||
|
||||
self.assertEqual(
|
||||
zeilen,
|
||||
[
|
||||
{"ident": "1", "Link": "http://server/wfs", "Provider": "WFS"},
|
||||
{"ident": "2", "Link": "http://server/rest", "Provider": "REST"},
|
||||
],
|
||||
)
|
||||
|
||||
def test_fehlende_datei_liefert_leere_liste(self):
|
||||
"""Prüft, dass eine nicht vorhandene Datei eine leere Liste liefert, statt eine Ausnahme zu werfen."""
|
||||
pruefmanager = _AufzeichnenderPruefmanager()
|
||||
importer = ExcelImporter(filepath="C:/nicht/vorhanden/linkliste.xlsx", pruefmanager=cast(Any, pruefmanager))
|
||||
|
||||
zeilen = importer.import_xlsx()
|
||||
|
||||
self.assertEqual(zeilen, [])
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("datei_nicht_gefunden", aktionen)
|
||||
|
||||
def test_arbeitsblatt_ohne_header_liefert_leere_liste_und_meldet_kein_header(self):
|
||||
"""Prüft, dass ein leeres Arbeitsblatt ohne Header-Zeile 'kein_header' meldet und eine leere Liste liefert."""
|
||||
pruefmanager = _AufzeichnenderPruefmanager()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
pfad = Path(tmp) / "leer.xlsx"
|
||||
Workbook().save(pfad)
|
||||
|
||||
importer = ExcelImporter(filepath=str(pfad), pruefmanager=cast(Any, pruefmanager))
|
||||
zeilen = importer.import_xlsx()
|
||||
|
||||
self.assertEqual(zeilen, [])
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("kein_header", aktionen)
|
||||
|
||||
def test_header_zeile_nur_leere_werte_liefert_kein_header(self):
|
||||
"""Prüft, dass eine Header-Zeile aus ausschließlich leeren Zellen ebenfalls als 'kein_header' gilt."""
|
||||
pruefmanager = _AufzeichnenderPruefmanager()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
pfad = Path(tmp) / "leerer_header.xlsx"
|
||||
wb = Workbook()
|
||||
ws = wb.active
|
||||
ws.append([None, None, None])
|
||||
wb.save(pfad)
|
||||
|
||||
importer = ExcelImporter(filepath=str(pfad), pruefmanager=cast(Any, pruefmanager))
|
||||
zeilen = importer.import_xlsx()
|
||||
|
||||
self.assertEqual(zeilen, [])
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("kein_header", aktionen)
|
||||
|
||||
def test_nicht_lesbare_datei_liefert_leere_liste_und_meldet_read_error(self):
|
||||
"""Prüft, dass eine Datei mit ungültigem Excel-Inhalt 'read_error' meldet und eine leere Liste liefert."""
|
||||
pruefmanager = _AufzeichnenderPruefmanager()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
pfad = Path(tmp) / "kaputt.xlsx"
|
||||
pfad.write_text("Das ist keine gültige Excel-Datei", encoding="utf-8")
|
||||
|
||||
importer = ExcelImporter(filepath=str(pfad), pruefmanager=cast(Any, pruefmanager))
|
||||
zeilen = importer.import_xlsx()
|
||||
|
||||
self.assertEqual(zeilen, [])
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("read_error", aktionen)
|
||||
|
||||
|
||||
class ExcelImporterKonstruktorTests(unittest.TestCase):
|
||||
def test_ohne_dateipfad_wirft_valueerror(self):
|
||||
"""Prüft, dass ExcelImporter ohne Dateipfad einen ValueError wirft."""
|
||||
with self.assertRaises(ValueError):
|
||||
ExcelImporter(filepath="", pruefmanager=cast(Any, _AufzeichnenderPruefmanager()))
|
||||
|
||||
def test_ohne_pruefmanager_wirft_valueerror(self):
|
||||
"""Prüft, dass ExcelImporter ohne Pruefmanager einen ValueError wirft."""
|
||||
with self.assertRaises(ValueError):
|
||||
ExcelImporter(filepath="C:/tmp/x.xlsx", pruefmanager=cast(Any, None))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,427 @@
|
||||
"""Tests für sn_basis/modules/LayerLoader.py (Layer erzeugen/filtern/laden, Issue #63)."""
|
||||
|
||||
import unittest
|
||||
from typing import Any, List, cast
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from sn_basis.modules.LayerLoader import LayerLoader
|
||||
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
|
||||
|
||||
|
||||
class _AufzeichnenderPruefmanager:
|
||||
def __init__(self) -> None:
|
||||
self.ergebnisse: List[Any] = []
|
||||
|
||||
def verarbeite(self, ergebnis: Any):
|
||||
self.ergebnisse.append(ergebnis)
|
||||
return ergebnis
|
||||
|
||||
|
||||
class _FakeCrs:
|
||||
def __init__(self, authid: str) -> None:
|
||||
self._authid = authid
|
||||
|
||||
def authid(self) -> str:
|
||||
return self._authid
|
||||
|
||||
|
||||
class _FakeGeometry:
|
||||
def __init__(self, empty: bool = False) -> None:
|
||||
self.transformed = False
|
||||
self._empty = empty
|
||||
|
||||
def clone(self) -> "_FakeGeometry":
|
||||
return _FakeGeometry(self._empty)
|
||||
|
||||
def transform(self, ct: Any) -> None:
|
||||
self.transformed = True
|
||||
|
||||
def combine(self, other: "_FakeGeometry") -> "_FakeGeometry":
|
||||
return self
|
||||
|
||||
def isEmpty(self) -> bool:
|
||||
return self._empty
|
||||
|
||||
def boundingBox(self) -> str:
|
||||
return "BBOX"
|
||||
|
||||
def intersects(self, other: Any) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
class _FakeFeature:
|
||||
def __init__(self, geometry: Any = None) -> None:
|
||||
self._geometry = geometry if geometry is not None else _FakeGeometry()
|
||||
|
||||
def geometry(self) -> Any:
|
||||
return self._geometry
|
||||
|
||||
|
||||
class _FakeDataProvider:
|
||||
def __init__(self) -> None:
|
||||
self.added_features: List[Any] = []
|
||||
self.added_attributes: Any = None
|
||||
|
||||
def addAttributes(self, fields: Any) -> None:
|
||||
self.added_attributes = fields
|
||||
|
||||
def addFeatures(self, features: Any) -> None:
|
||||
self.added_features = list(features)
|
||||
|
||||
|
||||
class _FakeFilteredLayer:
|
||||
"""Ersetzt QgsVectorLayer sowohl als Klasse (VectorLayer-Konstante) als auch als Konstruktor."""
|
||||
|
||||
VectorLayer = 1
|
||||
|
||||
def __init__(self, uri: Any = None, name: Any = None, provider: Any = None) -> None:
|
||||
self.uri = uri
|
||||
self._name = name
|
||||
self._valid = True
|
||||
self._provider = _FakeDataProvider()
|
||||
|
||||
def isValid(self) -> bool:
|
||||
return self._valid
|
||||
|
||||
def dataProvider(self) -> _FakeDataProvider:
|
||||
return self._provider
|
||||
|
||||
def name(self) -> Any:
|
||||
return self._name
|
||||
|
||||
def updateFields(self) -> None:
|
||||
return None
|
||||
|
||||
def updateExtents(self) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class _FakeSourceLayer:
|
||||
def __init__(
|
||||
self,
|
||||
features: Any = None,
|
||||
valid: bool = True,
|
||||
layer_type: int = 1,
|
||||
geometry_type: int = 2,
|
||||
authid: str = "EPSG:25833",
|
||||
name: str = "Quelle",
|
||||
) -> None:
|
||||
self._features = features if features is not None else []
|
||||
self._valid = valid
|
||||
self._type = layer_type
|
||||
self._geometry_type = geometry_type
|
||||
self._crs = _FakeCrs(authid)
|
||||
self._name = name
|
||||
|
||||
def isValid(self) -> bool:
|
||||
return self._valid
|
||||
|
||||
def type(self) -> int:
|
||||
return self._type
|
||||
|
||||
def getFeatures(self, request: Any = None) -> List[Any]:
|
||||
return list(self._features)
|
||||
|
||||
def geometryType(self) -> int:
|
||||
return self._geometry_type
|
||||
|
||||
def crs(self) -> _FakeCrs:
|
||||
return self._crs
|
||||
|
||||
def name(self) -> str:
|
||||
return self._name
|
||||
|
||||
def fields(self) -> list:
|
||||
return ["feld1"]
|
||||
|
||||
|
||||
def _make_loader() -> tuple[LayerLoader, _AufzeichnenderPruefmanager]:
|
||||
pruefmanager = _AufzeichnenderPruefmanager()
|
||||
return LayerLoader(pruefmanager=cast(Any, pruefmanager)), pruefmanager
|
||||
|
||||
|
||||
class CreateLayerTests(unittest.TestCase):
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer")
|
||||
def test_wfs_baut_uri_ohne_getfeature_parameter(self, mock_cls):
|
||||
"""Prüft, dass für WFS die GetFeature-spezifischen Parameter aus der URI entfernt werden."""
|
||||
mock_cls.return_value = MagicMock(isValid=lambda: True)
|
||||
loader, _pm = _make_loader()
|
||||
|
||||
loader.create_layer("wfs", "http://server/wfs?typename=Flurstueck&request=GetFeature&BBOX=1,2,3,4", "Thema")
|
||||
|
||||
uri = mock_cls.call_args[0][0]
|
||||
self.assertIn("url=http://server/wfs?", uri)
|
||||
self.assertIn("typename=Flurstueck", uri)
|
||||
self.assertNotIn("request=", uri)
|
||||
self.assertNotIn("BBOX", uri)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer")
|
||||
def test_wfs_mit_url_praefix_wird_nicht_doppelt_gesetzt(self, mock_cls):
|
||||
"""Prüft, dass ein bereits vorhandenes 'url='-Präfix im WFS-Link nicht verdoppelt wird."""
|
||||
mock_cls.return_value = MagicMock(isValid=lambda: True)
|
||||
loader, _pm = _make_loader()
|
||||
|
||||
loader.create_layer("wfs", "url=http://server/wfs?typename=Flurstueck", "Thema")
|
||||
|
||||
uri = mock_cls.call_args[0][0]
|
||||
self.assertNotIn("url=url=", uri)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsRasterLayer")
|
||||
def test_wms_baut_url_praefix_und_nutzt_rasterlayer(self, mock_cls):
|
||||
"""Prüft, dass WMS-Layer über QgsRasterLayer mit 'url='-Präfix erzeugt werden."""
|
||||
mock_cls.return_value = MagicMock(isValid=lambda: True)
|
||||
loader, _pm = _make_loader()
|
||||
|
||||
loader.create_layer("wms", "http://server/wms", "Thema")
|
||||
|
||||
uri = mock_cls.call_args[0][0]
|
||||
self.assertEqual(uri, "url=http://server/wms")
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer")
|
||||
def test_rest_featureserver_ergaenzt_layer_index(self, mock_cls):
|
||||
"""Prüft, dass ein REST-Link auf '.../FeatureServer' automatisch um '/0' ergänzt wird."""
|
||||
mock_cls.return_value = MagicMock(isValid=lambda: True)
|
||||
loader, _pm = _make_loader()
|
||||
|
||||
loader.create_layer("rest", "http://server/arcgis/rest/services/x/FeatureServer", "Thema")
|
||||
|
||||
uri = mock_cls.call_args[0][0]
|
||||
self.assertTrue(uri.startswith("url=") and uri.endswith("/FeatureServer/0"))
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer")
|
||||
def test_ogr_provider_nutzt_link_direkt(self, mock_cls):
|
||||
"""Prüft, dass ein OGR-Provider den Link unverändert an QgsVectorLayer übergibt."""
|
||||
mock_cls.return_value = MagicMock(isValid=lambda: True)
|
||||
loader, _pm = _make_loader()
|
||||
|
||||
loader.create_layer("ogr", "C:/daten/flurstuecke.gpkg", "Thema")
|
||||
|
||||
mock_cls.assert_called_once_with("C:/daten/flurstuecke.gpkg", "Thema", "ogr")
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer")
|
||||
def test_ungueltiger_layer_meldet_layer_nicht_verfuegbar(self, mock_cls):
|
||||
"""Prüft, dass ein ungültig erzeugter Layer als 'layer_nicht_verfuegbar' gemeldet wird und None liefert."""
|
||||
mock_cls.return_value = MagicMock(isValid=lambda: False)
|
||||
loader, pruefmanager = _make_loader()
|
||||
|
||||
ergebnis = loader.create_layer("ogr", "C:/daten/x.gpkg", "Thema")
|
||||
|
||||
self.assertIsNone(ergebnis)
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("layer_nicht_verfuegbar", aktionen)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer")
|
||||
def test_ausnahme_beim_erstellen_wird_abgefangen(self, mock_cls):
|
||||
"""Prüft, dass eine Ausnahme beim Layer-Erstellen abgefangen und als Fehlerergebnis gemeldet wird."""
|
||||
mock_cls.side_effect = RuntimeError("kaputt")
|
||||
loader, pruefmanager = _make_loader()
|
||||
|
||||
ergebnis = loader.create_layer("ogr", "C:/daten/x.gpkg", "Thema")
|
||||
|
||||
self.assertIsNone(ergebnis)
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("layer_nicht_verfuegbar", aktionen)
|
||||
|
||||
|
||||
class TransformGeometryTests(unittest.TestCase):
|
||||
def test_fehlender_layer_liefert_geometrie_unveraendert(self):
|
||||
"""Prüft, dass ohne Quell- oder Ziel-Layer die Geometrie unverändert zurückgegeben wird."""
|
||||
loader, _pm = _make_loader()
|
||||
geom = _FakeGeometry()
|
||||
|
||||
ergebnis = loader._transform_geometry_to_layer_crs(geom, None, _FakeSourceLayer())
|
||||
|
||||
self.assertIs(ergebnis, geom)
|
||||
|
||||
def test_gleiche_crs_liefert_geometrie_unveraendert(self):
|
||||
"""Prüft, dass bei identischem CRS von Quelle und Ziel keine Transformation stattfindet."""
|
||||
loader, _pm = _make_loader()
|
||||
geom = _FakeGeometry()
|
||||
quelle = _FakeSourceLayer(authid="EPSG:25833")
|
||||
ziel = _FakeSourceLayer(authid="EPSG:25833")
|
||||
|
||||
ergebnis = loader._transform_geometry_to_layer_crs(geom, quelle, ziel)
|
||||
|
||||
self.assertIs(ergebnis, geom)
|
||||
self.assertFalse(geom.transformed)
|
||||
|
||||
def test_unterschiedliche_crs_klont_und_transformiert_geometrie(self):
|
||||
"""Prüft, dass bei unterschiedlichem CRS eine geklonte, transformierte Geometrie geliefert wird."""
|
||||
loader, _pm = _make_loader()
|
||||
geom = _FakeGeometry()
|
||||
quelle = _FakeSourceLayer(authid="EPSG:4326")
|
||||
ziel = _FakeSourceLayer(authid="EPSG:25833")
|
||||
|
||||
ergebnis = loader._transform_geometry_to_layer_crs(geom, quelle, ziel)
|
||||
|
||||
self.assertIsNot(ergebnis, geom)
|
||||
self.assertTrue(ergebnis.transformed)
|
||||
|
||||
|
||||
class TransformExtentTests(unittest.TestCase):
|
||||
def test_gleiche_crs_liefert_extent_unveraendert(self):
|
||||
"""Prüft, dass bei identischem CRS von Quelle und Ziel der Extent unverändert bleibt."""
|
||||
loader, _pm = _make_loader()
|
||||
quelle = _FakeSourceLayer(authid="EPSG:25833")
|
||||
ziel = _FakeSourceLayer(authid="EPSG:25833")
|
||||
|
||||
ergebnis = loader._transform_extent_to_layer_crs("EXTENT", quelle, ziel)
|
||||
|
||||
self.assertEqual(ergebnis, "EXTENT")
|
||||
|
||||
def test_unterschiedliche_crs_nutzt_coordinate_transform(self):
|
||||
"""Prüft, dass bei unterschiedlichem CRS der Extent über QgsCoordinateTransform umgerechnet wird."""
|
||||
|
||||
class _FakeTransform:
|
||||
def __init__(self, source_crs: Any, target_crs: Any, project: Any) -> None:
|
||||
pass
|
||||
|
||||
def transformBoundingBox(self, rect: Any) -> str:
|
||||
return "TRANSFORMIERT"
|
||||
|
||||
loader, _pm = _make_loader()
|
||||
quelle = _FakeSourceLayer(authid="EPSG:4326")
|
||||
ziel = _FakeSourceLayer(authid="EPSG:25833")
|
||||
|
||||
with patch("sn_basis.modules.LayerLoader.QgsCoordinateTransform", _FakeTransform):
|
||||
ergebnis = loader._transform_extent_to_layer_crs("EXTENT", quelle, ziel)
|
||||
|
||||
self.assertEqual(ergebnis, "TRANSFORMIERT")
|
||||
|
||||
|
||||
class FilterByExtentTests(unittest.TestCase):
|
||||
def test_ungueltiger_layer_wird_unveraendert_zurueckgegeben(self):
|
||||
"""Prüft, dass ein ungültiger Layer von filter_by_extent unverändert zurückgegeben wird."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer(valid=False)
|
||||
|
||||
self.assertIs(loader.filter_by_extent(layer, "EXTENT"), layer)
|
||||
|
||||
def test_ohne_extent_wird_layer_unveraendert_zurueckgegeben(self):
|
||||
"""Prüft, dass ohne Extent (None) der Layer unverändert zurückgegeben wird."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer()
|
||||
|
||||
self.assertIs(loader.filter_by_extent(layer, None), layer)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer", _FakeFilteredLayer)
|
||||
def test_falscher_layertyp_wird_unveraendert_zurueckgegeben(self):
|
||||
"""Prüft, dass ein Nicht-Vektor-Layer (z. B. Rasterlayer) unverändert zurückgegeben wird."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer(layer_type=2)
|
||||
|
||||
self.assertIs(loader.filter_by_extent(layer, "EXTENT"), layer)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer", _FakeFilteredLayer)
|
||||
def test_keine_treffer_liefert_none(self):
|
||||
"""Prüft, dass ein Raumfilter ohne Treffer None statt eines leeren Layers liefert."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer(features=[])
|
||||
|
||||
self.assertIsNone(loader.filter_by_extent(layer, "EXTENT"))
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer", _FakeFilteredLayer)
|
||||
def test_abbruch_liefert_none_und_meldet_needs_user_action(self):
|
||||
"""Prüft, dass ein Abbruch während des Raumfilters None liefert und gemeldet wird."""
|
||||
loader, pruefmanager = _make_loader()
|
||||
layer = _FakeSourceLayer(features=[_FakeFeature()])
|
||||
|
||||
ergebnis = loader.filter_by_extent(layer, "EXTENT", cancel_callback=lambda: True)
|
||||
|
||||
self.assertIsNone(ergebnis)
|
||||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||||
self.assertIn("needs_user_action", aktionen)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer", _FakeFilteredLayer)
|
||||
def test_erfolgreicher_filter_liefert_neuen_layer_mit_features(self):
|
||||
"""Prüft, dass ein erfolgreicher BBOX-Filter einen neuen Speicherlayer mit den gefundenen Features liefert."""
|
||||
loader, _pm = _make_loader()
|
||||
features = [_FakeFeature(), _FakeFeature()]
|
||||
layer = _FakeSourceLayer(features=features, name="Flurstuecke")
|
||||
|
||||
ergebnis = loader.filter_by_extent(layer, "EXTENT")
|
||||
|
||||
self.assertIsInstance(ergebnis, _FakeFilteredLayer)
|
||||
self.assertEqual(ergebnis.name(), "Flurstuecke_bbox")
|
||||
self.assertEqual(len(ergebnis.dataProvider().added_features), 2)
|
||||
|
||||
|
||||
class FilterByLayerTests(unittest.TestCase):
|
||||
def test_ungueltiger_layer_oder_filterlayer_wird_unveraendert_zurueckgegeben(self):
|
||||
"""Prüft, dass ein ungültiger Haupt- oder Filter-Layer unverändert zurückgegeben wird."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer()
|
||||
ungueltiger_filter = _FakeSourceLayer(valid=False)
|
||||
|
||||
self.assertIs(loader.filter_by_layer(layer, ungueltiger_filter), layer)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer", _FakeFilteredLayer)
|
||||
def test_filterlayer_ohne_features_liefert_none(self):
|
||||
"""Prüft, dass ein Filter-Layer ohne Features (leere Vereinigungsgeometrie) None liefert."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer(features=[_FakeFeature()])
|
||||
filter_layer = _FakeSourceLayer(features=[])
|
||||
|
||||
self.assertIsNone(loader.filter_by_layer(layer, filter_layer))
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer", _FakeFilteredLayer)
|
||||
def test_erfolgreicher_geometriefilter_liefert_neuen_layer(self):
|
||||
"""Prüft, dass ein erfolgreicher Geometriefilter einen neuen Speicherlayer mit den passenden Features liefert."""
|
||||
loader, _pm = _make_loader()
|
||||
filter_layer = _FakeSourceLayer(features=[_FakeFeature()], name="Verfahrensgebiet")
|
||||
layer = _FakeSourceLayer(features=[_FakeFeature(), _FakeFeature()], name="Flurstuecke")
|
||||
|
||||
ergebnis = loader.filter_by_layer(layer, filter_layer)
|
||||
|
||||
self.assertIsInstance(ergebnis, _FakeFilteredLayer)
|
||||
self.assertEqual(ergebnis.name(), "Flurstuecke_filtered")
|
||||
self.assertEqual(len(ergebnis.dataProvider().added_features), 2)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsVectorLayer", _FakeFilteredLayer)
|
||||
def test_keine_ueberschneidung_liefert_none(self):
|
||||
"""Prüft, dass ohne überschneidende Features None statt eines leeren Layers geliefert wird."""
|
||||
|
||||
class _NichtSchneidendeGeometrie(_FakeGeometry):
|
||||
def intersects(self, other: Any) -> bool:
|
||||
return False
|
||||
|
||||
loader, _pm = _make_loader()
|
||||
filter_layer = _FakeSourceLayer(features=[_FakeFeature()], name="Verfahrensgebiet")
|
||||
layer = _FakeSourceLayer(features=[_FakeFeature(_NichtSchneidendeGeometrie())], name="Flurstuecke")
|
||||
|
||||
self.assertIsNone(loader.filter_by_layer(layer, filter_layer))
|
||||
|
||||
|
||||
class AddToProjectTests(unittest.TestCase):
|
||||
@patch("sn_basis.modules.LayerLoader.QgsProject")
|
||||
def test_gueltiger_layer_wird_dem_projekt_hinzugefuegt(self, mock_project):
|
||||
"""Prüft, dass ein gültiger Layer über QgsProject.instance().addMapLayer hinzugefügt wird."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer(valid=True)
|
||||
|
||||
loader.add_to_project(cast(Any, layer))
|
||||
|
||||
mock_project.instance.return_value.addMapLayer.assert_called_once_with(layer)
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsProject")
|
||||
def test_ungueltiger_layer_wird_nicht_hinzugefuegt(self, mock_project):
|
||||
"""Prüft, dass ein ungültiger Layer nicht ans Projekt übergeben wird."""
|
||||
loader, _pm = _make_loader()
|
||||
layer = _FakeSourceLayer(valid=False)
|
||||
|
||||
loader.add_to_project(cast(Any, layer))
|
||||
|
||||
mock_project.instance.return_value.addMapLayer.assert_not_called()
|
||||
|
||||
@patch("sn_basis.modules.LayerLoader.QgsProject")
|
||||
def test_none_layer_wird_nicht_hinzugefuegt(self, mock_project):
|
||||
"""Prüft, dass ein fehlender Layer (None) nicht ans Projekt übergeben wird."""
|
||||
loader, _pm = _make_loader()
|
||||
|
||||
loader.add_to_project(cast(Any, None))
|
||||
|
||||
mock_project.instance.return_value.addMapLayer.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user