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

from __future__ import print_function

import os
import re

import mock

from chromite.cbuildbot import cbuildbot_unittest
from chromite.cbuildbot import commands
from chromite.cbuildbot.stages import generic_stages_unittest
from chromite.cbuildbot.stages import vm_test_stages
from chromite.lib import cgroups
from chromite.lib import config_lib
from chromite.lib import constants
from chromite.lib import cros_logging
from chromite.lib import cros_test_lib
from chromite.lib import failures_lib
from chromite.lib import gs
from chromite.lib import moblab_vm
from chromite.lib import osutils
from chromite.lib import path_util
from chromite.lib import results_lib
from chromite.lib.buildstore import FakeBuildStore

# pylint: disable=too-many-ancestors


class GCETestStageTest(generic_stages_unittest.AbstractStageTestCase,
                       cbuildbot_unittest.SimpleBuilderTestCase):
  """Tests for the GCETest stage."""

  BOT_ID = 'betty-full'
  RELEASE_TAG = ''

  def setUp(self):
    self.PatchObject(commands, 'CreateTestRoot', return_value=self.tempdir)
    for cmd in ('GenerateStackTraces', 'ArchiveFile',
                'UploadArchivedFile', 'BuildAndArchiveTestResultsTarball'):
      self.PatchObject(commands, cmd, autospec=True)
    for cmd in (
        'RunTestSuite',
        'ArchiveTestResults',
        'ArchiveVMFiles',
        'RunDevModeTest',
        'RunCrosVMTest',
        'ListTests',
        'GetTestResultsDir',
    ):
      self.PatchObject(vm_test_stages, cmd, autospec=True)
    self.PatchObject(
        vm_test_stages.VMTestStage,
        '_NoTestResults',
        autospec=True,
        return_value=False)
    self.PatchObject(osutils, 'RmDir', autospec=True)
    self.PatchObject(cgroups, 'SimpleContainChildren', autospec=True)
    self._Prepare()
    self.buildstore = FakeBuildStore()

    # Simulate breakpad symbols being ready.
    board_runattrs = self._run.GetBoardRunAttrs(self._current_board)
    board_runattrs.SetParallel('breakpad_symbols_generated', True)
    board_runattrs.SetParallel('autotest_tarball_generated', True)

  def ConstructStage(self):
    # pylint: disable=protected-access
    self._run.GetArchive().SetupArchivePath()
    stage = vm_test_stages.GCETestStage(self._run, self.buildstore,
                                        self._current_board)
    image_dir = stage.GetImageDirSymlink()
    osutils.Touch(
        os.path.join(image_dir, constants.TEST_KEY_PRIVATE), makedirs=True)
    return stage

  def testGceTests(self):
    """Verifies that GCE_SUITE_TEST_TYPE tests are run on GCE."""
    self._run.config['gce_tests'] = [
        config_lib.GCETestConfig(
            constants.GCE_SUITE_TEST_TYPE, test_suite='gce-smoke')
    ]
    gce_tarball = constants.TEST_IMAGE_GCE_TAR

    # pylint: disable=unused-argument
    def _MockRunTestSuite(buildroot, board, image_path, results_dir,
                          test_config, *args, **kwargs):
      test_type = test_config.test_type
      self.assertEndsWith(image_path, gce_tarball)
      self.assertEqual(test_type, constants.GCE_SUITE_TEST_TYPE)
      self.assertEqual(test_config.test_suite, 'gce-smoke')

    # pylint: enable=unused-argument

    vm_test_stages.RunTestSuite.side_effect = _MockRunTestSuite

    self.RunStage()

    self.assertTrue(vm_test_stages.RunTestSuite.called and
                    vm_test_stages.RunTestSuite.call_count == 1)


