Fix kernel error paths test.

server/site_tests/platform_KernelErrorPaths/platform_KernelErrorPaths.py

Index: server/site_tests/platform_KernelErrorPaths/platform_KernelErrorPaths.py
diff --git a/server/site_tests/platform_KernelErrorPaths/platform_KernelErrorPaths.py b/server/site_tests/platform_KernelErrorPaths/platform_KernelErrorPaths.py
index 6ef099fb987317007b97cc7e35c3db046807acc2..7a597efbd3adfe45c74e6fc588d0641e0977e2b5 100644
--- a/server/site_tests/platform_KernelErrorPaths/platform_KernelErrorPaths.py
+++ b/server/site_tests/platform_KernelErrorPaths/platform_KernelErrorPaths.py
@@ -2,103 +2,63 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
-import logging, os, re, shutil
-from autotest_lib.server import test, autotest
+import logging, os
+
 from autotest_lib.client.common_lib import error
+from autotest_lib.client.cros.crash_test import CrashTest as CrashTestDefs
+from autotest_lib.server import test
 
 class platform_KernelErrorPaths(test.test):
     version = 1
 
-    def breakme(self, command):
-        logging.info('KernelErrorPaths: causing %s on host %s' %
+    def breakme(self, text):
+        command = "echo %s > /proc/breakme" % text
+        logging.info("KernelErrorPaths: executing '%s' on %s" %
                      (command, self.client.hostname))
         try:
-            self.client.run("echo %s > /proc/breakme" % command)
+            self.client.run(command)
         except error.AutoservRunError, e:
             # it is expected that this will cause a non-zero exit status
             pass
 
+    def pause_crash_sending(self):
+        result = self.client.run(
+            'ls %s' % os.path.dirname(CrashTestDefs._PAUSE_FILE))
+        self.client.run('touch %s' % CrashTestDefs._PAUSE_FILE)
+        return not os.path.basename(
+            CrashTestDefs._PAUSE_FILE) in result.stdout.splitlines()
 
-    def test_bug(self):
-        """
-        Cause the target to log a kernel bug, and then check in the
-        messages to make sure it did.
-        """
-        # Clear the messages so we can compare.
-        self.client.run('dmesg -c')
-        # Cause the client to report a kernel BUG.
-        self.breakme('bug')
-        # Now get messages and check to make sure it's in there.
-        result = self.client.run('dmesg')
-        marker = "Kernel BUG at"
-        found = False
-        for line in result.stdout.split('\n'):
-            if line.find(marker) != -1:
-                found = True
-                break
-        if not found:
-            raise error.TestFail("Kernel BUG reporting not working.")
-
-
-    def test_deadlock(self):
-        # Cause the target to go into a deadlock (have to run it twice).
-        self.breakme('deadlock')
-        self.breakme('deadlock')
-
-
-    def test_soft_lockup(self):
-        # Cause the target to go into an infinite loop.
-        self.breakme('softlockup')
-
-
-    def test_irq_lockup(self):
-        # Cause the target to go into a lockup with interrupts enabled.
-        self.breakme('irqlockup')
-
-
-    def test_no_irq_lockup(self):
-        # Cause the target to go into a lockup with interrupts disabled.
-        self.breakme('nmiwatchdog')
-
-
-    def test_null_dereference(self):
-        # Clear the messages so we can compare.
-        self.client.run('dmesg -c')
-        # Cause the target to dereference a null pointer.
-        self.breakme('nullptr')
-        # Now get messages and check to make sure it was noticed.
-        result = self.client.run('dmesg')
-        found = False
-        marker = "BUG: unable to handle kernel NULL pointer dereference"
-        for line in result.stdout.split('\n'):
-            if line.find(marker) != -1:
-                found = True
-                break
-        if not found:
-            raise error.TestFail("Kernel NULL pointer dereference detection "
-                                 "not working.")
-
-
-    def test_panic(self):
-        # Cause the target to panic.
-        self.breakme('panic')
-        if not self.client.wait_down(timeout=30):
-            raise error.TestFail("Kernel panic went unnoticed.")
-        if not self.client.wait_up(timeout=40):
-            raise error.TestFail("Kernel panic didn't cause successful reboot.")
-
+    def resume_crash_sending(self):

[petkov, 2011/02/25 19:54:35]

who calls this method? it seems it needs to be called in cleanup or something...

[vb, 2011/02/25 23:27:37]

right, this got lost when I cleaned up the code before sending it for review.

I spent some more time on this: I also need to manage the crash consent file, so
I changed the code to keep track of both and restore the status quo after test
finishes.

+        self.client.run('rm -f %s' % CrashTestDefs._PAUSE_FILE)
 
     def run_once(self, host=None):
         self.client = host
 
-        self.test_bug()
-        # TODO: Fill in the tests for these.
-        # self.test_deadlock()
-        # self.test_soft_lockup()
-        # self.test_irq_lockup()
-        # self.test_no_irq_lockup()
-        self.test_null_dereference()
-        # TODO(mbligh): crosbug.com/2269 - panic currently halts the
-        # DUT instead of rebooting it.  Commenting out until
-        # fixed.
-        #self.test_panic()
+        crash_sending_needs_resuming = self.pause_crash_sending()

[petkov, 2011/02/25 19:54:35]

unused?

+
+        crash_log_dir = CrashTestDefs._SYSTEM_CRASH_DIR
+
+        # Each tuple consists of two strings: the 'breakme' string to send
+        # into /proc/breakme on the target, and the crash report string to
+        # look for in the crash dump after target restarts.
+        # TODO(vbendeb): add the following breakme strings after fixing kernel
+        # bugs:
+        # 'deadlock' (has to be sent twice), 'softlockup', 'irqlockup'
+       test_tuples = (
+            ('bug', 'kernel BUG at'),
+            ('nmiwatchdog', 'BUG: NMI Watchdog detected LOCKUP'),
+            ('nullptr',
+             'BUG: unable to handle kernel NULL pointer dereference at'),
+            ('panic', 'Kernel panic - not syncing:'),
+            )
+
+        for action, text in test_tuples:
+            # Delete crash results, if any
+            self.client.run('rm -f %s/*' % crash_log_dir)
+            boot_id = self.client.get_boot_id()
+            self.breakme(action)  # This should cause target reset.
+            self.client.wait_for_restart(timeout=25, old_boot_id=boot_id)
+            result = self.client.run('cat %s/kernel.*.kcrash' % crash_log_dir)
+            if text not in result.stdout:
+                raise error.TestFail(
+                    "No '%s' in the log after sending '%s'" % (text, cmd))