Merge b300d0ccbc into 7e7c1c4c0a

thousandeyes-sdk-core/src/thousandeyes_sdk/core/exceptions.py

@@ -14,7 +14,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from typing import Any, Optional, Type
+from typing import Any, Optional
 
 from typing_extensions import Self
 
@@ -135,22 +135,6 @@ class ApiException(OpenApiException):
                     pass
             self.headers = http_resp.getheaders()
 
-    @classmethod
-    def exception_class_for_http_status(cls, status: int) -> Type["ApiException"]:
-        if status == 400:
-            return BadRequestException
-        if status == 401:
-            return UnauthorizedException
-        if status == 403:
-            return ForbiddenException
-        if status == 404:
-            return NotFoundException
-        if status == 429:
-            return TooManyRequestsException
-        if 500 <= status <= 599:
-            return ServiceException
-        return ApiException
-
     @classmethod
     def from_response(
             cls,
@@ -159,8 +143,24 @@ class ApiException(OpenApiException):
             body: Optional[str],
             data: Optional[Any],
     ) -> Self:
-        exc_class = cls.exception_class_for_http_status(http_resp.status)
+        if http_resp.status == 400:
-        raise exc_class(http_resp=http_resp, body=body, data=data)
+            raise BadRequestException(http_resp=http_resp, body=body, data=data)
+
+        if http_resp.status == 401:
+            raise UnauthorizedException(http_resp=http_resp, body=body, data=data)
+
+        if http_resp.status == 403:
+            raise ForbiddenException(http_resp=http_resp, body=body, data=data)
+
+        if http_resp.status == 404:
+            raise NotFoundException(http_resp=http_resp, body=body, data=data)
+
+        if http_resp.status == 429:
+            raise TooManyRequestsException(http_resp=http_resp, body=body, data=data)
+
+        if 500 <= http_resp.status <= 599:
+            raise ServiceException(http_resp=http_resp, body=body, data=data)
+        raise ApiException(http_resp=http_resp, body=body, data=data)
 
     def __str__(self):
         """Custom error messages for exception"""

thousandeyes-sdk-core/test/conftest.py

@@ -1,6 +0,0 @@
-import sys
-from pathlib import Path
-
-_core_test_support = Path(__file__).resolve().parent
-if str(_core_test_support) not in sys.path:
-    sys.path.insert(0, str(_core_test_support))

thousandeyes-sdk-core/test/sdk_test_support/__init__.py

@@ -1 +0,0 @@
-# Test-only helpers for SDK integration tests. Not shipped in the published package.

thousandeyes-sdk-core/test/sdk_test_support/mock_server.py

