diff --git a/tests/test_api.py b/tests/test_api.py index 9755b43..eed75c7 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -78,14 +78,6 @@ def api_hidden_resources(config_hidden_resources): return API(config_hidden_resources) -# API using PostgreSQL provider -@pytest.fixture() -def pg_api_(): - with open(get_test_file_path('pygeoapi-test-config-postgresql.yml')) as fh: - config = yaml_load(fh) - return API(config) - - def test_apirequest(api_): # Test without (valid) locales with pytest.raises(ValueError): @@ -868,180 +860,6 @@ def test_get_collection_items(config, api_): assert code == HTTPStatus.BAD_REQUEST -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_request({ - 'filter-lang': 'cql-text', - 'filter': cql_query - }) - rsp_headers, code, response = pg_api_.get_collection_items( - 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_request({ - 'filter': cql_query - }) - rsp_headers, code, response = pg_api_.get_collection_items( - 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_request({ - 'filter-lang': 'cql-json', # Only cql-text is valid for GET - 'filter': cql_query - }) - rsp_headers, code, response = pg_api_.get_collection_items( - 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_request({ - 'filter': bad_cql - }) - rsp_headers, code, response = pg_api_.get_collection_items( - 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_request({ - 'filter-lang': 'cql-json' - }, data=cql, **headers) - rsp_headers, code, response = pg_api_.post_collection_items( - 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_request({ - 'filter-lang': 'cql-text' # Only cql-json is valid for POST - }, data=cql, **headers) - rsp_headers, code, response = pg_api_.post_collection_items( - 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_request({ - 'filter-lang': 'cql-json' - }, data=bad_cql, **headers) - rsp_headers, code, response = pg_api_.post_collection_items( - 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_json_ld(config, api_): req = mock_request({ 'f': 'jsonld', diff --git a/tests/test_postgresql_provider.py b/tests/test_postgresql_provider.py index 6a68c15..3ff6c1f 100644 --- a/tests/test_postgresql_provider.py +++ b/tests/test_postgresql_provider.py @@ -36,10 +36,15 @@ # See pygeoapi/provider/postgresql.py for instructions on setting up # test database in Docker +import os +import json import pytest +from http import HTTPStatus from pygeofilter.parsers.ecql import parse +from pygeoapi.api import API + from pygeoapi.provider.base import ( ProviderConnectionError, ProviderItemNotFoundError, @@ -48,7 +53,10 @@ from pygeoapi.provider.base import ( from pygeoapi.provider.postgresql import PostgreSQLProvider import pygeoapi.provider.postgresql as postgresql_provider_module -import os +from pygeoapi.util import yaml_load + +from .util import get_test_file_path, mock_request + PASSWORD = os.environ.get('POSTGRESQL_PASSWORD', 'postgres') @@ -69,6 +77,14 @@ def config(): } +# API using PostgreSQL provider +@pytest.fixture() +def pg_api_(): + with open(get_test_file_path('pygeoapi-test-config-postgresql.yml')) as fh: + config = yaml_load(fh) + return API(config) + + def test_query(config): """Testing query for a valid JSON object with geometry""" p = PostgreSQLProvider(config) @@ -361,3 +377,178 @@ 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_request({ + 'filter-lang': 'cql-text', + 'filter': cql_query + }) + rsp_headers, code, response = pg_api_.get_collection_items( + 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_request({ + 'filter': cql_query + }) + rsp_headers, code, response = pg_api_.get_collection_items( + 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_request({ + 'filter-lang': 'cql-json', # Only cql-text is valid for GET + 'filter': cql_query + }) + rsp_headers, code, response = pg_api_.get_collection_items( + 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_request({ + 'filter': bad_cql + }) + rsp_headers, code, response = pg_api_.get_collection_items( + 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_request({ + 'filter-lang': 'cql-json' + }, data=cql, **headers) + rsp_headers, code, response = pg_api_.post_collection_items( + 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_request({ + 'filter-lang': 'cql-text' # Only cql-json is valid for POST + }, data=cql, **headers) + rsp_headers, code, response = pg_api_.post_collection_items( + 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_request({ + 'filter-lang': 'cql-json' + }, data=bad_cql, **headers) + rsp_headers, code, response = pg_api_.post_collection_items( + 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')