diff --git a/default-sample.cfg b/default-sample.cfg index 6c4fb673d..19908d040 100644 --- a/default-sample.cfg +++ b/default-sample.cfg @@ -1,12 +1,38 @@ +# Configuration file sample +# +# Comment line with leading '#' to fallback to the hardcoded default value +# +# The configuration is load in this order: +# - First the hardcoded value are loaded +# - If no configuration file is provided in wsgi script, the first file that exist will be loaded: +# * on Unix/Linux System: +# - `/etc/pywps.cfg` +# - `$HOME/.pywps.cfg` +# * on Windows: +# - `pywps\\etc\\default.cfg` +# - Then if PYWPS_CFG environment variable is set, this file will be loaded +# +# Last loaded file override setting from the previous one. + + [metadata:main] +# Setup the title in GetCapabilities identification_title=PyWPS Demo server -identification_abstract=PyWPS testing and development server. Do NOT use this server in production environement. You shall setup PyWPS as WSGI application for production. Please refer documentation for further detials. +# Setup the abstract in GetCapabilities +identification_abstract=PyWPS testing and development server. Do NOT use this server in production environment. You shall setup PyWPS as WSGI application for production. Please refer documentation for further detials. +# Setup the keywords in GetCapabilities identification_keywords=WPS,GRASS,PyWPS, Demo, Dev identification_keywords_type=theme +# Setup the fees in GetCapabilities identification_fees=None +# Setup the AccessConstraints in GetCapabilities identification_accessconstraints=None -provider_name=PyWPS Developement team -provider_url=https://pywps.org/' +# Setup provider name in GetCapabilities +provider_name=PyWPS Development team +# Setup provider URL (informative) in GetCapabilities +provider_url=https://pywps.org/ + +# Setup Contacts information for GetCapabilities (informative) contact_name=Your Name contact_position=Developer contact_address=My Street @@ -14,39 +40,78 @@ contact_city=My City contact_stateorprovince=None contact_postalcode=000 00 contact_country=World, Internet -contact_phone=+00 00 11 22 33 -contact_fax=+00 99 88 77 66 -contact_email=info@yourdomain.org +contact_phone=+xx-xxx-xxx-xxxx +contact_fax=+xx-xxx-xxx-xxxx +contact_email=contact@yourdomain.org contact_url=https://pywps.org contact_hours=8:00-20:00UTC contact_instructions=Knock on the door contact_role=pointOfContact [server] +encoding=utf-8 +language=en-US +url=http://localhost:5000/wps maxsingleinputsize=1mb maxrequestsize=3mb -url=http://localhost:5000/wps outputurl=http://localhost:5000/outputs/ outputpath=outputs workdir=workdir -maxprocesses=10 +maxprocesses=30 parallelprocesses=2 storagetype=file +# hardcoded default : tempfile.gettempdir() +#temp_path=/tmp + +processes_path= +# list of allowed input paths (file url input) seperated by ':' +allowedinputpaths= + +# hardcoded default : tempfile.gettempdir() +#workdir= + +# If this flag is enabled it will set the HOME environment for each process to +# its current workdir (a temp folder). +sethomedir=false + +# If this flag is true PyWPS will remove the process temporary workdir after +# process has finished. +cleantempdir=true + +storagetype=file + +# File storage outputs can be copied, moved or linked from the workdir to the +# output folder. +# Allowed functions: "copy", "move", "link" (hardcoded default "copy") +storage_copy_function=copy + +# Handles the default mimetype for requests. +# valid options: "text/xml", "application/json" +default_mimetype=text/xml + +# Default indentation used for json data responses. +json_indent=2 + [processing] mode=default +# hardcoded default: os.path.dirname(os.path.realpath(sys.argv[0])) +#path= + +# https://github.com/natefoo/slurm-drmaa +drmaa_native_specification= + [logging] level=INFO file=logs/pywps.log -database=sqlite:///logs/pywps-logs.sqlite3 +database=sqlite:///:memory: format=%(asctime)s] [%(levelname)s] file=%(pathname)s line=%(lineno)s module=%(module)s function=%(funcName)s %(message)s - +prefix=pywps_ [grass] gisbase=/usr/local/grass-7.3.svn/ - [s3] bucket=my-org-wps region=us-east-1 diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 000000000..1f1f6bb94 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +addopts = --import-mode=importlib +pythonpath = tests +testpaths = tests diff --git a/pywps/configuration.py b/pywps/configuration.py index 592680f45..ec77e30be 100755 --- a/pywps/configuration.py +++ b/pywps/configuration.py @@ -88,21 +88,16 @@ def get_config_value(section, option, default_value=''): return value -def load_configuration(cfgfiles=None): - """Load PyWPS configuration from configuration files. - - The later configuration file in the array overwrites configuration - from the first. - - :param cfgfiles: list of configuration files - """ +def load_hardcoded_configuration(): + """Load PyWPS hardcoded configuration.""" global CONFIG - LOGGER.info('loading configuration') + LOGGER.debug('loading harcoded configuration') CONFIG = configparser.ConfigParser(os.environ, interpolation=EnvInterpolation()) - LOGGER.debug('setting default values') + tmpdir = tempfile.gettempdir() + CONFIG.add_section('server') CONFIG.set('server', 'encoding', 'utf-8') CONFIG.set('server', 'language', 'en-US') @@ -110,38 +105,23 @@ def load_configuration(cfgfiles=None): CONFIG.set('server', 'maxprocesses', '30') CONFIG.set('server', 'maxsingleinputsize', '1mb') CONFIG.set('server', 'maxrequestsize', '3mb') - CONFIG.set('server', 'temp_path', tempfile.gettempdir()) + CONFIG.set('server', 'temp_path', tmpdir) CONFIG.set('server', 'processes_path', '') - outputpath = tempfile.gettempdir() - CONFIG.set('server', 'outputurl', file_uri(outputpath)) - CONFIG.set('server', 'outputpath', outputpath) - # list of allowed input paths (file url input) seperated by ':' + CONFIG.set('server', 'outputurl', file_uri(tmpdir)) + CONFIG.set('server', 'outputpath', tmpdir) CONFIG.set('server', 'allowedinputpaths', '') - CONFIG.set('server', 'workdir', tempfile.gettempdir()) + CONFIG.set('server', 'workdir', tmpdir) CONFIG.set('server', 'parallelprocesses', '2') - # If this flag is enabled it will set the HOME environment - # for each process to its current workdir (a temp folder). CONFIG.set('server', 'sethomedir', 'false') - # If this flag is enabled PyWPS will remove the process temporary workdir - # after process has finished. CONFIG.set('server', 'cleantempdir', 'true') CONFIG.set('server', 'storagetype', 'file') - # File storage outputs can be copied, moved or linked - # from the workdir to the output folder. - # Allowed functions: "copy", "move", "link" (default "copy") CONFIG.set('server', 'storage_copy_function', 'copy') - - # handles the default mimetype for requests. - # available options: "text/xml", "application/json" CONFIG.set("server", "default_mimetype", "text/xml") - - # default json indentation for responses. CONFIG.set("server", "json_indent", "2") CONFIG.add_section('processing') CONFIG.set('processing', 'mode', 'default') CONFIG.set('processing', 'path', os.path.dirname(os.path.realpath(sys.argv[0]))) - # https://github.com/natefoo/slurm-drmaa CONFIG.set('processing', 'drmaa_native_specification', '') CONFIG.add_section('logging') @@ -185,6 +165,20 @@ def load_configuration(cfgfiles=None): CONFIG.set('s3', 'encrypt', 'false') CONFIG.set('s3', 'region', '') + +def load_configuration(cfgfiles=None): + """Load PyWPS configuration from configuration files. + + The later configuration file in the array overwrites configuration + from the first. + + :param cfgfiles: list of configuration files + """ + + global CONFIG + + load_hardcoded_configuration() + if not cfgfiles: cfgfiles = _get_default_config_files_location() diff --git a/pywps/tests.py b/pywps/tests.py index 366e98167..67f617434 100644 --- a/pywps/tests.py +++ b/pywps/tests.py @@ -212,6 +212,3 @@ def assert_wps_version(response, version="1.0.0"): '/ows:ServiceTypeVersion') found_version = elem[0].text assert version == found_version - tmp = Path(tempfile.mkdtemp()) - with open(tmp / "out.xml", "wb") as out: - out.writelines(response.response) diff --git a/tests/__init__.py b/tests/__init__.py index 7f84d2ce4..560137a2f 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -12,28 +12,28 @@ import configparser import pywps.configuration as config -from tests import test_capabilities -from tests import test_configuration -from tests import test_describe -from tests import test_execute -from tests import test_exceptions -from tests import test_inout -from tests import test_literaltypes -from tests import validator -from tests import test_ows -from tests import test_formats -from tests import test_dblog -from tests import test_wpsrequest -from tests import test_service -from tests import test_process -from tests import test_processing -from tests import test_assync -from tests import test_grass_location -from tests import test_storage -from tests import test_filestorage -from tests import test_s3storage -from tests.validator import test_complexvalidators -from tests.validator import test_literalvalidators +import test_capabilities +import test_configuration +import test_describe +import test_execute +import test_exceptions +import test_inout +import test_literaltypes +import validator +import test_ows +import test_formats +import test_dblog +import test_wpsrequest +import test_service +import test_process +import test_processing +import test_assync +import test_grass_location +import test_storage +import test_filestorage +import test_s3storage +from validator import test_complexvalidators +from validator import test_literalvalidators def find_grass(): diff --git a/tests/basic.py b/tests/basic.py new file mode 100644 index 000000000..4f3d0ae6b --- /dev/null +++ b/tests/basic.py @@ -0,0 +1,37 @@ +# coding=utf-8 +import os +import unittest +import pywps.configuration +from tempfile import TemporaryDirectory + + +class TestBase(unittest.TestCase): + + def setUp(self) -> None: + # Do not use load_configuration() that will load system configuration + # files such as /etc/pywps.cfg + pywps.configuration.load_hardcoded_configuration() + + # Ensure all data goes into ontime temporary directory + self.tmpdir = TemporaryDirectory(prefix="pywps_test_") + + # shortcut + set = pywps.configuration.CONFIG.set + + set('server', 'temp_path', f"{self.tmpdir.name}/temp_path") + set('server', 'outputpath', f"{self.tmpdir.name}/outputpath") + set('server', 'workdir', f"{self.tmpdir.name}/workdir") + + set('logging', 'level', 'DEBUG') + set('logging', 'file', f"{self.tmpdir.name}/logging-file.log") + set("logging", "database", f"sqlite:///{self.tmpdir.name}/test-pywps-logs.sqlite3") + + set('processing', 'path', f"{self.tmpdir.name}/processing_path") + + os.mkdir(f"{self.tmpdir.name}/temp_path") + os.mkdir(f"{self.tmpdir.name}/outputpath") + os.mkdir(f"{self.tmpdir.name}/workdir") + os.mkdir(f"{self.tmpdir.name}/processing_path") + + def tearDown(self) -> None: + self.tmpdir.cleanup() diff --git a/tests/test_app_exceptions.py b/tests/test_app_exceptions.py index c1fa00a67..480fee48a 100644 --- a/tests/test_app_exceptions.py +++ b/tests/test_app_exceptions.py @@ -3,14 +3,11 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase from pywps.app.exceptions import format_message, ProcessError, DEFAULT_ALLOWED_CHARS -class AppExceptionsTest(unittest.TestCase): - - def setUp(self): - pass +class AppExceptionsTest(TestBase): def test_format_message(self): assert format_message('no data available') == 'no data available' diff --git a/tests/test_assync.py b/tests/test_assync.py index d3bd87d2e..296bc2c07 100644 --- a/tests/test_assync.py +++ b/tests/test_assync.py @@ -3,13 +3,13 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase import pytest import time from pywps import Service, configuration from pywps import get_ElementMakerForVersion from pywps.tests import client_for, assert_response_accepted, assert_response_success -from .processes import Sleep +from processes import Sleep from owslib.wps import WPSExecution from pathlib import Path from tempfile import TemporaryDirectory @@ -20,22 +20,13 @@ WPS, OWS = get_ElementMakerForVersion(VERSION) -class ExecuteTest(unittest.TestCase): - def setUp(self) -> None: - - # Create temporary directory to create test database. - self.tmpdir = TemporaryDirectory() +class ExecuteTest(TestBase): + def setUp(self) -> None: + super().setUp() # Running processes using the MultiProcessing scheduler and a file-based database configuration.CONFIG.set('processing', 'mode', 'distributed') - configuration.CONFIG.set("logging", "database", f"sqlite:///{self.tmpdir.name}/test-pywps-logs.sqlite3") - - def tearDown(self) -> None: - # Cleanup temporary database - configuration.load_configuration() - self.tmpdir.cleanup() - def test_async(self): client = client_for(Service(processes=[Sleep()])) wps = WPSExecution() @@ -56,9 +47,10 @@ def test_async(self): # Parse response to extract the status file path url = resp.xml.xpath("//@statusLocation")[0] + print(url) # OWSlib only reads from URLs, not local files. So we need to read the response manually. - p = Path(url[6:]) + p = Path(configuration.get_config_value('server', 'outputpath')) / url.split('/')[-1] # Poll the process until it completes total_time = 0 @@ -77,6 +69,7 @@ def test_async(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_assync_inout.py b/tests/test_assync_inout.py index 496fcd714..206c6c182 100644 --- a/tests/test_assync_inout.py +++ b/tests/test_assync_inout.py @@ -3,6 +3,7 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## +from basic import TestBase from pywps import Service, Process, LiteralInput, ComplexOutput from pywps import FORMATS from pywps import get_ElementMakerForVersion @@ -37,32 +38,34 @@ def inout(request, response): ) -def test_assync_inout(): - client = client_for(Service(processes=[create_inout()])) - request_doc = WPS.Execute( - OWS.Identifier('inout'), - WPS.DataInputs( - WPS.Input( - OWS.Identifier('text'), - WPS.Data( - WPS.LiteralData( - "Hello World" +class TestAsyncInout(TestBase): + + def test_assync_inout(self): + client = client_for(Service(processes=[create_inout()])) + request_doc = WPS.Execute( + OWS.Identifier('inout'), + WPS.DataInputs( + WPS.Input( + OWS.Identifier('text'), + WPS.Data( + WPS.LiteralData( + "Hello World" + ) ) ) - ) - ), - WPS.ResponseForm( - WPS.ResponseDocument( - WPS.Output( - OWS.Identifier("text") + ), + WPS.ResponseForm( + WPS.ResponseDocument( + WPS.Output( + OWS.Identifier("text") + ), ), ), - ), - version="1.0.0" - ) - resp = client.post_xml(doc=request_doc) - assert resp.status_code == 200 + version="1.0.0" + ) + resp = client.post_xml(doc=request_doc) + assert resp.status_code == 200 - # TODO: - # . extract the status URL from the response - # . send a status request + # TODO: + # . extract the status URL from the response + # . send a status request diff --git a/tests/test_capabilities.py b/tests/test_capabilities.py index 8df785736..428c23de0 100644 --- a/tests/test_capabilities.py +++ b/tests/test_capabilities.py @@ -3,7 +3,7 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase from pywps import configuration from pywps.app import Process, Service from pywps.app.Common import Metadata @@ -13,7 +13,7 @@ WPS, OWS = get_ElementMakerForVersion("1.0.0") -class BadRequestTest(unittest.TestCase): +class BadRequestTest(TestBase): def test_bad_http_verb(self): client = client_for(Service()) @@ -42,9 +42,11 @@ def test_bad_request_type_with_post(self): assert resp.status_code == 400 -class CapabilitiesTest(unittest.TestCase): +class CapabilitiesTest(TestBase): def setUp(self): + super().setUp() + def pr1(): pass def pr2(): pass self.client = client_for( @@ -136,9 +138,9 @@ def test_version2(self): assert_wps_version(resp, version="2.0.0") -class CapabilitiesTranslationsTest(unittest.TestCase): +class CapabilitiesTranslationsTest(TestBase): def setUp(self): - configuration.load_configuration() + super().setUp() configuration.CONFIG.set('server', 'language', 'en-US,fr-CA') self.client = client_for( Service( @@ -161,9 +163,6 @@ def setUp(self): ) ) - def tearDown(self): - configuration.CONFIG.set('server', 'language', 'en-US') - def test_get_translated(self): resp = self.client.get('?Request=GetCapabilities&service=WPS&language=fr-CA') @@ -181,6 +180,8 @@ def test_get_translated(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_complexdata_io.py b/tests/test_complexdata_io.py index 84ab9a46b..dc495ab19 100644 --- a/tests/test_complexdata_io.py +++ b/tests/test_complexdata_io.py @@ -1,6 +1,6 @@ """Test embedding different file formats and different encodings within the tag.""" -import unittest +from basic import TestBase import os from pywps import get_ElementMakerForVersion from pywps.app.basic import get_xpath_ns @@ -66,7 +66,7 @@ def get_data(fn, encoding=None): return data -class RawInput(unittest.TestCase): +class RawInput(TestBase): def make_request(self, name, fn, fmt): """Create XML request embedding encoded data.""" diff --git a/tests/test_configuration.py b/tests/test_configuration.py index 925f55aba..1b3901404 100644 --- a/tests/test_configuration.py +++ b/tests/test_configuration.py @@ -5,26 +5,39 @@ """Tests for the configuration.""" +from basic import TestBase import os -import unittest + +import random from pywps import configuration -class TestEnvInterpolation(unittest.TestCase): +class TestEnvInterpolation(TestBase): """Test cases for env variable interpolation within the configuration.""" + test_value = "SOME_RANDOM_VALUE" + + def setUp(self) -> None: + super().setUp() + # Generate an unused environment key + self.test_key = "SOME_RANDOM_KEY" + while self.test_key in os.environ: + self.test_key = "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ", k=32)) + os.environ[self.test_key] = self.test_value + + def tearDown(self) -> None: + del os.environ[self.test_key] + super().tearDown() def test_expand_user(self): """Ensure we can parse a value with the $USER entry.""" - user = os.environ.get("USER") - configuration.CONFIG.read_string("[envinterpolationsection]\nuser=$USER") - assert user == configuration.CONFIG["envinterpolationsection"]["user"] + configuration.CONFIG.read_string(f"[envinterpolationsection]\nuser=${self.test_key}") + assert self.test_value == configuration.CONFIG["envinterpolationsection"]["user"] def test_expand_user_with_some_text(self): """Ensure we can parse a value with the $USER entry and some more text.""" - user = os.environ.get("USER") - new_user = "new_" + user - configuration.CONFIG.read_string("[envinterpolationsection]\nuser=new_${USER}") + new_user = "new_" + self.test_value + configuration.CONFIG.read_string(f"[envinterpolationsection]\nuser=new_${{{self.test_key}}}") assert new_user == configuration.CONFIG["envinterpolationsection"]["user"] def test_dont_expand_value_without_env_variable(self): @@ -41,6 +54,8 @@ def test_dont_expand_value_without_env_variable(self): def load_tests(loader=None, tests=None, pattern=None): """Load the tests and return the test suite for this file.""" + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_dblog.py b/tests/test_dblog.py index 7e26a685c..d3707b52b 100644 --- a/tests/test_dblog.py +++ b/tests/test_dblog.py @@ -6,18 +6,18 @@ """Unit tests for dblog """ -import unittest +from basic import TestBase from pywps import configuration from pywps.dblog import get_session from pywps.dblog import ProcessInstance -class DBLogTest(unittest.TestCase): +class DBLogTest(TestBase): """DBGLog test cases""" def setUp(self): - + super().setUp() self.database = configuration.get_config_value('logging', 'database') def test_0_dblog(self): @@ -47,6 +47,8 @@ def test_db_content(self): def load_tests(loader=None, tests=None, pattern=None): """Load local tests """ + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_describe.py b/tests/test_describe.py index 17b6d9a91..df21fc2eb 100644 --- a/tests/test_describe.py +++ b/tests/test_describe.py @@ -3,7 +3,8 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase + import pytest from collections import namedtuple from pywps import Process, Service, LiteralInput, ComplexInput, BoundingBoxInput @@ -76,9 +77,11 @@ def get_describe_result(resp): return result -class DescribeProcessTest(unittest.TestCase): +class DescribeProcessTest(TestBase): def setUp(self): + super().setUp() + def hello(request): pass @@ -139,9 +142,11 @@ def test_post_two_args(self): assert [pr.identifier for pr in result] == ['hello', 'ping'] -class DescribeProcessTranslationsTest(unittest.TestCase): +class DescribeProcessTranslationsTest(TestBase): def setUp(self): + super().setUp() + configuration.get_config_value('server', 'language') configuration.CONFIG.set('server', 'language', 'en-US,fr-CA') self.client = client_for( @@ -197,9 +202,6 @@ def setUp(self): ) ) - def tearDown(self): - configuration.CONFIG.set('server', 'language', 'en-US') - def test_get_describe_translations(self): resp = self.client.get('?Request=DescribeProcess&service=wps&version=1.0.0&identifier=all&language=fr-CA') @@ -223,7 +225,7 @@ def test_get_describe_translations(self): assert output_abstracts == ["Description", "Description"] -class DescribeProcessInputTest(unittest.TestCase): +class DescribeProcessInputTest(TestBase): def describe_process(self, process): client = client_for(Service(processes=[process])) @@ -271,7 +273,7 @@ def hello(request): assert result.inputs == [('the_number', 'literal', 'integer', "0")] -class InputDescriptionTest(unittest.TestCase): +class InputDescriptionTest(TestBase): def test_literal_integer_input(self): literal = LiteralInput('foo', 'Literal foo', data_type='positiveInteger', @@ -334,7 +336,7 @@ def test_bbox_input(self): assert data["dimensions"] == 2 -class OutputDescriptionTest(unittest.TestCase): +class OutputDescriptionTest(TestBase): @pytest.mark.skip(reason="not working") def test_literal_output(self): @@ -395,6 +397,8 @@ def test_bbox_output(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py index 578568d8e..b85cb1cbd 100644 --- a/tests/test_exceptions.py +++ b/tests/test_exceptions.py @@ -3,7 +3,7 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase from pywps import Service, get_ElementMakerForVersion from pywps.app.basic import get_xpath_ns from pywps.tests import assert_pywps_version, client_for @@ -15,9 +15,10 @@ xpath_ns = get_xpath_ns(VERSION) -class ExceptionsTest(unittest.TestCase): +class ExceptionsTest(TestBase): def setUp(self): + super().setUp() self.client = client_for(Service(processes=[])) def test_invalid_parameter_value(self): @@ -50,6 +51,8 @@ def test_bad_request(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_execute.py b/tests/test_execute.py index d3432c05d..bcc2f0a30 100644 --- a/tests/test_execute.py +++ b/tests/test_execute.py @@ -3,11 +3,11 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase import pytest from pywps import xml_util as etree import json -import tempfile + import os.path from pywps import Service, Process, LiteralOutput, LiteralInput,\ BoundingBoxOutput, BoundingBoxInput, Format, ComplexInput, ComplexOutput, FORMATS @@ -201,7 +201,7 @@ def _handler(request, response): def create_metalink_process(): - from .processes.metalinkprocess import MultipleOutputs + from processes.metalinkprocess import MultipleOutputs return MultipleOutputs() @@ -232,7 +232,7 @@ def get_output(doc): return output -class ExecuteTest(unittest.TestCase): +class ExecuteTest(TestBase): """Test for Exeucte request KVP request""" @pytest.mark.xfail(reason="test.opendap.org is offline") @@ -563,15 +563,13 @@ def test_output_response_dataType(self): assert el.attrib['dataType'] == 'string' -class AllowedInputReferenceExecuteTest(unittest.TestCase): +class AllowedInputReferenceExecuteTest(TestBase): def setUp(self): + super().setUp() self.allowedinputpaths = configuration.get_config_value('server', 'allowedinputpaths') configuration.CONFIG.set('server', 'allowedinputpaths', DATA_DIR) - def tearDown(self): - configuration.CONFIG.set('server', 'allowedinputpaths', self.allowedinputpaths) - def test_geojson_input_reference_rest(self): geojson = os.path.join(DATA_DIR, 'json', 'point.geojson') with open(geojson) as f: @@ -596,15 +594,13 @@ def test_geojson_input_reference_rest(self): assert_response_success_json(resp, result) -class ExecuteTranslationsTest(unittest.TestCase): +class ExecuteTranslationsTest(TestBase): def setUp(self): + super().setUp() configuration.get_config_value('server', 'language') configuration.CONFIG.set('server', 'language', 'en-US,fr-CA') - def tearDown(self): - configuration.CONFIG.set('server', 'language', 'en-US') - def test_translations(self): client = client_for(Service(processes=[create_translated_greeter()])) @@ -649,7 +645,7 @@ def test_translations(self): assert output_abstract == ["Description"] -class ExecuteXmlParserTest(unittest.TestCase): +class ExecuteXmlParserTest(TestBase): """Tests for Execute request XML Parser """ @@ -796,7 +792,7 @@ def test_build_input_file_name(self): from pywps.inout.basic import ComplexInput h = ComplexInput('ci') - h.workdir = workdir = tempfile.mkdtemp() + h.workdir = workdir = configuration.get_config_value('server', 'workdir') self.assertEqual( h._build_file_name('http://path/to/test.txt'), @@ -824,6 +820,8 @@ def test_build_input_file_name(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_filestorage.py b/tests/test_filestorage.py index 96fb232f7..e97180ac4 100644 --- a/tests/test_filestorage.py +++ b/tests/test_filestorage.py @@ -2,6 +2,8 @@ # Copyright 2018 Open Source Geospatial Foundation and others # # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## +from basic import TestBase + from pathlib import Path from pywps.inout.storage.file import FileStorageBuilder, FileStorage, _build_output_name @@ -11,15 +13,12 @@ from pywps import configuration, FORMATS -import tempfile - -import unittest - -class FileStorageTests(unittest.TestCase): +class FileStorageTests(TestBase): def setUp(self): - self.tmp_dir = tempfile.mkdtemp() + super().setUp() + self.tmp_dir = self.tmpdir.name def test_build_output_name(self): storage = FileStorageBuilder().build() @@ -85,6 +84,8 @@ def test_location(self): def load_tests(loader=None, tests=None, pattern=None): """Load local tests """ + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_formats.py b/tests/test_formats.py index 2dfff8ff5..048f9000b 100644 --- a/tests/test_formats.py +++ b/tests/test_formats.py @@ -5,7 +5,7 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase from pywps.inout.formats import Format, get_format, FORMATS from pywps.app.basic import get_xpath_ns @@ -14,10 +14,11 @@ xpath_ns = get_xpath_ns("1.0.0") -class FormatsTest(unittest.TestCase): +class FormatsTest(TestBase): """Formats test cases""" def setUp(self): + super().setUp() def validate(self, inpt, level=None): """fake validate method @@ -26,9 +27,6 @@ def validate(self, inpt, level=None): self.validate = validate - def tearDown(self): - pass - def test_format_class(self): """Test pywps.formats.Format class """ @@ -114,6 +112,8 @@ def test_json_in(self): def load_tests(loader=None, tests=None, pattern=None): """Load local tests """ + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_grass_location.py b/tests/test_grass_location.py index 2afb8c8e0..d075327b5 100644 --- a/tests/test_grass_location.py +++ b/tests/test_grass_location.py @@ -3,7 +3,8 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase + import pywps.configuration as config from pywps import Service, Process, ComplexInput, get_format, \ get_ElementMakerForVersion @@ -55,11 +56,13 @@ def file_location(request, response): grass_location="complexinput:input1") -class GRASSTests(unittest.TestCase): +class GRASSTests(TestBase): """Test creating GRASS locations and mapsets in different ways.""" def setUp(self): """Skip test if GRASS is not installed on the machine.""" + super().setUp() + if not config.CONFIG.get('grass', 'gisbase'): self.skipTest('GRASS lib not found') @@ -100,6 +103,8 @@ def test_file_based_location(self): def load_tests(loader=None, tests=None, pattern=None): """Load tests.""" + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_inout.py b/tests/test_inout.py index 7755e5c72..6d532892a 100644 --- a/tests/test_inout.py +++ b/tests/test_inout.py @@ -7,11 +7,11 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## +from basic import TestBase import requests import os import tempfile -import unittest import json from pywps import inout import base64 @@ -42,17 +42,15 @@ def get_data_format(mime_type): return Format(mime_type=mime_type, validate=get_validator(mime_type)) -class IOHandlerTest(unittest.TestCase): +class IOHandlerTest(TestBase): """IOHandler test cases""" def setUp(self): - tmp_dir = tempfile.mkdtemp() + super().setUp() + tmp_dir = self.tmpdir.name self.iohandler = IOHandler(workdir=tmp_dir) self._value = 'lalala' - def tearDown(self): - pass - def test_basic_IOHandler(self): """Test basic IOHandler""" self.assertTrue(os.path.isdir(self.iohandler.workdir)) @@ -120,7 +118,7 @@ def test_stream(self): def test_file(self): """Test file input IOHandler""" - (fd, tmp_file) = tempfile.mkstemp() + (fd, tmp_file) = tempfile.mkstemp(dir=self.tmpdir.name) source = tmp_file file_handler = open(tmp_file, 'w') file_handler.write(self._value) @@ -143,12 +141,12 @@ def test_url(self): def test_workdir(self): """Test workdir""" - workdir = tempfile.mkdtemp() + workdir = tempfile.mkdtemp(dir=self.tmpdir.name) self.iohandler.workdir = workdir self.assertTrue(os.path.isdir(self.iohandler.workdir)) # make another - workdir = tempfile.mkdtemp() + workdir = tempfile.mkdtemp(dir=self.tmpdir.name) self.iohandler.workdir = workdir self.assertTrue(os.path.isdir(self.iohandler.workdir)) @@ -185,11 +183,12 @@ def test_is_textfile(self): self.assertTrue(_is_textfile(geojson)) -class ComplexInputTest(unittest.TestCase): +class ComplexInputTest(TestBase): """ComplexInput test cases""" def setUp(self): - self.tmp_dir = tempfile.mkdtemp() + super().setUp() + self.tmp_dir = self.tmpdir.name data_format = get_data_format('application/json') self.complex_in = inout.inputs.ComplexInput(identifier="complexinput", title='MyComplex', @@ -220,11 +219,12 @@ def test_data_format(self): self.assertIsInstance(self.complex_in.supported_formats[0], Format) -class SerializationComplexInputTest(unittest.TestCase): +class SerializationComplexInputTest(TestBase): """ComplexInput test cases""" def setUp(self): - self.tmp_dir = tempfile.mkdtemp() + super().setUp() + self.tmp_dir = self.tmpdir.name def make_complex_input(self): complex = inout.inputs.ComplexInput( @@ -340,11 +340,12 @@ def test_complex_input_url(self): self.assertEqual(complex.prop, 'url') -class SerializationLiteralInputTest(unittest.TestCase): +class SerializationLiteralInputTest(TestBase): """LiteralInput test cases""" def setUp(self): - self.tmp_dir = tempfile.mkdtemp() + super().setUp() + self.tmp_dir = self.tmpdir.name def make_literal_input(self): literal = inout.inputs.LiteralInput( @@ -393,11 +394,12 @@ def test_literal_input(self): self.assert_literal_equals(literal, literal2) -class SerializationBoundingBoxInputTest(unittest.TestCase): +class SerializationBoundingBoxInputTest(TestBase): """LiteralInput test cases""" def setUp(self): - self.tmp_dir = tempfile.mkdtemp() + super().setUp() + self.tmp_dir = self.tmpdir.name def make_bbox_input(self): bbox = inout.inputs.BoundingBoxInput( @@ -440,11 +442,12 @@ def test_bbox_input(self): self.assert_bbox_equals(bbox, bbox2) -class DodsComplexInputTest(unittest.TestCase): +class DodsComplexInputTest(TestBase): """ComplexInput test cases""" def setUp(self): - self.tmp_dir = tempfile.mkdtemp() + super().setUp() + self.tmp_dir = self.tmpdir.name data_format = get_data_format('application/x-ogc-dods') self.complex_in = ComplexInput(identifier="complexinput", title='MyComplex', @@ -473,11 +476,12 @@ def test_contruct(self): self.assertIsInstance(self.complex_in, ComplexInput) -class ComplexOutputTest(unittest.TestCase): +class ComplexOutputTest(TestBase): """ComplexOutput test cases""" def setUp(self): - tmp_dir = tempfile.mkdtemp() + super().setUp() + tmp_dir = self.tmpdir.name data_format = get_data_format('application/json') self.complex_out = inout.outputs.ComplexOutput( identifier="complexoutput", @@ -568,10 +572,11 @@ def test_json(self): ) -class SimpleHandlerTest(unittest.TestCase): +class SimpleHandlerTest(TestBase): """SimpleHandler test cases""" def setUp(self): + super().setUp() data_type = 'integer' @@ -584,10 +589,11 @@ def test_data_type(self): self.assertEqual(convert(self.simple_handler.data_type, '1'), 1) -class LiteralInputTest(unittest.TestCase): +class LiteralInputTest(TestBase): """LiteralInput test cases""" def setUp(self): + super().setUp() self.literal_input = inout.inputs.LiteralInput( identifier="literalinput", @@ -690,10 +696,11 @@ def test_translations(self): assert identifier == self.literal_input.identifier -class LiteralOutputTest(unittest.TestCase): +class LiteralOutputTest(TestBase): """LiteralOutput test cases""" def setUp(self): + super().setUp() self.literal_output = inout.outputs.LiteralOutput( "literaloutput", @@ -725,11 +732,11 @@ def test_json(self): ) -class BBoxInputTest(unittest.TestCase): +class BBoxInputTest(TestBase): """BountingBoxInput test cases""" def setUp(self): - + super().setUp() self.bbox_input = inout.inputs.BoundingBoxInput( "bboxinput", title="BBox input", @@ -758,10 +765,11 @@ def test_json_out(self): ) -class BBoxOutputTest(unittest.TestCase): +class BBoxOutputTest(TestBase): """BoundingBoxOutput test cases""" def setUp(self): + super().setUp() self.bbox_out = inout.outputs.BoundingBoxOutput( "bboxoutput", title="BBox output", @@ -790,8 +798,11 @@ def test_json(self): ) -class TestMetaLink(unittest.TestCase): - tmp_dir = tempfile.mkdtemp() +class TestMetaLink(TestBase): + + def setUp(self) -> None: + super().setUp() + self.tmp_dir = self.tmpdir.name def metafile(self): mf = MetaFile('identifier', 'title', fmt=FORMATS.JSON) @@ -875,6 +886,8 @@ def test_set_size(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_literaltypes.py b/tests/test_literaltypes.py index 7573ae808..4024e07ea 100644 --- a/tests/test_literaltypes.py +++ b/tests/test_literaltypes.py @@ -5,7 +5,7 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase import datetime from pywps.inout.literaltypes import ( convert_integer, @@ -21,10 +21,11 @@ ) -class ValuesReferenceTest(unittest.TestCase): +class ValuesReferenceTest(TestBase): """ValuesReference test cases""" def setUp(self): + super().setUp() self.reference = "https://en.wikipedia.org/w/api.php?action=opensearch&search=scotland&format=json" def test_json(self): @@ -34,7 +35,7 @@ def test_json(self): self.assertEqual(new_val_ref.reference, self.reference) -class ConvertorTest(unittest.TestCase): +class ConvertorTest(TestBase): """IOHandler test cases""" def test_integer(self): @@ -121,6 +122,8 @@ def test_anyuri(self): convert_anyURI("ftp:///deep/path/;params?query#fragment") def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_ows.py b/tests/test_ows.py index cdd796ac5..de354c65d 100644 --- a/tests/test_ows.py +++ b/tests/test_ows.py @@ -6,9 +6,9 @@ __author__ = "Luis de Sousa" __date__ = "10-03-2015" +from basic import TestBase import os import tempfile -import unittest from pywps import Service, Process, ComplexInput, ComplexOutput, Format, FORMATS, get_format from pywps.exceptions import NoApplicableCode from pywps import get_ElementMakerForVersion @@ -21,76 +21,75 @@ WPS, OWS = get_ElementMakerForVersion("1.0.0") -def create_feature(): - - def feature(request, response): - input = request.inputs['input'][0].file - response.outputs['output'].data_format = FORMATS.GML - response.outputs['output'].file = input - return response - - return Process(handler=feature, - identifier='feature', - title='Process Feature', - inputs=[ComplexInput( - 'input', - title='Input', - supported_formats=[get_format('GML')])], - outputs=[ComplexOutput( - 'output', - title='Output', - supported_formats=[get_format('GML')])]) - - -def create_sum_one(): - - def sum_one(request, response): - input = request.inputs['input'][0].file - # What do we need to assert a Complex input? - # assert type(input) is str - - import grass.script as grass - - # Import the raster and set the region - if grass.run_command("r.in.gdal", flags="o", out="input", - input=input, quiet=True) != 0: - raise NoApplicableCode("Could not import cost map. " - "Please check the WCS service.") - - if grass.run_command("g.region", flags="a", rast="input") != 0: - raise NoApplicableCode("Could not set GRASS region.") - - # Add 1 - if grass.mapcalc("$output = $input + $value", output="output", - input="input", value=1.0, quiet=True): - raise NoApplicableCode("Could not use GRASS map calculator.") - - # Export the result - _, out = tempfile.mkstemp() - os.environ['GRASS_VERBOSE'] = '-1' - if grass.run_command("r.out.gdal", flags="f", input="output", - type="UInt16", output=out, overwrite=True) != 0: - raise NoApplicableCode("Could not export result from GRASS.") - del os.environ['GRASS_VERBOSE'] - - response.outputs['output'].file = out - return response - - return Process(handler=sum_one, - identifier='sum_one', - title='Process Sum One', - inputs=[ComplexInput( - 'input', - title='Input', - supported_formats=[Format('image/img')])], - outputs=[ComplexOutput( - 'output', - title='Output', - supported_formats=[get_format('GEOTIFF')])], - grass_location='epsg:4326') - - -class ExecuteTests(unittest.TestCase): +class ExecuteTests(TestBase): + + def create_feature(self): + + def feature(request, response): + input = request.inputs['input'][0].file + response.outputs['output'].data_format = FORMATS.GML + response.outputs['output'].file = input + return response + + return Process(handler=feature, + identifier='feature', + title='Process Feature', + inputs=[ComplexInput( + 'input', + title='Input', + supported_formats=[get_format('GML')])], + outputs=[ComplexOutput( + 'output', + title='Output', + supported_formats=[get_format('GML')])]) + + def create_sum_one(self): + + def sum_one(request, response): + input = request.inputs['input'][0].file + # What do we need to assert a Complex input? + # assert type(input) is str + + import grass.script as grass + + # Import the raster and set the region + if grass.run_command("r.in.gdal", flags="o", out="input", + input=input, quiet=True) != 0: + raise NoApplicableCode("Could not import cost map. " + "Please check the WCS service.") + + if grass.run_command("g.region", flags="a", rast="input") != 0: + raise NoApplicableCode("Could not set GRASS region.") + + # Add 1 + if grass.mapcalc("$output = $input + $value", output="output", + input="input", value=1.0, quiet=True): + raise NoApplicableCode("Could not use GRASS map calculator.") + + # Export the result + _, out = tempfile.mkstemp(dir=self.tmpdir.name) + os.environ['GRASS_VERBOSE'] = '-1' + if grass.run_command("r.out.gdal", flags="f", input="output", + type="UInt16", output=out, + overwrite=True) != 0: + raise NoApplicableCode("Could not export result from GRASS.") + del os.environ['GRASS_VERBOSE'] + + response.outputs['output'].file = out + return response + + return Process(handler=sum_one, + identifier='sum_one', + title='Process Sum One', + inputs=[ComplexInput( + 'input', + title='Input', + supported_formats=[Format('image/img')])], + outputs=[ComplexOutput( + 'output', + title='Output', + supported_formats=[get_format('GEOTIFF')])], + grass_location='epsg:4326') def test_wfs(self): if not service_ok('https://demo.mapserver.org'): @@ -124,7 +123,7 @@ def test_wcs(self): if not service_ok('https://demo.mapserver.org'): self.skipTest("mapserver is unreachable") - client = client_for(Service(processes=[create_sum_one()])) + client = client_for(Service(processes=[self.create_sum_one()])) request_doc = WPS.Execute( OWS.Identifier('sum_one'), WPS.DataInputs( @@ -144,6 +143,8 @@ def test_wcs(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_process.py b/tests/test_process.py index b1b2ebaa2..09e21fc7a 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -6,7 +6,7 @@ """Test process """ -import unittest +from basic import TestBase from pywps import Process from pywps.app.Common import Metadata @@ -38,9 +38,10 @@ def donothing(request, response): pass -class ProcessTestCase(unittest.TestCase): +class ProcessTestCase(TestBase): def setUp(self): + super().setUp() self.process = DoNothing() def test_get_input_title(self): @@ -76,6 +77,8 @@ def test_get_translations(self): def load_tests(loader=None, tests=None, pattern=None): """Load local tests """ + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_processing.py b/tests/test_processing.py index 70d342de9..c7fa2f046 100644 --- a/tests/test_processing.py +++ b/tests/test_processing.py @@ -6,7 +6,7 @@ """Unit tests for processing """ -import unittest +from basic import TestBase import json import uuid @@ -18,17 +18,21 @@ from pywps.app import WPSRequest from pywps.response.execute import ExecuteResponse -from .processes import Greeter, InOut, BBox +from processes import Greeter, InOut, BBox -class GreeterProcessingTest(unittest.TestCase): +class GreeterProcessingTest(TestBase): """Processing test case with Greeter process""" def setUp(self): + super().setUp() + + self.workdir = pywps.configuration.get_config_value('server', 'workdir') + self.uuid = uuid.uuid1() self.dummy_process = Greeter() self.dummy_process._set_uuid(self.uuid) - self.dummy_process.set_workdir('/tmp') + self.dummy_process.set_workdir(self.workdir) self.wps_request = WPSRequest() self.wps_response = ExecuteResponse(self.wps_request, self.uuid, process=self.dummy_process) @@ -53,25 +57,29 @@ def test_job_json(self): new_job = Job.from_json(json.loads(self.job.json)) self.assertEqual(new_job.name, 'greeter') self.assertEqual(new_job.uuid, str(self.uuid)) - self.assertEqual(new_job.workdir, '/tmp') + self.assertEqual(new_job.workdir, self.workdir) self.assertEqual(len(new_job.process.inputs), 1) def test_job_dump(self): new_job = Job.load(self.job.dump()) self.assertEqual(new_job.name, 'greeter') self.assertEqual(new_job.uuid, str(self.uuid)) - self.assertEqual(new_job.workdir, '/tmp') + self.assertEqual(new_job.workdir, self.workdir) self.assertEqual(len(new_job.process.inputs), 1) -class InOutProcessingTest(unittest.TestCase): +class InOutProcessingTest(TestBase): """Processing test case with InOut process""" def setUp(self): + super().setUp() + + self.workdir = pywps.configuration.get_config_value('server', 'workdir') + self.uuid = uuid.uuid1() self.dummy_process = InOut() self.dummy_process._set_uuid(self.uuid) - self.dummy_process.set_workdir('/tmp') + self.dummy_process.set_workdir(self.workdir) self.wps_request = WPSRequest() self.wps_response = ExecuteResponse(self.wps_request, self.uuid, process=self.dummy_process) @@ -84,7 +92,7 @@ def test_job_json(self): new_job = Job.from_json(json.loads(self.job.json)) self.assertEqual(new_job.name, 'inout') self.assertEqual(new_job.uuid, str(self.uuid)) - self.assertEqual(new_job.workdir, '/tmp') + self.assertEqual(new_job.workdir, self.workdir) self.assertEqual(len(new_job.process.inputs), 3) self.assertEqual(new_job.json, self.job.json) # idempotent test @@ -92,19 +100,23 @@ def test_job_dump(self): new_job = Job.load(self.job.dump()) self.assertEqual(new_job.name, 'inout') self.assertEqual(new_job.uuid, str(self.uuid)) - self.assertEqual(new_job.workdir, '/tmp') + self.assertEqual(new_job.workdir, self.workdir) self.assertEqual(len(new_job.process.inputs), 3) self.assertEqual(new_job.json, self.job.json) # idempotent test -class BBoxProcessingTest(unittest.TestCase): +class BBoxProcessingTest(TestBase): """Processing test case with BBox input and output process""" def setUp(self): + super().setUp() + + self.workdir = pywps.configuration.get_config_value('server', 'workdir') + self.uuid = uuid.uuid1() self.dummy_process = BBox() self.dummy_process._set_uuid(self.uuid) - self.dummy_process.set_workdir('/tmp') + self.dummy_process.set_workdir(self.workdir) self.wps_request = WPSRequest() self.wps_response = ExecuteResponse(self.wps_request, self.uuid, process=self.dummy_process) @@ -117,7 +129,7 @@ def test_job_json(self): new_job = Job.from_json(json.loads(self.job.json)) self.assertEqual(new_job.name, 'bbox_test') self.assertEqual(new_job.uuid, str(self.uuid)) - self.assertEqual(new_job.workdir, '/tmp') + self.assertEqual(new_job.workdir, self.workdir) self.assertEqual(len(new_job.process.inputs), 1) self.assertEqual(new_job.json, self.job.json) # idempotent test @@ -125,7 +137,7 @@ def test_job_dump(self): new_job = Job.load(self.job.dump()) self.assertEqual(new_job.name, 'bbox_test') self.assertEqual(new_job.uuid, str(self.uuid)) - self.assertEqual(new_job.workdir, '/tmp') + self.assertEqual(new_job.workdir, self.workdir) self.assertEqual(len(new_job.process.inputs), 1) self.assertEqual(new_job.json, self.job.json) # idempotent test @@ -133,6 +145,8 @@ def test_job_dump(self): def load_tests(loader=None, tests=None, pattern=None): """Load local tests """ + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_s3storage.py b/tests/test_s3storage.py index 7998abac7..8ee694fab 100644 --- a/tests/test_s3storage.py +++ b/tests/test_s3storage.py @@ -3,29 +3,25 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## +from basic import TestBase from pywps.inout.storage.s3 import S3StorageBuilder, S3Storage from pywps.inout.storage import STORE_TYPE from pywps.inout.basic import ComplexOutput from pywps import configuration, FORMATS -import tempfile - -import unittest from unittest.mock import patch -class S3StorageTests(unittest.TestCase): - - def setUp(self): - self.tmp_dir = tempfile.mkdtemp() +class S3StorageTests(TestBase): @patch('pywps.inout.storage.s3.S3Storage.uploadData') def test_store(self, uploadData): configuration.CONFIG.set('s3', 'bucket', 'notrealbucket') configuration.CONFIG.set('s3', 'prefix', 'wps') storage = S3StorageBuilder().build() - output = ComplexOutput('testme', 'Test', supported_formats=[FORMATS.TEXT], workdir=self.tmp_dir) + output = ComplexOutput('testme', 'Test', supported_formats=[FORMATS.TEXT], + workdir=configuration.get_config_value('server', 'workdir')) output.data = "Hello World!" store_type, filename, url = storage.store(output) @@ -58,6 +54,8 @@ def test_write(self, uploadData): def load_tests(loader=None, tests=None, pattern=None): """Load local tests """ + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_service.py b/tests/test_service.py index b67685c39..035b55072 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1,10 +1,10 @@ -import unittest +from basic import TestBase from pywps.app.Service import _validate_file_input from pywps.exceptions import FileURLNotSupported -class ServiceTest(unittest.TestCase): +class ServiceTest(TestBase): def test_validate_file_input(self): try: @@ -16,6 +16,8 @@ def test_validate_file_input(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_storage.py b/tests/test_storage.py index 8aea0c0b3..59ac081cc 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -2,6 +2,7 @@ # Copyright 2018 Open Source Geospatial Foundation and others # # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## +from basic import TestBase import pytest from pywps.inout.storage.builder import StorageBuilder @@ -11,43 +12,49 @@ from pywps import configuration from pathlib import Path -import tempfile +import os -@pytest.fixture -def fake_output(tmp_path): - class FakeOutput(object): - """Fake output object for testing.""" - def __init__(self): - self.identifier = "fake_output" - self.file = self._get_file() - self.uuid = None +class FakeOutput(object): + """Fake output object for testing.""" - def _get_file(self): - fn = tmp_path / 'file.tiff' - fn.touch() - return str(fn.absolute()) + def __init__(self, tmp_path): + self.identifier = "fake_output" + fn = Path(tmp_path) / "file.tiff" + fn.touch() + self.file = str(fn.absolute()) + self.uuid = None - return FakeOutput() - -class TestStorageBuilder(): +class TestDefaultStorageBuilder(TestBase): def test_default_storage(self): storage = StorageBuilder.buildStorage() assert isinstance(storage, FileStorage) - def test_s3_storage(self): + +class TestS3StorageBuilder(TestBase): + + def setUp(self) -> None: + super().setUp() configuration.CONFIG.set('server', 'storagetype', 's3') + + def test_s3_storage(self): storage = StorageBuilder.buildStorage() assert isinstance(storage, S3Storage) - def test_recursive_directory_creation(self, fake_output): - """Test that outputpath is created.""" + +class TestFileStorageBuilder(TestBase): + + def setUp(self) -> None: + super().setUp() configuration.CONFIG.set('server', 'storagetype', 'file') - outputpath = Path(tempfile.gettempdir()) / "a" / "b" / "c" - configuration.CONFIG.set('server', 'outputpath', str(outputpath)) - storage = StorageBuilder.buildStorage() + self.opath = os.path.join(self.tmpdir.name, "a", "b", "c") + configuration.CONFIG.set('server', 'outputpath', self.opath) - storage.store(fake_output) - assert outputpath.exists() + def test_recursive_directory_creation(self): + """Test that outputpath is created.""" + storage = StorageBuilder.buildStorage() + fn = FakeOutput(self.tmpdir.name) + storage.store(fn) + assert os.path.exists(self.opath) diff --git a/tests/test_wpsrequest.py b/tests/test_wpsrequest.py index 837755523..cc5868295 100644 --- a/tests/test_wpsrequest.py +++ b/tests/test_wpsrequest.py @@ -3,7 +3,7 @@ # licensed under MIT, Please consult LICENSE.txt for details # ################################################################## -import unittest +from basic import TestBase from pywps.app import WPSRequest import tempfile import datetime @@ -12,12 +12,13 @@ from pywps.inout.literaltypes import AnyValue -class WPSRequestTest(unittest.TestCase): +class WPSRequestTest(TestBase): def setUp(self): + super().setUp() self.request = WPSRequest() - self.tempfile = tempfile.mktemp() + self.tempfile = tempfile.mktemp(dir=self.tmpdir.name) x = open(self.tempfile, 'w') x.write("ahoj") @@ -158,6 +159,8 @@ def test_json_inout_bbox(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/test_xml_util.py b/tests/test_xml_util.py index e9325afc3..627509fd7 100644 --- a/tests/test_xml_util.py +++ b/tests/test_xml_util.py @@ -1,3 +1,5 @@ +from basic import TestBase + from pywps import xml_util as etree from io import StringIO diff --git a/tests/validator/test_complexvalidators.py b/tests/validator/test_complexvalidators.py index 070c2f769..1c868cab1 100644 --- a/tests/validator/test_complexvalidators.py +++ b/tests/validator/test_complexvalidators.py @@ -6,7 +6,7 @@ """Unit tests for complex validator """ -import unittest +from basic import TestBase import pytest from pywps.validator.mode import MODE from pywps.validator.complexvalidator import ( @@ -35,51 +35,44 @@ WITH_NC4 = True -def get_input(name, schema, mime_type): - - class FakeFormat(object): - mimetype = 'text/plain' - schema = None - units = None - - def validate(self, data): - return True - - class FakeInput(object): - tempdir = tempfile.mkdtemp() - file = os.path.join( - os.path.abspath(os.path.dirname(__file__)), - '..', 'data', name) - format = FakeFormat() +class ValidateTest(TestBase): + """Complex validator test cases""" - class data_format(object): - file = os.path.join( - os.path.abspath(os.path.dirname(__file__)), - '..', 'data', str(schema)) + def get_input(self, name, schema, mime_type): - 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 + class FakeFormat(object): + mimetype = 'text/plain' + schema = None + units = None - return fake_input + def validate(self, data): + return True + class FakeInput(object): + tempdir = tempfile.mkdtemp(dir=self.tmpdir.name) + file = os.path.join( + os.path.abspath(os.path.dirname(__file__)), + '..', 'data', name) + format = FakeFormat() -class ValidateTest(unittest.TestCase): - """Complex validator test cases""" + class data_format(object): + file = os.path.join( + os.path.abspath(os.path.dirname(__file__)), + '..', 'data', str(schema)) - def setUp(self): - pass + 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 - def tearDown(self): - pass + return fake_input def test_gml_validator(self): """Test GML validator """ - gml_input = get_input('gml/point.gml', 'point.xsd', FORMATS.GML.mime_type) + gml_input = self.get_input('gml/point.gml', 'point.xsd', FORMATS.GML.mime_type) self.assertTrue(validategml(gml_input, MODE.NONE), 'NONE validation') self.assertTrue(validategml(gml_input, MODE.SIMPLE), 'SIMPLE validation') self.assertTrue(validategml(gml_input, MODE.STRICT), 'STRICT validation') @@ -89,7 +82,7 @@ def test_gml_validator(self): def test_json_validator(self): """Test GeoJSON validator """ - json_input = get_input('json/point.geojson', None, FORMATS.JSON.mime_type) + json_input = self.get_input('json/point.geojson', None, FORMATS.JSON.mime_type) self.assertTrue(validatejson(json_input, MODE.NONE), 'NONE validation') self.assertTrue(validatejson(json_input, MODE.SIMPLE), 'SIMPLE validation') self.assertTrue(validatejson(json_input, MODE.STRICT), 'STRICT validation') @@ -98,7 +91,7 @@ def test_json_validator(self): def test_geojson_validator(self): """Test GeoJSON validator """ - geojson_input = get_input('json/point.geojson', 'json/schema/geojson.json', + geojson_input = self.get_input('json/point.geojson', 'json/schema/geojson.json', FORMATS.GEOJSON.mime_type) self.assertTrue(validategeojson(geojson_input, MODE.NONE), 'NONE validation') self.assertTrue(validategeojson(geojson_input, MODE.SIMPLE), 'SIMPLE validation') @@ -109,7 +102,7 @@ def test_geojson_validator(self): def test_shapefile_validator(self): """Test ESRI Shapefile validator """ - shapefile_input = get_input('shp/point.shp.zip', None, + shapefile_input = self.get_input('shp/point.shp.zip', None, FORMATS.SHP.mime_type) self.assertTrue(validateshapefile(shapefile_input, MODE.NONE), 'NONE validation') self.assertTrue(validateshapefile(shapefile_input, MODE.SIMPLE), 'SIMPLE validation') @@ -119,7 +112,7 @@ def test_shapefile_validator(self): def test_geotiff_validator(self): """Test GeoTIFF validator """ - geotiff_input = get_input('geotiff/dem.tiff', None, + geotiff_input = self.get_input('geotiff/dem.tiff', None, FORMATS.GEOTIFF.mime_type) self.assertTrue(validategeotiff(geotiff_input, MODE.NONE), 'NONE validation') self.assertTrue(validategeotiff(geotiff_input, MODE.SIMPLE), 'SIMPLE validation') @@ -129,7 +122,7 @@ def test_geotiff_validator(self): def test_netcdf_validator(self): """Test netCDF validator """ - netcdf_input = get_input('netcdf/time.nc', None, FORMATS.NETCDF.mime_type) + netcdf_input = self.get_input('netcdf/time.nc', None, FORMATS.NETCDF.mime_type) self.assertTrue(validatenetcdf(netcdf_input, MODE.NONE), 'NONE validation') self.assertTrue(validatenetcdf(netcdf_input, MODE.SIMPLE), 'SIMPLE validation') netcdf_input.stream.close() @@ -161,12 +154,14 @@ def test_dods_default(self): mode=MODE.SIMPLE) def test_fail_validator(self): - fake_input = get_input('point.xsd', 'point.xsd', FORMATS.SHP.mime_type) + fake_input = self.get_input('point.xsd', 'point.xsd', FORMATS.SHP.mime_type) self.assertFalse(validategml(fake_input, MODE.SIMPLE), 'SIMPLE validation invalid') fake_input.stream.close() def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [ diff --git a/tests/validator/test_literalvalidators.py b/tests/validator/test_literalvalidators.py index 5b23615c2..81b205380 100644 --- a/tests/validator/test_literalvalidators.py +++ b/tests/validator/test_literalvalidators.py @@ -6,7 +6,7 @@ """Unit tests for literal validator """ -import unittest +from basic import TestBase from pywps.validator.literalvalidator import * from pywps.inout.literaltypes import AllowedValue, AnyValue, ValuesReference @@ -24,15 +24,9 @@ class FakeInput(object): return fake_input -class ValidateTest(unittest.TestCase): +class ValidateTest(TestBase): """Literal validator test cases""" - def setUp(self): - pass - - def tearDown(self): - pass - def test_value_validator(self): """Test simple validator for string, integer, etc""" inpt = get_input(allowed_values=None, data='test') @@ -123,6 +117,8 @@ def test_combined_validator(self): def load_tests(loader=None, tests=None, pattern=None): + import unittest + if not loader: loader = unittest.TestLoader() suite_list = [