xref: /aosp_15_r20/external/autotest/tko/site_parse_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7) 
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