aboutsummaryrefslogtreecommitdiff
path: root/tests/test_dbs.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_dbs.py')
-rw-r--r--tests/test_dbs.py108
1 files changed, 108 insertions, 0 deletions
diff --git a/tests/test_dbs.py b/tests/test_dbs.py
new file mode 100644
index 0000000..157c423
--- /dev/null
+++ b/tests/test_dbs.py
@@ -0,0 +1,108 @@
1from .helper import *
2import dbs, main
3
4@unittest.skipUnless("unit" in limits, "Unit skipped")
5class 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())