autotest: Test new meta files in crash directory

client/bin/site_crash_test.py

 # Copyright (c) 2010 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 logging, os, re
 from autotest_lib.client.bin import site_log_reader, site_utils, test
 from autotest_lib.client.common_lib import error, utils
 
 
 class CrashTest(test.test):
@@ skipping 74 matching lines @@
         if username == 'chronos':
             return self._USER_CRASH_DIR
         else:
             return self._SYSTEM_CRASH_DIR
 
 
     def _initialize_crash_reporter(self):
         utils.system('%s --init --nounclean_check' % self._CRASH_REPORTER_PATH)
 
 
-    def create_fake_crash_dir_entry(self, name):
+    def write_crash_dir_entry(self, name, contents):
         entry = os.path.join(self._SYSTEM_CRASH_DIR, name)
         if not os.path.exists(self._SYSTEM_CRASH_DIR):
             os.makedirs(self._SYSTEM_CRASH_DIR)
-        utils.system('touch ' + entry)
+        utils.open_write_close(entry, contents)
         return entry
 
 
+    def write_fake_meta(self, name, exec_name):
+        return self.write_crash_dir_entry(name,
+                                          'exec_name=%s\n'
+                                          'ver=my_ver\n'
+                                          'done=1\n' % exec_name)
+
+
     def _prepare_sender_one_crash(self,
                                   send_success,
                                   reports_enabled,
                                   username,
                                   report):
         self._set_sending_mock(mock_enabled=True, send_success=send_success)
         self._set_consent(reports_enabled)
         if report is None:
-            report = self.create_fake_crash_dir_entry('fake.dmp')
+            self.write_crash_dir_entry('fake.dmp', '')
+            report = self.write_fake_meta('fake.meta', 'fake')
         return report
 
 
     def _parse_sender_output(self, output):
         """Parse the log output from the crash_sender script.
 
         This script can run on the logs from either a mocked or true
         crash send.
 
         Args:
           output: output from the script
 
         Returns:
           A dictionary with these values:
             exec_name: name of executable which crashed
+            meta_path: path to the report metadata file
+            output: the output from the script, copied
             report_kind: kind of report sent (minidump vs kernel)
-            report_name: name of the report sent
             send_attempt: did the script attempt to send a crash.
             send_success: if it attempted, was the crash send successful.
             sleep_time: if it attempted, how long did it sleep before
               sending (if mocked, how long would it have slept)
-            output: the output from the script, copied
         """
         sleep_match = re.search('Scheduled to send in (\d+)s', output)
         send_attempt = sleep_match is not None
         if send_attempt:
             sleep_time = int(sleep_match.group(1))
         else:
             sleep_time = None
-        report_kind_match = re.search('Report: (\S+) \((\S+)\)', output)
-        if report_kind_match:
-            report_name = report_kind_match.group(1)
-            report_kind = report_kind_match.group(2)
+        meta_match = re.search('Metadata: (\S+) \((\S+)\)', output)
+        if meta_match:
+            meta_path = meta_match.group(1)
+            report_kind = meta_match.group(2)
         else:
-            report_name = None
+            meta_path = None
             report_kind = None
+        payload_match = re.search('Payload: (\S+)', output)
+        if payload_match:
+            report_payload = payload_match.group(1)
+        else:
+            report_payload = None
         exec_name_match = re.search('Exec name: (\S+)', output)
         if exec_name_match:
             exec_name = exec_name_match.group(1)
         else:
             exec_name = None
         send_success = 'Mocking successful send' in output
         return {'exec_name': exec_name,
                 'report_kind': report_kind,
-                'report_name': report_name,
+                'meta_path': meta_path,
+                'report_payload': report_payload,
                 'send_attempt': send_attempt,
                 'send_success': send_success,
                 'sleep_time': sleep_time,
                 'output': output}
 
 
     def _call_sender_one_crash(self,
                                send_success=True,
                                reports_enabled=True,
                                username='root',
@@ skipping 58 matching lines @@
         del debug_result['output']
         logging.debug('Result of send (besides output): %s' % debug_result)
 
         return result
 
 
     def initialize(self):
         test.test.initialize(self)
         self._log_reader = site_log_reader.LogReader()
         self._leave_crash_sending = True
+        self._automatic_consent_saving = True
 
 
     def cleanup(self):
         self._reset_rate_limiting()
         self._clear_spooled_crashes()
         self._set_sending(self._leave_crash_sending)
         self._set_sending_mock(mock_enabled=False)
-        self._pop_consent()
+        if self._automatic_consent_saving:
+            self._pop_consent()
         test.test.cleanup(self)
 
 
     def run_crash_tests(self,
                         test_names,
                         initialize_crash_reporter=False,
                         clear_spool_first=True,
                         must_run_all=True):
         """Run crash tests defined in this class.
 
         Args:
           test_names: array of test names
           initialize_crash_reporter: should set up crash reporter for every run
           must_run_all: should make sure every test in this class is mentioned
             in test_names
         """
-        self._push_consent()
+        if self._automatic_consent_saving:
+            self._push_consent()
 
         if must_run_all:
             # Sanity check test_names is complete
             for attr in dir(self):
                 if attr.find('_test_') == 0:
                     test_name = attr[6:]
                     if not test_name in test_names:
                         raise error.TestError('Test %s is missing' % test_name)
 
         for test_name in test_names:
             logging.info(('=' * 20) + ('Running %s' % test_name) + ('=' * 20))
             if initialize_crash_reporter:
                 self._initialize_crash_reporter()
+            # Disable crash_sender and kill off any running ones.
+            self._set_sending(False)
             self._kill_running_sender()
             self._reset_rate_limiting()
             if clear_spool_first:
                 self._clear_spooled_crashes()
             self._set_sending(False)
             getattr(self, '_test_' + test_name)()