From 7efd933ff08cefbcb2d7d2d965890b74f9dd2a25 Mon Sep 17 00:00:00 2001 From: David Marteau Date: Wed, 21 Aug 2024 18:08:03 +0200 Subject: [PATCH] Remove postgres cache handler * Use generic Qgis storage instead * Remove broken validators --- CHANGELOG.md | 1 + Makefile | 3 + docker/Dockerfile | 3 + pyproject.toml | 2 + pyqgiswps/config.py | 2 +- pyqgiswps/executors/processfactory.py | 3 +- pyqgiswps/inout/basic.py | 10 +- pyqgiswps/poolserver/utils.py | 4 +- pyqgiswps/qgscache/cachemanager.py | 2 +- pyqgiswps/qgscache/handlers/__init__.py | 1 - .../qgscache/handlers/postgres_handler.py | 110 ------- .../qgscache/handlers/storagehandlers.py | 12 + pyqgiswps/tests.py | 26 +- pyqgiswps/utils/lru.py | 4 +- pyqgiswps/validator/__init__.py | 11 +- pyqgiswps/validator/complexvalidator.py | 281 ------------------ tests/docker-compose.yml | 1 + .../{validator => }/test_literalvalidators.py | 0 .../validator/test_complexvalidators.py | 116 -------- 19 files changed, 58 insertions(+), 534 deletions(-) delete mode 100644 pyqgiswps/qgscache/handlers/postgres_handler.py create mode 100644 pyqgiswps/qgscache/handlers/storagehandlers.py delete mode 100644 pyqgiswps/validator/complexvalidator.py rename tests/unittests/{validator => }/test_literalvalidators.py (100%) delete mode 100644 tests/unittests/validator/test_complexvalidators.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d06607d..8a3984d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ * Remove deprecated store apis: - `/store/` - `/ows/store/` +* Remove postgres cache handler in favor of generic Qgis storage ## 1.8.7 - 2024-07-15 diff --git a/Makefile b/Makefile index 38de4d2..6344fe5 100644 --- a/Makefile +++ b/Makefile @@ -67,3 +67,6 @@ typing: client-test: cd tests/clienttests && pytest -v $(PYTEST_ADDOPTS) +scan: + @bandit -c pyproject.toml -r pyqgiswps + diff --git a/docker/Dockerfile b/docker/Dockerfile index 4e97a6d..050a7e9 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -49,6 +49,9 @@ RUN mkdir -p /opt/local/ \ COPY docker/docker-entrypoint.sh / RUN chmod 0755 /docker-entrypoint.sh +# Expose to all interfaces in container +ENV QGSWPS_SERVER_INTERFACES=0.0.0.0 + EXPOSE 8080 ENTRYPOINT ["/docker-entrypoint.sh"] diff --git a/pyproject.toml b/pyproject.toml index c1a3968..b1861dc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -66,3 +66,5 @@ include = [ "pyqgisservercontrib.*", ] +[tool.bandit] +skips = ["B410", "B320", "B403", "B301"] diff --git a/pyqgiswps/config.py b/pyqgiswps/config.py index 9bbcda4..b7263a4 100644 --- a/pyqgiswps/config.py +++ b/pyqgiswps/config.py @@ -66,7 +66,7 @@ def load_configuration(): CONFIG.set('server', 'encoding', 'utf-8') CONFIG.set('server', 'language', 'en-US') CONFIG.set('server', 'port', getenv('QGSWPS_SERVER_HTTP_PORT', '8080')) - CONFIG.set('server', 'interfaces', getenv('QGSWPS_SERVER_INTERFACES', '0.0.0.0')) + CONFIG.set('server', 'interfaces', getenv('QGSWPS_SERVER_INTERFACES', '127.0.0.1')) # Select if output files are returned by reference or in response body CONFIG.set('server', 'outputfile_as_reference', getenv('QGSWPS_SERVER_OUTPUTFILE_AS_REFERENCE', 'yes')) # Max input file fetched as remote reference diff --git a/pyqgiswps/executors/processfactory.py b/pyqgiswps/executors/processfactory.py index daf6d35..98e53a7 100644 --- a/pyqgiswps/executors/processfactory.py +++ b/pyqgiswps/executors/processfactory.py @@ -28,6 +28,7 @@ from ..config import confservice from ..exceptions import ProcessException from ..poolserver.server import create_poolserver +from ..utils.conditions import assert_precondition from ..utils.plugins import WPSServerInterfaceImpl from ..utils.qgis import setup_qgis_paths, start_qgis_application from .logstore import logstore @@ -141,7 +142,7 @@ def initialize(self, load_qgis_processing: bool = False) -> Optional[List[WPSPro Should be called once """ - assert not self._initialized + assert_precondition(not self._initialized) self._config = confservice['processing'] diff --git a/pyqgiswps/inout/basic.py b/pyqgiswps/inout/basic.py index 9c135b6..c7b07c7 100644 --- a/pyqgiswps/inout/basic.py +++ b/pyqgiswps/inout/basic.py @@ -22,8 +22,14 @@ from pyqgiswps.exceptions import InvalidParameterValue from pyqgiswps.inout.formats import Format -from pyqgiswps.inout.literaltypes import LITERAL_DATA_TYPES, convert, is_anyvalue, to_json_serializable +from pyqgiswps.inout.literaltypes import ( + LITERAL_DATA_TYPES, + convert, + is_anyvalue, + to_json_serializable, +) from pyqgiswps.inout.uoms import UOM +from pyqgiswps.utils.conditions import assert_precondition from pyqgiswps.validator import get_validator from pyqgiswps.validator.base import emptyvalidator from pyqgiswps.validator.literalvalidator import validate_allowed_values, validate_anyvalue @@ -189,7 +195,7 @@ class BasicLiteral: """ def __init__(self, data_type, uoms=None): - assert data_type in LITERAL_DATA_TYPES, f"data type {data_type} no supported" + assert_precondition(data_type in LITERAL_DATA_TYPES, f"data type {data_type} no supported") self.data_type = data_type # current uom self.supported_uoms = uoms diff --git a/pyqgiswps/poolserver/utils.py b/pyqgiswps/poolserver/utils.py index 96d1437..e9fe486 100644 --- a/pyqgiswps/poolserver/utils.py +++ b/pyqgiswps/poolserver/utils.py @@ -11,11 +11,13 @@ import os +from tempfile import gettempdir + _pid = os.getpid() def _get_ipc(name: str) -> str: - ipc_path = f'/tmp/qgswps/{name}_{_pid}' + ipc_path = f'{gettempdir()}/qgswps/{name}_{_pid}' os.makedirs(os.path.dirname(ipc_path), exist_ok=True) return f'ipc://{ipc_path}' diff --git a/pyqgiswps/qgscache/cachemanager.py b/pyqgiswps/qgscache/cachemanager.py index 4ab8721..789fb65 100644 --- a/pyqgiswps/qgscache/cachemanager.py +++ b/pyqgiswps/qgscache/cachemanager.py @@ -212,7 +212,7 @@ def get_project_factory(self, key: str) -> Callable: # Retrieve the protocol-handler try: - store = componentmanager.get_service('@3liz.org/cache/protocol-handler;1?scheme=%s' % scheme) + store = componentmanager.get_service(f'@3liz.org/cache/protocol-handler;1?scheme={scheme}') except componentmanager.FactoryNotFoundError: LOGGER.warning("No protocol handler found for %s: using Qgis storage handler", scheme) # Fallback to Qgis storage handler diff --git a/pyqgiswps/qgscache/handlers/__init__.py b/pyqgiswps/qgscache/handlers/__init__.py index 20ba134..66b922c 100644 --- a/pyqgiswps/qgscache/handlers/__init__.py +++ b/pyqgiswps/qgscache/handlers/__init__.py @@ -8,6 +8,5 @@ from .file_handler import * # noqa: F403 -from .postgres_handler import * # noqa: F403 __all__ = [] diff --git a/pyqgiswps/qgscache/handlers/postgres_handler.py b/pyqgiswps/qgscache/handlers/postgres_handler.py deleted file mode 100644 index b82815c..0000000 --- a/pyqgiswps/qgscache/handlers/postgres_handler.py +++ /dev/null @@ -1,110 +0,0 @@ -# -# Copyright 2020 3liz -# Author David Marteau -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" Postgres protocol handler - - See - * http://www.bostongis.com/blog/index.php?/archives/271-New-in-QGIS-3.2-Save-Project-to-PostgreSQL.html - * https://github.com/qgis/QGIS-Enhancement-Proposals/issues/118 - - project filename syntax for posgresql - - ``` - postgres://[user[:pass]@]host[:port]/?dbname=X&schema=Y&project=Z - ``` - - How does QGIS store the project? - - QGIS creates a table called qgis_projects in whatever schema you had specified, - each project is stored as a separate row. - - columns are: 'name', 'metadata', 'content' - - Metadata is a json data that contains the 'last_modified_time' field. - - -""" -import logging -import urllib.parse - -from datetime import datetime -from typing import Optional, Tuple -from urllib.parse import parse_qs - -import psycopg2 - -from qgis.core import QgsProject - -from pyqgisservercontrib.core import componentmanager - -LOGGER = logging.getLogger('SRVLOG') - -__all__ = [] - - -@componentmanager.register_factory('@3liz.org/cache/protocol-handler;1?scheme=postgres') -class PostgresProtocolHandler: - """ Handle postgres protocol - """ - - def __init__(self): - pass - - def get_project( - self, - url: urllib.parse.ParseResult, - project: Optional[QgsProject] = None, - timestamp: Optional[datetime] = None, - ) -> Tuple[QgsProject, datetime]: - """ Create or return a proect - """ - params = {k: v[0] for k, v in parse_qs(url.query).items()} - - try: - project = params.pop('project') - schema = params.pop('schema', 'public') - database = params.pop('dbname', None) - except KeyError as exc: - LOGGER.error("Postgres handler: Missing parameter %s: %s", url.geturl(), str(exc)) - raise FileNotFoundError(url.geturl()) - - connexion_params = dict( - host=url.hostname, - port=url.port, - user=url.username, - password=url.password, - database=database, - # Treats remaining params as supported psql client options - **params, - ) - - # Connect to database and check modification time - try: - LOGGER.debug("**** Postgresql connection params %s", connexion_params) - conn = psycopg2.connect(**connexion_params) - cursor = conn.cursor() - cursor.execute(f"select metadata from {schema}.qgis_projects where name='{project}'") - if cursor.rowcount <= 0: - raise FileNotFoundError(url.geturl()) - metadata = cursor.fetchone()[0] - LOGGER.debug("**** Postgres metadata for '%s': %s", project, metadata) - conn.close() - except psycopg2.OperationalError as e: - LOGGER.error("Postgres handler Connection error: %s", str(e)) - raise FileNotFoundError(url.geturl()) - except psycopg2.Error as e: - LOGGER.error("Postgres handler Connection error: %s", str(e)) - raise RuntimeError("Connection failed: %s", url.geturl()) - - modified_time = datetime.fromisoformat(metadata['last_modified_time']) - if timestamp is None or timestamp < modified_time: - cachmngr = componentmanager.get_service('@3liz.org/cache-manager;1') - project = cachmngr.read_project(url.geturl()) - timestamp = modified_time - - return project, timestamp diff --git a/pyqgiswps/qgscache/handlers/storagehandlers.py b/pyqgiswps/qgscache/handlers/storagehandlers.py new file mode 100644 index 0000000..809b661 --- /dev/null +++ b/pyqgiswps/qgscache/handlers/storagehandlers.py @@ -0,0 +1,12 @@ +# +# Copyright 2020 3liz +# Author David Marteau +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +""" Qgis storage protocol handler + + Register storage handlers as protocol handler +""" diff --git a/pyqgiswps/tests.py b/pyqgiswps/tests.py index cdeb4d9..0635033 100644 --- a/pyqgiswps/tests.py +++ b/pyqgiswps/tests.py @@ -237,44 +237,44 @@ def options(self, headers: Optional[Dict] = None, path: str = '/ows/') -> WpsTes def assert_response_accepted(resp): - assert resp.status_code == 200, "resp.status_code = %s" % resp.status_code - assert resp.headers['Content-Type'] == 'text/xml;charset=utf-8' + assert resp.status_code == 200, f"resp.status_code = {resp.status_code}" # nosec + assert resp.headers['Content-Type'] == 'text/xml;charset=utf-8' # nosec success = resp.xpath_text( '/wps:ExecuteResponse' '/wps:Status' '/wps:ProcessAccepted', ) - assert success is not None + assert success is not None # nosec # TODO: assert status URL is present def assert_process_started(resp): - assert resp.status_code == 200, "resp.status_code = %s" % resp.status_code - assert resp.headers['Content-Type'] == 'text/xml;charset=utf-8' + assert resp.status_code == 200, f"resp.status_code = {resp.status_code}" # nosec + assert resp.headers['Content-Type'] == 'text/xml;charset=utf-8' # nosec success = resp.xpath_text( '/wps:ExecuteResponse' '/wps:Status' 'ProcessStarted', ) # Is it still like this in PyWPS-4 ? - assert success.split[0] == "processstarted" + assert success.split[0] == "processstarted" # nosec def assert_response_success(resp): - assert resp.status_code == 200, "resp.status_code = %s" % resp.status_code - assert resp.headers['Content-Type'] == 'text/xml;charset=utf-8' + assert resp.status_code == 200, f"resp.status_code = {resp.status_code}" # nosec + assert resp.headers['Content-Type'] == 'text/xml;charset=utf-8' # nosec success = resp.xpath('/wps:ExecuteResponse/wps:Status/wps:ProcessSucceeded') - assert len(success) == 1 + assert len(success) == 1 # nosec def assert_pyqgiswps_version(resp): # get first child of root element root_firstchild = resp.xpath('/*')[0].getprevious() - assert isinstance(root_firstchild, lxml.etree._Comment) + assert isinstance(root_firstchild, lxml.etree._Comment) # nosec tokens = root_firstchild.text.split() - assert len(tokens) == 2 - assert tokens[0] == 'py-qgis-wps' - assert tokens[1] == __version__ + assert len(tokens) == 2 # nosec + assert tokens[0] == 'py-qgis-wps' # nosec + assert tokens[1] == __version__ # nosec # diff --git a/pyqgiswps/utils/lru.py b/pyqgiswps/utils/lru.py index 2256a44..8f15bf2 100644 --- a/pyqgiswps/utils/lru.py +++ b/pyqgiswps/utils/lru.py @@ -10,6 +10,8 @@ """ from collections import OrderedDict +from .conditions import assert_precondition + class lrucache(): @@ -104,7 +106,7 @@ def size(self, size=None): :param int size: maximum number of elements in the cache """ if size is not None: - assert size > 0 + assert_precondition(size > 0) if size < self._capacity: d = self._table # Remove extra items diff --git a/pyqgiswps/validator/__init__.py b/pyqgiswps/validator/__init__.py index 1a8ffbd..514b7f2 100644 --- a/pyqgiswps/validator/__init__.py +++ b/pyqgiswps/validator/__init__.py @@ -18,16 +18,15 @@ import logging from pyqgiswps.validator.base import emptyvalidator -from pyqgiswps.validator.complexvalidator import validategeojson, validategeotiff, validategml, validateshapefile LOGGER = logging.getLogger('SRVLOG') _VALIDATORS = { - 'application/vnd.geo+json': validategeojson, - 'application/json': validategeojson, - 'application/x-zipped-shp': validateshapefile, - 'application/gml+xml': validategml, - 'image/tiff; subtype=geotiff': validategeotiff, + 'application/vnd.geo+json': emptyvalidator, + 'application/json': emptyvalidator, + 'application/x-zipped-shp': emptyvalidator, + 'application/gml+xml': emptyvalidator, + 'image/tiff; subtype=geotiff': emptyvalidator, 'application/xogc-wcs': emptyvalidator, 'application/x-ogc-wcs; version=1.0.0': emptyvalidator, 'application/x-ogc-wcs; version=1.1.0': emptyvalidator, diff --git a/pyqgiswps/validator/complexvalidator.py b/pyqgiswps/validator/complexvalidator.py deleted file mode 100644 index ddd344a..0000000 --- a/pyqgiswps/validator/complexvalidator.py +++ /dev/null @@ -1,281 +0,0 @@ -# -# Copyright 2018 3liz -# Author: David Marteau -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. -# -# Original parts are Copyright 2016 OSGeo Foundation, -# represented by PyWPS Project Steering Committee, -# and released under MIT license. -# Please consult PYWPS_LICENCE.txt for details -# -"""Validator classes are used for ComplexInputs, to validate the content -""" - -import logging -import mimetypes -import os - -from pyqgiswps.validator.formats import FORMATS -from pyqgiswps.validator.mode import MODE - -LOGGER = logging.getLogger('SRVLOG') - - -def validategml(data_input, mode): - """GML validation function - - :param data_input: :class:`ComplexInput` - :param pyqgiswps.validator.mode.MODE mode: - - This function validates GML input based on given validation mode. Following - happens, if `mode` parameter is given: - - `MODE.NONE` - it will return always `True` - `MODE.SIMPLE` - the mimetype will be checked - `MODE.STRICT` - `GDAL/OGR `_ is used for getting the propper format. - `MODE.VERYSTRICT` - the :class:`lxml.etree` is used along with given input `schema` and the - GML file is properly validated against given schema. - """ - - LOGGER.info('validating GML; Mode: %s', mode) - passed = False - - if mode >= MODE.NONE: - passed = True - - if mode >= MODE.SIMPLE: - - name = data_input.file - (mtype, _encoding) = mimetypes.guess_type(name, strict=False) - passed = data_input.data_format.mime_type in {mtype, FORMATS.GML.mime_type} - - if mode >= MODE.STRICT: - - from pyqgiswps.dependencies import ogr - data_source = ogr.Open(data_input.file) - if data_source: - passed = (data_source.GetDriver().GetName() == "GML") - else: - passed = False - - if mode >= MODE.VERYSTRICT: - - from urllib.request import urlopen - - from lxml import etree - - try: - schema_url = data_input.data_format.schema - gmlschema_doc = etree.parse(urlopen(schema_url)) - gmlschema = etree.XMLSchema(gmlschema_doc) - passed = gmlschema.validate(etree.parse(data_input.stream)) - except Exception as e: - LOGGER.warning(e) - passed = False - - return passed - - -def validategeojson(data_input, mode): - """GeoJSON validation example - - >>> import StringIO - >>> class FakeInput(object): - ... json = open('point.geojson','w') - ... json.write(''' - { - "type":"Feature", - "properties":{}, - "geometry":{ - "type":"Point", - "coordinates":[8.5781228542328, 22.87500500679] - }, - "crs":{ - "type":"name", - "properties":{"name":"urn:ogc:def:crs:OGC:1.3:CRS84"} - } - } - ''') - ... json.close() - ... file = 'point.geojson' - >>> class fake_data_format(object): - ... mimetype = 'application/geojson' - >>> fake_input = FakeInput() - >>> fake_input.data_format = fake_data_format() - >>> validategeojson(fake_input, MODE.SIMPLE) - True - """ - - LOGGER.info('validating GeoJSON; Mode: %s', mode) - passed = False - - if mode >= MODE.NONE: - passed = True - - """ - # - # This is broken in many ways: - # * Using deprecated jsonschema validation methods - # * Segmentation fault when trying to validate in - # VERY_STRICT mode. - # - if mode >= MODE.SIMPLE: - - name = data_input.file - (mtype, encoding) = mimetypes.guess_type(name, strict=False) - passed = data_input.data_format.mime_type in {mtype, FORMATS.GEOJSON.mime_type} - - if mode >= MODE.STRICT: - - from pyqgiswps.dependencies import ogr - data_source = ogr.Open(data_input.file) - if data_source: - passed = (data_source.GetDriver().GetName() == "GeoJSON") - else: - passed = False - - # if mode >= MODE.VERYSTRICT: - - # import jsonschema - f rom jsonschema.referencing import ( - Registry, - DRAFT4, - ) - import json - - # this code comes from - # https://github.com/om-henners/GeoJSON_Validation/blob/master/geojsonvalidation/geojson_validation.py - schema_home = os.path.join(_get_schemas_home(), "geojson") - base_schema = os.path.join(schema_home, "geojson.json") - - with open(base_schema) as fh: - geojson_base = json.load(fh) - - with open(os.path.join(schema_home, "crs.json")) as fh: - crs_json = json.load(fh) - - with open(os.path.join(schema_home, "bbox.json")) as fh: - bbox_json = json.load(fh) - - with open(os.path.join(schema_home, "geometry.json")) as fh: - geometry_json = json.load(fh) - - cached_json = { - "http://json-schema.org/geojson/crs.json": crs_json, - "http://json-schema.org/geojson/bbox.json": bbox_json, - "http://json-schema.org/geojson/geometry.json": geometry_json - } - - resolver = jsonschema.RefResolver( - "http://json-schema.org/geojson/geojson.json", - geojson_base, store=cached_json - ) - registry = Registry().with_resource( - "http://json-schema.org/geojson/geojson.json", - - - validator = jsonschema.Draft4Validator(geojson_base, resolver=resolver) - try: - validator.validate(json.loads(data_input.stream.read())) - passed = True - except jsonschema.ValidationError: - passed = False - """ - return passed - - -def validateshapefile(data_input, mode): - """ESRI Shapefile validation example - - """ - - LOGGER.info('validating Shapefile; Mode: %s', mode) - passed = False - - if mode >= MODE.NONE: - passed = True - - if mode >= MODE.SIMPLE: - - name = data_input.file - (mtype, _encoding) = mimetypes.guess_type(name, strict=False) - passed = data_input.data_format.mime_type in {mtype, FORMATS.SHP.mime_type} - - if mode >= MODE.STRICT: - - import zipfile - - from pyqgiswps.dependencies import ogr - z = zipfile.ZipFile(data_input.file) - shape_name = None - for name in z.namelist(): - z.extract(name, data_input.tempdir) - if os.path.splitext(name)[1].lower() == '.shp': - shape_name = name - - if shape_name: - data_source = ogr.Open(os.path.join(data_input.tempdir, shape_name)) - - if data_source: - passed = (data_source.GetDriver().GetName() == "ESRI Shapefile") - else: - passed = False - - return passed - - -def validategeotiff(data_input, mode): - """GeoTIFF validation example - """ - - LOGGER.info('Validating Shapefile; Mode: %s', mode) - passed = False - - if mode >= MODE.NONE: - passed = True - - if mode >= MODE.SIMPLE: - - name = data_input.file - (mtype, _encoding) = mimetypes.guess_type(name, strict=False) - passed = data_input.data_format.mime_type in {mtype, FORMATS.GEOTIFF.mime_type} - - if mode >= MODE.STRICT: - - from pyqgiswps.dependencies import gdal - data_source = gdal.Open(data_input.file) - if data_source: - passed = (data_source.GetDriver().ShortName == "GTiff") - else: - passed = False - - return passed - - -def _get_schemas_home(): - """Get path to schemas directory - """ - schema_dir = os.path.join( - os.path.abspath( - os.path.dirname(__file__), - ), - os.path.pardir, - "schemas") - LOGGER.debug('Schemas directory: %s', schema_dir) - return schema_dir - - -if __name__ == "__main__": - import doctest - - from pyqgiswps.tests import temp_dir - with temp_dir() as tmp: - os.chdir(tmp) - doctest.testmod() diff --git a/tests/docker-compose.yml b/tests/docker-compose.yml index 6193835..ed786d0 100644 --- a/tests/docker-compose.yml +++ b/tests/docker-compose.yml @@ -15,6 +15,7 @@ services: QGSWPS_LOGLEVEL: DEBUG QGSWPS_REDIS_HOST: redis QGSWPS_EXPOSE_SERVER_INFOS: 'yes' + QGSWPS_SERVER_INTERFACES: '0.0.0.0' PYTEST_ADDOPTS: ${PYTEST_ADDOPTS} user: "${BECOME_USER}:${BECOME_USER}" volumes: diff --git a/tests/unittests/validator/test_literalvalidators.py b/tests/unittests/test_literalvalidators.py similarity index 100% rename from tests/unittests/validator/test_literalvalidators.py rename to tests/unittests/test_literalvalidators.py diff --git a/tests/unittests/validator/test_complexvalidators.py b/tests/unittests/validator/test_complexvalidators.py deleted file mode 100644 index cef0700..0000000 --- a/tests/unittests/validator/test_complexvalidators.py +++ /dev/null @@ -1,116 +0,0 @@ -################################################################## -# Copyright 2016 OSGeo Foundation, # -# represented by PyWPS Project Steering Committee, # -# licensed under MIT, Please consult LICENSE.txt for details # -################################################################## - -"""Unit tests for complex validator -""" - -import os -import tempfile - -from pyqgiswps.inout.formats import FORMATS -from pyqgiswps.validator.complexvalidator import ( - MODE, - validategeojson, - validategeotiff, - validategml, - validateshapefile, -) - -try: - import osgeo # noqa F401 -except ImportError: - WITH_GDAL = False -else: - WITH_GDAL = True - - -def get_input(name, schema, mime_type): - - class FakeFormat: - mimetype = 'text/plain' - schema = None - units = None - - def validate(self, data): - return True - - class FakeInput: - tempdir = tempfile.mkdtemp() - file = os.path.join( - os.path.abspath(os.path.dirname(__file__)), - '..', 'data', name) - format = FakeFormat() - - class data_format: - file = os.path.join( - os.path.abspath(os.path.dirname(__file__)), - '..', 'data', str(schema)) - - fake_input = FakeInput() - fake_input.stream = open(fake_input.file) - fake_input.data_format = data_format() - if schema: - fake_input.data_format.schema = 'file://' + fake_input.data_format.file - fake_input.data_format.mime_type = mime_type - - return fake_input - - -def test_gml_validator(): - """Test GML validator - """ - gml_input = get_input('gml/point.gml', 'point.xsd', FORMATS.GML.mime_type) - assert validategml(gml_input, MODE.NONE), 'NONE validation' - assert validategml(gml_input, MODE.SIMPLE), 'SIMPLE validation' - if WITH_GDAL: - assert validategml(gml_input, MODE.STRICT), 'STRICT validation' - # XXX Depends on external connection, may fail if not online - # Prevent test to fail if site is down - # assert validategml(gml_input, MODE.VERYSTRICT), 'VERYSTRICT validation' - gml_input.stream.close() - - -def test_geojson_validator(): - """Test GeoJSON validator - """ - geojson_input = get_input('json/point.geojson', 'json/schema/geojson.json', - FORMATS.GEOJSON.mime_type) - assert validategeojson(geojson_input, MODE.NONE), 'NONE validation' - assert validategeojson(geojson_input, MODE.SIMPLE), 'SIMPLE validation' - if WITH_GDAL: - assert validategeojson(geojson_input, MODE.STRICT), 'STRICT validation' - assert validategeojson(geojson_input, MODE.VERYSTRICT), 'VERYSTRICT validation' - geojson_input.stream.close() - - -def test_shapefile_validator(): - """Test ESRI Shapefile validator - """ - shapefile_input = get_input('shp/point.shp.zip', None, - FORMATS.SHP.mime_type) - assert validateshapefile(shapefile_input, MODE.NONE), 'NONE validation' - assert validateshapefile(shapefile_input, MODE.SIMPLE), 'SIMPLE validation' - if WITH_GDAL: - assert validateshapefile(shapefile_input, MODE.STRICT), 'STRICT validation' - shapefile_input.stream.close() - - -def test_geotiff_validator(): - """Test GeoTIFF validator - """ - geotiff_input = get_input('geotiff/dem.tiff', None, - FORMATS.GEOTIFF.mime_type) - assert validategeotiff(geotiff_input, MODE.NONE), 'NONE validation' - assert validategeotiff(geotiff_input, MODE.SIMPLE), 'SIMPLE validation' - if WITH_GDAL: - assert validategeotiff(geotiff_input, MODE.STRICT), 'STRICT validation' - geotiff_input.stream.close() - - -def test_fail_validator(): - fake_input = get_input('point.xsd', 'point.xsd', FORMATS.SHP.mime_type) - assert not validategml(fake_input, MODE.SIMPLE), 'SIMPLE validation invalid' - fake_input.stream.close()