346
268
osutils.make_readonly('dangling')
347
269
osutils.make_writable('dangling')
349
272
def test_kind_marker(self):
350
273
self.assertEqual("", osutils.kind_marker("file"))
351
274
self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
352
275
self.assertEqual("@", osutils.kind_marker("symlink"))
353
276
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
355
def test_host_os_dereferences_symlinks(self):
356
osutils.host_os_dereferences_symlinks()
359
class TestCanonicalRelPath(TestCaseInTempDir):
361
_test_needs_features = [CaseInsCasePresFilenameFeature]
363
def test_canonical_relpath_simple(self):
364
f = file('MixedCaseName', 'w')
366
self.failUnlessEqual(
367
canonical_relpath(self.test_base_dir, 'mixedcasename'),
368
'work/MixedCaseName')
370
def test_canonical_relpath_missing_tail(self):
371
os.mkdir('MixedCaseParent')
372
self.failUnlessEqual(
373
canonical_relpath(self.test_base_dir, 'mixedcaseparent/nochild'),
374
'work/MixedCaseParent/nochild')
377
class TestPumpFile(TestCase):
378
"""Test pumpfile method."""
381
# create a test datablock
382
self.block_size = 512
383
pattern = '0123456789ABCDEF'
384
self.test_data = pattern * (3 * self.block_size / len(pattern))
385
self.test_data_len = len(self.test_data)
387
def test_bracket_block_size(self):
388
"""Read data in blocks with the requested read size bracketing the
390
# make sure test data is larger than max read size
391
self.assertTrue(self.test_data_len > self.block_size)
393
from_file = FakeReadFile(self.test_data)
396
# read (max / 2) bytes and verify read size wasn't affected
397
num_bytes_to_read = self.block_size / 2
398
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
399
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
400
self.assertEqual(from_file.get_read_count(), 1)
402
# read (max) bytes and verify read size wasn't affected
403
num_bytes_to_read = self.block_size
404
from_file.reset_read_count()
405
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
406
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
407
self.assertEqual(from_file.get_read_count(), 1)
409
# read (max + 1) bytes and verify read size was limited
410
num_bytes_to_read = self.block_size + 1
411
from_file.reset_read_count()
412
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
413
self.assertEqual(from_file.get_max_read_size(), self.block_size)
414
self.assertEqual(from_file.get_read_count(), 2)
416
# finish reading the rest of the data
417
num_bytes_to_read = self.test_data_len - to_file.tell()
418
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
420
# report error if the data wasn't equal (we only report the size due
421
# to the length of the data)
422
response_data = to_file.getvalue()
423
if response_data != self.test_data:
424
message = "Data not equal. Expected %d bytes, received %d."
425
self.fail(message % (len(response_data), self.test_data_len))
427
def test_specified_size(self):
428
"""Request a transfer larger than the maximum block size and verify
429
that the maximum read doesn't exceed the block_size."""
430
# make sure test data is larger than max read size
431
self.assertTrue(self.test_data_len > self.block_size)
433
# retrieve data in blocks
434
from_file = FakeReadFile(self.test_data)
436
pumpfile(from_file, to_file, self.test_data_len, self.block_size)
438
# verify read size was equal to the maximum read size
439
self.assertTrue(from_file.get_max_read_size() > 0)
440
self.assertEqual(from_file.get_max_read_size(), self.block_size)
441
self.assertEqual(from_file.get_read_count(), 3)
443
# report error if the data wasn't equal (we only report the size due
444
# to the length of the data)
445
response_data = to_file.getvalue()
446
if response_data != self.test_data:
447
message = "Data not equal. Expected %d bytes, received %d."
448
self.fail(message % (len(response_data), self.test_data_len))
450
def test_to_eof(self):
451
"""Read to end-of-file and verify that the reads are not larger than
452
the maximum read size."""
453
# make sure test data is larger than max read size
454
self.assertTrue(self.test_data_len > self.block_size)
456
# retrieve data to EOF
457
from_file = FakeReadFile(self.test_data)
459
pumpfile(from_file, to_file, -1, self.block_size)
461
# verify read size was equal to the maximum read size
462
self.assertEqual(from_file.get_max_read_size(), self.block_size)
463
self.assertEqual(from_file.get_read_count(), 4)
465
# report error if the data wasn't equal (we only report the size due
466
# to the length of the data)
467
response_data = to_file.getvalue()
468
if response_data != self.test_data:
469
message = "Data not equal. Expected %d bytes, received %d."
470
self.fail(message % (len(response_data), self.test_data_len))
472
def test_defaults(self):
473
"""Verifies that the default arguments will read to EOF -- this
474
test verifies that any existing usages of pumpfile will not be broken
475
with this new version."""
476
# retrieve data using default (old) pumpfile method
477
from_file = FakeReadFile(self.test_data)
479
pumpfile(from_file, to_file)
481
# report error if the data wasn't equal (we only report the size due
482
# to the length of the data)
483
response_data = to_file.getvalue()
484
if response_data != self.test_data:
485
message = "Data not equal. Expected %d bytes, received %d."
486
self.fail(message % (len(response_data), self.test_data_len))
488
def test_report_activity(self):
490
def log_activity(length, direction):
491
activity.append((length, direction))
492
from_file = StringIO(self.test_data)
494
pumpfile(from_file, to_file, buff_size=500,
495
report_activity=log_activity, direction='read')
496
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
497
(36, 'read')], activity)
499
from_file = StringIO(self.test_data)
502
pumpfile(from_file, to_file, buff_size=500,
503
report_activity=log_activity, direction='write')
504
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
505
(36, 'write')], activity)
507
# And with a limited amount of data
508
from_file = StringIO(self.test_data)
511
pumpfile(from_file, to_file, buff_size=500, read_length=1028,
512
report_activity=log_activity, direction='read')
513
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
517
class TestPumpStringFile(TestCase):
519
def test_empty(self):
521
pump_string_file("", output)
522
self.assertEqual("", output.getvalue())
524
def test_more_than_segment_size(self):
526
pump_string_file("123456789", output, 2)
527
self.assertEqual("123456789", output.getvalue())
529
def test_segment_size(self):
531
pump_string_file("12", output, 2)
532
self.assertEqual("12", output.getvalue())
534
def test_segment_size_multiple(self):
536
pump_string_file("1234", output, 2)
537
self.assertEqual("1234", output.getvalue())
540
279
class TestSafeUnicode(TestCase):
956
647
new_dirblock.append((info[0], info[1], info[2], info[4]))
957
648
dirblock[:] = new_dirblock
959
def _save_platform_info(self):
960
cur_winver = win32utils.winver
961
cur_fs_enc = osutils._fs_enc
962
cur_dir_reader = osutils._selected_dir_reader
964
win32utils.winver = cur_winver
965
osutils._fs_enc = cur_fs_enc
966
osutils._selected_dir_reader = cur_dir_reader
967
self.addCleanup(restore)
969
def assertReadFSDirIs(self, expected):
970
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
971
# Force it to redetect
972
osutils._selected_dir_reader = None
973
# Nothing to list, but should still trigger the selection logic
974
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
975
self.assertIsInstance(osutils._selected_dir_reader, expected)
977
def test_force_walkdirs_utf8_fs_utf8(self):
978
self.requireFeature(UTF8DirReaderFeature)
979
self._save_platform_info()
980
win32utils.winver = None # Avoid the win32 detection code
981
osutils._fs_enc = 'UTF-8'
982
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
984
def test_force_walkdirs_utf8_fs_ascii(self):
985
self.requireFeature(UTF8DirReaderFeature)
986
self._save_platform_info()
987
win32utils.winver = None # Avoid the win32 detection code
988
osutils._fs_enc = 'US-ASCII'
989
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
991
def test_force_walkdirs_utf8_fs_ANSI(self):
992
self.requireFeature(UTF8DirReaderFeature)
993
self._save_platform_info()
994
win32utils.winver = None # Avoid the win32 detection code
995
osutils._fs_enc = 'ANSI_X3.4-1968'
996
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
998
def test_force_walkdirs_utf8_fs_latin1(self):
999
self._save_platform_info()
1000
win32utils.winver = None # Avoid the win32 detection code
1001
osutils._fs_enc = 'latin1'
1002
self.assertReadFSDirIs(osutils.UnicodeDirReader)
1004
def test_force_walkdirs_utf8_nt(self):
1005
# Disabled because the thunk of the whole walkdirs api is disabled.
1006
self.requireFeature(Win32ReadDirFeature)
1007
self._save_platform_info()
1008
win32utils.winver = 'Windows NT'
1009
from bzrlib._walkdirs_win32 import Win32ReadDir
1010
self.assertReadFSDirIs(Win32ReadDir)
1012
def test_force_walkdirs_utf8_98(self):
1013
self.requireFeature(Win32ReadDirFeature)
1014
self._save_platform_info()
1015
win32utils.winver = 'Windows 98'
1016
self.assertReadFSDirIs(osutils.UnicodeDirReader)
1018
650
def test_unicode_walkdirs(self):
1019
651
"""Walkdirs should always return unicode paths."""
1020
652
name0 = u'0file-\xb6'
1174
result = list(osutils._walkdirs_utf8('.'))
1175
self._filter_out_stat(result)
1176
self.assertEqual(expected_dirblocks, result)
1178
def test__walkdirs_utf8_win32readdir(self):
1179
self.requireFeature(Win32ReadDirFeature)
1180
self.requireFeature(tests.UnicodeFilenameFeature)
1181
from bzrlib._walkdirs_win32 import Win32ReadDir
1182
self._save_platform_info()
1183
osutils._selected_dir_reader = Win32ReadDir()
1184
name0u = u'0file-\xb6'
1185
name1u = u'1dir-\u062c\u0648'
1186
name2u = u'2file-\u0633'
1190
name1u + '/' + name0u,
1191
name1u + '/' + name1u + '/',
1194
self.build_tree(tree)
1195
name0 = name0u.encode('utf8')
1196
name1 = name1u.encode('utf8')
1197
name2 = name2u.encode('utf8')
1199
# All of the abspaths should be in unicode, all of the relative paths
1201
expected_dirblocks = [
1203
[(name0, name0, 'file', './' + name0u),
1204
(name1, name1, 'directory', './' + name1u),
1205
(name2, name2, 'file', './' + name2u),
1208
((name1, './' + name1u),
1209
[(name1 + '/' + name0, name0, 'file', './' + name1u
1211
(name1 + '/' + name1, name1, 'directory', './' + name1u
1215
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1220
result = list(osutils._walkdirs_utf8(u'.'))
1221
self._filter_out_stat(result)
1222
self.assertEqual(expected_dirblocks, result)
1224
def assertStatIsCorrect(self, path, win32stat):
1225
os_stat = os.stat(path)
1226
self.assertEqual(os_stat.st_size, win32stat.st_size)
1227
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1228
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1229
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1230
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1231
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1232
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1234
def test__walkdirs_utf_win32_find_file_stat_file(self):
1235
"""make sure our Stat values are valid"""
1236
self.requireFeature(Win32ReadDirFeature)
1237
self.requireFeature(tests.UnicodeFilenameFeature)
1238
from bzrlib._walkdirs_win32 import Win32ReadDir
1239
name0u = u'0file-\xb6'
1240
name0 = name0u.encode('utf8')
1241
self.build_tree([name0u])
1242
# I hate to sleep() here, but I'm trying to make the ctime different
1245
f = open(name0u, 'ab')
1247
f.write('just a small update')
1251
result = Win32ReadDir().read_dir('', u'.')
1253
self.assertEqual((name0, name0, 'file'), entry[:3])
1254
self.assertEqual(u'./' + name0u, entry[4])
1255
self.assertStatIsCorrect(entry[4], entry[3])
1256
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1258
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1259
"""make sure our Stat values are valid"""
1260
self.requireFeature(Win32ReadDirFeature)
1261
self.requireFeature(tests.UnicodeFilenameFeature)
1262
from bzrlib._walkdirs_win32 import Win32ReadDir
1263
name0u = u'0dir-\u062c\u0648'
1264
name0 = name0u.encode('utf8')
1265
self.build_tree([name0u + '/'])
1267
result = Win32ReadDir().read_dir('', u'.')
1269
self.assertEqual((name0, name0, 'directory'), entry[:3])
1270
self.assertEqual(u'./' + name0u, entry[4])
1271
self.assertStatIsCorrect(entry[4], entry[3])
802
result = list(osutils._walkdirs_unicode_to_utf8('.'))
803
self._filter_out_stat(result)
804
self.assertEqual(expected_dirblocks, result)
1273
806
def assertPathCompare(self, path_less, path_greater):
1274
807
"""check that path_less and path_greater compare correctly."""
1474
1018
self.assertTrue(isinstance(offset, int))
1475
1019
eighteen_hours = 18 * 3600
1476
1020
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1479
class TestSizeShaFile(TestCaseInTempDir):
1481
def test_sha_empty(self):
1482
self.build_tree_contents([('foo', '')])
1483
expected_sha = osutils.sha_string('')
1485
self.addCleanup(f.close)
1486
size, sha = osutils.size_sha_file(f)
1487
self.assertEqual(0, size)
1488
self.assertEqual(expected_sha, sha)
1490
def test_sha_mixed_endings(self):
1491
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1492
self.build_tree_contents([('foo', text)])
1493
expected_sha = osutils.sha_string(text)
1495
self.addCleanup(f.close)
1496
size, sha = osutils.size_sha_file(f)
1497
self.assertEqual(38, size)
1498
self.assertEqual(expected_sha, sha)
1501
class TestShaFileByName(TestCaseInTempDir):
1503
def test_sha_empty(self):
1504
self.build_tree_contents([('foo', '')])
1505
expected_sha = osutils.sha_string('')
1506
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1508
def test_sha_mixed_endings(self):
1509
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1510
self.build_tree_contents([('foo', text)])
1511
expected_sha = osutils.sha_string(text)
1512
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1515
class TestResourceLoading(TestCaseInTempDir):
1517
def test_resource_string(self):
1518
# test resource in bzrlib
1519
text = osutils.resource_string('bzrlib', 'debug.py')
1520
self.assertContainsRe(text, "debug_flags = set()")
1521
# test resource under bzrlib
1522
text = osutils.resource_string('bzrlib.ui', 'text.py')
1523
self.assertContainsRe(text, "class TextUIFactory")
1524
# test unsupported package
1525
self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1527
# test unknown resource
1528
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1531
class TestReCompile(TestCase):
1533
def test_re_compile_checked(self):
1534
r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1535
self.assertTrue(r.match('aaaa'))
1536
self.assertTrue(r.match('aAaA'))
1538
def test_re_compile_checked_error(self):
1539
# like https://bugs.launchpad.net/bzr/+bug/251352
1540
err = self.assertRaises(
1541
errors.BzrCommandError,
1542
osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1544
"Invalid regular expression in test case: '*': "
1545
"nothing to repeat",