class VMTestStageTest(generic_stages_unittest.AbstractStageTestCase,
                      cbuildbot_unittest.SimpleBuilderTestCase):
  """Tests for the VMTest stage."""

  BOT_ID = 'betty-full'
  RELEASE_TAG = ''

  def setUp(self):
    self.PatchObject(commands, 'CreateTestRoot', return_value=self.tempdir)
    for cmd in ('GenerateStackTraces', 'ArchiveFile',
                'UploadArchivedFile', 'BuildAndArchiveTestResultsTarball'):
      self.PatchObject(commands, cmd, autospec=True)
    for cmd in (
        'RunTestSuite',
        'ArchiveTestResults',
        'ArchiveVMFiles',
        'RunDevModeTest',
        'RunCrosVMTest',
        'ListTests',
        'GetTestResultsDir',
    ):
      self.PatchObject(vm_test_stages, cmd, autospec=True)
    self.PatchObject(
        vm_test_stages.VMTestStage,
        '_NoTestResults',
        autospec=True,
        return_value=False)
    self.PatchObject(osutils, 'RmDir', autospec=True)
    self.PatchObject(cgroups, 'SimpleContainChildren', autospec=True)
    self._Prepare()
    self.buildstore = FakeBuildStore()

    # Simulate breakpad symbols being ready.
    board_runattrs = self._run.GetBoardRunAttrs(self._current_board)
    board_runattrs.SetParallel('breakpad_symbols_generated', True)
    board_runattrs.SetParallel('autotest_tarball_generated', True)

  def ConstructStage(self):
    # pylint: disable=protected-access
    self._run.GetArchive().SetupArchivePath()
    stage = vm_test_stages.VMTestStage(self._run, self.buildstore,
                                       self._current_board)
    image_dir = stage.GetImageDirSymlink()
    osutils.Touch(
        os.path.join(image_dir, constants.TEST_KEY_PRIVATE), makedirs=True)
    return stage

  def testQuickTests(self):
    """Tests if quick unit and cros_au_test_harness tests are run correctly."""
    self._run.config['vm_tests'] = [
        config_lib.VMTestConfig(constants.SIMPLE_AU_TEST_TYPE)
    ]
    self.RunStage()

  def testFailedTest(self):
    """Tests if quick unit and cros_au_test_harness tests are run correctly."""
    self.PatchObject(
        vm_test_stages.VMTestStage,
        '_RunTest',
        autospec=True,
        side_effect=Exception())
    self.assertRaises(failures_lib.StepFailure, self.RunStage)

  def testRaisesInfraFail(self):
    """Tests that a infra failures has been raised."""
    commands.BuildAndArchiveTestResultsTarball.side_effect = (
        OSError('Cannot archive'))
    stage = self.ConstructStage()
    self.assertRaises(failures_lib.InfrastructureFailure, stage.PerformStage)

  def testSkipVMTest(self):
    """Tests trybot with no vm test."""
    extra_cmd_args = ['--novmtests']
    self._Prepare(extra_cmd_args=extra_cmd_args)
    # _Prepare resets the board_runattrs.
    board_runattrs = self._run.GetBoardRunAttrs(self._current_board)
    board_runattrs.SetParallel('autotest_tarball_generated', True)
    self.RunStage()

  def testReportTestResults(self):
    """Test trybot with reporting function."""
    self._run.config['vm_tests'] = [
        config_lib.VMTestConfig(constants.SIMPLE_AU_TEST_TYPE)
    ]
    self._run.config['vm_test_report_to_dashboards'] = True
    self.RunStage()

  def testForgivingVMTest(self):
    """Test if a test is warn-only, it actually warns."""
    self._run.config['vm_tests'] = [
        config_lib.VMTestConfig(
            constants.VM_SUITE_TEST_TYPE,
            warn_only=True,
            test_suite='bvt-perbuild'),
        config_lib.VMTestConfig(
            constants.VM_SUITE_TEST_TYPE, warn_only=False, test_suite='bvt-arc')
    ]

    # pylint: disable=unused-argument
    def _MockRunTestSuite(buildroot, board, image_path, results_dir,
                          test_config, *args, **kwargs):
      # Only raise exception in one test.
      if test_config.test_suite == 'bvt-perbuild':
        raise Exception()

    # pylint: enable=unused-argument

    self.PatchObject(
        vm_test_stages,
        'RunTestSuite',
        autospec=True,
        side_effect=_MockRunTestSuite)
    results_lib.Results.Clear()
    self.RunStage()
    result = results_lib.Results.Get()[0]
    self.assertEqual(result.result, results_lib.Results.FORGIVEN)
    # Make sure that all tests were actually run.
    self.assertEqual(vm_test_stages.RunTestSuite.call_count,
                     len(self._run.config['vm_tests']))


