]>
git.immae.eu Git - perso/Immae/Projets/Cryptomonnaies/Cryptoportfolio/Trader.git/blob - tests/test_dbs.py
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(**{
12 "db_password": "password",
13 "db_database": "database",
15 psycopg2
.connect
.return_value
= "pg_connection"
16 dbs
.connect_psql(args
)
18 psycopg2
.connect
.assert_called_once_with(host
="host",
19 port
="port", user
="user", password
="password",
21 self
.assertEqual("pg_connection", dbs
.psql
)
22 with self
.assertRaises(AttributeError):
25 psycopg2
.connect
.reset_mock()
26 args
= main
.configargparse
.Namespace(**{
30 "db_password": "password",
31 "db_database": "database",
33 dbs
.connect_psql(args
)
34 psycopg2
.connect
.assert_not_called()
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(**{
42 "redis_database": "database",
44 redis
.Redis
.return_value
= "redis_connection"
45 dbs
.connect_redis(args
)
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):
53 redis
.Redis
.reset_mock()
54 args
= main
.configargparse
.Namespace(**{
57 "redis_database": "database",
59 dbs
.connect_redis(args
)
60 redis
.Redis
.assert_not_called()
63 with self
.subTest(redis_host
="socket"):
64 args
= main
.configargparse
.Namespace(**{
65 "redis_host": "/run/foo",
67 "redis_database": "database",
69 redis
.Redis
.return_value
= "redis_socket"
70 dbs
.connect_redis(args
)
72 redis
.Redis
.assert_called_once_with(unix_socket_path
="/run/foo", db
="database")
73 self
.assertEqual("redis_socket", dbs
.redis
)
75 def test_redis_connected(self
):
76 with self
.subTest(redis
=None):
78 self
.assertFalse(dbs
.redis_connected())
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())
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())
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())
95 def test_psql_connected(self
):
96 with self
.subTest(psql
=None):
98 self
.assertFalse(dbs
.psql_connected())
100 with self
.subTest(psql
="connected"):
101 dbs
.psql
= mock
.Mock()
103 self
.assertTrue(dbs
.psql_connected())
105 with self
.subTest(psql
="not connected"):
106 dbs
.psql
= mock
.Mock()
108 self
.assertFalse(dbs
.psql_connected())