Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions schemathesis.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
base-url = "https://${API_HOST}"
Comment thread
dbasunag marked this conversation as resolved.
20 changes: 15 additions & 5 deletions tests/model_registry/conftest.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import os
import pytest
from pytest import Config
import schemathesis
from typing import Generator, Any
from ocp_resources.pod import Pod
Expand Down Expand Up @@ -259,15 +261,17 @@ def model_registry_instance_rest_endpoint(


@pytest.fixture(scope="class")
def generated_schema(model_registry_instance_rest_endpoint: str) -> BaseOpenAPISchema:
def generated_schema(pytestconfig: Config, model_registry_instance_rest_endpoint: str) -> BaseOpenAPISchema:
os.environ["API_HOST"] = model_registry_instance_rest_endpoint
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"
url="https://raw.githubusercontent.com/kubeflow/model-registry/main/api/openapi/model-registry.yaml",
config=config,
)
schema.configure(base_url=f"https://{model_registry_instance_rest_endpoint}/")
return schema


@pytest.fixture
@pytest.fixture()
def state_machine(generated_schema: BaseOpenAPISchema, current_client_token: str) -> APIStateMachine:
BaseAPIWorkflow = generated_schema.as_state_machine()

Expand All @@ -277,12 +281,18 @@ class APIWorkflow(BaseAPIWorkflow): # type: ignore
def setup(self) -> None:
self.headers = {"Authorization": f"Bearer {current_client_token}", "Content-Type": "application/json"}

def before_call(self, case: Case) -> None:
LOGGER.info(f"Checking: {case.method} {case.path}")

# 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}")
LOGGER.info(
f"Method tested: {case.method}, API: {case.path}, response code:{response.status_code},"
f" Full Response:{response.text}"
)

return APIWorkflow

Expand Down
Loading