autotest: Test new crash_reporter error diagnostics

client/site_tests/logging_CrashSender/logging_CrashSender.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_crash_test, site_utils, test
 from autotest_lib.client.common_lib import error, utils
 
 _25_HOURS_AGO = -25 * 60 * 60
 _CRASH_SENDER_CRON_PATH = '/etc/cron.hourly/crash_sender.hourly'
@@ skipping 44 matching lines @@
 
     def _shift_file_mtime(self, path, delta):
         statinfo = os.stat(path)
         os.utime(path, (statinfo.st_atime,
                         statinfo.st_mtime + delta))
 
 
     def _test_sender_simple_old_minidump(self):
         """Test that old minidumps and metadata are sent."""
         dmp_path = self.write_crash_dir_entry('fake.dmp', '')
-        meta_path = self.write_fake_meta('fake.meta', 'fake')
+        meta_path = self.write_fake_meta('fake.meta', 'fake', dmp_path)
         self._shift_file_mtime(dmp_path, _25_HOURS_AGO)
         self._shift_file_mtime(meta_path, _25_HOURS_AGO)
         self._check_simple_minidump_send(meta_path)
 
 
     def _test_sender_simple_kernel_crash(self):
         """Test sending a single kcrash report."""
         kcrash_fake_report = self.write_crash_dir_entry(
             'kernel.today.kcrash', '')
-        self.write_fake_meta('kernel.today.meta', 'kernel')
+        self.write_fake_meta('kernel.today.meta',
+                             'kernel',
+                             kcrash_fake_report)
         result = self._call_sender_one_crash(report=kcrash_fake_report)
         if (result['report_exists'] or
             result['rate_count'] != 1 or
             not result['send_attempt'] or
             not result['send_success'] or
             result['sleep_time'] < 0 or
             result['sleep_time'] >= _SECONDS_SEND_SPREAD or
             result['report_kind'] != 'kcrash' or
             (result['report_payload'] !=
              '/var/spool/crash/kernel.today.kcrash') or
@@ skipping 105 matching lines @@
         self._shift_file_mtime(unknown_file, _25_HOURS_AGO)
         results = self._call_sender_one_crash()
         if (not 'Removing old orphaned file' in results['output'] or
             os.path.exists(core_file) or os.path.exists(unknown_file)):
             raise error.TestFail(
                 'Old orphaned files were not removed')
 
 
     def _test_sender_incomplete_metadata(self):
         """Test that incomplete metadata file is removed once old."""
-        meta_file = self.write_crash_dir_entry('incomplete.meta', 'half=1')
         dmp_file = self.write_crash_dir_entry('incomplete.dmp', '')
+        meta_file = self.write_fake_meta('incomplete.meta',
+                                         'unknown',
+                                         dmp_file,
+                                         complete=False)
         # As new files, we expect crash_sender to leave these alone.
         results = self._call_sender_one_crash()
         if ('Removing recent incomplete report' in results['output'] or
             not os.path.exists(meta_file) or
             not os.path.exists(dmp_file)):
             raise error.TestFail('New unknown files were removed')
         self._shift_file_mtime(meta_file, _25_HOURS_AGO)
         results = self._call_sender_one_crash()
         if (not 'Removing old incomplete metadata' in results['output'] or
             os.path.exists(meta_file) or os.path.exists(dmp_file)):
             raise error.TestFail(
                 'Old unknown/incomplete files were not removed')
 
 
+    def _test_sender_missing_payload(self):
+        meta_file = self.write_fake_meta('bad.meta',
+                                         'unknown',
+                                         'bad.dmp')
+        other_file = self.write_crash_dir_entry('bad.other', '')
+        results = self._call_sender_one_crash(report=meta_file)
+        # Should remove this file.
+        if (not 'Missing payload' in results['output'] or
+            os.path.exists(meta_file) or
+            os.path.exists(other_file)):
+            raise error.TestFail('Missing payload case handled wrong')
+
+
     def _test_cron_runs(self):
         """Test sender runs successfully as part of the hourly cron job.
 
         Assuming we've run test_sender_simple which shows that a minidump
         gets removed as part of sending, we run the cron job (which is
         asynchronous) and wait for that file to be removed to just verify
         the job eventually runs the sender."""
         minidump = self._prepare_sender_one_crash(send_success=True,
                                                   reports_enabled=True,
                                                   username='root',
@@ skipping 18 matching lines @@
             'sender_simple_minidump',
             'sender_simple_old_minidump',
             'sender_simple_kernel_crash',
             'sender_pausing',
             'sender_reports_disabled',
             'sender_rate_limiting',
             'sender_single_instance',
             'sender_send_fails',
             'sender_orphaned_files',
             'sender_incomplete_metadata',
+            'sender_missing_payload',
             'cron_runs'])