Add api tests stateless

clients/python/.gitignore

@@ -7,3 +7,4 @@
 __pycache__/
 venv/
 .port-forwards.pid
+.hypothesis/

clients/python/noxfile.py

@@ -60,6 +60,7 @@ def tests(session: Session) -> None:
         "pytest-asyncio",
         "uvloop",
         "olot",
+        "schemathesis",
     )
     session.run(
         "pytest",
@@ -83,6 +84,7 @@ def e2e_tests(session: Session) -> None:
         "boto3",
         "olot",
         "uvloop",
+        "schemathesis",
     )
     try:
         session.run(

clients/python/pyproject.toml

@@ -60,6 +60,7 @@ requests = "^2.32.2"
 black = ">=24.4.2,<26.0.0"
 types-python-dateutil = "^2.9.0.20240906"
 pytest-html = "^4.1.1"
+schemathesis = "^4.0.2"
 
 [tool.coverage.run]
 branch = true

clients/python/schemathesis.toml

@@ -0,0 +1 @@
+base-url = "${API_HOST}"

clients/python/tests/conftest.py

@@ -15,7 +15,9 @@
 
 import pytest
 import requests
+import schemathesis
 import uvloop
+from schemathesis.specs.openapi.schemas import BaseOpenAPISchema
 
 from model_registry import ModelRegistry
 from model_registry.utils import BackendDefinition, _get_skopeo_backend
@@ -317,3 +319,14 @@ def mock_override(base_image, dest_dir, params):
         skopeo_pull_mock.side_effect = mock_override
         skopeo_push_mock.side_effect = mock_override
         yield backend, skopeo_pull_mock, skopeo_push_mock, generic_auth_vars
+
+@pytest.fixture(scope="class")
+def generated_schema(pytestconfig: pytest.Config ) -> BaseOpenAPISchema:
+    os.environ["API_HOST"] = REGISTRY_URL
+    config = schemathesis.config.SchemathesisConfig.from_path(f"{pytestconfig.rootpath}/schemathesis.toml")
+    schema = schemathesis.openapi.from_url(
+        url="https://raw.githubusercontent.com/kubeflow/model-registry/main/api/openapi/model-registry.yaml",
+        config=config,
+    )
+    schema.config.output.sanitization.update(enabled=False)
+    return schema

clients/python/tests/test_api.py

@@ -0,0 +1,40 @@
+import pytest
+import schemathesis
+from hypothesis import HealthCheck, settings
+
+schema = schemathesis.pytest.from_fixture("generated_schema")
+
+
+@schema.parametrize()
+@settings(
+    deadline=None,
+    suppress_health_check=[
+        HealthCheck.filter_too_much,
+    ]
+)
+@pytest.mark.e2e
+def test_mr_api_stateless(setup_env_user_token, case):
+    """Test the Model Registry API endpoints.
+
+    This test uses schemathesis to generate and validate API requests
+    """
+
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {setup_env_user_token}"
+    }
+
+    # We need to ensure valid data
+    if case.path == "/api/model_registry/v1alpha3/model_versions/{modelversionId}/artifacts":
+        case.path_parameters["modelversionId"] = "test-model-version-1"
+
+        if case.method == "POST":
+            case.body = {
+                "artifactType": "model-artifact",
+                "name": "test-artifact",
+                "uri": "s3://test-bucket/test-artifact",
+                "metadata": {}
+            }
+
+    case.call_and_validate(headers=headers)
+