0
# Testing Utilities
1
2
Comprehensive test utilities for asserting Prometheus metric values and changes in test suites. Enables precise testing of monitoring functionality and metric behavior.
3
4
## Capabilities
5
6
### Metric Assertion Functions
7
8
Functions for testing metric values and changes in unit tests and integration tests.
9
10
```python { .api }
11
def assert_metric_equal(expected_value, metric_name: str, registry=REGISTRY, **labels):
12
"""
13
Asserts that metric_name{**labels} == expected_value.
14
15
Parameters:
16
- expected_value: Expected metric value (int or float)
17
- metric_name: str, name of the Prometheus metric
18
- registry: Prometheus registry to query (default: REGISTRY)
19
- **labels: Label filters as keyword arguments
20
21
Raises:
22
AssertionError: If metric value doesn't match expected value
23
"""
24
25
def assert_metric_not_equal(expected_value, metric_name: str, registry=REGISTRY, **labels):
26
"""
27
Asserts that metric_name{**labels} != expected_value.
28
29
Parameters:
30
- expected_value: Value that metric should NOT equal
31
- metric_name: str, name of the Prometheus metric
32
- registry: Prometheus registry to query (default: REGISTRY)
33
- **labels: Label filters as keyword arguments
34
35
Raises:
36
AssertionError: If metric value equals expected value
37
"""
38
39
def assert_metric_diff(frozen_registry, expected_diff, metric_name: str, registry=REGISTRY, **labels):
40
"""
41
Asserts that metric_name{**labels} changed by expected_diff between
42
the frozen registry and current state.
43
44
Parameters:
45
- frozen_registry: Previously saved registry state from save_registry()
46
- expected_diff: Expected change in metric value (int or float)
47
- metric_name: str, name of the Prometheus metric
48
- registry: Prometheus registry to query (default: REGISTRY)
49
- **labels: Label filters as keyword arguments
50
51
Raises:
52
AssertionError: If metric change doesn't match expected difference
53
"""
54
55
def assert_metric_no_diff(frozen_registry, expected_diff, metric_name: str, registry=REGISTRY, **labels):
56
"""
57
Asserts that metric_name{**labels} did NOT change by expected_diff
58
between the frozen registry and current state.
59
60
Parameters:
61
- frozen_registry: Previously saved registry state from save_registry()
62
- expected_diff: Difference that should NOT have occurred
63
- metric_name: str, name of the Prometheus metric
64
- registry: Prometheus registry to query (default: REGISTRY)
65
- **labels: Label filters as keyword arguments
66
67
Raises:
68
AssertionError: If metric change matches the expected difference
69
"""
70
71
def assert_metric_compare(frozen_registry, predicate, metric_name: str, registry=REGISTRY, **labels):
72
"""
73
Asserts that metric_name{**labels} changed according to a provided
74
predicate function between the frozen registry and current state.
75
76
Parameters:
77
- frozen_registry: Previously saved registry state from save_registry()
78
- predicate: Function that takes (old_value, new_value) and returns bool
79
- metric_name: str, name of the Prometheus metric
80
- registry: Prometheus registry to query (default: REGISTRY)
81
- **labels: Label filters as keyword arguments
82
83
Raises:
84
AssertionError: If predicate returns False for the metric change
85
"""
86
```
87
88
### Registry Management Functions
89
90
Functions for managing metric registry state during testing.
91
92
```python { .api }
93
def save_registry(registry=REGISTRY):
94
"""
95
Freezes a registry for later comparison.
96
97
This lets a user test changes to a metric instead of testing
98
the absolute value. Typical usage:
99
100
registry = save_registry()
101
doStuff()
102
assert_metric_diff(registry, 1, 'stuff_done_total')
103
104
Parameters:
105
- registry: Prometheus registry to freeze (default: REGISTRY)
106
107
Returns:
108
Frozen registry state (deep copy of metric samples)
109
"""
110
```
111
112
### Metric Query Functions
113
114
Functions for retrieving metric values and information during testing.
115
116
```python { .api }
117
def get_metric(metric_name: str, registry=REGISTRY, **labels):
118
"""
119
Gets the current value of a single metric.
120
121
Parameters:
122
- metric_name: str, name of the Prometheus metric
123
- registry: Prometheus registry to query (default: REGISTRY)
124
- **labels: Label filters as keyword arguments
125
126
Returns:
127
Metric value (int/float) or None if metric not found
128
"""
129
130
def get_metrics_vector(metric_name: str, registry=REGISTRY):
131
"""
132
Returns the values for all label combinations of a given metric.
133
134
Parameters:
135
- metric_name: str, name of the Prometheus metric
136
- registry: Prometheus registry to query (default: REGISTRY)
137
138
Returns:
139
List of (labels_dict, value) tuples for all label combinations
140
"""
141
142
def get_metric_from_frozen_registry(metric_name: str, frozen_registry, **labels):
143
"""
144
Gets a single metric value from a previously frozen registry.
145
146
Parameters:
147
- metric_name: str, name of the Prometheus metric
148
- frozen_registry: Previously saved registry state
149
- **labels: Label filters as keyword arguments
150
151
Returns:
152
Metric value (int/float) or None if metric not found
153
"""
154
155
def get_metric_vector_from_frozen_registry(metric_name: str, frozen_registry):
156
"""
157
Gets all label combinations and values for a metric from frozen registry.
158
159
Parameters:
160
- metric_name: str, name of the Prometheus metric
161
- frozen_registry: Previously saved registry state
162
163
Returns:
164
List of (labels_dict, value) tuples for all label combinations
165
"""
166
```
167
168
### Formatting Utilities
169
170
Helper functions for formatting metric information in test output.
171
172
```python { .api }
173
def format_labels(labels: dict):
174
"""
175
Format a set of labels to Prometheus representation.
176
177
Parameters:
178
- labels: dict, label key-value pairs
179
180
Returns:
181
str, formatted labels like '{method="GET",port="80"}'
182
183
Example:
184
>>> format_labels({'method': 'GET', 'port': '80'})
185
'{method="GET",port="80"}'
186
"""
187
188
def format_vector(vector):
189
"""
190
Formats a list of (labels, value) tuples into human-readable representation.
191
192
Parameters:
193
- vector: List of (labels_dict, value) tuples
194
195
Returns:
196
str, formatted vector with one metric per line
197
"""
198
```
199
200
## Usage Examples
201
202
### Basic Metric Testing
203
204
```python
205
import pytest
206
from django.test import TestCase
207
from django_prometheus.testutils import (
208
assert_metric_equal, assert_metric_diff, save_registry
209
)
210
211
class MetricsTestCase(TestCase):
212
def test_request_counter(self):
213
"""Test that HTTP requests increment the counter."""
214
# Save initial state
215
registry = save_registry()
216
217
# Make a request
218
response = self.client.get('/')
219
220
# Assert counter increased by 1
221
assert_metric_diff(
222
registry, 1,
223
'django_http_requests_total_by_method',
224
method='GET'
225
)
226
227
# Assert final count
228
assert_metric_equal(
229
1,
230
'django_http_requests_total_by_method',
231
method='GET'
232
)
233
```
234
235
### Model Operation Testing
236
237
```python
238
from myapp.models import User
239
from django_prometheus.testutils import assert_metric_diff, save_registry
240
241
class ModelMetricsTestCase(TestCase):
242
def test_model_operations(self):
243
"""Test that model operations are tracked."""
244
registry = save_registry()
245
246
# Create a user
247
User.objects.create(username='test_user')
248
249
# Assert insert counter increased
250
assert_metric_diff(
251
registry, 1,
252
'django_model_inserts_total',
253
model='user'
254
)
255
256
# Update the user
257
user = User.objects.get(username='test_user')
258
user.email = 'test@example.com'
259
user.save()
260
261
# Assert update counter increased
262
assert_metric_diff(
263
registry, 1,
264
'django_model_updates_total',
265
model='user'
266
)
267
```
268
269
### Cache Testing
270
271
```python
272
from django.core.cache import cache
273
from django_prometheus.testutils import assert_metric_diff, save_registry
274
275
class CacheMetricsTestCase(TestCase):
276
def test_cache_operations(self):
277
"""Test cache hit/miss tracking."""
278
registry = save_registry()
279
280
# Cache miss
281
value = cache.get('nonexistent_key')
282
assert value is None
283
284
assert_metric_diff(
285
registry, 1,
286
'django_cache_get_total',
287
backend='locmem'
288
)
289
assert_metric_diff(
290
registry, 1,
291
'django_cache_get_misses_total',
292
backend='locmem'
293
)
294
295
# Cache set and hit
296
cache.set('test_key', 'test_value')
297
value = cache.get('test_key')
298
299
assert_metric_diff(
300
registry, 2, # Total gets: miss + hit
301
'django_cache_get_total',
302
backend='locmem'
303
)
304
assert_metric_diff(
305
registry, 1,
306
'django_cache_get_hits_total',
307
backend='locmem'
308
)
309
```
310
311
### Custom Predicate Testing
312
313
```python
314
from django_prometheus.testutils import assert_metric_compare, save_registry
315
316
class CustomMetricsTestCase(TestCase):
317
def test_response_time_increase(self):
318
"""Test that response times are recorded."""
319
registry = save_registry()
320
321
# Perform operation that should increase response time metric
322
response = self.client.get('/slow-endpoint/')
323
324
# Custom predicate to check response time increased
325
def response_time_increased(old_val, new_val):
326
return (new_val or 0) > (old_val or 0)
327
328
assert_metric_compare(
329
registry,
330
response_time_increased,
331
'django_http_requests_latency_seconds_by_view_method',
332
view='slow_endpoint',
333
method='GET'
334
)
335
```
336
337
### Multiple Label Testing
338
339
```python
340
from django_prometheus.testutils import get_metrics_vector, assert_metric_equal
341
342
class MultiLabelTestCase(TestCase):
343
def test_multiple_status_codes(self):
344
"""Test metrics with multiple label combinations."""
345
# Generate different status codes
346
self.client.get('/') # 200
347
self.client.get('/nonexistent/') # 404
348
349
# Get all status code combinations
350
vector = get_metrics_vector('django_http_responses_total_by_status')
351
352
# Find specific status codes
353
status_200_count = None
354
status_404_count = None
355
356
for labels, value in vector:
357
if labels.get('status') == '200':
358
status_200_count = value
359
elif labels.get('status') == '404':
360
status_404_count = value
361
362
assert status_200_count >= 1
363
assert status_404_count >= 1
364
365
# Or test specific combinations
366
assert_metric_equal(1, 'django_http_responses_total_by_status', status='200')
367
assert_metric_equal(1, 'django_http_responses_total_by_status', status='404')
368
```
369
370
### Error Condition Testing
371
372
```python
373
from django_prometheus.testutils import get_metric, assert_metric_equal
374
375
class ErrorMetricsTestCase(TestCase):
376
def test_nonexistent_metric(self):
377
"""Test behavior with nonexistent metrics."""
378
# Non-existent metric returns None
379
value = get_metric('nonexistent_metric')
380
assert value is None
381
382
# Non-existent labels return None
383
value = get_metric(
384
'django_http_requests_total_by_method',
385
method='NONEXISTENT'
386
)
387
assert value is None
388
389
def test_exception_tracking(self):
390
"""Test that exceptions are tracked in metrics."""
391
registry = save_registry()
392
393
# Trigger an exception in a view
394
with pytest.raises(Exception):
395
self.client.get('/error-endpoint/')
396
397
# Check exception was tracked
398
assert_metric_diff(
399
registry, 1,
400
'django_http_exceptions_total_by_type',
401
type='ValueError' # or whatever exception type
402
)
403
```
404
405
### Performance Testing
406
407
```python
408
from django_prometheus.testutils import save_registry, get_metric
409
import time
410
411
class PerformanceTestCase(TestCase):
412
def test_metric_collection_performance(self):
413
"""Test that metric collection doesn't significantly impact performance."""
414
# Measure time with metrics
415
start = time.time()
416
registry = save_registry()
417
418
for i in range(100):
419
self.client.get('/')
420
421
metrics_time = time.time() - start
422
423
# Verify metrics were collected
424
final_count = get_metric(
425
'django_http_requests_total_by_method',
426
method='GET'
427
)
428
assert final_count >= 100
429
430
# Performance assertion (adjust based on requirements)
431
assert metrics_time < 5.0 # Should complete in under 5 seconds
432
```
433
434
## Testing Best Practices
435
436
### Test Isolation
437
438
```python
439
class IsolatedMetricsTestCase(TestCase):
440
def setUp(self):
441
"""Save registry state before each test."""
442
self.registry = save_registry()
443
444
def test_isolated_operation(self):
445
"""Each test starts with known metric state."""
446
# Test operations knowing the baseline
447
self.client.get('/')
448
assert_metric_diff(self.registry, 1, 'django_http_requests_total_by_method', method='GET')
449
```
450
451
### Async Testing
452
453
```python
454
import asyncio
455
from django.test import TransactionTestCase
456
457
class AsyncMetricsTestCase(TransactionTestCase):
458
async def test_async_operations(self):
459
"""Test metrics with async operations."""
460
registry = save_registry()
461
462
# Async operations that generate metrics
463
await asyncio.gather(
464
self.async_operation_1(),
465
self.async_operation_2(),
466
)
467
468
# Assert metrics were collected
469
assert_metric_diff(registry, 2, 'custom_async_operations_total')
470
```
471
472
### Integration Testing
473
474
```python
475
from django.test import override_settings
476
477
class IntegrationTestCase(TestCase):
478
@override_settings(
479
PROMETHEUS_METRIC_NAMESPACE='test',
480
PROMETHEUS_EXPORT_MIGRATIONS=True
481
)
482
def test_full_integration(self):
483
"""Test complete monitoring integration."""
484
registry = save_registry()
485
486
# Complex workflow that exercises multiple metric types
487
user = User.objects.create(username='integration_test') # Model metrics
488
self.client.force_login(user) # Auth metrics
489
response = self.client.get('/api/data/') # HTTP metrics
490
cache.set('integration_key', 'value') # Cache metrics
491
492
# Verify all metric types were updated
493
assert_metric_diff(registry, 1, 'test_model_inserts_total', model='user')
494
assert_metric_diff(registry, 1, 'test_http_requests_total_by_method', method='GET')
495
# ... other assertions
496
```