• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python
2
3import unittest
4import common
5from autotest_lib.client.common_lib.test_utils import mock
6from autotest_lib.database import migrate, db_utils
7
8class 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
80if __name__ == '__main__':
81    unittest.main()
82