RO/osm_ro/tests/test_db.py

   1 # -*- coding: utf-8 -*-
   2 ##
   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
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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
  13 # under the License.
  14 ##
  15
  16 # pylint: disable=E1101
  17 import unittest
  18
  19 from MySQLdb import connect, cursors, DatabaseError, IntegrityError
  20
  21 from ..db_base import retry, with_transaction
  22 from ..nfvo_db import nfvo_db
  23 from .db_helpers import TestCaseWithDatabase
  24
  25
  26 class TestDbDecorators(TestCaseWithDatabase):
  27     @classmethod
  28     def setUpClass(cls):
  29         connection = connect(cls.host, cls.user, cls.password)
  30         cursor = connection.cursor()
  31         cursor.execute(
  32             "CREATE DATABASE IF NOT EXISTS {};".format(
  33                 connection.escape_string(cls.database)))
  34         cursor.execute("use {};".format(cls.database))
  35         cursor.execute("""\
  36             CREATE TABLE IF NOT EXISTS `test_table` (\
  37                 `id` int(11) NOT NULL,
  38                 PRIMARY KEY (`id`)\
  39             );\
  40         """)
  41         cursor.close()
  42         connection.close()
  43
  44     @classmethod
  45     def tearDownClass(cls):
  46         cls.empty_database()
  47
  48     def setUp(self):
  49         self.maxDiff = None
  50         self.db = nfvo_db(self.host, self.user, self.password, self.database)
  51         self.db.connect()
  52         self.addCleanup(lambda: self.db.disconnect())
  53
  54     def db_run(self, query, cursor=None):
  55         cursor = cursor or self.db.con.cursor()
  56         cursor.execute(query)
  57         return cursor.fetchone()
  58
  59     def test_retry_inject_attempt(self):
  60         @retry
  61         def _fn(db, attempt=None):
  62             self.assertIsNotNone(attempt)
  63             self.assertEqual(attempt.number, 1)
  64
  65         _fn(self.db)
  66
  67     def test_retry_accept_max_attempts(self):
  68         success = []
  69         failures = []
  70
  71         @retry(max_attempts=5)
  72         def _fn(db, attempt=None):
  73             if attempt.count < 4:
  74                 failures.append(attempt.count)
  75                 raise DatabaseError("Emulate DB error", "msg")
  76             success.append(attempt.count)
  77
  78         _fn(self.db)
  79         self.assertEqual(failures, [0, 1, 2, 3])
  80         self.assertEqual(success, [4])
  81
  82     def test_retry_reconnect_auctomatically(self):
  83         success = []
  84         failures = []
  85
  86         @retry(max_attempts=3)
  87         def _fn(db, attempt=None):
  88             if attempt.count < 2:
  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)
  93             return result
  94
  95         result = _fn(self.db)
  96         self.assertEqual(failures, [0, 1])
  97         self.assertEqual(success, [2])
  98         self.assertEqual(result, (2, 4))
  99
 100     def test_retry_reraise_non_db_errors(self):
 101         failures = []
 102
 103         @retry
 104         def _fn(db, attempt=None):
 105             failures.append(attempt.count)
 106             raise SystemError("Non Correlated Error")
 107
 108         with self.assertRaises(SystemError):
 109             _fn(self.db)
 110
 111         self.assertEqual(failures, [0])
 112
 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
 122
 123         count = self.db_run('select count(*) FROM test_table')
 124         self.assertEqual(count, (0,))
 125
 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',
 129                                 cursor)
 130
 131         self.assertEqual(count, {'counter': 0})
 132
 133
 134 if __name__ == '__main__':
 135     unittest.main()