Add Starlette mock client and API rules test (#1134)

pygeoapi/starlette_app.py

@@ -43,6 +43,7 @@
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.responses import Response, JSONResponse, HTMLResponse
+from starlette.types import ASGIApp, Scope, Send, Receive
 import uvicorn
 
 from pygeoapi.api import API
@@ -56,7 +57,7 @@
 
 p = Path(__file__)
 
-app = Starlette(debug=True)
+APP = Starlette(debug=True)
 STATIC_DIR = Path(p).parent.resolve() / 'static'
 
 try:
@@ -468,6 +469,28 @@ async def stac_catalog_path(request: Request):
     return get_response(api_.get_stac_path(request, path))
 
 
+class ApiRulesMiddleware:
+    """ Custom middleware to properly deal with trailing slashes.
+    See https://github.com/encode/starlette/issues/869.
+    """
+    def __init__(
+            self,
+            app: ASGIApp
+    ) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope,
+                       receive: Receive, send: Send) -> None:
+        path = scope["path"]
+        if scope["type"] == "http" and API_RULES.strict_slashes \
+                and path != '/' and path.endswith('/'):
+            response = Response(status_code=404)
+            await response(scope, receive, send)
+            return
+
+        await self.app(scope, receive, send)
+
+
 api_routes = [
     Route('/', landing_page),
     Route('/openapi', openapi),
@@ -510,18 +533,22 @@ async def stac_catalog_path(request: Request):
 ]
 
 url_prefix = API_RULES.get_version_prefix('starlette')
-app = Starlette(
+APP = Starlette(
     routes=[
         Mount(f'{url_prefix}/static', StaticFiles(directory=STATIC_DIR)),
-        Mount(url_prefix, routes=api_routes) if url_prefix else api_routes
+        Mount(url_prefix, routes=api_routes)
     ]
 )
-app.router.redirect_slashes = not API_RULES.strict_slashes
+
+# If API rules require strict slashes, do not redirect
+if API_RULES.strict_slashes:
+    APP.router.redirect_slashes = False
+    APP.add_middleware(ApiRulesMiddleware)
 
 # CORS: optionally enable from config.
 if CONFIG['server'].get('cors', False):
     from starlette.middleware.cors import CORSMiddleware
-    app.add_middleware(CORSMiddleware, allow_origins=['*'])
+    APP.add_middleware(CORSMiddleware, allow_origins=['*'])
 
 try:
     OGC_SCHEMAS_LOCATION = Path(CONFIG['server']['ogc_schemas_location'])
@@ -532,7 +559,7 @@ async def stac_catalog_path(request: Request):
         not OGC_SCHEMAS_LOCATION.name.startswith('http')):
     if not OGC_SCHEMAS_LOCATION.exists():
         raise RuntimeError('OGC schemas misconfigured')
