Formatting: Format all python code with black.
This CL is probably not what you're looking for, it's only
automated formatting. Ignore it with
`git blame --ignore-rev <revision>` for this commit.
BUG=b:233893248
TEST=CQ
Change-Id: I66591d7a738d241aed3290138c0f68065ab10a6d
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/chromite/+/3879174
Reviewed-by: Mike Frysinger <vapier@chromium.org>
Tested-by: Alex Klein <saklein@chromium.org>
diff --git a/cli/command_vm_test.py b/cli/command_vm_test.py
index f9b823a..d1f3b62 100644
--- a/cli/command_vm_test.py
+++ b/cli/command_vm_test.py
@@ -19,245 +19,289 @@
class Error(Exception):
- """Base exception for CLI command VM tests."""
+ """Base exception for CLI command VM tests."""
class SetupError(Error):
- """Raised when error occurs during test environment setup."""
+ """Raised when error occurs during test environment setup."""
class TestError(Error):
- """Raised when a command test has failed."""
+ """Raised when a command test has failed."""
class CommandError(Error):
- """Raised when error occurs during a command test."""
+ """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)
+ """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."""
+ """Decorator that runs the command test function."""
- def Decorator(test_function):
- """Inner decorator that actually wraps the 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)
+ 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 Wrapper
- return Decorator
+ return Decorator
class CommandVMTest(object):
- """Base class for CLI command VM tests.
+ """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):
- """Initializes CommandVMTest.
-
- Args:
- board: Board for the VM to run tests.
- image_path: Path to the image for the VM to run 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.
"""
- 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):
- """Builds a CLI command.
+ def __init__(self, board, image_path):
+ """Initializes CommandVMTest.
- 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()
+ 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 SetUp(self):
- """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 BuildCommand(self, command, device=None, pos_args=None, opt_args=None):
+ """Builds a CLI command.
- def TearDown(self):
- """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)
+ 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()
- @test_command_decorator('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'
+ def SetUp(self):
+ """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)
- cmd = self.BuildCommand('shell', device=self.device_addr,
- opt_args=['--no-known-hosts'])
+ def TearDown(self):
+ """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,
+ )
- 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)
+ @test_command_decorator("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"
- 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)
+ cmd = self.BuildCommand(
+ "shell", device=self.device_addr, opt_args=["--no-known-hosts"]
+ )
- 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)
+ 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)
- @test_command_decorator('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.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 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 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)
+ 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('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.device_addr,
- pos_args=['latest'],
- opt_args=['--no-wipe', '--no-reboot'])
+ @test_command_decorator("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.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 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)
+ 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('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.device_addr,
- pos_args=packages, opt_args=['--log-level=info',
- '--root=/usr/local'])
+ @test_command_decorator("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.device_addr,
+ pos_args=["latest"],
+ opt_args=["--no-wipe", "--no-reboot"],
+ )
- logging.info('Test to uninstall packages on the VM device.')
- with outcap.OutputCapturer() as output:
- result = cros_build_lib.run(cmd + ['--unmerge'], check=False)
+ 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)
- if result.returncode:
- logging.error('Failed to uninstall packages on the VM device.')
- raise CommandError(result.stderr)
+ @test_command_decorator("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.device_addr,
+ pos_args=packages,
+ opt_args=["--log-level=info", "--root=/usr/local"],
+ )
- 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 uninstall packages on the VM device.")
+ with outcap.OutputCapturer() as output:
+ result = cros_build_lib.run(cmd + ["--unmerge"], check=False)
- 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 uninstall packages on the VM device.")
+ raise CommandError(result.stderr)
- 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.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()
- 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()
+ logging.info("Test to install packages on the VM device.")
+ with outcap.OutputCapturer() as output:
+ result = cros_build_lib.run(cmd, check=False)
- # 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()
+ if result.returncode:
+ logging.error("Failed to install packages on the VM device.")
+ raise CommandError(result.stderr)
- def RunTests(self):
- """Calls the test functions."""
- self.TestShell()
- # TestDebug broken (crbug.com/863122)
- self.TestFlash()
- self.TestDeploy()
+ 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()
- def Run(self):
- """Runs the tests."""
- try:
- self.SetUp()
- self.RunTests()
- logging.info('All tests completed successfully.')
- finally:
- self.TearDown()
+ # 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):
+ """Calls the test functions."""
+ self.TestShell()
+ # TestDebug broken (crbug.com/863122)
+ self.TestFlash()
+ self.TestDeploy()
+
+ def Run(self):
+ """Runs the tests."""
+ try:
+ self.SetUp()
+ self.RunTests()
+ logging.info("All tests completed successfully.")
+ finally:
+ self.TearDown()