Merge branch 'main' into scc

Author: dbasunag

pyproject.toml

@@ -50,7 +50,7 @@ dependencies = [
     "pyyaml",
     "tenacity",
     "types-requests>=2.32.0.20241016",
-    "schemathesis>=3.38.10",
+    "schemathesis>=4.0.0a8",
     "requests",
     "pytest-asyncio",
     "syrupy",

pytest.ini

@@ -15,6 +15,7 @@ markers =
     slow: Mark tests which take more than 10 minutes as slow tests.
     pre_upgrade: Mark tests which should be run before upgrading the product.
     post_upgrade: Mark tests which should be run after upgrading the product.
+    fuzzer: Mark tests that use fuzzing and are probably going to generate unanticipated failures.
 
     # Model server
     modelmesh: Mark tests which are model mesh tests

tests/model_registry/conftest.py

@@ -12,6 +12,11 @@
 from ocp_resources.deployment import Deployment
 
 from ocp_resources.model_registry import ModelRegistry
+import schemathesis.schemas
+from schemathesis.specs.openapi.schemas import BaseOpenAPISchema
+from schemathesis.generation.stateful.state_machine import APIStateMachine
+from schemathesis.core.transport import Response
+from schemathesis.generation.case import Case
 from ocp_resources.resource import ResourceEditor
 
 from pytest import FixtureRequest
@@ -44,6 +49,10 @@
 
 @pytest.fixture(scope="class")
 def model_registry_namespace(request: FixtureRequest, admin_client: DynamicClient) -> Generator[Namespace, Any, Any]:
+    # TODO: model_registry_namespace fixture should basically be doing this 1) check the ns exists, 2) it is in ACTIVE
+    # state and return. But it should not create the ns. If DSC manages registriesNamespace, and namespace was not
+    # created when mr is updated to Managed state, it would be a bug and we should catch it
+    # To be handled in upcoming PR.
     with create_ns(
         name=request.param.get("namespace_name", MR_NAMESPACE),
         admin_client=admin_client,
@@ -301,11 +310,32 @@ def model_registry_instance_rest_endpoint(
 
 
 @pytest.fixture(scope="class")
-def generated_schema(model_registry_instance_rest_endpoint: str) -> Any:
-    return schemathesis.from_uri(
-        uri="https://raw.githubusercontent.com/kubeflow/model-registry/main/api/openapi/model-registry.yaml",
-        base_url=f"https://{model_registry_instance_rest_endpoint}/",
+def generated_schema(model_registry_instance_rest_endpoint: str) -> BaseOpenAPISchema:
+    schema = schemathesis.openapi.from_url(
+        url="https://raw.githubusercontent.com/kubeflow/model-registry/main/api/openapi/model-registry.yaml"
     )
+    schema.configure(base_url=f"https://{model_registry_instance_rest_endpoint}/")
+    return schema
+
+
+@pytest.fixture
+def state_machine(generated_schema: BaseOpenAPISchema, current_client_token: str) -> APIStateMachine:
+    BaseAPIWorkflow = generated_schema.as_state_machine()
+
+    class APIWorkflow(BaseAPIWorkflow):  # type: ignore
+        headers: dict[str, str]
+
+        def setup(self) -> None:
+            self.headers = {"Authorization": f"Bearer {current_client_token}", "Content-Type": "application/json"}
+
+        # these kwargs are passed to requests.request()
+        def get_call_kwargs(self, case: Case) -> dict[str, Any]:
+            return {"verify": False, "headers": self.headers}
+
+        def after_call(self, response: Response, case: Case) -> None:
+            LOGGER.info(f"{case.method} {case.path} -> {response.status_code}")
+
+    return APIWorkflow
 
 
 @pytest.fixture(scope="class")

tests/model_registry/test_rest_api.py

@@ -4,15 +4,13 @@
 
 LOGGER = get_logger(name=__name__)
 
-schema = schemathesis.from_pytest_fixture("generated_schema")
+schema = schemathesis.pytest.from_fixture("generated_schema")
 
 
-# TODO: This is a Stateless test due to how the openAPI spec is currently defined in upstream.
-# Once it is updated to support Stateful testing of the API we can enable this to run every time,
-# but for now having it run manually to check the existing failures is more than enough.
 @pytest.mark.skip(reason="Only run manually for now")
+@pytest.mark.fuzzer
 @schema.parametrize()
-def test_mr_api(case, current_client_token):
+def test_mr_api_stateless(case, current_client_token):
     case.headers["Authorization"] = f"Bearer {current_client_token}"
     case.headers["Content-Type"] = "application/json"
     case.call_and_validate(verify=False)

tests/model_registry/test_rest_api_stateful.py

@@ -0,0 +1,31 @@
+import pytest
+from simple_logger.logger import get_logger
+from tests.model_registry.constants import MR_NAMESPACE
+from utilities.constants import DscComponents
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.fuzzer
+@pytest.mark.parametrize(
+    "model_registry_namespace, updated_dsc_component_state_scope_class",
+    [
+        pytest.param(
+            {"namespace_name": MR_NAMESPACE},
+            {
+                "component_patch": {
+                    DscComponents.MODELREGISTRY: {
+                        "managementState": DscComponents.ManagementState.MANAGED,
+                        "registriesNamespace": MR_NAMESPACE,
+                    },
+                },
+            },
+        )
+    ],
+    indirect=True,
+)
+@pytest.mark.usefixtures("model_registry_namespace", "updated_dsc_component_state_scope_class")
+class TestRestAPIStateful:
+    def test_mr_api_stateful(self, state_machine):
+        """Launches stateful tests against the Model Registry API endpoints defined in its openAPI yaml spec file"""
+        state_machine.run()