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