Sander Schaminee
41f3bd801a
* API design rule support (#1134): - Change config model - Change Flask, Starlette, and Django apps for API rule adherence - Add Flask and Starlette mock clients to test API rule adherence - Add get_base_url() util function to replace all config['server']['url'] refs - Ensure that any internal links have URL prefixes if needed - Add tests and update docs * Prevent fcntl import error (breaks tests when running locally on Windows) * Prefer trailing slash in landing page URL when strict_slashes=True (#1134)
237 lines
8.1 KiB
Python
237 lines
8.1 KiB
Python
# =================================================================
|
|
#
|
|
# Authors: Tom Kralidis <tomkralidis@gmail.com>
|
|
#
|
|
# Copyright (c) 2021 Tom Kralidis
|
|
#
|
|
# 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.
|
|
#
|
|
# =================================================================
|
|
|
|
from datetime import datetime, date, time
|
|
from decimal import Decimal
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from pygeoapi import util
|
|
from pygeoapi.api import __version__
|
|
from pygeoapi.provider.base import ProviderTypeError
|
|
|
|
from .util import get_test_file_path
|
|
|
|
|
|
@pytest.fixture()
|
|
def config():
|
|
with open(get_test_file_path('pygeoapi-test-config.yml')) as fh:
|
|
return util.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 util.yaml_load(fh)
|
|
|
|
|
|
def test_get_typed_value():
|
|
value = util.get_typed_value('2')
|
|
assert isinstance(value, int)
|
|
|
|
value = util.get_typed_value('1.2')
|
|
assert isinstance(value, float)
|
|
|
|
value = util.get_typed_value('1.c2')
|
|
assert isinstance(value, str)
|
|
|
|
|
|
def test_yaml_load(config):
|
|
assert isinstance(config, dict)
|
|
with pytest.raises(FileNotFoundError):
|
|
with open(get_test_file_path('404.yml')) as fh:
|
|
util.yaml_load(fh)
|
|
|
|
|
|
def test_str2bool():
|
|
assert not util.str2bool(False)
|
|
assert not util.str2bool('0')
|
|
assert not util.str2bool('no')
|
|
assert util.str2bool('yes')
|
|
assert util.str2bool('1')
|
|
assert util.str2bool(True)
|
|
assert util.str2bool('true')
|
|
assert util.str2bool('True')
|
|
assert util.str2bool('TRUE')
|
|
assert util.str2bool('tRuE')
|
|
assert util.str2bool('on')
|
|
assert util.str2bool('On')
|
|
assert not util.str2bool('off')
|
|
|
|
|
|
def test_json_serial():
|
|
d = datetime(1972, 10, 30)
|
|
assert util.json_serial(d) == '1972-10-30T00:00:00'
|
|
|
|
d = date(2010, 7, 31)
|
|
assert util.json_serial(d) == '2010-07-31'
|
|
|
|
d = time(11)
|
|
assert util.json_serial(d) == '11:00:00'
|
|
|
|
d = Decimal(1.0)
|
|
assert util.json_serial(d) == 1.0
|
|
|
|
with pytest.raises(TypeError):
|
|
util.json_serial('foo')
|
|
|
|
|
|
def test_mimetype():
|
|
assert util.get_mimetype('file.xml') == 'application/xml'
|
|
assert util.get_mimetype('file.yml') == 'text/plain'
|
|
assert util.get_mimetype('file.yaml') == 'text/plain'
|
|
|
|
|
|
def test_get_breadcrumbs():
|
|
path = '/dataset/model-run/forecast-hour/variable.grib2'
|
|
breadcrumbs = util.get_breadcrumbs(path)
|
|
|
|
assert len(breadcrumbs) == 5
|
|
assert breadcrumbs[3]['href'] == 'dataset/model-run/forecast-hour'
|
|
|
|
|
|
def test_path_basename():
|
|
assert util.get_path_basename('/path/to/file.txt') == 'file.txt'
|
|
assert util.get_path_basename('/path/to/dir') == 'dir'
|
|
|
|
|
|
def test_filter_dict_by_key_value(config):
|
|
collections = util.filter_dict_by_key_value(config['resources'],
|
|
'type', 'collection')
|
|
assert len(collections) == 7
|
|
|
|
notfound = util.filter_dict_by_key_value(config['resources'],
|
|
'type', 'foo')
|
|
|
|
assert len(notfound) == 0
|
|
|
|
|
|
def test_get_provider_by_type(config):
|
|
p = util.get_provider_by_type(config['resources']['obs']['providers'],
|
|
'feature')
|
|
|
|
assert isinstance(p, dict)
|
|
assert p['type'] == 'feature'
|
|
assert p['name'] == 'CSV'
|
|
|
|
with pytest.raises(ProviderTypeError):
|
|
p = util.get_provider_by_type(config['resources']['obs']['providers'],
|
|
'something-else')
|
|
|
|
|
|
def test_get_provider_default(config):
|
|
pd = util.get_provider_default(config['resources']['obs']['providers'])
|
|
|
|
assert pd['type'] == 'feature'
|
|
assert pd['name'] == 'CSV'
|
|
|
|
pd = util.get_provider_default(config['resources']['obs']['providers'])
|
|
|
|
|
|
def test_read_data():
|
|
data = util.read_data(get_test_file_path('pygeoapi-test-config.yml'))
|
|
|
|
assert isinstance(data, bytes)
|
|
|
|
|
|
def test_url_join():
|
|
f = util.url_join
|
|
assert f('http://localhost:5000') == 'http://localhost:5000'
|
|
assert f('http://localhost:5000/') == 'http://localhost:5000'
|
|
assert f('http://localhost:5000', '') == 'http://localhost:5000'
|
|
assert f('http://localhost:5000/', '') == 'http://localhost:5000'
|
|
assert f('http://localhost:5000/', '/') == 'http://localhost:5000'
|
|
assert f('http://localhost:5000/api', '/') == 'http://localhost:5000/api'
|
|
assert f('http://localhost:5000/api', '/v0') == 'http://localhost:5000/api/v0' # noqa
|
|
assert f('http://localhost:5000/api', '/v0/') == 'http://localhost:5000/api/v0' # noqa
|
|
assert f('http://localhost:5000', 'api', 'v0') == 'http://localhost:5000/api/v0' # noqa
|
|
|
|
|
|
def test_get_base_url(config, config_with_rules):
|
|
assert util.get_base_url(config) == 'http://localhost:5000'
|
|
assert util.get_base_url(config_with_rules) == 'http://localhost:5000/api/v0' # noqa
|
|
|
|
|
|
def test_get_api_rules(config, config_with_rules):
|
|
# Test unset/default rules
|
|
rules = util.get_api_rules(config)
|
|
assert not rules.strict_slashes
|
|
assert not rules.url_prefix
|
|
assert rules.api_version == __version__
|
|
assert rules.version_header == ''
|
|
assert rules.get_url_prefix() == ''
|
|
assert rules.response_headers == {}
|
|
|
|
# Test configured rules
|
|
rules = util.get_api_rules(config_with_rules)
|
|
assert rules.strict_slashes
|
|
assert rules.url_prefix
|
|
assert rules.api_version == __version__
|
|
assert rules.version_header == 'X-API-Version'
|
|
assert rules.response_headers == {'X-API-Version': __version__}
|
|
|
|
# Test specific version override
|
|
config_changed = deepcopy(config_with_rules)
|
|
config_changed['server']['api_rules']['api_version'] = '1.2.3'
|
|
rules = util.get_api_rules(config_changed)
|
|
assert rules.api_version == '1.2.3'
|
|
assert rules.get_url_prefix() == 'v1'
|
|
assert rules.get_url_prefix('flask') == '/v1'
|
|
assert rules.get_url_prefix('starlette') == '/v1'
|
|
assert rules.get_url_prefix('django') == r'^v1/'
|
|
|
|
# Test prefix without version
|
|
config_changed = deepcopy(config_with_rules)
|
|
config_changed['server']['api_rules']['url_prefix'] = 'test'
|
|
rules = util.get_api_rules(config_changed)
|
|
assert rules.get_url_prefix() == 'test'
|
|
assert rules.get_url_prefix('flask') == '/test'
|
|
assert rules.get_url_prefix('starlette') == '/test'
|
|
assert rules.get_url_prefix('django') == r'^test/'
|
|
|
|
|
|
def test_prefetcher():
|
|
prefetcher = util.UrlPrefetcher()
|
|
assert prefetcher.get_headers('bad_url') == {}
|
|
# URL below will redirect once
|
|
url = 'https://github.com/geopython/pygeoapi/raw/4a18393662583e53b8c7d591130246d9cd2c3f3f/pygeoapi/static/img/pygeoapi.png' # noqa
|
|
headers = prefetcher.get_headers(url)
|
|
length = int(headers.get('content-length', 0))
|
|
assert length > 0
|
|
# Test without redirect
|
|
headers = prefetcher.get_headers(url, allow_redirects=False)
|
|
assert headers.get('content-length') in (0, '0', None)
|
|
assert headers.get('content-type') != 'image/png'
|
|
# Test using redirect location from header
|
|
headers = prefetcher.get_headers(headers['location'])
|
|
assert int(headers.get('content-length', 0)) == length
|
|
assert headers.get('content-type') == 'image/png'
|