au_test_harness/gce_au_worker_unittest.py

#!/usr/bin/python2
#
# 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.

"""Tests for gce_au_worker."""

from __future__ import print_function

import os
import sys
import unittest

import constants
sys.path.append(constants.CROS_PLATFORM_ROOT)
sys.path.append(constants.SOURCE_ROOT)

from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils
from chromite.lib import path_util
from crostestutils.au_test_harness.gce_au_worker import GCEAUWorker
from crostestutils.lib.gce import GceContext

class Options(object):
  """A fake class to hold command line options."""

  def __init__(self):
    self.board = 'fake-board'
    self.delta = False
    self.verbose = False
    self.quick_test = False
    self.verify_suite_name = 'smoke'


class GceAuWorkerTest(cros_test_lib.MockTempDirTestCase):
  """Test suite for GCEAUWorker."""

  PROJECT = 'test-project'
  ZONE = 'test-zone'
  NETWORK = 'default'
  BUCKET = 'foo-bucket'
  CLIENT_EMAIL = 'test-account@testdomain.com'
  GCE_TARBALL = 'chromiumos_test_image_gce_tar.gz'

  def setUp(self):
    # Fake out environment.
    options = Options()
    options.ssh_private_key = os.path.join(self.tempdir, 'ssh-private-key')
    self.ssh_private_key = options.ssh_private_key
    osutils.Touch(self.ssh_private_key)

    test_results_root = os.path.join(self.tempdir, 'test-results')
    self.test_results_all = os.path.join(test_results_root, 'all')
    self.test_results_failed = os.path.join(test_results_root, 'failed')
    osutils.SafeMakedirs(self.test_results_all)

    self.json_key_file = os.path.join(self.tempdir, 'service_account.json')
    osutils.Touch(self.json_key_file)

    self.image_path = os.path.join(self.tempdir, self.GCE_TARBALL)
    osutils.Touch(self.image_path)

    self.PatchObject(GceContext, 'ForServiceAccount', autospec=True)
    self.worker = GCEAUWorker(options, test_results_root, project=self.PROJECT,
                              zone=self.ZONE, network=self.NETWORK,
                              gcs_bucket=self.BUCKET,
                              json_key_file=self.json_key_file)

    # Mock out methods.
    for cmd in ['CreateInstance', 'CreateImage', 'GetInstanceIP',
                'DeleteInstance', 'DeleteImage', 'ListInstances', 'ListImages']:
      self.PatchObject(self.worker.gce_context, cmd, autospec=True)

    for cmd in ['CopyInto', 'DoCommand']:
      self.PatchObject(self.worker.gscontext, cmd, autospec=True)

    self.PatchObject(self.worker, 'GetNextResultsPath', autospec=True,
                     return_value=(self.test_results_all,
                                   self.test_results_failed))

  def testUpdateImage(self):
    """Tests that UpdateImage creates a GCE VM using the given tarball."""

    def _CopyInto(src, _):
      self.assertEqual(self.image_path, src)

    self.PatchObject(self.worker.gscontext, 'CopyInto', autospec=True,
                     side_effect=_CopyInto)
    self.PatchObject(self.worker, '_DeleteInstanceIfExists', autospec=True)
    self.PatchObject(self.worker, 'GetNextResultsPath', autospec=True,
                     return_value=('test-resultsi-all', 'test-results-failed'))
    self.worker.UpdateImage(self.image_path)

    #pylint: disable=protected-access
    self.worker._DeleteInstanceIfExists.assert_called_once_with()
    #pylint: enable=protected-access
    self.assertNotEqual(self.worker.instance, '')
    self.assertNotEqual(self.worker.image, '')
    self.assertTrue(self.worker.gscontext.CopyInto.called)

  def testCleanUp(self):
    """Tests that CleanUp deletes all instances and doesn't leak processes."""
    for _ in range(3):
      self.worker.UpdateImage(self.image_path)
    self.assertEqual(len(self.worker.bg_delete_processes), 2)

    self.worker.CleanUp()
    self.assertEqual(len(self.worker.bg_delete_processes), 0)

  def testVerifyImage(self):
    """Tests that VerifyImage calls out to test_that with correct args."""

    def _RunCommand(cmd, *args, **kwargs):
      expected_cmd = ['test_that', '-b', 'fake-board', '--no-quickmerge',
                      '--results_dir=%s' % self.test_results_all, '1.2.3.4',
                      'suite:smoke']
      for i, arg in enumerate(expected_cmd):
        self.assertEqual(arg, cmd[i])

      return cros_build_lib.CommandResult()

    self.PatchObject(cros_build_lib, 'RunCommand', autospec=True,
                     side_effect=_RunCommand)
    self.PatchObject(self.worker, 'AssertEnoughTestsPassed', autospec=True)
    self.PatchObject(self.worker, '_DeleteInstanceIfExists', autospec=True)
    self.PatchObject(self.worker.gce_context, 'GetInstanceIP', autospec=True,
                     return_value='1.2.3.4')
    self.PatchObject(path_util, 'ToChrootPath', autospec=True,
                     return_value='x/y/z')
    self.worker.UpdateImage(self.image_path)
    self.worker.VerifyImage(None)
    self.assertTrue(cros_build_lib.RunCommand.called)

  def testHandleFail(self):
    """Tests that _HandleFail copies necessary files for repro."""
    self.PatchObject(cros_build_lib, 'RunCommand', autospec=True)
    self.PatchObject(self.worker, '_DeleteInstanceIfExists', autospec=True)
    self.PatchObject(path_util, 'ToChrootPath', autospec=True,
                     return_value='x/y/z')
    self.PatchObject(self.worker, 'AssertEnoughTestsPassed', autospec=True,
                     return_value=False)
    self.worker.UpdateImage(self.image_path)
    self.worker.VerifyImage(None)
    self.assertExists(os.path.join(self.test_results_failed, self.GCE_TARBALL))
    self.assertExists(os.path.join(self.test_results_failed,
                                   os.path.basename(self.ssh_private_key)))


if __name__ == '__main__':
  unittest.main()