]>
Commit | Line | Data |
---|---|---|
30700830 IB |
1 | from .helper import * |
2 | import dbs, main | |
3 | ||
4 | @unittest.skipUnless("unit" in limits, "Unit skipped") | |
5 | class DbsTest(WebMockTestCase): | |
6 | @mock.patch.object(dbs, "psycopg2") | |
7 | def test_connect_psql(self, psycopg2): | |
8 | args = main.configargparse.Namespace(**{ | |
9 | "db_host": "host", | |
10 | "db_port": "port", | |
11 | "db_user": "user", | |
12 | "db_password": "password", | |
13 | "db_database": "database", | |
14 | }) | |
15 | psycopg2.connect.return_value = "pg_connection" | |
16 | dbs.connect_psql(args) | |
17 | ||
18 | psycopg2.connect.assert_called_once_with(host="host", | |
19 | port="port", user="user", password="password", | |
20 | database="database") | |
21 | self.assertEqual("pg_connection", dbs.psql) | |
22 | with self.assertRaises(AttributeError): | |
23 | args.db_password | |
24 | ||
25 | psycopg2.connect.reset_mock() | |
26 | args = main.configargparse.Namespace(**{ | |
27 | "db_host": "host", | |
28 | "db_port": "port", | |
29 | "db_user": "user", | |
30 | "db_password": "password", | |
31 | "db_database": "database", | |
32 | }) | |
33 | dbs.connect_psql(args) | |
34 | psycopg2.connect.assert_not_called() | |
35 | ||
36 | @mock.patch.object(dbs, "_redis") | |
37 | def test_connect_redis(self, redis): | |
38 | with self.subTest(redis_host="tcp"): | |
39 | args = main.configargparse.Namespace(**{ | |
40 | "redis_host": "host", | |
41 | "redis_port": "port", | |
42 | "redis_database": "database", | |
43 | }) | |
44 | redis.Redis.return_value = "redis_connection" | |
45 | dbs.connect_redis(args) | |
46 | ||
47 | redis.Redis.assert_called_once_with(host="host", | |
48 | port="port", db="database") | |
49 | self.assertEqual("redis_connection", dbs.redis) | |
50 | with self.assertRaises(AttributeError): | |
51 | args.redis_database | |
52 | ||
53 | redis.Redis.reset_mock() | |
54 | args = main.configargparse.Namespace(**{ | |
55 | "redis_host": "host", | |
56 | "redis_port": "port", | |
57 | "redis_database": "database", | |
58 | }) | |
59 | dbs.connect_redis(args) | |
60 | redis.Redis.assert_not_called() | |
61 | ||
62 | dbs.redis = None | |
63 | with self.subTest(redis_host="socket"): | |
64 | args = main.configargparse.Namespace(**{ | |
65 | "redis_host": "/run/foo", | |
66 | "redis_port": "port", | |
67 | "redis_database": "database", | |
68 | }) | |
69 | redis.Redis.return_value = "redis_socket" | |
70 | dbs.connect_redis(args) | |
71 | ||
72 | redis.Redis.assert_called_once_with(unix_socket_path="/run/foo", db="database") | |
73 | self.assertEqual("redis_socket", dbs.redis) | |
74 | ||
75 | def test_redis_connected(self): | |
76 | with self.subTest(redis=None): | |
77 | dbs.redis = None | |
78 | self.assertFalse(dbs.redis_connected()) | |
79 | ||
80 | with self.subTest(redis="mocked_true"): | |
81 | dbs.redis = mock.Mock() | |
82 | dbs.redis.ping.return_value = True | |
83 | self.assertTrue(dbs.redis_connected()) | |
84 | ||
85 | with self.subTest(redis="mocked_false"): | |
86 | dbs.redis = mock.Mock() | |
87 | dbs.redis.ping.return_value = False | |
88 | self.assertFalse(dbs.redis_connected()) | |
89 | ||
90 | with self.subTest(redis="mocked_raise"): | |
91 | dbs.redis = mock.Mock() | |
92 | dbs.redis.ping.side_effect = Exception("bouh") | |
93 | self.assertFalse(dbs.redis_connected()) | |
94 | ||
95 | def test_psql_connected(self): | |
96 | with self.subTest(psql=None): | |
97 | dbs.psql = None | |
98 | self.assertFalse(dbs.psql_connected()) | |
99 | ||
100 | with self.subTest(psql="connected"): | |
101 | dbs.psql = mock.Mock() | |
102 | dbs.psql.closed = 0 | |
103 | self.assertTrue(dbs.psql_connected()) | |
104 | ||
105 | with self.subTest(psql="not connected"): | |
106 | dbs.psql = mock.Mock() | |
107 | dbs.psql.closed = 3 | |
108 | self.assertFalse(dbs.psql_connected()) |