1 # -*- coding: utf-8 -*-
3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
4 # not use this file except in compliance with the License. You may obtain
5 # a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 # License for the specific language governing permissions and limitations
16 # pylint: disable=E1101
19 from MySQLdb
import connect
, cursors
, DatabaseError
, IntegrityError
21 from ..db_base
import retry
, with_transaction
22 from ..nfvo_db
import nfvo_db
23 from .db_helpers
import TestCaseWithDatabase
26 class TestDbDecorators(TestCaseWithDatabase
):
29 connection
= connect(cls
.host
, cls
.user
, cls
.password
)
30 cursor
= connection
.cursor()
32 "CREATE DATABASE IF NOT EXISTS {};".format(
33 connection
.escape_string(cls
.database
)))
34 cursor
.execute("use {};".format(cls
.database
))
36 CREATE TABLE IF NOT EXISTS `test_table` (\
37 `id` int(11) NOT NULL,
45 def tearDownClass(cls
):
50 self
.db
= nfvo_db(self
.host
, self
.user
, self
.password
, self
.database
)
52 self
.addCleanup(lambda: self
.db
.disconnect())
54 def db_run(self
, query
, cursor
=None):
55 cursor
= cursor
or self
.db
.con
.cursor()
57 return cursor
.fetchone()
59 def test_retry_inject_attempt(self
):
61 def _fn(db
, attempt
=None):
62 self
.assertIsNotNone(attempt
)
63 self
.assertEqual(attempt
.number
, 1)
67 def test_retry_accept_max_attempts(self
):
71 @retry(max_attempts
=5)
72 def _fn(db
, attempt
=None):
74 failures
.append(attempt
.count
)
75 raise DatabaseError("Emulate DB error", "msg")
76 success
.append(attempt
.count
)
79 self
.assertEqual(failures
, [0, 1, 2, 3])
80 self
.assertEqual(success
, [4])
82 def test_retry_reconnect_auctomatically(self
):
86 @retry(max_attempts
=3)
87 def _fn(db
, attempt
=None):
89 failures
.append(attempt
.count
)
90 db
.con
.close() # Simulate connection failure
91 result
= self
.db_run('select 1+1, 2+2;')
92 success
.append(attempt
.count
)
96 self
.assertEqual(failures
, [0, 1])
97 self
.assertEqual(success
, [2])
98 self
.assertEqual(result
, (2, 4))
100 def test_retry_reraise_non_db_errors(self
):
104 def _fn(db
, attempt
=None):
105 failures
.append(attempt
.count
)
106 raise SystemError("Non Correlated Error")
108 with self
.assertRaises(SystemError):
111 self
.assertEqual(failures
, [0])
113 def test_transaction_rollback(self
):
114 with self
.assertRaises(IntegrityError
), \
115 self
.db
.transaction() as cursor
:
116 # The first row is created normally
117 self
.db_run('insert into test_table (id) values (1)', cursor
)
118 # The second row fails due to repeated id
119 self
.db_run('insert into test_table (id) values (1)', cursor
)
120 # The entire transaction will rollback then, and therefore the
121 # first operation will be undone
123 count
= self
.db_run('select count(*) FROM test_table')
124 self
.assertEqual(count
, (0,))
126 def test_transaction_cursor(self
):
127 with self
.db
.transaction(cursors
.DictCursor
) as cursor
:
128 count
= self
.db_run('select count(*) as counter FROM test_table',
131 self
.assertEqual(count
, {'counter': 0})
134 if __name__
== '__main__':