0
# Plugin System
1
2
Extensible plugin architecture supporting HTTP/REST, MQTT, and gRPC protocols with standardized interfaces for requests, responses, and session management.
3
4
## Capabilities
5
6
### Base Request Interface
7
8
Abstract base class defining the interface for all request implementations across different protocols.
9
10
```python { .api }
11
class BaseRequest:
12
"""
13
Abstract base class for protocol-specific request implementations.
14
"""
15
16
def __init__(
17
self,
18
session: Any,
19
rspec: dict,
20
test_block_config: TestConfig
21
) -> None:
22
"""
23
Initialize request with session, specification, and configuration.
24
25
Parameters:
26
- session: Protocol-specific session object
27
- rspec: Request specification dictionary from YAML
28
- test_block_config: Test configuration and variables
29
"""
30
31
@property
32
def request_vars(self) -> box.Box:
33
"""
34
Variables used in the request for templating in subsequent stages.
35
36
Returns:
37
box.Box: Boxed request variables accessible via dot notation
38
"""
39
40
def run(self):
41
"""
42
Execute the request and return response.
43
44
Returns:
45
Protocol-specific response object
46
"""
47
```
48
49
### Base Response Interface
50
51
Abstract base class for response verification and validation across all protocols.
52
53
```python { .api }
54
@dataclasses.dataclass
55
class BaseResponse:
56
"""
57
Abstract base class for protocol-specific response verification.
58
"""
59
60
name: str
61
expected: Any
62
test_block_config: TestConfig
63
response: Optional[Any] = None
64
validate_functions: list[Any] = dataclasses.field(init=False, default_factory=list)
65
errors: list[str] = dataclasses.field(init=False, default_factory=list)
66
67
def verify(self, response):
68
"""
69
Verify response against expected values and return saved variables.
70
71
Parameters:
72
- response: Actual response object from request execution
73
74
Returns:
75
Dictionary of variables to save for future test stages
76
77
Raises:
78
TestFailError: If response verification fails
79
"""
80
81
def recurse_check_key_match(
82
self,
83
expected_block: Optional[Mapping],
84
block: Mapping,
85
blockname: str,
86
strict: StrictOption,
87
) -> None:
88
"""
89
Recursively validate response data against expected data.
90
91
Parameters:
92
- expected_block: Expected data structure
93
- block: Actual response data
94
- blockname: Name of the block being validated (for error messages)
95
- strict: Strictness level for validation
96
"""
97
```
98
99
## Built-in Plugins
100
101
### HTTP/REST Plugin
102
103
Default plugin for HTTP/REST API testing using the requests library.
104
105
```python { .api }
106
class TavernRestPlugin:
107
"""
108
HTTP/REST protocol plugin using requests library.
109
"""
110
111
session_type = "requests.Session"
112
request_type = "RestRequest"
113
request_block_name = "request"
114
verifier_type = "RestResponse"
115
response_block_name = "response"
116
```
117
118
**Entry Point Configuration:**
119
```
120
[project.entry-points.tavern_http]
121
requests = "tavern._plugins.rest.tavernhook:TavernRestPlugin"
122
```
123
124
**Usage in YAML:**
125
```yaml
126
test_name: HTTP API test
127
stages:
128
- name: Make HTTP request
129
request: # Maps to request_block_name
130
url: https://api.example.com/users
131
method: POST
132
json:
133
name: "John Doe"
134
email: "john@example.com"
135
headers:
136
Authorization: "Bearer {token}"
137
response: # Maps to response_block_name
138
status_code: 201
139
json:
140
id: !anyint
141
name: "John Doe"
142
save:
143
json:
144
user_id: id
145
```
146
147
### MQTT Plugin
148
149
Plugin for testing MQTT publish/subscribe operations using paho-mqtt.
150
151
```python { .api }
152
# MQTT Plugin Configuration
153
session_type = "MQTTClient"
154
request_type = "MQTTRequest"
155
request_block_name = "mqtt_publish"
156
verifier_type = "MQTTResponse"
157
response_block_name = "mqtt_response"
158
```
159
160
**Entry Point Configuration:**
161
```
162
[project.entry-points.tavern_mqtt]
163
paho-mqtt = "tavern._plugins.mqtt.tavernhook"
164
```
165
166
**Usage in YAML:**
167
```yaml
168
test_name: MQTT publish/subscribe test
169
stages:
170
- name: Publish and verify message
171
mqtt_publish: # Maps to request_block_name
172
topic: "sensor/temperature"
173
payload:
174
value: 23.5
175
unit: "celsius"
176
timestamp: "{timestamp}"
177
qos: 1
178
mqtt_response: # Maps to response_block_name
179
topic: "sensor/temperature/ack"
180
payload:
181
status: "received"
182
message_id: !anystr
183
timeout: 5
184
```
185
186
### gRPC Plugin
187
188
Plugin for testing gRPC services with protocol buffer support.
189
190
```python { .api }
191
# gRPC Plugin Configuration
192
session_type = "GRPCClient"
193
request_type = "GRPCRequest"
194
request_block_name = "grpc_request"
195
verifier_type = "GRPCResponse"
196
response_block_name = "grpc_response"
197
```
198
199
**Entry Point Configuration:**
200
```
201
[project.entry-points.tavern_grpc]
202
grpc = "tavern._plugins.grpc.tavernhook"
203
```
204
205
**Usage in YAML:**
206
```yaml
207
test_name: gRPC service test
208
stages:
209
- name: Call gRPC method
210
grpc_request: # Maps to request_block_name
211
service: "UserService"
212
method: "CreateUser"
213
body:
214
name: "John Doe"
215
email: "john@example.com"
216
metadata:
217
authorization: "Bearer {token}"
218
grpc_response: # Maps to response_block_name
219
status: "OK"
220
body:
221
id: !anyint
222
name: "John Doe"
223
created_at: !anything
224
```
225
226
## Plugin Development
227
228
### Creating Custom Plugins
229
230
**Step 1: Implement Request Class**
231
232
```python
233
from tavern.request import BaseRequest
234
import box
235
236
class CustomProtocolRequest(BaseRequest):
237
def __init__(self, session, rspec, test_block_config):
238
self.session = session
239
self.rspec = rspec
240
self.test_block_config = test_block_config
241
self._request_vars = {}
242
243
@property
244
def request_vars(self) -> box.Box:
245
return box.Box(self._request_vars)
246
247
def run(self):
248
# Implement protocol-specific request logic
249
endpoint = self.rspec['endpoint']
250
data = self.rspec.get('data', {})
251
252
# Store variables for templating
253
self._request_vars.update({
254
'endpoint': endpoint,
255
'request_time': time.time()
256
})
257
258
# Execute request and return response
259
return self.session.send_request(endpoint, data)
260
```
261
262
**Step 2: Implement Response Class**
263
264
```python
265
from tavern.response import BaseResponse
266
267
@dataclasses.dataclass
268
class CustomProtocolResponse(BaseResponse):
269
def verify(self, response):
270
saved_variables = {}
271
272
# Validate response
273
if self.expected.get('success') is not None:
274
if response.success != self.expected['success']:
275
self._adderr("Expected success=%s, got %s",
276
self.expected['success'], response.success)
277
278
# Save variables for future stages
279
if 'save' in self.expected:
280
for key, path in self.expected['save'].items():
281
saved_variables[key] = getattr(response, path)
282
283
if self.errors:
284
raise TestFailError(
285
f"Response verification failed:\n{self._str_errors()}"
286
)
287
288
return saved_variables
289
```
290
291
**Step 3: Create Plugin Hook**
292
293
```python
294
from tavern._core.plugins import PluginHelperBase
295
296
class CustomProtocolPlugin(PluginHelperBase):
297
session_type = CustomProtocolSession
298
request_type = CustomProtocolRequest
299
request_block_name = "custom_request"
300
verifier_type = CustomProtocolResponse
301
response_block_name = "custom_response"
302
```
303
304
**Step 4: Register Plugin Entry Point**
305
306
```python
307
# setup.py or pyproject.toml
308
entry_points = {
309
'tavern_custom': {
310
'my_protocol = mypackage.plugin:CustomProtocolPlugin'
311
}
312
}
313
```
314
315
### Plugin Configuration
316
317
**Backend Selection:**
318
```bash
319
# Use custom plugin via CLI
320
tavern-ci --tavern-custom-backend my_protocol test.yaml
321
322
# Or programmatically
323
from tavern.core import run
324
run("test.yaml", tavern_custom_backend="my_protocol")
325
```
326
327
**Plugin Registration Discovery:**
328
```python
329
# Tavern discovers plugins via entry points
330
import pkg_resources
331
332
def discover_plugins():
333
plugins = {}
334
for entry_point in pkg_resources.iter_entry_points('tavern_custom'):
335
plugins[entry_point.name] = entry_point.load()
336
return plugins
337
```
338
339
## Plugin Architecture
340
341
### Session Management
342
343
Each plugin manages protocol-specific sessions:
344
345
```python
346
# HTTP plugin uses requests.Session
347
session = requests.Session()
348
session.headers.update({'User-Agent': 'Tavern/2.17.0'})
349
350
# MQTT plugin uses paho-mqtt client
351
client = mqtt.Client()
352
client.on_connect = handle_connect
353
client.connect(broker_host, broker_port)
354
355
# Custom plugin session
356
class CustomSession:
357
def __init__(self, **config):
358
self.connection = create_connection(config)
359
360
def send_request(self, endpoint, data):
361
return self.connection.call(endpoint, data)
362
```
363
364
### Request/Response Flow
365
366
```python
367
# 1. Plugin creates session
368
session = plugin.session_type(**session_config)
369
370
# 2. Plugin creates request
371
request = plugin.request_type(session, request_spec, test_config)
372
373
# 3. Execute request
374
response = request.run()
375
376
# 4. Plugin creates response verifier
377
verifier = plugin.verifier_type(
378
name=stage_name,
379
expected=expected_spec,
380
test_block_config=test_config,
381
response=response
382
)
383
384
# 5. Verify response and extract variables
385
saved_vars = verifier.verify(response)
386
```
387
388
### Error Handling
389
390
```python
391
class CustomProtocolError(TavernException):
392
"""Custom protocol-specific errors."""
393
pass
394
395
class CustomRequest(BaseRequest):
396
def run(self):
397
try:
398
return self.session.send_request(...)
399
except ConnectionError as e:
400
raise CustomProtocolError(f"Connection failed: {e}")
401
except TimeoutError as e:
402
raise CustomProtocolError(f"Request timeout: {e}")
403
```
404
405
## Plugin Examples
406
407
### Database Plugin Example
408
409
```python
410
@dataclasses.dataclass
411
class DatabaseRequest(BaseRequest):
412
def run(self):
413
query = self.rspec['query']
414
params = self.rspec.get('params', {})
415
416
cursor = self.session.execute(query, params)
417
return cursor.fetchall()
418
419
@dataclasses.dataclass
420
class DatabaseResponse(BaseResponse):
421
def verify(self, rows):
422
if 'row_count' in self.expected:
423
if len(rows) != self.expected['row_count']:
424
self._adderr("Expected %d rows, got %d",
425
self.expected['row_count'], len(rows))
426
427
return {'result_count': len(rows), 'first_row': rows[0] if rows else None}
428
429
# Usage in YAML:
430
# db_query:
431
# query: "SELECT * FROM users WHERE age > ?"
432
# params: [18]
433
# db_response:
434
# row_count: !anyint
435
```
436
437
## Types
438
439
```python { .api }
440
import dataclasses
441
from typing import Any, Optional, Mapping
442
from abc import abstractmethod
443
import box
444
445
TestConfig = "tavern._core.pytest.config.TestConfig"
446
StrictOption = "tavern._core.strict_util.StrictOption"
447
PluginHelperBase = "tavern._core.plugins.PluginHelperBase"
448
TestFailError = "tavern._core.exceptions.TestFailError"
449
```