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

"""Unit tests for the deploy module."""

import json
import multiprocessing
import os
import sys
from unittest import mock

from chromite.cli import command
from chromite.cli import deploy
from chromite.lib import build_target_lib
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils
from chromite.lib import remote_access
from chromite.lib import sysroot_lib
from chromite.lib import unittest_lib
from chromite.lib.parser import package_info


pytestmark = [cros_test_lib.pytestmark_inside_only]


if cros_build_lib.IsInsideChroot():
  import portage  # pylint: disable=import-error


# pylint: disable=protected-access


class ChromiumOSDeviceFake(object):
  """Fake for device."""

  def __init__(self):
    self.board = 'board'
    self.hostname = None
    self.username = None
    self.port = None
    self.lsb_release = None
    self.cmds = []
    self.work_dir = '/testdir/'
    self.selinux_available = False

  def MountRootfsReadWrite(self):
    return True

  def IsSELinuxAvailable(self):
    return self.selinux_available

  def IsSELinuxEnforced(self):
    return True

  def run(self, cmd, **_kwargs):
    self.cmds.append(cmd)

  def CopyToDevice(self, _src, _dest, _mode='rsync', **_kwargs):
    return True


class ChromiumOSDeviceHandlerFake(object):
  """Fake for chromite.lib.remote_access.ChomiumOSDeviceHandler."""

  class RemoteAccessFake(object):
    """Fake for chromite.lib.remote_access.RemoteAccess."""

    def __init__(self):
      self.remote_sh_output = None

    def RemoteSh(self, *_args, **_kwargs):
      return cros_build_lib.CommandResult(output=self.remote_sh_output)

  def __init__(self, *_args, **_kwargs):
    self._agent = self.RemoteAccessFake()
    self.device = ChromiumOSDeviceFake()

  # TODO(dpursell): Mock remote access object in cros_test_lib (brbug.com/986).
  def GetAgent(self):
    return self._agent

  def __exit__(self, _type, _value, _traceback):
    pass

  def __enter__(self):
    return self.device


class BrilloDeployOperationFake(deploy.BrilloDeployOperation):
  """Fake for deploy.BrilloDeployOperation."""
  def __init__(self, emerge, queue):
    super().__init__(emerge)
    self._queue = queue

  def ParseOutput(self, output=None):
    super().ParseOutput(output)
    self._queue.put('advance')


class DbApiFake(object):
  """Fake for Portage dbapi."""

  def __init__(self, pkgs):
    self.pkg_db = {}
    for cpv, slot, rdeps_raw, build_time in pkgs:
      self.pkg_db[cpv] = {
          'SLOT': slot, 'RDEPEND': rdeps_raw, 'BUILD_TIME': build_time}

  def cpv_all(self):
    return list(self.pkg_db)

  def aux_get(self, cpv, keys):
    pkg_info = self.pkg_db[cpv]
    return [pkg_info[key] for key in keys]


class PackageScannerFake(object):
  """Fake for PackageScanner."""

  def __init__(self, packages, pkgs_attrs, packages_cpvs=None):
    self.pkgs = packages
    self.cpvs = packages_cpvs or packages
    self.listed = []
    self.num_updates = 0
    self.pkgs_attrs = pkgs_attrs

  def Run(self, _device, _root, _packages, _update, _deep, _deep_rev):
    return self.cpvs, self.listed, self.num_updates, self.pkgs_attrs


class PortageTreeFake(object):
  """Fake for Portage tree."""

  def __init__(self, dbapi):
    self.dbapi = dbapi


