# -*- coding: utf-8 -*-

"""

    vcs

    ~~~

    Various version Control System (vcs) management abstraction layer for

    Python.

    :created_on: Apr 8, 2010

    :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.

"""

__version__ = '.'.join((str(each) for each in VERSION[:4]))

__all__ = [

    'get_version', 'get_repo', 'get_backend',

    'VCSError', 'RepositoryError', 'ChangesetError'

]

import sys

from rhodecode.lib.vcs.backends import get_repo, get_backend

from rhodecode.lib.vcs.exceptions import VCSError, RepositoryError, ChangesetError

def get_version():

    """

    Returns shorter version (digit parts only) as string.

    """

    return '.'.join((str(each) for each in VERSION[:3]))

def main(argv=None):

    if argv is None:

        argv = sys.argv

    from rhodecode.lib.vcs.cli import ExecutionManager

    manager = ExecutionManager(argv)

    manager.execute()

    return 0

if __name__ == '__main__':

    sys.exit(main(sys.argv))

import re

from itertools import chain

from dulwich import objects

from subprocess import Popen, PIPE

from rhodecode.lib.vcs.conf import settings

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

from rhodecode.lib.vcs.exceptions import (

    RepositoryError, ChangesetError, NodeDoesNotExistError, VCSError,

    ChangesetDoesNotExistError, ImproperArchiveTypeError

)

from rhodecode.lib.vcs.nodes import (

    FileNode, DirNode, NodeKind, RootNode, RemovedFileNode, SubModuleNode,

    ChangedFileNodesGenerator, AddedFileNodesGenerator, RemovedFileNodesGenerator

)

from rhodecode.lib.vcs.utils import (

    safe_unicode, safe_str, safe_int, date_fromtimestamp

)

from rhodecode.lib.vcs.utils.lazy import LazyProperty

class GitChangeset(BaseChangeset):

    """

    Represents state of the repository at single revision.

    """

    def __init__(self, repository, revision):

        self._stat_modes = {}

        self.repository = repository

        try:

            if isinstance(commit, objects.Tag):

                revision = commit.object[1]

                commit = self.repository._repo.get_object(commit.object[1])

        except KeyError:

            raise RepositoryError("Cannot get object with id %s" % revision)

        self.raw_id = revision

        self.id = self.raw_id

        self.short_id = self.raw_id[:12]

        self._commit = commit

        self._tree_id = commit.tree

        self._committer_property = 'committer'

        self._author_property = 'author'

        self._date_property = 'commit_time'

        self._date_tz_property = 'commit_timezone'

        self.revision = repository.revisions.index(revision)

        self.nodes = {}

        self._paths = {}

    @LazyProperty

    def committer(self):

        return safe_unicode(getattr(self._commit, self._committer_property))

    @LazyProperty

    def author(self):

        return safe_unicode(getattr(self._commit, self._author_property))

    @LazyProperty

    def date(self):

        return date_fromtimestamp(getattr(self._commit, self._date_property),

                                  getattr(self._commit, self._date_tz_property))

    @LazyProperty

    def _timestamp(self):

        return getattr(self._commit, self._date_property)

    @LazyProperty

    def status(self):

        """

        Returns modified, added, removed, deleted files for current changeset

        """

        return self.changed, self.added, self.removed

    @LazyProperty

    def tags(self):

        _tags = []

        for tname, tsha in self.repository.tags.iteritems():

            if tsha == self.raw_id:

                _tags.append(tname)

        return _tags

    @LazyProperty

    def branch(self):

        heads = self.repository._heads(reverse=False)

        ref = heads.get(self.raw_id)

        if ref:

            return safe_unicode(ref)

    def _fix_path(self, path):

        """

        Paths are stored without trailing slash so we need to get rid off it if

        needed.

        """

        if path.endswith('/'):

            path = path.rstrip('/')

        return path

# -*- coding: utf-8 -*-

"""

    vcs.backends.git.repository

    ~~~~~~~~~~~~~~~~~~~~~~~~~~~

    Git repository implementation.

    :created_on: Apr 8, 2010

    :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.

"""

import os

import re

import time

import urllib

