diff --git a/geopackage_validator/utils.py b/geopackage_validator/utils.py index e974c5a..b95ece6 100644 --- a/geopackage_validator/utils.py +++ b/geopackage_validator/utils.py @@ -6,6 +6,8 @@ from pathlib import Path import json +from time import sleep + import yaml from typing import Callable, Optional @@ -57,7 +59,12 @@ def silence_gdal(): gdal.PushErrorHandler(error_handler) driver = ogr.GetDriverByName("GPKG") - dataset = driver.Open(filename, 0) + + dataset = None + try: + dataset = driver.Open(filename, 0) + except Exception as e: + error_handler(gdal.CE_Fatal, 0, e.args[0]) if dataset is not None: dataset.silence_gdal = silence_gdal diff --git a/geopackage_validator/validate.py b/geopackage_validator/validate.py index 832dd5d..92c0264 100644 --- a/geopackage_validator/validate.py +++ b/geopackage_validator/validate.py @@ -2,6 +2,7 @@ import logging import sys import traceback +from time import sleep from osgeo import gdal @@ -77,29 +78,31 @@ def validators_to_use( return [validator_dict[code] for code in codes] +class GdalErrorHandler(object): + def __init__(self): + self.gdal_error_traces = [] + self.gdal_warning_traces = [] + + def handler(self, err_level, err_no, err_msg): + trace = err_msg.replace("\n", " ") + if err_level == gdal.CE_Warning: + self.gdal_warning_traces.append(trace) + else: + self.gdal_error_traces.append(trace) + + def validate( gpkg_path, table_definitions_path=None, validations_path=None, validations="" ): """Starts the geopackage validations.""" utils.check_gdal_version() - gdal_error_traces = [] - gdal_warning_traces = [] - - # Register GDAL error handler function - def gdal_error_handler(err_class, err_num, error): - trace = error.replace("\n", " ") - # import pdb - # pdb.set_trace() - if err_class == gdal.CE_Warning: - gdal_warning_traces.append(trace) - else: - gdal_error_traces.append(trace) + errHandler = GdalErrorHandler() + dataset = utils.open_dataset(gpkg_path, errHandler.handler) - dataset = utils.open_dataset(gpkg_path, gdal_error_handler) - if len(gdal_error_traces): + if len(errHandler.gdal_error_traces): initial_gdal_traces = [ - gdal_error_traces.pop() for _ in range(len(gdal_error_traces)) + errHandler.gdal_error_traces.pop() for _ in range(len(errHandler.gdal_error_traces)) ] initial_gdal_errors = [ format_result( @@ -163,7 +166,7 @@ def gdal_error_handler(err_class, err_num, error): validation_error = True success = False current_gdal_error_traces = [ - gdal_error_traces.pop() for _ in range(len(gdal_error_traces)) + errHandler.gdal_error_traces.pop() for _ in range(len(errHandler.gdal_error_traces)) ] if current_gdal_error_traces: success = False @@ -178,12 +181,12 @@ def gdal_error_handler(err_class, err_num, error): ) validation_results.append(output) - if gdal_warning_traces: + if errHandler.gdal_warning_traces: output = format_result( validation_code="UNKNOWN_WARNINGS", validation_description="It is recommended that these unexpected (GDAL) warnings are looked into.", level=ValidationLevel.UNKNOWN_WARNING, - trace=gdal_warning_traces, + trace=errHandler.gdal_warning_traces, ) validation_results.append(output) diff --git a/pyproject.toml b/pyproject.toml index 56e5aba..7e34b0d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,17 +10,18 @@ license = {text = "MIT"} classifiers = [ "Programming Language :: Python :: 3", "Programming Language :: Python :: 3 :: Only", - "Programming Language :: Python :: 3.6", "Programming Language :: Python :: 3.7", "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", ] dynamic = ["version", "readme"] dependencies = [ "setuptools>=42.0,!=58.*,!=59.*,!=60.*,!=61.*", "Click >= 8.0", "click-log >=0.3", - "gdal >=3.0.4", + "gdal >=3.4", "minio", "pyyaml", ] diff --git a/tests/test_utils.py b/tests/test_utils.py index c3a936f..6ad2b2f 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -24,9 +24,15 @@ def gdal_error_handler(err_class, err_num, error): results.append("GDAL_ERROR") dataset = open_dataset("tests/data/test_gdal_error.gpkg", gdal_error_handler) - validations = dataset.ExecuteSQL('select rtreecheck("rtree_table_geom");') - dataset.ReleaseResultSet(validations) - assert results[0] == "GDAL_ERROR" + + # Since GDAL 3.7 the exceptions work more pythonic + try: + validations = dataset.ExecuteSQL('select rtreecheck("rtree_table_geom");') + dataset.ReleaseResultSet(validations) + except RuntimeError as e: + results.append("GDAL_TRY_ERROR") + + assert results[0] == "GDAL_ERROR" or results[0] == "GDAL_TRY_ERROR" assert len(results) == 1 @@ -39,7 +45,7 @@ def gdal_error_handler(err_class, err_num, error): dataset = open_dataset("tests/data/test_gdal_error.gpkg", gdal_error_handler) with dataset.silence_gdal(): - validations = dataset.ExecuteSQL('select rtreecheck("rtree_table_geom");') + validations = dataset.ExecuteSQL('select rtreecheck("rtree_cbs_arbeidsmarktregio_2014_gegeneraliseerd_geom_parent");') dataset.ReleaseResultSet(validations) assert len(results) == 0 @@ -64,9 +70,17 @@ def gdal_error_handler(err_class, err_num, error): results.append("GDAL_ERROR") dataset = open_dataset("tests/data/test_gdal_error.gpkg", gdal_error_handler) - do_something_with_error_gdal(dataset) - do_something_silenced_gdal(dataset) - do_something_with_error_gdal(dataset) - assert len(results) == 2 - assert results == ["GDAL_ERROR", "GDAL_ERROR"] + # Since GDAL 3.7 this test will not work because these errors are thrown through the try except clause + try: + do_something_with_error_gdal(dataset) + do_something_silenced_gdal(dataset) + do_something_with_error_gdal(dataset) + except RuntimeError as e: + results.append("GDAL_TRY_ERROR") + + if len(results) == 1: + assert results == ["GDAL_TRY_ERROR"] + else: + assert len(results) == 2 + assert results == ["GDAL_ERROR", "GDAL_ERROR"] diff --git a/tests/test_validate.py b/tests/test_validate.py index bdeff13..cde6516 100644 --- a/tests/test_validate.py +++ b/tests/test_validate.py @@ -111,9 +111,16 @@ def test_validate_all_validations_no_error(): def test_validate_all_validations_with_broken_gpkg_throws_gdal_error(): + + # try: results, validations_executed, success = validate( gpkg_path="tests/data/test_broken_geopackage.gpkg", validations="ALL" ) + # except Exception: + # print('verwachte fout') + # else: + # print('foute fout') + assert len(results) == 1 print(results) assert results[0]["locations"] == [