common_util_unittest.py

#!/usr/bin/python
#
# Copyright (c) 2012 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 common_util module."""

import os
import shutil
import subprocess
import tempfile
import unittest

import mox

import build_artifact
import common_util
import gsutil_util


# Fake Dev Server Layout:
TEST_LAYOUT = {
    'test-board-1': ['R17-1413.0.0-a1-b1346', 'R17-18.0.0-a1-b1346'],
    'test-board-2': ['R16-2241.0.0-a0-b2', 'R17-2.0.0-a1-b1346'],
    'test-board-3': []
}


class CommonUtilTest(mox.MoxTestBase):

  def setUp(self):
    mox.MoxTestBase.setUp(self)
    self._static_dir = tempfile.mkdtemp('common_util_unittest')
    self._outside_sandbox_dir = tempfile.mkdtemp('common_util_unittest')
    self._install_dir = tempfile.mkdtemp('common_util_unittest')

    for board, builds in TEST_LAYOUT.iteritems():
      board_path = os.path.join(self._static_dir, board)
      os.mkdir(board_path)
      for build in builds:
        build_path = os.path.join(board_path, build)
        os.mkdir(build_path)
        with open(os.path.join(
          build_path, build_artifact.TEST_IMAGE), 'w') as f:
          f.write('TEST_IMAGE')
        with open(os.path.join(
          build_path, build_artifact.STATEFUL_UPDATE), 'w') as f:
          f.write('STATEFUL_UPDATE')
        with open(os.path.join(
          build_path, build_artifact.ROOT_UPDATE), 'w') as f:
          f.write('ROOT_UPDATE')
        # AU payloads.
        au_dir = os.path.join(build_path, common_util.AU_BASE)
        nton_dir = os.path.join(au_dir, build + common_util.NTON_DIR_SUFFIX)
        os.makedirs(nton_dir)
        with open(os.path.join(nton_dir, build_artifact.ROOT_UPDATE), 'w') as f:
          f.write('ROOT_UPDATE')
        mton_dir = os.path.join(au_dir, build + common_util.MTON_DIR_SUFFIX)
        os.makedirs(mton_dir)
        with open(os.path.join(mton_dir, build_artifact.ROOT_UPDATE), 'w') as f:
          f.write('ROOT_UPDATE')

    self._good_mock_process = self.mox.CreateMock(subprocess.Popen)
    self._good_mock_process.returncode = 0
    self._bad_mock_process = self.mox.CreateMock(subprocess.Popen)
    self._bad_mock_process.returncode = 1

  def tearDown(self):
    shutil.rmtree(self._static_dir)
    shutil.rmtree(self._outside_sandbox_dir)
    shutil.rmtree(self._install_dir)

  def testParsePayloadListWithoutDeltas(self):
    """Tests we can parse the payload list when no delta updates exist."""
    archive_url_prefix = ('gs://chromeos-image-archive/x86-mario-release/'
                          'R17-1413.0.0-a1-b1346')
    full_basename = ('chromeos_R17-1413.0.0-a1_x86-mario_full_dev.bin')
    full_url = '/'.join([archive_url_prefix, full_basename])
    full_url_out, nton_url_out, mton_url_out, fw_url_out = (
        common_util.ParsePayloadList(
            archive_url_prefix, [full_basename, '', '']))
    self.assertEqual([full_url, None, None, None],
                     [full_url_out, nton_url_out, mton_url_out, fw_url_out])

  def testParsePartialPayloadList(self):
    """Tests that we can parse a payload list with missing optional payload."""
    archive_url_prefix = ('gs://chromeos-image-archive/x86-mario-release/'
                          'R17-1413.0.0-a1-b1346/')
    nton_basename = ('chromeos_R17-1413.0.0-a1_R17-1413.0.0-a1_x86-'
                     'mario_delta_dev.bin')
    full_basename = ('chromeos_R17-1413.0.0-a1_x86-mario_full_dev.bin')

    nton_url = '/'.join([archive_url_prefix, nton_basename])
    full_url = '/'.join([archive_url_prefix, full_basename])

    full_url_out, nton_url_out, mton_url_out, fw_url_out = (
        common_util.ParsePayloadList(archive_url_prefix,
                                     [full_basename, nton_basename]))
    self.assertEqual([full_url, nton_url, None, None],
                     [full_url_out, nton_url_out, mton_url_out, fw_url_out])

  def testInstallBuild(self):
    # TODO(frankf): Implement this test
    # self.fail('Not implemented.')
    pass

  def testPrepareAutotestPkgs(self):
    # TODO(frankf): Implement this test
    # self.fail('Not implemented.')
    # TODO: implement
    pass

  def testSafeSandboxAccess(self):
    # Path is in sandbox.
    self.assertTrue(
        common_util.SafeSandboxAccess(
            self._static_dir, os.path.join(self._static_dir, 'some-board')))

    # Path is sandbox.
    self.assertFalse(
        common_util.SafeSandboxAccess(self._static_dir, self._static_dir))

    # Path is outside the sandbox.
    self.assertFalse(
        common_util.SafeSandboxAccess(
          self._static_dir, self._outside_sandbox_dir))

    # Path contains '..'.
    self.assertFalse(
        common_util.SafeSandboxAccess(
            self._static_dir, os.path.join(self._static_dir, os.pardir)))

    # Path contains symbolic link references.
    os.chdir(self._static_dir)
    os.symlink(os.pardir, 'parent')
    self.assertFalse(
        common_util.SafeSandboxAccess(
            self._static_dir, os.path.join(self._static_dir, os.pardir)))

  def testAcquireReleaseLocks(self):
    # Successful lock and unlock, removing the newly created directory.
    lock_file = common_util.AcquireLock(self._static_dir, 'test-lock')
    self.assertTrue(os.path.exists(lock_file))
    common_util.ReleaseLock(self._static_dir, 'test-lock', destroy=True)
    self.assertFalse(os.path.exists(lock_file))

    # Attempt to freshly create and lock an existing directory.
    common_util.AcquireLock(self._static_dir, 'test-lock')
    common_util.ReleaseLock(self._static_dir, 'test-lock')
    self.assertRaises(common_util.CommonUtilError, common_util.AcquireLock,
                      self._static_dir, 'test-lock')
    common_util.AcquireLock(self._static_dir, 'test-lock', create_once=False)
    common_util.ReleaseLock(self._static_dir, 'test-lock', destroy=True)

    # Sucessfully re-lock a pre-existing directory.
    common_util.AcquireLock(self._static_dir, 'test-lock')
    common_util.ReleaseLock(self._static_dir, 'test-lock')
    common_util.AcquireLock(self._static_dir, 'test-lock', create_once=False)
    common_util.ReleaseLock(self._static_dir, 'test-lock', destroy=True)

    # Attempt to lock an already locked directory.
    common_util.AcquireLock(self._static_dir, 'test-lock')
    self.assertRaises(common_util.CommonUtilError, common_util.AcquireLock,
                      self._static_dir, 'test-lock')
    common_util.ReleaseLock(self._static_dir, 'test-lock', destroy=True)

  def testGetLatestBuildVersion(self):
    self.assertEqual(
        common_util.GetLatestBuildVersion(self._static_dir, 'test-board-1'),
        'R17-1413.0.0-a1-b1346')

  def testGetLatestBuildVersionLatest(self):
    """Test that we raise CommonUtilError when a build dir is empty."""
    self.assertRaises(common_util.CommonUtilError,
                      common_util.GetLatestBuildVersion,
                      self._static_dir, 'test-board-3')

  def testGetLatestBuildVersionUnknownBuild(self):
    """Test that we raise CommonUtilError when a build dir does not exist."""
    self.assertRaises(common_util.CommonUtilError,
                      common_util.GetLatestBuildVersion,
                      self._static_dir, 'bad-dir')

  def testGetLatestBuildVersionMilestone(self):
    """Test that we can get builds based on milestone."""
    expected_build_str = 'R16-2241.0.0-a0-b2'
    milestone = 'R16'
    build_str = common_util.GetLatestBuildVersion(
        self._static_dir, 'test-board-2', milestone)
    self.assertEqual(expected_build_str, build_str)

  def testGetControlFile(self):
    control_file_dir = os.path.join(
        self._static_dir, 'test-board-1', 'R17-1413.0.0-a1-b1346', 'autotest',
        'server', 'site_tests', 'network_VPN')
    os.makedirs(control_file_dir)
    with open(os.path.join(control_file_dir, 'control'), 'w') as f:
      f.write('hello!')

    control_content = common_util.GetControlFile(
        self._static_dir, 'test-board-1/R17-1413.0.0-a1-b1346',
        os.path.join('server', 'site_tests', 'network_VPN', 'control'))
    self.assertEqual(control_content, 'hello!')

  def commonGatherArtifactDownloads(self, payload_names):
    """Tests that we can gather the correct download requirements."""
    build = 'R17-1413.0.0-a1-b1346'
    archive_url_prefix = ('gs://chromeos-image-archive/x86-mario-release/' +
                          build)
    mock_data = 'mock data\nmock_data'
    payloads = map(lambda x: '/'.join([archive_url_prefix, x]), payload_names)
    expected_payloads = payloads + map(
        lambda x: '/'.join([archive_url_prefix, x]),
            [build_artifact.STATEFUL_UPDATE,
             build_artifact.AUTOTEST_ZIPPED_PACKAGE,
             build_artifact.TEST_SUITES_PACKAGE])
    self.mox.StubOutWithMock(gsutil_util, 'GSUtilRun')
    self.mox.StubOutWithMock(common_util, 'IsAvailable')
    self.mox.StubOutWithMock(common_util, 'ParsePayloadList')

    # GSUtil cat gs://archive_url_prefix/UPLOADED.
    gsutil_util.GSUtilRun(mox.StrContains(common_util.UPLOADED_LIST),
                          mox.IgnoreArg()).AndReturn(mock_data)
    common_util.IsAvailable(
        mox.IgnoreArg(), mock_data.splitlines()).AndReturn(True)
    common_util.ParsePayloadList(archive_url_prefix,
                                 mock_data.splitlines()).AndReturn(
        payloads + [None] * (4 - len(payload_names)))
    self.mox.ReplayAll()
    artifacts = common_util.GatherArtifactDownloads(
        self._static_dir, archive_url_prefix, self._install_dir, build)
    for index, artifact in enumerate(artifacts):
      self.assertEqual(artifact._gs_path, expected_payloads[index])
      self.assertTrue(artifact._tmp_staging_dir.startswith(self._static_dir))

    self.mox.VerifyAll()


  def testGatherArtifactDownloadsWithoutFirmware(self):
    """Tests that we can gather the correct download requirements."""
    self.commonGatherArtifactDownloads(['p1', 'p2', 'p3'])

  def testGatherArtifactDownloads(self):
    """Tests that we can gather the correct download requirements."""
    self.commonGatherArtifactDownloads(['p1', 'p2', 'p3', 'p4'])

  def testGatherArtifactDownloadsWithoutMton(self):
    """Gather the correct download requirements without mton delta."""
    self.commonGatherArtifactDownloads(['p1', 'p2'])

  def testGatherArtifactDownloadsWithoutMtonOrNton(self):
    """Gather the correct download requirements without delta payloads."""
    self.commonGatherArtifactDownloads(['p1'])

  def testGatherSymbolArtifactDownloads(self):
    """Tests that we can find debug symbol artifacts to download."""
    build = 'R17-1413.0.0-a1-b1346'
    archive_url_prefix = ('gs://chromeos-image-archive/x86-mario-release/' +
                          build)
    symbol_url = '/'.join([archive_url_prefix,
                           build_artifact.DEBUG_SYMBOLS])
    mock_data = 'mock-tarball.tgz\nmock-debug.tgz'
    self.mox.StubOutWithMock(gsutil_util, 'GSUtilRun')

    # GSUtil cat gs://archive_url_prefix/UPLOADED.
    gsutil_util.GSUtilRun(mox.StrContains(common_util.UPLOADED_LIST),
                          mox.IgnoreArg()).AndReturn(mock_data)

    self.mox.ReplayAll()
    artifacts = common_util.GatherSymbolArtifactDownloads(
        self._static_dir, archive_url_prefix, self._install_dir)
    for _, artifact in enumerate(artifacts):
      self.assertEqual(artifact._gs_path, symbol_url)
      self.assertTrue(artifact._tmp_staging_dir.startswith(self._static_dir))

    self.mox.VerifyAll()

  def testIsAvailable(self):
    """Test that we can detect whether the target artifacts are avaialble."""
    # Test when the all target files are available
    pattern_list = ['_full_', 'autotest.tar']
    uploaded_list = ['chromeos_R17-1413.0.0-a1_x86-mario_full_dev.bin',
                     'debug.tgz',
                     'autotest.tar.bz2']

    available = common_util.IsAvailable(pattern_list, uploaded_list)
    self.assertTrue(available)

    # Test when some target files are missing
    pattern_list = ['_full_', 'autotest.tar']
    uploaded_list = ['chromeos_R17-1413.0.0-a1_x86-mario_full_dev.bin',
                     'debug.tgz']

    available = common_util.IsAvailable(pattern_list, uploaded_list)
    self.assertFalse(available)

  def testWaitUntilAvailable(self):
    """Test that we can poll until all target artifacts are available."""
    archive_url = ('gs://chromeos-image-archive/x86-mario-release/'
                   'R17-1413.0.0-a1-b1346')
    to_wait_list = ['_full_']
    mock_data = 'mock data\nmock_data\nmock_data'

    self.mox.StubOutWithMock(gsutil_util, 'GSUtilRun')
    self.mox.StubOutWithMock(common_util, 'IsAvailable')

    # GSUtil cat gs://archive_url_prefix/UPLOADED.
    gsutil_util.GSUtilRun(mox.StrContains(common_util.UPLOADED_LIST),
                          mox.IgnoreArg()).AndReturn(mock_data)
    common_util.IsAvailable(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(True)

    self.mox.ReplayAll()
    uploaded_list = common_util.WaitUntilAvailable(
        to_wait_list, archive_url, 'UNIT TEST', delay=1)
    self.assertEqual(uploaded_list, mock_data.splitlines())
    self.mox.VerifyAll()

  def testWaitUntilAvailableWithRetry(self):
    """Test that we can poll until all target artifacts are available."""
    archive_url = ('gs://chromeos-image-archive/x86-mario-release/'
                   'R17-1413.0.0-a1-b1346')
    to_wait_list = ['_full_']
    mock_data = 'mock data\nmock_data\nmock_data'

    self.mox.StubOutWithMock(gsutil_util, 'GSUtilRun')
    self.mox.StubOutWithMock(common_util, 'IsAvailable')

    # GSUtil cat gs://archive_url_prefix/UPLOADED.
    gsutil_util.GSUtilRun(mox.StrContains(common_util.UPLOADED_LIST),
                          mox.IgnoreArg()).AndReturn(mock_data)
    common_util.IsAvailable(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(False)

    gsutil_util.GSUtilRun(mox.StrContains(common_util.UPLOADED_LIST),
                          mox.IgnoreArg()).AndReturn(mock_data)
    common_util.IsAvailable(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(True)

    self.mox.ReplayAll()
    uploaded_list = common_util.WaitUntilAvailable(
        to_wait_list, archive_url, 'UNIT TEST', delay=1)
    self.assertEqual(uploaded_list, mock_data.splitlines())
    self.mox.VerifyAll()

  def testWaitUntilAvailableTimeout(self):
    """Test that we wait for the target artifacts until timeout occurs."""
    archive_url = ('gs://chromeos-image-archive/x86-mario-release/'
                   'R17-1413.0.0-a1-b1346')
    to_wait_list = ['_full_']
    mock_data = 'mock data\nmock_data\nmock_data'

    self.mox.StubOutWithMock(gsutil_util, 'GSUtilRun')
    self.mox.StubOutWithMock(common_util, 'IsAvailable')

    # GSUtil cat gs://archive_url_prefix/UPLOADED.
    gsutil_util.GSUtilRun(mox.StrContains(common_util.UPLOADED_LIST),
                          mox.IgnoreArg()).AndReturn(mock_data)
    common_util.IsAvailable(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(False)

    self.mox.ReplayAll()
    self.assertRaises(common_util.CommonUtilError,
                      common_util.WaitUntilAvailable,
                      to_wait_list,
                      archive_url,
                      'UNIT TEST',
                      delay=2,
                      timeout=1)
    self.mox.VerifyAll()

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