class BackendTestMixin(object):
    """
    This is a backend independent test case class which should be created
    with ``type`` method.

    It is required to set following attributes at subclass:

    - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)
    - ``repo_path``: path to the repository which would be created for set of
      tests
    - ``recreate_repo_per_test``: If set to ``False``, repo would NOT be created
      before every single test. Defaults to ``True``.
    """
    recreate_repo_per_test = True

    @classmethod
    def get_backend(cls):
        return vcs.get_backend(cls.backend_alias)

    @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

    @classmethod
    def setUpClass(cls):
        logging.basicConfig(level=logging.INFO)
        Backend = cls.get_backend()
        cls.backend_class = Backend
        cls.repo_path = get_new_dir(str(time.time()))
        cls.repo = Backend(cls.repo_path, create=True)
        cls.imc = cls.repo.in_memory_changeset
        cls.default_branch = cls.repo.DEFAULT_BRANCH_NAME

        for commit in cls._get_commits():
            for node in commit.get('added', []):
                cls.imc.add(FileNode(node.path, content=node.content))
            for node in commit.get('changed', []):
                cls.imc.change(FileNode(node.path, content=node.content))
            for node in commit.get('removed', []):
                cls.imc.remove(FileNode(node.path))

            cls.tip = cls.imc.commit(message=unicode(commit['message']),
                                     author=unicode(commit['author']),
                                     date=commit['date'])

    @classmethod
    def tearDownClass(cls):
        if not getattr(cls, 'recreate_repo_per_test', False) and \
            'VCS_REMOVE_TEST_DIRS' in os.environ:
            shutil.rmtree(cls.repo_path)

    def setUp(self):
        if getattr(self, 'recreate_repo_per_test', False):
            self.__class__.setUpClass()

    def tearDown(self):
        if getattr(self, 'recreate_repo_per_test', False) and \
            'VCS_REMOVE_TEST_DIRS' in os.environ:
            shutil.rmtree(self.repo_path) SCM_TESTS = ['hg', 'git'] class BaseChangeset(object):
    """
    Each backend should implement it's changeset representation.

    **Attributes**

        ``repository``
            repository object within which changeset exists

        ``id``
            may be ``raw_id`` or i.e. for mercurial's tip just ``tip``

        ``raw_id``
            raw changeset representation (i.e. full 40 length sha for git
            backend)

        ``short_id``
            shortened (if apply) version of ``raw_id``; it would be simple
            shortcut for ``raw_id[:12]`` for git/mercurial backends or same
            as ``raw_id`` for subversion

        ``revision``
            revision number as integer

        ``files``
            list of ``FileNode`` (``Node`` with NodeKind.FILE) objects

        ``dirs``
            list of ``DirNode`` (``Node`` with NodeKind.DIR) objects

        ``nodes``
            combined list of ``Node`` objects

        ``author``
            author of the changeset, as unicode

        ``message``
            message of the changeset, as unicode

        ``parents``
            list of parent changesets

        ``last``
            ``True`` if this is last changeset in repository, ``False``
            otherwise; trying to access this attribute while there is no
            changesets would raise ``EmptyRepositoryError``
    """
    def __str__(self):
        return '<%s at %s:%s>' % (self.__class__.__name__, self.revision,
            self.short_id)

    def __repr__(self):
        return self.__str__()

    def __unicode__(self):
        return u'%s:%s' % (self.revision, self.short_id)

    def __eq__(self, other):
        return self.raw_id == other.raw_id

    @LazyProperty
    def last(self):
        if self.repository is None:
            raise ChangesetError("Cannot check if it's most recent revision")
        return self.raw_id == self.repository.revisions[-1]

    @LazyProperty
    def parents(self):
        """
        Returns list of parents changesets.
        """
        raise NotImplementedError

    @LazyProperty
    def children(self):
        """
        Returns list of children changesets.
        """
        raise NotImplementedError

    @LazyProperty
    def id(self):
        """
        Returns string identifying this changeset.
        """
        raise NotImplementedError

    @LazyProperty
    def raw_id(self):
        """
        Returns raw string identifying this changeset.
        """
        raise NotImplementedError

    @LazyProperty
    def short_id(self):
        """
        Returns shortened version of ``raw_id`` attribute, as string,
        identifying this changeset, useful for web representation.
        """
        raise NotImplementedError

    @LazyProperty
    def revision(self):
        """
        Returns integer identifying this changeset.

        """
        raise NotImplementedError

    @LazyProperty
    def committer(self):
        """
        Returns Committer for given commit
        """

        raise NotImplementedError

    @LazyProperty
    def committer_name(self):
        """
        Returns Author name for given commit
        """

        return author_name(self.committer)

    @LazyProperty
    def committer_email(self):
        """
        Returns Author email address for given commit
        """

        return author_email(self.committer)

    @LazyProperty
    def author(self):
        """
        Returns Author for given commit
        """

        raise NotImplementedError

    @LazyProperty
    def author_name(self):
        """
        Returns Author name for given commit
        """

        return author_name(self.author)

    @LazyProperty
    def author_email(self):
        """
        Returns Author email address for given commit
        """

        return author_email(self.author)

    def get_file_mode(self, path):
        """
        Returns stat mode of the file at the given ``path``.
        """
        raise NotImplementedError

    def get_file_content(self, path):
        """
        Returns content of the file at the given ``path``.
        """
        raise NotImplementedError

    def get_file_size(self, path):
        """
        Returns size of the file at the given ``path``.
        """
        raise NotImplementedError

    def get_file_changeset(self, path):
        """
        Returns last commit of the file at the given ``path``.
        """
        raise NotImplementedError

    def get_file_history(self, path):
        """
        Returns history of file as reversed list of ``Changeset`` objects for
        which file at given ``path`` has been modified.
        """
        raise NotImplementedError

    def get_nodes(self, path):
        """
        Returns combined ``DirNode`` and ``FileNode`` objects list representing
        state of changeset at the given ``path``.

        :raises ``ChangesetError``: if node at the given ``path`` is not
          instance of ``DirNode``
        """
        raise NotImplementedError

    def get_node(self, path):
        """
        Returns ``Node`` object from the given ``path``.

        :raises ``NodeDoesNotExistError``: if there is no node at the given
          ``path``
        """
        raise NotImplementedError

    def fill_archive(self, stream=None, kind='tgz', prefix=None):
        """
        Fills up given stream.

        :param stream: file like object.
        :param kind: one of following: ``zip``, ``tar``, ``tgz``
            or ``tbz2``. Default: ``tgz``.
        :param prefix: name of root directory in archive.
            Default is repository name and changeset's raw_id joined with dash.

            repo-tip.<kind>
        """

        raise NotImplementedError

    def get_chunked_archive(self, **kwargs):
        """
        Returns iterable archive. Tiny wrapper around ``fill_archive`` method.

        :param chunk_size: extra parameter which controls size of returned
            chunks. Default:8k.
        """

        chunk_size = kwargs.pop('chunk_size', 8192)
        stream = kwargs.get('stream')
        self.fill_archive(**kwargs)
        while True:
            data = stream.read(chunk_size)
            if not data:
                break
            yield data

    @LazyProperty
    def root(self):
        """
        Returns ``RootNode`` object for this changeset.
        """
        return self.get_node('')

    def next(self, branch=None):
        """
        Returns next changeset from current, if branch is gives it will return
        next changeset belonging to this branch

        :param branch: show changesets within the given named branch
        """
        raise NotImplementedError

    def prev(self, branch=None):
        """
        Returns previous changeset from current, if branch is gives it will
        return previous changeset belonging to this branch

        :param branch: show changesets within the given named branch
        """
        raise NotImplementedError

    @LazyProperty
    def added(self):
        """
        Returns list of added ``FileNode`` objects.
        """
        raise NotImplementedError

    @LazyProperty
    def changed(self):
        """
        Returns list of modified ``FileNode`` objects.
        """
        raise NotImplementedError

    @LazyProperty
    def removed(self):
        """
        Returns list of removed ``FileNode`` objects.
        """
        raise NotImplementedError

    @LazyProperty
    def size(self):
        """
        Returns total number of bytes from contents of all filenodes.
        """
        return sum((node.size for node in self.get_filenodes_generator()))

    def walk(self, topurl=''):
        """
        Similar to os.walk method. Insted of filesystem it walks through
        changeset starting at given ``topurl``.  Returns generator of tuples
        (topnode, dirnodes, filenodes).
        """
        topnode = self.get_node(topurl)
        yield (topnode, topnode.dirs, topnode.files)
        for dirnode in topnode.dirs:
            for tup in self.walk(dirnode.path):
                yield tup

    def get_filenodes_generator(self):
        """
        Returns generator that yields *all* file nodes.
        """
        for topnode, dirs, files in self.walk():
            for node in files:
                yield node

    def as_dict(self):
        """
        Returns dictionary with changeset's attributes and their values.
        """
        data = get_dict_for_attrs(self, ['id', 'raw_id', 'short_id',
            'revision', 'date', 'message'])
        data['author'] = {'name': self.author_name, 'email': self.author_email}
        data['added'] = [node.path for node in self.added]
        data['changed'] = [node.path for node in self.changed]
        data['removed'] = [node.path for node in self.removed]
        return data class FileNode(Node):
    """
    Class representing file nodes.

    :attribute: path: path to the node, relative to repostiory's root
    :attribute: content: if given arbitrary sets content of the file
    :attribute: changeset: if given, first time content is accessed, callback
    :attribute: mode: octal stat mode for a node. Default is 0100644.
    """

    def __init__(self, path, content=None, changeset=None, mode=None):
        """
        Only one of ``content`` and ``changeset`` may be given. Passing both
        would raise ``NodeError`` exception.

        :param path: relative path to the node
        :param content: content may be passed to constructor
        :param changeset: if given, will use it to lazily fetch content
        :param mode: octal representation of ST_MODE (i.e. 0100644)
        """

        if content and changeset:
            raise NodeError("Cannot use both content and changeset")
        super(FileNode, self).__init__(path, kind=NodeKind.FILE)
        self.changeset = changeset
        self._content = content
        self._mode = mode or 0100644

    @LazyProperty
    def mode(self):
        """
        Returns lazily mode of the FileNode. If ``changeset`` is not set, would
        use value given at initialization or 0100644 (default).
        """
        if self.changeset:
            mode = self.changeset.get_file_mode(self.path)
        else:
            mode = self._mode
        return mode

    def _get_content(self):
        if self.changeset:
            content = self.changeset.get_file_content(self.path)
        else:
            content = self._content
        return content

    @property
    def content(self):
        """
        Returns lazily content of the FileNode. If possible, would try to
        decode content from UTF-8.
        """
        content = self._get_content()

        if bool(content and '\0' in content):
            return content
        return safe_unicode(content)

    @LazyProperty
    def size(self):
        if self.changeset:
            return self.changeset.get_file_size(self.path)
        raise NodeError("Cannot retrieve size of the file without related "
            "changeset attribute")

    @LazyProperty
    def message(self):
        if self.changeset:
            return self.last_changeset.message
        raise NodeError("Cannot retrieve message of the file without related "
            "changeset attribute")

    @LazyProperty
    def last_changeset(self):
        if self.changeset:
            return self.changeset.get_file_changeset(self.path)
        raise NodeError("Cannot retrieve last changeset of the file without "
            "related changeset attribute")

    def get_mimetype(self):
        """
        Mimetype is calculated based on the file's content. If ``_mimetype``
        attribute is available, it will be returned (backends which store
        mimetypes or can easily recognize them, should set this private
        attribute to indicate that type should *NOT* be calculated).
        """
        if hasattr(self, '_mimetype'):
            if (isinstance(self._mimetype, (tuple, list,)) and
                len(self._mimetype) == 2):
                return self._mimetype
            else:
                raise NodeError('given _mimetype attribute must be an 2 '
                               'element list or tuple')

        mtype, encoding = mimetypes.guess_type(self.name)

        if mtype is None:
            if self.is_binary:
                mtype = 'application/octet-stream'
                encoding = None
            else:
                mtype = 'text/plain'
                encoding = None
        return mtype, encoding

    @LazyProperty
    def mimetype(self):
        """
        Wrapper around full mimetype info. It returns only type of fetched
        mimetype without the encoding part. use get_mimetype function to fetch
        full set of (type,encoding)
        """
        return self.get_mimetype()[0]

    @LazyProperty
    def mimetype_main(self):
        return self.mimetype.split('/')[0]

    @LazyProperty
    def lexer(self):
        """
        Returns pygment's lexer class. Would try to guess lexer taking file's
        content, name and mimetype.
        """

        try:
            lexer = lexers.guess_lexer_for_filename(self.name, self.content, stripnl=False)
        except lexers.ClassNotFound:
            lexer = lexers.TextLexer(stripnl=False)
        # returns first alias
        return lexer

    @LazyProperty
    def lexer_alias(self):
        """
        Returns first alias of the lexer guessed for this file.
        """
        return self.lexer.aliases[0]

    @LazyProperty
    def history(self):
        """
        Returns a list of changeset for this file in which the file was changed
        """
        if self.changeset is None:
            raise NodeError('Unable to get changeset for this FileNode')
        return self.changeset.get_file_history(self.path)

    @LazyProperty
    def annotate(self):
        """
        Returns a list of three element tuples with lineno,changeset and line
        """
        if self.changeset is None:
            raise NodeError('Unable to get changeset for this FileNode')
        return self.changeset.get_file_annotate(self.path)

    @LazyProperty
    def state(self):
        if not self.changeset:
            raise NodeError("Cannot check state of the node if it's not "
                "linked with changeset")
        elif self.path in (node.path for node in self.changeset.added):
            return NodeState.ADDED
        elif self.path in (node.path for node in self.changeset.changed):
            return NodeState.CHANGED
        else:
            return NodeState.NOT_CHANGED

    @property
    def is_binary(self):
        """
        Returns True if file has binary content.
        """
        _bin = '\0' in self._get_content()
        return _bin

    @LazyProperty
    def extension(self):
        """Returns filenode extension"""
        return self.name.split('.')[-1]

    def is_executable(self):
        """
        Returns ``True`` if file has executable flag turned on.
        """
        return bool(self.mode & stat.S_IXUSR)

    def __repr__(self):
        return '<%s %r @ %s>' % (self.__class__.__name__, self.path,
                                 getattr(self.changeset, 'short_id', '')) class AddedFileNodesGenerator(NodeGeneratorBase):
    """
    Class holding Added files for current changeset
    """
    pass class ChangedFileNodesGenerator(NodeGeneratorBase):
    """
    Class holding Changed files for current changeset
    """
    pass class RemovedFileNodesGenerator(NodeGeneratorBase):
    """
    Class holding removed files for current changeset
    """
    def __iter__(self):
        for p in self.current_paths:
            yield RemovedFileNode(path=p)

    def __getslice__(self, i, j):
        for p in self.current_paths[i:j]:
            yield RemovedFileNode(path=p) class BranchDoesNotExistError(RepositoryError):
    pass class ChangesetDoesNotExistError(ChangesetError):
    pass class RepositoryError(VCSError):
    pass class deque(object):