class MoblabVMTestStageTestCase(
    cros_test_lib.RunCommandTestCase,
    generic_stages_unittest.AbstractStageTestCase,
    cbuildbot_unittest.SimpleBuilderTestCase,
):
  """Does what it says above."""

  BOT_ID = 'moblab-generic-vm-full'
  RELEASE_TAG = ''

  def setUp(self):
    self.CreateMockOverlay('moblab-generic-vm')
    self._temp_chroot_prefix = os.path.join(self.tempdir, 'chroot')
    osutils.SafeMakedirsNonRoot(self._temp_chroot_prefix)
    self._temp_host_prefix = os.path.join(self.tempdir, 'host')
    osutils.SafeMakedirsNonRoot(self._temp_host_prefix)
    self.PatchObject(commands, 'UploadArchivedFile', autospec=True)
    self._Prepare()
    self.buildstore = FakeBuildStore()

  def ConstructStage(self):
    self._run.GetArchive().SetupArchivePath()
    self._run.config['moblab_vm_tests'] = [
        config_lib.MoblabVMTestConfig(constants.MOBLAB_VM_SMOKE_TEST_TYPE),
    ]
    stage = vm_test_stages.MoblabVMTestStage(self._run, self.buildstore,
                                             self._current_board)
    image_dir = stage.GetImageDirSymlink()
    osutils.Touch(
        os.path.join(image_dir, constants.TEST_KEY_PRIVATE), makedirs=True)
    osutils.Touch(
        os.path.join(image_dir, constants.TEST_IMAGE_BIN), makedirs=True)
    return stage

  def _temp_chroot_path(self, suffix):
    return os.path.join(self._temp_chroot_prefix, suffix)

  def _temp_host_path(self, suffix):
    return os.path.join(self._temp_host_prefix, suffix)

  def _strip_path_prefix(self, full_path):
    """Strips the host / chroot prefix from the given path."""
    if full_path.startswith(self._temp_chroot_prefix):
      return full_path.lstrip(self._temp_chroot_prefix)
    elif full_path.startswith(self._temp_host_prefix):
      return full_path.lstrip(self._temp_host_prefix)

  def testPerformStageSuccess(self):
    mock_create_test_root = self.PatchObject(
        commands,
        'CreateTestRoot',
        autospec=True,
        return_value=self._temp_chroot_prefix)
    self.PatchObject(
        path_util,
        'FromChrootPath',
        new=lambda x: self._temp_host_path(self._strip_path_prefix(x)),
    )
    self.PatchObject(
        path_util,
        'ToChrootPath',
        new=lambda x: self._temp_chroot_path(self._strip_path_prefix(x)),
    )

    mock_gs_context = mock.create_autospec(gs.GSContext)
    self.PatchObject(
        gs, 'GSContext', autospec=True, return_value=mock_gs_context)
    mock_buildbot_link = self.PatchObject(cros_logging, 'PrintBuildbotLink')
    mock_generate_payloads = self.PatchObject(commands, 'GeneratePayloads')
    mock_qp_payloads = self.PatchObject(commands,
                                        'GenerateQuickProvisionPayloads')
    self.PatchObject(commands, 'BuildAutotestTarballsForHWTest')
    # self.PatchObject(vm_test_stages, 'StageArtifactsOnMoblab', autospec=True)
    mock_run_moblab_tests = self.PatchObject(
        vm_test_stages, 'RunMoblabTests', autospec=True)
    mock_validate_results = self.PatchObject(
        vm_test_stages, 'ValidateMoblabTestSuccess', autospec=True)

    disk_dir = os.path.join(self.tempdir, 'moblab_mounted_disk')
    osutils.SafeMakedirsNonRoot(disk_dir)
    mock_context_manager = mock.MagicMock()
    mock_context_manager.__enter__.return_value = disk_dir
    mock_moblab_vm = mock.create_autospec(moblab_vm.MoblabVm)
    mock_moblab_vm.MountedMoblabDiskContext.return_value = mock_context_manager
    self.PatchObject(
        moblab_vm, 'MoblabVm', autospec=True, return_value=mock_moblab_vm)

    # Prepopulate results in the results directory to test result link printing.
    osutils.SafeMakedirsNonRoot(
        os.path.join(
            self._temp_host_prefix,
            'results',
            'results-1-moblab_DummyServerNoSspSuite',
            'moblab_RunSuite',
            'sysinfo',
            'var',
            'log',
            'bootup',
        ))
    osutils.SafeMakedirsNonRoot(
        os.path.join(
            self._temp_host_prefix,
            'results',
            'results-1-moblab_DummyServerNoSspSuite',
            'moblab_RunSuite',
            'sysinfo',
            'var',
            'log',
            'autotest',
        ))
    osutils.SafeMakedirsNonRoot(
        os.path.join(
            self._temp_host_prefix,
            'results',
            'results-1-moblab_DummyServerNoSspSuite',
            'moblab_RunSuite',
            'sysinfo',
            'var',
            'log_diff',
            'autotest',
        ))
    osutils.SafeMakedirsNonRoot(
        os.path.join(
            self._temp_host_prefix,
            'results',
            'results-1-moblab_DummyServerNoSspSuite',
            'sysinfo',
            'mnt',
            'moblab',
            'results',
        ))
    self.RunStage()

    self.assertEqual(mock_create_test_root.call_count, 1)

    mock_moblab_vm.Create.assert_called_once_with(mock.ANY, mock.ANY)
    self.assertEqual(mock_moblab_vm.Start.call_count, 1)
    self.assertEqual(mock_generate_payloads.call_count, 1)
    self.assertEqual(mock_qp_payloads.call_count, 1)
    generate_payloads_kwargs = mock_generate_payloads.call_args_list[0][1]
    self.assertTrue(os.path.isdir(generate_payloads_kwargs['archive_dir']))
    self.assertTrue(
        os.path.isfile(generate_payloads_kwargs['target_image_path']))
    mock_qp_kwargs = mock_qp_payloads.call_args_list[0][1]
    print(mock_qp_kwargs)
    self.assertTrue(
        os.path.isfile(mock_qp_kwargs['target_image_path']))
    self.assertTrue(os.path.isdir(mock_qp_kwargs['archive_dir']))
    self.assertEqual(mock_run_moblab_tests.call_count, 1)
    run_moblab_tests_kwargs = mock_run_moblab_tests.call_args_list[0][1]
    self.assertEqual(run_moblab_tests_kwargs['moblab_board'],
                     'moblab-generic-vm')

    self.assertEqual(mock_validate_results.call_count, 1)
    # 1 for the overall results during _Upload, 4 more for the detailed logs.
    self.assertEqual(mock_buildbot_link.call_count, 5)

    self.assertEqual(mock_moblab_vm.Stop.call_count, 1)
    self.assertEqual(mock_moblab_vm.Destroy.call_count, 1)


