cli/command_vm_test.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.

"""Module for integration VM tests for CLI commands.

This module contains the basic functionalities for setting up a VM and testing
the CLI commands.
"""

from __future__ import print_function

from chromite.lib import cros_build_lib
from chromite.lib import cros_logging as logging
from chromite.lib import remote_access
from chromite.lib import vm


class Error(Exception):
  """Base exception for CLI command VM tests."""


class SetupError(Error):
  """Raised when error occurs during test environment setup."""


class TestError(Error):
  """Raised when a command test has failed."""


class CommandError(Error):
  """Raised when error occurs during a command test."""


def _PrintCommandLog(command, content):
  """Print out the log |content| for |command|."""
  if content:
    logging.info('\n----------- Start of %s log -----------\n%s\n'
                 '-----------  End of %s log  -----------',
                 command, content.rstrip(), command)


def TestCommandDecorator(command_name):
  """Decorator that runs the command test function."""

  def Decorator(test_function):
    """Inner decorator that actually wraps the function."""

    def Wrapper(command_test):
      """Wrapper for the test function."""
      command = cros_build_lib.CmdToStr(command_test.BuildCommand(command_name))
      logging.info('Running test for %s.', command)
      try:
        test_function(command_test)
        logging.info('Test for %s passed.', command)
      except CommandError as e:
        _PrintCommandLog(command, str(e))
        raise TestError('Test for %s failed.' % command)

    return Wrapper

  return Decorator


