server/site_tests/brillo_RecoverFromBadImage/brillo_RecoverFromBadImage.py

# Copyright 2015 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 common
from autotest_lib.client.common_lib import error
from autotest_lib.server import test


# The /dev directory mapping partition names to block devices.
_BLK_DEV_BY_NAME_DIR = '/dev/block/by-name'
# By default, we kill and recover the active system partition.
_DEFAULT_PART_NAME = 'system_X'


class brillo_RecoverFromBadImage(test.test):
    """Ensures that a Brillo device can recover from a bad image."""
    version = 1


    def resolve_slot(self, host, partition):
        """Resolves a partition slot (if any).

        @param host: A host object representing the DUT.
        @param partition: The name of the partition we are using. If it ends
                          with '_X' then we attempt to substitute it with some
                          non-active slot.

        @return A pair consisting of a fully resolved partition name and slot
                index; the latter is None if the partition is not slotted.

        @raise TestError: If a target slot could not be resolved.
        """
        # Check if the partition is slotted.
        if not re.match('.+_[a-zX]$', partition):
            return partition, None

        try:
            current_slot = int(
                    host.run_output('bootctl get-current-slot').strip())
            if partition[-1] == 'X':
                # Find a non-active target slot we could use.
                num_slots = int(
                        host.run_output('bootctl get-number-slots').strip())
                if num_slots < 2:
                    raise error.TestError(
                            'Device has no non-active slot that we can use')
                target_slot = 0 if current_slot else 1
                partition = partition[:-1] + chr(ord('a') + target_slot)
                logging.info(
                        'Current slot is %d, partition resolved to %s '
                        '(slot %d)', current_slot, partition, target_slot)
            else:
                # Make sure the partition slot is different from the active one.
                target_slot = ord(partition[-1]) - ord('a')
                if target_slot == current_slot:
                    target_slot = None
                    logging.warning(
                            'Partition %s is associated with the current boot '
                            'slot (%d), wiping it might fail if it is mounted',
                            partition, current_slot)
        except error.AutoservError:
            raise error.TestError('Error resolving device slots')

        return partition, target_slot


    def find_partition_device(self, host, partition):
        """Returns the block device of the partition.

        @param host: A host object representing the DUT.
        @param partition: The name of the partition we are using.

        @return Path to the device containing the partition.

        @raise TestError: If the partition name could not be mapped to a device.
        """
        try:
            cmd = 'find %s -type l' % os.path.join(_BLK_DEV_BY_NAME_DIR, '')
            for device in host.run_output(cmd).splitlines():
                if os.path.basename(device) == partition:
                    logging.info('Mapped partition %s to device %s',
                                 partition, device)
                    return device
        except error.AutoservError:
            raise error.TestError(
                    'Error finding device for partition %s' % partition)
        raise error.TestError(
                'No device found for partition %s' % partition)


    def get_device_block_info(self, host, device):
        """Returns the block size and count for a device.

        @param host: A host object representing the DUT.
        @param device: Path to a block device.

        @return A pair consisting of the block size (in bytes) and the total
                number of blocks on the device.

        @raise TestError: If we failed to get the block info for the device.
        """
        try:
            block_size = int(
                    host.run_output('blockdev --getbsz %s' % device).strip())
            device_size = int(
                    host.run_output('blockdev --getsize64 %s' % device).strip())
        except error.AutoservError:
            raise error.TestError(
                    'Failed to get block info for device %s' % device)
        return block_size, device_size / block_size


    def run_once(self, host=None, image_file=None, partition=_DEFAULT_PART_NAME,
                 device=None):
        """Runs the test.

        @param host: A host object representing the DUT.
        @param image_file: Image file to flash to the partition.
        @param partition: Name of the partition to wipe/recover.
        @param device: Path to the partition block device.

        @raise TestError: Something went wrong while trying to execute the test.
        @raise TestFail: The test failed.
        """
        # Check that the image file exists.
        if image_file is None:
            raise error.TestError('No image file provided')
        if not os.path.isfile(image_file):
            raise error.TestError('Image file %s not found' % image_file)

        try:
            # Resolve partition name and slot.
            partition, target_slot = self.resolve_slot(host, partition)

            # Figure out the partition device.
            if device is None:
                device = self.find_partition_device(host, partition)

            # Find the block size and count for the device.
            block_size, num_blocks = self.get_device_block_info(host, device)

            # Wipe the partition.
            logging.info('Wiping partition %s (%s)', partition, device)
            cmd = ('dd if=/dev/zero of=%s bs=%d count=%d' %
                   (device, block_size, num_blocks))
            run_err = 'Failed to wipe partition using %s' % cmd
            host.run(cmd)

            # Switch to the target slot, if required.
            if target_slot is not None:
                run_err = 'Error setting the active boot slot'
                host.run('bootctl set-active-boot-slot %d' % target_slot)

            # Re-flash the partition with fastboot.
            run_err = 'Failed to reboot the device into fastboot'
            host.ensure_bootloader_mode()
            run_err = 'Failed to flash image to partition %s' % partition
            host.fastboot_run('flash', args=(partition, image_file))

            # Reboot the device.
            run_err = 'Failed to reboot the device after flashing image'
            host.ensure_adb_mode()

            # Make sure we've booted from the alternate slot, if required.
            if target_slot is not None:
                run_err = 'Error checking the current boot slot'
                current_slot = int(
                        host.run_output('bootctl get-current-slot').strip())
                if current_slot != target_slot:
                    logging.error('Rebooted from slot %d instead of %d',
                                  current_slot, target_slot)
                    raise error.TestError(
                            'Device did not reboot from the expected slot')
        except error.AutoservError:
            raise error.TestFail(run_err)