/brz/remove-bazaar : contents of breezy/tests/test__dirstate

: (revision 7490.7.6)

To get this branch, use:

bzr branch
http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

# Copyright (C) 2007-2011 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tests for the compiled dirstate helpers."""

import bisect
import os
import time

from .. import (
    osutils,
    tests,
    )
from ..bzr import (
    dirstate,
    _dirstate_helpers_py,
    )
from . import (
    test_dirstate,
    )
from .test_osutils import dir_reader_scenarios
from .scenarios import (
    load_tests_apply_scenarios,
    multiply_scenarios,
    )
from . import (
    features,
    )


load_tests = load_tests_apply_scenarios


compiled_dirstate_helpers_feature = features.ModuleAvailableFeature(
    'breezy.bzr._dirstate_helpers_pyx')


# FIXME: we should also parametrize against SHA1Provider !

ue_scenarios = [('dirstate_Python',
                 {'update_entry': dirstate.py_update_entry})]
if compiled_dirstate_helpers_feature.available():
    update_entry = compiled_dirstate_helpers_feature.module.update_entry
    ue_scenarios.append(('dirstate_Pyrex', {'update_entry': update_entry}))

pe_scenarios = [('dirstate_Python',
                 {'_process_entry': dirstate.ProcessEntryPython})]
if compiled_dirstate_helpers_feature.available():
    process_entry = compiled_dirstate_helpers_feature.module.ProcessEntryC
    pe_scenarios.append(('dirstate_Pyrex', {'_process_entry': process_entry}))

helper_scenarios = [('dirstate_Python', {'helpers': _dirstate_helpers_py})]
if compiled_dirstate_helpers_feature.available():
    helper_scenarios.append(('dirstate_Pyrex',
                             {'helpers': compiled_dirstate_helpers_feature.module}))


class TestBisectPathMixin(object):
    """Test that _bisect_path_*() returns the expected values.

    _bisect_path_* is intended to work like bisect.bisect_*() except it
    knows it is working on paths that are sorted by ('path', 'to', 'foo')
    chunks rather than by raw 'path/to/foo'.

    Test Cases should inherit from this and override ``get_bisect_path`` return
    their implementation, and ``get_bisect`` to return the matching
    bisect.bisect_* function.
    """

    def get_bisect_path(self):
        """Return an implementation of _bisect_path_*"""
        raise NotImplementedError

    def get_bisect(self):
        """Return a version of bisect.bisect_*.

        Also, for the 'exists' check, return the offset to the real values.
        For example bisect_left returns the index of an entry, while
        bisect_right returns the index *after* an entry

        :return: (bisect_func, offset)
        """
        raise NotImplementedError

    def assertBisect(self, paths, split_paths, path, exists=True):
        """Assert that bisect_split works like bisect_left on the split paths.

        :param paths: A list of path names
        :param split_paths: A list of path names that are already split up by directory
            ('path/to/foo' => ('path', 'to', 'foo'))
        :param path: The path we are indexing.
        :param exists: The path should be present, so make sure the
            final location actually points to the right value.

        All other arguments will be passed along.
        """
        bisect_path = self.get_bisect_path()
        self.assertIsInstance(paths, list)
        bisect_path_idx = bisect_path(paths, path)
        split_path = self.split_for_dirblocks([path])[0]
        bisect_func, offset = self.get_bisect()
        bisect_split_idx = bisect_func(split_paths, split_path)
        self.assertEqual(bisect_split_idx, bisect_path_idx,
                         '%s disagreed. %s != %s'
                         ' for key %r'
                         % (bisect_path.__name__,
                            bisect_split_idx, bisect_path_idx, path)
                         )
        if exists:
            self.assertEqual(path, paths[bisect_path_idx + offset])

    def split_for_dirblocks(self, paths):
        dir_split_paths = []
        for path in paths:
            dirname, basename = os.path.split(path)
            dir_split_paths.append((dirname.split(b'/'), basename))
        dir_split_paths.sort()
        return dir_split_paths

    def test_simple(self):
        """In the simple case it works just like bisect_left"""
        paths = [b'', b'a', b'b', b'c', b'd']
        split_paths = self.split_for_dirblocks(paths)
        for path in paths:
            self.assertBisect(paths, split_paths, path, exists=True)
        self.assertBisect(paths, split_paths, b'_', exists=False)
        self.assertBisect(paths, split_paths, b'aa', exists=False)
        self.assertBisect(paths, split_paths, b'bb', exists=False)
        self.assertBisect(paths, split_paths, b'cc', exists=False)
        self.assertBisect(paths, split_paths, b'dd', exists=False)
        self.assertBisect(paths, split_paths, b'a/a', exists=False)
        self.assertBisect(paths, split_paths, b'b/b', exists=False)
        self.assertBisect(paths, split_paths, b'c/c', exists=False)
        self.assertBisect(paths, split_paths, b'd/d', exists=False)

    def test_involved(self):
        """This is where bisect_path_* diverges slightly."""
        # This is the list of paths and their contents
        # a/
        #   a/
        #     a
        #     z
        #   a-a/
        #     a
        #   a-z/
        #     z
        #   a=a/
        #     a
        #   a=z/
        #     z
        #   z/
        #     a
        #     z
        #   z-a
        #   z-z
        #   z=a
        #   z=z
        # a-a/
        #   a
        # a-z/
        #   z
        # a=a/
        #   a
        # a=z/
        #   z
        # This is the exact order that is stored by dirstate
        # All children in a directory are mentioned before an children of
        # children are mentioned.
        # So all the root-directory paths, then all the
        # first sub directory, etc.
        paths = [  # content of '/'
            b'', b'a', b'a-a', b'a-z', b'a=a', b'a=z',
                 # content of 'a/'
                 b'a/a', b'a/a-a', b'a/a-z',
                 b'a/a=a', b'a/a=z',
                 b'a/z', b'a/z-a', b'a/z-z',
                 b'a/z=a', b'a/z=z',
                 # content of 'a/a/'
                 b'a/a/a', b'a/a/z',
                 # content of 'a/a-a'
                 b'a/a-a/a',
                 # content of 'a/a-z'
                 b'a/a-z/z',
                 # content of 'a/a=a'
                 b'a/a=a/a',
                 # content of 'a/a=z'
                 b'a/a=z/z',
                 # content of 'a/z/'
                 b'a/z/a', b'a/z/z',
                 # content of 'a-a'
                 b'a-a/a',
                 # content of 'a-z'
                 b'a-z/z',
                 # content of 'a=a'
                 b'a=a/a',
                 # content of 'a=z'
                 b'a=z/z',
            ]
        split_paths = self.split_for_dirblocks(paths)
        sorted_paths = []
        for dir_parts, basename in split_paths:
            if dir_parts == [b'']:
                sorted_paths.append(basename)
            else:
                sorted_paths.append(b'/'.join(dir_parts + [basename]))

        self.assertEqual(sorted_paths, paths)

        for path in paths:
            self.assertBisect(paths, split_paths, path, exists=True)


class TestBisectPathLeft(tests.TestCase, TestBisectPathMixin):
    """Run all Bisect Path tests against _bisect_path_left."""

    def get_bisect_path(self):
        from breezy.bzr._dirstate_helpers_py import _bisect_path_left
        return _bisect_path_left

    def get_bisect(self):
        return bisect.bisect_left, 0