class _Condition(_Verbose):
class _Event(_Verbose):
    def __init__(self, iterable=(), maxlen= -1):
    def append(self, x):
    def appendleft(self, x):
    def pop(self):
    def popleft(self):
    def clear(self):
    def extend(self, iterable):
    def extendleft(self, iterable):
    def rotate(self, n=1):
    def __getitem__(self, i):
    def __setitem__(self, i, value):
    def __delitem__(self, i):
    def __len__(self):
    def __cmp__(self, other):
    def __repr__(self, _track=[]):
    def __getstate__(self):
    def __setstate__(self, s):
    def __hash__(self):
    def __copy__(self):
    def __deepcopy__(self, memo={}):
def Condition(*args, **kwargs):
    def __init__(self, lock=None, verbose=None):
    def __enter__(self):
    def __exit__(self, *args):
    def __repr__(self):
    def _release_save(self):
    def _acquire_restore(self, x):
    def _is_owned(self):
    def wait(self, timeout=None):
    def notify(self, n=1):
    def notifyAll(self):
def Event(*args, **kwargs):
    def __init__(self, verbose=None):
    def isSet(self):
    def set(self):
    def clear(self):
    def wait(self, timeout=None):
import datetime
import vcs
from vcs.tests.base import BackendTestMixin
from vcs.tests.conf import SCM_TESTS
from vcs.backends.base import BaseChangeset
from vcs.nodes import (
    FileNode, AddedFileNodesGenerator,
    ChangedFileNodesGenerator, RemovedFileNodesGenerator
)
from vcs.exceptions import (
    BranchDoesNotExistError, ChangesetDoesNotExistError,
    RepositoryError
)
from vcs.utils.compat import unittest
        end_date = datetime.datetime(2010, 1, 3)
        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_and_branch_name(self):
        start_date = datetime.datetime(2010, 1, 1)
        end_date = datetime.datetime(2010, 1, 3)
        for cs in self.repo.get_changesets(start_date=start_date,
                                           end_date=end_date,
                                           branch_name=self.repo.DEFAULT_BRANCH_NAME):
            self.assertGreaterEqual(cs.date, start_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))

    def test_get_changesets_numerical_id_reversed(self):
        with self.assertRaises(RepositoryError):
            [x for x in self.repo.get_changesets(start=3, end=2)]

    def test_get_changesets_numerical_id_respects_both_start_and_end_last(self):
        with self.assertRaises(RepositoryError):
            last = len(self.repo.revisions)
            list(self.repo.get_changesets(start=last-1, end=last-2))

    def test_get_changesets_numerical_id_last_zero_error(self):
        with self.assertRaises(RepositoryError):
            last = len(self.repo.revisions)
            list(self.repo.get_changesets(start=last-1, end=0))


