0
# Testing Utilities
1

2
Mock connection pools and transports for unit testing HTTP caching behavior. These utilities allow you to test caching logic without making actual network requests.
3

4
## Capabilities
5

6
### Mock Synchronous Transport
7

8
Mock transport for testing synchronous HTTP client caching behavior.
9

10
```python { .api }
11
class MockTransport(httpx.BaseTransport):
12
    def handle_request(self, request: httpx.Request) -> httpx.Response:
13
        """Handle request by returning pre-configured mock response"""
14
    
15
    def add_responses(self, responses: list[httpx.Response]) -> None:
16
        """
17
        Add mock responses to be returned in order.
18
        
19
        Parameters:
20
        - responses: List of httpx.Response objects to return
21
        """
22
```
23

24
**Usage Examples:**
25

26
```python
27
import httpx
28
import hishel
29

30
# Create mock responses
31
response1 = httpx.Response(200, json={"data": "first"})
32
response2 = httpx.Response(200, json={"data": "second"})
33

34
# Set up mock transport
35
mock_transport = hishel.MockTransport()
36
mock_transport.add_responses([response1, response2])
37

38
# Use with cache client for testing
39
cache_transport = hishel.CacheTransport(transport=mock_transport)
40
with httpx.Client(transport=cache_transport) as client:
41
    # First request uses first mock response
42
    resp1 = client.get("https://api.example.com/data")
43
    assert resp1.json() == {"data": "first"}
44
    
45
    # Second request should be served from cache
46
    resp2 = client.get("https://api.example.com/data")  
47
    assert resp2.json() == {"data": "first"}  # Same as first (cached)
48
```
49

50
### Mock Asynchronous Transport
51

52
Mock transport for testing asynchronous HTTP client caching behavior.
53

54
```python { .api }
55
class MockAsyncTransport(httpx.AsyncBaseTransport):
56
    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
57
        """Handle async request by returning pre-configured mock response"""
58
    
59
    def add_responses(self, responses: list[httpx.Response]) -> None:
60
        """
61
        Add mock responses to be returned in order.
62
        
63
        Parameters:
64
        - responses: List of httpx.Response objects to return
65
        """
66
```
67

68
**Usage Examples:**
69

70
```python
71
import httpx
72
import hishel
73
import asyncio
74

75
async def test_async_caching():
76
    # Create mock responses with cache headers
77
    response = httpx.Response(
78
        200,
79
        headers={"cache-control": "max-age=3600"},
80
        json={"data": "cached"}
81
    )
82
    
83
    # Set up mock async transport
84
    mock_transport = hishel.MockAsyncTransport()
85
    mock_transport.add_responses([response])
86
    
87
    # Use with async cache client
88
    cache_transport = hishel.AsyncCacheTransport(transport=mock_transport)
89
    async with httpx.AsyncClient(transport=cache_transport) as client:
90
        # First request uses mock response
91
        resp1 = await client.get("https://api.example.com/data")
92
        assert resp1.json() == {"data": "cached"}
93
        
94
        # Second request served from cache (no additional mock response needed)
95
        resp2 = await client.get("https://api.example.com/data")
96
        assert resp2.json() == {"data": "cached"}
97

98
asyncio.run(test_async_caching())
99
```
100

101
### Mock Connection Pools
102

103
Lower-level mock connection pools for testing httpcore-level caching.
104

105
```python { .api }
106
class MockConnectionPool:
107
    def handle_request(self, request: httpcore.Request) -> httpcore.Response:
108
        """Handle httpcore request with mock response"""
109
    
110
    def add_responses(self, responses: list[httpcore.Response]) -> None:
111
        """
112
        Add mock httpcore responses.
113
        
114
        Parameters:
115
        - responses: List of httpcore.Response objects
116
        """
117

118
class MockAsyncConnectionPool:
119
    async def handle_async_request(self, request: httpcore.Request) -> httpcore.Response:
120
        """Handle async httpcore request with mock response"""
121
    
122
    def add_responses(self, responses: list[httpcore.Response]) -> None:
123
        """
124
        Add mock httpcore responses.
125
        
126
        Parameters:
127
        - responses: List of httpcore.Response objects
128
        """
129
```
130

131
**Usage Examples:**
132

133
```python
134
import httpcore
135
import hishel
136

137
# Create mock httpcore responses
138
response = httpcore.Response(
139
    200,
140
    headers=[(b"content-type", b"application/json")],
141
    content=b'{"data": "test"}'
142
)
143

144
# Test with mock connection pool
145
mock_pool = hishel.MockConnectionPool()
146
mock_pool.add_responses([response])
147

148
# Use for lower-level testing
149
request = httpcore.Request(b"GET", b"https://api.example.com/data")
150
response = mock_pool.handle_request(request)
151
assert response.status == 200
152
```
153

154
## Testing Cache Behavior
155

156
### Test Cache Hit/Miss
157

158
```python
159
import httpx
160
import hishel
161
import pytest
162

163
def test_cache_hit_miss():
164
    # Mock response with cache headers
165
    cached_response = httpx.Response(
166
        200,
167
        headers={"cache-control": "max-age=3600", "etag": '"abc123"'},
168
        json={"data": "original"}
169
    )
170
    
171
    # Fresh response for revalidation
172
    fresh_response = httpx.Response(
173
        200,
174
        headers={"cache-control": "max-age=3600", "etag": '"def456"'},
175
        json={"data": "updated"}
176
    )
177
    
178
    mock_transport = hishel.MockTransport()
179
    mock_transport.add_responses([cached_response, fresh_response])
180
    
181
    storage = hishel.InMemoryStorage()
182
    cache_transport = hishel.CacheTransport(
183
        transport=mock_transport,
184
        storage=storage
185
    )
186
    
187
    with httpx.Client(transport=cache_transport) as client:
188
        # First request - cache miss
189
        resp1 = client.get("https://api.example.com/data")
190
        assert resp1.json() == {"data": "original"}
191
        
192
        # Second request - cache hit
193
        resp2 = client.get("https://api.example.com/data")
194
        assert resp2.json() == {"data": "original"}  # Served from cache
195
        
196
        # Verify only one network request was made
197
        # (second response in mock is unused)
198
```
199