class TestCompiledBisectPathLeft(TestBisectPathLeft):
    """Run all Bisect Path tests against _bisect_path_lect"""

    _test_needs_features = [compiled_dirstate_helpers_feature]

    def get_bisect_path(self):
        from breezy.bzr._dirstate_helpers_pyx import _bisect_path_left
        return _bisect_path_left


class TestBisectPathRight(tests.TestCase, TestBisectPathMixin):
    """Run all Bisect Path tests against _bisect_path_right"""

    def get_bisect_path(self):
        from breezy.bzr._dirstate_helpers_py import _bisect_path_right
        return _bisect_path_right

    def get_bisect(self):
        return bisect.bisect_right, -1


class TestCompiledBisectPathRight(TestBisectPathRight):
    """Run all Bisect Path tests against _bisect_path_right"""

    _test_needs_features = [compiled_dirstate_helpers_feature]

    def get_bisect_path(self):
        from breezy.bzr._dirstate_helpers_pyx import _bisect_path_right
        return _bisect_path_right


class TestBisectDirblock(tests.TestCase):
    """Test that bisect_dirblock() returns the expected values.

    bisect_dirblock is intended to work like bisect.bisect_left() except it
    knows it is working on dirblocks and that dirblocks are sorted by ('path',
    'to', 'foo') chunks rather than by raw 'path/to/foo'.

    This test is parameterized by calling get_bisect_dirblock(). Child test
    cases can override this function to test against a different
    implementation.
    """

    def get_bisect_dirblock(self):
        """Return an implementation of bisect_dirblock"""
        from breezy.bzr._dirstate_helpers_py import bisect_dirblock
        return bisect_dirblock

    def assertBisect(self, dirblocks, split_dirblocks, path, *args, **kwargs):
        """Assert that bisect_split works like bisect_left on the split paths.

        :param dirblocks: A list of (path, [info]) pairs.
        :param split_dirblocks: A list of ((split, path), [info]) pairs.
        :param path: The path we are indexing.

        All other arguments will be passed along.
        """
        bisect_dirblock = self.get_bisect_dirblock()
        self.assertIsInstance(dirblocks, list)
        bisect_split_idx = bisect_dirblock(dirblocks, path, *args, **kwargs)
        split_dirblock = (path.split(b'/'), [])
        bisect_left_idx = bisect.bisect_left(split_dirblocks, split_dirblock,
                                             *args)
        self.assertEqual(bisect_left_idx, bisect_split_idx,
                         'bisect_split disagreed. %s != %s'
                         ' for key %r'
                         % (bisect_left_idx, bisect_split_idx, path)
                         )

    def paths_to_dirblocks(self, paths):
        """Convert a list of paths into dirblock form.

        Also, ensure that the paths are in proper sorted order.
        """
        dirblocks = [(path, []) for path in paths]
        split_dirblocks = [(path.split(b'/'), []) for path in paths]
        self.assertEqual(sorted(split_dirblocks), split_dirblocks)
        return dirblocks, split_dirblocks

    def test_simple(self):
        """In the simple case it works just like bisect_left"""
        paths = [b'', b'a', b'b', b'c', b'd']
        dirblocks, split_dirblocks = self.paths_to_dirblocks(paths)
        for path in paths:
            self.assertBisect(dirblocks, split_dirblocks, path)
        self.assertBisect(dirblocks, split_dirblocks, b'_')
        self.assertBisect(dirblocks, split_dirblocks, b'aa')
        self.assertBisect(dirblocks, split_dirblocks, b'bb')
        self.assertBisect(dirblocks, split_dirblocks, b'cc')
        self.assertBisect(dirblocks, split_dirblocks, b'dd')
        self.assertBisect(dirblocks, split_dirblocks, b'a/a')
        self.assertBisect(dirblocks, split_dirblocks, b'b/b')
        self.assertBisect(dirblocks, split_dirblocks, b'c/c')
        self.assertBisect(dirblocks, split_dirblocks, b'd/d')

    def test_involved(self):
        """This is where bisect_left diverges slightly."""
        paths = [b'', b'a',
                 b'a/a', b'a/a/a', b'a/a/z', b'a/a-a', b'a/a-z',
                 b'a/z', b'a/z/a', b'a/z/z', b'a/z-a', b'a/z-z',
                 b'a-a', b'a-z',
                 b'z', b'z/a/a', b'z/a/z', b'z/a-a', b'z/a-z',
                 b'z/z', b'z/z/a', b'z/z/z', b'z/z-a', b'z/z-z',
                 b'z-a', b'z-z',
                 ]
        dirblocks, split_dirblocks = self.paths_to_dirblocks(paths)
        for path in paths:
            self.assertBisect(dirblocks, split_dirblocks, path)

    def test_involved_cached(self):
        """This is where bisect_left diverges slightly."""
        paths = [b'', b'a',
                 b'a/a', b'a/a/a', b'a/a/z', b'a/a-a', b'a/a-z',
                 b'a/z', b'a/z/a', b'a/z/z', b'a/z-a', b'a/z-z',
                 b'a-a', b'a-z',
                 b'z', b'z/a/a', b'z/a/z', b'z/a-a', b'z/a-z',
                 b'z/z', b'z/z/a', b'z/z/z', b'z/z-a', b'z/z-z',
                 b'z-a', b'z-z',
                 ]
        cache = {}
        dirblocks, split_dirblocks = self.paths_to_dirblocks(paths)
        for path in paths:
            self.assertBisect(dirblocks, split_dirblocks, path, cache=cache)


class TestCompiledBisectDirblock(TestBisectDirblock):
    """Test that bisect_dirblock() returns the expected values.

    bisect_dirblock is intended to work like bisect.bisect_left() except it
    knows it is working on dirblocks and that dirblocks are sorted by ('path',
    'to', 'foo') chunks rather than by raw 'path/to/foo'.

    This runs all the normal tests that TestBisectDirblock did, but uses the
    compiled version.
    """

    _test_needs_features = [compiled_dirstate_helpers_feature]

    def get_bisect_dirblock(self):
        from breezy.bzr._dirstate_helpers_pyx import bisect_dirblock
        return bisect_dirblock