class TestInstallPackageScanner(cros_test_lib.MockOutputTestCase):
  """Test the update package scanner."""
  _BOARD = 'foo_board'
  _BUILD_ROOT = '/build/%s' % _BOARD
  _VARTREE = [
      ('foo/app1-1.2.3-r4', '0', 'foo/app2 !foo/app3', '1413309336'),
      ('foo/app2-4.5.6-r7', '0', '', '1413309336'),
      ('foo/app4-2.0.0-r1', '0', 'foo/app1 foo/app5', '1413309336'),
      ('foo/app5-3.0.7-r3', '0', '', '1413309336'),
  ]

  def setUp(self):
    """Patch imported modules."""
    self.PatchObject(cros_build_lib, 'GetChoice', return_value=0)
    self.device = ChromiumOSDeviceHandlerFake()
    self.scanner = deploy._InstallPackageScanner(self._BUILD_ROOT)
    self.PatchObject(deploy, '_GetDLCInfo', return_value=(None, None))

  def SetupVartree(self, vartree_pkgs):
    self.device.GetAgent().remote_sh_output = json.dumps(vartree_pkgs)

  def SetupBintree(self, bintree_pkgs):
    bintree = PortageTreeFake(DbApiFake(bintree_pkgs))
    build_root = os.path.join(self._BUILD_ROOT, '')
    portage_db = {build_root: {'bintree': bintree}}
    self.PatchObject(portage, 'create_trees', return_value=portage_db)

  def ValidatePkgs(self, actual, expected, constraints=None):
    # Containing exactly the same packages.
    self.assertEqual(sorted(expected), sorted(actual))
    # Packages appear in the right order.
    if constraints is not None:
      for needs, needed in constraints:
        self.assertGreater(actual.index(needs), actual.index(needed))

  def testRunUpdatedVersion(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r4'
    self.SetupBintree([
        (app1, '0', 'foo/app2 !foo/app3', '1413309336'),
        ('foo/app2-4.5.6-r7', '0', '', '1413309336'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 1)

  def testRunUpdatedBuildTime(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.3-r4'
    self.SetupBintree([
        (app1, '0', 'foo/app2 !foo/app3', '1413309350'),
        ('foo/app2-4.5.6-r7', '0', '', '1413309336'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 1)

  def testRunExistingDepUpdated(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r2'
    app2 = 'foo/app2-4.5.8-r3'
    self.SetupBintree([
        (app1, '0', 'foo/app2 !foo/app3', '1413309350'),
        (app2, '0', '', '1413309350'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1, app2], constraints=[(app1, app2)])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 2)

  def testRunMissingDepUpdated(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r2'
    app6 = 'foo/app6-1.0.0-r1'
    self.SetupBintree([
        (app1, '0', 'foo/app2 !foo/app3 foo/app6', '1413309350'),
        ('foo/app2-4.5.6-r7', '0', '', '1413309336'),
        (app6, '0', '', '1413309350'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1, app6], constraints=[(app1, app6)])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 1)

  def testRunExistingRevDepUpdated(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r2'
    app4 = 'foo/app4-2.0.1-r3'
    self.SetupBintree([
        (app1, '0', 'foo/app2 !foo/app3', '1413309350'),
        (app4, '0', 'foo/app1 foo/app5', '1413309350'),
        ('foo/app5-3.0.7-r3', '0', '', '1413309336'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1, app4], constraints=[(app4, app1)])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 2)

  def testRunMissingRevDepNotUpdated(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r2'
    app6 = 'foo/app6-1.0.0-r1'
    self.SetupBintree([
        (app1, '0', 'foo/app2 !foo/app3', '1413309350'),
        (app6, '0', 'foo/app1', '1413309350'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 1)

  def testRunTransitiveDepsUpdated(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r2'
    app2 = 'foo/app2-4.5.8-r3'
    app4 = 'foo/app4-2.0.0-r1'
    app5 = 'foo/app5-3.0.8-r2'
    self.SetupBintree([
        (app1, '0', 'foo/app2 !foo/app3', '1413309350'),
        (app2, '0', '', '1413309350'),
        (app4, '0', 'foo/app1 foo/app5', '1413309350'),
        (app5, '0', '', '1413309350'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1, app2, app4, app5],
                      constraints=[(app1, app2), (app4, app1), (app4, app5)])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 4)

  def testRunDisjunctiveDepsExistingUpdated(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r2'
    self.SetupBintree([
        (app1, '0', '|| ( foo/app6 foo/app2 ) !foo/app3', '1413309350'),
        ('foo/app2-4.5.6-r7', '0', '', '1413309336'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 1)

  def testRunDisjunctiveDepsDefaultUpdated(self):
    self.SetupVartree(self._VARTREE)
    app1 = 'foo/app1-1.2.5-r2'
    app7 = 'foo/app7-1.0.0-r1'
    self.SetupBintree([
        (app1, '0', '|| ( foo/app6 foo/app7 ) !foo/app3', '1413309350'),
        (app7, '0', '', '1413309350'),
    ])
    installs, listed, num_updates, _ = self.scanner.Run(
        self.device, '/', ['app1'], True, True, True)
    self.ValidatePkgs(installs, [app1, app7], constraints=[(app1, app7)])
    self.ValidatePkgs(listed, [app1])
    self.assertEqual(num_updates, 1)


class TestDeploy(cros_test_lib.ProgressBarTestCase,
                 cros_test_lib.MockTempDirTestCase):
  """Test deploy.Deploy."""

  @staticmethod
  def FakeGetPackagesByCPV(cpvs, _strip, _sysroot):
    return ['/path/to/%s.tbz2' % cpv.pv for cpv in cpvs]

  def setUp(self):
    # Fake being root to avoid running filesystem commands with sudo_run.
    self.PatchObject(os, 'getuid', return_value=0)
    self.PatchObject(os, 'geteuid', return_value=0)
    self._sysroot = os.path.join(self.tempdir, 'sysroot')
    osutils.SafeMakedirs(self._sysroot)
    self.device = ChromiumOSDeviceHandlerFake()
    self.PatchObject(
        remote_access, 'ChromiumOSDeviceHandler', return_value=self.device)
    self.PatchObject(cros_build_lib, 'GetBoard', return_value=None)
    self.PatchObject(build_target_lib, 'get_default_sysroot_path',
                     return_value=self._sysroot)
    self.package_scanner = self.PatchObject(deploy, '_InstallPackageScanner')
    self.get_packages_paths = self.PatchObject(
        deploy, '_GetPackagesByCPV', side_effect=self.FakeGetPackagesByCPV)
    self.emerge = self.PatchObject(deploy, '_Emerge', return_value=None)
    self.unmerge = self.PatchObject(deploy, '_Unmerge', return_value=None)
    self.PatchObject(deploy, '_GetDLCInfo', return_value=(None, None))
    # Avoid running the portageq command.
    sysroot_lib.Sysroot(self._sysroot).WriteConfig(
        'ARCH="amd64"\nPORTDIR_OVERLAY="%s"' % '/nothing/here')
    # make.conf needs to exist to correctly read back config.
    unittest_lib.create_stub_make_conf(self._sysroot)


  def testDeployEmerge(self):
    """Test that deploy._Emerge is called for each package."""

    _BINPKG = '/path/to/bar-1.2.5.tbz2'
    def FakeIsFile(fname):
      return fname == _BINPKG

    packages = ['some/foo-1.2.3', _BINPKG, 'some/foobar-2.0']
    cpvs = ['some/foo-1.2.3', 'to/bar-1.2.5', 'some/foobar-2.0']
    self.package_scanner.return_value = PackageScannerFake(
        packages,
        {'some/foo-1.2.3': {}, _BINPKG: {}, 'some/foobar-2.0': {}},
        cpvs)
    self.PatchObject(os.path, 'isfile', side_effect=FakeIsFile)

    deploy.Deploy(None, ['package'], force=True, clean_binpkg=False)

    # Check that package names were correctly resolved into binary packages.
    self.get_packages_paths.assert_called_once_with(
        [package_info.SplitCPV(p) for p in cpvs], True, self._sysroot)
    # Check that deploy._Emerge is called the right number of times.
    self.emerge.assert_called_once_with(mock.ANY, [
        '/path/to/foo-1.2.3.tbz2', '/path/to/bar-1.2.5.tbz2',
        '/path/to/foobar-2.0.tbz2'], '/', extra_args=None)
    self.assertEqual(self.unmerge.call_count, 0)

  def testDeployEmergeDLC(self):
    """Test that deploy._Emerge installs images for DLC packages."""
    packages = ['some/foodlc-1.0', 'some/bardlc-2.0']
    cpvs = ['some/foodlc-1.0', 'some/bardlc-2.0']
    self.package_scanner.return_value = PackageScannerFake(
        packages, {'some/foodlc-1.0': {}, 'some/bardlc-2.0': {}}, cpvs)
    self.PatchObject(deploy, '_GetDLCInfo',
                     return_value=('foo_id', 'foo_package'))

    deploy.Deploy(None, ['package'], force=True, clean_binpkg=False)
    # Check that dlcservice is restarted (DLC modules are deployed).
    self.assertTrue(['restart', 'dlcservice'] in self.device.device.cmds)

  def testDeployEmergeSELinux(self):
    """Test deploy progress when the device has SELinux"""

    _BINPKG = '/path/to/bar-1.2.5.tbz2'
    def FakeIsFile(fname):
      return fname == _BINPKG

    def GetRestoreconCommand(pkgfile):
      remote_path = os.path.join('/testdir/packages/to/', pkgfile)
      return [['cd', '/', '&&',
               'tar', 'tf', remote_path, '|',
               'restorecon', '-i', '-f', '-']]

    self.device.device.selinux_available = True
    packages = ['some/foo-1.2.3', _BINPKG, 'some/foobar-2.0']
    cpvs = ['some/foo-1.2.3', 'to/bar-1.2.5', 'some/foobar-2.0']
    self.package_scanner.return_value = PackageScannerFake(
        packages,
        {'some/foo-1.2.3': {}, _BINPKG: {}, 'some/foobar-2.0': {}},
        cpvs)
    self.PatchObject(os.path, 'isfile', side_effect=FakeIsFile)

    deploy.Deploy(None, ['package'], force=True, clean_binpkg=False)

    # Check that package names were correctly resolved into binary packages.
    self.get_packages_paths.assert_called_once_with(
        [package_info.SplitCPV(p) for p in cpvs], True, self._sysroot)
    # Check that deploy._Emerge is called the right number of times.
    self.assertEqual(self.emerge.call_count, 1)
    self.assertEqual(self.unmerge.call_count, 0)

    self.assertEqual(self.device.device.cmds,
                     [['setenforce', '0']] +
                     GetRestoreconCommand('foo-1.2.3.tbz2') +
                     GetRestoreconCommand('bar-1.2.5.tbz2') +
                     GetRestoreconCommand('foobar-2.0.tbz2') +
                     [['setenforce', '1']])

  def testDeployUnmerge(self):
    """Test that deploy._Unmerge is called for each package."""
    packages = ['foo', 'bar', 'foobar', 'foodlc']
    self.package_scanner.return_value = PackageScannerFake(
        packages, {'foo': {}, 'bar': {}, 'foobar': {},
                   'foodlc': {deploy._DLC_ID: 'foodlc',
                              deploy._DLC_PACKAGE: 'foopackage'}})

    deploy.Deploy(None, ['package'], force=True, clean_binpkg=False,
                  emerge=False)

    # Check that deploy._Unmerge is called the right number of times.
    self.assertEqual(self.emerge.call_count, 0)
    self.unmerge.assert_called_once_with(mock.ANY, packages, '/')

    self.assertEqual(
        self.device.device.cmds,
        [['dlcservice_util', '--uninstall', '--id=foodlc'],
         ['restart', 'dlcservice']])

  def testDeployMergeWithProgressBar(self):
    """Test that BrilloDeployOperation.Run() is called for merge."""
    packages = ['foo', 'bar', 'foobar']
    self.package_scanner.return_value = PackageScannerFake(
        packages, {'foo': {}, 'bar': {}, 'foobar': {}})

    run = self.PatchObject(deploy.BrilloDeployOperation, 'Run',
                           return_value=None)

    self.PatchObject(command, 'UseProgressBar', return_value=True)
    deploy.Deploy(None, ['package'], force=True, clean_binpkg=False)

    # Check that BrilloDeployOperation.Run was called.
    self.assertTrue(run.called)

  def testDeployUnmergeWithProgressBar(self):
    """Test that BrilloDeployOperation.Run() is called for unmerge."""
    packages = ['foo', 'bar', 'foobar']
    self.package_scanner.return_value = PackageScannerFake(
        packages, {'foo': {}, 'bar': {}, 'foobar': {}})

    run = self.PatchObject(deploy.BrilloDeployOperation, 'Run',
                           return_value=None)

    self.PatchObject(command, 'UseProgressBar', return_value=True)
    deploy.Deploy(None, ['package'], force=True, clean_binpkg=False,
                  emerge=False)

    # Check that BrilloDeployOperation.Run was called.
    self.assertTrue(run.called)

  def testBrilloDeployMergeOperation(self):
    """Test that BrilloDeployOperation works for merge."""
    def func(queue):
      for event in op.MERGE_EVENTS:
        queue.get()
        print(event)
        sys.stdout.flush()

    queue = multiprocessing.Queue()
    # Emerge one package.
    op = BrilloDeployOperationFake(True, queue)

    with self.OutputCapturer():
      op.Run(func, queue)

    # Check that the progress bar prints correctly.
    self.AssertProgressBarAllEvents(len(op.MERGE_EVENTS))

  def testBrilloDeployUnmergeOperation(self):
    """Test that BrilloDeployOperation works for unmerge."""
    def func(queue):
      for event in op.UNMERGE_EVENTS:
        queue.get()
        print(event)
        sys.stdout.flush()

    queue = multiprocessing.Queue()
    # Unmerge one package.
    op = BrilloDeployOperationFake(False, queue)

    with self.OutputCapturer():
      op.Run(func, queue)

    # Check that the progress bar prints correctly.
    self.AssertProgressBarAllEvents(len(op.UNMERGE_EVENTS))