0
# Testing Utilities
1
2
In-memory testing framework for mocking HTTP requests without network access, enabling fast and reliable unit tests for HTTP-based applications. treq.testing provides a complete testing environment that simulates HTTP interactions.
3
4
## Capabilities
5
6
### StubTreq Class
7
8
Main testing client that provides all treq HTTP methods while routing requests to local Twisted resources instead of making network calls.
9
10
```python { .api }
11
class StubTreq:
12
def __init__(self, resource):
13
"""
14
Create a testing HTTP client.
15
16
Provides all the function calls exposed in treq.__all__ (get, post, put,
17
delete, patch, head, request) plus content processing functions.
18
19
Parameters:
20
- resource: Resource - Twisted Resource object providing fake responses
21
"""
22
23
# HTTP methods (same signatures as module-level functions)
24
def get(self, url, **kwargs): ...
25
def post(self, url, data=None, **kwargs): ...
26
def put(self, url, data=None, **kwargs): ...
27
def patch(self, url, data=None, **kwargs): ...
28
def delete(self, url, **kwargs): ...
29
def head(self, url, **kwargs): ...
30
def request(self, method, url, **kwargs): ...
31
32
# Content processing functions
33
def collect(self, response, collector): ...
34
def content(self, response): ...
35
def text_content(self, response, encoding="ISO-8859-1"): ...
36
def json_content(self, response, **kwargs): ...
37
38
def flush(self):
39
"""
40
Process all pending requests.
41
42
StubTreq processes requests synchronously but may defer
43
some operations. Call flush() to ensure all processing
44
is complete.
45
"""
46
```
47
48
### RequestTraversalAgent
49
50
Agent implementation that traverses Twisted resources to generate responses, used internally by StubTreq.
51
52
```python { .api }
53
class RequestTraversalAgent:
54
def __init__(self, resource):
55
"""
56
Agent that traverses a resource tree to handle requests.
57
58
Parameters:
59
- resource: Resource - Root resource for request routing
60
"""
61
62
def request(self, method, uri, headers=None, bodyProducer=None):
63
"""
64
Process request through resource tree.
65
66
Returns:
67
Deferred that fires with IResponse
68
"""
69
70
def flush(self):
71
"""Flush any pending operations."""
72
```
73
74
### StringStubbingResource
75
76
Convenient resource for returning simple string responses with customizable status codes and headers.
77
78
```python { .api }
79
class StringStubbingResource(Resource):
80
def __init__(self, response_string):
81
"""
82
Resource that returns a fixed string response.
83
84
Parameters:
85
- response_string: str or bytes - Response content to return
86
"""
87
88
def render(self, request):
89
"""
90
Handle request and return configured response.
91
92
Returns:
93
bytes - Response content
94
"""
95
```
96
97
### Request Matching Utilities
98
99
Tools for validating that requests match expected patterns during testing.
100
101
```python { .api }
102
class HasHeaders:
103
def __init__(self, headers):
104
"""
105
Matcher for validating request headers.
106
107
Parameters:
108
- headers: dict - Expected headers to match
109
"""
110
111
class RequestSequence:
112
def __init__(self, requests, resource):
113
"""
114
Resource that expects a specific sequence of requests.
115
116
Parameters:
117
- requests: list - Expected sequence of request patterns
118
- resource: Resource - Resource to handle requests after validation
119
"""
120
```
121
122
## Usage Examples
123
124
### Basic Request Stubbing
125
126
```python
127
from treq.testing import StubTreq, StringStubbingResource
128
from twisted.web.resource import Resource
129
from twisted.internet import defer
130
import json
131
132
# Create a simple resource
133
class APIResource(Resource):
134
def render_GET(self, request):
135
return json.dumps({'message': 'Hello, World!'}).encode('utf-8')
136
137
def render_POST(self, request):
138
# Echo back the posted data
139
content = request.content.read()
140
return json.dumps({'received': content.decode('utf-8')}).encode('utf-8')
141
142
@defer.inlineCallbacks
143
def test_basic_requests():
144
# Create stub client
145
resource = APIResource()
146
client = StubTreq(resource)
147
148
# Test GET request
149
response = yield client.get('http://example.com/api')
150
data = yield client.json_content(response)
151
assert data['message'] == 'Hello, World!'
152
153
# Test POST request
154
response = yield client.post(
155
'http://example.com/api',
156
data='test data'
157
)
158
data = yield client.json_content(response)
159
assert data['received'] == 'test data'
160
161
# Ensure all processing is complete
162
client.flush()
163
```
164
165
### Testing with String Responses
166
167
```python
168
from treq.testing import StubTreq, StringStubbingResource
169
170
@defer.inlineCallbacks
171
def test_string_responses():
172
# Simple string resource
173
resource = StringStubbingResource("Hello, World!")
174
client = StubTreq(resource)
175
176
response = yield client.get('http://example.com')
177
text = yield client.text_content(response)
178
assert text == "Hello, World!"
179
180
# JSON string resource
181
json_resource = StringStubbingResource('{"status": "ok"}')
182
json_client = StubTreq(json_resource)
183
184
response = yield json_client.get('http://example.com/status')
185
data = yield json_client.json_content(response)
186
assert data['status'] == 'ok'
187
```
188
189
### Advanced Resource Testing
190
191
```python
192
from twisted.web.resource import Resource
193
from twisted.web import server
194
195
class UserResource(Resource):
196
def __init__(self):
197
Resource.__init__(self)
198
self.users = {}
199
self.user_id_counter = 1
200
201
def render_GET(self, request):
202
# List all users
203
return json.dumps(list(self.users.values())).encode('utf-8')
204
205
def render_POST(self, request):
206
# Create new user
207
data = json.loads(request.content.read().decode('utf-8'))
208
user_id = self.user_id_counter
209
user = {'id': user_id, 'name': data['name'], 'email': data['email']}
210
self.users[user_id] = user
211
self.user_id_counter += 1
212
213
request.setResponseCode(201)
214
return json.dumps(user).encode('utf-8')
215
216
class UserDetailResource(Resource):
217
def __init__(self, users):
218
Resource.__init__(self)
219
self.users = users
220
221
def render_GET(self, request):
222
user_id = int(request.path.decode('utf-8').split('/')[-1])
223
if user_id in self.users:
224
return json.dumps(self.users[user_id]).encode('utf-8')
225
else:
226
request.setResponseCode(404)
227
return b'{"error": "User not found"}'
228
229
@defer.inlineCallbacks
230
def test_rest_api():
231
# Set up resource tree
232
root = Resource()
233
user_resource = UserResource()
234
root.putChild(b'users', user_resource)
235
236
# Create client
237
client = StubTreq(root)
238
239
# Test creating user
240
response = yield client.post(
241
'http://example.com/users',
242
json={'name': 'John Doe', 'email': 'john@example.com'}
243
)
244
assert response.code == 201
245
user = yield client.json_content(response)
246
assert user['name'] == 'John Doe'
247
248
# Test listing users
249
response = yield client.get('http://example.com/users')
250
users = yield client.json_content(response)
251
assert len(users) == 1
252
assert users[0]['name'] == 'John Doe'
253
```
254
255
### Error Response Testing
256
257
```python
258
class ErrorResource(Resource):
259
def render_GET(self, request):
260
error_type = request.args.get(b'error', [b''])[0].decode('utf-8')
261
262
if error_type == '404':
263
request.setResponseCode(404)
264
return b'Not Found'
265
elif error_type == '500':
266
request.setResponseCode(500)
267
return b'Internal Server Error'
268
elif error_type == 'timeout':
269
# Simulate timeout by not responding
270
return server.NOT_DONE_YET
271
else:
272
return b'OK'
273
274
@defer.inlineCallbacks
275
def test_error_handling():
276
client = StubTreq(ErrorResource())
277
278
# Test 404 error
279
response = yield client.get('http://example.com?error=404')
280
assert response.code == 404
281
text = yield client.text_content(response)
282
assert text == 'Not Found'
283
284
# Test 500 error
285
response = yield client.get('http://example.com?error=500')
286
assert response.code == 500
287
288
# Test successful response
289
response = yield client.get('http://example.com')
290
assert response.code == 200
291
```
292
293
### Request Validation Testing
294
295
```python
296
from treq.testing import HasHeaders
297
298
class ValidatingResource(Resource):
299
def render_POST(self, request):
300
# Validate headers
301
if b'content-type' not in request.requestHeaders:
302
request.setResponseCode(400)
303
return b'Missing Content-Type'
304
305
# Validate authentication
306
auth_header = request.getHeader('authorization')
307
if not auth_header or not auth_header.startswith('Bearer '):
308
request.setResponseCode(401)
309
return b'Unauthorized'
310
311
return b'{"status": "success"}'
312
313
@defer.inlineCallbacks
314
def test_request_validation():
315
client = StubTreq(ValidatingResource())
316
317
# Test missing headers
318
response = yield client.post('http://example.com/api', data='test')
319
assert response.code == 400
320
321
# Test with proper headers
322
response = yield client.post(
323
'http://example.com/api',
324
json={'data': 'test'},
325
headers={
326
'Authorization': 'Bearer token123',
327
'Content-Type': 'application/json'
328
}
329
)
330
assert response.code == 200
331
```
332
333
### Asynchronous Resource Testing
334
335
```python
336
from twisted.internet import defer
337
338
class AsyncResource(Resource):
339
@defer.inlineCallbacks
340
def render_GET(self, request):
341
# Simulate async operation
342
yield defer.sleep(0.1) # Simulated delay
343
344
# Return response
345
request.write(b'{"async": "response"}')
346
request.finish()
347
348
defer.returnValue(server.NOT_DONE_YET)
349
350
@defer.inlineCallbacks
351
def test_async_resource():
352
client = StubTreq(AsyncResource())
353
354
response = yield client.get('http://example.com/async')
355
data = yield client.json_content(response)
356
assert data['async'] == 'response'
357
358
# Ensure async operations complete
359
client.flush()
360
```
361
362
### Integration Testing Pattern
363
364
```python
365
class TestHTTPClient:
366
def setUp(self):
367
self.resource = self.create_test_resource()
368
self.client = StubTreq(self.resource)
369
370
def create_test_resource(self):
371
root = Resource()
372
api = Resource()
373
root.putChild(b'api', api)
374
375
# Add various endpoints
376
api.putChild(b'users', UserResource())
377
api.putChild(b'posts', PostResource())
378
return root
379
380
@defer.inlineCallbacks
381
def test_user_workflow(self):
382
# Create user
383
user_response = yield self.client.post(
384
'http://example.com/api/users',
385
json={'name': 'Test User'}
386
)
387
user = yield self.client.json_content(user_response)
388
389
# Create post for user
390
post_response = yield self.client.post(
391
'http://example.com/api/posts',
392
json={'title': 'Test Post', 'user_id': user['id']}
393
)
394
post = yield self.client.json_content(post_response)
395
396
# Verify relationships
397
assert post['user_id'] == user['id']
398
399
self.client.flush()
400
```
401
402
## Types
403
404
Testing-related types:
405
406
```python { .api }
407
# Resource types from Twisted
408
from twisted.web.resource import Resource
409
410
# Agent interface
411
from twisted.web.iweb import IAgent
412
413
# Request type
414
from twisted.web.server import Request
415
416
# Testing utilities
417
StubTreqType = StubTreq
418
RequestTraversalAgentType = RequestTraversalAgent
419
```
420
421
## Testing Best Practices
422
423
### Resource Design
424
425
- **Stateless resources**: Design resources to be stateless when possible
426
- **Deterministic responses**: Ensure consistent behavior across test runs
427
- **Error simulation**: Include error conditions in test resources
428
- **Realistic delays**: Use `defer.sleep()` for timing-sensitive tests
429
430
### Test Organization
431
432
- **Setup/teardown**: Use proper test setup and cleanup
433
- **Resource reuse**: Share common resources across related tests
434
- **Isolation**: Ensure tests don't interfere with each other
435
- **Assertions**: Use clear, specific assertions for request/response validation
436
437
### Performance Considerations
438
439
- **Memory usage**: Clean up resources after tests
440
- **Test speed**: StubTreq eliminates network latency for fast tests
441
- **Flush operations**: Always call `flush()` to ensure completion
442
- **Resource complexity**: Keep test resources simple and focused
443
444
### Common Patterns
445
446
- **Mock external APIs**: Replace external service calls with controlled responses
447
- **Error condition testing**: Systematically test error scenarios
448
- **Authentication testing**: Verify auth flows without real credentials
449
- **Integration testing**: Test complete request/response workflows