/
base.py
102 lines (78 loc) · 2.78 KB
/
base.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
from __future__ import unicode_literals
import sys
if sys.version_info >= (2, 7):
import unittest
else:
import unittest2 as unittest
from limpyd.database import DEFAULT_CONNECTION_SETTINGS
from limpyd.contrib.database import PipelineDatabase
TEST_CONNECTION_SETTINGS = DEFAULT_CONNECTION_SETTINGS.copy()
TEST_CONNECTION_SETTINGS['db'] = 15
test_database = PipelineDatabase(**TEST_CONNECTION_SETTINGS)
class LimpydBaseTest(unittest.TestCase):
database = test_database
@property
def connection(self):
return self.database.connection
def setUp(self):
# Ensure that we are on the right DB before flushing
current_db_id = self.connection.connection_pool.connection_kwargs['db']
assert current_db_id != DEFAULT_CONNECTION_SETTINGS['db']
assert current_db_id == TEST_CONNECTION_SETTINGS['db']
self.connection.flushdb()
def tearDown(self):
self.connection.flushdb()
def count_commands(self):
"""
Helper method to only count redis commands that work on keys (ie ignore
commands like info...)
"""
return self.connection.info()['total_commands_processed']
def assertNumCommands(self, num, func=None, *args, **kwargs):
"""
A context assert, to use with "with":
with self.assertNumCommands(2):
obj.field.set(1)
obj.field.get()
"""
context = _AssertNumCommandsContext(self, num)
if func is None:
return context
# Basically emulate the `with` statement here.
context.__enter__()
try:
func(*args, **kwargs)
except:
context.__exit__(*sys.exc_info())
raise
else:
context.__exit__(*sys.exc_info())
class _AssertNumCommandsContext(object):
"""
A context to count commands occured
"""
def __init__(self, test_case, num):
self.test_case = test_case
self.num = num
def __enter__(self):
self.starting_commands = self.test_case.count_commands()
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type is not None:
return
# we remove 1 to ignore the "info" called in __enter__
final_commands = self.test_case.count_commands() - 1
executed = final_commands - self.starting_commands
self.test_case.assertEqual(
executed, self.num, "%d commands executed, %d expected" % (
executed, self.num
)
)
class LimpydBaseTestTest(LimpydBaseTest):
"""
Test parts of LimpydBaseTest
"""
def test_assert_num_commands_is_ok(self):
with self.assertNumCommands(1):
# we know that info do only one command
self.connection.info()