#!/usr/bin/python

# Copyright (c) 2011 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 lkgm_manager. Needs to be run inside of chroot for mox."""

import mox
import os
import shutil
import sys
import tempfile
import threading
import time
import unittest

if __name__ == '__main__':
  import constants
  sys.path.append(constants.SOURCE_ROOT)

from chromite.buildbot import lkgm_manager
from chromite.buildbot import manifest_version
from chromite.buildbot import manifest_version_unittest


FAKE_VERSION_STRING = '1.2.3.4-rc3'
FAKE_VERSION_STRING_NEXT = '1.2.3.4-rc4'


class LKGMCandidateInfoTest(mox.MoxTestBase):
  """Test methods testing methods in _LKGMCandidateInfo class."""

  def setUp(self):
    mox.MoxTestBase.setUp(self)
    self.tmpdir = tempfile.mkdtemp()

  def testLoadFromString(self):
    """Tests whether we can load from a string."""
    info = lkgm_manager._LKGMCandidateInfo(version_string=FAKE_VERSION_STRING)
    self.assertEqual(info.VersionString(), FAKE_VERSION_STRING)

  def testIncrementVersionPatch(self):
    """Tests whether we can increment a lkgm info."""
    info = lkgm_manager._LKGMCandidateInfo(version_string=FAKE_VERSION_STRING)
    info.IncrementVersion()
    self.assertEqual(info.VersionString(), FAKE_VERSION_STRING_NEXT)

  def testVersionCompare(self):
    """Tests whether our comparision method works."""
    info1 = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc1')
    info2 = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc2')
    info3 = lkgm_manager._LKGMCandidateInfo('1.2.200.4-rc1')
    info4 = lkgm_manager._LKGMCandidateInfo('1.4.3.4-rc1')

    self.assertTrue(info2 > info1)
    self.assertTrue(info3 > info1)
    self.assertTrue(info3 > info2)
    self.assertTrue(info4 > info1)
    self.assertTrue(info4 > info2)
    self.assertTrue(info4 > info3)

  def tearDown(self):
    shutil.rmtree(self.tmpdir)


