extract models from openapi specs

Author: bisgaard-itis

clients/python/test/test_osparc/conftest.py

@@ -10,6 +10,11 @@
 from urllib3 import HTTPResponse
 from pydantic import BaseModel
 from typing import Callable, Generator
+from prance import ResolvingParser
+import json
+from tempfile import NamedTemporaryFile
+from pathlib import Path
+from typing import Any, Type, TypeVar
 
 
 @pytest.fixture
@@ -21,6 +26,16 @@ def cfg(faker: Faker) -> osparc.Configuration:
     )
 
 
+@pytest.fixture
+def osparc_openapi_specs() -> Generator[dict[str, Any], None, None]:
+    with NamedTemporaryFile(suffix=".json") as file:
+        file = Path(file.name)
+        file.write_text(json.dumps(osparc.openapi()))
+        osparc_spec = ResolvingParser(f"{file.resolve()}").specification
+    assert osparc_spec is not None
+    yield osparc_spec
+
+
 @pytest.fixture
 def api_client(cfg: osparc.Configuration) -> osparc.ApiClient:
     return osparc.ApiClient(configuration=cfg)
@@ -55,53 +70,20 @@ def _sideeffect(
     yield _mock_server
 
 
-@pytest.fixture
-def job_metadata_update(faker: Faker):
-    return osparc.JobMetadataUpdate(
-        metadata={
-            "boolean": faker.boolean(),
-            "float": faker.pyfloat(),
-            "int": faker.pyint(),
-            "str": faker.text(),
-            "None": None,
-        }
-    )
-
-
-@pytest.fixture
-def job_metadata(
-    faker: Faker, job_metadata_update: osparc.JobMetadataUpdate
-) -> osparc.JobMetadata:
-    return osparc.JobMetadata(
-        job_id=f"{faker.uuid4()}",
-        metadata=job_metadata_update.metadata,
-        url=faker.url(),
-    )
+T = TypeVar("T", bound=BaseModel)
 
 
 @pytest.fixture
-def job_inputs(faker: Faker) -> osparc.JobInputs:
-    return osparc.JobInputs(
-        {
-            "File": osparc.File(id=f"{faker.uuid4()}", filename=faker.file_name()),
-            "bool": faker.boolean(),
-            "float": faker.pyfloat(),
-            "int": faker.pyint(),
-            "str": faker.text(),
-            "None": None,
-        }
-    )
-
-
-@pytest.fixture
-def job(faker: Faker) -> osparc.Job:
-    return osparc.Job(
-        id=f"{faker.uuid4()}",
-        name=faker.file_name(),
-        inputs_checksum=f"{faker.sha256()}",
-        created_at=faker.date_time(),
-        runner_name="runner1",  # must validate regexp, hence hardcoded
-        url=None,
-        runner_url=None,
-        outputs_url=None,
-    )
+def create_osparc_response_model(
+    osparc_openapi_specs: dict[str, Any],
+) -> Generator[Callable[[Type[T]], T], None, None]:
+    def _create_model(model_type: Type[T]) -> T:
+        schemas = osparc_openapi_specs.get("components", {}).get("schemas", {})
+        example_data = schemas.get(model_type.__name__, {}).get("example", {})
+        assert example_data, (
+            "Could not extract example data for",
+            " '{model_type.__name__}' from openapi specs",
+        )
+        return model_type.model_validate(example_data)
+
+    yield _create_model

clients/python/test/test_osparc/test_solvers_api.py

@@ -3,6 +3,9 @@
 from typing import Callable, Generator
 from pydantic import BaseModel
 import pytest
+from typing import Type, TypeVar
+
+T = TypeVar("T", bound=BaseModel)
 
 
 @pytest.fixture
@@ -12,10 +15,12 @@ def solvers_api(api_client: ApiClient) -> Generator[SolversApi, None, None]:
 
 def test_create_job(
     create_server_mock: Callable[[int, BaseModel], None],
-    job_inputs: JobInputs,
-    job: Job,
+    create_osparc_response_model: Callable,
     solvers_api: SolversApi,
 ):
+    job_inputs = create_osparc_response_model(JobInputs)
+    job = create_osparc_response_model(Job)
+
     create_server_mock(201, job)
 
     _job = solvers_api.create_job(
@@ -26,10 +31,11 @@ def test_create_job(
 
 def test_get_job_custom_metadata(
     create_server_mock: Callable[[int, BaseModel], None],
-    job_metadata: JobMetadata,
+    create_osparc_response_model: Callable[[Type[JobMetadata]], JobMetadata],
     faker: Faker,
     solvers_api: SolversApi,
 ):
+    job_metadata = create_osparc_response_model(JobMetadata)
     create_server_mock(200, job_metadata)
 
     metadata = solvers_api.get_job_custom_metadata(
@@ -40,11 +46,12 @@ def test_get_job_custom_metadata(
 
 def test_replace_job_custom_metadata(
     create_server_mock: Callable[[int, BaseModel], None],
-    job_metadata: JobMetadata,
-    job_metadata_update: JobMetadataUpdate,
+    create_osparc_response_model: Callable,
     solvers_api: SolversApi,
     faker: Faker,
 ):
+    job_metadata = create_osparc_response_model(JobMetadata)
+    job_metadata_update = create_osparc_response_model(JobMetadataUpdate)
     create_server_mock(200, job_metadata)
 
     _job_metadata = solvers_api.replace_job_custom_metadata(