class TestLtByDirs(tests.TestCase):
    """Test an implementation of lt_by_dirs()

    lt_by_dirs() compares 2 paths by their directory sections, rather than as
    plain strings.

    Child test cases can override ``get_lt_by_dirs`` to test a specific
    implementation.
    """

    def get_lt_by_dirs(self):
        """Get a specific implementation of lt_by_dirs."""
        from ..bzr._dirstate_helpers_py import lt_by_dirs
        return lt_by_dirs

    def assertCmpByDirs(self, expected, str1, str2):
        """Compare the two strings, in both directions.

        :param expected: The expected comparison value. -1 means str1 comes
            first, 0 means they are equal, 1 means str2 comes first
        :param str1: string to compare
        :param str2: string to compare
        """
        lt_by_dirs = self.get_lt_by_dirs()
        if expected == 0:
            self.assertEqual(str1, str2)
            self.assertFalse(lt_by_dirs(str1, str2))
            self.assertFalse(lt_by_dirs(str2, str1))
        elif expected > 0:
            self.assertFalse(lt_by_dirs(str1, str2))
            self.assertTrue(lt_by_dirs(str2, str1))
        else:
            self.assertTrue(lt_by_dirs(str1, str2))
            self.assertFalse(lt_by_dirs(str2, str1))

    def test_cmp_empty(self):
        """Compare against the empty string."""
        self.assertCmpByDirs(0, b'', b'')
        self.assertCmpByDirs(1, b'a', b'')
        self.assertCmpByDirs(1, b'ab', b'')
        self.assertCmpByDirs(1, b'abc', b'')
        self.assertCmpByDirs(1, b'abcd', b'')
        self.assertCmpByDirs(1, b'abcde', b'')
        self.assertCmpByDirs(1, b'abcdef', b'')
        self.assertCmpByDirs(1, b'abcdefg', b'')
        self.assertCmpByDirs(1, b'abcdefgh', b'')
        self.assertCmpByDirs(1, b'abcdefghi', b'')
        self.assertCmpByDirs(1, b'test/ing/a/path/', b'')

    def test_cmp_same_str(self):
        """Compare the same string"""
        self.assertCmpByDirs(0, b'a', b'a')
        self.assertCmpByDirs(0, b'ab', b'ab')
        self.assertCmpByDirs(0, b'abc', b'abc')
        self.assertCmpByDirs(0, b'abcd', b'abcd')
        self.assertCmpByDirs(0, b'abcde', b'abcde')
        self.assertCmpByDirs(0, b'abcdef', b'abcdef')
        self.assertCmpByDirs(0, b'abcdefg', b'abcdefg')
        self.assertCmpByDirs(0, b'abcdefgh', b'abcdefgh')
        self.assertCmpByDirs(0, b'abcdefghi', b'abcdefghi')
        self.assertCmpByDirs(0, b'testing a long string',
                             b'testing a long string')
        self.assertCmpByDirs(0, b'x' * 10000, b'x' * 10000)
        self.assertCmpByDirs(0, b'a/b', b'a/b')
        self.assertCmpByDirs(0, b'a/b/c', b'a/b/c')
        self.assertCmpByDirs(0, b'a/b/c/d', b'a/b/c/d')
        self.assertCmpByDirs(0, b'a/b/c/d/e', b'a/b/c/d/e')

    def test_simple_paths(self):
        """Compare strings that act like normal string comparison"""
        self.assertCmpByDirs(-1, b'a', b'b')
        self.assertCmpByDirs(-1, b'aa', b'ab')
        self.assertCmpByDirs(-1, b'ab', b'bb')
        self.assertCmpByDirs(-1, b'aaa', b'aab')
        self.assertCmpByDirs(-1, b'aab', b'abb')
        self.assertCmpByDirs(-1, b'abb', b'bbb')
        self.assertCmpByDirs(-1, b'aaaa', b'aaab')
        self.assertCmpByDirs(-1, b'aaab', b'aabb')
        self.assertCmpByDirs(-1, b'aabb', b'abbb')
        self.assertCmpByDirs(-1, b'abbb', b'bbbb')
        self.assertCmpByDirs(-1, b'aaaaa', b'aaaab')
        self.assertCmpByDirs(-1, b'a/a', b'a/b')
        self.assertCmpByDirs(-1, b'a/b', b'b/b')
        self.assertCmpByDirs(-1, b'a/a/a', b'a/a/b')
        self.assertCmpByDirs(-1, b'a/a/b', b'a/b/b')
        self.assertCmpByDirs(-1, b'a/b/b', b'b/b/b')
        self.assertCmpByDirs(-1, b'a/a/a/a', b'a/a/a/b')
        self.assertCmpByDirs(-1, b'a/a/a/b', b'a/a/b/b')
        self.assertCmpByDirs(-1, b'a/a/b/b', b'a/b/b/b')
        self.assertCmpByDirs(-1, b'a/b/b/b', b'b/b/b/b')
        self.assertCmpByDirs(-1, b'a/a/a/a/a', b'a/a/a/a/b')

    def test_tricky_paths(self):
        self.assertCmpByDirs(1, b'ab/cd/ef', b'ab/cc/ef')
        self.assertCmpByDirs(1, b'ab/cd/ef', b'ab/c/ef')
        self.assertCmpByDirs(-1, b'ab/cd/ef', b'ab/cd-ef')
        self.assertCmpByDirs(-1, b'ab/cd', b'ab/cd-')
        self.assertCmpByDirs(-1, b'ab/cd', b'ab-cd')

    def test_cmp_unicode_not_allowed(self):
        lt_by_dirs = self.get_lt_by_dirs()
        self.assertRaises(TypeError, lt_by_dirs, u'Unicode', b'str')
        self.assertRaises(TypeError, lt_by_dirs, b'str', u'Unicode')
        self.assertRaises(TypeError, lt_by_dirs, u'Unicode', u'Unicode')

    def test_cmp_non_ascii(self):
        self.assertCmpByDirs(-1, b'\xc2\xb5', b'\xc3\xa5')  # u'\xb5', u'\xe5'
        self.assertCmpByDirs(-1, b'a', b'\xc3\xa5')  # u'a', u'\xe5'
        self.assertCmpByDirs(-1, b'b', b'\xc2\xb5')  # u'b', u'\xb5'
        self.assertCmpByDirs(-1, b'a/b', b'a/\xc3\xa5')  # u'a/b', u'a/\xe5'
        self.assertCmpByDirs(-1, b'b/a', b'b/\xc2\xb5')  # u'b/a', u'b/\xb5'


class TestCompiledLtByDirs(TestLtByDirs):
    """Test the pyrex implementation of lt_by_dirs"""

    _test_needs_features = [compiled_dirstate_helpers_feature]

    def get_lt_by_dirs(self):
        from ..bzr._dirstate_helpers_pyx import lt_by_dirs
        return lt_by_dirs