200
### Test Cache Revalidation
201

202
```python
203
import httpx
204
import hishel
205

206
def test_cache_revalidation():
207
    # Original response
208
    original_response = httpx.Response(
209
        200,
210
        headers={"cache-control": "max-age=0", "etag": '"version1"'},
211
        json={"version": 1}
212
    )
213
    
214
    # 304 Not Modified response
215
    not_modified_response = httpx.Response(
216
        304,
217
        headers={"cache-control": "max-age=3600", "etag": '"version1"'}
218
    )
219
    
220
    mock_transport = hishel.MockTransport()
221
    mock_transport.add_responses([original_response, not_modified_response])
222
    
223
    cache_transport = hishel.CacheTransport(transport=mock_transport)
224
    
225
    with httpx.Client(transport=cache_transport) as client:
226
        # First request
227
        resp1 = client.get("https://api.example.com/data")
228
        assert resp1.json() == {"version": 1}
229
        
230
        # Second request triggers revalidation due to max-age=0
231
        resp2 = client.get("https://api.example.com/data")
232
        assert resp2.json() == {"version": 1}  # Same content (304 response)
233
        assert resp2.status_code == 200       # But status is 200 (from cache)
234
```
235

236
### Test Different Storage Backends
237

238
```python
239
import httpx
240
import hishel
241
import tempfile
242
import pytest
243

244
@pytest.mark.parametrize("storage_class", [
245
    hishel.InMemoryStorage,
246
    hishel.FileStorage,
247
])
248
def test_storage_backends(storage_class):
249
    response = httpx.Response(
250
        200,
251
        headers={"cache-control": "max-age=3600"},
252
        json={"backend": storage_class.__name__}
253
    )
254
    
255
    mock_transport = hishel.MockTransport()
256
    mock_transport.add_responses([response])
257
    
258
    # Configure storage based on type
259
    if storage_class == hishel.FileStorage:
260
        with tempfile.TemporaryDirectory() as tmpdir:
261
            storage = storage_class(base_path=tmpdir)
262
    else:
263
        storage = storage_class()
264
    
265
    cache_transport = hishel.CacheTransport(
266
        transport=mock_transport,
267
        storage=storage
268
    )
269
    
270
    with httpx.Client(transport=cache_transport) as client:
271
        resp1 = client.get("https://api.example.com/data")
272
        resp2 = client.get("https://api.example.com/data")  # From cache
273
        
274
        assert resp1.json() == resp2.json()
275
```
276

277
### Test Cache Control Directives
278

279
```python
280
import httpx
281
import hishel
282

283
def test_no_cache_directive():
284
    # Response with no-cache directive
285
    response = httpx.Response(
286
        200,
287
        headers={"cache-control": "no-cache"},
288
        json={"directive": "no-cache"}
289
    )
290
    
291
    revalidation_response = httpx.Response(
292
        200,
293
        headers={"cache-control": "max-age=3600"},
294
        json={"directive": "revalidated"}
295
    )
296
    
297
    mock_transport = hishel.MockTransport()
298
    mock_transport.add_responses([response, revalidation_response])
299
    
300
    cache_transport = hishel.CacheTransport(transport=mock_transport)
301
    
302
    with httpx.Client(transport=cache_transport) as client:
303
        # First request
304
        resp1 = client.get("https://api.example.com/data")
305
        assert resp1.json() == {"directive": "no-cache"}
306
        
307
        # Second request should revalidate due to no-cache
308
        resp2 = client.get("https://api.example.com/data")
309
        assert resp2.json() == {"directive": "revalidated"}
310
```
311

312
## Test Utilities
313

314
### Custom Test Storage
315

316
```python
317
import hishel
318

319
class TestStorage(hishel.BaseStorage):
320
    """Test storage that tracks operations"""
321
    
322
    def __init__(self):
323
        super().__init__()
324
        self.stored_keys = []
325
        self.retrieved_keys = []
326
        self.cache = {}
327
    
328
    def store(self, key, response, request, metadata=None):
329
        self.stored_keys.append(key)
330
        self.cache[key] = (response, request, metadata)
331
    
332
    def retrieve(self, key):
333
        self.retrieved_keys.append(key)
334
        return self.cache.get(key)
335
    
336
    def remove(self, key):
337
        self.cache.pop(key, None)
338
    
339
    def update_metadata(self, key, response, request, metadata):
340
        if key in self.cache:
341
            self.cache[key] = (response, request, metadata)
342
    
343
    def close(self):
344
        pass
345

346
# Use in tests
347
def test_with_custom_storage():
348
    test_storage = TestStorage()
349
    
350
    response = httpx.Response(200, json={"test": True})
351
    mock_transport = hishel.MockTransport()
352
    mock_transport.add_responses([response])
353
    
354
    cache_transport = hishel.CacheTransport(
355
        transport=mock_transport,
356
        storage=test_storage
357
    )
358
    
359
    with httpx.Client(transport=cache_transport) as client:
360
        client.get("https://api.example.com/data")
361
        client.get("https://api.example.com/data")  # Cache hit
362
    
363
    assert len(test_storage.stored_keys) == 1
364
    assert len(test_storage.retrieved_keys) == 1
365
```