253
347
self.assertEqual("@", osutils.kind_marker("symlink"))
254
348
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
350
def test_host_os_dereferences_symlinks(self):
351
osutils.host_os_dereferences_symlinks()
354
class TestCanonicalRelPath(TestCaseInTempDir):
356
_test_needs_features = [CaseInsCasePresFilenameFeature]
358
def test_canonical_relpath_simple(self):
359
f = file('MixedCaseName', 'w')
361
# Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
362
real_base_dir = osutils.realpath(self.test_base_dir)
363
actual = osutils.canonical_relpath(real_base_dir, 'mixedcasename')
364
self.failUnlessEqual('work/MixedCaseName', actual)
366
def test_canonical_relpath_missing_tail(self):
367
os.mkdir('MixedCaseParent')
368
# Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
369
real_base_dir = osutils.realpath(self.test_base_dir)
370
actual = osutils.canonical_relpath(real_base_dir,
371
'mixedcaseparent/nochild')
372
self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
375
class TestPumpFile(TestCase):
376
"""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
osutils.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
osutils.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
osutils.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
osutils.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
osutils.pumpfile(from_file, to_file, self.test_data_len,
437
# verify read size was equal to the maximum read size
438
self.assertTrue(from_file.get_max_read_size() > 0)
439
self.assertEqual(from_file.get_max_read_size(), self.block_size)
440
self.assertEqual(from_file.get_read_count(), 3)
442
# report error if the data wasn't equal (we only report the size due
443
# to the length of the data)
444
response_data = to_file.getvalue()
445
if response_data != self.test_data:
446
message = "Data not equal. Expected %d bytes, received %d."
447
self.fail(message % (len(response_data), self.test_data_len))
449
def test_to_eof(self):
450
"""Read to end-of-file and verify that the reads are not larger than
451
the maximum read size."""
452
# make sure test data is larger than max read size
453
self.assertTrue(self.test_data_len > self.block_size)
455
# retrieve data to EOF
456
from_file = FakeReadFile(self.test_data)
458
osutils.pumpfile(from_file, to_file, -1, self.block_size)
460
# verify read size was equal to the maximum read size
461
self.assertEqual(from_file.get_max_read_size(), self.block_size)
462
self.assertEqual(from_file.get_read_count(), 4)
464
# report error if the data wasn't equal (we only report the size due
465
# to the length of the data)
466
response_data = to_file.getvalue()
467
if response_data != self.test_data:
468
message = "Data not equal. Expected %d bytes, received %d."
469
self.fail(message % (len(response_data), self.test_data_len))
471
def test_defaults(self):
472
"""Verifies that the default arguments will read to EOF -- this
473
test verifies that any existing usages of pumpfile will not be broken
474
with this new version."""
475
# retrieve data using default (old) pumpfile method
476
from_file = FakeReadFile(self.test_data)
478
osutils.pumpfile(from_file, to_file)
480
# report error if the data wasn't equal (we only report the size due
481
# to the length of the data)
482
response_data = to_file.getvalue()
483
if response_data != self.test_data:
484
message = "Data not equal. Expected %d bytes, received %d."
485
self.fail(message % (len(response_data), self.test_data_len))
487
def test_report_activity(self):
489
def log_activity(length, direction):
490
activity.append((length, direction))
491
from_file = StringIO(self.test_data)
493
osutils.pumpfile(from_file, to_file, buff_size=500,
494
report_activity=log_activity, direction='read')
495
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
496
(36, 'read')], activity)
498
from_file = StringIO(self.test_data)
501
osutils.pumpfile(from_file, to_file, buff_size=500,
502
report_activity=log_activity, direction='write')
503
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
504
(36, 'write')], activity)
506
# And with a limited amount of data
507
from_file = StringIO(self.test_data)
510
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
511
report_activity=log_activity, direction='read')
512
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
516
class TestPumpStringFile(TestCase):
518
def test_empty(self):
520
osutils.pump_string_file("", output)
521
self.assertEqual("", output.getvalue())
523
def test_more_than_segment_size(self):
525
osutils.pump_string_file("123456789", output, 2)
526
self.assertEqual("123456789", output.getvalue())
528
def test_segment_size(self):
530
osutils.pump_string_file("12", output, 2)
531
self.assertEqual("12", output.getvalue())
533
def test_segment_size_multiple(self):
535
osutils.pump_string_file("1234", output, 2)
536
self.assertEqual("1234", output.getvalue())
257
539
class TestSafeUnicode(TestCase):
625
955
new_dirblock.append((info[0], info[1], info[2], info[4]))
626
956
dirblock[:] = new_dirblock
958
def _save_platform_info(self):
959
cur_winver = win32utils.winver
960
cur_fs_enc = osutils._fs_enc
961
cur_dir_reader = osutils._selected_dir_reader
963
win32utils.winver = cur_winver
964
osutils._fs_enc = cur_fs_enc
965
osutils._selected_dir_reader = cur_dir_reader
966
self.addCleanup(restore)
968
def assertReadFSDirIs(self, expected):
969
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
970
# Force it to redetect
971
osutils._selected_dir_reader = None
972
# Nothing to list, but should still trigger the selection logic
973
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
974
self.assertIsInstance(osutils._selected_dir_reader, expected)
976
def test_force_walkdirs_utf8_fs_utf8(self):
977
self.requireFeature(UTF8DirReaderFeature)
978
self._save_platform_info()
979
win32utils.winver = None # Avoid the win32 detection code
980
osutils._fs_enc = 'UTF-8'
981
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
983
def test_force_walkdirs_utf8_fs_ascii(self):
984
self.requireFeature(UTF8DirReaderFeature)
985
self._save_platform_info()
986
win32utils.winver = None # Avoid the win32 detection code
987
osutils._fs_enc = 'US-ASCII'
988
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
990
def test_force_walkdirs_utf8_fs_ANSI(self):
991
self.requireFeature(UTF8DirReaderFeature)
992
self._save_platform_info()
993
win32utils.winver = None # Avoid the win32 detection code
994
osutils._fs_enc = 'ANSI_X3.4-1968'
995
self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
997
def test_force_walkdirs_utf8_fs_latin1(self):
998
self._save_platform_info()
999
win32utils.winver = None # Avoid the win32 detection code
1000
osutils._fs_enc = 'latin1'
1001
self.assertReadFSDirIs(osutils.UnicodeDirReader)
1003
def test_force_walkdirs_utf8_nt(self):
1004
# Disabled because the thunk of the whole walkdirs api is disabled.
1005
self.requireFeature(Win32ReadDirFeature)
1006
self._save_platform_info()
1007
win32utils.winver = 'Windows NT'
1008
from bzrlib._walkdirs_win32 import Win32ReadDir
1009
self.assertReadFSDirIs(Win32ReadDir)
1011
def test_force_walkdirs_utf8_98(self):
1012
self.requireFeature(Win32ReadDirFeature)
1013
self._save_platform_info()
1014
win32utils.winver = 'Windows 98'
1015
self.assertReadFSDirIs(osutils.UnicodeDirReader)
628
1017
def test_unicode_walkdirs(self):
629
1018
"""Walkdirs should always return unicode paths."""
630
1019
name0 = u'0file-\xb6'
780
result = list(osutils._walkdirs_unicode_to_utf8('.'))
781
self._filter_out_stat(result)
782
self.assertEqual(expected_dirblocks, result)
1173
result = list(osutils._walkdirs_utf8('.'))
1174
self._filter_out_stat(result)
1175
self.assertEqual(expected_dirblocks, result)
1177
def test__walkdirs_utf8_win32readdir(self):
1178
self.requireFeature(Win32ReadDirFeature)
1179
self.requireFeature(tests.UnicodeFilenameFeature)
1180
from bzrlib._walkdirs_win32 import Win32ReadDir
1181
self._save_platform_info()
1182
osutils._selected_dir_reader = Win32ReadDir()
1183
name0u = u'0file-\xb6'
1184
name1u = u'1dir-\u062c\u0648'
1185
name2u = u'2file-\u0633'
1189
name1u + '/' + name0u,
1190
name1u + '/' + name1u + '/',
1193
self.build_tree(tree)
1194
name0 = name0u.encode('utf8')
1195
name1 = name1u.encode('utf8')
1196
name2 = name2u.encode('utf8')
1198
# All of the abspaths should be in unicode, all of the relative paths
1200
expected_dirblocks = [
1202
[(name0, name0, 'file', './' + name0u),
1203
(name1, name1, 'directory', './' + name1u),
1204
(name2, name2, 'file', './' + name2u),
1207
((name1, './' + name1u),
1208
[(name1 + '/' + name0, name0, 'file', './' + name1u
1210
(name1 + '/' + name1, name1, 'directory', './' + name1u
1214
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1219
result = list(osutils._walkdirs_utf8(u'.'))
1220
self._filter_out_stat(result)
1221
self.assertEqual(expected_dirblocks, result)
1223
def assertStatIsCorrect(self, path, win32stat):
1224
os_stat = os.stat(path)
1225
self.assertEqual(os_stat.st_size, win32stat.st_size)
1226
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1227
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1228
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1229
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1230
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1231
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1233
def test__walkdirs_utf_win32_find_file_stat_file(self):
1234
"""make sure our Stat values are valid"""
1235
self.requireFeature(Win32ReadDirFeature)
1236
self.requireFeature(tests.UnicodeFilenameFeature)
1237
from bzrlib._walkdirs_win32 import Win32ReadDir
1238
name0u = u'0file-\xb6'
1239
name0 = name0u.encode('utf8')
1240
self.build_tree([name0u])
1241
# I hate to sleep() here, but I'm trying to make the ctime different
1244
f = open(name0u, 'ab')
1246
f.write('just a small update')
1250
result = Win32ReadDir().read_dir('', u'.')
1252
self.assertEqual((name0, name0, 'file'), entry[:3])
1253
self.assertEqual(u'./' + name0u, entry[4])
1254
self.assertStatIsCorrect(entry[4], entry[3])
1255
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1257
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1258
"""make sure our Stat values are valid"""
1259
self.requireFeature(Win32ReadDirFeature)
1260
self.requireFeature(tests.UnicodeFilenameFeature)
1261
from bzrlib._walkdirs_win32 import Win32ReadDir
1262
name0u = u'0dir-\u062c\u0648'
1263
name0 = name0u.encode('utf8')
1264
self.build_tree([name0u + '/'])
1266
result = Win32ReadDir().read_dir('', u'.')
1268
self.assertEqual((name0, name0, 'directory'), entry[:3])
1269
self.assertEqual(u'./' + name0u, entry[4])
1270
self.assertStatIsCorrect(entry[4], entry[3])
784
1272
def assertPathCompare(self, path_less, path_greater):
785
1273
"""check that path_less and path_greater compare correctly."""
996
1473
self.assertTrue(isinstance(offset, int))
997
1474
eighteen_hours = 18 * 3600
998
1475
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1478
class TestSizeShaFile(TestCaseInTempDir):
1480
def test_sha_empty(self):
1481
self.build_tree_contents([('foo', '')])
1482
expected_sha = osutils.sha_string('')
1484
self.addCleanup(f.close)
1485
size, sha = osutils.size_sha_file(f)
1486
self.assertEqual(0, size)
1487
self.assertEqual(expected_sha, sha)
1489
def test_sha_mixed_endings(self):
1490
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1491
self.build_tree_contents([('foo', text)])
1492
expected_sha = osutils.sha_string(text)
1494
self.addCleanup(f.close)
1495
size, sha = osutils.size_sha_file(f)
1496
self.assertEqual(38, size)
1497
self.assertEqual(expected_sha, sha)
1500
class TestShaFileByName(TestCaseInTempDir):
1502
def test_sha_empty(self):
1503
self.build_tree_contents([('foo', '')])
1504
expected_sha = osutils.sha_string('')
1505
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1507
def test_sha_mixed_endings(self):
1508
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1509
self.build_tree_contents([('foo', text)])
1510
expected_sha = osutils.sha_string(text)
1511
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1514
class TestResourceLoading(TestCaseInTempDir):
1516
def test_resource_string(self):
1517
# test resource in bzrlib
1518
text = osutils.resource_string('bzrlib', 'debug.py')
1519
self.assertContainsRe(text, "debug_flags = set()")
1520
# test resource under bzrlib
1521
text = osutils.resource_string('bzrlib.ui', 'text.py')
1522
self.assertContainsRe(text, "class TextUIFactory")
1523
# test unsupported package
1524
self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1526
# test unknown resource
1527
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1530
class TestReCompile(TestCase):
1532
def test_re_compile_checked(self):
1533
r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1534
self.assertTrue(r.match('aaaa'))
1535
self.assertTrue(r.match('aAaA'))
1537
def test_re_compile_checked_error(self):
1538
# like https://bugs.launchpad.net/bzr/+bug/251352
1539
err = self.assertRaises(
1540
errors.BzrCommandError,
1541
osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1543
"Invalid regular expression in test case: '*': "
1544
"nothing to repeat",