class TestLtPathByDirblock(tests.TestCase):
    """Test an implementation of _lt_path_by_dirblock()

    _lt_path_by_dirblock() compares two paths using the sort order used by
    DirState. All paths in the same directory are sorted together.

    Child test cases can override ``get_lt_path_by_dirblock`` to test a specific
    implementation.
    """

    def get_lt_path_by_dirblock(self):
        """Get a specific implementation of _lt_path_by_dirblock."""
        from ..bzr._dirstate_helpers_py import _lt_path_by_dirblock
        return _lt_path_by_dirblock

    def assertLtPathByDirblock(self, paths):
        """Compare all paths and make sure they evaluate to the correct order.

        This does N^2 comparisons. It is assumed that ``paths`` is properly
        sorted list.

        :param paths: a sorted list of paths to compare
        """
        # First, make sure the paths being passed in are correct
        def _key(p):
            dirname, basename = os.path.split(p)
            return dirname.split(b'/'), basename
        self.assertEqual(sorted(paths, key=_key), paths)

        lt_path_by_dirblock = self.get_lt_path_by_dirblock()
        for idx1, path1 in enumerate(paths):
            for idx2, path2 in enumerate(paths):
                lt_result = lt_path_by_dirblock(path1, path2)
                self.assertEqual(idx1 < idx2, lt_result,
                                 '%s did not state that %r < %r, lt=%s'
                                 % (lt_path_by_dirblock.__name__,
                                    path1, path2, lt_result))

    def test_cmp_simple_paths(self):
        """Compare against the empty string."""
        self.assertLtPathByDirblock(
            [b'', b'a', b'ab', b'abc', b'a/b/c', b'b/d/e'])
        self.assertLtPathByDirblock([b'kl', b'ab/cd', b'ab/ef', b'gh/ij'])

    def test_tricky_paths(self):
        self.assertLtPathByDirblock([
            # Contents of ''
            b'', b'a', b'a-a', b'a=a', b'b',
            # Contents of 'a'
            b'a/a', b'a/a-a', b'a/a=a', b'a/b',
            # Contents of 'a/a'
            b'a/a/a', b'a/a/a-a', b'a/a/a=a',
            # Contents of 'a/a/a'
            b'a/a/a/a', b'a/a/a/b',
            # Contents of 'a/a/a-a',
            b'a/a/a-a/a', b'a/a/a-a/b',
            # Contents of 'a/a/a=a',
            b'a/a/a=a/a', b'a/a/a=a/b',
            # Contents of 'a/a-a'
            b'a/a-a/a',
            # Contents of 'a/a-a/a'
            b'a/a-a/a/a', b'a/a-a/a/b',
            # Contents of 'a/a=a'
            b'a/a=a/a',
            # Contents of 'a/b'
            b'a/b/a', b'a/b/b',
            # Contents of 'a-a',
            b'a-a/a', b'a-a/b',
            # Contents of 'a=a',
            b'a=a/a', b'a=a/b',
            # Contents of 'b',
            b'b/a', b'b/b',
            ])
        self.assertLtPathByDirblock([
            # content of '/'
            b'', b'a', b'a-a', b'a-z', b'a=a', b'a=z',
                 # content of 'a/'
                 b'a/a', b'a/a-a', b'a/a-z',
                 b'a/a=a', b'a/a=z',
                 b'a/z', b'a/z-a', b'a/z-z',
                 b'a/z=a', b'a/z=z',
                 # content of 'a/a/'
                 b'a/a/a', b'a/a/z',
                 # content of 'a/a-a'
                 b'a/a-a/a',
                 # content of 'a/a-z'
                 b'a/a-z/z',
                 # content of 'a/a=a'
                 b'a/a=a/a',
                 # content of 'a/a=z'
                 b'a/a=z/z',
                 # content of 'a/z/'
                 b'a/z/a', b'a/z/z',
                 # content of 'a-a'
                 b'a-a/a',
                 # content of 'a-z'
                 b'a-z/z',
                 # content of 'a=a'
                 b'a=a/a',
                 # content of 'a=z'
                 b'a=z/z',
            ])

    def test_unicode_not_allowed(self):
        lt_path_by_dirblock = self.get_lt_path_by_dirblock()
        self.assertRaises(TypeError, lt_path_by_dirblock, u'Uni', 'str')
        self.assertRaises(TypeError, lt_path_by_dirblock, 'str', u'Uni')
        self.assertRaises(TypeError, lt_path_by_dirblock, u'Uni', u'Uni')
        self.assertRaises(TypeError, lt_path_by_dirblock, u'x/Uni', 'x/str')
        self.assertRaises(TypeError, lt_path_by_dirblock, 'x/str', u'x/Uni')
        self.assertRaises(TypeError, lt_path_by_dirblock, u'x/Uni', u'x/Uni')

    def test_nonascii(self):
        self.assertLtPathByDirblock([
            # content of '/'
            b'', b'a', b'\xc2\xb5', b'\xc3\xa5',
            # content of 'a'
            b'a/a', b'a/\xc2\xb5', b'a/\xc3\xa5',
            # content of 'a/a'
            b'a/a/a', b'a/a/\xc2\xb5', b'a/a/\xc3\xa5',
            # content of 'a/\xc2\xb5'
            b'a/\xc2\xb5/a', b'a/\xc2\xb5/\xc2\xb5', b'a/\xc2\xb5/\xc3\xa5',
            # content of 'a/\xc3\xa5'
            b'a/\xc3\xa5/a', b'a/\xc3\xa5/\xc2\xb5', b'a/\xc3\xa5/\xc3\xa5',
            # content of '\xc2\xb5'
            b'\xc2\xb5/a', b'\xc2\xb5/\xc2\xb5', b'\xc2\xb5/\xc3\xa5',
            # content of '\xc2\xe5'
            b'\xc3\xa5/a', b'\xc3\xa5/\xc2\xb5', b'\xc3\xa5/\xc3\xa5',
            ])


class TestCompiledLtPathByDirblock(TestLtPathByDirblock):
    """Test the pyrex implementation of _lt_path_by_dirblock"""

    _test_needs_features = [compiled_dirstate_helpers_feature]

    def get_lt_path_by_dirblock(self):
        from ..bzr._dirstate_helpers_pyx import _lt_path_by_dirblock
        return _lt_path_by_dirblock


class TestMemRChr(tests.TestCase):
    """Test memrchr functionality"""

    _test_needs_features = [compiled_dirstate_helpers_feature]

    def assertMemRChr(self, expected, s, c):
        from breezy.bzr._dirstate_helpers_pyx import _py_memrchr
        self.assertEqual(expected, _py_memrchr(s, c))

    def test_missing(self):
        self.assertMemRChr(None, b'', b'a')
        self.assertMemRChr(None, b'', b'c')
        self.assertMemRChr(None, b'abcdefghijklm', b'q')
        self.assertMemRChr(None, b'aaaaaaaaaaaaaaaaaaaaaaa', b'b')

    def test_single_entry(self):
        self.assertMemRChr(0, b'abcdefghijklm', b'a')
        self.assertMemRChr(1, b'abcdefghijklm', b'b')
        self.assertMemRChr(2, b'abcdefghijklm', b'c')
        self.assertMemRChr(10, b'abcdefghijklm', b'k')
        self.assertMemRChr(11, b'abcdefghijklm', b'l')
        self.assertMemRChr(12, b'abcdefghijklm', b'm')

    def test_multiple(self):
        self.assertMemRChr(10, b'abcdefjklmabcdefghijklm', b'a')
        self.assertMemRChr(11, b'abcdefjklmabcdefghijklm', b'b')
        self.assertMemRChr(12, b'abcdefjklmabcdefghijklm', b'c')
        self.assertMemRChr(20, b'abcdefjklmabcdefghijklm', b'k')
        self.assertMemRChr(21, b'abcdefjklmabcdefghijklm', b'l')
        self.assertMemRChr(22, b'abcdefjklmabcdefghijklm', b'm')
        self.assertMemRChr(22, b'aaaaaaaaaaaaaaaaaaaaaaa', b'a')

    def test_with_nulls(self):
        self.assertMemRChr(10, b'abc\0\0\0jklmabc\0\0\0ghijklm', b'a')
        self.assertMemRChr(11, b'abc\0\0\0jklmabc\0\0\0ghijklm', b'b')
        self.assertMemRChr(12, b'abc\0\0\0jklmabc\0\0\0ghijklm', b'c')
        self.assertMemRChr(20, b'abc\0\0\0jklmabc\0\0\0ghijklm', b'k')
        self.assertMemRChr(21, b'abc\0\0\0jklmabc\0\0\0ghijklm', b'l')
        self.assertMemRChr(22, b'abc\0\0\0jklmabc\0\0\0ghijklm', b'm')
        self.assertMemRChr(22, b'aaa\0\0\0aaaaaaa\0\0\0aaaaaaa', b'a')
        self.assertMemRChr(9, b'\0\0\0\0\0\0\0\0\0\0', b'\0')


