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
103
104
105
106
107
108
|
from .helper import *
import dbs, main
@unittest.skipUnless("unit" in limits, "Unit skipped")
class DbsTest(WebMockTestCase):
@mock.patch.object(dbs, "psycopg2")
def test_connect_psql(self, psycopg2):
args = main.configargparse.Namespace(**{
"db_host": "host",
"db_port": "port",
"db_user": "user",
"db_password": "password",
"db_database": "database",
})
psycopg2.connect.return_value = "pg_connection"
dbs.connect_psql(args)
psycopg2.connect.assert_called_once_with(host="host",
port="port", user="user", password="password",
database="database")
self.assertEqual("pg_connection", dbs.psql)
with self.assertRaises(AttributeError):
args.db_password
psycopg2.connect.reset_mock()
args = main.configargparse.Namespace(**{
"db_host": "host",
"db_port": "port",
"db_user": "user",
"db_password": "password",
"db_database": "database",
})
dbs.connect_psql(args)
psycopg2.connect.assert_not_called()
@mock.patch.object(dbs, "_redis")
def test_connect_redis(self, redis):
with self.subTest(redis_host="tcp"):
args = main.configargparse.Namespace(**{
"redis_host": "host",
"redis_port": "port",
"redis_database": "database",
})
redis.Redis.return_value = "redis_connection"
dbs.connect_redis(args)
redis.Redis.assert_called_once_with(host="host",
port="port", db="database")
self.assertEqual("redis_connection", dbs.redis)
with self.assertRaises(AttributeError):
args.redis_database
redis.Redis.reset_mock()
args = main.configargparse.Namespace(**{
"redis_host": "host",
"redis_port": "port",
"redis_database": "database",
})
dbs.connect_redis(args)
redis.Redis.assert_not_called()
dbs.redis = None
with self.subTest(redis_host="socket"):
args = main.configargparse.Namespace(**{
"redis_host": "/run/foo",
"redis_port": "port",
"redis_database": "database",
})
redis.Redis.return_value = "redis_socket"
dbs.connect_redis(args)
redis.Redis.assert_called_once_with(unix_socket_path="/run/foo", db="database")
self.assertEqual("redis_socket", dbs.redis)
def test_redis_connected(self):
with self.subTest(redis=None):
dbs.redis = None
self.assertFalse(dbs.redis_connected())
with self.subTest(redis="mocked_true"):
dbs.redis = mock.Mock()
dbs.redis.ping.return_value = True
self.assertTrue(dbs.redis_connected())
with self.subTest(redis="mocked_false"):
dbs.redis = mock.Mock()
dbs.redis.ping.return_value = False
self.assertFalse(dbs.redis_connected())
with self.subTest(redis="mocked_raise"):
dbs.redis = mock.Mock()
dbs.redis.ping.side_effect = Exception("bouh")
self.assertFalse(dbs.redis_connected())
def test_psql_connected(self):
with self.subTest(psql=None):
dbs.psql = None
self.assertFalse(dbs.psql_connected())
with self.subTest(psql="connected"):
dbs.psql = mock.Mock()
dbs.psql.closed = 0
self.assertTrue(dbs.psql_connected())
with self.subTest(psql="not connected"):
dbs.psql = mock.Mock()
dbs.psql.closed = 3
self.assertFalse(dbs.psql_connected())
|
