Home | History | Annotate | Download | only in database 
      1 #!/usr/bin/python
      2 
      3 import unittest
      4 import common
      5 from autotest_lib.client.common_lib.test_utils import mock
      6 from autotest_lib.database import migrate, db_utils
      7 
      8 class UtilsTest(unittest.TestCase):
      9 
     10     EXISTS_QUERY_BASE = ('SELECT table_name FROM information_schema.%s '
     11                          'WHERE table_schema = %%s')
     12     DB_NAME = 'test_db'
     13 
     14 
     15     def setUp(self):
     16         self.god = mock.mock_god()
     17         self.manager = self.god.create_mock_class(migrate.MigrationManager,
     18                                                   'manager')
     19 
     20         self.god.stub_function(self.manager, 'execute')
     21         self.god.stub_function(self.manager, 'get_db_name')
     22 
     23 
     24     def tearDown(self):
     25         self.god.unstub_all()
     26 
     27 
     28     def test_check_exists(self):
     29         views = ('view1', 'view2')
     30         def _call_check_exists():
     31             db_utils.check_exists(self.manager, views, db_utils.VIEW_TYPE)
     32 
     33         self._setup_exists_expects(views, 'VIEWS')
     34         _call_check_exists()
     35         self.god.check_playback()
     36 
     37         self._setup_exists_expects(('view1',), 'VIEWS')
     38         self.assertRaises(Exception, _call_check_exists)
     39         self.god.check_playback()
     40 
     41 
     42     def test_drop_views(self):
     43         views = ('view1', 'view2')
     44         self._setup_exists_expects(views, 'VIEWS')
     45 
     46         for view in views:
     47             self.manager.execute.expect_call('DROP VIEW `%s`' % view)
     48 
     49         db_utils.drop_views(self.manager, views)
     50         self.god.check_playback()
     51 
     52 
     53     def test_rename(self):
     54         mapping = {
     55                 'table1' : 'new_table1',
     56                 'table2' : 'new_table2',
     57                 }
     58         self._setup_exists_expects((name for name, _ in mapping.iteritems()),
     59                                    'TABLES')
     60 
     61         for name, new_name in mapping.iteritems():
     62             self.manager.execute.expect_call(
     63                     'RENAME TABLE `%s` TO `%s`' % (name, new_name))
     64 
     65         db_utils.rename(self.manager, mapping)
     66         self.god.check_playback()
     67 
     68 
     69     def _setup_exists_expects(self, names, table):
     70         self.manager.get_db_name.expect_call().and_return(self.DB_NAME)
     71         self.manager.execute.expect_call(
     72                 self.EXISTS_QUERY_BASE % table, self.DB_NAME).and_return(
     73                 self._create_exists_query_result(names))
     74 
     75 
     76     def _create_exists_query_result(self, names):
     77         return ((name, None) for name in names)
     78 
     79 
     80 if __name__ == '__main__':
     81     unittest.main()
     82