DanielandClaude Sonnet 5
e86bef40d2
Tests für Dateipruefer._ist_leer (None/Whitespace), Dateipruefer._handle_leere_eingabe (Priorität bei mehreren Flags), Datenschreiber._store_style_in_gpkg (nachgezogen aus #63, siehe #67) und Pruefmanager._handle_verfahrensgebiet_layer_existiert (entfernen/ersetzen) ergänzt. Linkpruefer._ist_url/_pruefe_url/_pruefe_dateipfad und Pruefmanager._handle_datei_existiert sind bereits über pruefe()/verarbeite() vollständig abgedeckt und bekommen den Vermerk "von Test abgesehen". Pruefmanager._ask_use_or_replace_pufferlayer ist toter Code (Bug #60) und bleibt bis zur Behebung unmarkiert getestet, nur vermerkt. sn_basis steht damit laut Prüfstand bei 0 ungeklärten Funktionen. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
299 lines
12 KiB
Python
299 lines
12 KiB
Python
"""Tests für sn_basis/modules/Datenschreiber.py (Schreib-/Ladepipeline, Issue #63)."""
|
||
|
||
import gc
|
||
import json
|
||
import shutil
|
||
import sqlite3
|
||
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)
|
||
|
||
|
||
class StoreStyleInGpkgTests(unittest.TestCase):
|
||
"""Tests für Datenschreiber._store_style_in_gpkg (Issue #70, nachgezogen aus #63)."""
|
||
|
||
def test_legt_tabelle_an_und_speichert_stil_als_default(self):
|
||
"""Prüft, dass _store_style_in_gpkg die layer_styles-Tabelle anlegt und den Stil als Default speichert."""
|
||
tmp = tempfile.mkdtemp()
|
||
try:
|
||
gpkg_path = str(Path(tmp) / "ziel.gpkg")
|
||
style_path = Path(tmp) / "stil.qml"
|
||
style_path.write_text("<qgis>Test-QML</qgis>", encoding="utf-8")
|
||
schreiber, _pm = _make_schreiber(gpkg_path=gpkg_path)
|
||
|
||
schreiber._store_style_in_gpkg("mein_layer", str(style_path))
|
||
|
||
with sqlite3.connect(gpkg_path) as conn:
|
||
zeile = conn.execute(
|
||
"SELECT f_table_name, styleName, styleQML, useAsDefault FROM layer_styles"
|
||
).fetchone()
|
||
conn.close()
|
||
finally:
|
||
# _store_style_in_gpkg schließt seine eigene sqlite3-Verbindung nicht (Produktionscode,
|
||
# nicht Teil von Issue #70) – ohne gc.collect() bleibt die .gpkg-Datei unter Windows
|
||
# gesperrt und shutil.rmtree schlägt fehl.
|
||
gc.collect()
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
|
||
self.assertEqual(zeile[0], "mein_layer")
|
||
self.assertEqual(zeile[1], "stil.qml")
|
||
self.assertEqual(zeile[2], "<qgis>Test-QML</qgis>")
|
||
self.assertEqual(zeile[3], 1)
|
||
|
||
def test_setzt_bisherigen_default_stil_desselben_layers_zurueck(self):
|
||
"""Prüft, dass _store_style_in_gpkg den useAsDefault-Wert vorhandener Stile desselben Layers zurücksetzt."""
|
||
tmp = tempfile.mkdtemp()
|
||
try:
|
||
gpkg_path = str(Path(tmp) / "ziel.gpkg")
|
||
style_path = Path(tmp) / "stil.qml"
|
||
style_path.write_text("<qgis>Neuer Stil</qgis>", encoding="utf-8")
|
||
schreiber, _pm = _make_schreiber(gpkg_path=gpkg_path)
|
||
|
||
schreiber._store_style_in_gpkg("mein_layer", str(style_path))
|
||
schreiber._store_style_in_gpkg("mein_layer", str(style_path))
|
||
|
||
with sqlite3.connect(gpkg_path) as conn:
|
||
werte = [
|
||
row[0]
|
||
for row in conn.execute(
|
||
"SELECT useAsDefault FROM layer_styles WHERE f_table_name = ? ORDER BY id",
|
||
("mein_layer",),
|
||
).fetchall()
|
||
]
|
||
conn.close()
|
||
finally:
|
||
gc.collect()
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
|
||
self.assertEqual(werte, [0, 1])
|
||
|
||
def test_liest_geometriespalte_ueber_geometrycolumn_aus(self):
|
||
"""Prüft, dass _store_style_in_gpkg die Geometriespalte eines Layers mit geometryColumn() übernimmt."""
|
||
|
||
class _LayerMitGeometrieSpalte:
|
||
def geometryColumn(self) -> str:
|
||
return "geom"
|
||
|
||
tmp = tempfile.mkdtemp()
|
||
try:
|
||
gpkg_path = str(Path(tmp) / "ziel.gpkg")
|
||
style_path = Path(tmp) / "stil.qml"
|
||
style_path.write_text("<qgis/>", encoding="utf-8")
|
||
schreiber, _pm = _make_schreiber(gpkg_path=gpkg_path)
|
||
|
||
schreiber._store_style_in_gpkg("mein_layer", str(style_path), layer=_LayerMitGeometrieSpalte())
|
||
|
||
with sqlite3.connect(gpkg_path) as conn:
|
||
spalte = conn.execute(
|
||
"SELECT f_geometry_column FROM layer_styles WHERE f_table_name = ?",
|
||
("mein_layer",),
|
||
).fetchone()[0]
|
||
conn.close()
|
||
finally:
|
||
gc.collect()
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
|
||
self.assertEqual(spalte, "geom")
|
||
|
||
def test_fehlender_stilpfad_meldet_fehler_ueber_pruefmanager_statt_absturz(self):
|
||
"""Prüft, dass ein nicht lesbarer Stilpfad über den Pruefmanager gemeldet wird statt eine Exception zu werfen."""
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
gpkg_path = str(Path(tmp) / "ziel.gpkg")
|
||
schreiber, pruefmanager = _make_schreiber(gpkg_path=gpkg_path)
|
||
|
||
schreiber._store_style_in_gpkg("mein_layer", str(Path(tmp) / "nicht_vorhanden.qml"))
|
||
|
||
aktionen = [getattr(e, "aktion", None) for e in pruefmanager.ergebnisse]
|
||
self.assertIn("style_gpkg_speichern_fehlgeschlagen", aktionen)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|