-    app.mount(
+    APP.mount(
         f'{url_prefix}/schemas', StaticFiles(directory=OGC_SCHEMAS_LOCATION)
     )
 

requirements-dev.txt

@@ -9,7 +9,11 @@ pytest
 pytest-cov
 pytest-env
 coverage
-pyld
+
+# Testing with mock Starlette client
+starlette
+uvicorn[standard]
+httpx
 
 # PEP8
 flake8

tests/pygeoapi-test-config-apirules.yml

@@ -33,7 +33,7 @@ server:
     bind:
         host: 0.0.0.0
         port: 5000
-    url: http://localhost:5000
+    url: http://localhost:5000/api
     mimetype: application/json; charset=UTF-8
     encoding: utf-8
     gzip: false

tests/test_api.py

@@ -38,13 +38,13 @@
 
 from pyld import jsonld
 import pytest
+
 from pygeoapi.api import (
     API, APIRequest, FORMAT_TYPES, validate_bbox, validate_datetime,
     validate_subset, F_HTML, F_JSON, F_JSONLD, F_GZIP, __version__
 )
 from pygeoapi.util import yaml_load, get_api_rules
-
-from .util import get_test_file_path, mock_request, mock_client
+from .util import get_test_file_path, mock_request, mock_flask, mock_starlette
 
 LOGGER = logging.getLogger(__name__)
 
@@ -235,38 +235,72 @@ def test_apirequest(api_):
 
 def test_apirules_active(config_with_rules, rules_api):
     assert rules_api.config == config_with_rules
-    flask_prefix = get_api_rules(config_with_rules).get_version_prefix('flask')
+    rules = get_api_rules(config_with_rules)
 
-    with mock_client('pygeoapi-test-config-apirules.yml') as flask_mock:
+    # Test Flask
+    flask_prefix = rules.get_version_prefix('flask')
+    with mock_flask('pygeoapi-test-config-apirules.yml') as flask_client:
         # Test happy path
-        response = flask_mock.get(f'{flask_prefix}/conformance')
+        response = flask_client.get(f'{flask_prefix}/conformance')
         assert response.status_code == 200
         assert response.headers['X-API-Version'] == __version__
         assert response.request.url == \
-               flask_mock.application.url_for('pygeoapi.conformance')
-        response = flask_mock.get(f'{flask_prefix}/static/img/pygeoapi.png')
+               flask_client.application.url_for('pygeoapi.conformance')
+        response = flask_client.get(f'{flask_prefix}/static/img/pygeoapi.png')
+        assert response.status_code == 200
+
+        # Test strict slashes
+        response = flask_client.get(f'{flask_prefix}/conformance/')
+        assert response.status_code == 404
+
+    # Test Starlette
+    starlette_prefix = rules.get_version_prefix('starlette')
+    with mock_starlette('pygeoapi-test-config-apirules.yml') as starlette_client:  # noqa
+        # Test happy path
+        response = starlette_client.get(f'{starlette_prefix}/conformance')
+        assert response.status_code == 200
+        assert response.headers['X-API-Version'] == __version__
+        response = starlette_client.get(f'{starlette_prefix}/static/img/pygeoapi.png')  # noqa
         assert response.status_code == 200
 
         # Test strict slashes
-        response = flask_mock.get(f'{flask_prefix}/conformance/')
+        response = starlette_client.get(f'{starlette_prefix}/conformance/')
         assert response.status_code == 404
 
 
 def test_apirules_inactive(config, api_):
     assert api_.config == config
-    flask_prefix = get_api_rules(config).get_version_prefix('flask')
+    rules = get_api_rules(config)
+
+    # Test Flask
+    flask_prefix = rules.get_version_prefix('flask')
     assert flask_prefix == ''
+    with mock_flask('pygeoapi-test-config.yml') as flask_client:
+        # Test happy path
+        response = flask_client.get('/conformance')
+        assert response.status_code == 200
+        assert 'X-API-Version' not in response.headers
+        response = flask_client.get('/static/img/pygeoapi.png')
+        assert response.status_code == 200
+
+        # Test slash redirect
+        response = flask_client.get('/conformance/')
+        assert response.status_code != 404
+        assert 'X-API-Version' not in response.headers
 
-    with mock_client('pygeoapi-test-config.yml') as flask_mock:
+    # Test Starlette
+    starlette_prefix = rules.get_version_prefix('starlette')
+    assert starlette_prefix == ''
+    with mock_starlette('pygeoapi-test-config.yml') as starlette_client:
         # Test happy path
-        response = flask_mock.get('/conformance')
+        response = starlette_client.get('/conformance')
         assert response.status_code == 200
         assert 'X-API-Version' not in response.headers
-        response = flask_mock.get('/static/img/pygeoapi.png')
+        response = starlette_client.get('/static/img/pygeoapi.png')
         assert response.status_code == 200
 
         # Test slash redirect
-        response = flask_mock.get('/conformance/')
+        response = starlette_client.get('/conformance/')
         assert response.status_code != 404
         assert 'X-API-Version' not in response.headers
 

tests/util.py

@@ -30,10 +30,12 @@
 import sys
 import logging
 import os.path
+from urllib.parse import urlsplit, urljoin
 from importlib import reload
 from contextlib import contextmanager
 
 from flask.testing import FlaskClient
+from starlette.testclient import TestClient as StarletteClient
 from werkzeug.test import create_environ
 from werkzeug.wrappers import Request
 from werkzeug.datastructures import ImmutableMultiDict
@@ -76,14 +78,12 @@ def mock_request(params: dict = None, data=None, **headers) -> Request:
 
 
 @contextmanager
-def mock_client(config_file: str = 'pygeoapi-test-config.yml',
-                use_cookies: bool = True, **kwargs) -> FlaskClient:
+def mock_flask(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> FlaskClient:  # noqa
     """
     Mocks a Flask client so we can test the API routing with applied API rules.
 
     :param config_file: Optional configuration YAML file to use.
                         If not set, the default test configuration is used.
-    :param use_cookies: Whether the mock Flask client should use cookies.
     """
     flask_app = None
     env_conf = os.getenv('PYGEOAPI_CONFIG')
@@ -97,17 +97,74 @@ def mock_client(config_file: str = 'pygeoapi-test-config.yml',
         # Force a module reload to make sure we really use another config
         reload(flask_app)
 
-        client = flask_app.APP.test_client(use_cookies, **kwargs)
+        # Set server root path
+        url_parts = urlsplit(flask_app.CONFIG['server']['url'])
+        app_root = url_parts.path.rstrip('/') or '/'
+        flask_app.APP.config['SERVER_NAME'] = url_parts.netloc
+        flask_app.APP.config['APPLICATION_ROOT'] = app_root
+
+        # Create and return test client
+        client = flask_app.APP.test_client(**kwargs)
         yield client
+
     finally:
         if env_conf is None:
             # Remove env variable again if it was not set initially
             del os.environ['PYGEOAPI_CONFIG']
-            # "Un-import" Flask app module
+            # Unload Flask app module
             del sys.modules['pygeoapi.flask_app']
         else:
             # Restore env variable to its original value and reload Flask app
             os.environ['PYGEOAPI_CONFIG'] = env_conf
             if flask_app:
                 reload(flask_app)
         del client
+
+
+@contextmanager
+def mock_starlette(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> StarletteClient:  # noqa
+    """
+    Mocks a Starlette client so we can test the API routing with applied
+    API rules.
+
+    :param config_file: Optional configuration YAML file to use.
+                        If not set, the default test configuration is used.
+    """
+    starlette_app = None
+    env_conf = os.getenv('PYGEOAPI_CONFIG')
+    try:
+        # Temporarily override environment variable to import Starlette app
+        os.environ['PYGEOAPI_CONFIG'] = get_test_file_path(config_file)
+
+        # Import current pygeoapi Starlette app module
+        from pygeoapi import starlette_app
+
+        # Force a module reload to make sure we really use another config
+        reload(starlette_app)
+
+        # Get server root path
+        base_url = starlette_app.CONFIG['server']['url']
+        root_path = urlsplit(base_url).path.rstrip('/') or '/'
+
+        # Create and return test client
+        # Note: the 'root_path' argument does not work as expected,
+        # so we join the base URL and the root path.
+        client = StarletteClient(
+            starlette_app.APP,
+            urljoin(base_url, root_path),
+            **kwargs
+        )
+        yield client
+
+    finally:
+        if env_conf is None:
+            # Remove env variable again if it was not set initially
+            del os.environ['PYGEOAPI_CONFIG']
+            # Unload Starlette app module
+            del sys.modules['pygeoapi.starlette_app']
+        else:
+            # Restore env variable to original value and reload Starlette app
+            os.environ['PYGEOAPI_CONFIG'] = env_conf
+            if starlette_app:
+                reload(starlette_app)
+        del client