-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconftest.py
More file actions
200 lines (152 loc) · 7.14 KB
/
Copy pathconftest.py
File metadata and controls
200 lines (152 loc) · 7.14 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
"""
Pytest configuration and fixtures for Stack0 SDK tests
"""
from __future__ import annotations
from typing import Any, Generator
from unittest.mock import MagicMock, patch
import httpx
import pytest
from stack0 import AsyncStack0, Stack0
@pytest.fixture
def api_key() -> str:
"""Test API key"""
return "stack0_test_key_123"
@pytest.fixture
def base_url() -> str:
"""Test base URL"""
return "https://api.stack0.dev/v1"
@pytest.fixture
def mock_response() -> dict[str, Any]:
"""Base mock response"""
return {"success": True}
def create_mock_response(data: dict[str, Any], status_code: int = 200) -> httpx.Response:
"""Create a mock httpx Response"""
return httpx.Response(
status_code=status_code,
json=data,
request=httpx.Request("GET", "https://api.stack0.dev/test"),
)
@pytest.fixture
def mock_httpx_client() -> Generator[MagicMock, None, None]:
"""Mock httpx Client"""
with patch("httpx.Client") as mock:
client_instance = MagicMock()
mock.return_value = client_instance
mock.return_value.__enter__ = MagicMock(return_value=client_instance)
mock.return_value.__exit__ = MagicMock(return_value=False)
yield client_instance
@pytest.fixture
def mock_async_httpx_client() -> Generator[MagicMock, None, None]:
"""Mock httpx AsyncClient"""
with patch("httpx.AsyncClient") as mock:
client_instance = MagicMock()
mock.return_value = client_instance
mock.return_value.__aenter__ = MagicMock(return_value=client_instance)
mock.return_value.__aexit__ = MagicMock(return_value=False)
yield client_instance
@pytest.fixture
def stack0_client(api_key: str, mock_httpx_client: MagicMock) -> Stack0:
"""Create a Stack0 client with mocked HTTP"""
with patch("stack0._client.http.httpx") as mock_httpx:
mock_httpx.Client.return_value = mock_httpx_client
client = Stack0(api_key=api_key)
return client
@pytest.fixture
def async_stack0_client(api_key: str, mock_async_httpx_client: MagicMock) -> AsyncStack0:
"""Create an AsyncStack0 client with mocked HTTP"""
with patch("stack0._client.http.httpx") as mock_httpx:
mock_httpx.AsyncClient.return_value = mock_async_httpx_client
client = AsyncStack0(api_key=api_key)
return client
class MockHTTPClient:
"""Mock HTTP client for testing"""
def __init__(self, responses: dict[str, dict[str, Any]] | None = None) -> None:
self.responses = responses or {}
self.requests: list[dict[str, Any]] = []
self.api_key = "test_key"
self.base_url = "https://api.stack0.dev/v1"
def _get_response(self, method: str, path: str) -> dict[str, Any]:
"""Get a mock response for the given request"""
key = f"{method}:{path}"
if key in self.responses:
return self.responses[key]
# Check for partial path match
for resp_key, resp_value in self.responses.items():
resp_method, resp_path = resp_key.split(":", 1)
if resp_method == method and path.startswith(resp_path.split("?")[0]):
return resp_value
# Default response
return {"success": True}
def get(self, path: str) -> dict[str, Any]:
self.requests.append({"method": "GET", "path": path})
return self._get_response("GET", path)
def post(self, path: str, body: dict[str, Any] | None = None) -> dict[str, Any]:
self.requests.append({"method": "POST", "path": path, "body": body or {}})
return self._get_response("POST", path)
def put(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "PUT", "path": path, "body": body})
return self._get_response("PUT", path)
def patch(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "PATCH", "path": path, "body": body})
return self._get_response("PATCH", path)
def delete(self, path: str) -> dict[str, Any]:
self.requests.append({"method": "DELETE", "path": path})
return self._get_response("DELETE", path)
def delete_with_body(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "DELETE", "path": path, "body": body})
return self._get_response("DELETE", path)
def upload(self, path: str, file: bytes, data: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "POST", "path": path, "data": data, "file_size": len(file)})
return self._get_response("POST", path)
def close(self) -> None:
pass
class MockAsyncHTTPClient:
"""Mock async HTTP client for testing"""
def __init__(self, responses: dict[str, dict[str, Any]] | None = None) -> None:
self.responses = responses or {}
self.requests: list[dict[str, Any]] = []
self.api_key = "test_key"
self.base_url = "https://api.stack0.dev/v1"
def _get_response(self, method: str, path: str) -> dict[str, Any]:
"""Get a mock response for the given request"""
key = f"{method}:{path}"
if key in self.responses:
return self.responses[key]
# Check for partial path match
for resp_key, resp_value in self.responses.items():
resp_method, resp_path = resp_key.split(":", 1)
if resp_method == method and path.startswith(resp_path.split("?")[0]):
return resp_value
# Default response
return {"success": True}
async def get(self, path: str) -> dict[str, Any]:
self.requests.append({"method": "GET", "path": path})
return self._get_response("GET", path)
async def post(self, path: str, body: dict[str, Any] | None = None) -> dict[str, Any]:
self.requests.append({"method": "POST", "path": path, "body": body or {}})
return self._get_response("POST", path)
async def put(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "PUT", "path": path, "body": body})
return self._get_response("PUT", path)
async def patch(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "PATCH", "path": path, "body": body})
return self._get_response("PATCH", path)
async def delete(self, path: str) -> dict[str, Any]:
self.requests.append({"method": "DELETE", "path": path})
return self._get_response("DELETE", path)
async def delete_with_body(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "DELETE", "path": path, "body": body})
return self._get_response("DELETE", path)
async def upload(self, path: str, file: bytes, data: dict[str, Any]) -> dict[str, Any]:
self.requests.append({"method": "POST", "path": path, "data": data, "file_size": len(file)})
return self._get_response("POST", path)
async def close(self) -> None:
pass
@pytest.fixture
def mock_http() -> MockHTTPClient:
"""Create a mock HTTP client"""
return MockHTTPClient()
@pytest.fixture
def mock_async_http() -> MockAsyncHTTPClient:
"""Create a mock async HTTP client"""
return MockAsyncHTTPClient()