@@ -1,228 +0,0 @@
-# Copyright 2024 Cisco Systems, Inc. and its affiliates
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-
-from __future__ import annotations
-
-import json
-import re
-import threading
-from datetime import datetime
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from typing import Any, Dict, Mapping, Optional
-from urllib.parse import urlparse
-
-from sdk_test_support.mock_server_types import OperationExpectation
-
-OPERATION_ID_HEADER = "X-TE-Operation-Id"
-ERROR_STATUS_HEADER = "X-TE-Error-Status"
-AUTHORIZATION_HEADER = "Authorization"
-
-
-def _normalize_scalar(value: Any) -> Any:
-    if isinstance(value, str) and "T" in value:
-        try:
-            return datetime.fromisoformat(value.replace("Z", "+00:00")).isoformat().replace("+00:00", "Z")
-        except ValueError:
-            return value
-    return value
-
-
-def _normalize_json(value: Any) -> Any:
-    if isinstance(value, dict):
-        return {key: _normalize_json(value[key]) for key in sorted(value.keys())}
-    if isinstance(value, list):
-        return [_normalize_json(item) for item in value]
-    return _normalize_scalar(value)
-
-
-def _json_body_matches(expected: Any, actual: Any) -> bool:
-    if isinstance(expected, dict) and isinstance(actual, dict):
-        for key in actual:
-            if key not in expected:
-                return False
-            if not _json_body_matches(expected[key], actual[key]):
-                return False
-        return True
-    if isinstance(expected, list) and isinstance(actual, list):
-        if len(expected) != len(actual):
-            return False
-        return all(_json_body_matches(expected_item, actual_item)
-                   for expected_item, actual_item in zip(expected, actual))
-    return _normalize_json(expected) == _normalize_json(actual)
-
-
-class MockApiServer:
-    def __init__(self, manifest: Mapping[str, OperationExpectation], host: str = "127.0.0.1", port: int = 0):
-        self._manifest = dict(manifest)
-        self._host = host
-        self._port = port
-        self._server: Optional[ThreadingHTTPServer] = None
-        self._thread: Optional[threading.Thread] = None
-
-    @property
-    def base_url(self) -> str:
-        if self._server is None:
-            raise RuntimeError("MockApiServer has not been started")
-        return f"http://{self._host}:{self._server.server_port}"
-
-    def start(self) -> None:
-        if self._server is not None:
-            return
-
-        manifest = self._manifest
-        handler = _build_handler(manifest)
-        self._server = ThreadingHTTPServer((self._host, self._port), handler)
-        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
-        self._thread.start()
-
-    def stop(self) -> None:
-        if self._server is None:
-            return
-        self._server.shutdown()
-        self._server.server_close()
-        if self._thread is not None:
-            self._thread.join(timeout=5)
-        self._server = None
-        self._thread = None
-
-    def __enter__(self) -> "MockApiServer":
-        self.start()
-        return self
-
-    def __exit__(self, exc_type, exc, tb) -> None:
-        self.stop()
-
-
-def _build_handler(manifest: Mapping[str, OperationExpectation]):
-    class MockApiRequestHandler(BaseHTTPRequestHandler):
-        def log_message(self, format: str, *args) -> None:
-            return
-
-        def do_GET(self) -> None:
-            self._handle_request("GET")
-
-        def do_POST(self) -> None:
-            self._handle_request("POST")
-
-        def do_PUT(self) -> None:
-            self._handle_request("PUT")
-
-        def do_PATCH(self) -> None:
-            self._handle_request("PATCH")
-
-        def do_DELETE(self) -> None:
-            self._handle_request("DELETE")
-
-        def _handle_request(self, method: str) -> None:
-            auth_error = _validate_authorization(self.headers.get(AUTHORIZATION_HEADER))
-            if auth_error is not None:
-                self._write_json(auth_error, 401)
-                return
-
-            operation_id = self.headers.get(OPERATION_ID_HEADER)
-            if not operation_id:
-                self._write_json({"detail": f"Missing required header {OPERATION_ID_HEADER}"}, 400)
-                return
-
-            expectation = manifest.get(operation_id)
-            if expectation is None:
-                self._write_json({"detail": f"Unknown operation id {operation_id}"}, 400)
-                return
-
-            if expectation.method.upper() != method.upper():
-                self._write_json(
-                    {"detail": f"Unexpected HTTP method {method} for operation {operation_id}"},
-                    400,
-                )
-                return
-
-            parsed = urlparse(self.path)
-            if not _path_matches(expectation.path, parsed.path):
-                self._write_json({"detail": "Path does not match operation expectation"}, 400)
-                return
-
-            error_status_header = self.headers.get(ERROR_STATUS_HEADER)
-            if error_status_header:
-                self._handle_error_response(expectation, error_status_header)
-                return
-
-            body_bytes = _read_body(self)
-            if expectation.request_body_example is not None:
-                if not body_bytes:
-                    self._write_json({"detail": "Expected request body"}, 400)
-                    return
-                try:
-                    request_json = json.loads(body_bytes.decode("utf-8"))
-                except json.JSONDecodeError:
-                    self._write_json({"detail": "Invalid JSON request body"}, 400)
-                    return
-                if not _json_body_matches(expectation.request_body_example, request_json):
-                    self._write_json({"detail": "Request body does not match OAS example"}, 400)
-                    return
-
-            if expectation.success_body is None:
-                self.send_response(expectation.success_status)
-                self.end_headers()
-                return
-
-            self._write_json(
-                expectation.success_body,
-                expectation.success_status,
-                expectation.success_content_type,
-            )
-
-        def _handle_error_response(self, expectation: OperationExpectation, error_status_header: str) -> None:
-            error_response = expectation.error_responses.get(error_status_header)
-            if error_response is None:
-                self._write_json(
-                    {"detail": f"No configured error response for status {error_status_header}"},
-                    400,
-                )
-                return
-            if error_response.body is None:
-                self.send_response(error_response.status)
-                self.end_headers()
-                return
-            self._write_json(error_response.body, error_response.status, error_response.content_type)
-
-        def _write_json(self, body: Any, status: int, content_type: str = "application/json") -> None:
-            payload = json.dumps(body).encode("utf-8")
-            self.send_response(status)
-            self.send_header("Content-Type", content_type)
-            self.send_header("Content-Length", str(len(payload)))
-            self.end_headers()
-            self.wfile.write(payload)
-
-    return MockApiRequestHandler
-
-
-def _validate_authorization(value: Optional[str]) -> Optional[Dict[str, str]]:
-    if value is None or not value.strip():
-        return {"detail": f"Missing or empty required header {AUTHORIZATION_HEADER}"}
-    return None
-
-
-def _read_body(handler: BaseHTTPRequestHandler) -> bytes:
-    length = handler.headers.get("Content-Length")
-    if not length:
-        return b""
-    return handler.rfile.read(int(length))
-
-
-def _path_matches(template: str, actual_path: str) -> bool:
-    pattern = re.sub(r"\{[^/]+\}", r"[^/]+", template)
-    pattern = f"^{pattern}$"
-    return re.match(pattern, actual_path) is not None

