318
351
self.assertEqual("@", osutils.kind_marker("symlink"))
319
352
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
354
def test_host_os_dereferences_symlinks(self):
355
osutils.host_os_dereferences_symlinks()
358
class TestCanonicalRelPath(TestCaseInTempDir):
360
_test_needs_features = [CaseInsCasePresFilenameFeature]
362
def test_canonical_relpath_simple(self):
363
f = file('MixedCaseName', 'w')
365
self.failUnlessEqual(
366
canonical_relpath(self.test_base_dir, 'mixedcasename'),
367
'work/MixedCaseName')
369
def test_canonical_relpath_missing_tail(self):
370
os.mkdir('MixedCaseParent')
371
self.failUnlessEqual(
372
canonical_relpath(self.test_base_dir, 'mixedcaseparent/nochild'),
373
'work/MixedCaseParent/nochild')
376
class TestPumpFile(TestCase):
377
"""Test pumpfile method."""
379
# create a test datablock
380
self.block_size = 512
381
pattern = '0123456789ABCDEF'
382
self.test_data = pattern * (3 * self.block_size / len(pattern))
383
self.test_data_len = len(self.test_data)
385
def test_bracket_block_size(self):
386
"""Read data in blocks with the requested read size bracketing the
388
# make sure test data is larger than max read size
389
self.assertTrue(self.test_data_len > self.block_size)
391
from_file = FakeReadFile(self.test_data)
394
# read (max / 2) bytes and verify read size wasn't affected
395
num_bytes_to_read = self.block_size / 2
396
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
397
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
398
self.assertEqual(from_file.get_read_count(), 1)
400
# read (max) bytes and verify read size wasn't affected
401
num_bytes_to_read = self.block_size
402
from_file.reset_read_count()
403
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
404
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
405
self.assertEqual(from_file.get_read_count(), 1)
407
# read (max + 1) bytes and verify read size was limited
408
num_bytes_to_read = self.block_size + 1
409
from_file.reset_read_count()
410
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
411
self.assertEqual(from_file.get_max_read_size(), self.block_size)
412
self.assertEqual(from_file.get_read_count(), 2)
414
# finish reading the rest of the data
415
num_bytes_to_read = self.test_data_len - to_file.tell()
416
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
418
# report error if the data wasn't equal (we only report the size due
419
# to the length of the data)
420
response_data = to_file.getvalue()
421
if response_data != self.test_data:
422
message = "Data not equal. Expected %d bytes, received %d."
423
self.fail(message % (len(response_data), self.test_data_len))
425
def test_specified_size(self):
426
"""Request a transfer larger than the maximum block size and verify
427
that the maximum read doesn't exceed the block_size."""
428
# make sure test data is larger than max read size
429
self.assertTrue(self.test_data_len > self.block_size)
431
# retrieve data in blocks
432
from_file = FakeReadFile(self.test_data)
434
pumpfile(from_file, to_file, self.test_data_len, self.block_size)
436
# verify read size was equal to the maximum read size
437
self.assertTrue(from_file.get_max_read_size() > 0)
438
self.assertEqual(from_file.get_max_read_size(), self.block_size)
439
self.assertEqual(from_file.get_read_count(), 3)
441
# report error if the data wasn't equal (we only report the size due
442
# to the length of the data)
443
response_data = to_file.getvalue()
444
if response_data != self.test_data:
445
message = "Data not equal. Expected %d bytes, received %d."
446
self.fail(message % (len(response_data), self.test_data_len))
448
def test_to_eof(self):
449
"""Read to end-of-file and verify that the reads are not larger than
450
the maximum read size."""
451
# make sure test data is larger than max read size
452
self.assertTrue(self.test_data_len > self.block_size)
454
# retrieve data to EOF
455
from_file = FakeReadFile(self.test_data)
457
pumpfile(from_file, to_file, -1, self.block_size)
459
# verify read size was equal to the maximum read size
460
self.assertEqual(from_file.get_max_read_size(), self.block_size)
461
self.assertEqual(from_file.get_read_count(), 4)
463
# report error if the data wasn't equal (we only report the size due
464
# to the length of the data)
465
response_data = to_file.getvalue()
466
if response_data != self.test_data:
467
message = "Data not equal. Expected %d bytes, received %d."
468
self.fail(message % (len(response_data), self.test_data_len))
470
def test_defaults(self):
471
"""Verifies that the default arguments will read to EOF -- this
472
test verifies that any existing usages of pumpfile will not be broken
473
with this new version."""
474
# retrieve data using default (old) pumpfile method
475
from_file = FakeReadFile(self.test_data)
477
pumpfile(from_file, to_file)
479
# report error if the data wasn't equal (we only report the size due
480
# to the length of the data)
481
response_data = to_file.getvalue()
482
if response_data != self.test_data:
483
message = "Data not equal. Expected %d bytes, received %d."
484
self.fail(message % (len(response_data), self.test_data_len))
486
def test_report_activity(self):
488
def log_activity(length, direction):
489
activity.append((length, direction))
490
from_file = StringIO(self.test_data)
492
pumpfile(from_file, to_file, buff_size=500,
493
report_activity=log_activity, direction='read')
494
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
495
(36, 'read')], activity)
497
from_file = StringIO(self.test_data)
500
pumpfile(from_file, to_file, buff_size=500,
501
report_activity=log_activity, direction='write')
502
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
503
(36, 'write')], activity)
505
# And with a limited amount of data
506
from_file = StringIO(self.test_data)
509
pumpfile(from_file, to_file, buff_size=500, read_length=1028,
510
report_activity=log_activity, direction='read')
511
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
515
class TestPumpStringFile(TestCase):
517
def test_empty(self):
519
pump_string_file("", output)
520
self.assertEqual("", output.getvalue())
522
def test_more_than_segment_size(self):
524
pump_string_file("123456789", output, 2)
525
self.assertEqual("123456789", output.getvalue())
527
def test_segment_size(self):
529
pump_string_file("12", output, 2)
530
self.assertEqual("12", output.getvalue())
532
def test_segment_size_multiple(self):
534
pump_string_file("1234", output, 2)
535
self.assertEqual("1234", output.getvalue())
322
538
class TestSafeUnicode(TestCase):
701
954
new_dirblock.append((info[0], info[1], info[2], info[4]))
702
955
dirblock[:] = new_dirblock
957
def _save_platform_info(self):
958
cur_winver = win32utils.winver
959
cur_fs_enc = osutils._fs_enc
960
cur_dir_reader = osutils._selected_dir_reader
962
win32utils.winver = cur_winver
963
osutils._fs_enc = cur_fs_enc
964
osutils._selected_dir_reader = cur_dir_reader
965
self.addCleanup(restore)
967
def assertReadFSDirIs(self, expected):
968
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
969
# Force it to redetect
970
osutils._selected_dir_reader = None
971
# Nothing to list, but should still trigger the selection logic
972
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
973
self.assertIsInstance(osutils._selected_dir_reader, expected)
975
def test_force_walkdirs_utf8_fs_utf8(self):
976
self.requireFeature(UTF8DirReaderFeature)
977
self._save_platform_info()
978
win32utils.winver = None # Avoid the win32 detection code
979
osutils._fs_enc = 'UTF-8'
980
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
982
def test_force_walkdirs_utf8_fs_ascii(self):
983
self.requireFeature(UTF8DirReaderFeature)
984
self._save_platform_info()
985
win32utils.winver = None # Avoid the win32 detection code
986
osutils._fs_enc = 'US-ASCII'
987
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
989
def test_force_walkdirs_utf8_fs_ANSI(self):
990
self.requireFeature(UTF8DirReaderFeature)
991
self._save_platform_info()
992
win32utils.winver = None # Avoid the win32 detection code
993
osutils._fs_enc = 'ANSI_X3.4-1968'
994
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
996
def test_force_walkdirs_utf8_fs_latin1(self):
997
self._save_platform_info()
998
win32utils.winver = None # Avoid the win32 detection code
999
osutils._fs_enc = 'latin1'
1000
self.assertReadFSDirIs(osutils.UnicodeDirReader)
1002
def test_force_walkdirs_utf8_nt(self):
1003
# Disabled because the thunk of the whole walkdirs api is disabled.
1004
self.requireFeature(Win32ReadDirFeature)
1005
self._save_platform_info()
1006
win32utils.winver = 'Windows NT'
1007
from bzrlib._walkdirs_win32 import Win32ReadDir
1008
self.assertReadFSDirIs(Win32ReadDir)
1010
def test_force_walkdirs_utf8_98(self):
1011
self.requireFeature(Win32ReadDirFeature)
1012
self._save_platform_info()
1013
win32utils.winver = 'Windows 98'
1014
self.assertReadFSDirIs(osutils.UnicodeDirReader)
704
1016
def test_unicode_walkdirs(self):
705
1017
"""Walkdirs should always return unicode paths."""
706
1018
name0 = u'0file-\xb6'
856
result = list(osutils._walkdirs_unicode_to_utf8('.'))
857
self._filter_out_stat(result)
858
self.assertEqual(expected_dirblocks, result)
1172
result = list(osutils._walkdirs_utf8('.'))
1173
self._filter_out_stat(result)
1174
self.assertEqual(expected_dirblocks, result)
1176
def test__walkdirs_utf8_win32readdir(self):
1177
self.requireFeature(Win32ReadDirFeature)
1178
self.requireFeature(tests.UnicodeFilenameFeature)
1179
from bzrlib._walkdirs_win32 import Win32ReadDir
1180
self._save_platform_info()
1181
osutils._selected_dir_reader = Win32ReadDir()
1182
name0u = u'0file-\xb6'
1183
name1u = u'1dir-\u062c\u0648'
1184
name2u = u'2file-\u0633'
1188
name1u + '/' + name0u,
1189
name1u + '/' + name1u + '/',
1192
self.build_tree(tree)
1193
name0 = name0u.encode('utf8')
1194
name1 = name1u.encode('utf8')
1195
name2 = name2u.encode('utf8')
1197
# All of the abspaths should be in unicode, all of the relative paths
1199
expected_dirblocks = [
1201
[(name0, name0, 'file', './' + name0u),
1202
(name1, name1, 'directory', './' + name1u),
1203
(name2, name2, 'file', './' + name2u),
1206
((name1, './' + name1u),
1207
[(name1 + '/' + name0, name0, 'file', './' + name1u
1209
(name1 + '/' + name1, name1, 'directory', './' + name1u
1213
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1218
result = list(osutils._walkdirs_utf8(u'.'))
1219
self._filter_out_stat(result)
1220
self.assertEqual(expected_dirblocks, result)
1222
def assertStatIsCorrect(self, path, win32stat):
1223
os_stat = os.stat(path)
1224
self.assertEqual(os_stat.st_size, win32stat.st_size)
1225
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1226
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1227
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1228
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1229
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1230
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1232
def test__walkdirs_utf_win32_find_file_stat_file(self):
1233
"""make sure our Stat values are valid"""
1234
self.requireFeature(Win32ReadDirFeature)
1235
self.requireFeature(tests.UnicodeFilenameFeature)
1236
from bzrlib._walkdirs_win32 import Win32ReadDir
1237
name0u = u'0file-\xb6'
1238
name0 = name0u.encode('utf8')
1239
self.build_tree([name0u])
1240
# I hate to sleep() here, but I'm trying to make the ctime different
1243
f = open(name0u, 'ab')
1245
f.write('just a small update')
1249
result = Win32ReadDir().read_dir('', u'.')
1251
self.assertEqual((name0, name0, 'file'), entry[:3])
1252
self.assertEqual(u'./' + name0u, entry[4])
1253
self.assertStatIsCorrect(entry[4], entry[3])
1254
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1256
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1257
"""make sure our Stat values are valid"""
1258
self.requireFeature(Win32ReadDirFeature)
1259
self.requireFeature(tests.UnicodeFilenameFeature)
1260
from bzrlib._walkdirs_win32 import Win32ReadDir
1261
name0u = u'0dir-\u062c\u0648'
1262
name0 = name0u.encode('utf8')
1263
self.build_tree([name0u + '/'])
1265
result = Win32ReadDir().read_dir('', u'.')
1267
self.assertEqual((name0, name0, 'directory'), entry[:3])
1268
self.assertEqual(u'./' + name0u, entry[4])
1269
self.assertStatIsCorrect(entry[4], entry[3])
860
1271
def assertPathCompare(self, path_less, path_greater):
861
1272
"""check that path_less and path_greater compare correctly."""
1077
1488
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1081
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1083
# This program is free software; you can redistribute it and/or modify
1084
# it under the terms of the GNU General Public License as published by
1085
# the Free Software Foundation; either version 2 of the License, or
1086
# (at your option) any later version.
1088
# This program is distributed in the hope that it will be useful,
1089
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1090
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
1091
# GNU General Public License for more details.
1093
# You should have received a copy of the GNU General Public License
1094
# along with this program; if not, write to the Free Software
1095
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
1098
# NOTE: If update these, please also update the help for global-options in
1099
# bzrlib/help_topics/__init__.py
1102
"""Set of flags that enable different debug behaviour.
1104
These are set with eg ``-Dlock`` on the bzr command line.
1108
* auth - show authentication sections used
1109
* error - show stack traces for all top level exceptions
1110
* evil - capture call sites that do expensive or badly-scaling operations.
1111
* fetch - trace history copying between repositories
1112
* graph - trace graph traversal information
1113
* hashcache - log every time a working file is read to determine its hash
1114
* hooks - trace hook execution
1115
* hpss - trace smart protocol requests and responses
1116
* http - trace http connections, requests and responses
1117
* index - trace major index operations
1118
* knit - trace knit operations
1119
* lock - trace when lockdir locks are taken or released
1120
* merge - emit information for debugging merges
1121
* pack - emit information about pack operations
1127
1491
class TestResourceLoading(TestCaseInTempDir):
1129
1493
def test_resource_string(self):
1130
1494
# test resource in bzrlib
1131
1495
text = osutils.resource_string('bzrlib', 'debug.py')
1132
self.assertEquals(_debug_text, text)
1496
self.assertContainsRe(text, "debug_flags = set()")
1133
1497
# test resource under bzrlib
1134
1498
text = osutils.resource_string('bzrlib.ui', 'text.py')
1135
1499
self.assertContainsRe(text, "class TextUIFactory")