import datetime

import os

import sys

import mock

import pytest

from kallithea.lib.vcs.backends.git import GitChangeset, GitRepository

from kallithea.lib.vcs.exceptions import NodeDoesNotExistError, RepositoryError, VCSError

from kallithea.lib.vcs.nodes import DirNode, FileNode, NodeKind, NodeState

from kallithea.model.scm import ScmModel

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, TESTS_TMP_PATH, get_new_dir

class TestGitRepository(object):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_GIT_REPO_CLONE):

            pytest.fail('Cannot test git clone repo as location %s already '

                      'exists. You should manually remove it first.'

                      % TEST_GIT_REPO_CLONE)

    def setup_method(self):

        self.repo = GitRepository(TEST_GIT_REPO)

    def test_wrong_repo_path(self):

        wrong_repo_path = os.path.join(TESTS_TMP_PATH, 'errorrepo')

        with pytest.raises(RepositoryError):

            GitRepository(wrong_repo_path)

    def test_git_cmd_injection(self):

            # Should fail because URL will contain the parts after ; too

            GitRepository(get_new_dir('injection-repo'), src_url=repo_inject_path, update_after_clone=True, create=True)

        with pytest.raises(RepositoryError):

            # Should fail on direct clone call, which as of this writing does not happen outside of class

            clone_fail_repo = GitRepository(get_new_dir('injection-repo'), create=True)

            clone_fail_repo.clone(repo_inject_path, update_after_clone=True,)

        # Verify correct quoting of evil characters that should work on posix file systems

        if sys.platform == 'win32':

            # windows does not allow '"' in dir names

            # and some versions of the git client don't like ` and '

            tricky_path = get_new_dir("tricky-path-repo-$")

        else:

            tricky_path = get_new_dir("tricky-path-repo-$'\"`")

        successfully_cloned = GitRepository(tricky_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)

        # Repo should have been created

        assert not successfully_cloned._repo.bare

        if sys.platform == 'win32':

            # windows does not allow '"' in dir names

            # and some versions of the git client don't like ` and '

            tricky_path_2 = get_new_dir("tricky-path-2-repo-$")

        else:

            tricky_path_2 = get_new_dir("tricky-path-2-repo-$'\"`")

        successfully_cloned2 = GitRepository(tricky_path_2, src_url=tricky_path, bare=True, create=True)

        # Repo should have been created and thus used correct quoting for clone

        assert successfully_cloned2._repo.bare

        # Should pass because URL has been properly quoted

        successfully_cloned.pull(tricky_path_2)

        successfully_cloned2.fetch(tricky_path)

    def test_repo_create_with_spaces_in_path(self):

        repo_path = get_new_dir("path with spaces")

        repo = GitRepository(repo_path, src_url=None, bare=True, create=True)

        # Repo should have been created

        assert repo._repo.bare

    def test_repo_clone(self):

        self.__check_for_existing_repo()

        repo = GitRepository(TEST_GIT_REPO)

        repo_clone = GitRepository(TEST_GIT_REPO_CLONE,

            src_url=TEST_GIT_REPO, create=True, update_after_clone=True)

        assert len(repo.revisions) == len(repo_clone.revisions)

        # Checking hashes of changesets should be enough

        for changeset in repo.get_changesets():

            raw_id = changeset.raw_id

            assert raw_id == repo_clone.get_changeset(raw_id).raw_id

    def test_repo_clone_with_spaces_in_path(self):

        repo_path = get_new_dir("path with spaces")

        successfully_cloned = GitRepository(repo_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)

        # Repo should have been created

        assert not successfully_cloned._repo.bare

        successfully_cloned.pull(TEST_GIT_REPO)

        self.repo.fetch(repo_path)

    def test_repo_clone_without_create(self):

        with pytest.raises(RepositoryError):

            GitRepository(TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)

    def test_repo_clone_with_update(self):

        repo = GitRepository(TEST_GIT_REPO)

        clone_path = TEST_GIT_REPO_CLONE + '_with_update'

        repo_clone = GitRepository(clone_path,

            create=True, src_url=TEST_GIT_REPO, update_after_clone=True)

        assert len(repo.revisions) == len(repo_clone.revisions)

        # check if current workdir was updated

        fpath = os.path.join(clone_path, 'MANIFEST.in')

        assert os.path.isfile(fpath) == True, 'Repo was cloned and updated but file %s could not be found' % fpath

    def test_repo_clone_without_update(self):

        repo = GitRepository(TEST_GIT_REPO)

        clone_path = TEST_GIT_REPO_CLONE + '_without_update'

        repo_clone = GitRepository(clone_path,

            create=True, src_url=TEST_GIT_REPO, update_after_clone=False)

        assert len(repo.revisions) == len(repo_clone.revisions)

        # check if current workdir was *NOT* updated

        fpath = os.path.join(clone_path, 'MANIFEST.in')

        # Make sure it's not bare repo

        assert not repo_clone._repo.bare

        assert os.path.isfile(fpath) == False, 'Repo was cloned and updated but file %s was found' % fpath

    def test_repo_clone_into_bare_repo(self):

        repo = GitRepository(TEST_GIT_REPO)

        clone_path = TEST_GIT_REPO_CLONE + '_bare.git'

        repo_clone = GitRepository(clone_path, create=True,

            src_url=repo.path, bare=True)

        assert repo_clone._repo.bare

    def test_create_repo_is_not_bare_by_default(self):

        repo = GitRepository(get_new_dir('not-bare-by-default'), create=True)

        assert not repo._repo.bare

    def test_create_bare_repo(self):

        repo = GitRepository(get_new_dir('bare-repo'), create=True, bare=True)

        assert repo._repo.bare

    def test_revisions(self):

        # there are 112 revisions (by now)

        # so we can assume they would be available from now on

        subset = set([

            'c1214f7e79e02fc37156ff215cd71275450cffc3',

            '38b5fe81f109cb111f549bfe9bb6b267e10bc557',

            'fa6600f6848800641328adbf7811fd2372c02ab2',

            '102607b09cdd60e2793929c4f90478be29f85a17',

            '49d3fd156b6f7db46313fac355dca1a0b94a0017',

            '2d1028c054665b962fa3d307adfc923ddd528038',

            'd7e0d30fbcae12c90680eb095a4f5f02505ce501',

            'ff7ca51e58c505fec0dd2491de52c622bb7a806b',

            'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',

            '8430a588b43b5d6da365400117c89400326e7992',

            'd955cd312c17b02143c04fa1099a352b04368118',

            'f67b87e5c629c2ee0ba58f85197e423ff28d735b',

            'add63e382e4aabc9e1afdc4bdc24506c269b7618',

            'f298fe1189f1b69779a4423f40b48edf92a703fc',

            'bd9b619eb41994cac43d67cf4ccc8399c1125808',

            '6e125e7c890379446e98980d8ed60fba87d0f6d1',

            'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',

            '0b05e4ed56c802098dfc813cbe779b2f49e92500',

            '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',

            '45223f8f114c64bf4d6f853e3c35a369a6305520',

            'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',

            'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',

            '27d48942240f5b91dfda77accd2caac94708cc7d',

            '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',

            'e686b958768ee96af8029fe19c6050b1a8dd3b2b'])

        assert subset.issubset(set(self.repo.revisions))

    def test_slicing(self):

        # 4 1 5 10 95

        for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),

                                 (10, 20, 10), (5, 100, 95)]:

            revs = list(self.repo[sfrom:sto])

            assert len(revs) == size

            assert revs[0] == self.repo.get_changeset(sfrom)

            assert revs[-1] == self.repo.get_changeset(sto - 1)

    def test_branches(self):

        # TODO: Need more tests here

        # Removed (those are 'remotes' branches for cloned repo)

        #assert 'master' in self.repo.branches

        #assert 'gittree' in self.repo.branches

        #assert 'web-branch' in self.repo.branches

        for name, id in self.repo.branches.items():

            assert isinstance(self.repo.get_changeset(id), GitChangeset)

    def test_tags(self):

        # TODO: Need more tests here

        assert 'v0.1.1' in self.repo.tags

        assert 'v0.1.2' in self.repo.tags

        for name, id in self.repo.tags.items():

            assert isinstance(self.repo.get_changeset(id), GitChangeset)

    def _test_single_changeset_cache(self, revision):

        chset = self.repo.get_changeset(revision)

        assert revision in self.repo.changesets

        assert chset is self.repo.changesets[revision]

    def test_initial_changeset(self):

        id = self.repo.revisions[0]

        init_chset = self.repo.get_changeset(id)

        assert init_chset.message == 'initial import\n'

        assert init_chset.author == 'Marcin Kuzminski <marcin@python-blog.com>'

        for path in ('vcs/__init__.py',

                     'vcs/backends/BaseRepository.py',

                     'vcs/backends/__init__.py'):

            assert isinstance(init_chset.get_node(path), FileNode)

        for path in ('', 'vcs', 'vcs/backends'):

            assert isinstance(init_chset.get_node(path), DirNode)

        with pytest.raises(NodeDoesNotExistError):

            init_chset.get_node(path='foobar')

        node = init_chset.get_node('vcs/')

        assert hasattr(node, 'kind')

        assert node.kind == NodeKind.DIR

        node = init_chset.get_node('vcs')

        assert hasattr(node, 'kind')

        assert node.kind == NodeKind.DIR

        node = init_chset.get_node('vcs/__init__.py')

        assert hasattr(node, 'kind')

        assert node.kind == NodeKind.FILE

    def test_not_existing_changeset(self):

        with pytest.raises(RepositoryError):

            self.repo.get_changeset('f' * 40)