From 189855e63b811dea71a5b38620f848765c4470d4 Mon Sep 17 00:00:00 2001 From: daniel Date: Tue, 21 Apr 2026 21:50:26 +0200 Subject: [PATCH] Flst_Laden, VerfGebiet bilden --- functions/flurstueck_alkis.py | 113 ++++ functions/verfahrensgebiet_edit_service.py | 358 ++++++++++++ styles/GIS_Flst_Beschriftung_ALKIS_NAS.qml | 349 ++++++++++++ styles/verfahrensgebiet.qml | 215 +++---- tests/test_flurstueck_alkis.py | 223 ++++++++ tests/test_verfahrensgebiet_edit_service.py | 595 ++++++++++++++++++++ tests/test_working_tab.py | 283 ++++++++++ ui/tabs/working_tab.py | 235 +++++++- 8 files changed, 2247 insertions(+), 124 deletions(-) create mode 100644 functions/flurstueck_alkis.py create mode 100644 functions/verfahrensgebiet_edit_service.py create mode 100644 styles/GIS_Flst_Beschriftung_ALKIS_NAS.qml create mode 100644 tests/test_flurstueck_alkis.py create mode 100644 tests/test_verfahrensgebiet_edit_service.py diff --git a/functions/flurstueck_alkis.py b/functions/flurstueck_alkis.py new file mode 100644 index 0000000..40876b2 --- /dev/null +++ b/functions/flurstueck_alkis.py @@ -0,0 +1,113 @@ +"""sn_verfahrensgebiet/functions/flurstueck_alkis.py – ALKIS-Integration für Flurstücke. + +Lädt Flurstücke aus dem ALKIS-WFS-Dienst (adv:AX_Flurstueck) anhand von +Gemarkungsschlüsseln, die der Nutzer interaktiv eingibt. + +Der WFS-Layer wird direkt (nicht als Memory-Layer) ins Projekt geladen, +da die einzelnen Features für die spätere Auswahl benötigt werden. +""" + +from sn_basis.functions import QgsVectorLayer, QgsProject +from sn_basis.functions import QMessageBox +from sn_basis.functions.dialog_wrapper import ask_text + +# LoadStatus aus verfahrensgebiet_alkis wiederverwenden (identische Semantik) +from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus + +ALKIS_NAS_URL = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs" +FLST_TYPENAME = "adv:AX_Flurstueck" +FLST_LAYER_NAME = "Flurstücke" + + +def _check_existing_flurstueck_layer(tab_widget): + """Prüft, ob ein Flurstück-Layer bereits im Projekt existiert. + + Returns + ------- + Tuple[Optional[QgsVectorLayer], LoadStatus] + Bei ``LoadStatus.KEEP`` enthält das Tuple den vorhandenen Layer, + bei ``LoadStatus.FIRST`` und ``LoadStatus.RELOAD`` ist es None. + """ + project = QgsProject.instance() + existing_layers = project.mapLayersByName(FLST_LAYER_NAME) + if not existing_layers: + return None, LoadStatus.FIRST + + reply = QMessageBox.question( + tab_widget, + "Layer bereits vorhanden", + f"Der Layer '{FLST_LAYER_NAME}' existiert bereits.\n" + "Möchten Sie ihn löschen und neu laden?", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No, + ) + if reply == QMessageBox.StandardButton.Yes: + for lyr in existing_layers: + project.removeMapLayer(lyr.id()) + return None, LoadStatus.RELOAD + else: + return existing_layers[0], LoadStatus.KEEP + + +def flurstueck_alkis(tab_widget): + """Lädt Flurstücke aus ALKIS-WFS anhand von Gemarkungsschlüsseln. + + Der Nutzer gibt einen oder mehrere Gemarkungsschlüssel (kommagetrennt) ein. + Der WFS wird per SQL-Filter auf + ``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN (...)`` abgefragt. + + Parameters + ---------- + tab_widget : + Eltern-Widget für Dialoge. + + Returns + ------- + Tuple[Optional[QgsVectorLayer], LoadStatus] + Geladener WFS-Layer und Ladestatus. + Bei Fehlern ist der Layer None und der Status LoadStatus.NONE. + """ + # 1. Vorhandenen Layer prüfen + existing, status = _check_existing_flurstueck_layer(tab_widget) + if status == LoadStatus.KEEP: + return existing, status + + # 2. Gemarkungsschlüssel abfragen + eingabe, ok = ask_text( + "Gemarkungsschlüssel", + "Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):", + parent=tab_widget, + ) + if not ok or not eingabe.strip(): + QMessageBox.warning( + tab_widget, + "Abgebrochen", + "Bitte mindestens einen gültigen Gemarkungsschlüssel eingeben.", + ) + return None, LoadStatus.NONE + + schluessel_liste = [s.strip() for s in eingabe.split(",") if s.strip()] + if not schluessel_liste: + QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.") + return None, LoadStatus.NONE + + schluessel_str = ", ".join(f"'{s}'" for s in schluessel_liste) + + # 3. WFS-URI aufbauen und Layer laden + wfs_uri = ( + f"url='{ALKIS_NAS_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' " + f"sql=SELECT * FROM AX_Flurstueck " + f"WHERE gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN ({schluessel_str})" + ) + layer = QgsVectorLayer(wfs_uri, FLST_LAYER_NAME, "WFS") + + if not layer.isValid(): + QMessageBox.critical( + tab_widget, + "Fehler", + "Flurstück-Layer konnte nicht geladen werden.\n" + "Bitte Gemarkungsschlüssel und Netzwerkverbindung prüfen.", + ) + return None, LoadStatus.NONE + + return layer, status diff --git a/functions/verfahrensgebiet_edit_service.py b/functions/verfahrensgebiet_edit_service.py new file mode 100644 index 0000000..cea40d7 --- /dev/null +++ b/functions/verfahrensgebiet_edit_service.py @@ -0,0 +1,358 @@ +"""sn_verfahrensgebiet/functions/verfahrensgebiet_edit_service.py + +Service-Modul für Verfahrensgebiet-Editoperationen aus Flurstück-Auswahl. + +Der Verfahrensgebiet-Layer ist ein Single-Feature-Memory-Layer: alle +Flurstück-Geometrien werden via ``QgsGeometry.unaryUnion`` zu einem einzigen +(Multi-)Polygon verschmolzen und als ein Feature gespeichert. Dadurch +erscheint das Verfahrensgebiet im Kartenfenster als zusammenhängendes Objekt. + +Die Flurstückskennzeichen werden als kommagetrennte Zeichenkette im +Attributfeld ``flurstueckskennzeichen`` des Single-Features gespeichert, +sodass Delta-Operationen (Ergänzen/Entfernen) ohne erneuten Zugriff auf den +Quell-Flurstück-Layer möglich bleiben und die ID-Menge via +``get_kennzeichen_menge()`` jederzeit abrufbar ist. +""" + +from typing import Optional, Tuple, List + +from sn_basis.functions import QgsVectorLayer, QgsFeature, QgsField +from sn_basis.functions.qgiscore_wrapper import QgsGeometry +from sn_basis.functions.qt_wrapper import QVariant +from sn_basis.modules.layerpruefer import Layerpruefer +from sn_basis.modules.pruef_ergebnis import pruef_ergebnis + +VERF_LAYER_NAME = "Verfahrensgebiet" +FELD_KENNZEICHEN = "flurstueckskennzeichen" +FELD_GEMARKUNG = "gemarkung_ax_gemarkung_schluessel_gemarkungsnummer" +PFLICHTFELDER: List[str] = [FELD_KENNZEICHEN, FELD_GEMARKUNG] + + +# --------------------------------------------------------------------------- +# Vorbedingungsprüfungen +# --------------------------------------------------------------------------- + +def pruefe_flurstueck_layer(layer) -> pruef_ergebnis: + """Prüft, ob ein Layer als Flurstück-Quelle geeignet ist. + + Prüfreihenfolge via Layerpruefer: + existiert → Vektorlayer → Polygon-Geometrie → nicht leer → Pflichtfelder vorhanden. + + Returns + ------- + pruef_ergebnis + ok=True, wenn alle Prüfungen bestanden wurden. + """ + pruefer = Layerpruefer( + layer=layer, + erwarteter_layertyp="vector", + erwarteter_geotyp="Polygon", + erforderliche_felder=PFLICHTFELDER, + ) + return pruefer.pruefe() + + +def pruefe_auswahl(layer) -> pruef_ergebnis: + """Prüft, ob auf dem Layer mindestens ein Feature ausgewählt ist. + + Returns + ------- + pruef_ergebnis + ok=True bei mindestens einer Auswahl, sonst ok=False mit aktion ``layer_leer``. + """ + try: + count = int(layer.selectedFeatureCount()) + except Exception: + count = 0 + + if count == 0: + return pruef_ergebnis( + ok=False, + meldung="Es sind keine Flurstücke ausgewählt. Bitte zuerst eine Auswahl treffen.", + aktion="layer_leer", + kontext=None, + ) + return pruef_ergebnis( + ok=True, + meldung=f"{count} Flurstücke ausgewählt.", + aktion="ok", + kontext=None, + ) + + +# --------------------------------------------------------------------------- +# Hilfsfunktionen +# --------------------------------------------------------------------------- + +def get_kennzeichen_menge(layer) -> set: + """Gibt die Menge aller ``flurstueckskennzeichen`` aus dem Layer zurück. + + Im Single-Feature-Modell sind die Kennzeichen als kommagetrennte + Zeichenkette im einzigen Feature gespeichert. + """ + result = set() + for feat in layer.getFeatures(): + val = feat[FELD_KENNZEICHEN] + if val: + for kz in str(val).split(","): + kz = kz.strip() + if kz: + result.add(kz) + return result + + +def _erstelle_memory_layer(crs_authid: str) -> QgsVectorLayer: + """Erzeugt einen leeren Multi-Feature-Memory-Layer für das Verfahrensgebiet.""" + layer = QgsVectorLayer( + f"MultiPolygon?crs={crs_authid}", VERF_LAYER_NAME, "memory" + ) + pr = layer.dataProvider() + pr.addAttributes([ + QgsField(FELD_KENNZEICHEN, QVariant.String), + QgsField(FELD_GEMARKUNG, QVariant.String), + ]) + layer.updateFields() + return layer + + +# --------------------------------------------------------------------------- +# Hauptoperationen +# --------------------------------------------------------------------------- + +def erstelle_verfahrensgebiet_aus_auswahl( + source_layer, +) -> Tuple[Optional[object], int, int]: + """Erstellt neuen Verfahrensgebiet-Memory-Layer aus der aktuellen Auswahl. + + Alle Geometrien werden via ``QgsGeometry.unaryUnion`` zu einem einzigen + (Multi-)Polygon verschmolzen. Der Layer enthält danach genau ein Feature. + Die Flurstückskennzeichen werden kommagetrennt im Attribut gespeichert. + + Parameters + ---------- + source_layer : + Flurstück-Layer mit aktiver Auswahl. + + Returns + ------- + Tuple[layer, erfolgreich, fehlgeschlagen] + """ + crs = source_layer.crs().authid() + mem_layer = _erstelle_memory_layer(crs) + pr = mem_layer.dataProvider() + + erfolgreich = 0 + fehlgeschlagen = 0 + geom_list = [] + kz_list = [] + gemarkung_set = set() + + for src_feat in source_layer.selectedFeatures(): + try: + geom = src_feat.geometry() + if not geom or geom.isEmpty(): + fehlgeschlagen += 1 + continue + geom_list.append(geom) + kz = str(src_feat[FELD_KENNZEICHEN] or "") + gm = str(src_feat[FELD_GEMARKUNG] or "") + if kz: + kz_list.append(kz) + if gm: + gemarkung_set.add(gm) + erfolgreich += 1 + except Exception: + fehlgeschlagen += 1 + + if geom_list: + union_geom = QgsGeometry.unaryUnion(geom_list) + if union_geom and not union_geom.isEmpty(): + feat = QgsFeature() + feat.setGeometry(union_geom) + feat.setAttributes([ + ",".join(kz_list), + ",".join(sorted(gemarkung_set)), + ]) + pr.addFeatures([feat]) + mem_layer.updateExtents() + + return mem_layer, erfolgreich, fehlgeschlagen + + +def ergaenze_verfahrensgebiet( + verf_layer, + source_layer, +) -> Tuple[object, int, int, int]: + """Fügt nur neue (noch nicht enthaltene) Flurstücke zum Verfahrensgebiet hinzu. + + Die Geometrien der neuen Flurstücke werden mit der bestehenden Union-Geometrie + verschmolzen. Das Single-Feature wird an Ort und Stelle aktualisiert. + + Parameters + ---------- + verf_layer : + Bestehender Verfahrensgebiet-Memory-Layer (Single-Feature). + source_layer : + Flurstück-Layer mit aktiver Auswahl. + + Returns + ------- + Tuple[verf_layer, hinzugefuegt, bereits_vorhanden, fehlgeschlagen] + """ + vorhandene_ids = get_kennzeichen_menge(verf_layer) + pr = verf_layer.dataProvider() + + hinzugefuegt = 0 + bereits_vorhanden = 0 + fehlgeschlagen = 0 + neue_geoms = [] + neue_kz = [] + neue_gemarkungen: set = set() + + for src_feat in source_layer.selectedFeatures(): + try: + kz = str(src_feat[FELD_KENNZEICHEN] or "") + if kz in vorhandene_ids: + bereits_vorhanden += 1 + continue + geom = src_feat.geometry() + if not geom or geom.isEmpty(): + fehlgeschlagen += 1 + continue + neue_geoms.append(geom) + if kz: + neue_kz.append(kz) + gm = str(src_feat[FELD_GEMARKUNG] or "") + if gm: + neue_gemarkungen.add(gm) + hinzugefuegt += 1 + except Exception: + fehlgeschlagen += 1 + + if neue_geoms: + existing_feats = list(verf_layer.getFeatures()) + if existing_feats: + existing_feat = existing_feats[0] + all_geoms = [existing_feat.geometry()] + neue_geoms + new_union = QgsGeometry.unaryUnion(all_geoms) + alle_kz = sorted(vorhandene_ids | set(neue_kz)) + existing_gm_str = existing_feat[FELD_GEMARKUNG] or "" + alle_gm = sorted( + {g.strip() for g in existing_gm_str.split(",") if g.strip()} | neue_gemarkungen + ) + kz_idx = verf_layer.fields().indexFromName(FELD_KENNZEICHEN) + gm_idx = verf_layer.fields().indexFromName(FELD_GEMARKUNG) + pr.changeAttributeValues({existing_feat.id(): { + kz_idx: ",".join(alle_kz), + gm_idx: ",".join(alle_gm), + }}) + pr.changeGeometryValues({existing_feat.id(): new_union}) + else: + # Kein vorhandenes Feature – neues anlegen + union_geom = QgsGeometry.unaryUnion(neue_geoms) + if union_geom and not union_geom.isEmpty(): + feat = QgsFeature() + feat.setGeometry(union_geom) + feat.setAttributes([ + ",".join(sorted(neue_kz)), + ",".join(sorted(neue_gemarkungen)), + ]) + pr.addFeatures([feat]) + verf_layer.updateExtents() + verf_layer.triggerRepaint() + + return verf_layer, hinzugefuegt, bereits_vorhanden, fehlgeschlagen + + +def entferne_aus_verfahrensgebiet( + verf_layer, + source_layer, +) -> Tuple[object, int, int, int]: + """Entfernt Flurstücke aus dem Verfahrensgebiet. + + Die Geometrien der zu entfernenden Flurstücke werden per ``difference()`` + von der bestehenden Union-Geometrie subtrahiert. + Nicht enthaltene Flurstücke werden gezählt, aber ignoriert. + + Parameters + ---------- + verf_layer : + Bestehender Verfahrensgebiet-Memory-Layer (Single-Feature). + source_layer : + Flurstück-Layer mit aktiver Auswahl. + + Returns + ------- + Tuple[verf_layer, entfernt, nicht_enthalten, fehlgeschlagen] + """ + existing_feats = list(verf_layer.getFeatures()) + if not existing_feats: + auswahl_kz = { + str(f[FELD_KENNZEICHEN] or "") + for f in source_layer.selectedFeatures() + if f[FELD_KENNZEICHEN] + } + return verf_layer, 0, len(auswahl_kz), 0 + + existing_feat = existing_feats[0] + existing_kz_str = existing_feat[FELD_KENNZEICHEN] or "" + vorhandene_ids = {k.strip() for k in existing_kz_str.split(",") if k.strip()} + + auswahl_kz = { + str(f[FELD_KENNZEICHEN] or "") + for f in source_layer.selectedFeatures() + if f[FELD_KENNZEICHEN] + } + nicht_enthalten = len(auswahl_kz - vorhandene_ids) + zu_entfernen_kz = auswahl_kz & vorhandene_ids + + if not zu_entfernen_kz: + return verf_layer, 0, nicht_enthalten, 0 + + result_geom = existing_feat.geometry() + fehlgeschlagen = 0 + tatsaechlich_entfernt = 0 + + for src_feat in source_layer.selectedFeatures(): + kz = str(src_feat[FELD_KENNZEICHEN] or "") + if kz not in zu_entfernen_kz: + continue + try: + geom = src_feat.geometry() + if geom and not geom.isEmpty(): + result_geom = result_geom.difference(geom) + tatsaechlich_entfernt += 1 + except Exception: + fehlgeschlagen += 1 + + pr = verf_layer.dataProvider() + verbleibende_kz = sorted(vorhandene_ids - zu_entfernen_kz) + + if not verbleibende_kz or result_geom is None or result_geom.isEmpty(): + pr.deleteFeatures([existing_feat.id()]) + else: + kz_idx = verf_layer.fields().indexFromName(FELD_KENNZEICHEN) + pr.changeGeometryValues({existing_feat.id(): result_geom}) + pr.changeAttributeValues({existing_feat.id(): { + kz_idx: ",".join(verbleibende_kz), + }}) + + verf_layer.updateExtents() + verf_layer.triggerRepaint() + return verf_layer, tatsaechlich_entfernt, nicht_enthalten, fehlgeschlagen + + +def ersetze_verfahrensgebiet( + source_layer, +) -> Tuple[Optional[object], int, int]: + """Erstellt einen neuen Verfahrensgebiet-Layer aus der Auswahl (Ersetzen-Semantik). + + Der Aufrufer ist verantwortlich, den alten Layer aus dem Projekt zu entfernen, + bevor der neue Layer hinzugefügt wird. + + Returns + ------- + Tuple[layer, erfolgreich, fehlgeschlagen] + Entspricht ``erstelle_verfahrensgebiet_aus_auswahl``. + """ + return erstelle_verfahrensgebiet_aus_auswahl(source_layer) diff --git a/styles/GIS_Flst_Beschriftung_ALKIS_NAS.qml b/styles/GIS_Flst_Beschriftung_ALKIS_NAS.qml new file mode 100644 index 0000000..5e40734 --- /dev/null +++ b/styles/GIS_Flst_Beschriftung_ALKIS_NAS.qml @@ -0,0 +1,349 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 0 + 0 + 2 + diff --git a/styles/verfahrensgebiet.qml b/styles/verfahrensgebiet.qml index 474e368..35485d7 100644 --- a/styles/verfahrensgebiet.qml +++ b/styles/verfahrensgebiet.qml @@ -1,133 +1,142 @@ - - - - - - - - - + + + + + - - - - + - - - - - - + + + + + + + + + + + + + + + + + + - + - + diff --git a/tests/test_flurstueck_alkis.py b/tests/test_flurstueck_alkis.py new file mode 100644 index 0000000..b49e6d0 --- /dev/null +++ b/tests/test_flurstueck_alkis.py @@ -0,0 +1,223 @@ +"""sn_verfahrensgebiet/tests/test_flurstueck_alkis.py + +Unit-Tests für das flurstueck_alkis-Modul. + +Getestet werden: +- Nutzerabbruch bei ask_text → (None, LoadStatus.NONE) +- Ungültiger WFS-Layer → (None, LoadStatus.NONE) +- Erfolgreicher Ladevorgang → gültiger Layer + LoadStatus.FIRST +- Bestehender Layer: Nutzer wählt "Neu laden" → LoadStatus.RELOAD +- Bestehender Layer: Nutzer wählt "Behalten" → LoadStatus.KEEP +""" + +import sys +import types +import unittest +from unittest.mock import patch, MagicMock + +# --------------------------------------------------------------------------- +# Minimales QGIS-Mock-Setup +# --------------------------------------------------------------------------- +if "qgis" not in sys.modules: + qgis_mod = types.ModuleType("qgis") + qgis_core = types.ModuleType("qgis.core") + qgis_utils = types.ModuleType("qgis.utils") + qgis_utils.iface = types.SimpleNamespace() + qgis_utils.plugins = {} + qgis_mod.utils = qgis_utils + qgis_mod.core = qgis_core + sys.modules["qgis"] = qgis_mod + sys.modules["qgis.core"] = qgis_core + sys.modules["qgis.utils"] = qgis_utils + +from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis, FLST_LAYER_NAME +from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus + + +# --------------------------------------------------------------------------- +# Dummies +# --------------------------------------------------------------------------- + +class _DummyLayer: + def __init__(self, valid: bool, layer_id="L1"): + self._valid = valid + self._id = layer_id + + def isValid(self): + return self._valid + + def id(self): + return self._id + + def extent(self): + return object() + + +class _DummyProject: + def __init__(self, named_layers=None): + self._named_layers = named_layers or [] + self.removed = [] + + def mapLayersByName(self, name): + return self._named_layers + + def removeMapLayer(self, lid): + self.removed.append(lid) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestFlurstueckAlkis(unittest.TestCase): + + # ------------------------------------------------------------------ # + # 1. Nutzerabbruch (ask_text liefert ok=False / None) # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QMessageBox") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_user_cancel_returns_none(self, mock_ask_text, mock_project, mock_msgbox): + """ask_text → ok=False, text="" → (None, NONE).""" + mock_ask_text.return_value = ("", False) + mock_project.return_value = _DummyProject() + + layer, status = flurstueck_alkis(None) + + self.assertIsNone(layer) + self.assertEqual(status, LoadStatus.NONE) + + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QMessageBox") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_empty_text_after_confirmation_returns_none(self, mock_ask_text, mock_project, mock_msgbox): + """ask_text → ok=True, aber Text leer → (None, NONE).""" + mock_ask_text.return_value = (" ", True) + mock_project.return_value = _DummyProject() + + layer, status = flurstueck_alkis(None) + + self.assertIsNone(layer) + self.assertEqual(status, LoadStatus.NONE) + + # ------------------------------------------------------------------ # + # 2. Ungültiger WFS-Layer # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QMessageBox") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_invalid_wfs_layer_returns_none( + self, mock_ask_text, mock_project, mock_wfs_layer_cls, mock_msgbox + ): + """WFS-Layer nicht valide → (None, NONE).""" + mock_ask_text.return_value = ("1234", True) + mock_project.return_value = _DummyProject() + mock_wfs_layer_cls.return_value = _DummyLayer(valid=False) + + layer, status = flurstueck_alkis(None) + + self.assertIsNone(layer) + self.assertEqual(status, LoadStatus.NONE) + + # ------------------------------------------------------------------ # + # 3. Erfolgreicher Erstladevorgang # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_success_returns_layer_and_first_status( + self, mock_ask_text, mock_project, mock_wfs_layer_cls + ): + """Kein bestehender Layer, WFS valide → (layer, FIRST).""" + mock_ask_text.return_value = ("1234, 5678", True) + mock_project.return_value = _DummyProject() + valid_layer = _DummyLayer(valid=True, layer_id="wfs_ok") + mock_wfs_layer_cls.return_value = valid_layer + + layer, status = flurstueck_alkis(None) + + self.assertIs(layer, valid_layer) + self.assertEqual(status, LoadStatus.FIRST) + + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_sql_filter_contains_all_schluessel( + self, mock_ask_text, mock_project, mock_wfs_layer_cls + ): + """Der WFS-URI enthält alle eingegebenen Gemarkungsschlüssel.""" + mock_ask_text.return_value = ("1111, 2222, 3333", True) + mock_project.return_value = _DummyProject() + mock_wfs_layer_cls.return_value = _DummyLayer(valid=True) + + flurstueck_alkis(None) + + # Erster Positionsarg von QgsVectorLayer ist der URI-String + call_args = mock_wfs_layer_cls.call_args + uri_str = call_args[0][0] if call_args[0] else call_args[1].get("path", "") + self.assertIn("1111", uri_str) + self.assertIn("2222", uri_str) + self.assertIn("3333", uri_str) + + # ------------------------------------------------------------------ # + # 4. Bestehender Layer – Nutzer wählt "Neu laden" # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QMessageBox") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_existing_layer_reload_yes( + self, mock_ask_text, mock_project, mock_msgbox_cls, mock_wfs_layer_cls + ): + """Bestehender Layer + Nutzer: 'Ja, neu laden' → LoadStatus.RELOAD.""" + existing = _DummyLayer(valid=True, layer_id="old_flst") + project = _DummyProject(named_layers=[existing]) + mock_ask_text.return_value = ("1234", True) + mock_project.return_value = project + + # StandardButton als int, damit | Operator funktioniert + mock_msgbox_cls.StandardButton.Yes = 0x4000 + mock_msgbox_cls.StandardButton.No = 0x10000 + # QMessageBox.question → Yes + mock_msgbox_cls.question.return_value = 0x4000 + + new_layer = _DummyLayer(valid=True, layer_id="new_flst") + mock_wfs_layer_cls.return_value = new_layer + + layer, status = flurstueck_alkis(None) + + self.assertIn(existing.id(), project.removed) + self.assertIs(layer, new_layer) + self.assertEqual(status, LoadStatus.RELOAD) + + # ------------------------------------------------------------------ # + # 5. Bestehender Layer – Nutzer wählt "Behalten" # + # ------------------------------------------------------------------ # + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QMessageBox") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance") + @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text") + def test_existing_layer_keep_no( + self, mock_ask_text, mock_project, mock_msgbox_cls + ): + """Bestehender Layer + Nutzer: 'Nein, behalten' → LoadStatus.KEEP.""" + existing = _DummyLayer(valid=True, layer_id="keep_flst") + project = _DummyProject(named_layers=[existing]) + mock_ask_text.return_value = ("1234", True) + mock_project.return_value = project + + # StandardButton als int, damit | Operator funktioniert + mock_msgbox_cls.StandardButton.Yes = 0x4000 + mock_msgbox_cls.StandardButton.No = 0x10000 + # QMessageBox.question → No + mock_msgbox_cls.question.return_value = 0x10000 + + layer, status = flurstueck_alkis(None) + + self.assertIs(layer, existing) + self.assertEqual(status, LoadStatus.KEEP) + self.assertEqual(project.removed, []) # nichts gelöscht + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_verfahrensgebiet_edit_service.py b/tests/test_verfahrensgebiet_edit_service.py new file mode 100644 index 0000000..8679e4e --- /dev/null +++ b/tests/test_verfahrensgebiet_edit_service.py @@ -0,0 +1,595 @@ +"""sn_verfahrensgebiet/tests/test_verfahrensgebiet_edit_service.py + +Unit-Tests für das Verfahrensgebiet-Edit-Service-Modul. + +Getestet werden: +- pruefe_flurstueck_layer (über Layerpruefer-Integration) +- pruefe_auswahl +- erstelle_verfahrensgebiet_aus_auswahl: Erfolgs- und Fehlerpfade +- ergaenze_verfahrensgebiet: nur neue IDs werden hinzugefügt +- entferne_aus_verfahrensgebiet: nur vorhandene IDs werden entfernt +- ersetze_verfahrensgebiet: Ergebnis entspricht Erstellen +- get_kennzeichen_menge: Hilfsfunktion +- Zähler (erfolgreich/fehlgeschlagen/bereits_vorhanden/nicht_enthalten) +""" + +import sys +import types +import unittest +from unittest.mock import patch, MagicMock + +# --------------------------------------------------------------------------- +# Minimales QGIS-Mock-Setup (muss vor allen sn_*-Importen erfolgen) +# --------------------------------------------------------------------------- +if "qgis" not in sys.modules: + qgis_mod = types.ModuleType("qgis") + qgis_core = types.ModuleType("qgis.core") + qgis_utils = types.ModuleType("qgis.utils") + qgis_utils.iface = types.SimpleNamespace() + qgis_utils.plugins = {} + qgis_mod.utils = qgis_utils + qgis_mod.core = qgis_core + sys.modules["qgis"] = qgis_mod + sys.modules["qgis.core"] = qgis_core + sys.modules["qgis.utils"] = qgis_utils + +from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import ( + pruefe_flurstueck_layer, + pruefe_auswahl, + erstelle_verfahrensgebiet_aus_auswahl, + ergaenze_verfahrensgebiet, + entferne_aus_verfahrensgebiet, + ersetze_verfahrensgebiet, + get_kennzeichen_menge, + FELD_KENNZEICHEN, + FELD_GEMARKUNG, +) + + +# --------------------------------------------------------------------------- +# Test-Dummies +# --------------------------------------------------------------------------- + +class _DummyGeom: + def __init__(self, empty=False): + self._empty = empty + + def isEmpty(self): + return self._empty + + def difference(self, other): + """Dummy-Differenz: immer nicht-leeres Ergebnis (kein echter Geometrieschnitt).""" + return _DummyGeom(empty=False) + + +class _DummyFeature: + """Feature-Dummy mit dict-artigem Attributzugriff und konfigurierbarer Geometrie.""" + + def __init__(self, kz: str, gemarkung: str = "1234", geom_empty: bool = False, feat_id: int = 0): + self._attrs = {FELD_KENNZEICHEN: kz, FELD_GEMARKUNG: gemarkung} + self._geom_empty = geom_empty + self._id = feat_id + self._set_geom = None + self._set_attrs = None + + def __getitem__(self, key): + return self._attrs.get(key, None) + + def geometry(self): + return _DummyGeom(self._geom_empty) + + def id(self): + return self._id + + def setGeometry(self, g): + self._set_geom = g + + def setAttributes(self, attrs): + self._set_attrs = attrs + + +class _DummyProvider: + """DataProvider-Dummy für Single-Feature-Verfahrensgebiet-Layer.""" + + def __init__(self, layer_ref): + self._layer = layer_ref + + def addAttributes(self, attrs): + pass + + def addFeatures(self, feats): + self._layer._features.extend(feats) + return True + + def deleteFeatures(self, ids): + self._layer._deleted_ids.extend(ids) + return True + + def changeAttributeValues(self, changes: dict): + """changes: {fid: {field_idx: value}}""" + for feat in self._layer._features: + fid = feat.id() if callable(getattr(feat, "id", None)) else None + if fid in changes: + for field_idx, value in changes[fid].items(): + feat._attr_by_idx[field_idx] = value + return True + + def changeGeometryValues(self, changes: dict): + """changes: {fid: new_geom}""" + for feat in self._layer._features: + fid = feat.id() if callable(getattr(feat, "id", None)) else None + if fid in changes: + feat._set_geom = changes[fid] + return True + + +class _DummySingleFeature: + """Single-Feature-Dummy für den Verfahrensgebiet-Layer. + + Speichert Kennzeichen als kommagetrennte Zeichenkette (new-style). + Feld-Indizes: 0=FELD_KENNZEICHEN, 1=FELD_GEMARKUNG + """ + + def __init__(self, kz_komma: str, gemarkung: str = "", feat_id: int = 0, geom_empty: bool = False): + self._kz_komma = kz_komma + self._gemarkung = gemarkung + self._id = feat_id + self._geom_empty = geom_empty + self._set_geom = None + # Attribut-Mapping nach Index für changeAttributeValues + self._attr_by_idx: dict = {} + + def __getitem__(self, key): + if key == FELD_KENNZEICHEN: + return self._attr_by_idx.get(0, self._kz_komma) + if key == FELD_GEMARKUNG: + return self._attr_by_idx.get(1, self._gemarkung) + return None + + def geometry(self): + return _DummyGeom(self._geom_empty) + + def id(self): + return self._id + + def setGeometry(self, g): + self._set_geom = g + + def setAttributes(self, attrs): + if len(attrs) > 0: + self._kz_komma = attrs[0] + if len(attrs) > 1: + self._gemarkung = attrs[1] + + +class _DummyVerfLayer: + """Memory-Layer-Dummy für Verfahrensgebiet (Single-Feature-Modell).""" + + def __init__(self, single_feature=None): + self._features = [single_feature] if single_feature else [] + self._deleted_ids = [] + + def getFeatures(self): + deleted = set(self._deleted_ids) + return [ + f for f in self._features + if (f.id() if callable(getattr(f, "id", None)) else None) not in deleted + ] + + def dataProvider(self): + return _DummyProvider(self) + + def fields(self): + class _Fields: + def indexFromName(self, name): + if name == FELD_KENNZEICHEN: + return 0 + if name == FELD_GEMARKUNG: + return 1 + return -1 + return _Fields() + + def updateExtents(self): + pass + + def triggerRepaint(self): + pass + + def updateFields(self): + pass + + def isValid(self): + return True + + def crs(self): + class _Crs: + def authid(self): + return "EPSG:25833" + return _Crs() + + +class _DummySourceLayer: + """Flurstück-Quell-Layer-Dummy mit konfigurierbarer Auswahl.""" + + def __init__(self, selected=None, all_features=None, feature_count=None): + self._selected = list(selected or []) + self._all = list(all_features or []) + self._count = feature_count if feature_count is not None else len(self._selected) + + def selectedFeatures(self): + return self._selected + + def selectedFeatureCount(self): + return self._count + + def getFeatures(self): + return self._all + + def crs(self): + class _Crs: + def authid(self): + return "EPSG:25833" + return _Crs() + + +class _MockLayerpruefer: + """Steuerbarer Mock für Layerpruefer.pruefe().""" + + def __init__(self, ok, aktion="ok", meldung=""): + self._ok = ok + self._aktion = aktion + self._meldung = meldung + + def pruefe(self): + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis + return pruef_ergebnis(ok=self._ok, aktion=self._aktion, meldung=self._meldung) + + +# --------------------------------------------------------------------------- +# Tests: pruefe_flurstueck_layer +# --------------------------------------------------------------------------- + +class TestPruefeFlurstueckLayer(unittest.TestCase): + + def test_layer_ok_passes(self): + """Gültiger Polygon-Vektor-Layer mit Pflichtfeldern → ok=True.""" + import sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service as module + original = module.Layerpruefer + + class _PassPruefer: + def __init__(self, **_): + pass + def pruefe(self): + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis + return pruef_ergebnis(ok=True, aktion="ok") + + module.Layerpruefer = _PassPruefer + try: + ergebnis = pruefe_flurstueck_layer(object()) + self.assertTrue(ergebnis.ok) + self.assertEqual(ergebnis.aktion, "ok") + finally: + module.Layerpruefer = original + + def test_layer_missing_field_fails(self): + """Layer ohne Pflichtfelder → ok=False, aktion=felder_fehlen.""" + import sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service as module + original = module.Layerpruefer + + class _FailPruefer: + def __init__(self, **_): + pass + def pruefe(self): + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis + return pruef_ergebnis(ok=False, aktion="felder_fehlen", + meldung="Feld fehlt") + + module.Layerpruefer = _FailPruefer + try: + ergebnis = pruefe_flurstueck_layer(object()) + self.assertFalse(ergebnis.ok) + self.assertEqual(ergebnis.aktion, "felder_fehlen") + finally: + module.Layerpruefer = original + + def test_layer_wrong_geometry_fails(self): + """Layer mit falschem Geometrietyp → ok=False, aktion=falscher_geotyp.""" + import sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service as module + original = module.Layerpruefer + + class _WrongGeomPruefer: + def __init__(self, **_): + pass + def pruefe(self): + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis + return pruef_ergebnis(ok=False, aktion="falscher_geotyp", + meldung="Kein Polygon") + + module.Layerpruefer = _WrongGeomPruefer + try: + ergebnis = pruefe_flurstueck_layer(object()) + self.assertFalse(ergebnis.ok) + self.assertEqual(ergebnis.aktion, "falscher_geotyp") + finally: + module.Layerpruefer = original + + +# --------------------------------------------------------------------------- +# Tests: pruefe_auswahl +# --------------------------------------------------------------------------- + +class TestPruefeAuswahl(unittest.TestCase): + + def test_no_selection_fails(self): + layer = _DummySourceLayer(selected=[], feature_count=0) + ergebnis = pruefe_auswahl(layer) + self.assertFalse(ergebnis.ok) + self.assertEqual(ergebnis.aktion, "layer_leer") + + def test_with_selection_passes(self): + feats = [_DummyFeature("1000-001")] + layer = _DummySourceLayer(selected=feats) + ergebnis = pruefe_auswahl(layer) + self.assertTrue(ergebnis.ok) + self.assertIn("1", ergebnis.meldung) # "1 Flurstücke ausgewählt." + + def test_selectedFeatureCount_exception_treated_as_zero(self): + class _BrokenLayer: + def selectedFeatureCount(self): + raise RuntimeError("broken") + ergebnis = pruefe_auswahl(_BrokenLayer()) + self.assertFalse(ergebnis.ok) + + +# --------------------------------------------------------------------------- +# Tests: get_kennzeichen_menge +# --------------------------------------------------------------------------- + +class TestGetKennzeichenMenge(unittest.TestCase): + + def test_returns_set_of_ids(self): + """Kommagetrennte Kennzeichen im Single-Feature werden korrekt geparst.""" + feat = _DummySingleFeature("A001,A002,A001", feat_id=1) # Duplikat → set + layer = _DummyVerfLayer(feat) + result = get_kennzeichen_menge(layer) + self.assertEqual(result, {"A001", "A002"}) + + def test_empty_layer_returns_empty_set(self): + layer = _DummyVerfLayer() + result = get_kennzeichen_menge(layer) + self.assertEqual(result, set()) + + def test_single_kennzeichen_without_comma(self): + """Einzelnes Kennzeichen ohne Komma wird korrekt erkannt.""" + feat = _DummySingleFeature("X999", feat_id=1) + layer = _DummyVerfLayer(feat) + result = get_kennzeichen_menge(layer) + self.assertEqual(result, {"X999"}) + + +# --------------------------------------------------------------------------- +# Tests: erstelle_verfahrensgebiet_aus_auswahl +# --------------------------------------------------------------------------- + +class TestErstelle(unittest.TestCase): + + def _patch_all(self, test_func, union_empty=False): + """Patcht _erstelle_memory_layer und QgsGeometry.unaryUnion.""" + import sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service as module + orig_erstelle = module._erstelle_memory_layer + orig_union = module.QgsGeometry.unaryUnion + + mem = _DummyVerfLayer() + module._erstelle_memory_layer = lambda _crs: mem + module.QgsGeometry.unaryUnion = staticmethod( + lambda _geoms: _DummyGeom(empty=union_empty) + ) + try: + test_func(mem) + finally: + module._erstelle_memory_layer = orig_erstelle + module.QgsGeometry.unaryUnion = staticmethod(orig_union) + + def test_success_counts(self): + """3 gültige Features → erfolgreich=3, fehlgeschlagen=0, 1 Union-Feature im Layer.""" + feats = [ + _DummyFeature("A001", feat_id=0), + _DummyFeature("A002", feat_id=1), + _DummyFeature("A003", feat_id=2), + ] + source = _DummySourceLayer(selected=feats) + + def run(mem): + layer, ok, fail = erstelle_verfahrensgebiet_aus_auswahl(source) + self.assertIs(layer, mem) + self.assertEqual(ok, 3) + self.assertEqual(fail, 0) + # Single-Feature-Modell: genau 1 Union-Feature + self.assertEqual(len(mem._features), 1) + + self._patch_all(run) + + def test_partial_failure_on_empty_geometry(self): + """2 ok + 1 leere Geometrie → erfolgreich=2, fehlgeschlagen=1.""" + feats = [ + _DummyFeature("A001"), + _DummyFeature("A002", geom_empty=True), + _DummyFeature("A003"), + ] + source = _DummySourceLayer(selected=feats) + + def run(mem): + _, ok, fail = erstelle_verfahrensgebiet_aus_auswahl(source) + self.assertEqual(ok, 2) + self.assertEqual(fail, 1) + + self._patch_all(run) + + def test_all_fail_on_empty_geometries(self): + """Alle Features haben leere Geometrie → erfolgreich=0, kein Feature im Layer.""" + feats = [_DummyFeature("A001", geom_empty=True)] + source = _DummySourceLayer(selected=feats) + + def run(mem): + _, ok, fail = erstelle_verfahrensgebiet_aus_auswahl(source) + self.assertEqual(ok, 0) + self.assertEqual(fail, 1) + self.assertEqual(len(mem._features), 0) # kein Union-Feature + + self._patch_all(run) + + +# --------------------------------------------------------------------------- +# Tests: ergaenze_verfahrensgebiet +# --------------------------------------------------------------------------- + +class TestErgaenzeVerfahrensgebiet(unittest.TestCase): + + def _patch_union(self): + import sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service as module + orig = module.QgsGeometry.unaryUnion + module.QgsGeometry.unaryUnion = staticmethod(lambda _g: _DummyGeom(empty=False)) + return module, orig + + def test_adds_only_new_ids(self): + """Auswahl enthält 1 neue + 1 bereits vorhandene ID → hinzugefügt=1.""" + verf = _DummyVerfLayer(_DummySingleFeature("A001", feat_id=1)) + selected = [ + _DummyFeature("A001"), # bereits vorhanden + _DummyFeature("A002"), # neu + ] + source = _DummySourceLayer(selected=selected) + module, orig = self._patch_union() + try: + _, hinzugefuegt, bereits_vorhanden, fehlgeschlagen = ergaenze_verfahrensgebiet( + verf, source + ) + finally: + module.QgsGeometry.unaryUnion = staticmethod(orig) + self.assertEqual(hinzugefuegt, 1) + self.assertEqual(bereits_vorhanden, 1) + self.assertEqual(fehlgeschlagen, 0) + + def test_skip_empty_geometry_counts_as_failed(self): + """Neue ID mit leerer Geometrie → fehlgeschlagen=1, hinzugefügt=0.""" + verf = _DummyVerfLayer() # leer + selected = [_DummyFeature("A001", geom_empty=True)] + source = _DummySourceLayer(selected=selected) + + _, hinzugefuegt, _, fehlgeschlagen = ergaenze_verfahrensgebiet(verf, source) + self.assertEqual(hinzugefuegt, 0) + self.assertEqual(fehlgeschlagen, 1) + + def test_all_already_present(self): + """Alle ausgewählten IDs bereits vorhanden → bereits_vorhanden=n, hinzugefügt=0.""" + verf = _DummyVerfLayer(_DummySingleFeature("A001,A002", feat_id=1)) + selected = [_DummyFeature("A001"), _DummyFeature("A002")] + source = _DummySourceLayer(selected=selected) + + _, hinzugefuegt, bereits_vorhanden, fehlgeschlagen = ergaenze_verfahrensgebiet( + verf, source + ) + self.assertEqual(hinzugefuegt, 0) + self.assertEqual(bereits_vorhanden, 2) + self.assertEqual(fehlgeschlagen, 0) + + +# --------------------------------------------------------------------------- +# Tests: entferne_aus_verfahrensgebiet +# --------------------------------------------------------------------------- + +class TestEntferneAusVerfahrensgebiet(unittest.TestCase): + + def test_removes_present_ids(self): + """Auswahl enthält 1 vorhandene ID → entfernt=1, nicht_enthalten=0.""" + verf = _DummyVerfLayer(_DummySingleFeature("A001,A002", feat_id=1)) + selected = [_DummyFeature("A001")] + source = _DummySourceLayer(selected=selected) + + _, entfernt, nicht_enthalten, fehlgeschlagen = entferne_aus_verfahrensgebiet( + verf, source + ) + self.assertEqual(entfernt, 1) + self.assertEqual(nicht_enthalten, 0) + self.assertEqual(fehlgeschlagen, 0) + + def test_counts_ids_not_in_layer(self): + """Auswahl enthält ID, die nicht im Verfahrensgebiet ist → nicht_enthalten=1.""" + verf = _DummyVerfLayer(_DummySingleFeature("A001", feat_id=1)) + selected = [ + _DummyFeature("A001"), # vorhanden + _DummyFeature("A999"), # nicht enthalten + ] + source = _DummySourceLayer(selected=selected) + + _, entfernt, nicht_enthalten, fehlgeschlagen = entferne_aus_verfahrensgebiet( + verf, source + ) + self.assertEqual(entfernt, 1) + self.assertEqual(nicht_enthalten, 1) + + def test_none_present_in_layer(self): + """Auswahl hat IDs, die alle nicht im Layer sind → entfernt=0, nicht_enthalten=n.""" + verf = _DummyVerfLayer(_DummySingleFeature("X001", feat_id=1)) + selected = [_DummyFeature("A999"), _DummyFeature("B888")] + source = _DummySourceLayer(selected=selected) + + _, entfernt, nicht_enthalten, fehlgeschlagen = entferne_aus_verfahrensgebiet( + verf, source + ) + self.assertEqual(entfernt, 0) + self.assertEqual(nicht_enthalten, 2) + + def test_empty_selection_removes_nothing(self): + """Leere Auswahl → nichts entfernt, nichts nicht_enthalten.""" + verf = _DummyVerfLayer(_DummySingleFeature("A001", feat_id=1)) + source = _DummySourceLayer(selected=[]) + + _, entfernt, nicht_enthalten, fehlgeschlagen = entferne_aus_verfahrensgebiet( + verf, source + ) + self.assertEqual(entfernt, 0) + self.assertEqual(nicht_enthalten, 0) + + def test_removes_last_id_deletes_feature(self): + """Wird das letzte Kennzeichen entfernt, wird das Feature gelöscht.""" + verf = _DummyVerfLayer(_DummySingleFeature("A001", feat_id=1)) + selected = [_DummyFeature("A001")] + source = _DummySourceLayer(selected=selected) + + _, entfernt, _, _ = entferne_aus_verfahrensgebiet(verf, source) + self.assertEqual(entfernt, 1) + # Feature soll gelöscht worden sein + self.assertIn(1, verf._deleted_ids) + + +# --------------------------------------------------------------------------- +# Tests: ersetze_verfahrensgebiet +# --------------------------------------------------------------------------- + +class TestErsetze(unittest.TestCase): + + def test_ersetze_delegates_to_erstelle(self): + """ersetze_verfahrensgebiet gibt dasselbe Ergebnis wie erstelle zurück.""" + feats = [_DummyFeature("A001")] + source = _DummySourceLayer(selected=feats) + + import sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service as module + orig_erstelle = module._erstelle_memory_layer + orig_union = module.QgsGeometry.unaryUnion + mem = _DummyVerfLayer() + module._erstelle_memory_layer = lambda _crs: mem + module.QgsGeometry.unaryUnion = staticmethod(lambda _g: _DummyGeom(empty=False)) + + try: + layer, ok, fail = ersetze_verfahrensgebiet(source) + self.assertIs(layer, mem) + self.assertEqual(ok, 1) + self.assertEqual(fail, 0) + finally: + module._erstelle_memory_layer = orig_erstelle + module.QgsGeometry.unaryUnion = staticmethod(orig_union) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_working_tab.py b/tests/test_working_tab.py index 683dd27..c2c23cc 100644 --- a/tests/test_working_tab.py +++ b/tests/test_working_tab.py @@ -82,9 +82,17 @@ class TestWorkingTab(unittest.TestCase): def _make_tab(self): tab = WorkingTab.__new__(WorkingTab) tab.btn_verf_alkis = _DummyButton() + tab.btn_flurst_alkis = _DummyButton() + tab.btn_verf_aus_auswahl = _DummyButton() + tab.btn_verf_aus_liste = _DummyButton() tab.verf_layer_combo = _DummyCombo() tab.haken_verf = object() + tab.haken_flurst = object() tab.setze_haken = lambda *_args, **_kwargs: None + # Pruefmanager-Mock: verarbeite() wird im Test ggf. überschrieben + tab.pruefmanager = types.SimpleNamespace( + verarbeite=lambda ergebnis: ergebnis + ) return tab def test_set_busy_updates_button_state(self): @@ -151,6 +159,281 @@ class TestWorkingTab(unittest.TestCase): mock_set_layer_id.assert_called_once_with("") + # ------------------------------------------------------------------ + # handle_flurst_alkis + # ------------------------------------------------------------------ + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.success") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.flurstueck_alkis") + def test_handle_flurst_alkis_success_sets_haken_and_style( + self, mock_alkis, mock_iface, mock_project, mock_apply_style, mock_success + ): + tab = self._make_tab() + haken_calls = [] + tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v)) + + layer = _DummyLayer(valid=True) + mock_alkis.return_value = (layer, LoadStatus.FIRST) + mock_project.return_value = _DummyProject() + mock_iface.mapCanvas.return_value = _DummyCanvas() + + tab.handle_flurst_alkis() + + mock_apply_style.assert_called_once_with(layer, "GIS_Flst_Beschriftung_ALKIS_NAS.qml") + mock_success.assert_called_once() + self.assertEqual(len(haken_calls), 1) + self.assertTrue(haken_calls[0][1]) # haken gesetzt + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.info") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.flurstueck_alkis") + def test_handle_flurst_alkis_existing_layer_keep_path(self, mock_alkis, mock_info): + tab = self._make_tab() + haken_calls = [] + tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v)) + + layer = _DummyLayer(valid=True) + mock_alkis.return_value = (layer, LoadStatus.KEEP) + + tab.handle_flurst_alkis() + + mock_info.assert_called_once() + self.assertEqual(len(haken_calls), 1) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.flurstueck_alkis") + def test_handle_flurst_alkis_invalid_input_returns_early(self, mock_alkis): + """Status NONE (z. B. Nutzerabbruch) → kein Layer, kein Haken.""" + tab = self._make_tab() + haken_calls = [] + tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v)) + mock_alkis.return_value = (None, LoadStatus.NONE) + + tab.handle_flurst_alkis() + + self.assertEqual(len(haken_calls), 0) + # Button muss am Ende wieder freigegeben sein (finally) + self.assertTrue(tab.btn_flurst_alkis.isEnabled()) + + # ------------------------------------------------------------------ + # handle_verf_aus_auswahl – Vorbedingungsprüfungen + # ------------------------------------------------------------------ + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.warning") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + def test_handle_verfahrensgebiet_from_selection_requires_active_layer( + self, mock_iface, mock_warning + ): + """Kein aktiver Layer → warning, kein Layer wird erstellt.""" + tab = self._make_tab() + mock_iface.activeLayer.return_value = None + + tab.handle_verf_aus_auswahl() + + mock_warning.assert_called_once() + self.assertIn("aktiver Layer", mock_warning.call_args[0][1]) + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.warning") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_flurstueck_layer") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + def test_handle_verfahrensgebiet_from_selection_requires_required_fields( + self, mock_iface, mock_pruefe, mock_warning + ): + """Layer ohne Pflichtfelder → warning, kein Layer wird erstellt.""" + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis as PruefErg + tab = self._make_tab() + active = _DummyLayer(valid=True) + mock_iface.activeLayer.return_value = active + mock_pruefe.return_value = PruefErg( + ok=False, aktion="felder_fehlen", meldung="Feld fehlt" + ) + + tab.handle_verf_aus_auswahl() + + mock_warning.assert_called_once() + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.warning") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_auswahl") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_flurstueck_layer") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + def test_handle_verfahrensgebiet_from_selection_requires_selection( + self, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, mock_warning + ): + """Keine Flurstück-Auswahl → warning, kein Layer erstellt.""" + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis as PruefErg + tab = self._make_tab() + active = _DummyLayer(valid=True) + mock_iface.activeLayer.return_value = active + mock_pruefe_layer.return_value = PruefErg(ok=True, aktion="ok") + mock_pruefe_auswahl.return_value = PruefErg( + ok=False, aktion="layer_leer", meldung="Keine Auswahl" + ) + + tab.handle_verf_aus_auswahl() + + mock_warning.assert_called_once() + + # ------------------------------------------------------------------ + # handle_verf_aus_auswahl – Erstlauf (kein bestehender Layer) + # ------------------------------------------------------------------ + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QMessageBox") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.erstelle_verfahrensgebiet_aus_auswahl") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_auswahl") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_flurstueck_layer") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + def test_handle_verfahrensgebiet_from_selection_new_layer_success( + self, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, + mock_erstelle, mock_project_inst, mock_apply_style, mock_msgbox + ): + """Erstlauf ohne bestehenden Layer → Layer wird erstellt und hinzugefügt.""" + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis as PruefErg + tab = self._make_tab() + haken_calls = [] + tab.setze_haken = lambda lbl, v: haken_calls.append(v) + active = _DummyLayer(valid=True) + mock_iface.activeLayer.return_value = active + mock_iface.mapCanvas.return_value = _DummyCanvas() + mock_pruefe_layer.return_value = PruefErg(ok=True, aktion="ok") + mock_pruefe_auswahl.return_value = PruefErg(ok=True, aktion="ok", meldung="3 Flurstücke") + + new_layer = _DummyLayer(valid=True, layer_id="new_verf") + mock_erstelle.return_value = (new_layer, 3, 0) + + project = _DummyProject() + project.mapLayersByName = lambda _name: [] # kein bestehender Layer + mock_project_inst.return_value = project + + tab.handle_verf_aus_auswahl() + + self.assertIn(new_layer, project.added) + mock_apply_style.assert_called_once_with(new_layer, "verfahrensgebiet.qml") + self.assertTrue(haken_calls[0]) + + # ------------------------------------------------------------------ + # handle_verf_aus_auswahl – Ergänzen / Entfernen / Ersetzen + # ------------------------------------------------------------------ + + def _make_entscheidungs_ergebnis(self, aktion: str): + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis as PruefErg + return PruefErg(ok=True, aktion=aktion) + + def _setup_existierenden_layer(self, tab, mock_iface, mock_pruefe_layer, + mock_pruefe_auswahl, mock_project_inst, + mock_canvas, entscheidung: str): + """Gemeinsames Setup für Delta-Operationen mit bestehendem Layer.""" + from sn_basis.modules.pruef_ergebnis import pruef_ergebnis as PruefErg + active = _DummyLayer(valid=True) + mock_iface.activeLayer.return_value = active + mock_iface.mapCanvas.return_value = mock_canvas + mock_pruefe_layer.return_value = PruefErg(ok=True, aktion="ok") + mock_pruefe_auswahl.return_value = PruefErg(ok=True, aktion="ok", meldung="2 Flurstücke") + + verf_layer = _DummyLayer(valid=True, layer_id="verf_existing") + project = _DummyProject() + project.added = [] + project.mapLayersByName = lambda _: [verf_layer] + project.removeMapLayer = lambda lid: None + mock_project_inst.return_value = project + + # Pruefmanager gibt Entscheidung zurück + tab.pruefmanager.verarbeite = lambda _e: self._make_entscheidungs_ergebnis(entscheidung) + + return active, verf_layer, project + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QMessageBox") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.ergaenze_verfahrensgebiet") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_auswahl") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_flurstueck_layer") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + def test_handle_verfahrensgebiet_from_selection_existing_layer_ergaenzen_dispatch( + self, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, + mock_project_inst, mock_ergaenze, mock_apply_style, mock_msgbox + ): + """Entscheidung 'ergaenzen' → ergaenze_verfahrensgebiet wird aufgerufen.""" + tab = self._make_tab() + canvas = _DummyCanvas() + active, verf_layer, _ = self._setup_existierenden_layer( + tab, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, + mock_project_inst, canvas, "verfahrensgebiet_ergaenzen" + ) + mock_ergaenze.return_value = (verf_layer, 2, 0, 0) + + tab.handle_verf_aus_auswahl() + + mock_ergaenze.assert_called_once_with(verf_layer, active) + mock_apply_style.assert_called_once_with(verf_layer, "verfahrensgebiet.qml") + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QMessageBox") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.entferne_aus_verfahrensgebiet") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_auswahl") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_flurstueck_layer") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + def test_handle_verfahrensgebiet_from_selection_existing_layer_entfernen_dispatch( + self, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, + mock_project_inst, mock_entferne, mock_apply_style, mock_msgbox + ): + """Entscheidung 'entfernen' → entferne_aus_verfahrensgebiet wird aufgerufen.""" + tab = self._make_tab() + canvas = _DummyCanvas() + active, verf_layer, _ = self._setup_existierenden_layer( + tab, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, + mock_project_inst, canvas, "verfahrensgebiet_entfernen" + ) + mock_entferne.return_value = (verf_layer, 1, 1, 0) + + tab.handle_verf_aus_auswahl() + + mock_entferne.assert_called_once_with(verf_layer, active) + mock_apply_style.assert_called_once_with(verf_layer, "verfahrensgebiet.qml") + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QMessageBox") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.ersetze_verfahrensgebiet") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_auswahl") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.pruefe_flurstueck_layer") + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface") + def test_handle_verfahrensgebiet_from_selection_existing_layer_ersetzen_dispatch( + self, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, + mock_project_inst, mock_ersetze, mock_apply_style, mock_msgbox + ): + """Entscheidung 'ersetzen' → alter Layer entfernt, neuer hinzugefügt.""" + tab = self._make_tab() + canvas = _DummyCanvas() + active, verf_layer, project = self._setup_existierenden_layer( + tab, mock_iface, mock_pruefe_layer, mock_pruefe_auswahl, + mock_project_inst, canvas, "verfahrensgebiet_ersetzen" + ) + removed = [] + project.removeMapLayer = lambda lid: removed.append(lid) + new_layer = _DummyLayer(valid=True, layer_id="new_verf_replaced") + mock_ersetze.return_value = (new_layer, 3, 0) + + tab.handle_verf_aus_auswahl() + + mock_ersetze.assert_called_once_with(active) + self.assertIn(verf_layer.id(), removed) + self.assertIn(new_layer, project.added) + mock_apply_style.assert_called_once_with(new_layer, "verfahrensgebiet.qml") + + # ------------------------------------------------------------------ + # handle_verf_aus_liste + # ------------------------------------------------------------------ + + @patch("sn_verfahrensgebiet.ui.tabs.working_tab.show_info_dialog") + def test_aus_liste_button_shows_not_implemented_hint(self, mock_info): + tab = self._make_tab() + tab.handle_verf_aus_liste() + mock_info.assert_called_once() + if __name__ == "__main__": unittest.main() diff --git a/ui/tabs/working_tab.py b/ui/tabs/working_tab.py index b0519db..1f47ee9 100644 --- a/ui/tabs/working_tab.py +++ b/ui/tabs/working_tab.py @@ -8,9 +8,21 @@ from sn_basis.functions import Qgis, QgsProject #QgsMessageLog from sn_basis.functions.qgisui_wrapper import QgsMapLayerComboBox from sn_basis.functions.qgiscore_wrapper import QgsMapLayerProxyModel from sn_basis.functions.verfahrensgebiet_manager import get_verfahrensgebiet_layer_id, set_verfahrensgebiet_layer_id +from sn_basis.functions.dialog_wrapper import show_info_dialog +from sn_basis.modules.Pruefmanager import Pruefmanager +from sn_basis.modules.pruef_ergebnis import pruef_ergebnis from qgis.utils import iface from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import verfahrensgebiet_alkis, LoadStatus +from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis +from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import ( + pruefe_flurstueck_layer, + pruefe_auswahl, + erstelle_verfahrensgebiet_aus_auswahl, + ergaenze_verfahrensgebiet, + entferne_aus_verfahrensgebiet, + ersetze_verfahrensgebiet, +) from sn_basis.functions.message_wrapper import success, info, warning, error from sn_basis.functions.ly_style_wrapper import apply_style @@ -24,17 +36,12 @@ class WorkingTab(QWidget): tab_title = "Bearbeitung" def __init__(self, parent=None): - """Initialisiert den Working-Tab mit Verfahrensgebiet-Verwaltung. - - Args: - parent: Übergeordnetes Widget - """ super().__init__(parent) layout = QVBoxLayout() layout.setAlignment(Qt.AlignmentFlag.AlignTop) - # Abschnitt 1: Verfahrensgebiet + # Abschnitt 1: Verfahrensgebiet laden row_verf_header = QHBoxLayout() lbl_verf = QLabel("Verfahrensgebiet") self.haken_verf = self._haken_label() @@ -64,8 +71,7 @@ class WorkingTab(QWidget): self.verf_layer_combo.layerChanged.connect(self._on_verf_layer_changed) layout.addWidget(self.verf_layer_combo) - # Abschnitt 2: Flurstücke (Platzhalter für spätere Implementierung) - + # Abschnitt 2: Flurstücke row_flurst_header = QHBoxLayout() lbl_flurst = QLabel("Flurstücke") self.haken_flurst = self._haken_label() @@ -81,20 +87,39 @@ class WorkingTab(QWidget): row_flurst_buttons.addWidget(self.btn_flurst_shape) layout.addLayout(row_flurst_buttons) + # Abschnitt 3: Verfahrensgebiet erstellen + row_verf_erstellen_header = QHBoxLayout() + lbl_verf_erstellen = QLabel("Verfahrensgebiet erstellen") + row_verf_erstellen_header.addWidget(lbl_verf_erstellen) + row_verf_erstellen_header.addStretch() + layout.addLayout(row_verf_erstellen_header) + + row_verf_erstellen_buttons = QHBoxLayout() + self.btn_verf_aus_auswahl = QPushButton("aus Auswahl") + self.btn_verf_aus_liste = QPushButton("Aus Liste") + row_verf_erstellen_buttons.addWidget(self.btn_verf_aus_auswahl) + row_verf_erstellen_buttons.addWidget(self.btn_verf_aus_liste) + layout.addLayout(row_verf_erstellen_buttons) + layout.addStretch() self.setLayout(layout) + # Pruefmanager für interaktive Entscheidungen (Ergänzen/Entfernen/Ersetzen) + self.pruefmanager = Pruefmanager(ui_modus="qgis", parent=self) + # Signale verbinden self.btn_verf_alkis.clicked.connect(self.handle_verf_alkis) + self.btn_flurst_alkis.clicked.connect(self.handle_flurst_alkis) + self.btn_verf_aus_auswahl.clicked.connect(self.handle_verf_aus_auswahl) + self.btn_verf_aus_liste.clicked.connect(self.handle_verf_aus_liste) # self.btn_verf_shape.clicked.connect(self.handle_verf_shape) - # self.btn_flurst_alkis.clicked.connect(self.handle_flurst_alkis) # self.btn_flurst_shape.clicked.connect(self.handle_flurst_shape) # Beim Init: aktuelle Verfahrensgebiet-Layer-ID laden und im Dropdown anzeigen self._restore_verf_layer_selection() def handle_verf_alkis(self): - self._set_busy(self.btn_verf_alkis, True) + self._set_busy(self.btn_verf_alkis, True, "Aus ALKIS laden") try: layer, status = verfahrensgebiet_alkis(self) if not layer or not layer.isValid(): @@ -102,20 +127,16 @@ class WorkingTab(QWidget): return if status in (LoadStatus.FIRST, LoadStatus.RELOAD): - # Gemeinsame Logik für erstmaliges und erneutes Laden QgsProject.instance().addMapLayer(layer) iface.mapCanvas().setExtent(layer.extent()) - #iface.mapCanvas().refresh() apply_style(layer, "verfahrensgebiet.qml") iface.mapCanvas().refresh() self.setze_haken(self.haken_verf, True) - # Dropdown + Variable explizit synchronisieren. self.verf_layer_combo.setLayer(layer) self._on_verf_layer_changed(layer) - # Unterschied nur in der Meldung msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen" success("Verfahrensgebiet", f"Layer wurde {msg}.", duration=5) @@ -129,14 +150,23 @@ class WorkingTab(QWidget): warning("Verfahrensgebiet", "Layer konnte nicht geladen werden.", duration=None) finally: - self._set_busy(self.btn_verf_alkis, False) + self._set_busy(self.btn_verf_alkis, False, "Aus ALKIS laden") - def _set_busy(self, button: QPushButton, busy: bool): + def _set_busy(self, button: QPushButton, busy: bool, original_text: str = "Aus ALKIS laden") -> None: + """Setzt den Busy-State eines Buttons. + + Parameters + ---------- + button : + Der zu sperrende/entsperrende Button. + busy : bool + True zum Sperren (Text → "Lade ..."), False zum Freigeben. + original_text : str + Text, der nach dem Freigeben wiederhergestellt wird. + Jeder Button gibt seinen eigenen Originaltext explizit an. + """ button.setEnabled(not busy) - if busy: - button.setText("Lade ...") - else: - button.setText("Aus ALKIS laden") + button.setText("Lade ..." if busy else original_text) def _haken_label(self) -> QLabel: label = QLabel() @@ -167,7 +197,6 @@ class WorkingTab(QWidget): """Beim Tab-Init: Lade die zuletzt gespeicherte Verfahrensgebiet-Layer-ID.""" layer_id = get_verfahrensgebiet_layer_id() if layer_id: - # Versuche, den Layer im Projekt zu finden try: project = QgsProject.instance() layer = project.mapLayer(layer_id) @@ -175,3 +204,167 @@ class WorkingTab(QWidget): self.verf_layer_combo.setLayer(layer) except Exception: pass + + # ------------------------------------------------------------------ + # Flurstücke-Handler + # ------------------------------------------------------------------ + + def handle_flurst_alkis(self) -> None: + """Lädt Flurstücke aus ALKIS-WFS anhand von Gemarkungsschlüsseln.""" + self._set_busy(self.btn_flurst_alkis, True, "Aus ALKIS laden") + try: + layer, status = flurstueck_alkis(self) + if not layer or not layer.isValid(): + return + + if status in (LoadStatus.FIRST, LoadStatus.RELOAD): + QgsProject.instance().addMapLayer(layer) + # iface.mapCanvas().setExtent(layer.extent()) # Kartenausschnitt bleibt unverändert + apply_style(layer, "GIS_Flst_Beschriftung_ALKIS_NAS.qml") + iface.mapCanvas().refresh() + self.setze_haken(self.haken_flurst, True) + msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen" + success("Flurstücke", f"Layer wurde {msg}.", duration=5) + + elif status == LoadStatus.KEEP: + self.setze_haken(self.haken_flurst, True) + info("Flurstücke", "Vorhandener Layer wurde behalten.", duration=4) + + finally: + self._set_busy(self.btn_flurst_alkis, False, "Aus ALKIS laden") + + # ------------------------------------------------------------------ + # Verfahrensgebiet-erstellen-Handler + # ------------------------------------------------------------------ + + def handle_verf_aus_auswahl(self) -> None: + """Erstellt oder aktualisiert den Verfahrensgebiet-Layer aus der Flurstück-Auswahl.""" + self._set_busy(self.btn_verf_aus_auswahl, True, "aus Auswahl") + try: + # 1. Aktiven Layer ermitteln + active_layer = iface.activeLayer() + if not active_layer: + warning( + "Verfahrensgebiet erstellen", + "Kein aktiver Layer. Bitte den Flurstück-Layer aktivieren.", + duration=None, + ) + return + + # 2. Vorbedingungen via Layerpruefer + ergebnis = pruefe_flurstueck_layer(active_layer) + if not ergebnis.ok: + warning("Verfahrensgebiet erstellen", ergebnis.meldung, duration=None) + return + + # 3. Auswahl prüfen + ergebnis_auswahl = pruefe_auswahl(active_layer) + if not ergebnis_auswahl.ok: + warning("Verfahrensgebiet erstellen", ergebnis_auswahl.meldung, duration=None) + return + + # 4. Bestehenden Verfahrensgebiet-Layer prüfen + existing_layers = QgsProject.instance().mapLayersByName("Verfahrensgebiet") + + if not existing_layers: + # Erstlauf: neuen Memory-Layer erzeugen + layer, erfolgreich, fehlgeschlagen = erstelle_verfahrensgebiet_aus_auswahl( + active_layer + ) + if not layer or not layer.isValid(): + error("Verfahrensgebiet erstellen", "Layer konnte nicht erstellt werden.") + return + QgsProject.instance().addMapLayer(layer) + apply_style(layer, "verfahrensgebiet.qml") + iface.mapCanvas().refresh() + self.verf_layer_combo.setLayer(layer) + self._on_verf_layer_changed(layer) + self.setze_haken(self.haken_verf, True) + QMessageBox.information( + self, + "Verfahrensgebiet erstellt", + f"Verfahrensgebiet aus Auswahl erstellt.\n" + f"Erfolgreich: {erfolgreich}\n" + f"Fehlgeschlagen: {fehlgeschlagen}", + ) + return + + # 5. Bestehender Layer: Nutzerentscheidung via Pruefmanager + entscheidung_ergebnis = self.pruefmanager.verarbeite( + pruef_ergebnis( + ok=False, + aktion="verfahrensgebiet_layer_existiert", + meldung=( + "Ein Verfahrensgebiet-Layer existiert bereits.\n" + "Was soll mit der Auswahl geschehen?" + ), + kontext=existing_layers[0], + ) + ) + + aktion = entscheidung_ergebnis.aktion + verf_layer = existing_layers[0] + + if aktion == "verfahrensgebiet_abbrechen" or not entscheidung_ergebnis.ok: + return + + elif aktion == "verfahrensgebiet_ergaenzen": + _, hinzugefuegt, bereits_vorhanden, fehlgeschlagen = ergaenze_verfahrensgebiet( + verf_layer, active_layer + ) + apply_style(verf_layer, "verfahrensgebiet.qml") + iface.mapCanvas().refresh() + self.setze_haken(self.haken_verf, True) + QMessageBox.information( + self, + "Flurstücke ergänzt", + f"Hinzugefügt: {hinzugefuegt}\n" + f"Bereits vorhanden (übersprungen): {bereits_vorhanden}\n" + f"Fehlgeschlagen: {fehlgeschlagen}", + ) + + elif aktion == "verfahrensgebiet_entfernen": + _, entfernt, nicht_enthalten, fehlgeschlagen = entferne_aus_verfahrensgebiet( + verf_layer, active_layer + ) + apply_style(verf_layer, "verfahrensgebiet.qml") + iface.mapCanvas().refresh() + QMessageBox.information( + self, + "Flurstücke entfernt", + f"Entfernt: {entfernt}\n" + f"Nicht im Verfahrensgebiet (übersprungen): {nicht_enthalten}\n" + f"Fehlgeschlagen: {fehlgeschlagen}", + ) + + elif aktion == "verfahrensgebiet_ersetzen": + for lyr in existing_layers: + QgsProject.instance().removeMapLayer(lyr.id()) + layer, erfolgreich, fehlgeschlagen = ersetze_verfahrensgebiet(active_layer) + if not layer or not layer.isValid(): + error("Verfahrensgebiet erstellen", "Layer konnte nicht erstellt werden.") + return + QgsProject.instance().addMapLayer(layer) + apply_style(layer, "verfahrensgebiet.qml") + iface.mapCanvas().refresh() + self.verf_layer_combo.setLayer(layer) + self._on_verf_layer_changed(layer) + self.setze_haken(self.haken_verf, True) + QMessageBox.information( + self, + "Verfahrensgebiet ersetzt", + f"Verfahrensgebiet neu erstellt.\n" + f"Erfolgreich: {erfolgreich}\n" + f"Fehlgeschlagen: {fehlgeschlagen}", + ) + + finally: + self._set_busy(self.btn_verf_aus_auswahl, False, "aus Auswahl") + + def handle_verf_aus_liste(self) -> None: + """Platzhalter – Verfahrensgebiet aus Liste erstellen (noch nicht implementiert).""" + show_info_dialog( + "Noch nicht implementiert", + "Die Funktion 'Aus Liste' steht in einer zukünftigen Version zur Verfügung.", + parent=self, + )