thousandeyes-sdk-core/test/sdk_test_support/mock_server_types.py

@@ -1,41 +0,0 @@
-# Copyright 2024 Cisco Systems, Inc. and its affiliates
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# SPDX-License-Identifier: Apache-2.0
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from typing import Any, Dict
-
-
-@dataclass(frozen=True)
-class ErrorResponseExpectation:
-    status: int
-    body: Any
-    content_type: str = "application/problem+json"
-
-
-@dataclass(frozen=True)
-class OperationExpectation:
-    operation_id: str
-    method: str
-    path: str
-    success_status: int
-    success_body: Any = None
-    success_content_type: str = "application/json"
-    request_body_example: Any = None
-    path_param_examples: Dict[str, str] = field(default_factory=dict)
-    query_param_examples: Dict[str, str] = field(default_factory=dict)
-    error_responses: Dict[str, ErrorResponseExpectation] = field(default_factory=dict)

thousandeyes-sdk-core/test/test_exceptions.py

@@ -1,28 +0,0 @@
-import pytest
-
-from thousandeyes_sdk.core.exceptions import (
-    ApiException,
-    BadRequestException,
-    ForbiddenException,
-    NotFoundException,
-    ServiceException,
-    TooManyRequestsException,
-    UnauthorizedException,
-)
-
-
-@pytest.mark.parametrize(
-    ("status", "expected"),
-    [
-        (400, BadRequestException),
-        (401, UnauthorizedException),
-        (403, ForbiddenException),
-        (404, NotFoundException),
-        (429, TooManyRequestsException),
-        (500, ServiceException),
-        (503, ServiceException),
-        (418, ApiException),
-    ],
-)
-def test_exception_class_for_http_status(status, expected):
-    assert ApiException.exception_class_for_http_status(status) is expected

thousandeyes-sdk-core/test/test_mock_server.py

