from rhodecode.lib.vcs.utils.hgcompat import (

    hg_url, httpbasicauthhandler, httpdigestauthhandler

)

from .changeset import GitChangeset

from .config import ConfigFile

from .inmemory import GitInMemoryChangeset

from .workdir import GitWorkdir

SHA_PATTERN = re.compile(r'^[[0-9a-fA-F]{12}|[0-9a-fA-F]{40}]$')

log = logging.getLogger(__name__)

class GitRepository(BaseRepository):

    """

    Git repository backend.

    """

    DEFAULT_BRANCH_NAME = 'master'

    scm = 'git'

    def __init__(self, repo_path, create=False, src_url=None,

                 update_after_clone=False, bare=False):

        self.path = abspath(repo_path)

        repo = self._get_repo(create, src_url, update_after_clone, bare)

        self.bare = repo.bare

    @property

    def _config_files(self):

        return [

            self.bare and abspath(self.path, 'config')

                      or abspath(self.path, '.git', 'config'),

             abspath(get_user_home(), '.gitconfig'),

         ]

    @property

    def _repo(self):

        return Repo(self.path)

    @property

    def head(self):

        try:

            return self._repo.head()

        except KeyError:

            return None

    @LazyProperty

    def revisions(self):

        """

        Returns list of revisions' ids, in ascending order.  Being lazy

        attribute allows external tools to inject shas from cache.

        """

        return self._get_all_revisions()

    @classmethod

    def _run_git_command(cls, cmd, **opts):

        """

        Runs given ``cmd`` as git command and returns tuple

        (stdout, stderr).

        :param cmd: git command to be executed

        :param opts: env options to pass into Subprocess command

        """

        if '_bare' in opts:

            _copts = []

            del opts['_bare']

        else:

            _copts = ['-c', 'core.quotepath=false', ]

        safe_call = False

        if '_safe' in opts:

            #no exc on failure

            del opts['_safe']

            safe_call = True

        _str_cmd = False

        if isinstance(cmd, basestring):

            cmd = [cmd]

            _str_cmd = True

        gitenv = os.environ

        # need to clean fix GIT_DIR !

        if 'GIT_DIR' in gitenv:

            del gitenv['GIT_DIR']

        gitenv['GIT_CONFIG_NOGLOBAL'] = '1'

        _git_path = settings.GIT_EXECUTABLE_PATH

        cmd = [_git_path] + _copts + cmd

        if _str_cmd:

            cmd = ' '.join(cmd)

        try:

            _opts = dict(

                env=gitenv,

        try:

            if create and src_url:

                GitRepository._check_url(src_url)

                self.clone(src_url, update_after_clone, bare)

                return Repo(self.path)

            elif create:

                os.mkdir(self.path)

                if bare:

                    return Repo.init_bare(self.path)

                else:

                    return Repo.init(self.path)

            else:

                return self._repo

        except (NotGitRepository, OSError), err:

            raise RepositoryError(err)

    def _get_all_revisions(self):

        # we must check if this repo is not empty, since later command

        # fails if it is. And it's cheaper to ask than throw the subprocess

        # errors

        try:

            self._repo.head()

        except KeyError:

            return []

        rev_filter = _git_path = settings.GIT_REV_FILTER

        cmd = 'rev-list %s --reverse --date-order' % (rev_filter)

        try:

            so, se = self.run_git_command(cmd)

        except RepositoryError:

            # Can be raised for empty repositories

            return []

        return so.splitlines()

    def _get_all_revisions2(self):

        #alternate implementation using dulwich

        includes = [x[1][0] for x in self._parsed_refs.iteritems()

                    if x[1][1] != 'T']

        return [c.commit.id for c in self._repo.get_walker(include=includes)]

    def _get_revision(self, revision):

        """

        For git backend we always return integer here. This way we ensure

        that changset's revision attribute would become integer.

        """

        is_null = lambda o: len(o) == revision.count('0')

            raise EmptyRepositoryError("There are no changesets yet")

        if revision in (None, '', 'tip', 'HEAD', 'head', -1):

            return self.revisions[-1]

        is_bstr = isinstance(revision, (str, unicode))

        if ((is_bstr and revision.isdigit() and len(revision) < 12)

            or isinstance(revision, int) or is_null(revision)):

            try:

                revision = self.revisions[int(revision)]

            except Exception:

                raise ChangesetDoesNotExistError("Revision %s does not exist "

                    "for this repository" % (revision))

        elif is_bstr:

            # get by branch/tag name

            _ref_revision = self._parsed_refs.get(revision)

            if _ref_revision:  # and _ref_revision[1] in ['H', 'RH', 'T']:

                return _ref_revision[0]

            _tags_shas = self.tags.values()

            # maybe it's a tag ? we don't have them in self.revisions

            if revision in _tags_shas:

                return _tags_shas[_tags_shas.index(revision)]

            elif not SHA_PATTERN.match(revision) or revision not in self.revisions:

                raise ChangesetDoesNotExistError("Revision %s does not exist "

                    "for this repository" % (revision))

        # Ensure we return full id

        if not SHA_PATTERN.match(str(revision)):

            raise ChangesetDoesNotExistError("Given revision %s not recognized"

                % revision)

        return revision

    def _get_archives(self, archive_name='tip'):

        for i in [('zip', '.zip'), ('gz', '.tar.gz'), ('bz2', '.tar.bz2')]:

                yield {"type": i[0], "extension": i[1], "node": archive_name}

    def _get_url(self, url):

        """

        Returns normalized url. If schema is not given, would fall to

        filesystem (``file:///``) schema.

        """

        url = str(url)

        if url != 'default' and not '://' in url:

            url = ':///'.join(('file', url))

        heads = {}

        for key, val in refs.items():

            for ref_key in ['refs/heads/', 'refs/remotes/origin/']:

                if key.startswith(ref_key):

                    n = key[len(ref_key):]

                    if n not in ['HEAD']:

                        heads[n] = val

        return heads if reverse else dict((y, x) for x, y in heads.iteritems())

    def get_changeset(self, revision=None):

        """

        Returns ``GitChangeset`` object representing commit from git repository

        at the given revision or head (most recent commit) if None given.

        """

        if isinstance(revision, GitChangeset):

            return revision

        revision = self._get_revision(revision)

        changeset = GitChangeset(repository=self, revision=revision)

        return changeset

    def get_changesets(self, start=None, end=None, start_date=None,

           end_date=None, branch_name=None, reverse=False):

        """

        Returns iterator of ``GitChangeset`` objects from start to end (both

        are inclusive), in ascending date order (unless ``reverse`` is set).

        :param start: changeset ID, as str; first returned changeset

        :param end: changeset ID, as str; last returned changeset

        :param start_date: if specified, changesets with commit date less than

          ``start_date`` would be filtered out from returned set

        :param end_date: if specified, changesets with commit date greater than

          ``end_date`` would be filtered out from returned set

        :param branch_name: if specified, changesets not reachable from given

          branch would be filtered out from returned set

        :param reverse: if ``True``, returned generator would be reversed

          (meaning that returned changesets would have descending date order)

        :raise BranchDoesNotExistError: If given ``branch_name`` does not

            exist.

        :raise ChangesetDoesNotExistError: If changeset for given ``start`` or

          ``end`` could not be found.

        """

        if branch_name and branch_name not in self.branches:

            raise BranchDoesNotExistError("Branch '%s' not found" \

                                          % branch_name)

        # %H at format means (full) commit hash, initial hashes are retrieved

        # in ascending date order

        cmd_template = 'log --date-order --reverse --pretty=format:"%H"'

        cmd_params = {}

        if start_date:

            cmd_template += ' --since "$since"'

            cmd_params['since'] = start_date.strftime('%m/%d/%y %H:%M:%S')

        if end_date:

            cmd_template += ' --until "$until"'

            cmd_params['until'] = end_date.strftime('%m/%d/%y %H:%M:%S')

        if branch_name:

            cmd_template += ' $branch_name'

            cmd_params['branch_name'] = branch_name

        else:

            rev_filter = _git_path = settings.GIT_REV_FILTER

            cmd_template += ' %s' % (rev_filter)

        cmd = string.Template(cmd_template).safe_substitute(**cmd_params)

        revs = self.run_git_command(cmd)[0].splitlines()

        start_pos = 0

        end_pos = len(revs)

        if start:

            _start = self._get_revision(start)

            try:

                start_pos = revs.index(_start)

            except ValueError:

                pass

        if end is not None:

            _end = self._get_revision(end)

            try:

                end_pos = revs.index(_end)

            except ValueError:

                pass

        if None not in [start, end] and start_pos > end_pos:

            raise RepositoryError('start cannot be after end')

        if end_pos is not None:

            end_pos += 1

        revs = revs[start_pos:end_pos]

        if reverse:

            revs = reversed(revs)

        return CollectionGenerator(self, revs)

    def get_diff(self, rev1, rev2, path=None, ignore_whitespace=False,

                 context=3):

from rhodecode.lib.vcs.utils.hgcompat import (

    ui, nullid, match, patch, diffopts, clone, get_contact, pull,

    localrepository, RepoLookupError, Abort, RepoError, hex, scmutil, hg_url,

    httpbasicauthhandler, httpdigestauthhandler

)

from .changeset import MercurialChangeset

from .inmemory import MercurialInMemoryChangeset

from .workdir import MercurialWorkdir

log = logging.getLogger(__name__)

class MercurialRepository(BaseRepository):

    """

    Mercurial repository backend

    """

    DEFAULT_BRANCH_NAME = 'default'

    scm = 'hg'

    def __init__(self, repo_path, create=False, baseui=None, src_url=None,

                 update_after_clone=False):

        """

        Raises RepositoryError if repository could not be find at the given

        ``repo_path``.

        :param repo_path: local path of the repository

        :param create=False: if set to True, would try to create repository if

           it does not exist rather than raising exception

        :param baseui=None: user data

        :param src_url=None: would try to clone repository from given location

        :param update_after_clone=False: sets update of working copy after

          making a clone

        """

        if not isinstance(repo_path, str):

            raise VCSError('Mercurial backend requires repository path to '

                           'be instance of <str> got %s instead' %

                           type(repo_path))

        self.path = abspath(repo_path)

        self.baseui = baseui or ui.ui()

        # We've set path and ui, now we can set _repo itself

        self._repo = self._get_repo(create, src_url, update_after_clone)

    @property

    def _empty(self):

        """

        """

        # TODO: Following raises errors when using InMemoryChangeset...

        # return len(self._repo.changelog) == 0

        return len(self.revisions) == 0

    @LazyProperty

    def revisions(self):

        """

        Returns list of revisions' ids, in ascending order.  Being lazy

        attribute allows external tools to inject shas from cache.

        """

        return self._get_all_revisions()

    @LazyProperty

    def name(self):

        return os.path.basename(self.path)

    @LazyProperty

    def branches(self):

        return self._get_branches()

    @LazyProperty

    def allbranches(self):

        """

        List all branches, including closed branches.

        """

        return self._get_branches(closed=True)

    def _get_branches(self, closed=False):

        """

        Get's branches for this repository

        Returns only not closed branches by default

        :param closed: return also closed branches for mercurial

        """

        if self._empty:

            return {}

        def _branchtags(localrepo):

            """

            Patched version of mercurial branchtags to not return the closed

            branches

            :param localrepo: locarepository instance

            """

            bt = {}

from __future__ import with_statement

import datetime

from rhodecode.lib import vcs

from rhodecode.tests.vcs.base import BackendTestMixin

from rhodecode.tests.vcs.conf import SCM_TESTS

from rhodecode.lib.vcs.backends.base import BaseChangeset

from rhodecode.lib.vcs.nodes import (

    FileNode, AddedFileNodesGenerator,

    ChangedFileNodesGenerator, RemovedFileNodesGenerator

)

from rhodecode.lib.vcs.exceptions import (

    BranchDoesNotExistError, ChangesetDoesNotExistError,

)

from rhodecode.lib.vcs.utils.compat import unittest

class TestBaseChangeset(unittest.TestCase):

    def test_as_dict(self):

        changeset = BaseChangeset()

        changeset.id = 'ID'

        changeset.raw_id = 'RAW_ID'

        changeset.short_id = 'SHORT_ID'

        changeset.revision = 1009

        changeset.date = datetime.datetime(2011, 1, 30, 1, 45)

        changeset.message = 'Message of a commit'

        changeset.author = 'Joe Doe <joe.doe@example.com>'

        changeset.added = [FileNode('foo/bar/baz'), FileNode('foobar')]

        changeset.changed = []

        changeset.removed = []

        self.assertEqual(changeset.as_dict(), {

            'id': 'ID',

            'raw_id': 'RAW_ID',

            'short_id': 'SHORT_ID',

            'revision': 1009,

            'date': datetime.datetime(2011, 1, 30, 1, 45),

            'message': 'Message of a commit',

            'author': {

                'name': 'Joe Doe',

                'email': 'joe.doe@example.com',

            },

            'added': ['foo/bar/baz', 'foobar'],

            'changed': [],

            'removed': [],

        })

class ChangesetsWithCommitsTestCaseixin(BackendTestMixin):

    recreate_repo_per_test = True

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

        for x in xrange(5):

            yield {

                'message': 'Commit %d' % x,

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test_simple(self):

        tip = self.repo.get_changeset()

        self.assertEqual(tip.date, datetime.datetime(2010, 1, 3, 20))

    def test_get_changesets_is_ordered_by_date(self):

        changesets = list(self.repo.get_changesets())

        ordered_by_date = sorted(changesets,

            key=lambda cs: cs.date)

        self.assertItemsEqual(changesets, ordered_by_date)

    def test_get_changesets_respects_start(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(start=second_id))

        self.assertEqual(len(changesets), 4)

    def test_get_changesets_numerical_id_respects_start(self):

        second_id = 1

        changesets = list(self.repo.get_changesets(start=second_id))

        self.assertEqual(len(changesets), 4)

    def test_get_changesets_includes_start_changeset(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(start=second_id))

        self.assertEqual(changesets[0].raw_id, second_id)

    def test_get_changesets_respects_end(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(end=second_id))

        self.assertEqual(changesets[-1].raw_id, second_id)

        self.assertEqual(len(changesets), 2)

    def test_get_changesets_numerical_id_respects_end(self):

        second_id = 1

        changesets = list(self.repo.get_changesets(end=second_id))

        self.assertEqual(changesets.index(changesets[-1]), second_id)

        self.assertEqual(len(changesets), 2)

    def test_get_changesets_respects_both_start_and_end(self):

        second_id = self.repo.revisions[1]

        third_id = self.repo.revisions[2]

        changesets = list(self.repo.get_changesets(start=second_id,

            end=third_id))

        self.assertEqual(len(changesets), 2)

    def test_get_changesets_numerical_id_respects_both_start_and_end(self):

        changesets = list(self.repo.get_changesets(start=2, end=3))

        self.assertEqual(len(changesets), 2)

    def test_get_changesets_includes_end_changeset(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(end=second_id))

        self.assertEqual(changesets[-1].raw_id, second_id)

    def test_get_changesets_respects_start_date(self):

        start_date = datetime.datetime(2010, 2, 1)

        for cs in self.repo.get_changesets(start_date=start_date):

            self.assertGreaterEqual(cs.date, start_date)

    def test_get_changesets_respects_end_date(self):

        start_date = datetime.datetime(2010, 1, 1)

        end_date = datetime.datetime(2010, 2, 1)

        for cs in self.repo.get_changesets(start_date=start_date,

                                           end_date=end_date):

            self.assertGreaterEqual(cs.date, start_date)

            self.assertLessEqual(cs.date, end_date)

    def test_get_changesets_respects_start_date_and_end_date(self):

        end_date = datetime.datetime(2010, 2, 1)

        for cs in self.repo.get_changesets(end_date=end_date):

            self.assertLessEqual(cs.date, end_date)

    def test_get_changesets_respects_reverse(self):

        changesets_id_list = [cs.raw_id for cs in

            self.repo.get_changesets(reverse=True)]

        self.assertItemsEqual(changesets_id_list, reversed(self.repo.revisions))

    def test_get_filenodes_generator(self):

        tip = self.repo.get_changeset()

        filepaths = [node.path for node in tip.get_filenodes_generator()]

        self.assertItemsEqual(filepaths, ['file_%d.txt' % x for x in xrange(5)])

    def test_size(self):

        tip = self.repo.get_changeset()

        size = 5 * len('Foobar N') # Size of 5 files

        self.assertEqual(tip.size, size)

    def test_author(self):

        tip = self.repo.get_changeset()

        self.assertEqual(tip.author, u'Joe Doe <joe.doe@example.com>')

    def test_author_name(self):

        tip = self.repo.get_changeset()

        self.assertEqual(tip.author_name, u'Joe Doe')

    def test_author_email(self):

        tip = self.repo.get_changeset()