Skip to content

Commit

Permalink
Update test_postgresql_provider.py
Browse files Browse the repository at this point in the history
  • Loading branch information
@webb-ben
webb-ben committed Jul 17, 2024
1 parent 2e599ec commit 7f17539
Showing 1 changed file with 0 additions and 353 deletions.
353 changes: 0 additions & 353 deletions tests/test_postgresql_provider.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,17 +39,10 @@
# test database in Docker

import os
import json
import pytest
import pyproj
from http import HTTPStatus

from pygeofilter.parsers.ecql import parse

from pygeoapi.api import API
from pygeoapi.api.itemtypes import (
get_collection_items, get_collection_item, post_collection_items
)
from pygeoapi.provider.base import (
ProviderConnectionError,
ProviderItemNotFoundError,
Expand All @@ -58,11 +51,6 @@
from pygeoapi.provider.postgresql import PostgreSQLProvider
import pygeoapi.provider.postgresql as postgresql_provider_module

from pygeoapi.util import (yaml_load, geojson_to_geom,
get_transform_from_crs, get_crs_from_uri)

from .util import get_test_file_path, mock_api_request

PASSWORD = os.environ.get('POSTGRESQL_PASSWORD', 'postgres')
DEFAULT_CRS = 'http://www.opengis.net/def/crs/OGC/1.3/CRS84'

Expand Down Expand Up @@ -107,20 +95,6 @@ def config_types():
}


@pytest.fixture()
def openapi():
with open(get_test_file_path('pygeoapi-test-openapi.yml')) as fh:
return yaml_load(fh)


# API using PostgreSQL provider
@pytest.fixture()
def pg_api_(openapi):
with open(get_test_file_path('pygeoapi-test-config-postgresql.yml')) as fh:
config = yaml_load(fh)
return API(config, openapi)


def test_valid_connection_options(config):
if config.get('options'):
keys = list(config['options'].keys())
Expand Down Expand Up @@ -468,330 +442,3 @@ def test_engine_and_table_model_stores(config):
provider3 = PostgreSQLProvider(different_host)
assert provider3._engine is not provider0._engine
assert provider3.table_model is not provider0.table_model


# START: EXTERNAL API TESTS
def test_get_collection_items_postgresql_cql(pg_api_):
"""
Test for PostgreSQL CQL - requires local PostgreSQL with appropriate
data. See pygeoapi/provider/postgresql.py for details.
"""
# Arrange
cql_query = 'osm_id BETWEEN 80800000 AND 80900000 AND name IS NULL'
expected_ids = [80835474, 80835483]

# Act
req = mock_api_request({
'filter-lang': 'cql-text',
'filter': cql_query
})
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'hot_osm_waterways')

# Assert
assert code == HTTPStatus.OK
features = json.loads(response)
ids = [item['id'] for item in features['features']]
assert ids == expected_ids

# Act, no filter-lang
req = mock_api_request({
'filter': cql_query
})
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'hot_osm_waterways')

# Assert
assert code == HTTPStatus.OK
features = json.loads(response)
ids = [item['id'] for item in features['features']]
assert ids == expected_ids


def test_get_collection_items_postgresql_cql_invalid_filter_language(pg_api_):
"""
Test for PostgreSQL CQL - requires local PostgreSQL with appropriate
data. See pygeoapi/provider/postgresql.py for details.
Test for invalid filter language
"""
# Arrange
cql_query = 'osm_id BETWEEN 80800000 AND 80900000 AND name IS NULL'

# Act
req = mock_api_request({
'filter-lang': 'cql-json', # Only cql-text is valid for GET
'filter': cql_query
})
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'hot_osm_waterways')

# Assert
assert code == HTTPStatus.BAD_REQUEST
error_response = json.loads(response)
assert error_response['code'] == 'InvalidParameterValue'
assert error_response['description'] == 'Invalid filter language'


@pytest.mark.parametrize("bad_cql", [
'id IN (1, ~)',
'id EATS (1, 2)', # Valid CQL relations only
'id IN (1, 2' # At some point this may return UnexpectedEOF
])
def test_get_collection_items_postgresql_cql_bad_cql(pg_api_, bad_cql):
"""
Test for PostgreSQL CQL - requires local PostgreSQL with appropriate
data. See pygeoapi/provider/postgresql.py for details.
Test for bad cql
"""
# Act
req = mock_api_request({
'filter': bad_cql
})
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'hot_osm_waterways')