@@ -1,315 +0,0 @@
-import json
-
-import pytest
-
-from sdk_test_support.mock_server import (
-    AUTHORIZATION_HEADER,
-    ERROR_STATUS_HEADER,
-    OPERATION_ID_HEADER,
-    MockApiServer,
-)
-from sdk_test_support.mock_server_types import ErrorResponseExpectation, OperationExpectation
-
-
-@pytest.fixture
-def manifest():
-    return {
-        "createAlertRule": OperationExpectation(
-            operation_id="createAlertRule",
-            method="POST",
-            path="/alerts/rules",
-            request_body_example={"ruleName": "Example"},
-            success_status=201,
-            success_body={"ruleId": "1"},
-            error_responses={
-                "400": ErrorResponseExpectation(
-                    status=400,
-                    body={"title": "Bad Request", "status": 400},
-                )
-            },
-        ),
-        "deleteAlertRule": OperationExpectation(
-            operation_id="deleteAlertRule",
-            method="DELETE",
-            path="/alerts/rules/{ruleId}",
-            path_param_examples={"ruleId": "127094"},
-            success_status=204,
-            success_body=None,
-        ),
-        "getAlertRule": OperationExpectation(
-            operation_id="getAlertRule",
-            method="GET",
-            path="/alerts/rules/{ruleId}",
-            path_param_examples={"ruleId": "127094"},
-            success_status=200,
-            success_body={"ruleId": "127094", "ruleName": "Example"},
-        ),
-    }
-
-
-def _request(server: MockApiServer, *, method: str, path: str, headers=None, body=None):
-    import urllib.request
-
-    request_headers = {
-        AUTHORIZATION_HEADER: "Bearer test-token",
-        OPERATION_ID_HEADER: "createAlertRule",
-    }
-    if headers:
-        request_headers.update(headers)
-
-    data = None
-    if body is not None:
-        data = json.dumps(body).encode("utf-8")
-        request_headers["Content-Type"] = "application/json"
-
-    request = urllib.request.Request(
-        server.base_url + path,
-        data=data,
-        headers=request_headers,
-        method=method,
-    )
-    try:
-        with urllib.request.urlopen(request) as response:
-            return response.status, response.read()
-    except urllib.error.HTTPError as exc:
-        return exc.code, exc.read()
-
-
-def test_mock_server_happy_path(manifest):
-    with MockApiServer(manifest) as server:
-        status, body = _request(
-            server,
-            method="POST",
-            path="/alerts/rules",
-            body={"ruleName": "Example"},
-        )
-        assert status == 201
-        assert json.loads(body.decode("utf-8")) == {"ruleId": "1"}
-
-
-def test_mock_server_rejects_missing_authorization(manifest):
-    with MockApiServer(manifest) as server:
-        status, _ = _request(
-            server,
-            method="POST",
-            path="/alerts/rules",
-            headers={AUTHORIZATION_HEADER: ""},
-            body={"ruleName": "Example"},
-        )
-        assert status == 401
-
-
-def test_mock_server_rejects_invalid_request_body(manifest):
-    with MockApiServer(manifest) as server:
-        status, _ = _request(
-            server,
-            method="POST",
-            path="/alerts/rules",
-            body={"ruleName": "Wrong"},
-        )
-        assert status == 400
-
-
-def test_mock_server_error_path(manifest):
-    with MockApiServer(manifest) as server:
-        status, body = _request(
-            server,
-            method="POST",
-            path="/alerts/rules",
-            headers={ERROR_STATUS_HEADER: "400"},
-            body={"unexpected": True},
-        )
-        assert status == 400
-        assert json.loads(body.decode("utf-8"))["title"] == "Bad Request"
-
-
-def test_mock_server_ignores_readonly_fields_in_expected_body(manifest):
-    readonly_manifest = {
-        **manifest,
-        "createAlertRule": OperationExpectation(
-            operation_id="createAlertRule",
-            method="POST",
-            path="/alerts/rules",
-            request_body_example={"ruleName": "Example", "ruleId": "read-only"},
-            success_status=201,
-            success_body={"ruleId": "1"},
-        ),
-    }
-    with MockApiServer(readonly_manifest) as server:
-        status, body = _request(
-            server,
-            method="POST",
-            path="/alerts/rules",
-            body={"ruleName": "Example"},
-        )
-        assert status == 201
-        assert json.loads(body.decode("utf-8")) == {"ruleId": "1"}
-
-
-def test_mock_server_no_content_response(manifest):
-    with MockApiServer(manifest) as server:
-        import urllib.request
-
-        request = urllib.request.Request(
-            server.base_url + "/alerts/rules/127094",
-            headers={
-                AUTHORIZATION_HEADER: "Bearer test-token",
-                OPERATION_ID_HEADER: "deleteAlertRule",
-            },
-            method="DELETE",
-        )
-        with urllib.request.urlopen(request) as response:
-            assert response.status == 204
-            assert response.read() == b""
-
-
-def test_mock_server_matches_path_variable(manifest):
-    with MockApiServer(manifest) as server:
-        status, body = _request(
-            server,
-            method="GET",
-            path="/alerts/rules/127094",
-            headers={OPERATION_ID_HEADER: "getAlertRule"},
-        )
-        assert status == 200
-        assert json.loads(body.decode("utf-8")) == {"ruleId": "127094", "ruleName": "Example"}
-
-
-def test_mock_server_rejects_path_missing_path_variable(manifest):
-    with MockApiServer(manifest) as server:
-        status, body = _request(
-            server,
-            method="GET",
-            path="/alerts/rules",
-            headers={OPERATION_ID_HEADER: "getAlertRule"},
-        )
-        assert status == 400
-        assert json.loads(body.decode("utf-8"))["detail"] == "Path does not match operation expectation"
-
-
-def test_mock_server_accepts_equivalent_iso8601_datetime_formats(manifest):
-    datetime_manifest = {
-        **manifest,
-        "createAlertRule": OperationExpectation(
-            operation_id="createAlertRule",
-            method="POST",
-            path="/alerts/rules",
-            request_body_example={
-                "ruleName": "Example",
-                "startDate": "2017-07-01T05:00:00Z",
-            },
-            success_status=201,
-            success_body={"ruleId": "1"},
-        ),
-    }
-    with MockApiServer(datetime_manifest) as server:
-        status, body = _request(
-            server,
-            method="POST",
-            path="/alerts/rules",
-            body={
-                "ruleName": "Example",
-                "startDate": "2017-07-01T05:00:00+00:00",
-            },
-        )
-        assert status == 201
-        assert json.loads(body.decode("utf-8")) == {"ruleId": "1"}
-
-
-def test_mock_server_ignores_readonly_fields_in_nested_request_objects(manifest):
-    nested_manifest = {
-        **manifest,
-        "createAlertRule": OperationExpectation(
-            operation_id="createAlertRule",
-            method="POST",
-            path="/alerts/rules",
-            request_body_example={
-                "ruleName": "Example",
-                "widgets": [
-                    {
-                        "title": "Widget Title",
-                        "id": "read-only-id",
-                        "embedUrl": "https://example.com/embed",
-                    }
-                ],
-            },
-            success_status=201,
-            success_body={"ruleId": "1"},
-        ),
-    }
-    with MockApiServer(nested_manifest) as server:
-        status, body = _request(
-            server,
-            method="POST",
-            path="/alerts/rules",
-            body={
-                "ruleName": "Example",
-                "widgets": [{"title": "Widget Title"}],
-            },
-        )
-        assert status == 201
-        assert json.loads(body.decode("utf-8")) == {"ruleId": "1"}
-
-
-def test_integration_error_assertion_fails_when_deserialized_error_does_not_match_oas_example():
-    """Generated error-path tests compare ApiException.data to the OAS error example."""
-    import unittest
-    import urllib.error
-    import urllib.request
-
-    from pydantic import BaseModel, ConfigDict
-
-    class Error(BaseModel):
-        title: str
-        status: int
-
-        model_config = ConfigDict(extra="allow")
-
-        def to_json(self) -> str:
-            return self.model_dump_json()
-
-    mismatched_manifest = {
-        "createAlertRule": OperationExpectation(
-            operation_id="createAlertRule",
-            method="POST",
-            path="/alerts/rules",
-            request_body_example={"ruleName": "Example"},
-            success_status=201,
-            success_body={"ruleId": "1"},
-            error_responses={
-                "400": ErrorResponseExpectation(
-                    status=400,
-                    body={"title": "Wrong Title", "status": 400},
-                )
-            },
-        ),
-    }
-    oas_example = {"title": "Bad Request", "status": 400}
-
-    with MockApiServer(mismatched_manifest) as server:
-        request = urllib.request.Request(
-            server.base_url + "/alerts/rules",
-            data=json.dumps({"unexpected": True}).encode("utf-8"),
-            headers={
-                AUTHORIZATION_HEADER: "Bearer test-token",
-                OPERATION_ID_HEADER: "createAlertRule",
-                ERROR_STATUS_HEADER: "400",
-                "Content-Type": "application/json",
-            },
-            method="POST",
-        )
-        with pytest.raises(urllib.error.HTTPError) as http_error:
-            urllib.request.urlopen(request)
-        wire_body = json.loads(http_error.value.read().decode("utf-8"))
-
-    exception_data = Error.model_validate(wire_body)
-
-    def assert_constructed_model_matches_example_json(model, loaded_json):
-        test_case = unittest.TestCase()
-        test_case.assertEqual(
-            json.dumps(loaded_json, sort_keys=True),
-            json.dumps(json.loads(model.to_json()), sort_keys=True),
-        )
-
-    with pytest.raises(AssertionError):
-        assert_constructed_model_matches_example_json(exception_data, oas_example)