mv to pytest_simcore:

Author: pcrespov

packages/pytest-simcore/src/pytest_simcore/openapi_specs.py

@@ -0,0 +1,66 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+# pylint: disable=too-many-arguments
+
+from copy import deepcopy
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, NamedTuple
+
+import jsonref
+import pytest
+import yaml
+
+
+class Entrypoint(NamedTuple):
+    name: str
+    method: str
+    path: str
+
+
+@lru_cache  # ANE: required to boost tests speed, gains 3.5s per test
+def _load(file: Path, base_uri: str = "") -> dict:
+    # SEE https://jsonref.readthedocs.io/en/latest/#lazy-load-and-load-on-repr
+    data: dict = jsonref.replace_refs(  # type: ignore
+        yaml.safe_load(file.read_text()),
+        base_uri=base_uri,
+        lazy_load=True,  # this data will be iterated
+        merge_props=False,
+    )
+    return data
+
+
+@pytest.fixture
+def openapi_specs_path() -> Path:
+    pytest.fail(reason="Must be overriden in caller tests package")
+
+
+@pytest.fixture
+def openapi_specs(openapi_specs_path: Path) -> dict[str, Any]:
+    assert openapi_specs_path.is_file()
+    # TODO: If openapi_specs_path is a url, download in tmp_dir and get path
+
+    openapi: dict[str, Any] = _load(
+        openapi_specs_path, base_uri=openapi_specs_path.as_uri()
+    )
+    return deepcopy(openapi)
+
+
+@pytest.fixture
+def openapi_specs_entrypoints(
+    openapi_specs: dict,
+) -> set[Entrypoint]:
+    entrypoints: set[Entrypoint] = set()
+
+    # openapi-specifications, i.e. "contract"
+    for path, path_obj in openapi_specs["paths"].items():
+        for operation, operation_obj in path_obj.items():
+            entrypoints.add(
+                Entrypoint(
+                    method=operation.upper(),
+                    path=path,
+                    name=operation_obj["operationId"],
+                )
+            )
+    return entrypoints

services/web/server/tests/conftest.py

@@ -75,6 +75,7 @@
     "pytest_simcore.simcore_service_library_fixtures",
     "pytest_simcore.simcore_services",
     "pytest_simcore.socketio_client",
+    "pytest_simcore.openapi_specs",
 ]
 
 

services/web/server/tests/unit/conftest.py

@@ -8,12 +8,10 @@
 import json
 import sys
 from collections.abc import Callable, Iterable
-from functools import lru_cache
 from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock
 
-import jsonref
 import pytest
 import yaml
 from pytest_simcore.helpers.dict_tools import ConfigDict
@@ -81,23 +79,7 @@ def disable_gc_manual_guest_users(mocker):
     )
 
 
-@lru_cache  # ANE: required to boost tests speed, gains 3.5s per test
-def _load_openapi_specs(spec_path: Path | None = None) -> dict:
-    if spec_path is None:
-        spec_path = get_openapi_specs_path()
-
-    with spec_path.open() as fh:
-        spec_dict = yaml.safe_load(fh)
-
-    # SEE https://jsonref.readthedocs.io/en/latest/#lazy-load-and-load-on-repr
-    openapi: dict = jsonref.replace_refs(
-        spec_dict, base_uri=spec_path.as_uri(), lazy_load=True, merge_props=False
-    )
-    return openapi
-
-
 @pytest.fixture
-def openapi_specs(api_version_prefix) -> dict:
-
-    spec_path = get_openapi_specs_path(api_version_prefix)
-    return _load_openapi_specs(spec_path)
+def openapi_specs_path(api_version_prefix: str) -> Path:
+    # overrides pytest_simcore.openapi_specs.app_openapi_specs_path fixture
+    return get_openapi_specs_path(api_version_prefix)

services/web/server/tests/unit/with_dbs/03/test__openapi_specs.py

@@ -3,23 +3,17 @@
 # pylint: disable=unused-argument
 # pylint: disable=unused-variable
 
-from typing import NamedTuple
 
 import pytest
 from aiohttp import web
 from faker import Faker
 from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
 from pytest_simcore.helpers.typing_env import EnvVarsDict
+from pytest_simcore.openapi_specs import Entrypoint
 from simcore_service_webserver.application import create_application
 from simcore_service_webserver.application_settings import get_application_settings
 
 
-class Entrypoint(NamedTuple):
-    name: str
-    method: str
-    path: str
-
-
 @pytest.fixture
 def app_environment(
     mock_env_devel_environment: EnvVarsDict,
@@ -60,24 +54,7 @@ def app(app_environment: EnvVarsDict) -> web.Application:
 
 
 @pytest.fixture
-def expected_openapi_entrypoints(openapi_specs: dict) -> set[Entrypoint]:
-    entrypoints: set[Entrypoint] = set()
-
-    # openapi-specifications, i.e. "contract"
-    for path, path_obj in openapi_specs["paths"].items():
-        for operation, operation_obj in path_obj.items():
-            entrypoints.add(
-                Entrypoint(
-                    method=operation.upper(),
-                    path=path,
-                    name=operation_obj["operationId"],
-                )
-            )
-    return entrypoints
-
-
-@pytest.fixture
-def app_entrypoints(app: web.Application) -> set[Entrypoint]:
+def app_rest_entrypoints(app: web.Application) -> set[Entrypoint]:
     entrypoints: set[Entrypoint] = set()
 
     # app routes, i.e. "exposed"
@@ -102,12 +79,12 @@ def app_entrypoints(app: web.Application) -> set[Entrypoint]:
 
 
 def test_app_named_resources_against_openapi_specs(
-    expected_openapi_entrypoints: set[Entrypoint],
-    app_entrypoints: set[Entrypoint],
+    openapi_specs_entrypoints: set[Entrypoint],
+    app_rest_entrypoints: set[Entrypoint],
 ):
     # check whether exposed routes implements openapi.json contract
 
-    assert app_entrypoints == expected_openapi_entrypoints
+    assert app_rest_entrypoints == openapi_specs_entrypoints
 
     # NOTE: missing here is:
     # - input schemas (path, query and body)