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

    def _get_id_for_path(self, path):

        # FIXME: Please, spare a couple of minutes and make those codes cleaner;

        if not path in self._paths:

            path = path.strip('/')

            # set root tree

            tree = self.repository._repo[self._tree_id]

            if path == '':

                self._paths[''] = tree.id

                return tree.id

            splitted = path.split('/')

            dirs, name = splitted[:-1], splitted[-1]

            curdir = ''

                # cache all items from the given traversed tree

                for item, stat, id in tree.iteritems():

                    if curdir:

                        name = '/'.join((curdir, item))

                    else:

                        name = item

                    self._paths[name] = id

                    self._stat_modes[name] = stat

            if not path in self._paths:

                raise NodeDoesNotExistError("There is no file nor directory "

                    "at the given path '%s' at revision %s"

        return self._paths[path]

    def _get_kind(self, path):

        obj = self.repository._repo[self._get_id_for_path(path)]

        if isinstance(obj, objects.Blob):

            return NodeKind.FILE

        elif isinstance(obj, objects.Tree):

            return NodeKind.DIR

    def _get_filectx(self, path):

        path = self._fix_path(path)

        if self._get_kind(path) != NodeKind.FILE:

    def diff(self, ignore_whitespace=True, context=3):

        rev1 = self.parents[0] if self.parents else self.repository.EMPTY_CHANGESET

        rev2 = self

        return ''.join(self.repository.get_diff(rev1, rev2,

                                    ignore_whitespace=ignore_whitespace,

                                    context=context))

    def get_file_mode(self, path):

        """

        Returns stat mode of the file at the given ``path``.

        """

        # ensure path is traversed

        self._get_id_for_path(path)

        return self._stat_modes[path]

    def get_file_content(self, path):

        """

        Returns content of the file at given ``path``.

        """

        id = self._get_id_for_path(path)

        blob = self.repository._repo[id]

        return blob.as_pretty_string()

    def get_file_size(self, path):

        repo = self.repository._repo

        object_store = repo.object_store

        ENCODING = "UTF-8"

        DIRMOD = 040000

        # Create tree and populates it with blobs

        commit_tree = self.parents[0] and repo[self.parents[0]._commit.tree] or\

            objects.Tree()

        for node in self.added + self.changed:

            # Compute subdirs if needed

            dirpath, nodename = posixpath.split(node.path)

            parent = commit_tree

            ancestors = [('', parent)]

            # Tries to dig for the deepest existing tree

            while dirnames:

                curdir = dirnames.pop(0)

                try:

                    dir_id = parent[curdir][1]

                except KeyError:

                    # put curdir back into dirnames and stops

                    dirnames.insert(0, curdir)

                    break

    :created_on: Apr 8, 2010

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

"""

import os

import stat

import posixpath

import mimetypes

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

from rhodecode.lib.vcs.exceptions import NodeError, RemovedFileNodeError

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

class NodeKind:

    SUBMODULE = -1

    DIR = 1

    FILE = 2

class NodeState:

    ADDED = u'added'

    CHANGED = u'changed'

    NOT_CHANGED = u'not changed'

    Simplest class representing file or directory on repository.  SCM backends

    should use ``FileNode`` and ``DirNode`` subclasses rather than ``Node``

    directly.

    Node's ``path`` cannot start with slash as we operate on *relative* paths

    only. Moreover, every single node is identified by the ``path`` attribute,

    so it cannot end with slash, too. Otherwise, path could lead to mistakes.

    """

    def __init__(self, path, kind):

        if path.startswith('/'):

            raise NodeError("Cannot initialize Node objects with slash at "

        if path == '' and kind != NodeKind.DIR:

            raise NodeError("Only DirNode and its subclasses may be "

                            "initialized with empty path")

        self.kind = kind

        #self.dirs, self.files = [], []

        if self.is_root() and not self.is_dir():

            raise NodeError("Root node cannot be FILE kind")

    @LazyProperty

    def parent(self):

        parent_path = self.get_parent_path()

        if parent_path:

        :returns: new commited changeset

        """

        user = self._get_user(user)

        scm_instance = repo.scm_instance_no_cache()

        processed_nodes = []

        for f_path in nodes:

            if f_path.startswith('/') or f_path.startswith('.') or '../' in f_path:

                raise NonRelativePathError('%s is not an relative path' % f_path)

            if f_path:

                f_path = os.path.normpath(f_path)

            f_path = safe_str(f_path)

            # decoding here will force that we have proper encoded values

            # in any other case this will throw exceptions and deny commit

            if isinstance(content, (basestring,)):

                content = safe_str(content)

            elif isinstance(content, (file, cStringIO.OutputType,)):

                content = content.read()

            else:

                raise Exception('Content is of unrecognized type %s' % (

                    type(content)

                ))

            processed_nodes.append((f_path, content))

from __future__ import with_statement

import time

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

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

        self.assertItemsEqual(changeset.changed, [

            changeset.get_node('foo/bar'),

            changeset.get_node('foobar'),

        ])

        self.assertTrue(isinstance(changeset.removed, RemovedFileNodesGenerator))

        self.assertEqual(len(changeset.removed), 1)

        self.assertEqual(list(changeset.removed)[0].path, 'qwe')

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    # tests with additional commits

    cls_name = ''.join(('%s changesets with commits test' % alias).title().split())

    bases = (ChangesetsWithCommitsTestCaseixin, unittest.TestCase)

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

    # tests without additional commits

"""

Tests so called "in memory changesets" commit API of vcs.

"""

from __future__ import with_statement

import time

import datetime

from rhodecode.lib import vcs

from rhodecode.tests.vcs.conf import SCM_TESTS, get_new_dir

from rhodecode.lib.vcs.exceptions import EmptyRepositoryError

from rhodecode.lib.vcs.exceptions import NodeAlreadyAddedError

from rhodecode.lib.vcs.exceptions import NodeAlreadyExistsError

from rhodecode.lib.vcs.exceptions import NodeAlreadyRemovedError

from rhodecode.lib.vcs.exceptions import NodeAlreadyChangedError

from rhodecode.lib.vcs.exceptions import NodeDoesNotExistError

from rhodecode.lib.vcs.exceptions import NodeNotChangedError

from rhodecode.lib.vcs.nodes import DirNode

from rhodecode.lib.vcs.nodes import FileNode

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

class InMemoryChangesetTestMixin(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

            FileNode('foo/baz.txt', content='foo'),

            FileNode('foobar/foobaz/file', content='foo'),

            FileNode('foobar/barbaz', content='foo'),

        ]

        self.imc.add(*to_add)

        changeset = self.imc.commit(u'Another', u'joe.doe@example.com')

        self.assertEqual(changeset.get_node('foo/bar/foobaz/bar').content, 'foo')

        self.assertEqual(changeset.get_node('foo/bar/another/bar').content, 'foo')

        self.assertEqual(changeset.get_node('foo/baz.txt').content, 'foo')

        self.assertEqual(changeset.get_node('foobar/foobaz/file').content, 'foo')

        self.assertEqual(changeset.get_node('foobar/barbaz').content, 'foo')

    def test_add_raise_already_added(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.assertRaises(NodeAlreadyAddedError, self.imc.add, node)

    def test_check_integrity_raise_already_exist(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.imc.commit(message=u'Added foobar', author=unicode(self))

        self.imc.add(node)

        self.assertRaises(NodeAlreadyExistsError, self.imc.commit,

            message='new message',

        self.imc.add(FileNode('foo/fbar', content='foobar'))

        tip = self.imc.commit(u'Initial', u'joe.doe@example.com')

        # Change node's content

        node = FileNode('foo/bar/baz', content='My **changed** content')

        self.imc.change(node)

        self.imc.commit(u'Changed %s' % node.path, u'joe.doe@example.com')

        newtip = self.repo.get_changeset()

        self.assertNotEqual(tip, newtip)

        self.assertNotEqual(tip.id, newtip.id)

        self.assertEqual(newtip.get_node('foo/bar/baz').content,

    def test_change_raise_empty_repository(self):

        node = FileNode('foobar')

        self.assertRaises(EmptyRepositoryError, self.imc.change, node)

    def test_check_integrity_change_raise_node_does_not_exist(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.imc.commit(message=u'Added foobar', author=unicode(self))

        node = FileNode('not-foobar', content='')

        self.imc.change(node)

        self.assertRaises(NodeDoesNotExistError, self.imc.commit,