# 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 repository.py."""

import os
import time
from unittest import mock

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


# pylint: disable=protected-access


class RepositoryTests(cros_test_lib.RunCommandTestCase):
    """Test cases related to repository checkout methods."""

    def testIsLocalPath(self) -> None:
        """test IsLocalPath."""
        self.assertTrue(repository._IsLocalPath("/tmp/chromiumos/"))
        self.assertTrue(repository._IsLocalPath("file:///chromiumos/"))
        self.assertFalse(repository._IsLocalPath("https://chromiumos/"))
        self.assertFalse(repository._IsLocalPath("http://chromiumos/"))
        self.assertFalse(repository._IsLocalPath("ssh://chromiumos/"))
        self.assertFalse(repository._IsLocalPath("git://chromiumos/"))


class RepoInitTests(cros_test_lib.TempDirTestCase, cros_test_lib.MockTestCase):
    """Test cases related to repository initialization."""

    def __init__(self, *args, **kwargs) -> None:
        self.repo = None
        super().__init__(*args, **kwargs)

    def setUp(self) -> None:
        self.PatchObject(time, "sleep")

    def _Initialize(self, branch="master") -> None:
        site_params = config_lib.GetSiteParams()
        self.repo = repository.RepoRepository(
            site_params.MANIFEST_URL, self.tempdir, branch=branch
        )
        self.repo.Initialize()

    @cros_test_lib.pytestmark_network_test
    def testReInitialization(self) -> None:
        """Test ability to switch between branches."""
        self._Initialize("release-R19-2046.B")
        self._Initialize("master")

        # Test that a failed re-init due to bad branch doesn't leave repo in bad
        # state.
        # repo init on 'monkey' will retry on failures.
        self.assertRaises(Exception, self._Initialize, "monkey")
        self._Initialize("release-R20-2268.B")

    @cros_test_lib.pytestmark_network_test
    def testBuildRootGitCleanup(self) -> None:
        """Test successful repo cleanup."""
        self._Initialize()
        run_cmd_mock = self.PatchObject(
            cros_build_lib, "run", wraps=cros_build_lib.run
        )
        self.repo.BuildRootGitCleanup(prune_all=True)

        # run should be called twice.
        self.assertEqual(run_cmd_mock.call_count, 2)

    @cros_test_lib.pytestmark_network_test
    def testCleanStaleLocks(self) -> None:
        """Test successful repo lock cleanup."""
        self._Initialize("release-R19-2046.B")
        self.PatchObject(git, "GetGitGitdir")
        dsl = self.PatchObject(git, "DeleteStaleLocks")
        self.repo.CleanStaleLocks()

        self.assertEqual(dsl.call_count, 198)

    def testInitializationWithRepoInitRetry(self) -> None:
        """Test Initialization with repo init retry."""
        self.PatchObject(repository.RepoRepository, "_RepoSelfupdate")
        mock_cleanup = self.PatchObject(
            repository.RepoRepository, "_CleanUpRepoManifest"
        )
        error_result = cros_build_lib.CompletedProcess(["cmd"], returncode=1)
        ex = cros_build_lib.RunCommandError("error_msg", error_result)
        mock_init = self.PatchObject(cros_build_lib, "run", side_effect=ex)

        self.assertRaises(Exception, self._Initialize)
        self.assertEqual(
            mock_cleanup.call_count, repository.REPO_INIT_RETRY_LIMIT + 1
        )
        self.assertEqual(
            mock_init.call_count, repository.REPO_INIT_RETRY_LIMIT + 1
        )

    def testInitializationWithoutRepoInitRetry(self) -> None:
        """Test Initialization without repo init retry."""
        self.PatchObject(repository.RepoRepository, "_RepoSelfupdate")
        mock_cleanup = self.PatchObject(
            repository.RepoRepository, "_CleanUpRepoManifest"
        )
        mock_init = self.PatchObject(cros_build_lib, "run")

        self._Initialize()
        self.assertEqual(mock_cleanup.call_count, 0)
        self.assertEqual(mock_init.call_count, 1)