class TestReadDirblocks(test_dirstate.TestCaseWithDirState):
    """Test an implementation of _read_dirblocks()

    _read_dirblocks() reads in all of the dirblock information from the disk
    file.

    Child test cases can override ``get_read_dirblocks`` to test a specific
    implementation.
    """

    # inherits scenarios from test_dirstate

    def get_read_dirblocks(self):
        from breezy.bzr._dirstate_helpers_py import _read_dirblocks
        return _read_dirblocks

    def test_smoketest(self):
        """Make sure that we can create and read back a simple file."""
        tree, state, expected = self.create_basic_dirstate()
        del tree
        state._read_header_if_needed()
        self.assertEqual(dirstate.DirState.NOT_IN_MEMORY,
                         state._dirblock_state)
        read_dirblocks = self.get_read_dirblocks()
        read_dirblocks(state)
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)

    def test_trailing_garbage(self):
        tree, state, expected = self.create_basic_dirstate()
        # On Unix, we can write extra data as long as we haven't read yet, but
        # on Win32, if you've opened the file with FILE_SHARE_READ, trying to
        # open it in append mode will fail.
        state.unlock()
        f = open('dirstate', 'ab')
        try:
            # Add bogus trailing garbage
            f.write(b'bogus\n')
        finally:
            f.close()
            state.lock_read()
        e = self.assertRaises(dirstate.DirstateCorrupt,
                              state._read_dirblocks_if_needed)
        # Make sure we mention the bogus characters in the error
        self.assertContainsRe(str(e), 'bogus')


class TestCompiledReadDirblocks(TestReadDirblocks):
    """Test the pyrex implementation of _read_dirblocks"""

    _test_needs_features = [compiled_dirstate_helpers_feature]

    def get_read_dirblocks(self):
        from breezy.bzr._dirstate_helpers_pyx import _read_dirblocks
        return _read_dirblocks


class TestUsingCompiledIfAvailable(tests.TestCase):
    """Check that any compiled functions that are available are the default.

    It is possible to have typos, etc in the import line, such that
    _dirstate_helpers_pyx is actually available, but the compiled functions are
    not being used.
    """

    def test_bisect_dirblock(self):
        if compiled_dirstate_helpers_feature.available():
            from breezy.bzr._dirstate_helpers_pyx import bisect_dirblock
        else:
            from breezy.bzr._dirstate_helpers_py import bisect_dirblock
        self.assertIs(bisect_dirblock, dirstate.bisect_dirblock)

    def test__bisect_path_left(self):
        if compiled_dirstate_helpers_feature.available():
            from breezy.bzr._dirstate_helpers_pyx import _bisect_path_left
        else:
            from breezy.bzr._dirstate_helpers_py import _bisect_path_left
        self.assertIs(_bisect_path_left, dirstate._bisect_path_left)

    def test__bisect_path_right(self):
        if compiled_dirstate_helpers_feature.available():
            from breezy.bzr._dirstate_helpers_pyx import _bisect_path_right
        else:
            from breezy.bzr._dirstate_helpers_py import _bisect_path_right
        self.assertIs(_bisect_path_right, dirstate._bisect_path_right)

    def test_lt_by_dirs(self):
        if compiled_dirstate_helpers_feature.available():
            from ..bzr._dirstate_helpers_pyx import lt_by_dirs
        else:
            from ..bzr._dirstate_helpers_py import lt_by_dirs
        self.assertIs(lt_by_dirs, dirstate.lt_by_dirs)

    def test__read_dirblocks(self):
        if compiled_dirstate_helpers_feature.available():
            from breezy.bzr._dirstate_helpers_pyx import _read_dirblocks
        else:
            from breezy.bzr._dirstate_helpers_py import _read_dirblocks
        self.assertIs(_read_dirblocks, dirstate._read_dirblocks)

    def test_update_entry(self):
        if compiled_dirstate_helpers_feature.available():
            from breezy.bzr._dirstate_helpers_pyx import update_entry
        else:
            from breezy.bzr.dirstate import update_entry
        self.assertIs(update_entry, dirstate.update_entry)

    def test_process_entry(self):
        if compiled_dirstate_helpers_feature.available():
            from breezy.bzr._dirstate_helpers_pyx import ProcessEntryC
            self.assertIs(ProcessEntryC, dirstate._process_entry)
        else:
            from breezy.bzr.dirstate import ProcessEntryPython
            self.assertIs(ProcessEntryPython, dirstate._process_entry)