class RunTestSuiteTest(cros_test_lib.RunCommandTempDirTestCase):
  """Test RunTestSuite functionality."""

  TEST_BOARD = 'betty'
  BUILD_ROOT = '/fake/root'
  RESULTS_DIR = '/tmp/taco'
  TEST_IMAGE_OUTSIDE_CHROOT = os.path.join(BUILD_ROOT, 'test.img')
  VM_IMAGE_INSIDE_CHROOT = os.path.join(constants.CHROOT_SOURCE_ROOT,
                                        constants.VM_IMAGE_BIN)
  PRIVATE_KEY_OUTSIDE_CHROOT = os.path.join(BUILD_ROOT, 'rsa')
  PRIVATE_KEY_INSIDE_CHROOT = os.path.join(constants.CHROOT_SOURCE_ROOT, 'rsa')
  SSH_PORT = 9226

  def setUp(self):
    self.PatchObject(
        path_util,
        'FromChrootPath',
        new=
        lambda path: re.sub(r'^{0}'.format(constants.CHROOT_SOURCE_ROOT),
                            self.BUILD_ROOT, path)
    )
    self.PatchObject(
        path_util,
        'ToChrootPath',
        new=
        lambda path: re.sub(r'^{0}'.format(self.BUILD_ROOT),
                            constants.CHROOT_SOURCE_ROOT, path)
    )

  def _RunTestSuite(self, test_config, allow_chrome_crashes=False):
    vm_test_stages.RunTestSuite(
        self.BUILD_ROOT,
        self.TEST_BOARD,
        self.TEST_IMAGE_OUTSIDE_CHROOT,
        self.RESULTS_DIR,
        allow_chrome_crashes=allow_chrome_crashes,
        test_config=test_config,
        ssh_private_key=self.PRIVATE_KEY_OUTSIDE_CHROOT,
        ssh_port=self.SSH_PORT)
    self.assertCommandContains(['--board=%s' % self.TEST_BOARD])
    if test_config.use_ctest:
      self.assertCommandContains([
          os.path.join(self.BUILD_ROOT, 'src', 'platform', 'crostestutils',
                       'au_test_harness', 'cros_au_test_harness.py'),
          '--no_graphics', '--verbose',
          '--target_image=%s' % self.TEST_IMAGE_OUTSIDE_CHROOT,
          '--test_prefix=SimpleTestVerify',
          '--ssh_private_key=%s' % self.PRIVATE_KEY_OUTSIDE_CHROOT
      ])
      self.assertCommandContains(enter_chroot=True, expected=False)
    else:
      self.assertCommandContains([
          'cros_run_test', '--debug',
          '--image-path=%s' % self.VM_IMAGE_INSIDE_CHROOT,
          '--results-dir=%s' % self.RESULTS_DIR,
          '--private-key=%s' % self.PRIVATE_KEY_INSIDE_CHROOT
      ])
      self.assertCommandContains(enter_chroot=True)
      self.assertCommandContains(check=False)

  def testSimple(self):
    """Test SIMPLE config."""
    config = config_lib.VMTestConfig(constants.SIMPLE_AU_TEST_TYPE)
    self._RunTestSuite(config)
    self.assertCommandContains(['--test_prefix=SimpleTestVerify'])

  def testSmoke(self):
    """Test SMOKE config."""
    config = config_lib.VMTestConfig(
        constants.VM_SUITE_TEST_TYPE, test_suite='smoke')
    self._RunTestSuite(config)
    self.assertCommandContains(['--verify_suite_name=smoke'])

  def testGceSmokeTestType(self):
    """Test GCE test with gce-smoke suite."""
    config = config_lib.GCETestConfig(
        constants.GCE_SUITE_TEST_TYPE, test_suite='gce-smoke')
    self._RunTestSuite(config)
    self.assertCommandContains(['--test_prefix=SimpleTestVerify'])
    self.assertCommandContains(['--type=gce'])
    self.assertCommandContains(['--verify_suite_name=gce-smoke'])

  def testGceSanityTestType(self):
    """Test GCE test with gce-sanity suite."""
    config = config_lib.GCETestConfig(
        constants.GCE_SUITE_TEST_TYPE, test_suite='gce-sanity')
    self._RunTestSuite(config)
    self.assertCommandContains(['--test_prefix=SimpleTestVerify'])
    self.assertCommandContains(['--type=gce'])
    self.assertCommandContains(['--verify_suite_name=gce-sanity'])

  def testSmokeChromite(self):
    """Test SMOKE config using chromite VM code path."""
    config = config_lib.VMTestConfig(
        constants.VM_SUITE_TEST_TYPE, test_suite='smoke', use_ctest=False)
    self._RunTestSuite(config)
    self.assertCommandContains(['--autotest=suite:smoke'])
    self.assertCommandContains(['---test_that-args=-allow_chrome_crashes'],
                               expected=False)

  def testWhitelistChromeCrashes(self):
    """Test SMOKE config with allowing chrome crashes."""
    config = config_lib.VMTestConfig(
        constants.VM_SUITE_TEST_TYPE, test_suite='smoke', use_ctest=False)
    self._RunTestSuite(config, allow_chrome_crashes=True)
    self.assertCommandContains(['--autotest=suite:smoke'])
    self.assertCommandContains(['--test_that-args=--allow-chrome-crashes'])