class LKGMManagerTest(mox.MoxTestBase):
  """Tests for the BuildSpecs manager."""

  def setUp(self):
    mox.MoxTestBase.setUp(self)

    self.tmpdir = tempfile.mkdtemp()
    self.source_repo = 'ssh://source/repo'
    self.manifest_repo = 'ssh://manifest/repo'
    self.version_file = 'version-file.sh'
    self.branch = 'master'
    self.build_name = 'x86-generic'
    self.incr_type = 'patch'

    self.manager = lkgm_manager.LKGMManager(
      self.tmpdir, self.source_repo, self.manifest_repo, self.branch,
      self.build_name, dry_run=True)

    self.manager.SLEEP_TIMEOUT = 1

  def _CommonTestLatestCandidateByVersion(self, version, expected_candidate,
                                          no_all=False):
    """Common helper function for latest candidate tests.

    Helper function to test given a version whether we can the right candidate
    back.

    Args:
      version: The Chrome OS version to look for the latest candidate of.
      expected_candidate: What we expect to come back.
    """
    if not no_all:
      self.manager.all = ['1.2.3.4-rc1',
                          '1.2.3.4-rc2',
                          '1.2.3.4-rc9',
                          '1.2.3.5-rc1',
                          '1.2.3.6-rc2',
                          '1.2.4.3-rc1',
                          ]

    info_for_test = manifest_version.VersionInfo(version)
    candidate = self.manager._GetLatestCandidateByVersion(info_for_test)
    self.assertEqual(candidate.VersionString(), expected_candidate)

  def testGetLatestCandidateByVersionCommonCase(self):
    """Tests whether we can get the latest candidate under the common case.

    This test tests whether or not we get the right candidate when we have
    many of the same candidate version around but with different rc's.
    """
    self._CommonTestLatestCandidateByVersion('1.2.3.4', '1.2.3.4-rc9')

  def testGetLatestCandidateByVersionOnlyOne(self):
    """Tests whether we can get the latest candidate with only one rc."""
    self._CommonTestLatestCandidateByVersion('1.2.3.6', '1.2.3.6-rc2')

  def testGetLatestCandidateByVersionNone(self):
    """Tests whether we can get the latest candidate with no rc's."""
    self._CommonTestLatestCandidateByVersion('1.2.5.7', '1.2.5.7-rc1')

  def testGetLatestCandidateByVersionNoneNoAll(self):
    """Tests whether we can get the latest candidate with no rc's at all."""
    self._CommonTestLatestCandidateByVersion('10.0.1.5', '10.0.1.5-rc1', True)

  def testCreateNewCandidate(self):
    """Tests that we can create a new candidate and uprev and old rc."""
    # Let's stub out other LGKMManager calls cause they're already
    # unit tested.
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager,
                             '_GetLatestCandidateByVersion')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_CreateNewBuildSpec')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_SetInFlight')

    my_info = manifest_version.VersionInfo('1.2.3.4')
    most_recent_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc12')
    new_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc13')

    lkgm_manager.LKGMManager._GetCurrentVersionInfo(
        self.version_file).AndReturn(my_info)
    lkgm_manager.LKGMManager._LoadSpecs(my_info)
    lkgm_manager.LKGMManager._GetLatestCandidateByVersion(my_info).AndReturn(
        most_recent_candidate)
    lkgm_manager.LKGMManager._CreateNewBuildSpec(
        most_recent_candidate).AndReturn(new_candidate.VersionString())
    lkgm_manager.LKGMManager._SetInFlight(
        mox.StrContains(new_candidate.VersionString()))

    self.mox.ReplayAll()
    candidate = self.manager.CreateNewCandidate(self.version_file)
    self.assertEqual(candidate, new_candidate.VersionString())
    self.mox.VerifyAll()

  def testCreateNewCandidateReturnNoneIfNoWorkToDo(self):
    """Tests that we return nothing if there is nothing to create."""
    # Let's stub out other LGKMManager calls cause they're already
    # unit tested.
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager,
                             '_GetLatestCandidateByVersion')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_CreateNewBuildSpec')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_SetInFlight')

    my_info = manifest_version.VersionInfo('1.2.3.4')
    most_recent_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc12')

    lkgm_manager.LKGMManager._GetCurrentVersionInfo(
        self.version_file).AndReturn(my_info)
    lkgm_manager.LKGMManager._LoadSpecs(my_info)
    lkgm_manager.LKGMManager._GetLatestCandidateByVersion(my_info).AndReturn(
        most_recent_candidate)
    lkgm_manager.LKGMManager._CreateNewBuildSpec(
        most_recent_candidate).AndReturn(None)

    self.mox.ReplayAll()
    candidate = self.manager.CreateNewCandidate(self.version_file)
    self.assertEqual(candidate, None)
    self.mox.VerifyAll()

  def testGetLatestCandidate(self):
    """Makes sure we can get the latest created candidate manifest."""
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_SetInFlight')

    my_info = manifest_version.VersionInfo('1.2.3.4')
    most_recent_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc12')

    lkgm_manager.LKGMManager._GetCurrentVersionInfo(
        self.version_file).AndReturn(my_info)
    lkgm_manager.LKGMManager._LoadSpecs(my_info)
    lkgm_manager.LKGMManager._SetInFlight(
        mox.StrContains(most_recent_candidate.VersionString()))

    self.mox.ReplayAll()
    self.manager.latest_unprocessed = '1.2.3.4-rc12'
    candidate = self.manager.GetLatestCandidate(self.version_file)
    self.assertEqual(candidate, most_recent_candidate.VersionString())
    self.mox.VerifyAll()

  def testGetLatestCandidateNone(self):
    """Makes sure we get nothing if there is no work to be done."""
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo')
    self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs')

    my_info = manifest_version.VersionInfo('1.2.3.4')

    lkgm_manager.LKGMManager._GetCurrentVersionInfo(
        self.version_file).AndReturn(my_info)
    lkgm_manager.LKGMManager._LoadSpecs(my_info)

    self.mox.ReplayAll()
    candidate = self.manager.GetLatestCandidate(self.version_file)
    self.assertEqual(candidate, None)
    self.mox.VerifyAll()

  def _CreateManifest(self):
    """Returns a created test manifest in tmpdir with its dir_pfx."""
    self.manager.current_version = '1.2.3.4-rc21'
    dir_pfx = '1.2'
    manifest = os.path.join(self.manager.manifests_dir,
                            lkgm_manager.LKGMManager.LGKM_SUBDIR, 'buildspecs',
                            dir_pfx, '1.2.3.4-rc21.xml')
    manifest_version_unittest.TouchFile(manifest)
    return manifest, dir_pfx

  @staticmethod
  def _FinishBuilderHelper(manifest, path_for_builder, dir_pfx, status, wait=0):
    time.sleep(wait)
    manifest_version._CreateSymlink(
        manifest, os.path.join(path_for_builder, status, dir_pfx,
                               os.path.basename(manifest)))

  def _FinishBuild(self, manifest, path_for_builder, dir_pfx, status, wait=0):
    """Finishes a build by marking a status with optional delay."""
    if wait > 0:
      thread = threading.Thread(target=self._FinishBuilderHelper,
                                args=(manifest, path_for_builder, dir_pfx,
                                      status, wait))
      thread.start()
      return thread
    else:
      self._FinishBuilderHelper(manifest, path_for_builder, dir_pfx, status, 0)

  def testGetBuildersStatusBothFinished(self):
    """Tests GetBuilderStatus where both builds have finished."""
    self.mox.StubOutWithMock(lkgm_manager, '_SyncGitRepo')

    manifest, dir_pfx = self._CreateManifest()
    for_build1 = os.path.join(self.manager.manifests_dir,
                              lkgm_manager.LKGMManager.LGKM_SUBDIR,
                              'build-name', 'build1')
    for_build2 = os.path.join(self.manager.manifests_dir,
                              lkgm_manager.LKGMManager.LGKM_SUBDIR,
                              'build-name', 'build2')

    self._FinishBuild(manifest, for_build1, dir_pfx, 'fail')
    self._FinishBuild(manifest, for_build2, dir_pfx, 'pass')

    lkgm_manager._SyncGitRepo(self.manager.manifests_dir)
    self.mox.ReplayAll()
    statuses = self.manager.GetBuildersStatus(['build1', 'build2'])
    self.assertEqual(statuses['build1'], 'fail')
    self.assertEqual(statuses['build2'], 'pass')
    self.mox.VerifyAll()

  def testGetBuildersStatusWaitForOne(self):
    """Tests GetBuilderStatus where both builds have finished with one delay."""
    self.mox.StubOutWithMock(lkgm_manager, '_SyncGitRepo')

    manifest, dir_pfx = self._CreateManifest()
    for_build1 = os.path.join(self.manager.manifests_dir,
                              lkgm_manager.LKGMManager.LGKM_SUBDIR,
                              'build-name', 'build1')
    for_build2 = os.path.join(self.manager.manifests_dir,
                              lkgm_manager.LKGMManager.LGKM_SUBDIR,
                              'build-name', 'build2')

    self._FinishBuild(manifest, for_build1, dir_pfx, 'fail')
    self._FinishBuild(manifest, for_build2, dir_pfx, 'pass', wait=3)

    lkgm_manager._SyncGitRepo(self.manager.manifests_dir).MultipleTimes()
    self.mox.ReplayAll()

    statuses = self.manager.GetBuildersStatus(['build1', 'build2'])
    self.assertEqual(statuses['build1'], 'fail')
    self.assertEqual(statuses['build2'], 'pass')
    self.mox.VerifyAll()

  def testGetBuildersStatusReachTimeout(self):
    """Tests GetBuilderStatus where one build finishes and one never does."""
    self.mox.StubOutWithMock(lkgm_manager, '_SyncGitRepo')

    manifest, dir_pfx = self._CreateManifest()
    for_build1 = os.path.join(self.manager.manifests_dir,
                              lkgm_manager.LKGMManager.LGKM_SUBDIR,
                              'build-name', 'build1')
    for_build2 = os.path.join(self.manager.manifests_dir,
                              lkgm_manager.LKGMManager.LGKM_SUBDIR,
                              'build-name', 'build2')

    self._FinishBuild(manifest, for_build1, dir_pfx, 'fail', wait=3)
    thread = self._FinishBuild(manifest, for_build2, dir_pfx, 'pass', wait=10)

    lkgm_manager._SyncGitRepo(self.manager.manifests_dir).MultipleTimes()
    self.mox.ReplayAll()
    # Let's reduce this.
    self.manager.MAX_TIMEOUT_SECONDS = 5
    statuses = self.manager.GetBuildersStatus(['build1', 'build2'])
    self.assertEqual(statuses['build1'], 'fail')
    self.assertEqual(statuses['build2'], None)
    thread.join()
    self.mox.VerifyAll()

  def tearDown(self):
    shutil.rmtree(self.tmpdir)


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