diff --git a/tests/test_postgresql_provider.py b/tests/test_postgresql_provider.py index a067979..a586336 100644 --- a/tests/test_postgresql_provider.py +++ b/tests/test_postgresql_provider.py @@ -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, @@ -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' @@ -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()) @@ -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