# Copyright 2012 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unittests for lkgm_manager"""

import contextlib
import os
import tempfile
from unittest import mock
from xml.dom import minidom

from chromite.cbuildbot import cbuildbot_alerts
from chromite.cbuildbot import lkgm_manager
from chromite.cbuildbot import manifest_version
from chromite.cbuildbot import repository
from chromite.lib import config_lib
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import git
from chromite.lib import osutils
from chromite.lib.buildstore import FakeBuildStore


FAKE_VERSION_STRING = "1.2.4-rc3"
FAKE_VERSION_STRING_NEXT = "1.2.4-rc4"
CHROME_BRANCH = "13"

FAKE_VERSION = """
CHROMEOS_BUILD=1
CHROMEOS_BRANCH=2
CHROMEOS_PATCH=4
CHROME_BRANCH=13
"""


# pylint: disable=protected-access


class LKGMCandidateInfoTest(cros_test_lib.TestCase):
    """Test methods testing methods in _LKGMCandidateInfo class."""

    def testLoadFromString(self):
        """Tests whether we can load from a string."""
        info = lkgm_manager._LKGMCandidateInfo(
            version_string=FAKE_VERSION_STRING, chrome_branch=CHROME_BRANCH
        )
        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, chrome_branch=CHROME_BRANCH
        )
        info.IncrementVersion()
        self.assertEqual(info.VersionString(), FAKE_VERSION_STRING_NEXT)

    def testVersionCompare(self):
        """Tests whether our comparision method works."""
        info0 = lkgm_manager._LKGMCandidateInfo("5.2.3-rc100")
        info1 = lkgm_manager._LKGMCandidateInfo("1.2.3-rc1")
        info2 = lkgm_manager._LKGMCandidateInfo("1.2.3-rc2")
        info3 = lkgm_manager._LKGMCandidateInfo("1.2.200-rc1")
        info4 = lkgm_manager._LKGMCandidateInfo("1.4.3-rc1")

        self.assertGreater(info0, info1)
        self.assertGreater(info0, info2)
        self.assertGreater(info0, info3)
        self.assertGreater(info0, info4)
        self.assertGreater(info2, info1)
        self.assertGreater(info3, info1)
        self.assertGreater(info3, info2)
        self.assertGreater(info4, info1)
        self.assertGreater(info4, info2)
        self.assertGreater(info4, info3)
        self.assertEqual(info0, info0)
        self.assertEqual(info1, info1)
        self.assertEqual(info2, info2)
        self.assertEqual(info3, info3)
        self.assertEqual(info4, info4)
        self.assertNotEqual(info0, info1)
        self.assertNotEqual(info0, info2)
        self.assertNotEqual(info0, info3)
        self.assertNotEqual(info0, info4)
        self.assertNotEqual(info1, info0)
        self.assertNotEqual(info1, info2)
        self.assertNotEqual(info1, info3)
        self.assertNotEqual(info1, info4)
        self.assertNotEqual(info2, info0)
        self.assertNotEqual(info2, info1)
        self.assertNotEqual(info2, info3)
        self.assertNotEqual(info2, info4)
        self.assertNotEqual(info3, info0)
        self.assertNotEqual(info3, info1)
        self.assertNotEqual(info3, info2)
        self.assertNotEqual(info3, info4)
        self.assertNotEqual(info4, info0)
        self.assertNotEqual(info4, info1)
        self.assertNotEqual(info4, info1)
        self.assertNotEqual(info4, info3)


@contextlib.contextmanager
def TemporaryManifest():
    with tempfile.NamedTemporaryFile(mode="w") as f:
        # Create fake but empty manifest file.
        new_doc = minidom.getDOMImplementation().createDocument(
            None, "manifest", None
        )
        print(new_doc.toxml())
        new_doc.writexml(f)
        f.flush()
        yield f


