713
762
_test_needs_features = [CompiledDirstateHelpersFeature]
715
764
def get_read_dirblocks(self):
716
from bzrlib._dirstate_helpers_c import _read_dirblocks_c
717
return _read_dirblocks_c
765
from bzrlib._dirstate_helpers_pyx import _read_dirblocks
766
return _read_dirblocks
720
769
class TestUsingCompiledIfAvailable(tests.TestCase):
721
770
"""Check that any compiled functions that are available are the default.
723
772
It is possible to have typos, etc in the import line, such that
724
_dirstate_helpers_c is actually available, but the compiled functions are
773
_dirstate_helpers_pyx is actually available, but the compiled functions are
728
777
def test_bisect_dirblock(self):
729
778
if CompiledDirstateHelpersFeature.available():
730
from bzrlib._dirstate_helpers_c import bisect_dirblock_c
731
self.assertIs(bisect_dirblock_c, dirstate.bisect_dirblock)
779
from bzrlib._dirstate_helpers_pyx import bisect_dirblock
733
from bzrlib._dirstate_helpers_py import bisect_dirblock_py
734
self.assertIs(bisect_dirblock_py, dirstate.bisect_dirblock)
781
from bzrlib._dirstate_helpers_py import bisect_dirblock
782
self.assertIs(bisect_dirblock, dirstate.bisect_dirblock)
736
784
def test__bisect_path_left(self):
737
785
if CompiledDirstateHelpersFeature.available():
738
from bzrlib._dirstate_helpers_c import _bisect_path_left_c
739
self.assertIs(_bisect_path_left_c, dirstate._bisect_path_left)
786
from bzrlib._dirstate_helpers_pyx import _bisect_path_left
741
from bzrlib._dirstate_helpers_py import _bisect_path_left_py
742
self.assertIs(_bisect_path_left_py, dirstate._bisect_path_left)
788
from bzrlib._dirstate_helpers_py import _bisect_path_left
789
self.assertIs(_bisect_path_left, dirstate._bisect_path_left)
744
791
def test__bisect_path_right(self):
745
792
if CompiledDirstateHelpersFeature.available():
746
from bzrlib._dirstate_helpers_c import _bisect_path_right_c
747
self.assertIs(_bisect_path_right_c, dirstate._bisect_path_right)
793
from bzrlib._dirstate_helpers_pyx import _bisect_path_right
749
from bzrlib._dirstate_helpers_py import _bisect_path_right_py
750
self.assertIs(_bisect_path_right_py, dirstate._bisect_path_right)
795
from bzrlib._dirstate_helpers_py import _bisect_path_right
796
self.assertIs(_bisect_path_right, dirstate._bisect_path_right)
752
798
def test_cmp_by_dirs(self):
753
799
if CompiledDirstateHelpersFeature.available():
754
from bzrlib._dirstate_helpers_c import cmp_by_dirs_c
755
self.assertIs(cmp_by_dirs_c, dirstate.cmp_by_dirs)
800
from bzrlib._dirstate_helpers_pyx import cmp_by_dirs
757
from bzrlib._dirstate_helpers_py import cmp_by_dirs_py
758
self.assertIs(cmp_by_dirs_py, dirstate.cmp_by_dirs)
802
from bzrlib._dirstate_helpers_py import cmp_by_dirs
803
self.assertIs(cmp_by_dirs, dirstate.cmp_by_dirs)
760
805
def test__read_dirblocks(self):
761
806
if CompiledDirstateHelpersFeature.available():
762
from bzrlib._dirstate_helpers_c import _read_dirblocks_c
763
self.assertIs(_read_dirblocks_c, dirstate._read_dirblocks)
765
from bzrlib._dirstate_helpers_py import _read_dirblocks_py
766
self.assertIs(_read_dirblocks_py, dirstate._read_dirblocks)
807
from bzrlib._dirstate_helpers_pyx import _read_dirblocks
809
from bzrlib._dirstate_helpers_py import _read_dirblocks
810
self.assertIs(_read_dirblocks, dirstate._read_dirblocks)
812
def test_update_entry(self):
813
if CompiledDirstateHelpersFeature.available():
814
from bzrlib._dirstate_helpers_pyx import update_entry
816
from bzrlib.dirstate import update_entry
817
self.assertIs(update_entry, dirstate.update_entry)
819
def test_process_entry(self):
820
if CompiledDirstateHelpersFeature.available():
821
from bzrlib._dirstate_helpers_pyx import ProcessEntryC
822
self.assertIs(ProcessEntryC, dirstate._process_entry)
824
from bzrlib.dirstate import ProcessEntryPython
825
self.assertIs(ProcessEntryPython, dirstate._process_entry)
828
class TestUpdateEntry(test_dirstate.TestCaseWithDirState):
829
"""Test the DirState.update_entry functions"""
835
super(TestUpdateEntry, self).setUp()
836
orig = dirstate.update_entry
838
dirstate.update_entry = orig
839
self.addCleanup(cleanup)
840
dirstate.update_entry = self.update_entry
842
def get_state_with_a(self):
843
"""Create a DirState tracking a single object named 'a'"""
844
state = test_dirstate.InstrumentedDirState.initialize('dirstate')
845
self.addCleanup(state.unlock)
846
state.add('a', 'a-id', 'file', None, '')
847
entry = state._get_entry(0, path_utf8='a')
850
def test_observed_sha1_cachable(self):
851
state, entry = self.get_state_with_a()
852
atime = time.time() - 10
853
self.build_tree(['a'])
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)
857
state._observed_sha1(entry, "foo", statvalue)
858
self.assertEqual('foo', entry[1][0][1])
859
packed_stat = dirstate.pack_stat(statvalue)
860
self.assertEqual(packed_stat, entry[1][0][4])
862
def test_observed_sha1_not_cachable(self):
863
state, entry = self.get_state_with_a()
864
oldval = entry[1][0][1]
865
oldstat = entry[1][0][4]
866
self.build_tree(['a'])
867
statvalue = os.lstat('a')
868
state._observed_sha1(entry, "foo", statvalue)
869
self.assertEqual(oldval, entry[1][0][1])
870
self.assertEqual(oldstat, entry[1][0][4])
872
def test_update_entry(self):
873
state, _ = self.get_state_with_a()
874
tree = self.make_branch_and_tree('tree')
876
empty_revid = tree.commit('empty')
877
self.build_tree(['tree/a'])
878
tree.add(['a'], ['a-id'])
879
with_a_id = tree.commit('with_a')
880
self.addCleanup(tree.unlock)
881
state.set_parent_trees(
882
[(empty_revid, tree.branch.repository.revision_tree(empty_revid))],
884
entry = state._get_entry(0, path_utf8='a')
885
self.build_tree(['a'])
886
# Add one where we don't provide the stat or sha already
887
self.assertEqual(('', 'a', 'a-id'), entry[0])
888
self.assertEqual(('f', '', 0, False, dirstate.DirState.NULLSTAT),
890
# Flush the buffers to disk
892
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
893
state._dirblock_state)
895
stat_value = os.lstat('a')
896
packed_stat = dirstate.pack_stat(stat_value)
897
link_or_sha1 = self.update_entry(state, entry, abspath='a',
898
stat_value=stat_value)
899
self.assertEqual(None, link_or_sha1)
901
# The dirblock entry should not have cached the file's sha1 (too new)
902
self.assertEqual(('f', '', 14, False, dirstate.DirState.NULLSTAT),
904
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
905
state._dirblock_state)
906
mode = stat_value.st_mode
907
self.assertEqual([('is_exec', mode, False)], state._log)
910
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
911
state._dirblock_state)
913
# If we do it again right away, we don't know if the file has changed
914
# so we will re-read the file. Roll the clock back so the file is
915
# guaranteed to look too new.
916
state.adjust_time(-10)
919
link_or_sha1 = self.update_entry(state, entry, abspath='a',
920
stat_value=stat_value)
921
self.assertEqual([('is_exec', mode, False)], state._log)
922
self.assertEqual(None, link_or_sha1)
923
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
924
state._dirblock_state)
925
self.assertEqual(('f', '', 14, False, dirstate.DirState.NULLSTAT),
929
# If it is cachable (the clock has moved forward) but new it still
930
# won't calculate the sha or cache it.
931
state.adjust_time(+20)
933
link_or_sha1 = dirstate.update_entry(state, entry, abspath='a',
934
stat_value=stat_value)
935
self.assertEqual(None, link_or_sha1)
936
self.assertEqual([('is_exec', mode, False)], state._log)
937
self.assertEqual(('f', '', 14, False, dirstate.DirState.NULLSTAT),
940
# If the file is no longer new, and the clock has been moved forward
941
# sufficiently, it will cache the sha.
943
state.set_parent_trees(
944
[(with_a_id, tree.branch.repository.revision_tree(with_a_id))],
946
entry = state._get_entry(0, path_utf8='a')
948
link_or_sha1 = self.update_entry(state, entry, abspath='a',
949
stat_value=stat_value)
950
self.assertEqual('b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6',
952
self.assertEqual([('is_exec', mode, False), ('sha1', 'a')],
954
self.assertEqual(('f', link_or_sha1, 14, False, packed_stat),
957
# Subsequent calls will just return the cached value
959
link_or_sha1 = self.update_entry(state, entry, abspath='a',
960
stat_value=stat_value)
961
self.assertEqual('b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6',
963
self.assertEqual([], state._log)
964
self.assertEqual(('f', link_or_sha1, 14, False, packed_stat),
967
def test_update_entry_symlink(self):
968
"""Update entry should read symlinks."""
969
self.requireFeature(tests.SymlinkFeature)
970
state, entry = self.get_state_with_a()
972
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
973
state._dirblock_state)
974
os.symlink('target', 'a')
976
state.adjust_time(-10) # Make the symlink look new
977
stat_value = os.lstat('a')
978
packed_stat = dirstate.pack_stat(stat_value)
979
link_or_sha1 = self.update_entry(state, entry, abspath='a',
980
stat_value=stat_value)
981
self.assertEqual('target', link_or_sha1)
982
self.assertEqual([('read_link', 'a', '')], state._log)
983
# Dirblock is not updated (the link is too new)
984
self.assertEqual([('l', '', 6, False, dirstate.DirState.NULLSTAT)],
986
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
987
state._dirblock_state)
989
# Because the stat_value looks new, we should re-read the target
990
link_or_sha1 = self.update_entry(state, entry, abspath='a',
991
stat_value=stat_value)
992
self.assertEqual('target', link_or_sha1)
993
self.assertEqual([('read_link', 'a', ''),
994
('read_link', 'a', ''),
996
self.assertEqual([('l', '', 6, False, dirstate.DirState.NULLSTAT)],
998
state.adjust_time(+20) # Skip into the future, all files look old
999
link_or_sha1 = self.update_entry(state, entry, abspath='a',
1000
stat_value=stat_value)
1001
self.assertEqual('target', link_or_sha1)
1002
# We need to re-read the link because only now can we cache it
1003
self.assertEqual([('read_link', 'a', ''),
1004
('read_link', 'a', ''),
1005
('read_link', 'a', ''),
1007
self.assertEqual([('l', 'target', 6, False, packed_stat)],
1010
# Another call won't re-read the link
1011
self.assertEqual([('read_link', 'a', ''),
1012
('read_link', 'a', ''),
1013
('read_link', 'a', ''),
1015
link_or_sha1 = self.update_entry(state, entry, abspath='a',
1016
stat_value=stat_value)
1017
self.assertEqual('target', link_or_sha1)
1018
self.assertEqual([('l', 'target', 6, False, packed_stat)],
1021
def do_update_entry(self, state, entry, abspath):
1022
stat_value = os.lstat(abspath)
1023
return self.update_entry(state, entry, abspath, stat_value)
1025
def test_update_entry_dir(self):
1026
state, entry = self.get_state_with_a()
1027
self.build_tree(['a/'])
1028
self.assertIs(None, self.do_update_entry(state, entry, 'a'))
1030
def test_update_entry_dir_unchanged(self):
1031
state, entry = self.get_state_with_a()
1032
self.build_tree(['a/'])
1033
state.adjust_time(+20)
1034
self.assertIs(None, self.do_update_entry(state, entry, 'a'))
1035
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
1036
state._dirblock_state)
1038
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
1039
state._dirblock_state)
1040
self.assertIs(None, self.do_update_entry(state, entry, 'a'))
1041
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
1042
state._dirblock_state)
1044
def test_update_entry_file_unchanged(self):
1045
state, _ = self.get_state_with_a()
1046
tree = self.make_branch_and_tree('tree')
1048
self.build_tree(['tree/a'])
1049
tree.add(['a'], ['a-id'])
1050
with_a_id = tree.commit('witha')
1051
self.addCleanup(tree.unlock)
1052
state.set_parent_trees(
1053
[(with_a_id, tree.branch.repository.revision_tree(with_a_id))],
1055
entry = state._get_entry(0, path_utf8='a')
1056
self.build_tree(['a'])
1057
sha1sum = 'b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6'
1058
state.adjust_time(+20)
1059
self.assertEqual(sha1sum, self.do_update_entry(state, entry, 'a'))
1060
self.assertEqual(dirstate.DirState.IN_MEMORY_MODIFIED,
1061
state._dirblock_state)
1063
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
1064
state._dirblock_state)
1065
self.assertEqual(sha1sum, self.do_update_entry(state, entry, 'a'))
1066
self.assertEqual(dirstate.DirState.IN_MEMORY_UNMODIFIED,
1067
state._dirblock_state)
1069
def test_update_entry_tree_reference(self):
1070
state = test_dirstate.InstrumentedDirState.initialize('dirstate')
1071
self.addCleanup(state.unlock)
1072
state.add('r', 'r-id', 'tree-reference', None, '')
1073
self.build_tree(['r/'])
1074
entry = state._get_entry(0, path_utf8='r')
1075
self.do_update_entry(state, entry, 'r')
1076
entry = state._get_entry(0, path_utf8='r')
1077
self.assertEqual('t', entry[1][0][0])
1079
def create_and_test_file(self, state, entry):
1080
"""Create a file at 'a' and verify the state finds it during update.
1082
The state should already be versioning *something* at 'a'. This makes
1083
sure that state.update_entry recognizes it as a file.
1085
self.build_tree(['a'])
1086
stat_value = os.lstat('a')
1087
packed_stat = dirstate.pack_stat(stat_value)
1089
link_or_sha1 = self.do_update_entry(state, entry, abspath='a')
1090
self.assertEqual(None, link_or_sha1)
1091
self.assertEqual([('f', '', 14, False, dirstate.DirState.NULLSTAT)],
1095
def create_and_test_dir(self, state, entry):
1096
"""Create a directory at 'a' and verify the state finds it.
1098
The state should already be versioning *something* at 'a'. This makes
1099
sure that state.update_entry recognizes it as a directory.
1101
self.build_tree(['a/'])
1102
stat_value = os.lstat('a')
1103
packed_stat = dirstate.pack_stat(stat_value)
1105
link_or_sha1 = self.do_update_entry(state, entry, abspath='a')
1106
self.assertIs(None, link_or_sha1)
1107
self.assertEqual([('d', '', 0, False, packed_stat)], entry[1])
1111
# FIXME: Add unicode version
1112
def create_and_test_symlink(self, state, entry):
1113
"""Create a symlink at 'a' and verify the state finds it.
1115
The state should already be versioning *something* at 'a'. This makes
1116
sure that state.update_entry recognizes it as a symlink.
1118
This should not be called if this platform does not have symlink
1121
# caller should care about skipping test on platforms without symlinks
1122
os.symlink('path/to/foo', 'a')
1124
stat_value = os.lstat('a')
1125
packed_stat = dirstate.pack_stat(stat_value)
1127
link_or_sha1 = self.do_update_entry(state, entry, abspath='a')
1128
self.assertEqual('path/to/foo', link_or_sha1)
1129
self.assertEqual([('l', 'path/to/foo', 11, False, packed_stat)],
1133
def test_update_file_to_dir(self):
1134
"""If a file changes to a directory we return None for the sha.
1135
We also update the inventory record.
1137
state, entry = self.get_state_with_a()
1138
# The file sha1 won't be cached unless the file is old
1139
state.adjust_time(+10)
1140
self.create_and_test_file(state, entry)
1142
self.create_and_test_dir(state, entry)
1144
def test_update_file_to_symlink(self):
1145
"""File becomes a symlink"""
1146
self.requireFeature(tests.SymlinkFeature)
1147
state, entry = self.get_state_with_a()
1148
# The file sha1 won't be cached unless the file is old
1149
state.adjust_time(+10)
1150
self.create_and_test_file(state, entry)
1152
self.create_and_test_symlink(state, entry)
1154
def test_update_dir_to_file(self):
1155
"""Directory becoming a file updates the entry."""
1156
state, entry = self.get_state_with_a()
1157
# The file sha1 won't be cached unless the file is old
1158
state.adjust_time(+10)
1159
self.create_and_test_dir(state, entry)
1161
self.create_and_test_file(state, entry)
1163
def test_update_dir_to_symlink(self):
1164
"""Directory becomes a symlink"""
1165
self.requireFeature(tests.SymlinkFeature)
1166
state, entry = self.get_state_with_a()
1167
# The symlink target won't be cached if it isn't old
1168
state.adjust_time(+10)
1169
self.create_and_test_dir(state, entry)
1171
self.create_and_test_symlink(state, entry)
1173
def test_update_symlink_to_file(self):
1174
"""Symlink becomes a file"""
1175
self.requireFeature(tests.SymlinkFeature)
1176
state, entry = self.get_state_with_a()
1177
# The symlink and file info won't be cached unless old
1178
state.adjust_time(+10)
1179
self.create_and_test_symlink(state, entry)
1181
self.create_and_test_file(state, entry)
1183
def test_update_symlink_to_dir(self):
1184
"""Symlink becomes a directory"""
1185
self.requireFeature(tests.SymlinkFeature)
1186
state, entry = self.get_state_with_a()
1187
# The symlink target won't be cached if it isn't old
1188
state.adjust_time(+10)
1189
self.create_and_test_symlink(state, entry)
1191
self.create_and_test_dir(state, entry)
1193
def test__is_executable_win32(self):
1194
state, entry = self.get_state_with_a()
1195
self.build_tree(['a'])
1197
# Make sure we are using the win32 implementation of _is_executable
1198
state._is_executable = state._is_executable_win32
1200
# The file on disk is not executable, but we are marking it as though
1201
# it is. With _is_executable_win32 we ignore what is on disk.
1202
entry[1][0] = ('f', '', 0, True, dirstate.DirState.NULLSTAT)
1204
stat_value = os.lstat('a')
1205
packed_stat = dirstate.pack_stat(stat_value)
1207
state.adjust_time(-10) # Make sure everything is new
1208
self.update_entry(state, entry, abspath='a', stat_value=stat_value)
1210
# The row is updated, but the executable bit stays set.
1211
self.assertEqual([('f', '', 14, True, dirstate.DirState.NULLSTAT)],
1214
# Make the disk object look old enough to cache (but it won't cache the
1215
# sha as it is a new file).
1216
state.adjust_time(+20)
1217
digest = 'b50e5406bb5e153ebbeb20268fcf37c87e1ecfb6'
1218
self.update_entry(state, entry, abspath='a', stat_value=stat_value)
1219
self.assertEqual([('f', '', 14, True, dirstate.DirState.NULLSTAT)],
1222
def _prepare_tree(self):
1224
text = 'Hello World\n'
1225
tree = self.make_branch_and_tree('tree')
1226
self.build_tree_contents([('tree/a file', text)])
1227
tree.add('a file', 'a-file-id')
1228
# Note: dirstate does not sha prior to the first commit
1229
# so commit now in order for the test to work
1230
tree.commit('first')
1233
def test_sha1provider_sha1_used(self):
1234
tree, text = self._prepare_tree()
1235
state = dirstate.DirState.from_tree(tree, 'dirstate',
1236
UppercaseSHA1Provider())
1237
self.addCleanup(state.unlock)
1238
expected_sha = osutils.sha_string(text.upper() + "foo")
1239
entry = state._get_entry(0, path_utf8='a file')
1240
state._sha_cutoff_time()
1241
state._cutoff_time += 10
1242
sha1 = self.update_entry(state, entry, 'tree/a file',
1243
os.lstat('tree/a file'))
1244
self.assertEqual(expected_sha, sha1)
1246
def test_sha1provider_stat_and_sha1_used(self):
1247
tree, text = self._prepare_tree()
1249
self.addCleanup(tree.unlock)
1250
state = tree._current_dirstate()
1251
state._sha1_provider = UppercaseSHA1Provider()
1252
# If we used the standard provider, it would look like nothing has
1254
file_ids_changed = [change[0] for change
1255
in tree.iter_changes(tree.basis_tree())]
1256
self.assertEqual(['a-file-id'], file_ids_changed)
1259
class UppercaseSHA1Provider(dirstate.SHA1Provider):
1260
"""A custom SHA1Provider."""
1262
def sha1(self, abspath):
1263
return self.stat_and_sha1(abspath)[1]
1265
def stat_and_sha1(self, abspath):
1266
file_obj = file(abspath, 'rb')
1268
statvalue = os.fstat(file_obj.fileno())
1269
text = ''.join(file_obj.readlines())
1270
sha1 = osutils.sha_string(text.upper() + "foo")
1273
return statvalue, sha1
1276
class TestProcessEntry(test_dirstate.TestCaseWithDirState):
1279
_process_entry = None
1282
super(TestProcessEntry, self).setUp()
1283
orig = dirstate._process_entry
1285
dirstate._process_entry = orig
1286
self.addCleanup(cleanup)
1287
dirstate._process_entry = self._process_entry
1289
def assertChangedFileIds(self, expected, tree):
1292
file_ids = [info[0] for info
1293
in tree.iter_changes(tree.basis_tree())]
1296
self.assertEqual(sorted(expected), sorted(file_ids))
1298
def test_simple_changes(self):
1299
tree = self.make_branch_and_tree('tree')
1300
self.build_tree(['tree/file'])
1301
tree.add(['file'], ['file-id'])
1302
self.assertChangedFileIds([tree.get_root_id(), 'file-id'], tree)
1304
self.assertChangedFileIds([], tree)
1306
def test_sha1provider_stat_and_sha1_used(self):
1307
tree = self.make_branch_and_tree('tree')
1308
self.build_tree(['tree/file'])
1309
tree.add(['file'], ['file-id'])
1312
self.addCleanup(tree.unlock)
1313
state = tree._current_dirstate()
1314
state._sha1_provider = UppercaseSHA1Provider()
1315
self.assertChangedFileIds(['file-id'], tree)
added
removed
bzrlib/tests/test__dirstate_helpers.py
