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
5import os, re
6from autotest_lib.client.bin import utils
7from autotest_lib.client.common_lib import error
8from autotest_lib.client.cros.crash import crash_test
9
10
11_SECONDS_SEND_SPREAD = 3600
12
13class logging_CrashSender(crash_test.CrashTest):
14    """
15      End-to-end test of crash_sender.
16    """
17    version = 1
18
19
20    def _check_hardware_info(self, result):
21        # Get board name
22        lsb_release = utils.read_file('/etc/lsb-release')
23        board_match = re.search(r'CHROMEOS_RELEASE_BOARD=(.*)', lsb_release)
24        if not ('Board: %s' % board_match.group(1)) in result['output']:
25            raise error.TestFail('Missing board name %s in output' %
26                                 board_match.group(1))
27        # Get hwid
28        with os.popen("crossystem hwid 2>/dev/null", "r") as hwid_proc:
29            hwclass = hwid_proc.read()
30        if not hwclass:
31            hwclass = 'undefined'
32        if not ('HWClass: %s' % hwclass) in result['output']:
33            raise error.TestFail('Expected hwclass %s in output' % hwclass)
34
35
36    def _check_send_result(self, result, report_kind, payload_name,
37                           exec_name):
38        if result['report_exists']:
39            raise error.TestFail('Test report was not deleted after sending')
40        if result['rate_count'] != 1:
41            raise error.TestFail('Rate limit was not updated properly: #%d' %
42                                 result['rate_count'])
43        if not result['send_attempt']:
44            raise error.TestFail('Sender did not attempt the send')
45        if not result['send_success']:
46            raise error.TestFail('Send did not complete successfully')
47        if (result['sleep_time'] < 0 or
48            result['sleep_time'] >= _SECONDS_SEND_SPREAD):
49            raise error.TestFail('Sender did not sleep for an appropriate '
50                                 'amount of time: #%d' % result['sleep_time'])
51        if result['report_kind'] != report_kind:
52            raise error.TestFail('Incorrect report kind "%s", expected "%s"',
53                                 result['report_kind'], report_kind)
54        desired_payload = self.get_crash_dir_name(payload_name)
55        if result['report_payload'] != desired_payload:
56            raise error.TestFail('Payload filename was incorrect, got "%s", '
57                                 'expected "%s"', result['report_payload'],
58                                 desired_payload)
59        if result['exec_name'] != exec_name:
60            raise error.TestFail('ExecName was incorrect, expected "%s", '
61                                 'got "%s"', exec_name, result['exec_name'])
62
63
64    def _test_sender_simple_minidump(self):
65        """Test sending a single minidump crash report."""
66        result = self._call_sender_one_crash()
67        self._check_send_result(result, 'minidump',
68                                '%s.dmp' % self._FAKE_TEST_BASENAME, 'fake')
69        if (not 'Version: my_ver' in result['output']):
70            raise error.TestFail('Simple minidump send failed')
71        self._check_hardware_info(result)
72        # Also test "Image type" field.  Note that it will not be "dev" even
73        # on a dev build because crash-test-in-progress will exist.
74        if result['image_type']:
75            raise error.TestFail('Image type "%s" should not exist' %
76                                 result['image_type'])
77        # Also test "Boot mode" field.  Note that it will not be "dev" even
78        # when booting in dev mode because crash-test-in-progress will exist.
79        if result['boot_mode']:
80            raise error.TestFail('Boot mode "%s" should not exist' %
81                                 result['boot_mode'])
82
83
84    def _test_sender_reports_disabled(self):
85        """Test that when reporting is disabled, we don't send."""
86        result = self._call_sender_one_crash(reports_enabled=False)
87        if (result['report_exists'] or
88            not 'Crash reporting is disabled' in result['output'] or
89            result['send_attempt']):
90            raise error.TestFail('Sender did not handle reports disabled')
91
92
93    def run_once(self):
94        """ Run all tests once """
95        self.run_crash_tests([
96            'sender_simple_minidump',
97            'sender_reports_disabled']);
98