|
1 | | -import os |
2 | 1 | import pytest |
3 | 2 | from pytest import Config |
4 | | -import schemathesis |
5 | 3 | from typing import Generator, Any |
6 | 4 |
|
7 | 5 | from ocp_resources.infrastructure import Infrastructure |
|
14 | 12 | from ocp_resources.deployment import Deployment |
15 | 13 |
|
16 | 14 | from ocp_resources.model_registry_modelregistry_opendatahub_io import ModelRegistry |
17 | | -from schemathesis.specs.openapi.schemas import BaseOpenAPISchema |
18 | | -from schemathesis.generation.stateful.state_machine import APIStateMachine |
19 | | -from schemathesis.core.transport import Response |
20 | | -from schemathesis.generation.case import Case |
21 | 15 | from ocp_resources.resource import ResourceEditor |
22 | 16 |
|
23 | 17 | from pytest import FixtureRequest |
@@ -299,43 +293,6 @@ def model_registry_instance_rest_endpoint( |
299 | 293 | return get_endpoint_from_mr_service(svc=model_registry_instance_service, protocol=Protocols.REST) |
300 | 294 |
|
301 | 295 |
|
302 | | -@pytest.fixture(scope="class") |
303 | | -def generated_schema(pytestconfig: Config, model_registry_instance_rest_endpoint: str) -> BaseOpenAPISchema: |
304 | | - os.environ["API_HOST"] = model_registry_instance_rest_endpoint |
305 | | - config = schemathesis.config.SchemathesisConfig.from_path(f"{pytestconfig.rootpath}/schemathesis.toml") |
306 | | - schema = schemathesis.openapi.from_url( |
307 | | - url="https://raw.githubusercontent.com/kubeflow/model-registry/main/api/openapi/model-registry.yaml", |
308 | | - config=config, |
309 | | - ) |
310 | | - return schema |
311 | | - |
312 | | - |
313 | | -@pytest.fixture() |
314 | | -def state_machine(generated_schema: BaseOpenAPISchema, current_client_token: str) -> APIStateMachine: |
315 | | - BaseAPIWorkflow = generated_schema.as_state_machine() |
316 | | - |
317 | | - class APIWorkflow(BaseAPIWorkflow): # type: ignore |
318 | | - headers: dict[str, str] |
319 | | - |
320 | | - def setup(self) -> None: |
321 | | - self.headers = {"Authorization": f"Bearer {current_client_token}", "Content-Type": "application/json"} |
322 | | - |
323 | | - def before_call(self, case: Case) -> None: |
324 | | - LOGGER.info(f"Checking: {case.method} {case.path}") |
325 | | - |
326 | | - # these kwargs are passed to requests.request() |
327 | | - def get_call_kwargs(self, case: Case) -> dict[str, Any]: |
328 | | - return {"verify": False, "headers": self.headers} |
329 | | - |
330 | | - def after_call(self, response: Response, case: Case) -> None: |
331 | | - LOGGER.info( |
332 | | - f"Method tested: {case.method}, API: {case.path}, response code:{response.status_code}," |
333 | | - f" Full Response:{response.text}" |
334 | | - ) |
335 | | - |
336 | | - return APIWorkflow |
337 | | - |
338 | | - |
339 | 296 | @pytest.fixture(scope="class") |
340 | 297 | def updated_dsc_component_state_scope_class( |
341 | 298 | pytestconfig: Config, |
|
0 commit comments