osm_ro/tests/test_db.py

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