lib/device_unittest.py

# -*- coding: utf-8 -*-
# Copyright 2019 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.

"""Unittests for device.py"""

from __future__ import print_function

import subprocess
import sys

import mock

from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import device
from chromite.lib import remote_access
from chromite.lib import vm


assert sys.version_info >= (3, 6), 'This module requires Python 3.6+'


# pylint: disable=protected-access


class DeviceTester(cros_test_lib.RunCommandTestCase):
  """Test device.Device."""

  def setUp(self):
    """Common set up method for all tests."""
    opts = device.Device.GetParser().parse_args(['--device', '190.0.2.130'])
    self._device = device.Device(opts)

  def CreateDevice(self, device_name, should_start_vm):
    """Creates a device.

    Args:
      device_name: Name of the device.
      should_start_vm: If True, then created device should be a VM.
    """
    created_device = device.Device.Create(
        vm.VM.GetParser().parse_args(
            ['--device', device_name] if device_name else []))
    self.assertEqual(isinstance(created_device, vm.VM), should_start_vm)

  def testWaitForBoot(self):
    self._device.WaitForBoot()
    # Verify that ssh command is called with all of the right configurations.
    self.assertCommandContains(
        ['ssh', 'root@190.0.2.130', '--', 'true'])

  @mock.patch('chromite.lib.cros_build_lib.run',
              side_effect=remote_access.SSHConnectionError())
  def testWaitForBootTimeOut(self, boot_mock):
    """Verify an exception is raised when the device takes to long to boot."""
    self.assertRaises(device.DeviceError, self._device.WaitForBoot, sleep=0)
    boot_mock.assert_called()

  @mock.patch('chromite.lib.device.Device.remote_run',
              return_value=cros_build_lib.CommandResult(returncode=1))
  def testWaitForBootReturnCode(self, boot_mock):
    """Verify an exception is raised when the returncode is not 0."""
    self.assertRaises(device.DeviceError, self._device.WaitForBoot)
    boot_mock.assert_called()

  def testRemoteCmd(self):
    """Verify remote command runs correctly with default arguments."""
    self._device.remote_run(['/usr/local/autotest/bin/vm_sanity'])
    self.assertCommandContains(['/usr/local/autotest/bin/vm_sanity'])
    self.assertCommandContains(capture_output=True, stderr=subprocess.STDOUT,
                               log_output=True)

  def testRemoteCmdStream(self):
    """Verify remote command for streaming output."""
    self._device.remote_run('/usr/local/autotest/bin/vm_sanity',
                            stream_output=True)
    self.assertCommandContains(capture_output=False)
    self.assertCommandContains(stderr=subprocess.STDOUT,
                               log_output=True, expected=False)

  def testCreate(self):
    """Verify Device/VM creation."""
    # Verify a VM is created when no IP is specified.
    self.CreateDevice(None, True)
    # Verify a VM isn't created, even if IP address is localhost. This can
    # happen when SSH tunneling to a remote DUT.
    self.CreateDevice('localhost:12345', False)
    # Verify a Device is created when an IP is specified.
    self.CreateDevice('190.0.2.130', False)