class TestUpdateEntry(test_dirstate.TestCaseWithDirState):
    """Test the DirState.update_entry functions"""

    scenarios = multiply_scenarios(
        dir_reader_scenarios(), ue_scenarios)

    # Set by load_tests
    update_entry = None

    def setUp(self):
        super(TestUpdateEntry, self).setUp()
        self.overrideAttr(dirstate, 'update_entry', self.update_entry)

    def get_state_with_a(self):
        """Create a DirState tracking a single object named 'a'"""
        state = test_dirstate.InstrumentedDirState.initialize('dirstate')
        self.addCleanup(state.unlock)
        state.add('a', b'a-id', 'file', None, b'')
        entry = state._get_entry(0, path_utf8=b'a')
        return state, entry

    def test_observed_sha1_cachable(self):
        state, entry = self.get_state_with_a()
        state.save()
        atime = time.time() - 10
        self.build_tree(['a'])
        statvalue = test_dirstate._FakeStat.from_stat(os.lstat('a'))
        statvalue.st_mtime = statvalue.st_ctime = atime
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        state._observed_sha1(entry, b"foo", statvalue)
        self.assertEqual(b'foo', entry[1][0][1])
        packed_stat = dirstate.pack_stat(statvalue)
        self.assertEqual(packed_stat, entry[1][0][4])
        self.assertEqual(dirstate.DirState.IN_MEMORY_HASH_MODIFIED,
                         state._dirblock_state)

    def test_observed_sha1_not_cachable(self):
        state, entry = self.get_state_with_a()
        state.save()
        oldval = entry[1][0][1]
        oldstat = entry[1][0][4]
        self.build_tree(['a'])
        statvalue = os.lstat('a')
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        state._observed_sha1(entry, "foo", statvalue)
        self.assertEqual(oldval, entry[1][0][1])
        self.assertEqual(oldstat, entry[1][0][4])
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)

    def test_update_entry(self):
        state, _ = self.get_state_with_a()
        tree = self.make_branch_and_tree('tree')
        tree.lock_write()
        empty_revid = tree.commit('empty')
        self.build_tree(['tree/a'])
        tree.add(['a'], [b'a-id'])
        with_a_id = tree.commit('with_a')
        self.addCleanup(tree.unlock)
        state.set_parent_trees(
            [(empty_revid, tree.branch.repository.revision_tree(empty_revid))],
            [])
        entry = state._get_entry(0, path_utf8=b'a')
        self.build_tree(['a'])
        # Add one where we don't provide the stat or sha already
        self.assertEqual((b'', b'a', b'a-id'), entry[0])
        self.assertEqual((b'f', b'', 0, False, dirstate.DirState.NULLSTAT),
                         entry[1][0])
        # Flush the buffers to disk
        state.save()
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)

        stat_value = os.lstat('a')
        packed_stat = dirstate.pack_stat(stat_value)
        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        self.assertEqual(None, link_or_sha1)

        # The dirblock entry should not have computed or cached the file's
        # sha1, but it did update the files' st_size. However, this is not
        # worth writing a dirstate file for, so we leave the state UNMODIFIED
        self.assertEqual((b'f', b'', 14, False, dirstate.DirState.NULLSTAT),
                         entry[1][0])
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        mode = stat_value.st_mode
        self.assertEqual([('is_exec', mode, False)], state._log)

        state.save()
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)

        # Roll the clock back so the file is guaranteed to look too new. We
        # should still not compute the sha1.
        state.adjust_time(-10)
        del state._log[:]

        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        self.assertEqual([('is_exec', mode, False)], state._log)
        self.assertEqual(None, link_or_sha1)
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        self.assertEqual((b'f', b'', 14, False, dirstate.DirState.NULLSTAT),
                         entry[1][0])
        state.save()

        # If it is cachable (the clock has moved forward) but new it still
        # won't calculate the sha or cache it.
        state.adjust_time(+20)
        del state._log[:]
        link_or_sha1 = dirstate.update_entry(state, entry, abspath=b'a',
                                             stat_value=stat_value)
        self.assertEqual(None, link_or_sha1)
        self.assertEqual([('is_exec', mode, False)], state._log)
        self.assertEqual((b'f', b'', 14, False, dirstate.DirState.NULLSTAT),
                         entry[1][0])
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)

        # If the file is no longer new, and the clock has been moved forward
        # sufficiently, it will cache the sha.
        del state._log[:]
        state.set_parent_trees(
            [(with_a_id, tree.branch.repository.revision_tree(with_a_id))],
            [])
        entry = state._get_entry(0, path_utf8=b'a')

        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        self.assertEqual(b'b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6',
                         link_or_sha1)
        self.assertEqual([('is_exec', mode, False), ('sha1', b'a')],
                         state._log)
        self.assertEqual((b'f', link_or_sha1, 14, False, packed_stat),
                         entry[1][0])

        # Subsequent calls will just return the cached value
        del state._log[:]
        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        self.assertEqual(b'b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6',
                         link_or_sha1)
        self.assertEqual([], state._log)
        self.assertEqual((b'f', link_or_sha1, 14, False, packed_stat),
                         entry[1][0])

    def test_update_entry_symlink(self):
        """Update entry should read symlinks."""
        self.requireFeature(features.SymlinkFeature)
        state, entry = self.get_state_with_a()
        state.save()
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        os.symlink('target', 'a')

        state.adjust_time(-10)  # Make the symlink look new
        stat_value = os.lstat('a')
        packed_stat = dirstate.pack_stat(stat_value)
        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        self.assertEqual(b'target', link_or_sha1)
        self.assertEqual([('read_link', b'a', b'')], state._log)
        # Dirblock is not updated (the link is too new)
        self.assertEqual([(b'l', b'', 6, False, dirstate.DirState.NULLSTAT)],
                         entry[1])
        # The file entry turned into a symlink, that is considered
        # HASH modified worthy.
        self.assertEqual(dirstate.DirState.IN_MEMORY_HASH_MODIFIED,
                         state._dirblock_state)

        # Because the stat_value looks new, we should re-read the target
        del state._log[:]
        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        self.assertEqual(b'target', link_or_sha1)
        self.assertEqual([('read_link', b'a', b'')], state._log)
        self.assertEqual([(b'l', b'', 6, False, dirstate.DirState.NULLSTAT)],
                         entry[1])
        state.save()
        state.adjust_time(+20)  # Skip into the future, all files look old
        del state._log[:]
        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        # The symlink stayed a symlink. So while it is new enough to cache, we
        # don't bother setting the flag, because it is not really worth saving
        # (when we stat the symlink, we'll have paged in the target.)
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        self.assertEqual(b'target', link_or_sha1)
        # We need to re-read the link because only now can we cache it
        self.assertEqual([('read_link', b'a', b'')], state._log)
        self.assertEqual([(b'l', b'target', 6, False, packed_stat)],
                         entry[1])

        del state._log[:]
        # Another call won't re-read the link
        self.assertEqual([], state._log)
        link_or_sha1 = self.update_entry(state, entry, abspath=b'a',
                                         stat_value=stat_value)
        self.assertEqual(b'target', link_or_sha1)
        self.assertEqual([(b'l', b'target', 6, False, packed_stat)],
                         entry[1])

    def do_update_entry(self, state, entry, abspath):
        stat_value = os.lstat(abspath)
        return self.update_entry(state, entry, abspath, stat_value)

    def test_update_entry_dir(self):
        state, entry = self.get_state_with_a()
        self.build_tree(['a/'])
        self.assertIs(None, self.do_update_entry(state, entry, b'a'))

    def test_update_entry_dir_unchanged(self):
        state, entry = self.get_state_with_a()
        self.build_tree(['a/'])
        state.adjust_time(+20)
        self.assertIs(None, self.do_update_entry(state, entry, b'a'))
        # a/ used to be a file, but is now a directory, worth saving
        self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
                         state._dirblock_state)
        state.save()
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        # No changes to a/ means not worth saving.
        self.assertIs(None, self.do_update_entry(state, entry, b'a'))
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        # Change the last-modified time for the directory
        t = time.time() - 100.0
        try:
            os.utime('a', (t, t))
        except OSError:
            # It looks like Win32 + FAT doesn't allow to change times on a dir.
            raise tests.TestSkipped("can't update mtime of a dir on FAT")
        saved_packed_stat = entry[1][0][-1]
        self.assertIs(None, self.do_update_entry(state, entry, b'a'))
        # We *do* go ahead and update the information in the dirblocks, but we
        # don't bother setting IN_MEMORY_MODIFIED because it is trivial to
        # recompute.
        self.assertNotEqual(saved_packed_stat, entry[1][0][-1])
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)

    def test_update_entry_file_unchanged(self):
        state, _ = self.get_state_with_a()
        tree = self.make_branch_and_tree('tree')
        tree.lock_write()
        self.build_tree(['tree/a'])
        tree.add(['a'], [b'a-id'])
        with_a_id = tree.commit('witha')
        self.addCleanup(tree.unlock)
        state.set_parent_trees(
            [(with_a_id, tree.branch.repository.revision_tree(with_a_id))],
            [])
        entry = state._get_entry(0, path_utf8=b'a')
        self.build_tree(['a'])
        sha1sum = b'b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6'
        state.adjust_time(+20)
        self.assertEqual(sha1sum, self.do_update_entry(state, entry, b'a'))
        self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
                         state._dirblock_state)
        state.save()
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)
        self.assertEqual(sha1sum, self.do_update_entry(state, entry, b'a'))
        self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
                         state._dirblock_state)

    def test_update_entry_tree_reference(self):
        state = test_dirstate.InstrumentedDirState.initialize('dirstate')
        self.addCleanup(state.unlock)
        state.add('r', b'r-id', 'tree-reference', None, b'')
        self.build_tree(['r/'])
        entry = state._get_entry(0, path_utf8=b'r')
        self.do_update_entry(state, entry, 'r')
        entry = state._get_entry(0, path_utf8=b'r')
        self.assertEqual(b't', entry[1][0][0])

    def create_and_test_file(self, state, entry):
        """Create a file at 'a' and verify the state finds it during update.

        The state should already be versioning *something* at 'a'. This makes
        sure that state.update_entry recognizes it as a file.
        """
        self.build_tree(['a'])
        stat_value = os.lstat('a')
        packed_stat = dirstate.pack_stat(stat_value)

        link_or_sha1 = self.do_update_entry(state, entry, abspath='a')
        self.assertEqual(None, link_or_sha1)
        self.assertEqual([(b'f', b'', 14, False, dirstate.DirState.NULLSTAT)],
                         entry[1])
        return packed_stat

    def create_and_test_dir(self, state, entry):
        """Create a directory at 'a' and verify the state finds it.

        The state should already be versioning *something* at 'a'. This makes
        sure that state.update_entry recognizes it as a directory.
        """
        self.build_tree(['a/'])
        stat_value = os.lstat('a')
        packed_stat = dirstate.pack_stat(stat_value)

        link_or_sha1 = self.do_update_entry(state, entry, abspath=b'a')
        self.assertIs(None, link_or_sha1)
        self.assertEqual([(b'd', b'', 0, False, packed_stat)], entry[1])

        return packed_stat

    # FIXME: Add unicode version
    def create_and_test_symlink(self, state, entry):
        """Create a symlink at 'a' and verify the state finds it.

        The state should already be versioning *something* at 'a'. This makes
        sure that state.update_entry recognizes it as a symlink.

        This should not be called if this platform does not have symlink
        support.
        """
        # caller should care about skipping test on platforms without symlinks
        os.symlink('path/to/foo', 'a')

        stat_value = os.lstat('a')
        packed_stat = dirstate.pack_stat(stat_value)

        link_or_sha1 = self.do_update_entry(state, entry, abspath=b'a')
        self.assertEqual(b'path/to/foo', link_or_sha1)
        self.assertEqual([(b'l', b'path/to/foo', 11, False, packed_stat)],
                         entry[1])
        return packed_stat

    def test_update_file_to_dir(self):
        """If a file changes to a directory we return None for the sha.
        We also update the inventory record.
        """
        state, entry = self.get_state_with_a()
        # The file sha1 won't be cached unless the file is old
        state.adjust_time(+10)
        self.create_and_test_file(state, entry)
        os.remove('a')
        self.create_and_test_dir(state, entry)

    def test_update_file_to_symlink(self):
        """File becomes a symlink"""
        self.requireFeature(features.SymlinkFeature)
        state, entry = self.get_state_with_a()
        # The file sha1 won't be cached unless the file is old
        state.adjust_time(+10)
        self.create_and_test_file(state, entry)
        os.remove('a')
        self.create_and_test_symlink(state, entry)

    def test_update_dir_to_file(self):
        """Directory becoming a file updates the entry."""
        state, entry = self.get_state_with_a()
        # The file sha1 won't be cached unless the file is old
        state.adjust_time(+10)
        self.create_and_test_dir(state, entry)
        os.rmdir('a')
        self.create_and_test_file(state, entry)

    def test_update_dir_to_symlink(self):
        """Directory becomes a symlink"""
        self.requireFeature(features.SymlinkFeature)
        state, entry = self.get_state_with_a()
        # The symlink target won't be cached if it isn't old
        state.adjust_time(+10)
        self.create_and_test_dir(state, entry)
        os.rmdir('a')
        self.create_and_test_symlink(state, entry)

    def test_update_symlink_to_file(self):
        """Symlink becomes a file"""
        self.requireFeature(features.SymlinkFeature)
        state, entry = self.get_state_with_a()
        # The symlink and file info won't be cached unless old
        state.adjust_time(+10)
        self.create_and_test_symlink(state, entry)
        os.remove('a')
        self.create_and_test_file(state, entry)

    def test_update_symlink_to_dir(self):
        """Symlink becomes a directory"""
        self.requireFeature(features.SymlinkFeature)
        state, entry = self.get_state_with_a()
        # The symlink target won't be cached if it isn't old
        state.adjust_time(+10)
        self.create_and_test_symlink(state, entry)
        os.remove('a')
        self.create_and_test_dir(state, entry)

    def test__is_executable_win32(self):
        state, entry = self.get_state_with_a()
        self.build_tree(['a'])

        # Make sure we are using the version of _is_executable that doesn't
        # check the filesystem mode.
        state._use_filesystem_for_exec = False

        # The file on disk is not executable, but we are marking it as though
        # it is. With _use_filesystem_for_exec disabled we ignore what is on
        # disk.
        entry[1][0] = (b'f', b'', 0, True, dirstate.DirState.NULLSTAT)

        stat_value = os.lstat('a')
        packed_stat = dirstate.pack_stat(stat_value)

        state.adjust_time(-10)  # Make sure everything is new
        self.update_entry(state, entry, abspath=b'a', stat_value=stat_value)

        # The row is updated, but the executable bit stays set.
        self.assertEqual([(b'f', b'', 14, True, dirstate.DirState.NULLSTAT)],
                         entry[1])

        # Make the disk object look old enough to cache (but it won't cache the
        # sha as it is a new file).
        state.adjust_time(+20)
        digest = b'b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6'
        self.update_entry(state, entry, abspath=b'a', stat_value=stat_value)
        self.assertEqual([(b'f', b'', 14, True, dirstate.DirState.NULLSTAT)],
                         entry[1])

    def _prepare_tree(self):
        # Create a tree
        text = b'Hello World\n'
        tree = self.make_branch_and_tree('tree')
        self.build_tree_contents([('tree/a file', text)])
        tree.add('a file', b'a-file-id')
        # Note: dirstate does not sha prior to the first commit
        # so commit now in order for the test to work
        tree.commit('first')
        return tree, text

    def test_sha1provider_sha1_used(self):
        tree, text = self._prepare_tree()
        state = dirstate.DirState.from_tree(tree, 'dirstate',
                                            UppercaseSHA1Provider())
        self.addCleanup(state.unlock)
        expected_sha = osutils.sha_string(text.upper() + b"foo")
        entry = state._get_entry(0, path_utf8=b'a file')
        self.assertNotEqual((None, None), entry)
        state._sha_cutoff_time()
        state._cutoff_time += 10
        sha1 = self.update_entry(state, entry, 'tree/a file',
                                 os.lstat('tree/a file'))
        self.assertEqual(expected_sha, sha1)

    def test_sha1provider_stat_and_sha1_used(self):
        tree, text = self._prepare_tree()
        tree.lock_write()
        self.addCleanup(tree.unlock)
        state = tree._current_dirstate()
        state._sha1_provider = UppercaseSHA1Provider()
        # If we used the standard provider, it would look like nothing has
        # changed
        file_ids_changed = [change.file_id for change
                            in tree.iter_changes(tree.basis_tree())]
        self.assertEqual([b'a-file-id'], file_ids_changed)