class CommandVMTest(object):
  """Base class for CLI command VM tests.

  This class provides the abstract interface for testing CLI commands on a VM.
  The sub-class must define the BuildCommand method in order to be usable. And
  the test functions must use the TestCommandDecorator decorator.
  """

  def __init__(self, board, image_path):
    """Initializes CommandVMTest.

    Args:
      board: Board for the VM to run tests.
      image_path: Path to the image for the VM to run tests.
    """
    self.board = board
    self.image_path = image_path
    self.working_image_path = None
    self.vm = None

  def BuildCommand(self, command, device=None, pos_args=None, opt_args=None):
    """Builds a CLI command.

    Args:
      command: The sub-command to build on (e.g. 'flash', 'deploy').
      device: The device's address for the command.
      pos_args: A list of positional arguments for the command.
      opt_args: A list of optional arguments for the command.
    """
    raise NotImplementedError()

  def SetUp(self):
    """Creates and starts the VM instance for testing."""
    try:
      logging.info('Setting up the VM for testing.')
      self.working_image_path = vm.CreateVMImage(
          image=self.image_path, board=self.board, updatable=True)
      self.vm = vm.VMInstance(self.working_image_path)
      self.vm.Start()
      logging.info('The VM has been successfully set up. Ready to run tests.')
    except vm.VMError as e:
      raise SetupError('Failed to set up the VM for testing: %s' % e)

  def TearDown(self):
    """Stops the VM instance after testing."""
    try:
      logging.info('Stopping the VM.')
      if self.vm:
        self.vm.Stop()
      logging.info('The VM has been stopped.')
    except vm.VMStopError as e:
      logging.warning('Failed to stop the VM: %s', e)

  @TestCommandDecorator('devices')
  def TestDevices(self):
    """Tests the devices command."""
    logging.info('Test to use devices command to set a user-friendly alias '
                 'name for the VM device.')
    alias = 'vm_device'
    cmd = self.BuildCommand('devices', device=self.vm.device_addr,
                            pos_args=['alias', alias])
    result = cros_build_lib.RunCommand(cmd, capture_output=True,
                                       error_code_ok=True)
    if result.returncode:
      logging.error('Failed to set an alias for the VM device.')
      raise CommandError(result.error)

    # Verify that the alias is set correctly.
    with remote_access.ChromiumOSDeviceHandler(
        remote_access.LOCALHOST, port=self.vm.port) as device:
      if device.alias != alias:
        logging.error('VM alias is "%s", which is not expected.', device.alias)
        raise CommandError()

  @TestCommandDecorator('shell')
  def TestShell(self):
    """Tests the shell command."""
    # The path and content of a temporary file for testing shell command.
    path = '/tmp/shell-test'
    content = 'shell command test file'

    cmd = self.BuildCommand('shell', device=self.vm.device_addr,
                            opt_args=['--no-known-hosts'])

    logging.info('Test to use shell command to write a file to the VM device.')
    write_cmd = cmd + ['--', 'echo "%s" > %s' % (content, path)]
    result = cros_build_lib.RunCommand(write_cmd, capture_output=True,
                                       error_code_ok=True)
    if result.returncode:
      logging.error('Failed to write the file to the VM device.')
      raise CommandError(result.error)

    logging.info('Test to use shell command to read a file on the VM device.')
    read_cmd = cmd + ['--', 'cat %s' % path]
    result = cros_build_lib.RunCommand(read_cmd, capture_output=True,
                                       error_code_ok=True)
    if result.returncode or result.output.rstrip() != content:
      logging.error('Failed to read the file on the VM device.')
      raise CommandError(result.error)

    logging.info('Test to use shell command to remove a file on the VM device.')
    remove_cmd = cmd + ['--', 'rm %s' % path]
    result = cros_build_lib.RunCommand(remove_cmd, capture_output=True,
                                       error_code_ok=True)
    if result.returncode:
      logging.error('Failed to remove the file on the VM device.')
      raise CommandError(result.error)

  @TestCommandDecorator('debug')
  def TestDebug(self):
    """Tests the debug command."""
    logging.info('Test to start and debug a new process on the VM device.')
    exe_path = '/bin/bash'
    start_cmd = self.BuildCommand('debug', device=self.vm.device_addr,
                                  opt_args=['--exe', exe_path])
    result = cros_build_lib.RunCommand(start_cmd, capture_output=True,
                                       error_code_ok=True, input='\n')
    if result.returncode:
      logging.error('Failed to start and debug a new process on the VM device.')
      raise CommandError(result.error)

    logging.info('Test to attach a running process on the VM device.')
    with remote_access.ChromiumOSDeviceHandler(
        remote_access.LOCALHOST, port=self.vm.port) as device:
      exe = 'update_engine'
      pids = device.GetRunningPids(exe, full_path=False)
      if not pids:
        logging.error('Failed to find any running process to debug.')
        raise CommandError()
      pid = pids[0]
      attach_cmd = self.BuildCommand('debug', device=self.vm.device_addr,
                                     opt_args=['--pid', str(pid)])
      result = cros_build_lib.RunCommand(attach_cmd, capture_output=True,
                                         error_code_ok=True, input='\n')
      if result.returncode:
        logging.error('Failed to attach a running process on the VM device.')
        raise CommandError(result.error)

  @TestCommandDecorator('flash')
  def TestFlash(self):
    """Tests the flash command."""
    # We explicitly disable reboot after the update because VMs sometimes do
    # not come back after reboot. The flash command does not need to verify
    # the integrity of the updated image. We have AU tests for that.
    cmd = self.BuildCommand('flash', device=self.vm.device_addr,
                            pos_args=['latest'],
                            opt_args=['--no-wipe', '--no-reboot'])

    logging.info('Test to flash the VM device with the latest image.')
    result = cros_build_lib.RunCommand(cmd, capture_output=True,
                                       error_code_ok=True)
    if result.returncode:
      logging.error('Failed to flash the VM device.')
      raise CommandError(result.error)

  @TestCommandDecorator('deploy')
  def TestDeploy(self):
    """Tests the deploy command."""
    packages = ['dev-python/cherrypy', 'app-portage/portage-utils']
    # Set the installation root to /usr/local so that the command does not
    # attempt to remount rootfs (which leads to VM reboot).
    cmd = self.BuildCommand('deploy', device=self.vm.device_addr,
                            pos_args=packages, opt_args=['--root=/usr/local'])

    logging.info('Test to uninstall packages on the VM device.')
    result = cros_build_lib.RunCommand(cmd + ['--unmerge'],
                                       capture_output=True,
                                       error_code_ok=True)
    if result.returncode:
      logging.error('Failed to uninstall packages on the VM device.')
      raise CommandError(result.error)

    logging.info('Test to install packages on the VM device.')
    result = cros_build_lib.RunCommand(cmd, capture_output=True,
                                       error_code_ok=True)
    if result.returncode:
      logging.error('Failed to install packages on the VM device.')
      raise CommandError(result.error)

    # Verify that the packages are installed.
    with remote_access.ChromiumOSDeviceHandler(
        remote_access.LOCALHOST, port=self.vm.port) as device:
      try:
        device.RunCommand(['python', '-c', '"import cherrypy"'])
        device.RunCommand(['qmerge', '-h'])
      except cros_build_lib.RunCommandError as e:
        logging.error('Unable to verify packages installed on VM: %s', e)
        raise CommandError()

  def RunTests(self):
    """Calls the test functions."""
    self.TestDevices()
    self.TestShell()
    self.TestDebug()
    self.TestFlash()
    self.TestDeploy()

  def Run(self):
    """Runs the tests."""
    try:
      self.SetUp()
      self.RunTests()
      logging.info('All tests completed successfully.')
    finally:
      self.TearDown()