class ChangesetsChangesTestCaseMixin(BackendTestMixin):
    recreate_repo_per_test = False

    @classmethod
    def _get_commits(cls):
        return [
            {
                'message': u'Initial',
                'author': u'Joe Doe <joe.doe@example.com>',
                'date': datetime.datetime(2010, 1, 1, 20),
                'added': [
                    FileNode('foo/bar', content='foo'),
                    FileNode('foobar', content='foo'),
                    FileNode('qwe', content='foo'),
                ],
            },
            {
                'message': u'Massive changes',
                'author': u'Joe Doe <joe.doe@example.com>',
                'date': datetime.datetime(2010, 1, 1, 22),
                'added': [FileNode('fallout', content='War never changes')],
                'changed': [
                    FileNode('foo/bar', content='baz'),
                    FileNode('foobar', content='baz'),
                ],
                'removed': [FileNode('qwe')],
            },
        ]

    def test_initial_commit(self):
        changeset = self.repo.get_changeset(0)
        self.assertItemsEqual(changeset.added, [
            changeset.get_node('foo/bar'),
            changeset.get_node('foobar'),
            changeset.get_node('qwe'),
        ])
        self.assertItemsEqual(changeset.changed, [])
        self.assertItemsEqual(changeset.removed, [])

    def test_head_added(self):
        changeset = self.repo.get_changeset()
        self.assertTrue(isinstance(changeset.added, AddedFileNodesGenerator))
        self.assertItemsEqual(changeset.added, [
            changeset.get_node('fallout'),
        ])

        self.assertTrue(isinstance(changeset.changed, ChangedFileNodesGenerator))