class UppercaseSHA1Provider(dirstate.SHA1Provider):
    """A custom SHA1Provider."""

    def sha1(self, abspath):
        return self.stat_and_sha1(abspath)[1]

    def stat_and_sha1(self, abspath):
        with open(abspath, 'rb') as file_obj:
            statvalue = os.fstat(file_obj.fileno())
            text = b''.join(file_obj.readlines())
            sha1 = osutils.sha_string(text.upper() + b"foo")
        return statvalue, sha1


class TestProcessEntry(test_dirstate.TestCaseWithDirState):

    scenarios = multiply_scenarios(dir_reader_scenarios(), pe_scenarios)

    # Set by load_tests
    _process_entry = None

    def setUp(self):
        super(TestProcessEntry, self).setUp()
        self.overrideAttr(dirstate, '_process_entry', self._process_entry)

    def assertChangedFileIds(self, expected, tree):
        with tree.lock_read():
            file_ids = [info.file_id for info
                        in tree.iter_changes(tree.basis_tree())]
        self.assertEqual(sorted(expected), sorted(file_ids))

    def test_exceptions_raised(self):
        # This is a direct test of bug #495023, it relies on osutils.is_inside
        # getting called in an inner function. Which makes it a bit brittle,
        # but at least it does reproduce the bug.
        tree = self.make_branch_and_tree('tree')
        self.build_tree(['tree/file', 'tree/dir/', 'tree/dir/sub',
                         'tree/dir2/', 'tree/dir2/sub2'])
        tree.add(['file', 'dir', 'dir/sub', 'dir2', 'dir2/sub2'])
        tree.commit('first commit')
        tree.lock_read()
        self.addCleanup(tree.unlock)
        basis_tree = tree.basis_tree()

        def is_inside_raises(*args, **kwargs):
            raise RuntimeError('stop this')
        self.overrideAttr(dirstate, 'is_inside', is_inside_raises)
        try:
            from breezy.bzr import _dirstate_helpers_pyx
        except ImportError:
            pass
        else:
            self.overrideAttr(_dirstate_helpers_pyx,
                              'is_inside', is_inside_raises)
        self.overrideAttr(osutils, 'is_inside', is_inside_raises)
        self.assertListRaises(RuntimeError, tree.iter_changes, basis_tree)

    def test_simple_changes(self):
        tree = self.make_branch_and_tree('tree')
        self.build_tree(['tree/file'])
        tree.add(['file'], [b'file-id'])
        self.assertChangedFileIds([tree.path2id(''), b'file-id'], tree)
        tree.commit('one')
        self.assertChangedFileIds([], tree)

    def test_sha1provider_stat_and_sha1_used(self):
        tree = self.make_branch_and_tree('tree')
        self.build_tree(['tree/file'])
        tree.add(['file'], [b'file-id'])
        tree.commit('one')
        tree.lock_write()
        self.addCleanup(tree.unlock)
        state = tree._current_dirstate()
        state._sha1_provider = UppercaseSHA1Provider()
        self.assertChangedFileIds([b'file-id'], tree)


