| #!/usr/bin/python -u |
| # |
| # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| # |
| |
| import glob, os, shutil, tempfile, time, unittest |
| |
| import common |
| from autotest_lib.client.bin import utils |
| from autotest_lib.client.common_lib import global_config |
| from autotest_lib.client.common_lib.test_utils import mock |
| from autotest_lib.tko.site_parse import StackTrace |
| |
| |
| class stack_trace_test(unittest.TestCase): |
| |
| |
| def setUp(self): |
| self._fake_results = tempfile.mkdtemp() |
| self._cros_src_dir = global_config.global_config.get_config_value( |
| 'CROS', 'source_tree', default=None) |
| |
| if not self._cros_src_dir: |
| self.fail('No Chrome OS source tree defined in global_config.ini') |
| |
| self._stack_trace = StackTrace( |
| self._fake_results, self._cros_src_dir) |
| |
| self._cache_dir = os.path.join( |
| self._cros_src_dir, 'chroot', self._stack_trace._CACHE_DIR) |
| |
| # Ensure we don't obliterate a live cache directory by accident. |
| if os.path.exists(self._cache_dir): |
| self.fail( |
| 'Symbol cache directory already exists. Cowardly refusing to' |
| ' run. Please remove this directory manually to continue.') |
| |
| |
| def tearDown(self): |
| shutil.rmtree(self._fake_results) |
| if os.path.exists(self._cache_dir): |
| shutil.rmtree(self._cache_dir) |
| |
| |
| def _setup_basic_cache(self, |
| job_name='x86-alex-r16-R16-1166.0.0-a1-b1118_bvt', |
| mkdir=True): |
| # Ensure cache directory is present. |
| self._stack_trace._get_cache_dir() |
| board, rev, version = self._stack_trace._parse_job_name(job_name) |
| |
| symbols_dir = os.path.join( |
| self._cache_dir, '-'.join([board, rev, version])) |
| if mkdir: |
| os.mkdir(symbols_dir) |
| |
| chroot_symbols_dir = os.sep + os.path.relpath( |
| symbols_dir, self._stack_trace._chroot_dir) |
| |
| return job_name, symbols_dir, chroot_symbols_dir |
| |
| |
| def test_get_job_name(self): |
| job_name = 'x86-alex-r16-R16-1166.0.0-a1-b1118_regression' |
| with open(os.path.join(self._fake_results, 'keyval'), 'w') as f: |
| f.write('label=%s' % job_name) |
| |
| self.assertEqual(self._stack_trace._get_job_name(), job_name) |
| |
| |
| def test_parse_3_tuple_job_name(self): |
| job_name = 'x86-alex-r16-R16-1166.0.0-a1-b1118_regression' |
| board, rev, version = self._stack_trace._parse_job_name(job_name) |
| self.assertEqual(board, 'x86-alex') |
| self.assertEqual(rev, 'r16') |
| self.assertEqual(version, '1166.0.0') |
| |
| |
| def test_parse_4_tuple_job_name(self): |
| job_name = 'x86-mario-r15-0.15.1011.74-a1-b61_bvt' |
| board, rev, version = self._stack_trace._parse_job_name(job_name) |
| self.assertEqual(board, 'x86-mario') |
| self.assertEqual(rev, 'r15') |
| self.assertEqual(version, '0.15.1011.74') |
| |
| |
| def test_parse_4_tuple_au_job_name(self): |
| job_name = 'x86-alex-r15-0.15.1011.81_to_0.15.1011.82-a1-b69_mton_au' |
| board, rev, version = self._stack_trace._parse_job_name(job_name) |
| self.assertEqual(board, 'x86-alex') |
| self.assertEqual(rev, 'r15') |
| self.assertEqual(version, '0.15.1011.82') |
| |
| |
| def test_parse_3_tuple_au_job_name(self): |
| job_name = 'x86-alex-r16-1165.0.0_to_R16-1166.0.0-a1-b69_mton_au' |
| board, rev, version = self._stack_trace._parse_job_name(job_name) |
| self.assertEqual(board, 'x86-alex') |
| self.assertEqual(rev, 'r16') |
| self.assertEqual(version, '1166.0.0') |
| |
| |
| def test_get_cache_dir(self): |
| # Test without existing cache dir. |
| self.assertFalse(os.path.exists(self._cache_dir)) |
| self.assertEquals(self._stack_trace._get_cache_dir(), self._cache_dir) |
| self.assertTrue(os.path.exists(self._cache_dir)) |
| |
| # Test with existing cache dir. |
| self.assertEquals(self._stack_trace._get_cache_dir(), self._cache_dir) |
| |
| |
| def test_get_symbol_dir_with_cache(self): |
| job_name, symbols_dir, chroot_symbols_dir = self._setup_basic_cache() |
| |
| # Create completed file. |
| open(os.path.join( |
| symbols_dir, self._stack_trace._COMPLETE_FILE), 'w').close() |
| self.assertEqual( |
| self._stack_trace._get_symbol_dir(job_name), chroot_symbols_dir) |
| |
| |
| def test_get_symbol_dir_with_incomplete_cache(self): |
| # Create symbol dir w/o completed file. |
| job_name, symbols_dir, chroot_symbols_dir = self._setup_basic_cache() |
| |
| self.god = mock.mock_god() |
| self.god.stub_function(utils, 'poll_for_condition') |
| utils.poll_for_condition.expect_any_call() |
| |
| self.assertEqual( |
| self._stack_trace._get_symbol_dir(job_name), chroot_symbols_dir) |
| |
| self.god.unstub_all() |
| |
| |
| def test_get_symbol_dir_with_incomplete_cache_timeout(self): |
| # Create symbol dir w/o completed file. |
| job_name, symbols_dir, chroot_symbols_dir = self._setup_basic_cache() |
| |
| # Adjust timeout to be very low. Ensure exception is raised. |
| self._stack_trace._SYMBOL_WAIT_TIMEOUT = 1 |
| self.assertRaises( |
| utils.TimeoutError, self._stack_trace._get_symbol_dir, job_name) |
| |
| |
| def test_trim_cache(self): |
| # Ensure cache dir is created. |
| self._stack_trace._get_cache_dir() |
| |
| # Create 4 folders, 2 >24hrs old, 2 recent. |
| dirs = [ |
| ('oldest', time.time() - 60*60*24*10), |
| ('old', time.time() - 60*60*24), |
| ('new', time.time() - 60*60*12), |
| ('newest', time.time())] |
| |
| for dname, dtime in dirs: |
| dpath = os.path.join(self._cache_dir, dname) |
| os.mkdir(dpath) |
| os.utime(dpath, (dtime, dtime)) |
| |
| self._stack_trace._trim_cache() |
| |
| # Ensure only folders < 24hrs old exist. |
| for dname, dtime in dirs: |
| dpath = os.path.join(self._cache_dir, dname) |
| self.assertEquals( |
| os.path.exists(dpath), time.time() - dtime < 60*60*24) |
| |
| |
| def test_setup_cleanup_results_in_chroot(self): |
| # Write junk file to results directory. |
| test_string = 'junkity junk junk junk.' |
| junk_results_file = os.path.join(self._fake_results, 'junk.txt') |
| with open(junk_results_file, 'w') as f: |
| f.write(test_string) |
| |
| # Mount results directory inside of chroot. |
| chroot_results_dir = self._stack_trace._setup_results_in_chroot() |
| |
| full_chroot_results_dir = os.path.join( |
| self._stack_trace._chroot_dir, chroot_results_dir.lstrip(os.sep)) |
| |
| # Ensure results directory is mounted into chroot w/ r,w privs. |
| test_string_2 = 'with a dash of salt and pepper.' |
| junk_chroot_file = os.path.join(full_chroot_results_dir, 'junk.txt') |
| with open(junk_chroot_file, 'a+') as f: |
| self.assertEquals(test_string, f.read()) |
| f.write(test_string_2) |
| |
| # Unmount results directory and ensure mount pt is cleaned up. |
| self._stack_trace._cleanup_results_in_chroot(chroot_results_dir) |
| self.assertFalse(os.path.exists(full_chroot_results_dir)) |
| |
| # Ensure data we wrote is still there. |
| with open(junk_results_file, 'r') as f: |
| self.assertEqual(test_string + test_string_2, f.read()) |
| |
| |
| def test_generate_stack_traces_with_live_data(self): |
| # Retrieve cores from results set with known crashes. Corresponds to |
| # beta build x86-alex-r15-0.15.1011.73-a1-b60. gsutil doesn't properly |
| # rebuild directory structure on download, so just grab the cores. |
| utils.run( |
| 'gsutil -m cp' |
| ' gs://chromeos-autotest-results/68803-chromeos-test/*.dmp' |
| ' %s' % self._fake_results) |
| |
| # Grab keyval file |
| utils.run( |
| 'gsutil -m cp' |
| ' gs://chromeos-autotest-results/68803-chromeos-test/group0/keyval' |
| ' %s' % self._fake_results) |
| |
| # Enumerate existing .dmp files. |
| cores = glob.glob('%s/*.dmp' % self._fake_results) |
| |
| # Generate stack traces... |
| self._stack_trace.generate() |
| |
| # Ensure each core has a .dmp.txt file. |
| for core in cores: |
| self.assertTrue(os.path.exists(core + '.txt')) |
| |
| # Check the cache to make sure it was properly setup. |
| job_name = self._stack_trace._get_job_name() |
| _, symbols_dir, chroot_symbols_dir = self._setup_basic_cache( |
| job_name=job_name, mkdir=False) |
| self.assertEqual( |
| self._stack_trace._get_symbol_dir(job_name), chroot_symbols_dir) |
| self.assertTrue(os.path.exists(os.path.join( |
| symbols_dir, self._stack_trace._COMPLETE_FILE))) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
