13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
17
"""Tests for the osutils wrapper."""
19
from cStringIO import StringIO
28
26
from bzrlib import (
31
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
32
from bzrlib.osutils import (
34
is_inside_or_parent_of_any,
37
37
from bzrlib.tests import (
44
class _UTF8DirReaderFeature(tests.Feature):
48
from bzrlib import _readdir_pyx
49
self.reader = _readdir_pyx.UTF8DirReader
54
def feature_name(self):
55
return 'bzrlib._readdir_pyx'
57
UTF8DirReaderFeature = _UTF8DirReaderFeature()
59
term_ios_feature = tests.ModuleAvailableFeature('termios')
62
def _already_unicode(s):
66
def _utf8_to_unicode(s):
67
return s.decode('UTF-8')
70
def dir_reader_scenarios():
71
# For each dir reader we define:
73
# - native_to_unicode: a function converting the native_abspath as returned
74
# by DirReader.read_dir to its unicode representation
76
# UnicodeDirReader is the fallback, it should be tested on all platforms.
77
scenarios = [('unicode',
78
dict(_dir_reader_class=osutils.UnicodeDirReader,
79
_native_to_unicode=_already_unicode))]
80
# Some DirReaders are platform specific and even there they may not be
82
if UTF8DirReaderFeature.available():
83
from bzrlib import _readdir_pyx
84
scenarios.append(('utf8',
85
dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
86
_native_to_unicode=_utf8_to_unicode)))
88
if test__walkdirs_win32.win32_readdir_feature.available():
90
from bzrlib import _walkdirs_win32
93
dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
94
_native_to_unicode=_already_unicode)))
100
def load_tests(basic_tests, module, loader):
101
suite = loader.suiteClass()
102
dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
103
basic_tests, tests.condition_isinstance(TestDirReader))
104
tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
105
suite.addTest(remaining_tests)
109
class TestContainsWhitespace(tests.TestCase):
38
probe_unicode_in_user_encoding,
47
class TestOSUtils(TestCaseInTempDir):
111
49
def test_contains_whitespace(self):
112
50
self.failUnless(osutils.contains_whitespace(u' '))
463
312
osutils.make_readonly('dangling')
464
313
osutils.make_writable('dangling')
466
def test_host_os_dereferences_symlinks(self):
467
osutils.host_os_dereferences_symlinks()
470
class TestCanonicalRelPath(tests.TestCaseInTempDir):
472
_test_needs_features = [tests.CaseInsCasePresFilenameFeature]
474
def test_canonical_relpath_simple(self):
475
f = file('MixedCaseName', 'w')
477
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
478
self.failUnlessEqual('work/MixedCaseName', actual)
480
def test_canonical_relpath_missing_tail(self):
481
os.mkdir('MixedCaseParent')
482
actual = osutils.canonical_relpath(self.test_base_dir,
483
'mixedcaseparent/nochild')
484
self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
487
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
489
def assertRelpath(self, expected, base, path):
490
actual = osutils._cicp_canonical_relpath(base, path)
491
self.assertEqual(expected, actual)
493
def test_simple(self):
494
self.build_tree(['MixedCaseName'])
495
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
496
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
498
def test_subdir_missing_tail(self):
499
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
500
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
501
self.assertRelpath('MixedCaseParent/a_child', base,
502
'MixedCaseParent/a_child')
503
self.assertRelpath('MixedCaseParent/a_child', base,
504
'MixedCaseParent/A_Child')
505
self.assertRelpath('MixedCaseParent/not_child', base,
506
'MixedCaseParent/not_child')
508
def test_at_root_slash(self):
509
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
511
if osutils.MIN_ABS_PATHLENGTH > 1:
512
raise tests.TestSkipped('relpath requires %d chars'
513
% osutils.MIN_ABS_PATHLENGTH)
514
self.assertRelpath('foo', '/', '/foo')
516
def test_at_root_drive(self):
517
if sys.platform != 'win32':
518
raise tests.TestNotApplicable('we can only test drive-letter relative'
519
' paths on Windows where we have drive'
522
# The specific issue is that when at the root of a drive, 'abspath'
523
# returns "C:/" or just "/". However, the code assumes that abspath
524
# always returns something like "C:/foo" or "/foo" (no trailing slash).
525
self.assertRelpath('foo', 'C:/', 'C:/foo')
526
self.assertRelpath('foo', 'X:/', 'X:/foo')
527
self.assertRelpath('foo', 'X:/', 'X://foo')
530
class TestPumpFile(tests.TestCase):
531
"""Test pumpfile method."""
534
tests.TestCase.setUp(self)
535
# create a test datablock
536
self.block_size = 512
537
pattern = '0123456789ABCDEF'
538
self.test_data = pattern * (3 * self.block_size / len(pattern))
539
self.test_data_len = len(self.test_data)
541
def test_bracket_block_size(self):
542
"""Read data in blocks with the requested read size bracketing the
544
# make sure test data is larger than max read size
545
self.assertTrue(self.test_data_len > self.block_size)
547
from_file = file_utils.FakeReadFile(self.test_data)
550
# read (max / 2) bytes and verify read size wasn't affected
551
num_bytes_to_read = self.block_size / 2
552
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
553
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
554
self.assertEqual(from_file.get_read_count(), 1)
556
# read (max) bytes and verify read size wasn't affected
557
num_bytes_to_read = self.block_size
558
from_file.reset_read_count()
559
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
560
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
561
self.assertEqual(from_file.get_read_count(), 1)
563
# read (max + 1) bytes and verify read size was limited
564
num_bytes_to_read = self.block_size + 1
565
from_file.reset_read_count()
566
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
567
self.assertEqual(from_file.get_max_read_size(), self.block_size)
568
self.assertEqual(from_file.get_read_count(), 2)
570
# finish reading the rest of the data
571
num_bytes_to_read = self.test_data_len - to_file.tell()
572
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
574
# report error if the data wasn't equal (we only report the size due
575
# to the length of the data)
576
response_data = to_file.getvalue()
577
if response_data != self.test_data:
578
message = "Data not equal. Expected %d bytes, received %d."
579
self.fail(message % (len(response_data), self.test_data_len))
581
def test_specified_size(self):
582
"""Request a transfer larger than the maximum block size and verify
583
that the maximum read doesn't exceed the block_size."""
584
# make sure test data is larger than max read size
585
self.assertTrue(self.test_data_len > self.block_size)
587
# retrieve data in blocks
588
from_file = file_utils.FakeReadFile(self.test_data)
590
osutils.pumpfile(from_file, to_file, self.test_data_len,
593
# verify read size was equal to the maximum read size
594
self.assertTrue(from_file.get_max_read_size() > 0)
595
self.assertEqual(from_file.get_max_read_size(), self.block_size)
596
self.assertEqual(from_file.get_read_count(), 3)
598
# report error if the data wasn't equal (we only report the size due
599
# to the length of the data)
600
response_data = to_file.getvalue()
601
if response_data != self.test_data:
602
message = "Data not equal. Expected %d bytes, received %d."
603
self.fail(message % (len(response_data), self.test_data_len))
605
def test_to_eof(self):
606
"""Read to end-of-file and verify that the reads are not larger than
607
the maximum read size."""
608
# make sure test data is larger than max read size
609
self.assertTrue(self.test_data_len > self.block_size)
611
# retrieve data to EOF
612
from_file = file_utils.FakeReadFile(self.test_data)
614
osutils.pumpfile(from_file, to_file, -1, self.block_size)
616
# verify read size was equal to the maximum read size
617
self.assertEqual(from_file.get_max_read_size(), self.block_size)
618
self.assertEqual(from_file.get_read_count(), 4)
620
# report error if the data wasn't equal (we only report the size due
621
# to the length of the data)
622
response_data = to_file.getvalue()
623
if response_data != self.test_data:
624
message = "Data not equal. Expected %d bytes, received %d."
625
self.fail(message % (len(response_data), self.test_data_len))
627
def test_defaults(self):
628
"""Verifies that the default arguments will read to EOF -- this
629
test verifies that any existing usages of pumpfile will not be broken
630
with this new version."""
631
# retrieve data using default (old) pumpfile method
632
from_file = file_utils.FakeReadFile(self.test_data)
634
osutils.pumpfile(from_file, to_file)
636
# report error if the data wasn't equal (we only report the size due
637
# to the length of the data)
638
response_data = to_file.getvalue()
639
if response_data != self.test_data:
640
message = "Data not equal. Expected %d bytes, received %d."
641
self.fail(message % (len(response_data), self.test_data_len))
643
def test_report_activity(self):
645
def log_activity(length, direction):
646
activity.append((length, direction))
647
from_file = StringIO(self.test_data)
649
osutils.pumpfile(from_file, to_file, buff_size=500,
650
report_activity=log_activity, direction='read')
651
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
652
(36, 'read')], activity)
654
from_file = StringIO(self.test_data)
657
osutils.pumpfile(from_file, to_file, buff_size=500,
658
report_activity=log_activity, direction='write')
659
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
660
(36, 'write')], activity)
662
# And with a limited amount of data
663
from_file = StringIO(self.test_data)
666
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
667
report_activity=log_activity, direction='read')
668
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
672
class TestPumpStringFile(tests.TestCase):
674
def test_empty(self):
676
osutils.pump_string_file("", output)
677
self.assertEqual("", output.getvalue())
679
def test_more_than_segment_size(self):
681
osutils.pump_string_file("123456789", output, 2)
682
self.assertEqual("123456789", output.getvalue())
684
def test_segment_size(self):
686
osutils.pump_string_file("12", output, 2)
687
self.assertEqual("12", output.getvalue())
689
def test_segment_size_multiple(self):
691
osutils.pump_string_file("1234", output, 2)
692
self.assertEqual("1234", output.getvalue())
695
class TestRelpath(tests.TestCase):
697
def test_simple_relpath(self):
698
cwd = osutils.getcwd()
699
subdir = cwd + '/subdir'
700
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
702
def test_deep_relpath(self):
703
cwd = osutils.getcwd()
704
subdir = cwd + '/sub/subsubdir'
705
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
707
def test_not_relative(self):
708
self.assertRaises(errors.PathNotChild,
709
osutils.relpath, 'C:/path', 'H:/path')
710
self.assertRaises(errors.PathNotChild,
711
osutils.relpath, 'C:/', 'H:/path')
714
class TestSafeUnicode(tests.TestCase):
315
def test_kind_marker(self):
316
self.assertEqual("", osutils.kind_marker("file"))
317
self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
318
self.assertEqual("@", osutils.kind_marker("symlink"))
319
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
322
class TestSafeUnicode(TestCase):
716
324
def test_from_ascii_string(self):
717
325
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
1173
701
new_dirblock.append((info[0], info[1], info[2], info[4]))
1174
702
dirblock[:] = new_dirblock
1176
def _save_platform_info(self):
1177
self.overrideAttr(win32utils, 'winver')
1178
self.overrideAttr(osutils, '_fs_enc')
1179
self.overrideAttr(osutils, '_selected_dir_reader')
1181
def assertDirReaderIs(self, expected):
1182
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
1183
# Force it to redetect
1184
osutils._selected_dir_reader = None
1185
# Nothing to list, but should still trigger the selection logic
1186
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1187
self.assertIsInstance(osutils._selected_dir_reader, expected)
1189
def test_force_walkdirs_utf8_fs_utf8(self):
1190
self.requireFeature(UTF8DirReaderFeature)
1191
self._save_platform_info()
1192
win32utils.winver = None # Avoid the win32 detection code
1193
osutils._fs_enc = 'UTF-8'
1194
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1196
def test_force_walkdirs_utf8_fs_ascii(self):
1197
self.requireFeature(UTF8DirReaderFeature)
1198
self._save_platform_info()
1199
win32utils.winver = None # Avoid the win32 detection code
1200
osutils._fs_enc = 'US-ASCII'
1201
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1203
def test_force_walkdirs_utf8_fs_ANSI(self):
1204
self.requireFeature(UTF8DirReaderFeature)
1205
self._save_platform_info()
1206
win32utils.winver = None # Avoid the win32 detection code
1207
osutils._fs_enc = 'ANSI_X3.4-1968'
1208
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1210
def test_force_walkdirs_utf8_fs_latin1(self):
1211
self._save_platform_info()
1212
win32utils.winver = None # Avoid the win32 detection code
1213
osutils._fs_enc = 'latin1'
1214
self.assertDirReaderIs(osutils.UnicodeDirReader)
1216
def test_force_walkdirs_utf8_nt(self):
1217
# Disabled because the thunk of the whole walkdirs api is disabled.
1218
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1219
self._save_platform_info()
1220
win32utils.winver = 'Windows NT'
1221
from bzrlib._walkdirs_win32 import Win32ReadDir
1222
self.assertDirReaderIs(Win32ReadDir)
1224
def test_force_walkdirs_utf8_98(self):
1225
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1226
self._save_platform_info()
1227
win32utils.winver = 'Windows 98'
1228
self.assertDirReaderIs(osutils.UnicodeDirReader)
1230
704
def test_unicode_walkdirs(self):
1231
705
"""Walkdirs should always return unicode paths."""
1232
self.requireFeature(tests.UnicodeFilenameFeature)
1233
706
name0 = u'0file-\xb6'
1234
707
name1 = u'1dir-\u062c\u0648'
1235
708
name2 = u'2file-\u0633'
1328
808
result.append((dirdetail, new_dirblock))
1329
809
self.assertEqual(expected_dirblocks, result)
1331
def test__walkdirs_utf8_with_unicode_fs(self):
1332
"""UnicodeDirReader should be a safe fallback everywhere
811
def test_unicode__walkdirs_unicode_to_utf8(self):
812
"""walkdirs_unicode_to_utf8 should be a safe fallback everywhere
1334
814
The abspath portion should be in unicode
1336
self.requireFeature(tests.UnicodeFilenameFeature)
1337
# Use the unicode reader. TODO: split into driver-and-driven unit
1339
self._save_platform_info()
1340
osutils._selected_dir_reader = osutils.UnicodeDirReader()
1341
name0u = u'0file-\xb6'
1342
name1u = u'1dir-\u062c\u0648'
1343
name2u = u'2file-\u0633'
1347
name1u + '/' + name0u,
1348
name1u + '/' + name1u + '/',
1351
self.build_tree(tree)
1352
name0 = name0u.encode('utf8')
1353
name1 = name1u.encode('utf8')
1354
name2 = name2u.encode('utf8')
1356
# All of the abspaths should be in unicode, all of the relative paths
1358
expected_dirblocks = [
1360
[(name0, name0, 'file', './' + name0u),
1361
(name1, name1, 'directory', './' + name1u),
1362
(name2, name2, 'file', './' + name2u),
1365
((name1, './' + name1u),
1366
[(name1 + '/' + name0, name0, 'file', './' + name1u
1368
(name1 + '/' + name1, name1, 'directory', './' + name1u
1372
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1377
result = list(osutils._walkdirs_utf8('.'))
1378
self._filter_out_stat(result)
1379
self.assertEqual(expected_dirblocks, result)
1381
def test__walkdirs_utf8_win32readdir(self):
1382
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1383
self.requireFeature(tests.UnicodeFilenameFeature)
1384
from bzrlib._walkdirs_win32 import Win32ReadDir
1385
self._save_platform_info()
1386
osutils._selected_dir_reader = Win32ReadDir()
1387
name0u = u'0file-\xb6'
1388
name1u = u'1dir-\u062c\u0648'
1389
name2u = u'2file-\u0633'
1393
name1u + '/' + name0u,
1394
name1u + '/' + name1u + '/',
1397
self.build_tree(tree)
1398
name0 = name0u.encode('utf8')
1399
name1 = name1u.encode('utf8')
1400
name2 = name2u.encode('utf8')
1402
# All of the abspaths should be in unicode, all of the relative paths
1404
expected_dirblocks = [
1406
[(name0, name0, 'file', './' + name0u),
1407
(name1, name1, 'directory', './' + name1u),
1408
(name2, name2, 'file', './' + name2u),
1411
((name1, './' + name1u),
1412
[(name1 + '/' + name0, name0, 'file', './' + name1u
1414
(name1 + '/' + name1, name1, 'directory', './' + name1u
1418
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1423
result = list(osutils._walkdirs_utf8(u'.'))
1424
self._filter_out_stat(result)
1425
self.assertEqual(expected_dirblocks, result)
1427
def assertStatIsCorrect(self, path, win32stat):
1428
os_stat = os.stat(path)
1429
self.assertEqual(os_stat.st_size, win32stat.st_size)
1430
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1431
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1432
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1433
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1434
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1435
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1437
def test__walkdirs_utf_win32_find_file_stat_file(self):
1438
"""make sure our Stat values are valid"""
1439
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1440
self.requireFeature(tests.UnicodeFilenameFeature)
1441
from bzrlib._walkdirs_win32 import Win32ReadDir
1442
name0u = u'0file-\xb6'
1443
name0 = name0u.encode('utf8')
1444
self.build_tree([name0u])
1445
# I hate to sleep() here, but I'm trying to make the ctime different
1448
f = open(name0u, 'ab')
816
name0u = u'0file-\xb6'
817
name1u = u'1dir-\u062c\u0648'
818
name2u = u'2file-\u0633'
822
name1u + '/' + name0u,
823
name1u + '/' + name1u + '/',
1450
f.write('just a small update')
1454
result = Win32ReadDir().read_dir('', u'.')
1456
self.assertEqual((name0, name0, 'file'), entry[:3])
1457
self.assertEqual(u'./' + name0u, entry[4])
1458
self.assertStatIsCorrect(entry[4], entry[3])
1459
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1461
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1462
"""make sure our Stat values are valid"""
1463
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1464
self.requireFeature(tests.UnicodeFilenameFeature)
1465
from bzrlib._walkdirs_win32 import Win32ReadDir
1466
name0u = u'0dir-\u062c\u0648'
827
self.build_tree(tree)
829
raise TestSkipped('Could not represent Unicode chars'
830
' in current encoding.')
1467
831
name0 = name0u.encode('utf8')
1468
self.build_tree([name0u + '/'])
832
name1 = name1u.encode('utf8')
833
name2 = name2u.encode('utf8')
1470
result = Win32ReadDir().read_dir('', u'.')
1472
self.assertEqual((name0, name0, 'directory'), entry[:3])
1473
self.assertEqual(u'./' + name0u, entry[4])
1474
self.assertStatIsCorrect(entry[4], entry[3])
835
# All of the abspaths should be in unicode, all of the relative paths
837
expected_dirblocks = [
839
[(name0, name0, 'file', './' + name0u),
840
(name1, name1, 'directory', './' + name1u),
841
(name2, name2, 'file', './' + name2u),
844
((name1, './' + name1u),
845
[(name1 + '/' + name0, name0, 'file', './' + name1u
847
(name1 + '/' + name1, name1, 'directory', './' + name1u
851
((name1 + '/' + name1, './' + name1u + '/' + name1u),
856
result = list(osutils._walkdirs_unicode_to_utf8('.'))
857
self._filter_out_stat(result)
858
self.assertEqual(expected_dirblocks, result)
1476
860
def assertPathCompare(self, path_less, path_greater):
1477
861
"""check that path_less and path_greater compare correctly."""
1653
1041
self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1656
class TestSizeShaFile(tests.TestCaseInTempDir):
1658
def test_sha_empty(self):
1659
self.build_tree_contents([('foo', '')])
1660
expected_sha = osutils.sha_string('')
1662
self.addCleanup(f.close)
1663
size, sha = osutils.size_sha_file(f)
1664
self.assertEqual(0, size)
1665
self.assertEqual(expected_sha, sha)
1667
def test_sha_mixed_endings(self):
1668
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1669
self.build_tree_contents([('foo', text)])
1670
expected_sha = osutils.sha_string(text)
1671
f = open('foo', 'rb')
1672
self.addCleanup(f.close)
1673
size, sha = osutils.size_sha_file(f)
1674
self.assertEqual(38, size)
1675
self.assertEqual(expected_sha, sha)
1678
class TestShaFileByName(tests.TestCaseInTempDir):
1680
def test_sha_empty(self):
1681
self.build_tree_contents([('foo', '')])
1682
expected_sha = osutils.sha_string('')
1683
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1685
def test_sha_mixed_endings(self):
1686
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1687
self.build_tree_contents([('foo', text)])
1688
expected_sha = osutils.sha_string(text)
1689
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1692
class TestResourceLoading(tests.TestCaseInTempDir):
1044
class TestLocalTimeOffset(TestCase):
1046
def test_local_time_offset(self):
1047
"""Test that local_time_offset() returns a sane value."""
1048
offset = osutils.local_time_offset()
1049
self.assertTrue(isinstance(offset, int))
1050
# Test that the offset is no more than a eighteen hours in
1052
# Time zone handling is system specific, so it is difficult to
1053
# do more specific tests, but a value outside of this range is
1055
eighteen_hours = 18 * 3600
1056
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1058
def test_local_time_offset_with_timestamp(self):
1059
"""Test that local_time_offset() works with a timestamp."""
1060
offset = osutils.local_time_offset(1000000000.1234567)
1061
self.assertTrue(isinstance(offset, int))
1062
eighteen_hours = 18 * 3600
1063
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1066
class TestShaFileByName(TestCaseInTempDir):
1068
def test_sha_empty(self):
1069
self.build_tree_contents([('foo', '')])
1070
expected_sha = osutils.sha_string('')
1071
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1073
def test_sha_mixed_endings(self):
1074
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1075
self.build_tree_contents([('foo', text)])
1076
expected_sha = osutils.sha_string(text)
1077
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
* hashcache - log every time a working file is read to determine its hash
1113
* hooks - trace hook execution
1114
* hpss - trace smart protocol requests and responses
1115
* http - trace http connections, requests and responses
1116
* index - trace major index operations
1117
* knit - trace knit operations
1118
* lock - trace when lockdir locks are taken or released
1119
* merge - emit information for debugging merges
1120
* pack - emit information about pack operations
1121
* selftest_debug - do not disable all debug flags when running selftest
1127
class TestResourceLoading(TestCaseInTempDir):
1694
1129
def test_resource_string(self):
1695
1130
# test resource in bzrlib
1696
1131
text = osutils.resource_string('bzrlib', 'debug.py')
1697
self.assertContainsRe(text, "debug_flags = set()")
1132
self.assertEquals(_debug_text, text)
1698
1133
# test resource under bzrlib
1699
1134
text = osutils.resource_string('bzrlib.ui', 'text.py')
1700
1135
self.assertContainsRe(text, "class TextUIFactory")
1704
1139
# test unknown resource
1705
1140
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1708
class TestReCompile(tests.TestCase):
1710
def _deprecated_re_compile_checked(self, *args, **kwargs):
1711
return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
1712
osutils.re_compile_checked, *args, **kwargs)
1714
def test_re_compile_checked(self):
1715
r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
1716
self.assertTrue(r.match('aaaa'))
1717
self.assertTrue(r.match('aAaA'))
1719
def test_re_compile_checked_error(self):
1720
# like https://bugs.launchpad.net/bzr/+bug/251352
1722
# Due to possible test isolation error, re.compile is not lazy at
1723
# this point. We re-install lazy compile.
1724
lazy_regex.install_lazy_compile()
1725
err = self.assertRaises(
1726
errors.BzrCommandError,
1727
self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
1729
'Invalid regular expression in test case: '
1730
'"*" nothing to repeat',
1734
class TestDirReader(tests.TestCaseInTempDir):
1737
_dir_reader_class = None
1738
_native_to_unicode = None
1741
tests.TestCaseInTempDir.setUp(self)
1742
self.overrideAttr(osutils,
1743
'_selected_dir_reader', self._dir_reader_class())
1745
def _get_ascii_tree(self):
1753
expected_dirblocks = [
1755
[('0file', '0file', 'file'),
1756
('1dir', '1dir', 'directory'),
1757
('2file', '2file', 'file'),
1760
(('1dir', './1dir'),
1761
[('1dir/0file', '0file', 'file'),
1762
('1dir/1dir', '1dir', 'directory'),
1765
(('1dir/1dir', './1dir/1dir'),
1770
return tree, expected_dirblocks
1772
def test_walk_cur_dir(self):
1773
tree, expected_dirblocks = self._get_ascii_tree()
1774
self.build_tree(tree)
1775
result = list(osutils._walkdirs_utf8('.'))
1776
# Filter out stat and abspath
1777
self.assertEqual(expected_dirblocks,
1778
[(dirinfo, [line[0:3] for line in block])
1779
for dirinfo, block in result])
1781
def test_walk_sub_dir(self):
1782
tree, expected_dirblocks = self._get_ascii_tree()
1783
self.build_tree(tree)
1784
# you can search a subdir only, with a supplied prefix.
1785
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1786
# Filter out stat and abspath
1787
self.assertEqual(expected_dirblocks[1:],
1788
[(dirinfo, [line[0:3] for line in block])
1789
for dirinfo, block in result])
1791
def _get_unicode_tree(self):
1792
name0u = u'0file-\xb6'
1793
name1u = u'1dir-\u062c\u0648'
1794
name2u = u'2file-\u0633'
1798
name1u + '/' + name0u,
1799
name1u + '/' + name1u + '/',
1802
name0 = name0u.encode('UTF-8')
1803
name1 = name1u.encode('UTF-8')
1804
name2 = name2u.encode('UTF-8')
1805
expected_dirblocks = [
1807
[(name0, name0, 'file', './' + name0u),
1808
(name1, name1, 'directory', './' + name1u),
1809
(name2, name2, 'file', './' + name2u),
1812
((name1, './' + name1u),
1813
[(name1 + '/' + name0, name0, 'file', './' + name1u
1815
(name1 + '/' + name1, name1, 'directory', './' + name1u
1819
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1824
return tree, expected_dirblocks
1826
def _filter_out(self, raw_dirblocks):
1827
"""Filter out a walkdirs_utf8 result.
1829
stat field is removed, all native paths are converted to unicode
1831
filtered_dirblocks = []
1832
for dirinfo, block in raw_dirblocks:
1833
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1836
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1837
filtered_dirblocks.append((dirinfo, details))
1838
return filtered_dirblocks
1840
def test_walk_unicode_tree(self):
1841
self.requireFeature(tests.UnicodeFilenameFeature)
1842
tree, expected_dirblocks = self._get_unicode_tree()
1843
self.build_tree(tree)
1844
result = list(osutils._walkdirs_utf8('.'))
1845
self.assertEqual(expected_dirblocks, self._filter_out(result))
1847
def test_symlink(self):
1848
self.requireFeature(tests.SymlinkFeature)
1849
self.requireFeature(tests.UnicodeFilenameFeature)
1850
target = u'target\N{Euro Sign}'
1851
link_name = u'l\N{Euro Sign}nk'
1852
os.symlink(target, link_name)
1853
target_utf8 = target.encode('UTF-8')
1854
link_name_utf8 = link_name.encode('UTF-8')
1855
expected_dirblocks = [
1857
[(link_name_utf8, link_name_utf8,
1858
'symlink', './' + link_name),],
1860
result = list(osutils._walkdirs_utf8('.'))
1861
self.assertEqual(expected_dirblocks, self._filter_out(result))
1864
class TestReadLink(tests.TestCaseInTempDir):
1865
"""Exposes os.readlink() problems and the osutils solution.
1867
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1868
unicode string will be returned if a unicode string is passed.
1870
But prior python versions failed to properly encode the passed unicode
1873
_test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1876
super(tests.TestCaseInTempDir, self).setUp()
1877
self.link = u'l\N{Euro Sign}ink'
1878
self.target = u'targe\N{Euro Sign}t'
1879
os.symlink(self.target, self.link)
1881
def test_os_readlink_link_encoding(self):
1882
if sys.version_info < (2, 6):
1883
self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1885
self.assertEquals(self.target, os.readlink(self.link))
1887
def test_os_readlink_link_decoding(self):
1888
self.assertEquals(self.target.encode(osutils._fs_enc),
1889
os.readlink(self.link.encode(osutils._fs_enc)))
1892
class TestConcurrency(tests.TestCase):
1895
super(TestConcurrency, self).setUp()
1896
self.overrideAttr(osutils, '_cached_local_concurrency')
1898
def test_local_concurrency(self):
1899
concurrency = osutils.local_concurrency()
1900
self.assertIsInstance(concurrency, int)
1902
def test_local_concurrency_environment_variable(self):
1903
os.environ['BZR_CONCURRENCY'] = '2'
1904
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1905
os.environ['BZR_CONCURRENCY'] = '3'
1906
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1907
os.environ['BZR_CONCURRENCY'] = 'foo'
1908
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1910
def test_option_concurrency(self):
1911
os.environ['BZR_CONCURRENCY'] = '1'
1912
self.run_bzr('rocks --concurrency 42')
1913
# Command line overrides envrionment variable
1914
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1915
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1918
class TestFailedToLoadExtension(tests.TestCase):
1920
def _try_loading(self):
1922
import bzrlib._fictional_extension_py
1923
except ImportError, e:
1924
osutils.failed_to_load_extension(e)
1928
super(TestFailedToLoadExtension, self).setUp()
1929
self.overrideAttr(osutils, '_extension_load_failures', [])
1931
def test_failure_to_load(self):
1933
self.assertLength(1, osutils._extension_load_failures)
1934
self.assertEquals(osutils._extension_load_failures[0],
1935
"No module named _fictional_extension_py")
1937
def test_report_extension_load_failures_no_warning(self):
1938
self.assertTrue(self._try_loading())
1939
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
1940
# it used to give a Python warning; it no longer does
1941
self.assertLength(0, warnings)
1943
def test_report_extension_load_failures_message(self):
1945
trace.push_log_file(log)
1946
self.assertTrue(self._try_loading())
1947
osutils.report_extension_load_failures()
1948
self.assertContainsRe(
1950
r"bzr: warning: some compiled extensions could not be loaded; "
1951
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1955
class TestTerminalWidth(tests.TestCase):
1958
tests.TestCase.setUp(self)
1959
self._orig_terminal_size_state = osutils._terminal_size_state
1960
self._orig_first_terminal_size = osutils._first_terminal_size
1961
self.addCleanup(self.restore_osutils_globals)
1962
osutils._terminal_size_state = 'no_data'
1963
osutils._first_terminal_size = None
1965
def restore_osutils_globals(self):
1966
osutils._terminal_size_state = self._orig_terminal_size_state
1967
osutils._first_terminal_size = self._orig_first_terminal_size
1969
def replace_stdout(self, new):
1970
self.overrideAttr(sys, 'stdout', new)
1972
def replace__terminal_size(self, new):
1973
self.overrideAttr(osutils, '_terminal_size', new)
1975
def set_fake_tty(self):
1977
class I_am_a_tty(object):
1981
self.replace_stdout(I_am_a_tty())
1983
def test_default_values(self):
1984
self.assertEqual(80, osutils.default_terminal_width)
1986
def test_defaults_to_BZR_COLUMNS(self):
1987
# BZR_COLUMNS is set by the test framework
1988
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
1989
os.environ['BZR_COLUMNS'] = '12'
1990
self.assertEqual(12, osutils.terminal_width())
1992
def test_falls_back_to_COLUMNS(self):
1993
del os.environ['BZR_COLUMNS']
1994
self.assertNotEqual('42', os.environ['COLUMNS'])
1996
os.environ['COLUMNS'] = '42'
1997
self.assertEqual(42, osutils.terminal_width())
1999
def test_tty_default_without_columns(self):
2000
del os.environ['BZR_COLUMNS']
2001
del os.environ['COLUMNS']
2003
def terminal_size(w, h):
2007
# We need to override the osutils definition as it depends on the
2008
# running environment that we can't control (PQM running without a
2009
# controlling terminal is one example).
2010
self.replace__terminal_size(terminal_size)
2011
self.assertEqual(42, osutils.terminal_width())
2013
def test_non_tty_default_without_columns(self):
2014
del os.environ['BZR_COLUMNS']
2015
del os.environ['COLUMNS']
2016
self.replace_stdout(None)
2017
self.assertEqual(None, osutils.terminal_width())
2019
def test_no_TIOCGWINSZ(self):
2020
self.requireFeature(term_ios_feature)
2021
termios = term_ios_feature.module
2022
# bug 63539 is about a termios without TIOCGWINSZ attribute
2024
orig = termios.TIOCGWINSZ
2025
except AttributeError:
2026
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
2029
self.overrideAttr(termios, 'TIOCGWINSZ')
2030
del termios.TIOCGWINSZ
2031
del os.environ['BZR_COLUMNS']
2032
del os.environ['COLUMNS']
2033
# Whatever the result is, if we don't raise an exception, it's ok.
2034
osutils.terminal_width()
2036
class TestCreationOps(tests.TestCaseInTempDir):
2037
_test_needs_features = [features.chown_feature]
2040
tests.TestCaseInTempDir.setUp(self)
2041
self.overrideAttr(os, 'chown', self._dummy_chown)
2043
# params set by call to _dummy_chown
2044
self.path = self.uid = self.gid = None
2046
def _dummy_chown(self, path, uid, gid):
2047
self.path, self.uid, self.gid = path, uid, gid
2049
def test_copy_ownership_from_path(self):
2050
"""copy_ownership_from_path test with specified src."""
2052
f = open('test_file', 'wt')
2053
osutils.copy_ownership_from_path('test_file', ownsrc)
2056
self.assertEquals(self.path, 'test_file')
2057
self.assertEquals(self.uid, s.st_uid)
2058
self.assertEquals(self.gid, s.st_gid)
2060
def test_copy_ownership_nonesrc(self):
2061
"""copy_ownership_from_path test with src=None."""
2062
f = open('test_file', 'wt')
2063
# should use parent dir for permissions
2064
osutils.copy_ownership_from_path('test_file')
2067
self.assertEquals(self.path, 'test_file')
2068
self.assertEquals(self.uid, s.st_uid)
2069
self.assertEquals(self.gid, s.st_gid)
2071
class TestGetuserUnicode(tests.TestCase):
2073
def test_ascii_user(self):
2074
osutils.set_or_unset_env('LOGNAME', 'jrandom')
2075
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2077
def test_unicode_user(self):
2078
ue = osutils.get_user_encoding()
2079
osutils.set_or_unset_env('LOGNAME', u'jrandom\xb6'.encode(ue))
2080
self.assertEqual(u'jrandom\xb6', osutils.getuser_unicode())
added
removed
bzrlib/tests/test_osutils.py
