ENH: Allow OGC formats to be optionally supported

ogc/core.py

@@ -62,7 +62,11 @@ def __init__(self, layers=[], **kwargs):
             service_abstract=self.service_abstract,
             service_group_title=self.service_group_title,
         )
-        self.edr_routes = EdrRoutes(base_url=f"{self.server_address}{self.endpoint}/edr", layers=layers)
+        self.edr_routes = (
+            EdrRoutes(base_url=f"{self.server_address}{self.endpoint}/edr", layers=layers)
+            if settings.EDR_ENABLED
+            else None
+        )
 
     def get_coverage_from_id(self, identifier):
         for coverage in self.wcs_capabilities.coverages:

ogc/edr/test/conftest.py

@@ -1,3 +1,4 @@
+import os
 import pytest
 import numpy as np
 import datetime
@@ -36,6 +37,12 @@
 )
 
 
+@pytest.fixture(scope="session", autouse=True)
+def set_env_vars():
+    """Setup the environmental variables for the session to support EDR."""
+    os.environ["OGC_SUPPORTED_FORMATS"] = "edr"
+
+
 @pytest.fixture()
 def layers() -> List[pogc.Layer]:
     """List of test layers.

ogc/servers.py

@@ -34,10 +34,9 @@ def home(endpoint):
     """Example API home page. Developers should make their own page similar to this one."""
     test_layer = "testLayerName"
     test_layer_time = "12:59:59"  # HH:MM:SS
-    return f"""<h2> OGC Server API </h2>
-    <p>This is the API endpoint served at {endpoint}. Add example usage here for your users.</p>
 
-    <ul>
+    wcs_list_item = (
+        f"""
         <li> WCS: Open Geospatial Consortium (OGC) Web Coverage Service (WCS) <i>(v1.0.0)</i>
         <ul>
             <li><a href="?SERVICE=WCS&REQUEST=GetCapabilities&VERSION=1.0.0">WCS GetCapabilities (XML)</a> <i>(v1.0.0)</i></li>
@@ -46,6 +45,13 @@ def home(endpoint):
             <li><a href="?SERVICE=WCS&VERSION=1.0.0&REQUEST=GetCoverage&FORMAT=GeoTIFF&COVERAGE={test_layer_time}&BBOX=34.3952751159668,38.26394082159894,34.398660063743584,38.26779045113519&CRS=EPSG:4326&RESPONSE_CRS=EPSG:4326&WIDTH=631&HEIGHT=914&TIME=2021-03-01T12:00:00.000Z">WCS GetCoverage Example (GeoTIFF)</a> dynamic layer <i>(v1.0.0)</i></li>
         </ul>
         </li>
+    """
+        if settings.WCS_ENABLED
+        else ""
+    )
+
+    wms_list_item = (
+        f"""
         <li> WMS: Open Geospatial Consortium (OGC) Web Map Service (WMS) <i>(v1.3.0)</i>
         <ul>
             <li><a href="?SERVICE=WMS&REQUEST=GetCapabilities&VERSION=1.3.0">WMS GetCapabilities (XML)</a> <i>(v1.3.0)</i></li>
@@ -54,13 +60,32 @@ def home(endpoint):
             <li><a href="?SERVICE=WMS&VERSION=1.3.0&REQUEST=GetLegendGraphic&LAYER={test_layer}&STYLE=default&FORMAT=image/png">WMS GetLegend Example (PNG)</a> <i>(v1.3.0)</i></li>
         </ul>
         </li>
+    """
+        if settings.WMS_ENABLED
+        else ""
+    )
+
+    edr_list_item = (
+        f"""
         <li> EDR: Open Geospatial Consortium (OGC) Environmental Data Retrieval (EDR) <i>(v1.0.1)</i>
         <ul>
             <li><a href="{endpoint}/edr?f=html">EDR Landing Page (HTML)</a> <i>(v1.0.1)</i></li>
             <li><a href="{endpoint}/edr/conformance?f=json">EDR Conformance (JSON)</a> <i>(v1.0.1)</i></li>
             <li><a href="{endpoint}/edr/collections?f=json">EDR Collections (JSON)</a> <i>(v1.0.1)</i></li>
         </ul>
         </li>
+    """
+        if settings.EDR_ENABLED
+        else ""
+    )
+
+    return f"""<h2> OGC Server API </h2>
+    <p>This is the API endpoint served at {endpoint}. Add example usage here for your users.</p>
+
+    <ul>
+        {wcs_list_item}
+        {wms_list_item}
+        {edr_list_item}
     </ul>
     """
 
