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