29
29
from bzrlib.tests import (
32
from bzrlib.tests.test_osutils import dir_reader_scenarios
33
from bzrlib.tests.scenarios import (
34
load_tests_apply_scenarios,
37
from bzrlib.tests import (
42
load_tests = load_tests_apply_scenarios
45
compiled_dirstate_helpers_feature = features.ModuleAvailableFeature(
46
'bzrlib._dirstate_helpers_pyx')
49
# FIXME: we should also parametrize against SHA1Provider !
51
ue_scenarios = [('dirstate_Python',
52
{'update_entry': dirstate.py_update_entry})]
53
if compiled_dirstate_helpers_feature.available():
54
update_entry = compiled_dirstate_helpers_feature.module.update_entry
55
ue_scenarios.append(('dirstate_Pyrex', {'update_entry': update_entry}))
57
pe_scenarios = [('dirstate_Python',
58
{'_process_entry': dirstate.ProcessEntryPython})]
59
if compiled_dirstate_helpers_feature.available():
60
process_entry = compiled_dirstate_helpers_feature.module.ProcessEntryC
61
pe_scenarios.append(('dirstate_Pyrex', {'_process_entry': process_entry}))
35
from bzrlib import _dirstate_helpers_pyx
36
has_dirstate_helpers_pyx = True
38
has_dirstate_helpers_pyx = False
41
class _CompiledDirstateHelpersFeature(tests.Feature):
43
return has_dirstate_helpers_pyx
45
def feature_name(self):
46
return 'bzrlib._dirstate_helpers_pyx'
47
CompiledDirstateHelpersFeature = _CompiledDirstateHelpersFeature()
50
def load_tests(basic_tests, module, loader):
51
# FIXME: we should also parametrize against SHA1Provider !
52
suite = loader.suiteClass()
53
remaining_tests = basic_tests
55
dir_reader_scenarios = test_osutils.dir_reader_scenarios()
57
ue_scenarios = [('dirstate_Python',
58
{'update_entry': dirstate.py_update_entry})]
59
if has_dirstate_helpers_pyx:
60
pyrex_scenario = ('dirstate_Pyrex',
61
{'update_entry': _dirstate_helpers_pyx.update_entry})
62
ue_scenarios.append(pyrex_scenario)
63
process_entry_tests, remaining_tests = tests.split_suite_by_condition(
64
remaining_tests, tests.condition_isinstance(TestUpdateEntry))
65
tests.multiply_tests(process_entry_tests,
66
tests.multiply_scenarios(dir_reader_scenarios,
70
pe_scenarios = [('dirstate_Python',
71
{'_process_entry': dirstate.ProcessEntryPython})]
72
if has_dirstate_helpers_pyx:
75
{'_process_entry': _dirstate_helpers_pyx.ProcessEntryC})
76
pe_scenarios.append(pyrex_scenario)
77
process_entry_tests, remaining_tests = tests.split_suite_by_condition(
78
remaining_tests, tests.condition_isinstance(TestProcessEntry))
79
tests.multiply_tests(process_entry_tests,
80
tests.multiply_scenarios(dir_reader_scenarios,
84
dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
85
remaining_tests, tests.condition_isinstance(
86
test_dirstate.TestCaseWithDirState))
87
tests.multiply_tests(dir_reader_tests, dir_reader_scenarios, suite)
88
suite.addTest(remaining_tests)
64
93
class TestBisectPathMixin(object):
717
744
def test_trailing_garbage(self):
718
745
tree, state, expected = self.create_basic_dirstate()
719
# On Unix, we can write extra data as long as we haven't read yet, but
720
# on Win32, if you've opened the file with FILE_SHARE_READ, trying to
721
# open it in append mode will fail.
746
# We can modify the file as long as it hasn't been read yet.
723
747
f = open('dirstate', 'ab')
725
749
# Add bogus trailing garbage
726
750
f.write('bogus\n')
730
753
e = self.assertRaises(errors.DirstateCorrupt,
731
754
state._read_dirblocks_if_needed)
732
755
# Make sure we mention the bogus characters in the error
754
777
def test_bisect_dirblock(self):
755
if compiled_dirstate_helpers_feature.available():
778
if CompiledDirstateHelpersFeature.available():
756
779
from bzrlib._dirstate_helpers_pyx import bisect_dirblock
758
781
from bzrlib._dirstate_helpers_py import bisect_dirblock
759
782
self.assertIs(bisect_dirblock, dirstate.bisect_dirblock)
761
784
def test__bisect_path_left(self):
762
if compiled_dirstate_helpers_feature.available():
785
if CompiledDirstateHelpersFeature.available():
763
786
from bzrlib._dirstate_helpers_pyx import _bisect_path_left
765
788
from bzrlib._dirstate_helpers_py import _bisect_path_left
766
789
self.assertIs(_bisect_path_left, dirstate._bisect_path_left)
768
791
def test__bisect_path_right(self):
769
if compiled_dirstate_helpers_feature.available():
792
if CompiledDirstateHelpersFeature.available():
770
793
from bzrlib._dirstate_helpers_pyx import _bisect_path_right
772
795
from bzrlib._dirstate_helpers_py import _bisect_path_right
773
796
self.assertIs(_bisect_path_right, dirstate._bisect_path_right)
775
798
def test_cmp_by_dirs(self):
776
if compiled_dirstate_helpers_feature.available():
799
if CompiledDirstateHelpersFeature.available():
777
800
from bzrlib._dirstate_helpers_pyx import cmp_by_dirs
779
802
from bzrlib._dirstate_helpers_py import cmp_by_dirs
780
803
self.assertIs(cmp_by_dirs, dirstate.cmp_by_dirs)
782
805
def test__read_dirblocks(self):
783
if compiled_dirstate_helpers_feature.available():
806
if CompiledDirstateHelpersFeature.available():
784
807
from bzrlib._dirstate_helpers_pyx import _read_dirblocks
786
809
from bzrlib._dirstate_helpers_py import _read_dirblocks
787
810
self.assertIs(_read_dirblocks, dirstate._read_dirblocks)
789
812
def test_update_entry(self):
790
if compiled_dirstate_helpers_feature.available():
813
if CompiledDirstateHelpersFeature.available():
791
814
from bzrlib._dirstate_helpers_pyx import update_entry
793
816
from bzrlib.dirstate import update_entry
794
817
self.assertIs(update_entry, dirstate.update_entry)
796
819
def test_process_entry(self):
797
if compiled_dirstate_helpers_feature.available():
820
if CompiledDirstateHelpersFeature.available():
798
821
from bzrlib._dirstate_helpers_pyx import ProcessEntryC
799
822
self.assertIs(ProcessEntryC, dirstate._process_entry)
805
828
class TestUpdateEntry(test_dirstate.TestCaseWithDirState):
806
829
"""Test the DirState.update_entry functions"""
808
scenarios = multiply_scenarios(
809
dir_reader_scenarios(), ue_scenarios)
811
831
# Set by load_tests
812
832
update_entry = None
815
835
super(TestUpdateEntry, self).setUp()
816
self.overrideAttr(dirstate, 'update_entry', self.update_entry)
836
orig = dirstate.update_entry
838
dirstate.update_entry = orig
839
self.addCleanup(cleanup)
840
dirstate.update_entry = self.update_entry
818
842
def get_state_with_a(self):
819
843
"""Create a DirState tracking a single object named 'a'"""
826
850
def test_observed_sha1_cachable(self):
827
851
state, entry = self.get_state_with_a()
829
852
atime = time.time() - 10
830
853
self.build_tree(['a'])
831
statvalue = test_dirstate._FakeStat.from_stat(os.lstat('a'))
832
statvalue.st_mtime = statvalue.st_ctime = atime
833
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
834
state._dirblock_state)
854
statvalue = os.lstat('a')
855
statvalue = test_dirstate._FakeStat(statvalue.st_size, atime, atime,
856
statvalue.st_dev, statvalue.st_ino, statvalue.st_mode)
835
857
state._observed_sha1(entry, "foo", statvalue)
836
858
self.assertEqual('foo', entry[1][0][1])
837
859
packed_stat = dirstate.pack_stat(statvalue)
838
860
self.assertEqual(packed_stat, entry[1][0][4])
839
self.assertEqual(dirstate.DirState.IN_MEMORY_HASH_MODIFIED,
840
state._dirblock_state)
842
862
def test_observed_sha1_not_cachable(self):
843
863
state, entry = self.get_state_with_a()
845
864
oldval = entry[1][0][1]
846
865
oldstat = entry[1][0][4]
847
866
self.build_tree(['a'])
848
867
statvalue = os.lstat('a')
849
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
850
state._dirblock_state)
851
868
state._observed_sha1(entry, "foo", statvalue)
852
869
self.assertEqual(oldval, entry[1][0][1])
853
870
self.assertEqual(oldstat, entry[1][0][4])
854
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
855
state._dirblock_state)
857
872
def test_update_entry(self):
858
873
state, _ = self.get_state_with_a()
883
898
stat_value=stat_value)
884
899
self.assertEqual(None, link_or_sha1)
886
# The dirblock entry should not have computed or cached the file's
887
# sha1, but it did update the files' st_size. However, this is not
888
# worth writing a dirstate file for, so we leave the state UNMODIFIED
901
# The dirblock entry should not have cached the file's sha1 (too new)
889
902
self.assertEqual(('f', '', 14, False, dirstate.DirState.NULLSTAT),
891
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
904
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
892
905
state._dirblock_state)
893
906
mode = stat_value.st_mode
894
907
self.assertEqual([('is_exec', mode, False)], state._log)
971
983
# Dirblock is not updated (the link is too new)
972
984
self.assertEqual([('l', '', 6, False, dirstate.DirState.NULLSTAT)],
974
# The file entry turned into a symlink, that is considered
975
# HASH modified worthy.
976
self.assertEqual(dirstate.DirState.IN_MEMORY_HASH_MODIFIED,
986
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
977
987
state._dirblock_state)
979
989
# Because the stat_value looks new, we should re-read the target
981
990
link_or_sha1 = self.update_entry(state, entry, abspath='a',
982
991
stat_value=stat_value)
983
992
self.assertEqual('target', link_or_sha1)
984
self.assertEqual([('read_link', 'a', '')], state._log)
993
self.assertEqual([('read_link', 'a', ''),
994
('read_link', 'a', ''),
985
996
self.assertEqual([('l', '', 6, False, dirstate.DirState.NULLSTAT)],
988
998
state.adjust_time(+20) # Skip into the future, all files look old
990
999
link_or_sha1 = self.update_entry(state, entry, abspath='a',
991
1000
stat_value=stat_value)
992
# The symlink stayed a symlink. So while it is new enough to cache, we
993
# don't bother setting the flag, because it is not really worth saving
994
# (when we stat the symlink, we'll have paged in the target.)
995
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
996
state._dirblock_state)
997
1001
self.assertEqual('target', link_or_sha1)
998
1002
# We need to re-read the link because only now can we cache it
999
self.assertEqual([('read_link', 'a', '')], state._log)
1003
self.assertEqual([('read_link', 'a', ''),
1004
('read_link', 'a', ''),
1005
('read_link', 'a', ''),
1000
1007
self.assertEqual([('l', 'target', 6, False, packed_stat)],
1004
1010
# Another call won't re-read the link
1005
self.assertEqual([], state._log)
1011
self.assertEqual([('read_link', 'a', ''),
1012
('read_link', 'a', ''),
1013
('read_link', 'a', ''),
1006
1015
link_or_sha1 = self.update_entry(state, entry, abspath='a',
1007
1016
stat_value=stat_value)
1008
1017
self.assertEqual('target', link_or_sha1)
1023
1032
self.build_tree(['a/'])
1024
1033
state.adjust_time(+20)
1025
1034
self.assertIs(None, self.do_update_entry(state, entry, 'a'))
1026
# a/ used to be a file, but is now a directory, worth saving
1027
1035
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
1028
1036
state._dirblock_state)
1030
1038
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
1031
1039
state._dirblock_state)
1032
# No changes to a/ means not worth saving.
1033
self.assertIs(None, self.do_update_entry(state, entry, 'a'))
1034
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
1035
state._dirblock_state)
1036
# Change the last-modified time for the directory
1037
t = time.time() - 100.0
1039
os.utime('a', (t, t))
1041
# It looks like Win32 + FAT doesn't allow to change times on a dir.
1042
raise tests.TestSkipped("can't update mtime of a dir on FAT")
1043
saved_packed_stat = entry[1][0][-1]
1044
self.assertIs(None, self.do_update_entry(state, entry, 'a'))
1045
# We *do* go ahead and update the information in the dirblocks, but we
1046
# don't bother setting IN_MEMORY_MODIFIED because it is trivial to
1048
self.assertNotEqual(saved_packed_stat, entry[1][0][-1])
1040
self.assertIs(None, self.do_update_entry(state, entry, 'a'))
1049
1041
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
1050
1042
state._dirblock_state)
1284
1276
class TestProcessEntry(test_dirstate.TestCaseWithDirState):
1286
scenarios = multiply_scenarios(dir_reader_scenarios(), pe_scenarios)
1288
1278
# Set by load_tests
1289
1279
_process_entry = None
1291
1281
def setUp(self):
1292
1282
super(TestProcessEntry, self).setUp()
1293
self.overrideAttr(dirstate, '_process_entry', self._process_entry)
1283
orig = dirstate._process_entry
1285
dirstate._process_entry = orig
1286
self.addCleanup(cleanup)
1287
dirstate._process_entry = self._process_entry
1295
1289
def assertChangedFileIds(self, expected, tree):
1296
1290
tree.lock_read()
1302
1296
self.assertEqual(sorted(expected), sorted(file_ids))
1304
def test_exceptions_raised(self):
1305
# This is a direct test of bug #495023, it relies on osutils.is_inside
1306
# getting called in an inner function. Which makes it a bit brittle,
1307
# but at least it does reproduce the bug.
1308
tree = self.make_branch_and_tree('tree')
1309
self.build_tree(['tree/file', 'tree/dir/', 'tree/dir/sub',
1310
'tree/dir2/', 'tree/dir2/sub2'])
1311
tree.add(['file', 'dir', 'dir/sub', 'dir2', 'dir2/sub2'])
1312
tree.commit('first commit')
1314
self.addCleanup(tree.unlock)
1315
basis_tree = tree.basis_tree()
1316
def is_inside_raises(*args, **kwargs):
1317
raise RuntimeError('stop this')
1318
self.overrideAttr(osutils, 'is_inside', is_inside_raises)
1319
self.assertListRaises(RuntimeError, tree.iter_changes, basis_tree)
1321
1298
def test_simple_changes(self):
1322
1299
tree = self.make_branch_and_tree('tree')
1323
1300
self.build_tree(['tree/file'])