| # Copyright 2015 The ChromiumOS Authors |
| # 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. |
| """ |
| |
| import logging |
| |
| from chromite.cli import deploy |
| from chromite.lib import constants |
| from chromite.lib import cros_build_lib |
| from chromite.lib import remote_access |
| from chromite.lib import vm |
| from chromite.utils import outcap |
| |
| |
| 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) -> None: |
| """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 test_command_decorator(command_name): |
| """Decorator that runs the command test function.""" |
| |
| def Decorator(test_function): |
| """Inner decorator that actually wraps the function.""" |
| |
| def Wrapper(command_test) -> None: |
| """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: |
| """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 test_command_decorator decorator. |
| """ |
| |
| def __init__(self, board, image_path) -> None: |
| """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.port = None |
| self.device_addr = None |
| |
| def BuildCommand( |
| self, command, device=None, pos_args=None, opt_args=None |
| ) -> 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) -> None: |
| """Creates and starts the VM instance for testing.""" |
| self.port = remote_access.GetUnusedPort() |
| self.device_addr = "ssh://%s:%d" % (remote_access.LOCALHOST, self.port) |
| vm_path = vm.CreateVMImage( |
| image=self.image_path, board=self.board, updatable=True |
| ) |
| vm_cmd = [ |
| "./cros_vm", |
| "--ssh-port=%d" % self.port, |
| "--copy-on-write", |
| "--board=%s" % self.board, |
| "--image-path=%s" % vm_path, |
| "--start", |
| ] |
| cros_build_lib.run(vm_cmd, cwd=constants.CHROMITE_BIN_DIR) |
| |
| def TearDown(self) -> None: |
| """Stops the VM instance after testing.""" |
| if not self.port: |
| return |
| cros_build_lib.run( |
| ["./cros_vm", "--stop", "--ssh-port=%d" % self.port], |
| cwd=constants.CHROMITE_BIN_DIR, |
| check=False, |
| ) |
| |
| @test_command_decorator("shell") |
| def TestShell(self) -> None: |
| """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.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.run(write_cmd, capture_output=True, check=False) |
| if result.returncode: |
| logging.error("Failed to write the file to the VM device.") |
| raise CommandError(result.stderr) |
| |
| 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.run( |
| read_cmd, capture_output=True, encoding="utf-8", check=False |
| ) |
| if result.returncode or result.stdout.rstrip() != content: |
| logging.error("Failed to read the file on the VM device.") |
| raise CommandError(result.stderr) |
| |
| 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.run( |
| remove_cmd, capture_output=True, check=False |
| ) |
| if result.returncode: |
| logging.error("Failed to remove the file on the VM device.") |
| raise CommandError(result.stderr) |
| |
| @test_command_decorator("debug") |
| def TestDebug(self) -> None: |
| """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.device_addr, opt_args=["--exe", exe_path] |
| ) |
| result = cros_build_lib.run( |
| start_cmd, capture_output=True, check=False, input="\n" |
| ) |
| if result.returncode: |
| logging.error( |
| "Failed to start and debug a new process on the VM device." |
| ) |
| raise CommandError(result.stderr) |
| |
| logging.info("Test to attach a running process on the VM device.") |
| with remote_access.ChromiumOSDeviceHandler( |
| remote_access.LOCALHOST, port=self.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.device_addr, opt_args=["--pid", str(pid)] |
| ) |
| result = cros_build_lib.run( |
| attach_cmd, capture_output=True, check=False, input="\n" |
| ) |
| if result.returncode: |
| logging.error( |
| "Failed to attach a running process on the VM device." |
| ) |
| raise CommandError(result.stderr) |
| |
| @test_command_decorator("flash") |
| def TestFlash(self) -> None: |
| """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.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.run(cmd, capture_output=True, check=False) |
| if result.returncode: |
| logging.error("Failed to flash the VM device.") |
| raise CommandError(result.stderr) |
| |
| @test_command_decorator("deploy") |
| def TestDeploy(self) -> None: |
| """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.device_addr, |
| pos_args=packages, |
| opt_args=["--log-level=info", "--root=/usr/local"], |
| ) |
| |
| logging.info("Test to uninstall packages on the VM device.") |
| with outcap.OutputCapturer() as output: |
| result = cros_build_lib.run(cmd + ["--unmerge"], check=False) |
| |
| if result.returncode: |
| logging.error("Failed to uninstall packages on the VM device.") |
| raise CommandError(result.stderr) |
| |
| captured_output = output.GetStdout() + output.GetStderr() |
| for event in deploy.BrilloDeployOperation.UNMERGE_EVENTS: |
| if event not in captured_output: |
| logging.error( |
| "Strings used by deploy.BrilloDeployOperation to update " |
| "the progress bar have been changed. Please update the " |
| "strings in UNMERGE_EVENTS" |
| ) |
| raise CommandError() |
| |
| logging.info("Test to install packages on the VM device.") |
| with outcap.OutputCapturer() as output: |
| result = cros_build_lib.run(cmd, check=False) |
| |
| if result.returncode: |
| logging.error("Failed to install packages on the VM device.") |
| raise CommandError(result.stderr) |
| |
| captured_output = output.GetStdout() + output.GetStderr() |
| for event in deploy.BrilloDeployOperation.MERGE_EVENTS: |
| if event not in captured_output: |
| logging.error( |
| "Strings used by deploy.BrilloDeployOperation to update " |
| "the progress bar have been changed. Please update the " |
| "strings in MERGE_EVENTS" |
| ) |
| raise CommandError() |
| |
| # Verify that the packages are installed. |
| with remote_access.ChromiumOSDeviceHandler( |
| remote_access.LOCALHOST, port=self.port |
| ) as device: |
| try: |
| device.run(["python", "-c", '"import cherrypy"']) |
| device.run(["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) -> None: |
| """Calls the test functions.""" |
| self.TestShell() |
| # TestDebug broken (crbug.com/863122) |
| self.TestFlash() |
| self.TestDeploy() |
| |
| def Run(self) -> None: |
| """Runs the tests.""" |
| try: |
| self.SetUp() |
| self.RunTests() |
| logging.info("All tests completed successfully.") |
| finally: |
| self.TearDown() |