server/site_tests/platform_CorruptRootfs/platform_CorruptRootfs.py

# Copyright (c) 2009 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
import os
import re
import traceback
from autotest_lib.client.common_lib import error
from autotest_lib.server import test
from autotest_lib.client.bin import utils

class platform_CorruptRootfs(test.test):
    """Tests how the system recovers when the root file system is corrupted

    1. Copies kernel A and rootfs A to kernel B and rootfs B.
    2. Sets the modes on the partitions.
    3. Corrupts rootfs A by writing some random data to a the first few sectors.
    4. If enabled, corrupts the bootcache area.
    5. Reboots the system.
    6. Runs the test again in the reverse direction, leaving kernel A and
       rootfs A in the same state.
    """

    version = 1


    def _get_bootcache_offset(self):
        """Gets the offset of the bootcache from the command line.

        @return
          if bootcache is found, returns offset as a string
          otherwise returns '0'
        """

        # Get the linux cmd line
        result = self.client.run('cat /proc/cmdline')

        m = re.search('dm="(.*)"', result.stdout)
        dm = m.group(1)
        i = dm.find('bootcache')
        if i > 0:
            s = dm[i:].split()
            return s[2]   # 2nd field after bootcache has sector offset
        return '0'


    def _get_partition_layout(self):
        """Get the partition layout

        @return
          dev - name of the device hosting the partition
          kernelA - partition used to boot kernel
          rootfsA - partition of current root file system
          kernelB - backup copy of kernel
          rootfsB - backup copy of root file system
        """

        # What is our root partition?
        # TODO(crbug.com/226082)
        result = self.client.run('rootdev -s')
        logging.info('Root partition %s', result.stdout)
        rootdev = result.stdout.strip()
        if os.path.basename(rootdev).startswith('mmc'):
            dev = rootdev[:-2]
        else:
            dev = rootdev[:-1]
        kernelA = utils.get_kernel_partition(rootdev)
        rootfsA = rootdev
        kernelB = utils.get_free_kernel_partition(rootdev)
        rootfsB = utils.get_free_root_partition(rootdev)
        return dev, kernelA, rootfsA, kernelB, rootfsB


    def _corrupt_rootfs(self, host):
        """Corrupt the root file system
        """

        self.client = host
        self.client_test = 'platform_CorruptRootfs'

        dev, kernelA, rootfsA, kernelB, rootfsB = self._get_partition_layout()
        bootcache_offset = self._get_bootcache_offset()

        # Copy kernel and rootfs paritions from A to B
        logging.info('CorruptRootfs: copy partitions A to B')
        self.client.run('dd if=%s of=%s bs=64K' % (kernelA, kernelB))
        self.client.run('dd if=%s of=%s bs=64K' % (rootfsA, rootfsB))

        # Set attribrutes on kernel A and B
        logging.info('CorruptRootfs: set attributes on kernal A and B')
        self.client.run('cgpt add -i 2 -T 5 -P 9 -S 1 %s' % dev)
        self.client.run('cgpt add -i 4 -T 5 -P 9 -S 1 %s' % dev)

        # Corrupt rootfs A and bootcache
        logging.info('CorruptRootfs: corrupt rootfs A ' + rootfsA)
        self.client.run('dd if=/dev/urandom of=%s count=16' % rootfsA)
        if bootcache_offset != '0':
            self.client.run('dd if=/dev/zero of=%s seek=%s count=4096' %
                (rootfsA, bootcache_offset))

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

        # Find what partition we are now running on
        result = self.client.run('rootdev -s')
        logging.info('Root partition %s', result.stdout)


    def run_once(self, host=None):
        """
        run_once actually runs the test twice. The second run "undoes"
        what was done in the first run.

        @param host - the host machine running the test
        """

        self._corrupt_rootfs(host)
        self._corrupt_rootfs(host)