1
+ import pytest
1
2
from django .core .cache import caches
2
- from django .test import TestCase
3
3
from redis import RedisError
4
4
5
5
from django_prometheus .testutils import assert_metric_equal , get_metric
6
6
7
+ _SUPPORTED_CACHES = ["memcached.PyLibMCCache" , "memcached.PyMemcacheCache" , "filebased" , "locmem" , "redis" ]
7
8
8
- class TestCachesMetrics (TestCase ):
9
- """Test django_prometheus.caches metrics."""
10
9
11
- def test_counters (self ):
12
- supported_caches = [
13
- "memcached.PyLibMCCache" ,
14
- "memcached.PyMemcacheCache" ,
15
- "filebased" ,
16
- "locmem" ,
17
- "redis" ,
18
- ]
10
+ class TestCachesMetrics :
11
+ """Test django_prometheus.caches metrics."""
19
12
13
+ @pytest .mark .parametrize ("supported_cache" , _SUPPORTED_CACHES )
14
+ def test_counters (self , supported_cache ):
20
15
# Note: those tests require a memcached server running
21
- for supported_cache in supported_caches :
22
- tested_cache = caches [supported_cache ]
23
- backend = supported_cache .split ("." )[0 ]
24
-
25
- total_before = get_metric ("django_cache_get_total" , backend = backend ) or 0
26
- hit_before = get_metric ("django_cache_get_hits_total" , backend = backend ) or 0
27
- miss_before = get_metric ("django_cache_get_misses_total" , backend = backend ) or 0
28
-
29
- tested_cache .set ("foo1" , "bar" )
30
- tested_cache .get ("foo1" )
31
- tested_cache .get ("foo1" )
32
- tested_cache .get ("foofoo" )
33
-
34
- result = tested_cache .get ("foofoo" , default = "default" )
35
-
36
- assert result == "default"
37
-
38
- assert_metric_equal (total_before + 4 , "django_cache_get_total" , backend = backend )
39
- assert_metric_equal (hit_before + 2 , "django_cache_get_hits_total" , backend = backend )
40
- assert_metric_equal (
41
- miss_before + 2 ,
42
- "django_cache_get_misses_total" ,
43
- backend = backend ,
44
- )
16
+ tested_cache = caches [supported_cache ]
17
+ backend = supported_cache .split ("." )[0 ]
18
+ total_before = get_metric ("django_cache_get_total" , backend = backend ) or 0
19
+ hit_before = get_metric ("django_cache_get_hits_total" , backend = backend ) or 0
20
+ miss_before = get_metric ("django_cache_get_misses_total" , backend = backend ) or 0
21
+ tested_cache .set ("foo1" , "bar" )
22
+ tested_cache .get ("foo1" )
23
+ tested_cache .get ("foo1" )
24
+ tested_cache .get ("foofoo" )
25
+ result = tested_cache .get ("foofoo" , default = "default" )
26
+ assert result == "default"
27
+ assert_metric_equal (total_before + 4 , "django_cache_get_total" , backend = backend )
28
+ assert_metric_equal (hit_before + 2 , "django_cache_get_hits_total" , backend = backend )
29
+ assert_metric_equal (
30
+ miss_before + 2 ,
31
+ "django_cache_get_misses_total" ,
32
+ backend = backend ,
33
+ )
45
34
46
35
def test_redis_cache_fail (self ):
47
36
# Note: test use fake service config (like if server was stopped)
48
37
supported_cache = "redis"
49
-
50
38
total_before = get_metric ("django_cache_get_total" , backend = supported_cache ) or 0
51
39
fail_before = get_metric ("django_cache_get_fail_total" , backend = supported_cache ) or 0
52
40
hit_before = get_metric ("django_cache_get_hits_total" , backend = supported_cache ) or 0
@@ -61,29 +49,19 @@ def test_redis_cache_fail(self):
61
49
assert_metric_equal (fail_before + 1 , "django_cache_get_fail_total" , backend = supported_cache )
62
50
63
51
tested_cache = caches ["stopped_redis" ]
64
- with self . assertRaises (RedisError ):
52
+ with pytest . raises (RedisError ):
65
53
tested_cache .get ("foo1" )
66
54
67
55
assert_metric_equal (hit_before , "django_cache_get_hits_total" , backend = supported_cache )
68
56
assert_metric_equal (miss_before , "django_cache_get_misses_total" , backend = supported_cache )
69
57
assert_metric_equal (total_before + 2 , "django_cache_get_total" , backend = supported_cache )
70
58
assert_metric_equal (fail_before + 2 , "django_cache_get_fail_total" , backend = supported_cache )
71
59
72
- def test_cache_version_support (self ):
73
- supported_caches = [
74
- "memcached.PyLibMCCache" ,
75
- "memcached.PyMemcacheCache" ,
76
- "filebased" ,
77
- "locmem" ,
78
- "redis" ,
79
- ]
80
-
60
+ @pytest .mark .parametrize ("supported_cache" , _SUPPORTED_CACHES )
61
+ def test_cache_version_support (self , supported_cache ):
81
62
# Note: those tests require a memcached server running
82
- for supported_cache in supported_caches :
83
- tested_cache = caches [supported_cache ]
84
-
85
- tested_cache .set ("foo1" , "bar v.1" , version = 1 )
86
- tested_cache .set ("foo1" , "bar v.2" , version = 2 )
87
-
88
- assert "bar v.1" == tested_cache .get ("foo1" , version = 1 )
89
- assert "bar v.2" == tested_cache .get ("foo1" , version = 2 )
63
+ tested_cache = caches [supported_cache ]
64
+ tested_cache .set ("foo1" , "bar v.1" , version = 1 )
65
+ tested_cache .set ("foo1" , "bar v.2" , version = 2 )
66
+ assert "bar v.1" == tested_cache .get ("foo1" , version = 1 )
67
+ assert "bar v.2" == tested_cache .get ("foo1" , version = 2 )
0 commit comments