class UnmockedTests(cros_test_lib.TempDirTestCase):
  """Test cases which really run tests, instead of using mocks."""

  def testListFaliedTests(self):
    """Tests if we can list failed tests."""
    test_report_1 = """
/tmp/taco/taste_tests/all/results-01-has_salsa              [  PASSED  ]
/tmp/taco/taste_tests/all/results-01-has_salsa/has_salsa    [  PASSED  ]
/tmp/taco/taste_tests/all/results-02-has_cheese             [  FAILED  ]
/tmp/taco/taste_tests/all/results-02-has_cheese/has_cheese  [  FAILED  ]
/tmp/taco/taste_tests/all/results-02-has_cheese/has_cheese   FAIL: No cheese.
"""
    test_report_2 = """
/tmp/taco/verify_tests/all/results-01-has_salsa              [  PASSED  ]
/tmp/taco/verify_tests/all/results-01-has_salsa/has_salsa    [  PASSED  ]
/tmp/taco/verify_tests/all/results-02-has_cheese             [  PASSED  ]
/tmp/taco/verify_tests/all/results-02-has_cheese/has_cheese  [  PASSED  ]
"""
    results_path = os.path.join(self.tempdir, 'tmp/taco')
    os.makedirs(results_path)
    # Create two reports with the same content to test that we don't
    # list the same test twice.
    osutils.WriteFile(
        os.path.join(results_path, 'taste_tests', 'all', 'test_report.log'),
        test_report_1,
        makedirs=True)
    osutils.WriteFile(
        os.path.join(results_path, 'taste_tests', 'failed', 'test_report.log'),
        test_report_1,
        makedirs=True)
    osutils.WriteFile(
        os.path.join(results_path, 'verify_tests', 'all', 'test_report.log'),
        test_report_2,
        makedirs=True)

    self.assertEqual(
        vm_test_stages.ListTests(results_path, show_passed=False),
        [('has_cheese', 'taste_tests/all/results-02-has_cheese')])

  def testArchiveTestResults(self):
    """Test if we can archive a test results dir."""
    test_results_dir = 'tmp/taco'
    results_path = os.path.join(self.tempdir, 'chroot', test_results_dir)
    archive_dir = os.path.join(self.tempdir, 'archived_taco')
    os.makedirs(results_path)
    os.makedirs(archive_dir)
    # File that should be archived.
    osutils.Touch(os.path.join(results_path, 'foo.txt'))
    # Flies that should be ignored.
    osutils.Touch(os.path.join(results_path, 'chromiumos_qemu_disk.bin.foo'))
    os.symlink('/src/foo', os.path.join(results_path, 'taco_link'))
    vm_test_stages.ArchiveTestResults(results_path, archive_dir)
    self.assertExists(os.path.join(archive_dir, 'foo.txt'))
    self.assertNotExists(
        os.path.join(archive_dir, 'chromiumos_qemu_disk.bin.foo'))
    self.assertNotExists(os.path.join(archive_dir, 'taco_link'))

  def testValidateMoblabTestSuccessNoLogsRaises(self):
    """ValidateMoblabTestSuccess raises when logs are missing."""
    os.makedirs(os.path.join(self.tempdir, 'debug'))
    with self.assertRaises(failures_lib.TestFailure):
      vm_test_stages.ValidateMoblabTestSuccess(self.tempdir)

  def testValidateMoblabTestSuccessTestNotRunRaises(self):
    """ValidateMoblabTestSuccess raises when logs indicate no test run."""
    os.makedirs(os.path.join(self.tempdir, 'debug'))
    osutils.WriteFile(
        os.path.join(self.tempdir, 'debug', 'test_that.INFO'), """
Some random stuff.
01/08 15:00:28.679 INFO  autoserv| [stderr] Suite job          [ PASSED ]
01/08 15:00:28.681 INFO  autoserv| [stderr]
01/08 15:00:28.681 INFO  autoserv| [stderr] Suite timings:""")
    with self.assertRaises(failures_lib.TestFailure):
      vm_test_stages.ValidateMoblabTestSuccess(self.tempdir)

  def testValidateMoblabTestSuccessTestFailedRaises(self):
    """ValidateMoblabTestSuccess raises when logs indicate test failed."""
    os.makedirs(os.path.join(self.tempdir, 'debug'))
    osutils.WriteFile(
        os.path.join(self.tempdir, 'debug', 'test_that.INFO'), """
Some random stuff.
01/08 15:00:28.679 INFO  autoserv| [stderr] Suite job          [ PASSED ]
01/08 15:00:28.680 INFO  autoserv| [stderr] dummy_PassServer   [ FAILED ]
01/08 15:00:28.681 INFO  autoserv| [stderr]
01/08 15:00:28.681 INFO  autoserv| [stderr] Suite timings:""")
    with self.assertRaises(failures_lib.TestFailure):
      vm_test_stages.ValidateMoblabTestSuccess(self.tempdir)

  def testValidateMoblabTestSuccessTestPassed(self):
    """ValidateMoblabTestSuccess succeeds when logs indicate test passed."""
    os.makedirs(os.path.join(self.tempdir, 'debug'))
    osutils.WriteFile(
        os.path.join(self.tempdir, 'debug', 'test_that.INFO'), """
Some random stuff.
01/08 15:00:28.679 INFO  autoserv| [stderr] Suite job          [ PASSED ]
01/08 15:00:28.680 INFO  autoserv| [stderr] dummy_PassServer   [ PASSED ]
01/08 15:00:28.681 INFO  autoserv| [stderr]
01/08 15:00:28.681 INFO  autoserv| [stderr] Suite timings:""")
    vm_test_stages.ValidateMoblabTestSuccess(self.tempdir)

  def testArchiveVMFiles(self):
    """Validate ArchiveVMFiles success in archiving files."""
    test_buildroot = os.path.join(self.tempdir, 'buildroot')
    image_dir = os.path.join(test_buildroot, 'chroot', 'testResultsDir')
    test_path_archive_output = os.path.join(self.tempdir, 'testResultsDir')
    os.makedirs(test_path_archive_output)
    vm_files = ['abc.txt', 'chromiumos_qemu_mem.bin']
    cros_test_lib.CreateOnDiskHierarchy(image_dir, vm_files)
    result = vm_test_stages.ArchiveVMFiles(test_buildroot, 'testResultsDir',
                                           test_path_archive_output)
    # The expected output is the test_path_archive_output with the one file that
    # matches the constants VM pattern prefix, which will be converted to a
    # .bin.tar file.
    expected_result = [
        os.path.join(test_path_archive_output, 'chromiumos_qemu_mem.bin.tar')
    ]
    self.assertEqual(result, expected_result)