import urllib2

import logging

import posixpath

import string

from dulwich.objects import Tag

from dulwich.repo import Repo, NotGitRepository

from rhodecode.lib.vcs import subprocessio

from rhodecode.lib.vcs.backends.base import BaseRepository, CollectionGenerator

from rhodecode.lib.vcs.conf import settings

from rhodecode.lib.vcs.exceptions import (

    BranchDoesNotExistError, ChangesetDoesNotExistError, EmptyRepositoryError,

    RepositoryError, TagAlreadyExistError, TagDoesNotExistError

)

from rhodecode.lib.vcs.utils import safe_unicode, makedate, date_fromtimestamp

from rhodecode.lib.vcs.utils.lazy import LazyProperty

from rhodecode.lib.vcs.utils.ordered_dict import OrderedDict

from rhodecode.lib.vcs.utils.paths import abspath, get_user_home

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

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 _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 src_url and not create:

            raise RepositoryError("Create should be set to True if src_url is "

                                  "given (clone operation creates repository)")

        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):

            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))

            # 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]

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

                return _tags_shas[_tags_shas.index(revision)]

                raise ChangesetDoesNotExistError("Revision %s does not exist "

                    "for this repository" % (revision))

        # Ensure we return full id

            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))

        return url

    def get_hook_location(self):

        """

        returns absolute path to location where hooks are stored

        """

        loc = os.path.join(self.path, 'hooks')

        if not self.bare:

            loc = os.path.join(self.path, '.git', 'hooks')

        return loc

    @LazyProperty

    def name(self):

        return os.path.basename(self.path)

    @LazyProperty

    def last_change(self):

        """

        Returns last change made on this repository as datetime object

        """

        return date_fromtimestamp(self._get_mtime(), makedate()[1])

    def _get_mtime(self):

        try:

            return time.mktime(self.get_changeset().date.timetuple())

        except RepositoryError:

            idx_loc = '' if self.bare else '.git'

            # fallback to filesystem

            in_path = os.path.join(self.path, idx_loc, "index")

            he_path = os.path.join(self.path, idx_loc, "HEAD")

            if os.path.exists(in_path):

    def get_hook_location(self):

        """

        returns absolute path to location where hooks are stored

        """

        return os.path.join(self.path, '.hg', '.hgrc')

    def get_changeset(self, revision=None):

        """

        Returns ``MercurialChangeset`` object representing repository's

        changeset at the given ``revision``.

        """

        revision = self._get_revision(revision)

        changeset = MercurialChangeset(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 ``MercurialChangeset`` objects from start to end

        (both are inclusive)

        :param start: None, str, int or mercurial lookup format

        :param end:  None, str, int or mercurial lookup format

        :param start_date:

        :param end_date:

        :param branch_name:

        :param reversed: return changesets in reversed order

        """

        start_raw_id = self._get_revision(start)

        start_pos = self.revisions.index(start_raw_id) if start else None

        end_raw_id = self._get_revision(end)

        end_pos = self.revisions.index(end_raw_id) if end else None

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

            raise RepositoryError("Start revision '%s' cannot be "

                                  "after end revision '%s'" % (start, end))

        if branch_name and branch_name not in self.allbranches.keys():

            raise BranchDoesNotExistError('Branch %s not found in'

                                  ' this repository' % branch_name)

        if end_pos is not None:

            end_pos += 1

        #filter branches

        filter_ = []

        if branch_name:

            filter_.append('branch("%s")' % (branch_name))

            filter_.append('date(">%s")' % start_date)

            filter_.append('date("<%s")' % end_date)

        if filter_:

            revisions = scmutil.revrange(self._repo, filter_)

        else:

            revisions = self.revisions

        revs = revisions[start_pos:end_pos]

        if reverse:

            revs = reversed(revs)

        return CollectionGenerator(self, revs)

    def pull(self, url):

        """

        Tries to pull changes from external location.

        """

        url = self._get_url(url)

        try:

            pull(self.baseui, self._repo, url)

        except Abort, err:

            # Propagate error but with vcs's type

            raise RepositoryError(str(err))

    @LazyProperty

    def workdir(self):

        """

        Returns ``Workdir`` instance for this repository.

        """

        return MercurialWorkdir(self)

    def get_config_value(self, section, name=None, config_file=None):

        """

        Returns configuration value for a given [``section``] and ``name``.

        :param section: Section we want to retrieve value from

        :param name: Name of configuration we want to retrieve

        :param config_file: A path to file which should be used to retrieve

          configuration from (might also be a list of file paths)

        """

        if config_file is None:

            config_file = []

        elif isinstance(config_file, basestring):

            config_file = [config_file]

        config = self._repo.ui

        for path in config_file:

            config.readconfig(path)

        return config.config(section, name)

        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):

        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()

        self.assertEqual(tip.author_email, u'joe.doe@example.com')

    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_start(self):

        with self.assertRaises(ChangesetDoesNotExistError):

            list(self.repo.get_changesets(start='foobar'))

    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_end(self):

        with self.assertRaises(ChangesetDoesNotExistError):

            list(self.repo.get_changesets(end='foobar'))

    def test_get_changesets_raise_branchdoesnotexist_for_wrong_branch_name(self):

        with self.assertRaises(BranchDoesNotExistError):

            list(self.repo.get_changesets(branch_name='foobar'))

    def test_get_changesets_raise_repositoryerror_for_wrong_start_end(self):

        start = self.repo.revisions[-1]

        end = self.repo.revisions[0]

        with self.assertRaises(RepositoryError):

            list(self.repo.get_changesets(start=start, end=end))

from rhodecode.lib.vcs.nodes import FileNode

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

from rhodecode.tests.vcs.base import BackendTestMixin

from rhodecode.tests.vcs.conf import SCM_TESTS

class WorkdirTestCaseMixin(BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        commits = [

            {

                'message': u'Initial commit',

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

                'date': datetime.datetime(2010, 1, 1, 20),

                'added': [

                    FileNode('foobar', content='Foobar'),

                    FileNode('foobar2', content='Foobar II'),

                    FileNode('foo/bar/baz', content='baz here!'),

                ],

            },

            {

                'message': u'Changes...',

                'author': u'Jane Doe <jane.doe@example.com>',

                'date': datetime.datetime(2010, 1, 1, 21),

                'added': [

                    FileNode('some/new.txt', content='news...'),

                ],

                'changed': [

                    FileNode('foobar', 'Foobar I'),

                ],

                'removed': [],

            },

        ]

        return commits

    def test_get_branch_for_default_branch(self):

        self.assertEqual(self.repo.workdir.get_branch(),

            self.repo.DEFAULT_BRANCH_NAME)

    def test_get_branch_after_adding_one(self):

        self.imc.add(FileNode('docs/index.txt',

            content='Documentation\n'))

        self.imc.commit(

            message=u'New branch: foobar',

            author=u'joe',

            branch='foobar',

        )

    def test_get_changeset(self):

        self.imc.add(FileNode('docs/index.txt',

            content='Documentation\n'))

        head = self.imc.commit(

            message=u'New branch: foobar',

            author=u'joe',

            branch='foobar',

        )

        self.assertEqual(self.repo.workdir.get_changeset(), head)

    def test_checkout_branch(self):

        from rhodecode.lib.vcs.exceptions import BranchDoesNotExistError

        # first, 'foobranch' does not exist.

        self.assertRaises(BranchDoesNotExistError, self.repo.workdir.checkout_branch,

                          branch='foobranch')

        # create new branch 'foobranch'.

        self.imc.add(FileNode('file1', content='blah'))

        self.imc.commit(message=u'asd', author=u'john', branch='foobranch')

        # go back to the default branch

        self.repo.workdir.checkout_branch()

        self.assertEqual(self.repo.workdir.get_branch(), self.backend_class.DEFAULT_BRANCH_NAME)

        # checkout 'foobranch'

        self.repo.workdir.checkout_branch('foobranch')

        self.assertEqual(self.repo.workdir.get_branch(), 'foobranch')

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s branch test' % alias).title().split())

    bases = (WorkdirTestCaseMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

if __name__ == '__main__':

    unittest.main()