server/site_tests/kernel_MemoryRamoop/kernel_MemoryRamoop.py

# Copyright (c) 2014 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, random, re, string, traceback
from autotest_lib.client.common_lib import error
from autotest_lib.server import autotest
from autotest_lib.server import hosts
from autotest_lib.server import test

class kernel_MemoryRamoop(test.test):
    """
    This test verify that /dev/pstore/console-ramoops is preserved correctly
    after system reboot/kernel crash and also verify that there is no memory
    corrupt in that log.

    There is also platform_KernelErrorPaths test that test kernel crash. But
    this test focuses on verify that kernel create console-ramoops file
    correctly and its content is not corrupt. Contrary to the other test
    that test at the bigger scope, i.e. the whole error reporting mechanism.
    """
    version = 1

    _RAMOOP_PATH = '/dev/pstore/console-ramoops'
    _KMSG_PATH = '/dev/kmsg'
    _LKDTM_PATH = '/sys/kernel/debug/provoke-crash/DIRECT'

    # ramopp have size of 128K so we will generate about 100K of random message
    _MSG_LINE_COUNT = 1000
    _MSG_LINE_LENGTH = 80
    _MSG_MAGIC = 'ramoop_test'


    def run_once(self, client_ip):
        """
        Run the test.
        """
        if not client_ip:
            error.TestError("Must provide client's IP address to test")

        self._client = hosts.create_host(client_ip)
        self._client_at = autotest.Autotest(self._client)

        self._run_test(self._do_reboot, '.*Restarting system.$')
        self._run_test(self._do_kernel_panic, '.*lkdtm:.*PANIC$')
        self._run_test(self._do_kernel_bug, '.*lkdtm:.*BUG$')
        self._run_test(self._do_reboot_with_suspend, '.*Restarting system.$')

    def _run_test(self, test_function, sig_pattern):
        """
        Run the test using by write random message to kernel log. Then
        restart/crash the kernel and then verify integrity of console-ramoop

        @param test_function: fuction to call to reboot / crash DUT
        @param sig_pattern: regex of kernel log message generate when reboot
                            or crash by test_function
        """

        msg = self._generate_random_msg()

        for line in msg:
            cmd = 'echo "%s" > %s' % (line, self._KMSG_PATH)
            self._client.run(cmd)

        test_function()

        cmd = 'cat %s' % self._RAMOOP_PATH
        ramoop = self._client.run(cmd).stdout

        self._verify_random_msg(ramoop, msg, sig_pattern)

    def _do_reboot(self):
        """
        Reboot host machine
        """
        logging.info('Server: reboot client')
        try:
            self._client.reboot()
        except error.AutoservRebootError as e:
            raise error.TestFail('%s.\nTest failed with error %s' % (
                    traceback.format_exc(), str(e)))

    def _do_reboot_with_suspend(self):
        """
        Reboot host machine after suspend once
        """
        self._client_at.run_test('power_Resume')

        logging.info('Server: reboot client')
        try:
            self._client.reboot()
        except error.AutoservRebootError as e:
            raise error.TestFail('%s.\nTest failed with error %s' % (
                    traceback.format_exc(), str(e)))

    def _do_kernel_panic(self):
        """
        Cause kernel panic using kernel dump test module
        """
        logging.info('Server: make client kernel panic')

        cmd = 'echo PANIC > %s' % self._LKDTM_PATH
        boot_id = self._client.get_boot_id()
        self._client.run(cmd, ignore_status=True)
        self._client.wait_for_restart(old_boot_id=boot_id)

    def _do_kernel_bug(self):
        """
        Cause kernel bug using kernel dump test module
        """
        logging.info('Server: make client kernel bug')

        cmd = 'echo BUG > %s' % self._LKDTM_PATH
        boot_id = self._client.get_boot_id()
        self._client.run(cmd, ignore_status=True)
        self._client.wait_for_restart(old_boot_id=boot_id)


    def _generate_random_msg(self):
        """
        Generate random message to put in kernel log
        The message format is [magic string]: [3 digit id] [random char/digit]
        """
        valid_char = string.letters + string.digits
        ret = []
        for i in range(self._MSG_LINE_COUNT):
            line = '%s: %03d ' % (self._MSG_MAGIC, i)
            for _ in range(self._MSG_LINE_LENGTH):
                line += random.choice(valid_char)
            ret += [line]
        return ret

    def _verify_random_msg(self, ramoop, src_msg, sig_pattern):
        """
        Verify random message generated by _generate_random_msg

        There are 3 things to verify.
        1. At least one random message exist. (earlier random message may be
           cutoff because console-ramoops has limited size.
        2. Integrity of random message.
        3. Signature of reboot / kernel crash

        @param ramoop: console-ramoops file in DUT
        @param src_msg: message write to kernel log
        @param sig_patterm: regex of kernel log to verify
        """
        #                   time stamp     magic   id      random
        pattern = str("\\[ *(\\d+\\.\\d+)\\] (%s: (\\d{3}) \\w{%d})$" %
            (self._MSG_MAGIC, self._MSG_LINE_LENGTH))
        matcher = re.compile(pattern)

        logging.info('%s', pattern)

        state = 'find_rand_msg'

        last_timestamp = 0
        for line in ramoop.split('\n'):
            if state == 'find_rand_msg':
                if not matcher.match(line):
                    continue
                last_id = int(matcher.split(line)[3]) - 1
                state = 'match_rand_pattern'
                logging.info("%s: %s", state, line)

            if state == 'match_rand_pattern':
                if not matcher.match(line):
                    continue
                components = matcher.split(line)
                timestamp = float(components[1])
                msg = components[2]
                id = int(components[3])

                if timestamp < last_timestamp:
                    logging.info("last_timestamp: %f, timestamp: %d",
                                 last_timestamp, timestamp)
                    raise error.TestFail('Found reverse time stamp.')
                last_timestamp = timestamp

                if id != last_id + 1:
                    logging.info("last_id: %d, id: %d", last_id, id)
                    raise error.TestFail('Found missing message.')
                last_id = id

                if msg != src_msg[id]:
                    logging.info("cur_msg: '%s'", msg)
                    logging.info("src_msg: '%s'", src_msg[id])
                    raise error.TestFail('Found corrupt message.')

                if id == self._MSG_LINE_COUNT - 1:
                    state = 'find_signature'

            if state == 'find_signature':
                if re.match(sig_pattern, line):
                    logging.info("%s: %s", state, line)
                    break

        # error case: successful run must break in find_sigature state
        else:
            raise error.TestFail(str('Verify failed at state %s' % state))