# Assert
assert code == HTTPStatus.BAD_REQUEST
error_response = json.loads(response)
assert error_response['code'] == 'InvalidParameterValue'
assert error_response['description'] == f'Bad CQL string : {bad_cql}'


def test_post_collection_items_postgresql_cql(pg_api_):
"""
Test for PostgreSQL CQL - requires local PostgreSQL with appropriate
data. See pygeoapi/provider/postgresql.py for details.
"""
# Arrange
cql = {"and": [{"between": {"value": {"property": "osm_id"},
"lower": 80800000,
"upper": 80900000}},
{"isNull": {"property": "name"}}]}
# werkzeug requests use a value of CONTENT_TYPE 'application/json'
# to create Content-Type in the Request object. So here we need to
# overwrite the default CONTENT_TYPE with the required one.
headers = {'CONTENT_TYPE': 'application/query-cql-json'}
expected_ids = [80835474, 80835483]

# Act
req = mock_api_request({
'filter-lang': 'cql-json'
}, data=cql, **headers)
rsp_headers, code, response = post_collection_items(
pg_api_, req, 'hot_osm_waterways')

# Assert
assert code == HTTPStatus.OK
features = json.loads(response)
ids = [item['id'] for item in features['features']]
assert ids == expected_ids


def test_post_collection_items_postgresql_cql_invalid_filter_language(pg_api_):
"""
Test for PostgreSQL CQL - requires local PostgreSQL with appropriate
data. See pygeoapi/provider/postgresql.py for details.
Test for invalid filter language
"""
# Arrange
# CQL should never be parsed
cql = {"in": {"value": {"property": "id"}, "list": [1, 2]}}
headers = {'CONTENT_TYPE': 'application/query-cql-json'}

# Act
req = mock_api_request({
'filter-lang': 'cql-text' # Only cql-json is valid for POST
}, data=cql, **headers)
rsp_headers, code, response = post_collection_items(
pg_api_, req, 'hot_osm_waterways')

# Assert
assert code == HTTPStatus.BAD_REQUEST
error_response = json.loads(response)
assert error_response['code'] == 'InvalidParameterValue'
assert error_response['description'] == 'Invalid filter language'


@pytest.mark.parametrize("bad_cql", [
# Valid CQL relations only
{"eats": {"value": {"property": "id"}, "list": [1, 2]}},
# At some point this may return UnexpectedEOF
'{"in": {"value": {"property": "id"}, "list": [1, 2}}'
])
def test_post_collection_items_postgresql_cql_bad_cql(pg_api_, bad_cql):
"""
Test for PostgreSQL CQL - requires local PostgreSQL with appropriate
data. See pygeoapi/provider/postgresql.py for details.
Test for bad cql
"""
# Arrange
headers = {'CONTENT_TYPE': 'application/query-cql-json'}

# Act
req = mock_api_request({
'filter-lang': 'cql-json'
}, data=bad_cql, **headers)
rsp_headers, code, response = post_collection_items(
pg_api_, req, 'hot_osm_waterways')

# Assert
assert code == HTTPStatus.BAD_REQUEST
error_response = json.loads(response)
assert error_response['code'] == 'InvalidParameterValue'
assert error_response['description'].startswith('Bad CQL string')


def test_get_collection_items_postgresql_crs(pg_api_):
"""Test the coordinates transformation implementation of
PostgreSQLProvider when using the crs parameter.
"""
storage_crs = DEFAULT_CRS
crs_32735 = 'http://www.opengis.net/def/crs/EPSG/0/32735'

# Without CRS query parameter -> no coordinates transformation
req = mock_api_request({'bbox': '29.0,-2.85,29.05,-2.8'})
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'hot_osm_waterways')

assert code == HTTPStatus.OK

features_orig = json.loads(response)
assert rsp_headers['Content-Crs'] == f'<{DEFAULT_CRS}>'

# With CRS query parameter not resulting in coordinates transformation
# (i.e. 'crs' query parameter is the same as 'storage_crs')
req = mock_api_request(
{'crs': storage_crs, 'bbox': '29.0,-2.85,29.05,-2.8'})
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'hot_osm_waterways')

assert code == HTTPStatus.OK
assert rsp_headers['Content-Crs'] == f'<{storage_crs}>'

features_storage_crs = json.loads(response)

