e152347e1b665355f301bdd58a38f9e79412b8c9
1 # -*- coding: utf-8 -*-
2 # pylint: disable=E1101
5 from MySQLdb
import connect
, cursors
, DatabaseError
, IntegrityError
9 from ..db_base
import retry
, with_transaction
10 from ..nfvo_db
import nfvo_db
11 from .db_helpers
import TestCaseWithDatabase
14 class TestDbDecorators(TestCaseWithDatabase
):
17 connection
= connect(cls
.host
, cls
.user
, cls
.password
)
18 cursor
= connection
.cursor()
20 "CREATE DATABASE IF NOT EXISTS {};".format(
21 connection
.escape_string(cls
.database
)))
22 cursor
.execute("use {};".format(cls
.database
))
24 CREATE TABLE IF NOT EXISTS `test_table` (\
25 `id` int(11) NOT NULL,
33 def tearDownClass(cls
):
38 self
.db
= nfvo_db(self
.host
, self
.user
, self
.password
, self
.database
)
40 self
.addCleanup(lambda: self
.db
.disconnect())
42 def db_run(self
, query
, cursor
=None):
43 cursor
= cursor
or self
.db
.con
.cursor()
45 return cursor
.fetchone()
47 def test_retry_inject_attempt(self
):
49 def _fn(db
, attempt
=None):
50 self
.assertIsNotNone(attempt
)
51 self
.assertEqual(attempt
.number
, 1)
55 def test_retry_accept_max_attempts(self
):
59 @retry(max_attempts
=5)
60 def _fn(db
, attempt
=None):
62 failures
.append(attempt
.count
)
63 raise DatabaseError("Emulate DB error", "msg")
64 success
.append(attempt
.count
)
67 self
.assertEqual(failures
, [0, 1, 2, 3])
68 self
.assertEqual(success
, [4])
70 def test_retry_reconnect_auctomatically(self
):
74 @retry(max_attempts
=3)
75 def _fn(db
, attempt
=None):
77 failures
.append(attempt
.count
)
78 db
.con
.close() # Simulate connection failure
79 result
= self
.db_run('select 1+1, 2+2;')
80 success
.append(attempt
.count
)
84 self
.assertEqual(failures
, [0, 1])
85 self
.assertEqual(success
, [2])
86 self
.assertEqual(result
, (2, 4))
88 def test_retry_reraise_non_db_errors(self
):
92 def _fn(db
, attempt
=None):
93 failures
.append(attempt
.count
)
94 raise SystemError("Non Correlated Error")
96 with self
.assertRaises(SystemError):
99 self
.assertEqual(failures
, [0])
101 def test_transaction_rollback(self
):
102 with self
.assertRaises(IntegrityError
), \
103 self
.db
.transaction() as cursor
:
104 # The first row is created normally
105 self
.db_run('insert into test_table (id) values (1)', cursor
)
106 # The second row fails due to repeated id
107 self
.db_run('insert into test_table (id) values (1)', cursor
)
108 # The entire transaction will rollback then, and therefore the
109 # first operation will be undone
111 count
= self
.db_run('select count(*) FROM test_table')
112 self
.assertEqual(count
, (0,))
114 def test_transaction_cursor(self
):
115 with self
.db
.transaction(cursors
.DictCursor
) as cursor
:
116 count
= self
.db_run('select count(*) as counter FROM test_table',
119 self
.assertEqual(count
, {'counter': 0})
122 if __name__
== '__main__':