class TestPackStat(tests.TestCase):
    """Check packed representaton of stat values is robust on all inputs"""

    scenarios = helper_scenarios

    def pack(self, statlike_tuple):
        return self.helpers.pack_stat(os.stat_result(statlike_tuple))

    @staticmethod
    def unpack_field(packed_string, stat_field):
        return _dirstate_helpers_py._unpack_stat(packed_string)[stat_field]

    def test_result(self):
        self.assertEqual(b"AAAQAAAAABAAAAARAAAAAgAAAAEAAIHk",
                         self.pack((33252, 1, 2, 0, 0, 0, 4096, 15.5, 16.5, 17.5)))

    def test_giant_inode(self):
        packed = self.pack((33252, 0xF80000ABC, 0, 0, 0, 0, 0, 0, 0, 0))
        self.assertEqual(0x80000ABC, self.unpack_field(packed, "st_ino"))

    def test_giant_size(self):
        packed = self.pack((33252, 0, 0, 0, 0, 0, (1 << 33) + 4096, 0, 0, 0))
        self.assertEqual(4096, self.unpack_field(packed, "st_size"))

    def test_fractional_mtime(self):
        packed = self.pack((33252, 0, 0, 0, 0, 0, 0, 0, 16.9375, 0))
        self.assertEqual(16, self.unpack_field(packed, "st_mtime"))

    def test_ancient_mtime(self):
        packed = self.pack((33252, 0, 0, 0, 0, 0, 0, 0, -11644473600.0, 0))
        self.assertEqual(1240428288, self.unpack_field(packed, "st_mtime"))

    def test_distant_mtime(self):
        packed = self.pack((33252, 0, 0, 0, 0, 0, 0, 0, 64060588800.0, 0))
        self.assertEqual(3931046656, self.unpack_field(packed, "st_mtime"))

    def test_fractional_ctime(self):
        packed = self.pack((33252, 0, 0, 0, 0, 0, 0, 0, 0, 17.5625))
        self.assertEqual(17, self.unpack_field(packed, "st_ctime"))

    def test_ancient_ctime(self):
        packed = self.pack((33252, 0, 0, 0, 0, 0, 0, 0, 0, -11644473600.0))
        self.assertEqual(1240428288, self.unpack_field(packed, "st_ctime"))

    def test_distant_ctime(self):
        packed = self.pack((33252, 0, 0, 0, 0, 0, 0, 0, 0, 64060588800.0))
        self.assertEqual(3931046656, self.unpack_field(packed, "st_ctime"))

    def test_negative_dev(self):
        packed = self.pack((33252, 0, -0xFFFFFCDE, 0, 0, 0, 0, 0, 0, 0))
        self.assertEqual(0x322, self.unpack_field(packed, "st_dev"))

5557.1.7 by John Arbash Meinel Merge in the bzr.dev 5582	1	# Copyright (C) 2007-2011 Canonical Ltd
2474.1.7 by John Arbash Meinel Add some tests for a helper function that lets us	2	#
	3	# This program is free software; you can redistribute it and/or modify
	4	# it under the terms of the GNU General Public License as published by
	5	# the Free Software Foundation; either version 2 of the License, or
	6	# (at your option) any later version.
	7	#
	8	# This program is distributed in the hope that it will be useful,
	9	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	10	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	11	# GNU General Public License for more details.
	12	#
	13	# You should have received a copy of the GNU General Public License
	14	# along with this program; if not, write to the Free Software
4183.7.1 by Sabin Iacob update FSF mailing address	15	# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
2474.1.7 by John Arbash Meinel Add some tests for a helper function that lets us	16
	17	"""Tests for the compiled dirstate helpers."""
	18
2474.1.57 by John Arbash Meinel Move code around to refactor according to our pyrex extension design.	19	import bisect
2474.1.58 by John Arbash Meinel (broken) Try to properly implement DirState._bisect*	20	import os
3696.5.1 by Robert Collins Integrate in stat-cache-updates-during-commit with faster iter-changes.	21	import time
2474.1.57 by John Arbash Meinel Move code around to refactor according to our pyrex extension design.	22
6624 by Jelmer Vernooĳ Merge Python3 porting work ('py3 pokes')	23	from .. import (
4132.2.4 by Ian Clatworthy John's enhancements/fixes to the tests	24	osutils,
2474.1.7 by John Arbash Meinel Add some tests for a helper function that lets us	25	tests,
6670.4.1 by Jelmer Vernooij Update imports.	26	)
	27	from ..bzr import (
	28	dirstate,
6169.2.2 by Martin Packman Merge fix for overflow issues in pack_stat from 2.4	29	_dirstate_helpers_py,
2474.1.7 by John Arbash Meinel Add some tests for a helper function that lets us	30	)
6624 by Jelmer Vernooĳ Merge Python3 porting work ('py3 pokes')	31	from . import (
4241.14.12 by Vincent Ladeuil Far too many modifications for a single commit, need to restart.	32	test_dirstate,
5559.2.2 by Martin Pool Change to using standard load_tests_apply_scenarios.	33	)
6624 by Jelmer Vernooĳ Merge Python3 porting work ('py3 pokes')	34	from .test_osutils import dir_reader_scenarios
	35	from .scenarios import (
5559.2.2 by Martin Pool Change to using standard load_tests_apply_scenarios.	36	load_tests_apply_scenarios,
	37	multiply_scenarios,
	38	)
6624 by Jelmer Vernooĳ Merge Python3 porting work ('py3 pokes')	39	from . import (
5967.12.1 by Martin Pool Move all test features into bzrlib.tests.features	40	features,