class LKGMManagerTest(cros_test_lib.MockTempDirTestCase):
    """Tests for the BuildSpecs manager."""

    def setUp(self):
        self.manager = None
        self.push_mock = self.PatchObject(git, "CreatePushBranch")

        self.source_repo = "ssh://source/repo"
        self.manifest_repo = "ssh://manifest/repo"
        self.version_file = "version-file.sh"
        self.branch = "master"
        self.build_name = "amd64-generic"
        self.incr_type = "branch"
        self.buildstore = FakeBuildStore()

        # Create tmp subdirs based on the one provided TempDirMixin.
        self.tmpdir = os.path.join(self.tempdir, "base")
        osutils.SafeMakedirs(self.tmpdir)
        self.tmpmandir = os.path.join(self.tempdir, "man")
        osutils.SafeMakedirs(self.tmpmandir)

    def _LKGMManager(self, build_type=constants.PFQ_TYPE, config=None):
        repo = repository.RepoRepository(
            self.source_repo, self.tmpdir, self.branch, depth=1
        )
        manager = lkgm_manager.LKGMManager(
            repo,
            self.manifest_repo,
            self.build_name,
            build_type,
            "branch",
            force=False,
            branch=self.branch,
            config=config,
            buildstore=self.buildstore,
            dry_run=True,
        )
        manager.manifest_dir = self.tmpmandir
        manager.lkgm_path = os.path.join(
            self.tmpmandir, constants.LKGM_MANIFEST
        )

        manager.all_specs_dir = "/LKGM/path"
        manifest_dir = manager.manifest_dir
        manager.specs_for_builder = os.path.join(
            manifest_dir, manager.rel_working_dir, "build-name", "%(builder)s"
        )
        manager.SLEEP_TIMEOUT = 0
        return manager

    def _GetPathToManifest(self, info):
        return os.path.join(
            self.manager.all_specs_dir, "%s.xml" % info.VersionString()
        )

    def testCreateFromManifest(self):
        """Tests that we can create a new candidate from another manifest."""
        self.manager = self._LKGMManager()
        # Let's stub out other LKGMManager calls cause they're already
        # unit tested.

        version = "2010.0.0-rc7"
        my_info = lkgm_manager._LKGMCandidateInfo("2010.0.0")
        new_candidate = lkgm_manager._LKGMCandidateInfo(version)
        manifest = (
            "/tmp/manifest-versions-internal/paladin/buildspecs/"
            "20/%s.xml" % version
        )
        new_manifest = "/path/to/tmp/file.xml"

        build_id = 20162

        site_params = config_lib.GetSiteParams()
        # Patch out our RepoRepository to make sure we don't corrupt real repo.
        self.PatchObject(self.manager, "cros_source")
        filter_mock = self.PatchObject(
            manifest_version, "FilterManifest", return_value=new_manifest
        )

        # Do manifest refresh work.
        self.PatchObject(
            lkgm_manager.LKGMManager,
            "GetCurrentVersionInfo",
            return_value=my_info,
        )
        self.PatchObject(lkgm_manager.LKGMManager, "RefreshManifestCheckout")
        init_mock = self.PatchObject(
            lkgm_manager.LKGMManager, "InitializeManifestVariables"
        )

        # Publish new candidate.
        publish_mock = self.PatchObject(
            lkgm_manager.LKGMManager, "PublishManifest"
        )

        candidate_path = self.manager.CreateFromManifest(
            manifest, build_id=build_id
        )
        self.assertEqual(candidate_path, self._GetPathToManifest(new_candidate))
        self.assertEqual(self.manager.current_version, version)

        filter_mock.assert_called_once_with(
            manifest, whitelisted_remotes=site_params.EXTERNAL_REMOTES
        )
        publish_mock.assert_called_once_with(
            new_manifest, version, build_id=build_id
        )
        init_mock.assert_called_once_with(my_info)
        self.push_mock.assert_called_once_with(mock.ANY, mock.ANY, sync=False)

    def testCreateNewCandidateReturnNoneIfNoWorkToDo(self):
        """Tests that we return nothing if there is nothing to create."""
        self.manager = self._LKGMManager()
        new_manifest = "some_manifest"
        my_info = lkgm_manager._LKGMCandidateInfo("1.2.3")

        # Patch out our RepoRepository to make sure we don't corrupt real repo.
        cros_source_mock = self.PatchObject(self.manager, "cros_source")
        cros_source_mock.branch = "master"
        cros_source_mock.directory = "/foo/repo"

        self.PatchObject(lkgm_manager.LKGMManager, "CheckoutSourceCode")
        self.PatchObject(
            lkgm_manager.LKGMManager,
            "CreateManifest",
            return_value=new_manifest,
        )
        self.PatchObject(lkgm_manager.LKGMManager, "RefreshManifestCheckout")
        self.PatchObject(
            lkgm_manager.LKGMManager,
            "GetCurrentVersionInfo",
            return_value=my_info,
        )
        init_mock = self.PatchObject(
            lkgm_manager.LKGMManager, "InitializeManifestVariables"
        )
        self.PatchObject(
            lkgm_manager.LKGMManager, "HasCheckoutBeenBuilt", return_value=True
        )

        candidate = self.manager.CreateNewCandidate()
        self.assertEqual(candidate, None)
        init_mock.assert_called_once_with(my_info)

    def _CreateManifest(self):
        """Returns a created test manifest in tmpdir with its dir_pfx."""
        self.manager.current_version = "1.2.4-rc21"
        dir_pfx = CHROME_BRANCH
        manifest = os.path.join(
            self.manager.manifest_dir,
            self.manager.rel_working_dir,
            "buildspecs",
            dir_pfx,
            "1.2.4-rc21.xml",
        )
        osutils.Touch(manifest)
        return manifest, dir_pfx

    def _MockParseGitLog(self, fake_git_log, project):
        exists_mock = self.PatchObject(os.path, "exists", return_value=True)
        link_mock = self.PatchObject(cbuildbot_alerts, "PrintBuildbotLink")
        fake_project_handler = mock.Mock(spec=git.Manifest)
        fake_project_handler.checkouts_by_path = {project["path"]: project}
        self.PatchObject(git, "Manifest", return_value=fake_project_handler)

        fake_result = cros_build_lib.CompletedProcess(stdout=fake_git_log)
        self.PatchObject(git, "RunGit", return_value=fake_result)

        return exists_mock, link_mock

    def testAddChromeVersionToManifest(self):
        """Tests whether we can write the chrome version to the manifest file."""
        self.manager = self._LKGMManager()
        with TemporaryManifest() as f:
            chrome_version = "35.0.1863.0"
            # Write the chrome element to manifest.
            self.manager._AddChromeVersionToManifest(f.name, chrome_version)

            # Read the manifest file.
            new_doc = minidom.parse(f.name)
            elements = new_doc.getElementsByTagName(lkgm_manager.CHROME_ELEMENT)
            self.assertEqual(len(elements), 1)
            self.assertEqual(
                elements[0].getAttribute(lkgm_manager.CHROME_VERSION_ATTR),
                chrome_version,
            )