# ================================================================= # # Authors: Tom Kralidis # John A Stevenson # Colin Blackburn # # Copyright (c) 2023 Tom Kralidis # Copyright (c) 2022 John A Stevenson and Colin Blackburn # # Permission is hereby granted, free of charge, to any person # obtaining a copy of this software and associated documentation # files (the "Software"), to deal in the Software without # restriction, including without limitation the rights to use, # copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the # Software is furnished to do so, subject to the following # conditions: # # The above copyright notice and this permission notice shall be # included in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES # OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR # OTHER DEALINGS IN THE SOFTWARE. # # ================================================================= import copy import json import logging import time import gzip from http import HTTPStatus from unittest import mock from pyld import jsonld import pytest import pyproj from shapely.geometry import Point from pygeoapi.api import ( API, APIRequest, FORMAT_TYPES, validate_bbox, validate_datetime, validate_subset, F_HTML, F_JSON, F_JSONLD, F_GZIP, __version__ ) from pygeoapi.util import (yaml_load, get_crs_from_uri, get_api_rules, get_base_url) from .util import (get_test_file_path, mock_request, mock_flask, mock_starlette) from pygeoapi.models.provider.base import TileMatrixSetEnum LOGGER = logging.getLogger(__name__) @pytest.fixture() def config(): with open(get_test_file_path('pygeoapi-test-config.yml')) as fh: return yaml_load(fh) @pytest.fixture() def config_with_rules() -> dict: """ Returns a pygeoapi configuration with default API rules. """ with open(get_test_file_path('pygeoapi-test-config-apirules.yml')) as fh: return yaml_load(fh) @pytest.fixture() def config_enclosure() -> dict: """ Returns a pygeoapi configuration with enclosure links. """ with open(get_test_file_path('pygeoapi-test-config-enclosure.yml')) as fh: return yaml_load(fh) @pytest.fixture() def config_hidden_resources(): filename = 'pygeoapi-test-config-hidden-resources.yml' with open(get_test_file_path(filename)) as fh: return yaml_load(fh) @pytest.fixture() def openapi(): with open(get_test_file_path('pygeoapi-test-openapi.yml')) as fh: return yaml_load(fh) @pytest.fixture() def api_(config, openapi): return API(config, openapi) @pytest.fixture() def enclosure_api(config_enclosure, openapi): """ Returns an API instance with a collection with enclosure links. """ return API(config_enclosure, openapi) @pytest.fixture() def rules_api(config_with_rules, openapi): """ Returns an API instance with URL prefix and strict slashes policy. The API version is extracted from the current version here. """ return API(config_with_rules, openapi) @pytest.fixture() def api_hidden_resources(config_hidden_resources, openapi): return API(config_hidden_resources, openapi) def test_apirequest(api_): # Test without (valid) locales with pytest.raises(ValueError): req = mock_request() APIRequest(req, []) APIRequest(req, None) APIRequest(req, ['zz']) # Test all supported formats from query args for f, mt in FORMAT_TYPES.items(): req = mock_request({'f': f}) apireq = APIRequest(req, api_.locales) assert apireq.is_valid() assert apireq.format == f assert apireq.get_response_headers()['Content-Type'] == mt # Test all supported formats from Accept header for f, mt in FORMAT_TYPES.items(): req = mock_request(HTTP_ACCEPT=mt) apireq = APIRequest(req, api_.locales) assert apireq.is_valid() assert apireq.format == f assert apireq.get_response_headers()['Content-Type'] == mt # Test nonsense format req = mock_request({'f': 'foo'}) apireq = APIRequest(req, api_.locales) assert not apireq.is_valid() assert apireq.format == 'foo' assert apireq.is_valid(('foo',)) assert apireq.get_response_headers()['Content-Type'] == \ FORMAT_TYPES[F_JSON] # Test without format req = mock_request() apireq = APIRequest(req, api_.locales) assert apireq.is_valid() assert apireq.format is None assert apireq.get_response_headers()['Content-Type'] == \ FORMAT_TYPES[F_JSON] assert apireq.get_linkrel(F_JSON) == 'self' assert apireq.get_linkrel(F_HTML) == 'alternate' # Test complex format string hh = 'text/html,application/xhtml+xml,application/xml;q=0.9,' req = mock_request(HTTP_ACCEPT=hh) apireq = APIRequest(req, api_.locales) assert apireq.is_valid() assert apireq.format == F_HTML assert apireq.get_response_headers()['Content-Type'] == \ FORMAT_TYPES[F_HTML] assert apireq.get_linkrel(F_HTML) == 'self' assert apireq.get_linkrel(F_JSON) == 'alternate' # Test accept header with multiple valid formats hh = 'plain/text,application/ld+json,application/json;q=0.9,' req = mock_request(HTTP_ACCEPT=hh) apireq = APIRequest(req, api_.locales) assert apireq.is_valid() assert apireq.format == F_JSONLD assert apireq.get_response_headers()['Content-Type'] == \ FORMAT_TYPES[F_JSONLD] assert apireq.get_linkrel(F_JSONLD) == 'self' assert apireq.get_linkrel(F_HTML) == 'alternate' # Overrule HTTP content negotiation req = mock_request({'f': 'html'}, HTTP_ACCEPT='application/json') # noqa apireq = APIRequest(req, api_.locales) assert apireq.is_valid() assert apireq.format == F_HTML assert apireq.get_response_headers()['Content-Type'] == \ FORMAT_TYPES[F_HTML] # Test data for d in (None, '', 'test', {'key': 'value'}): req = mock_request(data=d) apireq = APIRequest.with_data(req, api_.locales) if not d: assert apireq.data == b'' elif isinstance(d, dict): assert d == json.loads(apireq.data) else: assert apireq.data == d.encode() # Test multilingual test_lang = { 'nl': ('en', 'en-US'), # unsupported lang should return default 'en-US': ('en', 'en-US'), 'de_CH': ('en', 'en-US'), 'fr-CH, fr;q=0.9, en;q=0.8': ('fr', 'fr-CA'), 'fr-CH, fr-BE;q=0.9': ('fr', 'fr-CA'), } sup_lang = ('en-US', 'fr_CA') for lang_in, (lang_out, cl_out) in test_lang.items(): # Using l query parameter req = mock_request({'lang': lang_in}) apireq = APIRequest(req, sup_lang) assert apireq.raw_locale == lang_in assert apireq.locale.language == lang_out assert apireq.get_response_headers()['Content-Language'] == cl_out # Using Accept-Language header req = mock_request(HTTP_ACCEPT_LANGUAGE=lang_in) apireq = APIRequest(req, sup_lang) assert apireq.raw_locale == lang_in assert apireq.locale.language == lang_out assert apireq.get_response_headers()['Content-Language'] == cl_out # Test language override req = mock_request({'lang': 'fr'}, HTTP_ACCEPT_LANGUAGE='en_US') apireq = APIRequest(req, sup_lang) assert apireq.raw_locale == 'fr' assert apireq.locale.language == 'fr' assert apireq.get_response_headers()['Content-Language'] == 'fr-CA' # Test locale territory req = mock_request({'lang': 'en-GB'}) apireq = APIRequest(req, sup_lang) assert apireq.raw_locale == 'en-GB' assert apireq.locale.language == 'en' assert apireq.locale.territory == 'US' assert apireq.get_response_headers()['Content-Language'] == 'en-US' # Test without Accept-Language header or 'lang' query parameter # (should return default language from YAML config) req = mock_request() apireq = APIRequest(req, api_.locales) assert apireq.raw_locale is None assert apireq.locale.language == api_.default_locale.language assert apireq.get_response_headers()['Content-Language'] == 'en-US' # Test without Accept-Language header or 'lang' query param # (should return first in custom list of languages) sup_lang = ('de', 'fr', 'en') apireq = APIRequest(req, sup_lang) assert apireq.raw_locale is None assert apireq.locale.language == 'de' assert apireq.get_response_headers()['Content-Language'] == 'de' def test_apirules_active(config_with_rules, rules_api): assert rules_api.config == config_with_rules rules = get_api_rules(config_with_rules) base_url = get_base_url(config_with_rules) # Test Flask flask_prefix = rules.get_url_prefix('flask') with mock_flask('pygeoapi-test-config-apirules.yml') as flask_client: # Test happy path response = flask_client.get(f'{flask_prefix}/conformance') assert response.status_code == 200 assert response.headers['X-API-Version'] == __version__ assert response.request.url == \ flask_client.application.url_for('pygeoapi.conformance') response = flask_client.get(f'{flask_prefix}/static/img/pygeoapi.png') assert response.status_code == 200 # Test that static resources also work without URL prefix response = flask_client.get('/static/img/pygeoapi.png') assert response.status_code == 200 # Test strict slashes response = flask_client.get(f'{flask_prefix}/conformance/') assert response.status_code == 404 # For the landing page ONLY, trailing slashes are actually preferred. # See https://docs.opengeospatial.org/is/17-069r4/17-069r4.html#_api_landing_page # noqa # Omitting the trailing slash should lead to a redirect. response = flask_client.get(f'{flask_prefix}/') assert response.status_code == 200 response = flask_client.get(flask_prefix) assert response.status_code in (307, 308) # Test links on landing page for correct URLs response = flask_client.get(flask_prefix, follow_redirects=True) assert response.status_code == 200 assert response.is_json links = response.json['links'] assert all( href.startswith(base_url) for href in (rel['href'] for rel in links) # noqa ) # Test Starlette starlette_prefix = rules.get_url_prefix('starlette') with mock_starlette('pygeoapi-test-config-apirules.yml') as starlette_client: # noqa # Test happy path response = starlette_client.get(f'{starlette_prefix}/conformance') assert response.status_code == 200 assert response.headers['X-API-Version'] == __version__ response = starlette_client.get(f'{starlette_prefix}/static/img/pygeoapi.png') # noqa assert response.status_code == 200 # Test that static resources also work without URL prefix response = starlette_client.get('/static/img/pygeoapi.png') assert response.status_code == 200 # Test strict slashes response = starlette_client.get(f'{starlette_prefix}/conformance/') assert response.status_code == 404 # For the landing page ONLY, trailing slashes are actually preferred. # See https://docs.opengeospatial.org/is/17-069r4/17-069r4.html#_api_landing_page # noqa # Omitting the trailing slash should lead to a redirect. response = starlette_client.get(f'{starlette_prefix}/') assert response.status_code == 200 response = starlette_client.get(starlette_prefix) assert response.status_code in (307, 308) # Test links on landing page for correct URLs response = starlette_client.get(starlette_prefix, follow_redirects=True) # noqa assert response.status_code == 200 links = response.json()['links'] assert all( href.startswith(base_url) for href in (rel['href'] for rel in links) # noqa ) def test_apirules_inactive(config, api_): assert api_.config == config rules = get_api_rules(config) # Test Flask flask_prefix = rules.get_url_prefix('flask') assert flask_prefix == '' with mock_flask('pygeoapi-test-config.yml') as flask_client: response = flask_client.get('') assert response.status_code == 200 response = flask_client.get('/conformance') assert response.status_code == 200 assert 'X-API-Version' not in response.headers assert response.request.url == \ flask_client.application.url_for('pygeoapi.conformance') response = flask_client.get('/static/img/pygeoapi.png') assert response.status_code == 200 # Test trailing slashes response = flask_client.get('/') assert response.status_code == 200 response = flask_client.get('/conformance/') assert response.status_code == 200 assert 'X-API-Version' not in response.headers # Test Starlette starlette_prefix = rules.get_url_prefix('starlette') assert starlette_prefix == '' with mock_starlette('pygeoapi-test-config.yml') as starlette_client: response = starlette_client.get('') assert response.status_code == 200 response = starlette_client.get('/conformance') assert response.status_code == 200 assert 'X-API-Version' not in response.headers assert str(response.url) == f"{starlette_client.base_url}/conformance" response = starlette_client.get('/static/img/pygeoapi.png') assert response.status_code == 200 # Test trailing slashes response = starlette_client.get('/') assert response.status_code == 200 response = starlette_client.get('/conformance/', follow_redirects=True) assert response.status_code == 200 assert 'X-API-Version' not in response.headers def test_api(config, api_, openapi): assert api_.config == config assert isinstance(api_.config, dict) req = mock_request(HTTP_ACCEPT='application/json') rsp_headers, code, response = api_.openapi_(req) assert rsp_headers['Content-Type'] == 'application/vnd.oai.openapi+json;version=3.0' # noqa # No language requested: should be set to default from YAML assert rsp_headers['Content-Language'] == 'en-US' root = json.loads(response) assert isinstance(root, dict) a = 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8' req = mock_request(HTTP_ACCEPT=a) rsp_headers, code, response = api_.openapi_(req) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] == \ FORMAT_TYPES[F_HTML] assert 'Swagger UI' in response a = 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8' req = mock_request({'ui': 'redoc'}, HTTP_ACCEPT=a) rsp_headers, code, response = api_.openapi_(req) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] == \ FORMAT_TYPES[F_HTML] assert 'ReDoc' in response req = mock_request({'f': 'foo'}) rsp_headers, code, response = api_.openapi_(req) assert rsp_headers['Content-Language'] == 'en-US' assert code == HTTPStatus.BAD_REQUEST assert api_.get_collections_url() == 'http://localhost:5000/collections' def test_api_exception(config, api_): req = mock_request({'f': 'foo'}) rsp_headers, code, response = api_.landing_page(req) assert rsp_headers['Content-Language'] == 'en-US' assert code == HTTPStatus.BAD_REQUEST # When a language is set, the exception should still be English req = mock_request({'f': 'foo', 'lang': 'fr'}) rsp_headers, code, response = api_.landing_page(req) assert rsp_headers['Content-Language'] == 'en-US' assert code == HTTPStatus.BAD_REQUEST def test_gzip(config, api_): # Requests for each response type and gzip encoding req_gzip_json = mock_request(HTTP_ACCEPT=FORMAT_TYPES[F_JSON], HTTP_ACCEPT_ENCODING=F_GZIP) req_gzip_jsonld = mock_request(HTTP_ACCEPT=FORMAT_TYPES[F_JSONLD], HTTP_ACCEPT_ENCODING=F_GZIP) req_gzip_html = mock_request(HTTP_ACCEPT=FORMAT_TYPES[F_HTML], HTTP_ACCEPT_ENCODING=F_GZIP) req_gzip_gzip = mock_request(HTTP_ACCEPT='application/gzip', HTTP_ACCEPT_ENCODING=F_GZIP) # Responses from server config without gzip compression rsp_headers, _, rsp_json = api_.landing_page(req_gzip_json) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] rsp_headers, _, rsp_jsonld = api_.landing_page(req_gzip_jsonld) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSONLD] rsp_headers, _, rsp_html = api_.landing_page(req_gzip_html) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] rsp_headers, _, _ = api_.landing_page(req_gzip_gzip) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] # Add gzip to server and use utf-16 encoding config['server']['gzip'] = True enc_16 = 'utf-16' config['server']['encoding'] = enc_16 api_ = API(config, openapi) # Responses from server with gzip compression rsp_json_headers, _, rsp_gzip_json = api_.landing_page(req_gzip_json) rsp_jsonld_headers, _, rsp_gzip_jsonld = api_.landing_page(req_gzip_jsonld) rsp_html_headers, _, rsp_gzip_html = api_.landing_page(req_gzip_html) rsp_gzip_headers, _, rsp_gzip_gzip = api_.landing_page(req_gzip_gzip) # Validate compressed json response assert rsp_json_headers['Content-Type'] == \ f'{FORMAT_TYPES[F_JSON]}; charset={enc_16}' assert rsp_json_headers['Content-Encoding'] == F_GZIP parsed_gzip_json = gzip.decompress(rsp_gzip_json).decode(enc_16) assert isinstance(parsed_gzip_json, str) parsed_gzip_json = json.loads(parsed_gzip_json) assert isinstance(parsed_gzip_json, dict) assert parsed_gzip_json == json.loads(rsp_json) # Validate compressed jsonld response assert rsp_jsonld_headers['Content-Type'] == \ f'{FORMAT_TYPES[F_JSONLD]}; charset={enc_16}' assert rsp_jsonld_headers['Content-Encoding'] == F_GZIP parsed_gzip_jsonld = gzip.decompress(rsp_gzip_jsonld).decode(enc_16) assert isinstance(parsed_gzip_jsonld, str) parsed_gzip_jsonld = json.loads(parsed_gzip_jsonld) assert isinstance(parsed_gzip_jsonld, dict) assert parsed_gzip_jsonld == json.loads(rsp_jsonld) # Validate compressed html response assert rsp_html_headers['Content-Type'] == \ f'{FORMAT_TYPES[F_HTML]}; charset={enc_16}' assert rsp_html_headers['Content-Encoding'] == F_GZIP parsed_gzip_html = gzip.decompress(rsp_gzip_html).decode(enc_16) assert isinstance(parsed_gzip_html, str) assert parsed_gzip_html == rsp_html # Validate compressed gzip response assert rsp_gzip_headers['Content-Type'] == \ f'{FORMAT_TYPES[F_GZIP]}; charset={enc_16}' assert rsp_gzip_headers['Content-Encoding'] == F_GZIP parsed_gzip_gzip = gzip.decompress(rsp_gzip_gzip).decode(enc_16) assert isinstance(parsed_gzip_gzip, str) parsed_gzip_gzip = json.loads(parsed_gzip_gzip) assert isinstance(parsed_gzip_gzip, dict) # Requests without content encoding header req_json = mock_request(HTTP_ACCEPT=FORMAT_TYPES[F_JSON]) req_jsonld = mock_request(HTTP_ACCEPT=FORMAT_TYPES[F_JSONLD]) req_html = mock_request(HTTP_ACCEPT=FORMAT_TYPES[F_HTML]) # Responses without content encoding _, _, rsp_json_ = api_.landing_page(req_json) _, _, rsp_jsonld_ = api_.landing_page(req_jsonld) _, _, rsp_html_ = api_.landing_page(req_html) # Confirm each request is the same when decompressed assert rsp_json_ == rsp_json == \ gzip.decompress(rsp_gzip_json).decode(enc_16) assert rsp_jsonld_ == rsp_jsonld == \ gzip.decompress(rsp_gzip_jsonld).decode(enc_16) assert rsp_html_ == rsp_html == \ gzip.decompress(rsp_gzip_html).decode(enc_16) def test_gzip_csv(config, api_): req_csv = mock_request({'f': 'csv'}) rsp_csv_headers, _, rsp_csv = api_.get_collection_items(req_csv, 'obs') assert rsp_csv_headers['Content-Type'] == 'text/csv; charset=utf-8' rsp_csv = rsp_csv.decode('utf-8') req_csv = mock_request({'f': 'csv'}, HTTP_ACCEPT_ENCODING=F_GZIP) rsp_csv_headers, _, rsp_csv_gzip = api_.get_collection_items(req_csv, 'obs') # noqa assert rsp_csv_headers['Content-Type'] == 'text/csv; charset=utf-8' rsp_csv_ = gzip.decompress(rsp_csv_gzip).decode('utf-8') assert rsp_csv == rsp_csv_ # Use utf-16 encoding config['server']['encoding'] = 'utf-16' api_ = API(config, openapi) req_csv = mock_request({'f': 'csv'}, HTTP_ACCEPT_ENCODING=F_GZIP) rsp_csv_headers, _, rsp_csv_gzip = api_.get_collection_items(req_csv, 'obs') # noqa assert rsp_csv_headers['Content-Type'] == 'text/csv; charset=utf-8' rsp_csv_ = gzip.decompress(rsp_csv_gzip).decode('utf-8') assert rsp_csv == rsp_csv_ def test_root(config, api_): req = mock_request() rsp_headers, code, response = api_.landing_page(req) root = json.loads(response) assert rsp_headers['Content-Type'] == 'application/json' == \ FORMAT_TYPES[F_JSON] assert rsp_headers['X-Powered-By'].startswith('pygeoapi') assert rsp_headers['Content-Language'] == 'en-US' assert isinstance(root, dict) assert 'links' in root assert root['links'][0]['rel'] == 'self' assert root['links'][0]['type'] == FORMAT_TYPES[F_JSON] assert root['links'][0]['href'].endswith('?f=json') assert any(link['href'].endswith('f=jsonld') and link['rel'] == 'alternate' for link in root['links']) assert any(link['href'].endswith('f=html') and link['rel'] == 'alternate' for link in root['links']) assert len(root['links']) == 11 assert 'title' in root assert root['title'] == 'pygeoapi default instance' assert 'description' in root assert root['description'] == 'pygeoapi provides an API to geospatial data' req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.landing_page(req) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] assert rsp_headers['Content-Language'] == 'en-US' def test_root_structured_data(config, api_): req = mock_request({"f": "jsonld"}) rsp_headers, code, response = api_.landing_page(req) root = json.loads(response) assert rsp_headers['Content-Type'] == 'application/ld+json' == \ FORMAT_TYPES[F_JSONLD] assert rsp_headers['Content-Language'] == 'en-US' assert rsp_headers['X-Powered-By'].startswith('pygeoapi') assert isinstance(root, dict) assert 'description' in root assert root['description'] == 'pygeoapi provides an API to geospatial data' assert '@context' in root assert root['@context'] == 'https://schema.org/docs/jsonldcontext.jsonld' expanded = jsonld.expand(root)[0] assert '@type' in expanded assert 'http://schema.org/DataCatalog' in expanded['@type'] assert 'http://schema.org/description' in expanded assert root['description'] == expanded['http://schema.org/description'][0][ '@value'] assert 'http://schema.org/keywords' in expanded assert len(expanded['http://schema.org/keywords']) == 3 assert '@value' in expanded['http://schema.org/keywords'][0].keys() assert 'http://schema.org/provider' in expanded assert expanded['http://schema.org/provider'][0]['@type'][ 0] == 'http://schema.org/Organization' assert expanded['http://schema.org/name'][0]['@value'] == root['name'] def test_conformance(config, api_): req = mock_request() rsp_headers, code, response = api_.conformance(req) root = json.loads(response) assert isinstance(root, dict) assert 'conformsTo' in root assert len(root['conformsTo']) == 36 assert 'http://www.opengis.net/spec/ogcapi-features-2/1.0/conf/crs' \ in root['conformsTo'] req = mock_request({'f': 'foo'}) rsp_headers, code, response = api_.conformance(req) assert code == HTTPStatus.BAD_REQUEST req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.conformance(req) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] # No language requested: should be set to default from YAML assert rsp_headers['Content-Language'] == 'en-US' def test_tilematrixsets(config, api_): req = mock_request() rsp_headers, code, response = api_.tilematrixsets(req) root = json.loads(response) assert isinstance(root, dict) assert 'tileMatrixSets' in root assert len(root['tileMatrixSets']) == 2 assert 'http://www.opengis.net/def/tilematrixset/OGC/1.0/WorldCRS84Quad' \ in root['tileMatrixSets'][0]['uri'] assert 'http://www.opengis.net/def/tilematrixset/OGC/1.0/WebMercatorQuad' \ in root['tileMatrixSets'][1]['uri'] req = mock_request({'f': 'foo'}) rsp_headers, code, response = api_.tilematrixsets(req) assert code == HTTPStatus.BAD_REQUEST req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.tilematrixsets(req) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] # No language requested: should be set to default from YAML assert rsp_headers['Content-Language'] == 'en-US' def test_tilematrixset(config, api_): req = mock_request() enums = [e.value for e in TileMatrixSetEnum] enum = None for e in enums: enum = e.tileMatrixSet rsp_headers, code, response = api_.tilematrixset(req, enum) root = json.loads(response) assert isinstance(root, dict) assert 'id' in root assert root['id'] == enum assert 'tileMatrices' in root assert len(root['tileMatrices']) == 30 rsp_headers, code, response = api_.tilematrixset(req, 'foo') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.tilematrixset(req, enum) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] # No language requested: should be set to default from YAML assert rsp_headers['Content-Language'] == 'en-US' def test_describe_collections(config, api_): req = mock_request({"f": "foo"}) rsp_headers, code, response = api_.describe_collections(req) assert code == HTTPStatus.BAD_REQUEST req = mock_request({"f": "html"}) rsp_headers, code, response = api_.describe_collections(req) assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] req = mock_request() rsp_headers, code, response = api_.describe_collections(req) collections = json.loads(response) assert len(collections) == 2 assert len(collections['collections']) == 9 assert len(collections['links']) == 3 rsp_headers, code, response = api_.describe_collections(req, 'foo') collection = json.loads(response) assert code == HTTPStatus.NOT_FOUND rsp_headers, code, response = api_.describe_collections(req, 'obs') collection = json.loads(response) assert rsp_headers['Content-Language'] == 'en-US' assert collection['id'] == 'obs' assert collection['title'] == 'Observations' assert collection['description'] == 'My cool observations' assert len(collection['links']) == 14 assert collection['extent'] == { 'spatial': { 'bbox': [[-180, -90, 180, 90]], 'crs': 'http://www.opengis.net/def/crs/OGC/1.3/CRS84' }, 'temporal': { 'interval': [ ['2000-10-30T18:24:39+00:00', '2007-10-30T08:57:29+00:00'] ], 'trs': 'http://www.opengis.net/def/uom/ISO-8601/0/Gregorian' } } # OAPIF Part 2 CRS 6.2.1 A, B, configured CRS + defaults assert collection['crs'] is not None crs_set = [ 'http://www.opengis.net/def/crs/EPSG/0/28992', 'http://www.opengis.net/def/crs/OGC/1.3/CRS84', 'http://www.opengis.net/def/crs/EPSG/0/4326', ] for crs in crs_set: assert crs in collection['crs'] assert collection['storageCRS'] is not None assert collection['storageCRS'] == 'http://www.opengis.net/def/crs/OGC/1.3/CRS84' # noqa assert 'storageCrsCoordinateEpoch' not in collection # French language request req = mock_request({'lang': 'fr'}) rsp_headers, code, response = api_.describe_collections(req, 'obs') collection = json.loads(response) assert rsp_headers['Content-Language'] == 'fr-CA' assert collection['title'] == 'Observations' assert collection['description'] == 'Mes belles observations' # Check HTML request in an unsupported language req = mock_request({'f': 'html', 'lang': 'de'}) rsp_headers, code, response = api_.describe_collections(req, 'obs') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] assert rsp_headers['Content-Language'] == 'en-US' req = mock_request() rsp_headers, code, response = api_.describe_collections(req, 'gdps-temperature') collection = json.loads(response) assert collection['id'] == 'gdps-temperature' assert len(collection['links']) == 10 assert collection['extent']['spatial']['grid'][0]['cellsCount'] == 2400 assert collection['extent']['spatial']['grid'][0]['resolution'] == 0.15000000000000002 # noqa assert collection['extent']['spatial']['grid'][1]['cellsCount'] == 1201 assert collection['extent']['spatial']['grid'][1]['resolution'] == 0.15 # hiearchical collections rsp_headers, code, response = api_.describe_collections( req, 'naturalearth/lakes') collection = json.loads(response) assert collection['id'] == 'naturalearth/lakes' # OAPIF Part 2 CRS 6.2.1 B, defaults when not configured assert collection['crs'] is not None default_crs_list = [ 'http://www.opengis.net/def/crs/OGC/1.3/CRS84', 'http://www.opengis.net/def/crs/OGC/1.3/CRS84h', ] contains_default = False for crs in default_crs_list: if crs in default_crs_list: contains_default = True assert contains_default assert collection['storageCRS'] is not None assert collection['storageCRS'] == 'http://www.opengis.net/def/crs/OGC/1.3/CRS84' # noqa assert collection['storageCrsCoordinateEpoch'] == 2017.23 def test_describe_collections_hidden_resources( config_hidden_resources, api_hidden_resources): req = mock_request({}) rsp_headers, code, response = api_hidden_resources.describe_collections(req) # noqa assert code == HTTPStatus.OK assert len(config_hidden_resources['resources']) == 3 collections = json.loads(response) assert len(collections['collections']) == 1 def test_get_collection_schema(config, api_): req = mock_request() rsp_headers, code, response = api_.get_collection_schema(req, 'notfound') assert code == HTTPStatus.NOT_FOUND req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.get_collection_schema(req, 'obs') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] req = mock_request({'f': 'json'}) rsp_headers, code, response = api_.get_collection_schema(req, 'obs') assert rsp_headers['Content-Type'] == 'application/schema+json' schema = json.loads(response) assert 'properties' in schema assert len(schema['properties']) == 5 req = mock_request({'f': 'json'}) rsp_headers, code, response = api_.get_collection_schema( req, 'gdps-temperature') assert rsp_headers['Content-Type'] == 'application/schema+json' schema = json.loads(response) assert 'properties' in schema assert len(schema['properties']) == 1 assert schema['properties']['1']['type'] == 'number' def test_get_collection_queryables(config, api_): req = mock_request() rsp_headers, code, response = api_.get_collection_queryables(req, 'notfound') assert code == HTTPStatus.NOT_FOUND req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.get_collection_queryables(req, 'obs') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] req = mock_request({'f': 'json'}) rsp_headers, code, response = api_.get_collection_queryables(req, 'obs') assert rsp_headers['Content-Type'] == 'application/schema+json' queryables = json.loads(response) assert 'properties' in queryables assert len(queryables['properties']) == 5 # test with provider filtered properties api_.config['resources']['obs']['providers'][0]['properties'] = ['stn_id'] rsp_headers, code, response = api_.get_collection_queryables(req, 'obs') queryables = json.loads(response) assert 'properties' in queryables assert len(queryables['properties']) == 2 assert 'geometry' in queryables['properties'] assert queryables['properties']['geometry']['$ref'] == 'https://geojson.org/schema/Geometry.json' # noqa # No language requested: should be set to default from YAML assert rsp_headers['Content-Language'] == 'en-US' def test_describe_collections_json_ld(config, api_): req = mock_request({'f': 'jsonld'}) rsp_headers, code, response = api_.describe_collections(req, 'obs') collection = json.loads(response) assert '@context' in collection expanded = jsonld.expand(collection)[0] # Metadata is about a schema:DataCollection that contains a schema:Dataset assert not expanded['@id'].endswith('obs') assert 'http://schema.org/dataset' in expanded assert len(expanded['http://schema.org/dataset']) == 1 dataset = expanded['http://schema.org/dataset'][0] assert dataset['@type'][0] == 'http://schema.org/Dataset' assert len(dataset['http://schema.org/distribution']) == 14 assert all(dist['@type'][0] == 'http://schema.org/DataDownload' for dist in dataset['http://schema.org/distribution']) assert 'http://schema.org/Organization' in expanded[ 'http://schema.org/provider'][0]['@type'] assert 'http://schema.org/Place' in dataset[ 'http://schema.org/spatial'][0]['@type'] assert 'http://schema.org/GeoShape' in dataset[ 'http://schema.org/spatial'][0]['http://schema.org/geo'][0]['@type'] assert dataset['http://schema.org/spatial'][0]['http://schema.org/geo'][ 0]['http://schema.org/box'][0]['@value'] == '-180,-90 180,90' assert 'http://schema.org/temporalCoverage' in dataset assert dataset['http://schema.org/temporalCoverage'][0][ '@value'] == '2000-10-30T18:24:39+00:00/2007-10-30T08:57:29+00:00' # No language requested: should be set to default from YAML assert rsp_headers['Content-Language'] == 'en-US' def test_get_collection_items(config, api_): req = mock_request() rsp_headers, code, response = api_.get_collection_items(req, 'foo') features = json.loads(response) assert code == HTTPStatus.NOT_FOUND req = mock_request({'f': 'foo'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert code == HTTPStatus.BAD_REQUEST req = mock_request({'bbox': '1,2,3'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert code == HTTPStatus.BAD_REQUEST req = mock_request({'bbox': '1,2,3,4c'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'bbox': '1,2,3,4', 'bbox-crs': 'bad_value'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'bbox-crs': 'bad_value'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST # bbox-crs must be in configured values for Collection req = mock_request({'bbox': '1,2,3,4', 'bbox-crs': 'http://www.opengis.net/def/crs/EPSG/0/4258'}) # noqa rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST # bbox-crs must be in configured values for Collection (CSV will ignore) req = mock_request({'bbox': '52,4,53,5', 'bbox-crs': 'http://www.opengis.net/def/crs/EPSG/0/4326'}) # noqa rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK # bbox-crs can be a default even if not configured req = mock_request({'bbox': '4,52,5,53', 'bbox-crs': 'http://www.opengis.net/def/crs/OGC/1.3/CRS84'}) # noqa rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK # bbox-crs can be a default even if not configured req = mock_request({'bbox': '4,52,5,53'}) # noqa rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK req = mock_request({'f': 'html', 'lang': 'fr'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] assert rsp_headers['Content-Language'] == 'fr-CA' req = mock_request() rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) # No language requested: should be set to default from YAML assert rsp_headers['Content-Language'] == 'en-US' assert len(features['features']) == 5 req = mock_request({'resulttype': 'hits'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert len(features['features']) == 0 # Invalid limit req = mock_request({'limit': 0}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert code == HTTPStatus.BAD_REQUEST req = mock_request({'stn_id': '35'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert len(features['features']) == 2 assert features['numberMatched'] == 2 req = mock_request({'stn_id': '35', 'value': '93.9'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert len(features['features']) == 1 assert features['numberMatched'] == 1 req = mock_request({'limit': 2}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert len(features['features']) == 2 assert features['features'][1]['properties']['stn_id'] == 35 links = features['links'] assert len(links) == 4 assert '/collections/obs/items?f=json' in links[0]['href'] assert links[0]['rel'] == 'self' assert '/collections/obs/items?f=jsonld' in links[1]['href'] assert links[1]['rel'] == 'alternate' assert '/collections/obs/items?f=html' in links[2]['href'] assert links[2]['rel'] == 'alternate' assert '/collections/obs' in links[3]['href'] assert links[3]['rel'] == 'collection' # Invalid offset req = mock_request({'offset': -1}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert code == HTTPStatus.BAD_REQUEST req = mock_request({'offset': 2}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert len(features['features']) == 3 assert features['features'][1]['properties']['stn_id'] == 2147 links = features['links'] assert len(links) == 5 assert '/collections/obs/items?f=json' in links[0]['href'] assert links[0]['rel'] == 'self' assert '/collections/obs/items?f=jsonld' in links[1]['href'] assert links[1]['rel'] == 'alternate' assert '/collections/obs/items?f=html' in links[2]['href'] assert links[2]['rel'] == 'alternate' assert '/collections/obs/items?offset=0' in links[3]['href'] assert links[3]['rel'] == 'prev' assert '/collections/obs' in links[4]['href'] assert links[4]['rel'] == 'collection' req = mock_request({ 'offset': 1, 'limit': 1, 'bbox': '-180,90,180,90' }) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert len(features['features']) == 1 links = features['links'] assert len(links) == 5 assert '/collections/obs/items?f=json&limit=1&bbox=-180,90,180,90' in \ links[0]['href'] assert links[0]['rel'] == 'self' assert '/collections/obs/items?f=jsonld&limit=1&bbox=-180,90,180,90' in \ links[1]['href'] assert links[1]['rel'] == 'alternate' assert '/collections/obs/items?f=html&limit=1&bbox=-180,90,180,90' in \ links[2]['href'] assert links[2]['rel'] == 'alternate' assert '/collections/obs/items?offset=0&limit=1&bbox=-180,90,180,90' \ in links[3]['href'] assert links[3]['rel'] == 'prev' assert '/collections/obs' in links[4]['href'] assert links[4]['rel'] == 'collection' req = mock_request({ 'sortby': 'bad-property', 'stn_id': '35' }) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'sortby': 'stn_id'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert code == HTTPStatus.OK req = mock_request({'sortby': '+stn_id'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert code == HTTPStatus.OK req = mock_request({'sortby': '-stn_id'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') features = json.loads(response) assert code == HTTPStatus.OK req = mock_request({'f': 'csv'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert rsp_headers['Content-Type'] == 'text/csv; charset=utf-8' req = mock_request({'datetime': '2003'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK req = mock_request({'datetime': '1999'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'datetime': '2010-04-22'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'datetime': '2001-11-11/2003-12-18'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK req = mock_request({'datetime': '../2003-12-18'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK req = mock_request({'datetime': '2001-11-11/..'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK req = mock_request({'datetime': '1999/2005-04-22'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK req = mock_request({'datetime': '1999/2000-04-22'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST api_.config['resources']['obs']['extents'].pop('temporal') req = mock_request({'datetime': '2002/2014-04-22'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.OK req = mock_request({'scalerank': 1}) rsp_headers, code, response = api_.get_collection_items( req, 'naturalearth/lakes') features = json.loads(response) assert len(features['features']) == 10 assert features['numberMatched'] == 11 assert features['numberReturned'] == 10 req = mock_request({'datetime': '2005-04-22'}) rsp_headers, code, response = api_.get_collection_items( req, 'naturalearth/lakes') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'skipGeometry': 'true'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert json.loads(response)['features'][0]['geometry'] is None req = mock_request({'properties': 'foo,bar'}) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert code == HTTPStatus.BAD_REQUEST def test_get_collection_items_crs(config, api_): # Invalid CRS query parameter req = mock_request({'crs': '4326'}) rsp_headers, code, response = api_.get_collection_items(req, 'norway_pop') assert code == HTTPStatus.BAD_REQUEST # Unsupported CRS req = mock_request({'crs': 'http://www.opengis.net/def/crs/EPSG/0/32633'}) rsp_headers, code, response = api_.get_collection_items(req, 'norway_pop') assert code == HTTPStatus.BAD_REQUEST # Supported CRSs default_crs = 'http://www.opengis.net/def/crs/OGC/1.3/CRS84' storage_crs = 'http://www.opengis.net/def/crs/EPSG/0/25833' crs_4258 = 'http://www.opengis.net/def/crs/EPSG/0/4258' supported_crs_list = [default_crs, storage_crs, crs_4258] for crs in supported_crs_list: req = mock_request({'crs': crs}) rsp_headers, code, response = api_.get_collection_items( req, 'norway_pop', ) assert code == HTTPStatus.OK assert rsp_headers['Content-Crs'] == f'<{crs}>' # With CRS query parameter, using storageCRS req = mock_request({'crs': storage_crs}) rsp_headers, code, response = api_.get_collection_items(req, 'norway_pop') assert code == HTTPStatus.OK assert rsp_headers['Content-Crs'] == f'<{storage_crs}>' features_25833 = json.loads(response) # With CRS query parameter resulting in coordinates transformation req = mock_request({'crs': crs_4258}) rsp_headers, code, response = api_.get_collection_items(req, 'norway_pop') assert code == HTTPStatus.OK assert rsp_headers['Content-Crs'] == f'<{crs_4258}>' features_4258 = json.loads(response) transform_func = pyproj.Transformer.from_crs( pyproj.CRS.from_epsg(25833), pyproj.CRS.from_epsg(4258), always_xy=False, ).transform for feat_orig in features_25833['features']: id_ = feat_orig['id'] x, y, *_ = feat_orig['geometry']['coordinates'] loc_transf = Point(transform_func(x, y)) for feat_out in features_4258['features']: if id_ == feat_out['id']: loc_out = Point(feat_out['geometry']['coordinates'][:2]) assert loc_out.equals_exact(loc_transf, 1e-5) break # Without CRS query parameter: assume Transform to default WGS84 lon,lat req = mock_request({}) rsp_headers, code, response = api_.get_collection_items(req, 'norway_pop') assert code == HTTPStatus.OK assert rsp_headers['Content-Crs'] == f'<{default_crs}>' features_wgs84 = json.loads(response) # With CRS query parameter resulting in coordinates transformation transform_func = pyproj.Transformer.from_crs( pyproj.CRS.from_epsg(4258), get_crs_from_uri(default_crs), always_xy=False, ).transform for feat_orig in features_4258['features']: id_ = feat_orig['id'] x, y, *_ = feat_orig['geometry']['coordinates'] loc_transf = Point(transform_func(x, y)) for feat_out in features_wgs84['features']: if id_ == feat_out['id']: loc_out = Point(feat_out['geometry']['coordinates'][:2]) assert loc_out.equals_exact(loc_transf, 1e-5) break def test_manage_collection_item_read_only_options_req(config, api_): """Test OPTIONS request on a read-only items endpoint""" req = mock_request() _, code, _ = api_.manage_collection_item(req, 'options', 'foo') assert code == HTTPStatus.NOT_FOUND req = mock_request() rsp_headers, code, _ = api_.manage_collection_item(req, 'options', 'obs') assert code == HTTPStatus.OK assert rsp_headers['Allow'] == 'HEAD, GET' req = mock_request() rsp_headers, code, _ = api_.manage_collection_item( req, 'options', 'obs', 'ressource_id') assert code == HTTPStatus.OK assert rsp_headers['Allow'] == 'HEAD, GET' def test_manage_collection_item_editable_options_req(config): """Test OPTIONS request on a editable items endpoint""" config = copy.deepcopy(config) config['resources']['obs']['providers'][0]['editable'] = True api_ = API(config, openapi) req = mock_request() rsp_headers, code, _ = api_.manage_collection_item(req, 'options', 'obs') assert code == HTTPStatus.OK assert rsp_headers['Allow'] == 'HEAD, GET, POST' req = mock_request() rsp_headers, code, _ = api_.manage_collection_item( req, 'options', 'obs', 'ressource_id') assert code == HTTPStatus.OK assert rsp_headers['Allow'] == 'HEAD, GET, PUT, DELETE' def test_describe_collections_enclosures(config_enclosure, enclosure_api): original_enclosures = { lnk['title']: lnk for lnk in config_enclosure['resources']['objects']['links'] if lnk['rel'] == 'enclosure' } req = mock_request() _, _, response = enclosure_api.describe_collections(req, 'objects') features = json.loads(response) modified_enclosures = { lnk['title']: lnk for lnk in features['links'] if lnk['rel'] == 'enclosure' } # If type and length is set, do not verify/update link assert original_enclosures['download link 1'] == \ modified_enclosures['download link 1'] # If length is missing, modify link type and length assert original_enclosures['download link 2']['type'] == \ modified_enclosures['download link 2']['type'] assert modified_enclosures['download link 2']['type'] == \ modified_enclosures['download link 3']['type'] assert 'length' not in original_enclosures['download link 2'] assert modified_enclosures['download link 2']['length'] > 0 assert modified_enclosures['download link 2']['length'] == \ modified_enclosures['download link 3']['length'] assert original_enclosures['download link 3']['type'] != \ modified_enclosures['download link 3']['type'] def test_get_collection_items_json_ld(config, api_): req = mock_request({ 'f': 'jsonld', 'limit': 2 }) rsp_headers, code, response = api_.get_collection_items(req, 'obs') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSONLD] # No language requested: return default from YAML assert rsp_headers['Content-Language'] == 'en-US' collection = json.loads(response) assert '@context' in collection assert all((f in collection['@context'][0] for f in ('schema', 'type', 'features', 'FeatureCollection'))) assert len(collection['@context']) > 1 assert collection['@context'][1]['schema'] == 'https://schema.org/' expanded = jsonld.expand(collection)[0] featuresUri = 'https://schema.org/itemListElement' assert len(expanded[featuresUri]) == 2 def test_get_collection_item(config, api_): req = mock_request({'f': 'foo'}) rsp_headers, code, response = api_.get_collection_item(req, 'obs', '371') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'f': 'json'}) rsp_headers, code, response = api_.get_collection_item( req, 'gdps-temperature', '371') assert code == HTTPStatus.BAD_REQUEST req = mock_request() rsp_headers, code, response = api_.get_collection_item(req, 'foo', '371') assert code == HTTPStatus.NOT_FOUND rsp_headers, code, response = api_.get_collection_item( req, 'obs', 'notfound') assert code == HTTPStatus.NOT_FOUND req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.get_collection_item(req, 'obs', '371') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] assert rsp_headers['Content-Language'] == 'en-US' req = mock_request() rsp_headers, code, response = api_.get_collection_item(req, 'obs', '371') feature = json.loads(response) assert feature['properties']['stn_id'] == 35 assert 'prev' not in feature['links'] assert 'next' not in feature['links'] def test_get_collection_item_json_ld(config, api_): req = mock_request({'f': 'jsonld'}) rsp_headers, _, response = api_.get_collection_item(req, 'objects', '3') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSONLD] assert rsp_headers['Content-Language'] == 'en-US' feature = json.loads(response) assert '@context' in feature assert all((f in feature['@context'][0] for f in ('schema', 'type', 'gsp'))) assert len(feature['@context']) == 1 assert 'schema' in feature['@context'][0] assert feature['@context'][0]['schema'] == 'https://schema.org/' assert feature['id'] == 3 expanded = jsonld.expand(feature)[0] assert expanded['@id'].startswith('http://') assert expanded['@id'].endswith('/collections/objects/items/3') assert expanded['http://www.opengis.net/ont/geosparql#hasGeometry'][0][ 'http://www.opengis.net/ont/geosparql#asWKT'][0][ '@value'] == 'POINT (-85 33)' assert expanded['https://schema.org/geo'][0][ 'https://schema.org/latitude'][0][ '@value'] == 33 assert expanded['https://schema.org/geo'][0][ 'https://schema.org/longitude'][0][ '@value'] == -85 _, _, response = api_.get_collection_item(req, 'objects', '2') feature = json.loads(response) assert feature['geometry']['type'] == 'MultiPoint' expanded = jsonld.expand(feature)[0] assert expanded['http://www.opengis.net/ont/geosparql#hasGeometry'][0][ 'http://www.opengis.net/ont/geosparql#asWKT'][0][ '@value'] == 'MULTIPOINT (10 40, 40 30, 20 20, 30 10)' assert expanded['https://schema.org/geo'][0][ 'https://schema.org/polygon'][0][ '@value'] == "10.0,40.0 40.0,30.0 20.0,20.0 30.0,10.0 10.0,40.0" _, _, response = api_.get_collection_item(req, 'objects', '1') feature = json.loads(response) expanded = jsonld.expand(feature)[0] assert expanded['http://www.opengis.net/ont/geosparql#hasGeometry'][0][ 'http://www.opengis.net/ont/geosparql#asWKT'][0][ '@value'] == 'LINESTRING (30 10, 10 30, 40 40)' assert expanded['https://schema.org/geo'][0][ 'https://schema.org/line'][0][ '@value'] == '30.0,10.0 10.0,30.0 40.0,40.0' _, _, response = api_.get_collection_item(req, 'objects', '4') feature = json.loads(response) expanded = jsonld.expand(feature)[0] assert expanded['http://www.opengis.net/ont/geosparql#hasGeometry'][0][ 'http://www.opengis.net/ont/geosparql#asWKT'][0][ '@value'] == 'MULTILINESTRING ((10 10, 20 20, 10 40), ' \ '(40 40, 30 30, 40 20, 30 10))' assert expanded['https://schema.org/geo'][0][ 'https://schema.org/line'][0][ '@value'] == '10.0,10.0 20.0,20.0 10.0,40.0 40.0,40.0 ' \ '30.0,30.0 40.0,20.0 30.0,10.0' _, _, response = api_.get_collection_item(req, 'objects', '5') feature = json.loads(response) expanded = jsonld.expand(feature)[0] assert expanded['http://www.opengis.net/ont/geosparql#hasGeometry'][0][ 'http://www.opengis.net/ont/geosparql#asWKT'][0][ '@value'] == 'POLYGON ((30 10, 40 40, 20 40, 10 20, 30 10))' assert expanded['https://schema.org/geo'][0][ 'https://schema.org/polygon'][0][ '@value'] == '30.0,10.0 40.0,40.0 20.0,40.0 10.0,20.0 30.0,10.0' _, _, response = api_.get_collection_item(req, 'objects', '7') feature = json.loads(response) expanded = jsonld.expand(feature)[0] assert expanded['http://www.opengis.net/ont/geosparql#hasGeometry'][0][ 'http://www.opengis.net/ont/geosparql#asWKT'][0][ '@value'] == 'MULTIPOLYGON (((30 20, 45 40, 10 40, 30 20)), '\ '((15 5, 40 10, 10 20, 5 10, 15 5)))' assert expanded['https://schema.org/geo'][0][ 'https://schema.org/polygon'][0][ '@value'] == '15.0,5.0 5.0,10.0 10.0,40.0 '\ '45.0,40.0 40.0,10.0 15.0,5.0' req = mock_request({'f': 'jsonld', 'lang': 'fr'}) rsp_headers, code, response = api_.get_collection_item(req, 'obs', '371') assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSONLD] assert rsp_headers['Content-Language'] == 'fr-CA' def test_get_collection_coverage(config, api_): req = mock_request() rsp_headers, code, response = api_.get_collection_coverage( req, 'obs') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'properties': '12'}) rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'subset': 'bad_axis(10:20)'}) rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'f': 'blah'}) rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') assert code == HTTPStatus.BAD_REQUEST req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') assert code == HTTPStatus.BAD_REQUEST assert rsp_headers['Content-Type'] == 'text/html' req = mock_request(HTTP_ACCEPT='text/html') rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') # NOTE: This test used to assert the code to be 200 OK, # but it requested HTML, which is not available, # so it should be 400 Bad Request assert code == HTTPStatus.BAD_REQUEST assert rsp_headers['Content-Type'] == 'text/html' req = mock_request({'subset': 'Lat(5:10),Long(5:10)'}) rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') assert code == HTTPStatus.OK content = json.loads(response) assert content['domain']['axes']['x']['num'] == 35 assert content['domain']['axes']['y']['num'] == 35 assert 'TMP' in content['parameters'] assert 'TMP' in content['ranges'] assert content['ranges']['TMP']['axisNames'] == ['y', 'x'] req = mock_request({'bbox': '-79,45,-75,49'}) rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') assert code == HTTPStatus.OK content = json.loads(response) assert content['domain']['axes']['x']['start'] == -79.0 assert content['domain']['axes']['x']['stop'] == -75.0 assert content['domain']['axes']['y']['start'] == 49.0 assert content['domain']['axes']['y']['stop'] == 45.0 req = mock_request({ 'subset': 'Lat(5:10),Long(5:10)', 'f': 'GRIB' }) rsp_headers, code, response = api_.get_collection_coverage( req, 'gdps-temperature') assert code == HTTPStatus.OK assert isinstance(response, bytes) req = mock_request(HTTP_ACCEPT='application/x-netcdf') rsp_headers, code, response = api_.get_collection_coverage( req, 'cmip5') assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == 'application/x-netcdf' # req = mock_request({ # 'subset': 'time("2006-07-01T06:00:00":"2007-07-01T06:00:00")' # }) # rsp_headers, code, response = api_.get_collection_coverage(req, 'cmip5') # # assert code == HTTPStatus.OK # assert isinstance(json.loads(response), dict) # req = mock_request({'subset': 'lat(1:2'}) # rsp_headers, code, response = api_.get_collection_coverage(req, 'cmip5') # # assert code == HTTPStatus.BAD_REQUEST # # req = mock_request({'subset': 'lat(1:2)'}) # rsp_headers, code, response = api_.get_collection_coverage(req, 'cmip5') # # assert code == HTTPStatus.NO_CONTENT def test_get_collection_map(config, api_): req = mock_request() rsp_headers, code, response = api_.get_collection_map(req, 'notfound') assert code == HTTPStatus.NOT_FOUND req = mock_request() rsp_headers, code, response = api_.get_collection_map( req, 'mapserver_world_map') assert code == HTTPStatus.OK assert isinstance(response, bytes) assert response[1:4] == b'PNG' def test_get_collection_tiles(config, api_): req = mock_request() rsp_headers, code, response = api_.get_collection_tiles(req, 'obs') assert code == HTTPStatus.BAD_REQUEST rsp_headers, code, response = api_.get_collection_tiles( req, 'naturalearth/lakes') assert code == HTTPStatus.OK # Language settings should be ignored (return system default) req = mock_request({'lang': 'fr'}) rsp_headers, code, response = api_.get_collection_tiles( req, 'naturalearth/lakes') assert rsp_headers['Content-Language'] == 'en-US' content = json.loads(response) assert len(content['links']) > 0 assert len(content['tilesets']) > 0 def test_describe_processes(config, api_): req = mock_request({'limit': 1}) # Test for description of single processes rsp_headers, code, response = api_.describe_processes(req) data = json.loads(response) assert code == HTTPStatus.OK assert len(data['processes']) == 1 assert len(data['links']) == 3 req = mock_request() # Test for undefined process rsp_headers, code, response = api_.describe_processes(req, 'foo') data = json.loads(response) assert code == HTTPStatus.NOT_FOUND assert data['code'] == 'NoSuchProcess' # Test for description of all processes rsp_headers, code, response = api_.describe_processes(req) data = json.loads(response) assert code == HTTPStatus.OK assert len(data['processes']) == 2 assert len(data['links']) == 3 # Test for particular, defined process rsp_headers, code, response = api_.describe_processes(req, 'hello-world') process = json.loads(response) assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] assert process['id'] == 'hello-world' assert process['version'] == '0.2.0' assert process['title'] == 'Hello World' assert len(process['keywords']) == 3 assert len(process['links']) == 6 assert len(process['inputs']) == 2 assert len(process['outputs']) == 1 assert len(process['outputTransmission']) == 1 assert len(process['jobControlOptions']) == 2 assert 'sync-execute' in process['jobControlOptions'] assert 'async-execute' in process['jobControlOptions'] # Check HTML response when requested in headers req = mock_request(HTTP_ACCEPT='text/html') rsp_headers, code, response = api_.describe_processes(req, 'hello-world') assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] # No language requested: return default from YAML assert rsp_headers['Content-Language'] == 'en-US' # Check JSON response when requested in headers req = mock_request(HTTP_ACCEPT='application/json') rsp_headers, code, response = api_.describe_processes(req, 'hello-world') assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] assert rsp_headers['Content-Language'] == 'en-US' # Check HTML response when requested with query parameter req = mock_request({'f': 'html'}) rsp_headers, code, response = api_.describe_processes(req, 'hello-world') assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] # No language requested: return default from YAML assert rsp_headers['Content-Language'] == 'en-US' # Check JSON response when requested with query parameter req = mock_request({'f': 'json'}) rsp_headers, code, response = api_.describe_processes(req, 'hello-world') assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] assert rsp_headers['Content-Language'] == 'en-US' # Check JSON response when requested with French language parameter req = mock_request({'lang': 'fr'}) rsp_headers, code, response = api_.describe_processes(req, 'hello-world') assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] assert rsp_headers['Content-Language'] == 'fr-CA' process = json.loads(response) assert process['title'] == 'Bonjour le Monde' # Check JSON response when language requested in headers req = mock_request(HTTP_ACCEPT_LANGUAGE='fr') rsp_headers, code, response = api_.describe_processes(req, 'hello-world') assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] assert rsp_headers['Content-Language'] == 'fr-CA' # Test for undefined process req = mock_request() rsp_headers, code, response = api_.describe_processes(req, 'goodbye-world') data = json.loads(response) assert code == HTTPStatus.NOT_FOUND assert data['code'] == 'NoSuchProcess' assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_JSON] # Test describe doesn't crash if example is missing req = mock_request() processor = api_.manager.get_processor("hello-world") example = processor.metadata.pop("example") rsp_headers, code, response = api_.describe_processes(req) processor.metadata['example'] = example data = json.loads(response) assert code == HTTPStatus.OK assert len(data['processes']) == 2 def test_execute_process(config, api_): req_body_0 = { 'inputs': { 'name': 'Test' } } req_body_1 = { 'inputs': { 'name': 'Test' }, 'response': 'document' } req_body_2 = { 'inputs': { 'name': 'Tést' } } req_body_3 = { 'inputs': { 'name': 'Tést', 'message': 'This is a test.' } } req_body_4 = { 'inputs': { 'foo': 'Tést' } } req_body_5 = { 'inputs': {} } req_body_6 = { 'inputs': { 'name': None } } req_body_7 = { 'inputs': { 'name': 'Test' }, 'subscriber': { 'successUri': 'https://example.com/success', 'inProgressUri': 'https://example.com/inProgress', 'failedUri': 'https://example.com/failed', } } cleanup_jobs = set() # Test posting empty payload to existing process req = mock_request(data='') rsp_headers, code, response = api_.execute_process(req, 'hello-world') assert rsp_headers['Content-Language'] == 'en-US' data = json.loads(response) assert code == HTTPStatus.BAD_REQUEST assert 'Location' not in rsp_headers assert data['code'] == 'MissingParameterValue' req = mock_request(data=req_body_0) rsp_headers, code, response = api_.execute_process(req, 'foo') data = json.loads(response) assert code == HTTPStatus.NOT_FOUND assert 'Location' not in rsp_headers assert data['code'] == 'NoSuchProcess' rsp_headers, code, response = api_.execute_process(req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.OK assert 'Location' in rsp_headers assert len(data.keys()) == 2 assert data['id'] == 'echo' assert data['value'] == 'Hello Test!' cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_1) rsp_headers, code, response = api_.execute_process(req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.OK assert 'Location' in rsp_headers assert len(data.keys()) == 1 assert data['outputs'][0]['id'] == 'echo' assert data['outputs'][0]['value'] == 'Hello Test!' cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_2) rsp_headers, code, response = api_.execute_process(req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.OK assert 'Location' in rsp_headers assert data['value'] == 'Hello Tést!' cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_3) rsp_headers, code, response = api_.execute_process(req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.OK assert 'Location' in rsp_headers assert data['value'] == 'Hello Tést! This is a test.' cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_4) rsp_headers, code, response = api_.execute_process(req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.BAD_REQUEST assert 'Location' in rsp_headers assert data['code'] == 'InvalidParameterValue' cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_5) rsp_headers, code, response = api_.execute_process(req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.BAD_REQUEST assert 'Location' in rsp_headers assert data['code'] == 'InvalidParameterValue' assert data['description'] == 'Error updating job' cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_6) rsp_headers, code, response = api_.execute_process(req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.BAD_REQUEST assert 'Location' in rsp_headers assert data['code'] == 'InvalidParameterValue' assert data['description'] == 'Error updating job' cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_0) rsp_headers, code, response = api_.execute_process(req, 'goodbye-world') response = json.loads(response) assert code == HTTPStatus.NOT_FOUND assert 'Location' not in rsp_headers assert response['code'] == 'NoSuchProcess' rsp_headers, code, response = api_.execute_process(req, 'hello-world') response = json.loads(response) assert code == HTTPStatus.OK cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_1, HTTP_Prefer='respond-async') rsp_headers, code, response = api_.execute_process(req, 'hello-world') assert 'Location' in rsp_headers response = json.loads(response) assert isinstance(response, dict) assert code == HTTPStatus.CREATED cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) req = mock_request(data=req_body_7) with mock.patch( 'pygeoapi.process.manager.base.requests.post' ) as post_mocker: rsp_headers, code, response = api_.execute_process(req, 'hello-world') assert code == HTTPStatus.OK post_mocker.assert_any_call( req_body_7['subscriber']['inProgressUri'], json={} ) post_mocker.assert_any_call( req_body_7['subscriber']['successUri'], json={'id': 'echo', 'value': 'Hello Test!'} ) assert post_mocker.call_count == 2 cleanup_jobs.add(tuple(['hello-world', rsp_headers['Location'].split('/')[-1]])) # Cleanup time.sleep(2) # Allow time for any outstanding async jobs for _, job_id in cleanup_jobs: rsp_headers, code, response = api_.delete_job(mock_request(), job_id) assert code == HTTPStatus.OK def _execute_a_job(api_): req_body_sync = { 'inputs': { 'name': 'Sync Test' } } req = mock_request(data=req_body_sync) rsp_headers, code, response = api_.execute_process( req, 'hello-world') data = json.loads(response) assert code == HTTPStatus.OK assert 'Location' in rsp_headers assert data['value'] == 'Hello Sync Test!' job_id = rsp_headers['Location'].split('/')[-1] return job_id def test_delete_job(api_): rsp_headers, code, response = api_.delete_job( mock_request(), 'does-not-exist') assert code == HTTPStatus.NOT_FOUND req_body_async = { 'inputs': { 'name': 'Async Test Deletion' } } job_id = _execute_a_job(api_) rsp_headers, code, response = api_.delete_job(mock_request(), job_id) assert code == HTTPStatus.OK rsp_headers, code, response = api_.delete_job(mock_request(), job_id) assert code == HTTPStatus.NOT_FOUND req = mock_request(data=req_body_async, HTTP_Prefer='respond-async') rsp_headers, code, response = api_.execute_process( req, 'hello-world') assert code == HTTPStatus.CREATED assert 'Location' in rsp_headers time.sleep(2) # Allow time for async execution to complete job_id = rsp_headers['Location'].split('/')[-1] rsp_headers, code, response = api_.delete_job(mock_request(), job_id) assert code == HTTPStatus.OK rsp_headers, code, response = api_.delete_job(mock_request(), job_id) assert code == HTTPStatus.NOT_FOUND def test_get_job_result(api_): rsp_headers, code, response = api_.get_job_result(mock_request(), 'not-exist') assert code == HTTPStatus.NOT_FOUND job_id = _execute_a_job(api_) rsp_headers, code, response = api_.get_job_result(mock_request(), job_id) # default response is html assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == 'text/html' assert 'Hello Sync Test!' in response rsp_headers, code, response = api_.get_job_result( mock_request({'f': 'json'}), job_id, ) assert code == HTTPStatus.OK assert rsp_headers['Content-Type'] == 'application/json' assert json.loads(response)['value'] == "Hello Sync Test!" def test_get_collection_edr_query(config, api_): # edr resource req = mock_request() rsp_headers, code, response = api_.describe_collections(req, 'icoads-sst') collection = json.loads(response) parameter_names = list(collection['parameter_names'].keys()) parameter_names.sort() assert len(parameter_names) == 4 assert parameter_names == ['AIRT', 'SST', 'UWND', 'VWND'] # no coords parameter rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.BAD_REQUEST # bad query type req = mock_request({'coords': 'POINT(11 11)'}) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'corridor') assert code == HTTPStatus.BAD_REQUEST # bad coords parameter req = mock_request({'coords': 'gah'}) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.BAD_REQUEST # bad parameter_names parameter req = mock_request({ 'coords': 'POINT(11 11)', 'parameter_names': 'bad' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.BAD_REQUEST # all parameters req = mock_request({'coords': 'POINT(11 11)'}) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.OK data = json.loads(response) axes = list(data['domain']['axes'].keys()) axes.sort() assert len(axes) == 3 assert axes == ['TIME', 'x', 'y'] assert data['domain']['axes']['x']['start'] == 11.0 assert data['domain']['axes']['x']['stop'] == 11.0 assert data['domain']['axes']['y']['start'] == 11.0 assert data['domain']['axes']['y']['stop'] == 11.0 parameters = list(data['parameters'].keys()) parameters.sort() assert len(parameters) == 4 assert parameters == ['AIRT', 'SST', 'UWND', 'VWND'] # single parameter req = mock_request({ 'coords': 'POINT(11 11)', 'parameter_names': 'SST' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.OK data = json.loads(response) assert len(data['parameters'].keys()) == 1 assert list(data['parameters'].keys())[0] == 'SST' # Zulu time zone req = mock_request({ 'coords': 'POINT(11 11)', 'datetime': '2000-01-17T00:00:00Z/2000-06-16T23:00:00Z' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.OK # bounded date range req = mock_request({ 'coords': 'POINT(11 11)', 'datetime': '2000-01-17/2000-06-16' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.OK data = json.loads(response) time_dict = data['domain']['axes']['TIME'] assert time_dict['start'] == '2000-02-15T16:29:05.999999999' assert time_dict['stop'] == '2000-06-16T10:25:30.000000000' assert time_dict['num'] == 5 # unbounded date range - start req = mock_request({ 'coords': 'POINT(11 11)', 'datetime': '../2000-06-16' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.OK data = json.loads(response) time_dict = data['domain']['axes']['TIME'] assert time_dict['start'] == '2000-01-16T06:00:00.000000000' assert time_dict['stop'] == '2000-06-16T10:25:30.000000000' assert time_dict['num'] == 6 # unbounded date range - end req = mock_request({ 'coords': 'POINT(11 11)', 'datetime': '2000-06-16/..' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.OK data = json.loads(response) time_dict = data['domain']['axes']['TIME'] assert time_dict['start'] == '2000-06-16T10:25:30.000000000' assert time_dict['stop'] == '2000-12-16T01:20:05.999999996' assert time_dict['num'] == 7 # some data req = mock_request({ 'coords': 'POINT(11 11)', 'datetime': '2000-01-16' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.OK # no data req = mock_request({ 'coords': 'POINT(11 11)', 'datetime': '2000-01-17' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.NO_CONTENT # position no coords req = mock_request({ 'datetime': '2000-01-17' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'position') assert code == HTTPStatus.BAD_REQUEST # cube bbox parameter 4 dimensional req = mock_request({ 'bbox': '0,0,10,10' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'cube') assert code == HTTPStatus.OK # cube bad bbox parameter req = mock_request({ 'bbox': '0,0,10' }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'cube') assert code == HTTPStatus.BAD_REQUEST # cube no bbox parameter req = mock_request({}) rsp_headers, code, response = api_.get_collection_edr_query( req, 'icoads-sst', None, 'cube') assert code == HTTPStatus.BAD_REQUEST # cube decreasing latitude coords and S3 req = mock_request({ 'bbox': '-100,40,-99,45', 'parameter_names': 'tmn', 'datetime': '1994-01-01/1994-12-31', }) rsp_headers, code, response = api_.get_collection_edr_query( req, 'usgs-prism', None, 'cube') assert code == HTTPStatus.OK def test_validate_bbox(): assert validate_bbox('1,2,3,4') == [1, 2, 3, 4] assert validate_bbox('1,2,3,4,5,6') == [1, 2, 3, 4, 5, 6] assert validate_bbox('-142,42,-52,84') == [-142, 42, -52, 84] assert (validate_bbox('-142.1,42.12,-52.22,84.4') == [-142.1, 42.12, -52.22, 84.4]) assert (validate_bbox('-142.1,42.12,-5.28,-52.22,84.4,7.39') == [-142.1, 42.12, -5.28, -52.22, 84.4, 7.39]) assert (validate_bbox('177.0,65.0,-177.0,70.0') == [177.0, 65.0, -177.0, 70.0]) with pytest.raises(ValueError): validate_bbox('1,2,4') with pytest.raises(ValueError): validate_bbox('1,2,4,5,6') with pytest.raises(ValueError): validate_bbox('3,4,1,2') with pytest.raises(ValueError): validate_bbox('1,2,6,4,5,3') def test_validate_datetime(): config = yaml_load(''' temporal: begin: 2000-10-30T18:24:39Z end: 2007-10-30T08:57:29Z ''') # test time instant assert validate_datetime(config, '2004') == '2004' assert validate_datetime(config, '2004-10') == '2004-10' assert validate_datetime(config, '2001-10-30') == '2001-10-30' with pytest.raises(ValueError): _ = validate_datetime(config, '2009-10-30') with pytest.raises(ValueError): _ = validate_datetime(config, '2000-09-09') with pytest.raises(ValueError): _ = validate_datetime(config, '2000-10-30T17:24:39Z') with pytest.raises(ValueError): _ = validate_datetime(config, '2007-10-30T08:58:29Z') # test time envelope assert validate_datetime(config, '2004/2005') == '2004/2005' assert validate_datetime(config, '2004-10/2005-10') == '2004-10/2005-10' assert (validate_datetime(config, '2001-10-30/2002-10-30') == '2001-10-30/2002-10-30') assert validate_datetime(config, '2004/..') == '2004/..' assert validate_datetime(config, '../2005') == '../2005' assert validate_datetime(config, '2004/') == '2004/..' assert validate_datetime(config, '/2005') == '../2005' assert validate_datetime(config, '2004-10/2005-10') == '2004-10/2005-10' assert (validate_datetime(config, '2001-10-30/2002-10-30') == '2001-10-30/2002-10-30') with pytest.raises(ValueError): _ = validate_datetime(config, '2007-11-01/..') with pytest.raises(ValueError): _ = validate_datetime(config, '2009/..') with pytest.raises(ValueError): _ = validate_datetime(config, '../2000-09') with pytest.raises(ValueError): _ = validate_datetime(config, '../1999') @pytest.mark.parametrize("value, expected", [ ('time(2000-11-11)', {'time': ['2000-11-11']}), ('time("2000-11-11")', {'time': ['2000-11-11']}), ('time("2000-11-11T00:11:11")', {'time': ['2000-11-11T00:11:11']}), ('time("2000-11-11T11:12:13":"2021-12-22T:13:33:33")', {'time': ['2000-11-11T11:12:13', '2021-12-22T:13:33:33']}), # noqa ('lat(40)', {'lat': [40]}), ('lat(0:40)', {'lat': [0, 40]}), ('foo("bar")', {'foo': ['bar']}), ('foo("bar":"baz")', {'foo': ['bar', 'baz']}) ]) def test_validate_subset(value, expected): assert validate_subset(value) == expected with pytest.raises(ValueError): validate_subset('foo("bar)') def test_get_exception(config, api_): d = api_.get_exception(500, {}, 'json', 'NoApplicableCode', 'oops') assert d[0] == {} assert d[1] == 500 content = json.loads(d[2]) assert content['code'] == 'NoApplicableCode' assert content['description'] == 'oops' d = api_.get_exception(500, {}, 'html', 'NoApplicableCode', 'oops') assert d[0] == {'Content-Type': 'text/html'}