1 # -*- coding: utf-8 -*-
2 # pylint: disable=E1101
5 from MySQLdb
import connect
, cursors
, DatabaseError
, IntegrityError
7 from ..db_base
import retry
, with_transaction
8 from ..nfvo_db
import nfvo_db
9 from .db_helpers
import TestCaseWithDatabase
12 class TestDbDecorators(TestCaseWithDatabase
):
15 connection
= connect(cls
.host
, cls
.user
, cls
.password
)
16 cursor
= connection
.cursor()
18 "CREATE DATABASE IF NOT EXISTS {};".format(
19 connection
.escape_string(cls
.database
)))
20 cursor
.execute("use {};".format(cls
.database
))
22 CREATE TABLE IF NOT EXISTS `test_table` (\
23 `id` int(11) NOT NULL,
31 def tearDownClass(cls
):
36 self
.db
= nfvo_db(self
.host
, self
.user
, self
.password
, self
.database
)
38 self
.addCleanup(lambda: self
.db
.disconnect())
40 def db_run(self
, query
, cursor
=None):
41 cursor
= cursor
or self
.db
.con
.cursor()
43 return cursor
.fetchone()
45 def test_retry_inject_attempt(self
):
47 def _fn(db
, attempt
=None):
48 self
.assertIsNotNone(attempt
)
49 self
.assertEqual(attempt
.number
, 1)
53 def test_retry_accept_max_attempts(self
):
57 @retry(max_attempts
=5)
58 def _fn(db
, attempt
=None):
60 failures
.append(attempt
.count
)
61 raise DatabaseError("Emulate DB error", "msg")
62 success
.append(attempt
.count
)
65 self
.assertEqual(failures
, [0, 1, 2, 3])
66 self
.assertEqual(success
, [4])
68 def test_retry_reconnect_auctomatically(self
):
72 @retry(max_attempts
=3)
73 def _fn(db
, attempt
=None):
75 failures
.append(attempt
.count
)
76 db
.con
.close() # Simulate connection failure
77 result
= self
.db_run('select 1+1, 2+2;')
78 success
.append(attempt
.count
)
82 self
.assertEqual(failures
, [0, 1])
83 self
.assertEqual(success
, [2])
84 self
.assertEqual(result
, (2, 4))
86 def test_retry_reraise_non_db_errors(self
):
90 def _fn(db
, attempt
=None):
91 failures
.append(attempt
.count
)
92 raise SystemError("Non Correlated Error")
94 with self
.assertRaises(SystemError):
97 self
.assertEqual(failures
, [0])
99 def test_transaction_rollback(self
):
100 with self
.assertRaises(IntegrityError
), \
101 self
.db
.transaction() as cursor
:
102 # The first row is created normally
103 self
.db_run('insert into test_table (id) values (1)', cursor
)
104 # The second row fails due to repeated id
105 self
.db_run('insert into test_table (id) values (1)', cursor
)
106 # The entire transaction will rollback then, and therefore the
107 # first operation will be undone
109 count
= self
.db_run('select count(*) FROM test_table')
110 self
.assertEqual(count
, (0,))
112 def test_transaction_cursor(self
):
113 with self
.db
.transaction(cursors
.DictCursor
) as cursor
:
114 count
= self
.db_run('select count(*) as counter FROM test_table',
117 self
.assertEqual(count
, {'counter': 0})
120 if __name__
== '__main__':