Merge 0e89a00683 into 7e7c1c4c0a

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

@@ -19,6 +19,7 @@ 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
@@ -30,18 +31,36 @@ 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 value
+    return _normalize_scalar(value)
 
 
 def _json_body_matches(expected: Any, actual: Any) -> bool:
     if isinstance(expected, dict) and isinstance(actual, dict):
-        filtered_expected = {key: expected[key] for key in actual if key in expected}
-        return _normalize_json(filtered_expected) == _normalize_json(actual)
+        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)
 
 

thousandeyes-sdk-core/test/test_mock_server.py

@@ -185,3 +185,131 @@ def test_mock_server_rejects_path_missing_path_variable(manifest):
         )
         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)