Move PostgreSQL API tests out of test_api.py (#1160)
Moved Postgresql tests from `test_api.py` to `test_postgres_provider.py`.
This commit is contained in:
Just van den Broecke
committed by
GitHub
GitHub
parent
38299e2dcc
commit
57cfbb5e3f
@@ -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',
|
||||
|
||||
@@ -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')
|
||||
|
||||
Reference in New Issue
Block a user