fix(ogc): isolate public and internal pygeoapi module globals

core/pygeoapi.py

@@ -1,4 +1,4 @@
-import importlib
+import importlib.util
 import os
 import re
 import sys
@@ -9,6 +9,9 @@
 import yaml
 from fastapi import FastAPI
 
+# Consumed by pygeoapi at import time only; see _load_pygeoapi_app.
+_PYGEOAPI_ENV_KEYS = ("PYGEOAPI_CONFIG", "PYGEOAPI_OPENAPI")
+
 THING_COLLECTIONS = [
     {
         "id": "water_wells",
@@ -404,8 +407,10 @@ def _assert_server_settings_match(
     public_config_path: Path, internal_config_path: Path
 ) -> None:
     # pygeoapi.api.API.__init__ mutates process-wide, module-level globals
-    # (CHARSET, FORMAT_TYPES) that persist across the importlib.reload this
-    # scheme relies on -- whichever mount is constructed last wins for both.
+    # (CHARSET, FORMAT_TYPES). Loading each mount from its own copy of
+    # pygeoapi.starlette_app does not help here, since both copies still
+    # share the one pygeoapi.api module -- whichever mount is constructed
+    # last wins for both.
     # Inert as long as both configs agree on these settings; fail loudly at
     # startup rather than let a future divergence silently corrupt responses
     # on whichever mount lost the race.
@@ -437,12 +442,37 @@ def _generate_openapi(config_path: Path, openapi_path: Path) -> None:
     openapi_path.write_text(openapi, encoding="utf-8")
 
 
-def _load_pygeoapi_app():
+def _load_pygeoapi_app(instance: str, config_path: Path, openapi_path: Path):
+    # pygeoapi.starlette_app resolves PYGEOAPI_CONFIG at import time into a
+    # module-level `api_`, and every route handler looks that name up in the
+    # module's globals at request time. importlib.reload() rebinds those
+    # globals *in place*, so reloading for the second mount retargets the
+    # handlers of the app already built for the first one -- both mounts end
+    # up serving whichever config was loaded last. Give each mount its own
+    # module object so the two sets of globals can never alias.
     module_name = "pygeoapi.starlette_app"
-    if module_name in sys.modules:
-        module = importlib.reload(sys.modules[module_name])
-    else:
-        module = importlib.import_module(module_name)
+    spec = find_spec(module_name)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to locate {module_name} for the {instance} mount.")
+
+    module = importlib.util.module_from_spec(spec)
+    # Registered before exec_module so the module can survive importing itself.
+    sys.modules[f"{module_name}__ocotillo_{instance}"] = module
+
+    previous = {key: os.environ.get(key) for key in _PYGEOAPI_ENV_KEYS}
+    os.environ["PYGEOAPI_CONFIG"] = str(config_path)
+    os.environ["PYGEOAPI_OPENAPI"] = str(openapi_path)
+    try:
+        spec.loader.exec_module(module)
+    finally:
+        # These are read only during import, so leaving the last mount's paths
+        # behind would silently decide the config for any later importer.
+        for key, value in previous.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
     return module.APP
 
 
@@ -461,10 +491,7 @@ def mount_pygeoapi(app: FastAPI) -> None:
     _write_config(config_path, server_url=_server_url(), include_edr=True)
     _generate_openapi(config_path, openapi_path)
 
-    os.environ["PYGEOAPI_CONFIG"] = str(config_path)
-    os.environ["PYGEOAPI_OPENAPI"] = str(openapi_path)
-
-    pygeoapi_app = _load_pygeoapi_app()
+    pygeoapi_app = _load_pygeoapi_app("public", config_path, openapi_path)
     mount_path = _mount_path()
     app.mount(mount_path, pygeoapi_app)
 
@@ -507,10 +534,7 @@ def mount_pygeoapi_internal(app: FastAPI) -> None:
     _generate_openapi(config_path, openapi_path)
     _assert_server_settings_match(_pygeoapi_dir() / "pygeoapi-config.yml", config_path)
 
-    os.environ["PYGEOAPI_CONFIG"] = str(config_path)
-    os.environ["PYGEOAPI_OPENAPI"] = str(openapi_path)
-
-    pygeoapi_app = _load_pygeoapi_app()
+    pygeoapi_app = _load_pygeoapi_app("internal", config_path, openapi_path)
 
     from core.internal_ogc_auth import InternalOGCAuthMiddleware
 

tests/test_pygeoapi_mount.py

@@ -1,50 +1,111 @@
-import types
+"""Isolation guarantees for the public (/ogcapi) and internal (/ogcapi-internal) mounts.
 
-from core import pygeoapi
+Both mounts are built from the same pygeoapi.starlette_app source, which
+resolves PYGEOAPI_CONFIG at import time into a module-level ``api_`` that
+every route handler reads out of module globals at request time. Loading the
+second mount by reloading that module rebinds those globals in place, which
+silently retargets the already-built first mount -- the public mount then
+serves the unfiltered ogc_internal_* views, defeating the A1
+``release_status = 'public'`` filter. These tests pin the isolation that
+prevents that.
+
+Importing this module builds the app (via the tests package), so both
+runtime config files already exist on disk by the time a test runs.
+"""
 
+import os
+import sys
+
+from core import pygeoapi
 
-def test_load_pygeoapi_app_imports_when_module_not_loaded(monkeypatch):
-    fake_module = types.SimpleNamespace(APP=object())
-    import_calls = []
+PUBLIC_MODULE = "pygeoapi.starlette_app__ocotillo_public"
+INTERNAL_MODULE = "pygeoapi.starlette_app__ocotillo_internal"
 
-    def fake_import_module(name):
-        import_calls.append(name)
-        return fake_module
 
-    monkeypatch.delitem(
-        pygeoapi.sys.modules,
-        "pygeoapi.starlette_app",
-        raising=False,
+def _mount_args():
+    public_dir = pygeoapi._pygeoapi_dir()
+    internal_dir = pygeoapi._pygeoapi_dir(
+        "PYGEOAPI_INTERNAL_RUNTIME_DIR", "/tmp/pygeoapi-internal"
     )
-    monkeypatch.setattr(
-        pygeoapi.importlib,
-        "import_module",
-        fake_import_module,
+    return (
+        (
+            "public",
+            public_dir / "pygeoapi-config.yml",
+            public_dir / "pygeoapi-openapi.yml",
+        ),
+        (
+            "internal",
+            internal_dir / "pygeoapi-config.yml",
+            internal_dir / "pygeoapi-openapi.yml",
+        ),
     )
 
-    app = pygeoapi._load_pygeoapi_app()
 
-    assert app is fake_module.APP
-    assert import_calls == ["pygeoapi.starlette_app"]
+def _load_both():
+    # Internal last, matching create_api_app's order -- the order that used
+    # to leave the public mount pointing at ogc_internal_* relations.
+    public, internal = _mount_args()
+    pygeoapi._load_pygeoapi_app(*public)
+    pygeoapi._load_pygeoapi_app(*internal)
+    return sys.modules[PUBLIC_MODULE], sys.modules[INTERNAL_MODULE]
 
 
-def test_load_pygeoapi_app_reloads_when_module_already_loaded(monkeypatch):
-    existing_module = types.SimpleNamespace(APP=object())
-    reloaded_module = types.SimpleNamespace(APP=object())
-    reload_calls = []
+def _provider_tables(api):
+    return {
+        name: resource["providers"][0].get("table")
+        for name, resource in api.config["resources"].items()
+        if resource.get("providers")
+    }
 
-    def fake_reload(module):
-        reload_calls.append(module)
-        return reloaded_module
 
-    monkeypatch.setitem(
-        pygeoapi.sys.modules,
-        "pygeoapi.starlette_app",
-        existing_module,
-    )
-    monkeypatch.setattr(pygeoapi.importlib, "reload", fake_reload)
+def test_each_mount_gets_independent_module_globals():
+    public_module, internal_module = _load_both()
+
+    assert public_module is not internal_module
+    # The aliasing that caused the leak: one shared dict, so one shared api_.
+    assert public_module.__dict__ is not internal_module.__dict__
+    assert public_module.api_ is not internal_module.api_
+
+
+def test_public_mount_does_not_resolve_to_internal_relations():
+    public_module, internal_module = _load_both()
+
+    public_tables = _provider_tables(public_module.api_)
+    assert public_tables, "public config exposed no provider-backed collections"
+    leaked = {
+        name: table
+        for name, table in public_tables.items()
+        if table and table.startswith("ogc_internal_")
+    }
+    assert not leaked, f"public mount resolves to internal relations: {leaked}"
+
+    internal_tables = _provider_tables(internal_module.api_)
+    assert internal_tables, "internal config exposed no provider-backed collections"
+    misrouted = {
+        name: table
+        for name, table in internal_tables.items()
+        if table and not table.startswith("ogc_internal_")
+    }
+    assert not misrouted, f"internal mount resolves to public relations: {misrouted}"
+
+
+def test_each_mount_advertises_its_own_server_url():
+    public_module, internal_module = _load_both()
+
+    public_url = public_module.api_.config["server"]["url"]
+    internal_url = internal_module.api_.config["server"]["url"]
+
+    assert public_url != internal_url
+    assert public_url == pygeoapi._server_url()
+    assert internal_url == pygeoapi._internal_server_url()
+
+
+def test_loading_a_mount_restores_config_env_vars():
+    # Leaving the last-loaded mount's paths in the environment would decide
+    # the config for anything that imports pygeoapi later in the process.
+    before = {key: os.environ.get(key) for key in pygeoapi._PYGEOAPI_ENV_KEYS}
 
-    app = pygeoapi._load_pygeoapi_app()
+    _load_both()
 
-    assert app is reloaded_module.APP
-    assert reload_calls == [existing_module]
+    after = {key: os.environ.get(key) for key in pygeoapi._PYGEOAPI_ENV_KEYS}
+    assert after == before