• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python -u
2#
3# Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6#
7
8#pylint: disable-msg=C0111
9
10import mox, os, shutil, tempfile, unittest
11
12from django.conf import settings
13
14import common
15from autotest_lib.client.common_lib import global_config
16from autotest_lib.frontend import database_settings_helper
17from autotest_lib.frontend import setup_django_environment
18from autotest_lib.frontend import setup_test_environment
19from autotest_lib.frontend.afe import frontend_test_utils
20from autotest_lib.frontend.afe import models as django_afe_models
21from autotest_lib.frontend.tko import models as django_tko_models
22from autotest_lib.tko import db as tko_db
23from autotest_lib.tko.site_parse import StackTrace
24
25# Have to import this after setup_django_environment and setup_test_environment.
26# It creates a database connection, so the mocking has to be done first.
27from django.db import connections
28
29class stack_trace_test(unittest.TestCase):
30
31
32    def setUp(self):
33        setup_test_environment.set_up()
34        self._fake_results = tempfile.mkdtemp()
35        self._cros_src_dir = global_config.global_config.get_config_value(
36            'CROS', 'source_tree', default=None)
37
38        if not self._cros_src_dir:
39            self.fail('No Chrome OS source tree defined in global_config.ini')
40
41        self._stack_trace = StackTrace(
42            self._fake_results, self._cros_src_dir)
43
44        self._cache_dir = os.path.join(
45            self._cros_src_dir, 'chroot', self._stack_trace._CACHE_DIR)
46
47        # Ensure we don't obliterate a live cache directory by accident.
48        if os.path.exists(self._cache_dir):
49            self.fail(
50                'Symbol cache directory already exists. Cowardly refusing to'
51                ' run. Please remove this directory manually to continue.')
52
53
54    def tearDown(self):
55        setup_test_environment.tear_down()
56        shutil.rmtree(self._fake_results)
57        if os.path.exists(self._cache_dir):
58            shutil.rmtree(self._cache_dir)
59
60
61    def _setup_basic_cache(self,
62                           job_name='x86-alex-r16-R16-1166.0.0-a1-b1118_bvt',
63                           mkdir=True):
64        # Ensure cache directory is present.
65        self._stack_trace._get_cache_dir()
66        board, rev, version = self._stack_trace._parse_job_name(job_name)
67
68        symbols_dir = os.path.join(
69            self._cache_dir, '-'.join([board, rev, version]))
70        if mkdir:
71            os.mkdir(symbols_dir)
72
73        chroot_symbols_dir = os.sep + os.path.relpath(
74            symbols_dir, self._stack_trace._chroot_dir)
75
76        return job_name, symbols_dir, chroot_symbols_dir
77
78
79    def test_get_job_name(self):
80        job_name = 'x86-alex-r16-R16-1166.0.0-a1-b1118_regression'
81        with open(os.path.join(self._fake_results, 'keyval'), 'w') as f:
82            f.write('label=%s' % job_name)
83
84        self.assertEqual(self._stack_trace._get_job_name(), job_name)
85
86
87    def test_parse_3_tuple_job_name(self):
88        job_name = 'x86-alex-r16-R16-1166.0.0-a1-b1118_regression'
89        board, rev, version = self._stack_trace._parse_job_name(job_name)
90        self.assertEqual(board, 'x86-alex')
91        self.assertEqual(rev, 'r16')
92        self.assertEqual(version, '1166.0.0')
93
94
95    def test_parse_4_tuple_job_name(self):
96        job_name = 'x86-mario-r15-0.15.1011.74-a1-b61_bvt'
97        board, rev, version = self._stack_trace._parse_job_name(job_name)
98        self.assertEqual(board, 'x86-mario')
99        self.assertEqual(rev, 'r15')
100        self.assertEqual(version, '0.15.1011.74')
101
102
103    def test_parse_4_tuple_au_job_name(self):
104        job_name = 'x86-alex-r15-0.15.1011.81_to_0.15.1011.82-a1-b69_mton_au'
105        board, rev, version = self._stack_trace._parse_job_name(job_name)
106        self.assertEqual(board, 'x86-alex')
107        self.assertEqual(rev, 'r15')
108        self.assertEqual(version, '0.15.1011.82')
109
110
111    def test_parse_3_tuple_au_job_name(self):
112        job_name = 'x86-alex-r16-1165.0.0_to_R16-1166.0.0-a1-b69_mton_au'
113        board, rev, version = self._stack_trace._parse_job_name(job_name)
114        self.assertEqual(board, 'x86-alex')
115        self.assertEqual(rev, 'r16')
116        self.assertEqual(version, '1166.0.0')
117
118
119class database_selection_test(mox.MoxTestBase,
120                              frontend_test_utils.FrontendTestMixin):
121
122    def setUp(self):
123        super(database_selection_test, self).setUp()
124        self._frontend_common_setup(fill_data=False)
125
126
127    def tearDown(self):
128        super(database_selection_test, self).tearDown()
129        self._frontend_common_teardown()
130        global_config.global_config.reset_config_values()
131
132
133    def assertQueries(self, database, assert_in, assert_not_in):
134        assert_in_found = False
135        for query in connections[database].queries:
136            sql = query['sql']
137            # Ignore CREATE TABLE statements as they are always executed
138            if 'INSERT INTO' in sql or 'SELECT' in sql:
139                self.assertNotIn(assert_not_in, sql)
140                if assert_in in sql:
141                    assert_in_found = True
142        self.assertTrue(assert_in_found)
143
144
145    def testDjangoModels(self):
146        # If DEBUG=False connection.query will be empty
147        settings.DEBUG = True
148
149        afe_job = django_afe_models.Job.objects.create(created_on='2014-08-12')
150        # Machine has less dependencies than tko Job so it's easier to create
151        tko_job = django_tko_models.Machine.objects.create()
152
153        django_afe_models.Job.objects.get(pk=afe_job.id)
154        django_tko_models.Machine.objects.get(pk=tko_job.pk)
155
156        self.assertQueries('global', 'tko_machines', 'afe_jobs')
157        self.assertQueries('default', 'afe_jobs', 'tko_machines')
158
159        # Avoid unnecessary debug output from other tests
160        settings.DEBUG = True
161
162
163    def testRunOnShardWithoutGlobalConfigsFails(self):
164        global_config.global_config.override_config_value(
165                'SHARD', 'shard_hostname', 'host1')
166        from autotest_lib.frontend import settings
167        # settings module was already loaded during the imports of this file,
168        # so before the configuration setting was made, therefore reload it:
169        reload(database_settings_helper)
170        self.assertRaises(global_config.ConfigError,
171                          reload, settings)
172
173
174    def testRunOnMasterWithoutGlobalConfigsWorks(self):
175        global_config.global_config.override_config_value(
176                'SHARD', 'shard_hostname', '')
177        from autotest_lib.frontend import settings
178        # settings module was already loaded during the imports of this file,
179        # so before the configuration setting was made, therefore reload it:
180        reload(database_settings_helper)
181        reload(settings)
182
183
184    def testTkoDatabase(self):
185        global_host = 'GLOBAL_HOST'
186        global_user = 'GLOBAL_USER'
187        global_db = 'GLOBAL_DB'
188        global_pw = 'GLOBAL_PW'
189        global_port = ''
190        local_host = 'LOCAL_HOST'
191
192        global_config.global_config.override_config_value(
193                'AUTOTEST_WEB', 'global_db_type', '')
194
195        global_config.global_config.override_config_value(
196                'AUTOTEST_WEB', 'global_db_host', global_host)
197        global_config.global_config.override_config_value(
198                'AUTOTEST_WEB', 'global_db_database', global_db)
199        global_config.global_config.override_config_value(
200                'AUTOTEST_WEB', 'global_db_user', global_user)
201        global_config.global_config.override_config_value(
202                'AUTOTEST_WEB', 'global_db_password', global_pw)
203        global_config.global_config.override_config_value(
204                'AUTOTEST_WEB', 'host', local_host)
205
206        class ConnectCalledException(Exception):
207            pass
208
209        # We're only interested in the parameters connect is called with here.
210        # Take the fast path out so we don't have to mock all the other calls
211        # that will later be made on the connection
212        def fake_connect(*args, **kwargs):
213            raise ConnectCalledException
214
215        tko_db.db_sql.connect = None
216        self.mox.StubOutWithMock(tko_db.db_sql, 'connect')
217        tko_db.db_sql.connect(
218                global_host, global_db, global_user, global_pw,
219                global_port).WithSideEffects(fake_connect)
220
221        self.mox.ReplayAll()
222
223        self.assertRaises(ConnectCalledException, tko_db.db_sql)
224
225
226if __name__ == "__main__":
227    unittest.main()
228