@@ -115,88 +140,89 @@ def method():
             self.add_url_rule(endpoint, view_func=method, methods=["GET", "POST"])  # add render method as flask route
             setattr(self, method_name, method)  # bind route function call to instance method
 
-            # Set up the EDR endpoints for the server
-            strict_slashes = get_api_rules(ogc.edr_routes.api.config).strict_slashes
-            self.add_url_rule(
-                f"/{endpoint}/edr",
-                endpoint=f"{endpoint}_landing_page",
-                view_func=self.edr_render(ogc.edr_routes.landing_page),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/static/<path:file_path>",
-                endpoint=f"{endpoint}_static_files",
-                view_func=self.edr_render(ogc.edr_routes.static_files),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/api",
-                endpoint=f"{endpoint}_api",
-                view_func=self.edr_render(ogc.edr_routes.openapi),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/openapi",
-                endpoint=f"{endpoint}_openapi",
-                view_func=self.edr_render(ogc.edr_routes.openapi),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/conformance",
-                endpoint=f"{endpoint}_conformance",
-                view_func=self.edr_render(ogc.edr_routes.conformance),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/collections",
-                endpoint=f"{endpoint}_collections",
-                view_func=self.edr_render(ogc.edr_routes.describe_collections),
-                defaults={"collection_id": None},
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/collections/<path:collection_id>",
-                endpoint=f"{endpoint}_collection",
-                view_func=self.edr_render(ogc.edr_routes.describe_collections),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/collections/<path:collection_id>/instances",
-                endpoint=f"{endpoint}_instances",
-                view_func=self.edr_render(ogc.edr_routes.describe_instances),
-                defaults={"instance_id": None},
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/collections/<path:collection_id>/instances/<path:instance_id>",
-                endpoint=f"{endpoint}_instance",
-                view_func=self.edr_render(ogc.edr_routes.describe_instances),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/collections/<path:collection_id>/<path:query_type>",
-                endpoint=f"{endpoint}_collection_query",
-                view_func=self.edr_render(ogc.edr_routes.collection_query),
-                defaults={"instance_id": None},
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
-            self.add_url_rule(
-                f"/{endpoint}/edr/collections/<path:collection_id>/instances/<path:instance_id>/<path:query_type>",
-                endpoint=f"{endpoint}_instance_query",
-                view_func=self.edr_render(ogc.edr_routes.collection_query),
-                methods=["GET"],
-                strict_slashes=strict_slashes,
-            )
+            # Set up the EDR endpoints for the server if routes are available
+            if settings.EDR_ENABLED:
+                strict_slashes = get_api_rules(ogc.edr_routes.api.config).strict_slashes
+                self.add_url_rule(
+                    f"/{endpoint}/edr",
+                    endpoint=f"{endpoint}_landing_page",
+                    view_func=self.edr_render(ogc.edr_routes.landing_page),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/static/<path:file_path>",
+                    endpoint=f"{endpoint}_static_files",
+                    view_func=self.edr_render(ogc.edr_routes.static_files),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/api",
+                    endpoint=f"{endpoint}_api",
+                    view_func=self.edr_render(ogc.edr_routes.openapi),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/openapi",
+                    endpoint=f"{endpoint}_openapi",
+                    view_func=self.edr_render(ogc.edr_routes.openapi),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/conformance",
+                    endpoint=f"{endpoint}_conformance",
+                    view_func=self.edr_render(ogc.edr_routes.conformance),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/collections",
+                    endpoint=f"{endpoint}_collections",
+                    view_func=self.edr_render(ogc.edr_routes.describe_collections),
+                    defaults={"collection_id": None},
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/collections/<path:collection_id>",
+                    endpoint=f"{endpoint}_collection",
+                    view_func=self.edr_render(ogc.edr_routes.describe_collections),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/collections/<path:collection_id>/instances",
+                    endpoint=f"{endpoint}_instances",
+                    view_func=self.edr_render(ogc.edr_routes.describe_instances),
+                    defaults={"instance_id": None},
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/collections/<path:collection_id>/instances/<path:instance_id>",
+                    endpoint=f"{endpoint}_instance",
+                    view_func=self.edr_render(ogc.edr_routes.describe_instances),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/collections/<path:collection_id>/<path:query_type>",
+                    endpoint=f"{endpoint}_collection_query",
+                    view_func=self.edr_render(ogc.edr_routes.collection_query),
+                    defaults={"instance_id": None},
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
+                self.add_url_rule(
+                    f"/{endpoint}/edr/collections/<path:collection_id>/instances/<path:instance_id>/<path:query_type>",
+                    endpoint=f"{endpoint}_instance_query",
+                    view_func=self.edr_render(ogc.edr_routes.collection_query),
+                    methods=["GET"],
+                    strict_slashes=strict_slashes,
+                )
 
     def ogc_render(self, ogc_idx):
         logger.info("OGC server.ogc_render %i", ogc_idx)
@@ -230,9 +256,9 @@ def ogc_render(self, ogc_idx):
             else:
                 args["base_url"] = None
             ogc_response = None
-            if args["service"].lower() == "wcs":
+            if args["service"].lower() == "wcs" and settings.WCS_ENABLED:
                 ogc_response = ogc.handle_wcs_kv(args)
-            elif args["service"].lower() == "wms":
+            elif args["service"].lower() == "wms" and settings.WMS_ENABLED:
                 ogc_response = ogc.handle_wms_kv(args)
             if ogc_response is not None:
                 if isinstance(ogc_response, six.string_types):

ogc/settings.py

@@ -87,3 +87,9 @@
     EDR_CONFIGURATION_PATH = os.environ["EDR_CONFIGURATION_PATH"]
 except Exception:
     EDR_CONFIGURATION_PATH = None
+
+# get supported formats
+OGC_SUPPORTED_FORMATS = os.environ.get("OGC_SUPPORTED_FORMATS", "wms,wcs")
+WMS_ENABLED = "wms" in OGC_SUPPORTED_FORMATS.lower()
+WCS_ENABLED = "wcs" in OGC_SUPPORTED_FORMATS.lower()
+EDR_ENABLED = "edr" in OGC_SUPPORTED_FORMATS.lower()

ogc/test/test_servers.py

@@ -1,12 +1,40 @@
 from ogc import servers
 from ogc import core
 from ogc import podpac as pogc
+from ogc import settings
+from unittest.mock import patch
+from typing import Callable
 
+import importlib
 import podpac
 import pytest
 import numpy as np
 
 
+@pytest.fixture
+def supported_formats() -> Callable[[str], None]:
+    """Fixture used to patch OGC supported formats.
+
+    Returns
+    -------
+    Callable[[str], None]
+        A function which patches the OGC supported formats based on input string.
+    """
+
+    def _supported_formats(formats: str):
+        """Patch the supported formats setting.
+
+        Parameters
+        ----------
+        formats : str
+            The formats which should be supported by the server as a string.
+        """
+        with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": formats}):
+            importlib.reload(settings)
+
+    return _supported_formats
+
+
 @pytest.fixture
 def client():
     """
@@ -96,3 +124,49 @@ def test_server_ogc_render_invalid_request(client):
     """
     response = client.get("/ogc?service=WCS&request=InvalidRequest")
     assert response.status_code == 400
+
+
+def test_server_with_default_supported_services(client):
+    """
+    Test the server with the default supported services.
+    """
+    response = client.get("/ogc?service=WMS&request=GetCapabilities")
+    assert response.status_code == 200
+
+    response = client.get("/ogc?service=WCS&request=GetCapabilities")
+    assert response.status_code == 200
+
+    response = client.get("/ogc/edr")
+    assert response.status_code == 404
+
+
+def test_server_without_wcs_supported_service(supported_formats, client):
+    """
+    Test the WCS service is unavailable when WCS is not a supported format.
+    """
+    supported_formats("wms")
+
+    response = client.get("/ogc?service=WMS&request=GetCapabilities")
+    assert response.status_code == 200
+
+    response = client.get("/ogc?service=WCS&request=GetCapabilities")
+    assert response.status_code == 400
+
+    response = client.get("/ogc/edr")
+    assert response.status_code == 404
+
+
+def test_server_without_wms_supported_service(supported_formats, client):
+    """
+    Test the WMS service is unavailable when WMS is not a supported format.
+    """
+    supported_formats("wcs")
+
+    response = client.get("/ogc?service=WCS&request=GetCapabilities")
+    assert response.status_code == 200
+
+    response = client.get("/ogc?service=WMS&request=GetCapabilities")
+    assert response.status_code == 400
+
+    response = client.get("/ogc/edr")
+    assert response.status_code == 404