class RepoInitChromeBotTests(RepoInitTests):
    """Test that Re-init works with the chrome-bot account.

    In testing, repo init behavior on the buildbots is different from a
    local run, because there is some logic in 'repo' that filters changes based
    on GIT_COMMITTER_IDENT.  So for sanity's sake, try to emulate running on the
    buildbots.
    """

    def setUp(self) -> None:
        os.putenv("GIT_COMMITTER_EMAIL", "chrome-bot@chromium.org")
        os.putenv("GIT_AUTHOR_EMAIL", "chrome-bot@chromium.org")


class RepoSyncTests(cros_test_lib.TempDirTestCase, cros_test_lib.MockTestCase):
    """Test cases related to repository Sync"""

    def setUp(self) -> None:
        site_params = config_lib.GetSiteParams()
        self.repo = repository.RepoRepository(
            site_params.MANIFEST_URL, self.tempdir, branch="master"
        )
        self.PatchObject(repository.RepoRepository, "_RepoDebugInfo")
        self.PatchObject(repository.RepoRepository, "Initialize")
        self.PatchObject(repository.RepoRepository, "_EnsureMirroring")
        self.PatchObject(repository.RepoRepository, "BuildRootGitCleanup")
        self.PatchObject(time, "sleep")

    def testSyncWithException(self) -> None:
        """Test Sync retry on repo network sync failure"""
        # Return value here isn't super important.
        result = cros_build_lib.CompletedProcess(
            ["cmd"], returncode=0, stderr="error"
        )
        ex = cros_build_lib.RunCommandError("msg", result)

        run_cmd_mock = self.PatchObject(cros_build_lib, "run", side_effect=ex)

        # repo.Sync raises SrcCheckOutException.
        self.assertRaises(
            repository.SrcCheckOutException,
            self.repo.Sync,
            local_manifest="local_manifest",
            network_only=True,
        )

        # run should be called SYNC_RETRIES + 1 times.
        self.assertEqual(run_cmd_mock.call_count, constants.SYNC_RETRIES + 1)

    def testSyncWithoutException(self) -> None:
        """Test successful repo sync without exception and retry"""
        # Return value here isn't super important.
        run_cmd_mock = self.PatchObject(cros_build_lib, "run")
        self.repo.Sync(local_manifest="local_manifest", network_only=True)

        # run should be called once.
        self.assertEqual(run_cmd_mock.call_count, 1)

    def test_RepoSelfupdateRaisesWarning(self) -> None:
        """Test _RepoSelfupdate when repo version warning is raised."""
        warnning_stderr = """
info: A new version of repo is available

...

gpg: Can't check signature: public key not found

...

warning: Skipped upgrade to unverified version
"""
        cmd_result = cros_build_lib.CompletedProcess(stderr=warnning_stderr)
        self.PatchObject(cros_build_lib, "run", return_value=cmd_result)
        with mock.patch.object(osutils, "RmDir") as mock_rm:
            self.repo._RepoSelfupdate()
        mock_rm.assert_called_once_with(mock.ANY, ignore_missing=True)

    def test_RepoSelfupdateRaisesException(self) -> None:
        """Test _RepoSelfupdate when exception is raised."""
        ex = cros_build_lib.RunCommandError("msg")
        self.PatchObject(cros_build_lib, "run", side_effect=ex)
        with mock.patch.object(osutils, "RmDir") as mock_rm:
            self.repo._RepoSelfupdate()
        mock_rm.assert_called_once_with(mock.ANY, ignore_missing=True)

    def test_RepoSelfupdateSucceeds(self) -> None:
        cmd_result = cros_build_lib.CompletedProcess()
        self.PatchObject(cros_build_lib, "run", return_value=cmd_result)
        with mock.patch.object(osutils, "RmDir") as mock_rm:
            self.repo._RepoSelfupdate()
        self.assertFalse(mock_rm.called)