# With CRS query parameter resulting in coordinates transformation
req = mock_api_request({'crs': crs_32735, 'bbox': '29.0,-2.85,29.05,-2.8'})
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'hot_osm_waterways')

assert code == HTTPStatus.OK
assert rsp_headers['Content-Crs'] == f'<{crs_32735}>'

features_32735 = json.loads(response)

# Make sure that we compare the same features
assert (
sorted(f['id'] for f in features_orig['features'])
== sorted(f['id'] for f in features_storage_crs['features'])
== sorted(f['id'] for f in features_32735['features'])
)

# Without 'crs' query parameter or with 'crs' set to 'storage_crs', the
# geometries of the returned features should be the same
for feat_orig in features_orig['features']:
id_ = feat_orig['id']
for feat_storage_crs in features_storage_crs['features']:
if id_ == feat_storage_crs['id']:
assert feat_orig['geometry'] == feat_storage_crs['geometry']
break

transform_func = get_transform_from_crs(
get_crs_from_uri(DEFAULT_CRS),
pyproj.CRS.from_epsg(32735),
always_xy=False,
)
# Check that the coordinates of returned features were transformed
for feat_orig in features_orig['features']:
id_ = feat_orig['id']
geom_orig = geojson_to_geom(feat_orig['geometry'])
for feat_32735 in features_32735['features']:
if id_ == feat_32735['id']:
geom_32735 = geojson_to_geom(feat_32735['geometry'])

assert geom_32735.equals_exact(transform_func(geom_orig), 1)
break


def test_get_collection_item_postgresql_crs(pg_api_):
"""Test the coordinates transformation implementation of
PostgreSQLProvider when using the crs parameter.
"""
storage_crs = DEFAULT_CRS
crs_32735 = 'http://www.opengis.net/def/crs/EPSG/0/32735'
# List of feature IDs located in UTM zone 35S
fid_list = [
'439338397',
'198190856',
'93063941',
'586449587',
'80827793',
'587350255',
'586994284',
'587960337',
'586449586',
'422440125',
]
for fid in fid_list:
# Without CRS query parameter -> no coordinates transformation
req = mock_api_request({'f': 'json'})
rsp_headers, code, response = get_collection_item(
pg_api_, req, 'hot_osm_waterways', fid)

assert code == HTTPStatus.OK
assert rsp_headers['Content-Crs'] == f'<{DEFAULT_CRS}>'

feat_orig = json.loads(response)
geom_orig = geojson_to_geom(feat_orig['geometry'])

# With CRS query parameter not resulting in coordinates transformation
# (i.e. 'crs' query parameter is the same as 'storage_crs')
req = mock_api_request({'f': 'json', 'crs': storage_crs})
rsp_headers, code, response = get_collection_item(
pg_api_, req, 'hot_osm_waterways', fid)

assert code == HTTPStatus.OK
assert rsp_headers['Content-Crs'] == f'<{storage_crs}>'

feat_storage_crs = json.loads(response)

# Without 'crs' query parameter or with 'crs' set to 'storage_crs', the
# geometries should be identical, when storage CRS is WGS84 lon,lat.
assert feat_orig['geometry'] == feat_storage_crs['geometry']

# With CRS query parameter resulting in coordinates transformation
req = mock_api_request({'f': 'json', 'crs': crs_32735})
rsp_headers, code, response = get_collection_item(
pg_api_, req, 'hot_osm_waterways', fid)

assert code == HTTPStatus.OK
assert rsp_headers['Content-Crs'] == f'<{crs_32735}>'

feat_32735 = json.loads(response)
geom_32735 = geojson_to_geom(feat_32735['geometry'])

transform_func = get_transform_from_crs(
get_crs_from_uri(DEFAULT_CRS),
pyproj.CRS.from_epsg(32735),
always_xy=False,
)
# Check that the coordinates of returned feature were transformed
assert geom_32735.equals_exact(transform_func(geom_orig), 1)


def test_get_collection_items_postgresql_automap_naming_conflicts(pg_api_):
"""
Test that PostgreSQLProvider can handle naming conflicts when automapping
classes and relationships from database schema.
"""
req = mock_api_request()
rsp_headers, code, response = get_collection_items(
pg_api_, req, 'dummy_naming_conflicts')

assert code == HTTPStatus.OK
features = json.loads(response).get('features')
assert len(features) == 0

0 comments on commit 7f17539

Please sign in to comment.