0
# Engine Management
1
2
Engine control and state management functions for controlling pook's HTTP interception capabilities, networking modes, and mock lifecycle. These functions provide fine-grained control over when and how pook intercepts HTTP traffic.
3
4
## Capabilities
5
6
### Engine Activation and Deactivation
7
8
Controls when pook starts and stops intercepting HTTP traffic, with support for both programmatic control and decorator patterns.
9
10
```python { .api }
11
def activate(fn=None):
12
"""
13
Enables HTTP traffic interceptors. Can be used as a decorator or called directly.
14
15
Parameters:
16
- fn (function|coroutinefunction, optional): Function to decorate when used as decorator
17
18
Returns:
19
function: Decorator wrapper when used as decorator, None otherwise
20
"""
21
22
def on(fn=None):
23
"""
24
Alias to activate(). Enables HTTP traffic interceptors.
25
26
Parameters:
27
- fn (function|coroutinefunction, optional): Function to decorate when used as decorator
28
29
Returns:
30
function: Decorator wrapper when used as decorator, None otherwise
31
"""
32
33
def disable():
34
"""
35
Disables HTTP traffic interceptors without flushing mocks.
36
37
The mocks remain registered and can be reactivated later.
38
"""
39
40
def off():
41
"""
42
Disables mock engine, HTTP interceptors, and flushes all registered mocks.
43
44
This is equivalent to calling both disable() and reset().
45
"""
46
```
47
48
Usage examples:
49
50
```python
51
import pook
52
import requests
53
54
# Direct activation/deactivation
55
pook.activate()
56
pook.get('https://api.example.com/data').reply(200).json({'data': []})
57
58
response = requests.get('https://api.example.com/data') # Intercepted
59
print(response.json())
60
61
pook.disable() # Stop intercepting
62
pook.activate() # Resume with same mocks
63
64
pook.off() # Stop and clear all mocks
65
66
# Using as decorator
67
@pook.activate
68
def test_with_mocks():
69
pook.get('https://api.example.com/test').reply(200)
70
response = requests.get('https://api.example.com/test')
71
assert response.status_code == 200
72
# Pook automatically deactivated after function
73
74
@pook.on # Alias to activate
75
def another_test():
76
pook.post('https://api.example.com/create').reply(201)
77
# Test implementation
78
79
# Async function support
80
@pook.activate
81
async def async_test():
82
pook.get('https://api.example.com/async').reply(200)
83
# Async HTTP client calls will be intercepted
84
```
85
86
### Engine State Management
87
88
Functions for resetting and managing the internal state of pook's mock engine.
89
90
```python { .api }
91
def reset():
92
"""
93
Resets current mock engine state, flushing all registered mocks
94
without disabling the engine.
95
96
The engine remains active if it was previously activated.
97
"""
98
99
def engine():
100
"""
101
Returns the current running mock engine instance.
102
103
Returns:
104
Engine: Current engine instance for direct manipulation
105
"""
106
107
def set_mock_engine(engine):
108
"""
109
Sets a custom mock engine, replacing the built-in one.
110
111
Parameters:
112
- engine (MockEngine): Custom mock engine to use
113
"""
114
```
115
116
Usage examples:
117
118
```python
119
import pook
120
121
# Engine state management
122
pook.activate()
123
pook.get('https://api.example.com/test1').reply(200)
124
pook.get('https://api.example.com/test2').reply(200)
125
126
print(f"Pending mocks: {pook.pending()}") # 2
127
128
pook.reset() # Clear mocks but keep engine active
129
print(f"Pending mocks: {pook.pending()}") # 0
130
print(f"Engine active: {pook.isactive()}") # True
131
132
# Direct engine access
133
current_engine = pook.engine()
134
print(f"Engine networking mode: {current_engine.networking}")
135
print(f"Number of mocks: {len(current_engine.mocks)}")
136
137
# Custom engine (advanced usage)
138
from pook import Engine, MockEngine
139
custom_engine = MockEngine(Engine(network=True))
140
pook.set_mock_engine(custom_engine)
141
```
142
143
### Context Managers
144
145
Context manager functions for creating isolated mock scopes with automatic cleanup and optional networking support.
146
147
```python { .api }
148
def use(network=False):
149
"""
150
Creates a new isolated mock engine to be used via context manager.
151
152
Parameters:
153
- network (bool, optional): Enable networking mode. Defaults to False
154
155
Returns:
156
Context manager yielding Engine instance
157
"""
158
159
def use_network():
160
"""
161
Creates a new mock engine with networking enabled as context manager.
162
163
Equivalent to use(network=True).
164
165
Returns:
166
Context manager yielding Engine instance
167
"""
168
```
169
170
Usage examples:
171
172
```python
173
import pook
174
import requests
175
176
# Isolated mock scope
177
with pook.use() as engine:
178
# All mocks created here are isolated
179
pook.get('https://api.example.com/isolated').reply(200).json({'isolated': True})
180
181
response = requests.get('https://api.example.com/isolated')
182
assert response.json()['isolated'] == True
183
184
# Check engine state
185
print(f"Mocks in this engine: {len(engine.mocks)}")
186
187
# Outside context, mocks are gone and engine is restored
188
print(f"Global mocks: {pook.pending()}") # 0
189
190
# Network-enabled context
191
with pook.use_network() as engine:
192
# Only some requests are mocked, others hit real network
193
pook.get('https://httpbin.org/json').reply(200).json({'mocked': True})
194
195
# This hits the mock
196
mock_response = requests.get('https://httpbin.org/json')
197
print(mock_response.json()) # {'mocked': True}
198
199
# This hits real network (if allowed)
200
real_response = requests.get('https://httpbin.org/ip')
201
print(real_response.json()) # Real response from httpbin
202
203
# Nested contexts
204
with pook.use() as outer_engine:
205
pook.get('https://api.example.com/outer').reply(200)
206
207
with pook.use() as inner_engine:
208
pook.get('https://api.example.com/inner').reply(200)
209
# Inner context has different mocks
210
211
# Back to outer context
212
```
213
214
### Network Mode Control
215
216
Functions for controlling when pook allows real network traffic for unmatched requests.
217
218
```python { .api }
219
def enable_network(*hostnames):
220
"""
221
Enables real networking mode for unmatched mocks in the current mock engine.
222
223
Parameters:
224
- *hostnames (str): Optional hostnames to enable networking for.
225
If not provided, enables for all hosts.
226
"""
227
228
def disable_network():
229
"""
230
Disables real traffic networking mode in the current mock engine.
231
232
All unmatched requests will be blocked/rejected.
233
"""
234
235
def use_network_filter(*fn):
236
"""
237
Adds network filters to determine if certain outgoing unmatched
238
HTTP traffic can establish real network connections.
239
240
Parameters:
241
- *fn (function): Filter functions that return True to allow network access
242
"""
243
```
244
245
Usage examples:
246
247
```python
248
import pook
249
import requests
250
251
# Enable networking for all hosts
252
pook.activate()
253
pook.enable_network()
254
255
# Mock some requests, allow others through
256
pook.get('https://api.example.com/mock-me').reply(200).json({'mocked': True})
257
258
# This is mocked
259
mock_response = requests.get('https://api.example.com/mock-me')
260
print(mock_response.json()) # {'mocked': True}
261
262
# This goes to real network
263
real_response = requests.get('https://httpbin.org/ip')
264
print(real_response.json()) # Real IP response
265
266
# Enable networking for specific hosts only
267
pook.disable_network()
268
pook.enable_network('httpbin.org', 'api.github.com')
269
270
# This goes to real network
271
requests.get('https://httpbin.org/json') # Allowed
272
273
# This would be blocked
274
# requests.get('https://api.example.com/real') # Would raise exception
275
276
# Network filtering with custom logic
277
def allow_safe_hosts(request):
278
"""Allow requests to known safe hosts."""
279
safe_hosts = ['httpbin.org', 'jsonplaceholder.typicode.com']
280
return any(host in request.url for host in safe_hosts)
281
282
def allow_get_only(request):
283
"""Only allow GET requests through."""
284
return request.method.upper() == 'GET'
285
286
pook.use_network_filter(allow_safe_hosts, allow_get_only)
287
288
# Only GET requests to safe hosts will be allowed through
289
requests.get('https://httpbin.org/get') # Allowed
290
# requests.post('https://httpbin.org/post') # Blocked (POST method)
291
# requests.get('https://evil.com/data') # Blocked (unsafe host)
292
```
293
294
### Advanced Engine Patterns
295
296
Complex engine management scenarios and patterns for sophisticated testing setups.
297
298
```python
299
import pook
300
from pook import Engine, MockEngine
301
302
# Custom engine configuration
303
def create_test_engine():
304
"""Create a custom configured engine for testing."""
305
engine = Engine(network=True)
306
engine.debug = True # Enable debug mode
307
return MockEngine(engine)
308
309
# Use custom engine
310
original_engine = pook.engine()
311
test_engine = create_test_engine()
312
pook.set_mock_engine(test_engine)
313
314
try:
315
# Test with custom engine
316
pook.activate()
317
pook.get('https://api.test.com/data').reply(200)
318
319
# Your tests here
320
321
finally:
322
# Restore original engine
323
pook.set_mock_engine(original_engine)
324
325
# Conditional engine setup
326
def setup_test_environment(use_network=False, debug=False):
327
"""Setup test environment with specific engine configuration."""
328
if use_network:
329
with pook.use_network() as engine:
330
if debug:
331
engine.debug = True
332
yield engine
333
else:
334
with pook.use() as engine:
335
if debug:
336
engine.debug = True
337
yield engine
338
339
# Usage
340
with setup_test_environment(use_network=True, debug=True) as engine:
341
# Test in network-enabled, debug mode
342
pook.get('https://api.example.com/test').reply(200)
343
# Some requests mocked, others hit real network
344
345
# Engine state inspection and management
346
@pook.activate
347
def complex_test_scenario():
348
"""Test scenario with multiple mock phases."""
349
350
# Phase 1: Setup initial mocks
351
pook.get('https://api.example.com/setup').reply(200).json({'phase': 1})
352
353
# Verify setup
354
assert pook.pending() == 1
355
assert pook.isactive()
356
357
# Execute phase 1
358
response1 = requests.get('https://api.example.com/setup')
359
assert response1.json()['phase'] == 1
360
361
# Phase 2: Reset and add new mocks
362
pook.reset() # Clear mocks but keep engine active
363
pook.post('https://api.example.com/execute').reply(201).json({'phase': 2})
364
365
# Execute phase 2
366
response2 = requests.post('https://api.example.com/execute')
367
assert response2.json()['phase'] == 2
368
369
# Verify all mocks were consumed
370
assert pook.isdone()
371
```
372
373
## Engine Class Reference
374
375
```python { .api }
376
class Engine:
377
"""
378
Engine represents the mock interceptor and matcher engine responsible
379
for triggering interceptors and matching outgoing HTTP traffic.
380
"""
381
382
def __init__(self, network=False):
383
"""
384
Parameters:
385
- network (bool, optional): Enables/disables real networking mode
386
"""
387
388
# Key attributes
389
debug: bool # Enables/disables debug mode
390
active: bool # Current engine activation status
391
networking: bool # Current engine networking mode status
392
mocks: list # List of registered Mock instances
393
filters: list # Engine-level mock filter functions
394
mappers: list # Engine-level mock mapper functions
395
unmatched_reqs: list # Unmatched outgoing HTTP requests
396
network_filters: list # Real networking mode filters
397
398
def activate(self):
399
"""Activates registered interceptors."""
400
401
def disable(self):
402
"""Disables interceptors."""
403
404
def reset(self):
405
"""Resets and flushes engine state."""
406
407
def mock(self, url=None, **kw):
408
"""
409
Creates and registers new HTTP mock.
410
411
Parameters:
412
- url (str, optional): Request URL to mock
413
- **kw: Additional Mock constructor arguments
414
415
Returns:
416
Mock: New mock instance
417
"""
418
419
def enable_network(self, *hostnames):
420
"""
421
Enables real networking mode.
422
423
Parameters:
424
- *hostnames (str): Optional hostnames to enable networking for
425
"""
426
427
def disable_network(self):
428
"""Disables real networking mode."""
429
430
class MockEngine:
431
"""
432
MockEngine represents the low-level mocking engine abstraction layer
433
between pook and the underlying mocking mechanism.
434
"""
435
436
def __init__(self, engine):
437
"""
438
Parameters:
439
- engine (Engine): Injected pook engine to be used
440
"""
441
442
engine: Engine # Stores pook engine to be used
443
interceptors: list # HTTP traffic interceptors
444
445
def activate(self):
446
"""Activates registered interceptors."""
447
448
def disable(self):
449
"""Disables interceptors."""
450
```