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