54
83
def feature_name(self):
55
84
return 'bzrlib._readdir_pyx'
57
UTF8DirReaderFeature = features.ModuleAvailableFeature('bzrlib._readdir_pyx')
59
term_ios_feature = features.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
load_tests = load_tests_apply_scenarios
103
class TestContainsWhitespace(tests.TestCase):
86
ReadDirFeature = _ReadDirFeature()
89
class TestOSUtils(TestCaseInTempDir):
105
91
def test_contains_whitespace(self):
106
self.assertTrue(osutils.contains_whitespace(u' '))
107
self.assertTrue(osutils.contains_whitespace(u'hello there'))
108
self.assertTrue(osutils.contains_whitespace(u'hellothere\n'))
109
self.assertTrue(osutils.contains_whitespace(u'hello\nthere'))
110
self.assertTrue(osutils.contains_whitespace(u'hello\rthere'))
111
self.assertTrue(osutils.contains_whitespace(u'hello\tthere'))
92
self.failUnless(osutils.contains_whitespace(u' '))
93
self.failUnless(osutils.contains_whitespace(u'hello there'))
94
self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
95
self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
96
self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
97
self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
113
99
# \xa0 is "Non-breaking-space" which on some python locales thinks it
114
100
# is whitespace, but we do not.
115
self.assertFalse(osutils.contains_whitespace(u''))
116
self.assertFalse(osutils.contains_whitespace(u'hellothere'))
117
self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
120
class TestRename(tests.TestCaseInTempDir):
122
def create_file(self, filename, content):
123
f = open(filename, 'wb')
129
def _fancy_rename(self, a, b):
130
osutils.fancy_rename(a, b, rename_func=os.rename,
131
unlink_func=os.unlink)
101
self.failIf(osutils.contains_whitespace(u''))
102
self.failIf(osutils.contains_whitespace(u'hellothere'))
103
self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
133
105
def test_fancy_rename(self):
134
106
# This should work everywhere
135
self.create_file('a', 'something in a\n')
136
self._fancy_rename('a', 'b')
137
self.assertPathDoesNotExist('a')
138
self.assertPathExists('b')
108
osutils.fancy_rename(a, b,
109
rename_func=os.rename,
110
unlink_func=os.unlink)
112
open('a', 'wb').write('something in a\n')
114
self.failIfExists('a')
115
self.failUnlessExists('b')
139
116
self.check_file_contents('b', 'something in a\n')
141
self.create_file('a', 'new something in a\n')
142
self._fancy_rename('b', 'a')
118
open('a', 'wb').write('new something in a\n')
144
121
self.check_file_contents('a', 'something in a\n')
146
def test_fancy_rename_fails_source_missing(self):
147
# An exception should be raised, and the target should be left in place
148
self.create_file('target', 'data in target\n')
149
self.assertRaises((IOError, OSError), self._fancy_rename,
150
'missingsource', 'target')
151
self.assertPathExists('target')
152
self.check_file_contents('target', 'data in target\n')
154
def test_fancy_rename_fails_if_source_and_target_missing(self):
155
self.assertRaises((IOError, OSError), self._fancy_rename,
156
'missingsource', 'missingtarget')
158
123
def test_rename(self):
159
124
# Rename should be semi-atomic on all platforms
160
self.create_file('a', 'something in a\n')
125
open('a', 'wb').write('something in a\n')
161
126
osutils.rename('a', 'b')
162
self.assertPathDoesNotExist('a')
163
self.assertPathExists('b')
127
self.failIfExists('a')
128
self.failUnlessExists('b')
164
129
self.check_file_contents('b', 'something in a\n')
166
self.create_file('a', 'new something in a\n')
131
open('a', 'wb').write('new something in a\n')
167
132
osutils.rename('b', 'a')
169
134
self.check_file_contents('a', 'something in a\n')
476
354
osutils.make_readonly('dangling')
477
355
osutils.make_writable('dangling')
357
def test_kind_marker(self):
358
self.assertEqual("", osutils.kind_marker("file"))
359
self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
360
self.assertEqual("@", osutils.kind_marker("symlink"))
361
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
479
363
def test_host_os_dereferences_symlinks(self):
480
364
osutils.host_os_dereferences_symlinks()
483
class TestCanonicalRelPath(tests.TestCaseInTempDir):
485
_test_needs_features = [features.CaseInsCasePresFilenameFeature]
487
def test_canonical_relpath_simple(self):
488
f = file('MixedCaseName', 'w')
490
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
491
self.assertEqual('work/MixedCaseName', actual)
493
def test_canonical_relpath_missing_tail(self):
494
os.mkdir('MixedCaseParent')
495
actual = osutils.canonical_relpath(self.test_base_dir,
496
'mixedcaseparent/nochild')
497
self.assertEqual('work/MixedCaseParent/nochild', actual)
500
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
502
def assertRelpath(self, expected, base, path):
503
actual = osutils._cicp_canonical_relpath(base, path)
504
self.assertEqual(expected, actual)
506
def test_simple(self):
507
self.build_tree(['MixedCaseName'])
508
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
509
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
511
def test_subdir_missing_tail(self):
512
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
513
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
514
self.assertRelpath('MixedCaseParent/a_child', base,
515
'MixedCaseParent/a_child')
516
self.assertRelpath('MixedCaseParent/a_child', base,
517
'MixedCaseParent/A_Child')
518
self.assertRelpath('MixedCaseParent/not_child', base,
519
'MixedCaseParent/not_child')
521
def test_at_root_slash(self):
522
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
524
if osutils.MIN_ABS_PATHLENGTH > 1:
525
raise tests.TestSkipped('relpath requires %d chars'
526
% osutils.MIN_ABS_PATHLENGTH)
527
self.assertRelpath('foo', '/', '/foo')
529
def test_at_root_drive(self):
530
if sys.platform != 'win32':
531
raise tests.TestNotApplicable('we can only test drive-letter relative'
532
' paths on Windows where we have drive'
535
# The specific issue is that when at the root of a drive, 'abspath'
536
# returns "C:/" or just "/". However, the code assumes that abspath
537
# always returns something like "C:/foo" or "/foo" (no trailing slash).
538
self.assertRelpath('foo', 'C:/', 'C:/foo')
539
self.assertRelpath('foo', 'X:/', 'X:/foo')
540
self.assertRelpath('foo', 'X:/', 'X://foo')
543
class TestPumpFile(tests.TestCase):
367
class TestPumpFile(TestCase):
544
368
"""Test pumpfile method."""
547
tests.TestCase.setUp(self)
548
370
# create a test datablock
549
371
self.block_size = 512
550
372
pattern = '0123456789ABCDEF'
653
474
message = "Data not equal. Expected %d bytes, received %d."
654
475
self.fail(message % (len(response_data), self.test_data_len))
656
def test_report_activity(self):
658
def log_activity(length, direction):
659
activity.append((length, direction))
660
from_file = StringIO(self.test_data)
662
osutils.pumpfile(from_file, to_file, buff_size=500,
663
report_activity=log_activity, direction='read')
664
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
665
(36, 'read')], activity)
667
from_file = StringIO(self.test_data)
670
osutils.pumpfile(from_file, to_file, buff_size=500,
671
report_activity=log_activity, direction='write')
672
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
673
(36, 'write')], activity)
675
# And with a limited amount of data
676
from_file = StringIO(self.test_data)
679
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
680
report_activity=log_activity, direction='read')
681
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
685
class TestPumpStringFile(tests.TestCase):
478
class TestPumpStringFile(TestCase):
687
480
def test_empty(self):
688
481
output = StringIO()
689
osutils.pump_string_file("", output)
482
pump_string_file("", output)
690
483
self.assertEqual("", output.getvalue())
692
485
def test_more_than_segment_size(self):
693
486
output = StringIO()
694
osutils.pump_string_file("123456789", output, 2)
487
pump_string_file("123456789", output, 2)
695
488
self.assertEqual("123456789", output.getvalue())
697
490
def test_segment_size(self):
698
491
output = StringIO()
699
osutils.pump_string_file("12", output, 2)
492
pump_string_file("12", output, 2)
700
493
self.assertEqual("12", output.getvalue())
702
495
def test_segment_size_multiple(self):
703
496
output = StringIO()
704
osutils.pump_string_file("1234", output, 2)
497
pump_string_file("1234", output, 2)
705
498
self.assertEqual("1234", output.getvalue())
708
class TestRelpath(tests.TestCase):
710
def test_simple_relpath(self):
711
cwd = osutils.getcwd()
712
subdir = cwd + '/subdir'
713
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
715
def test_deep_relpath(self):
716
cwd = osutils.getcwd()
717
subdir = cwd + '/sub/subsubdir'
718
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
720
def test_not_relative(self):
721
self.assertRaises(errors.PathNotChild,
722
osutils.relpath, 'C:/path', 'H:/path')
723
self.assertRaises(errors.PathNotChild,
724
osutils.relpath, 'C:/', 'H:/path')
727
class TestSafeUnicode(tests.TestCase):
501
class TestSafeUnicode(TestCase):
729
503
def test_from_ascii_string(self):
730
504
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
1189
912
new_dirblock.append((info[0], info[1], info[2], info[4]))
1190
913
dirblock[:] = new_dirblock
915
def test__walkdirs_utf8_selection(self):
916
# Just trigger the function once, to make sure it has selected a real
918
list(osutils._walkdirs_utf8('.'))
919
if WalkdirsWin32Feature.available():
920
# If the compiled form is available, make sure it is used
921
from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
922
self.assertIs(_walkdirs_utf8_win32_find_file,
923
osutils._real_walkdirs_utf8)
924
elif sys.platform == 'win32':
925
self.assertIs(osutils._walkdirs_unicode_to_utf8,
926
osutils._real_walkdirs_utf8)
927
elif osutils._fs_enc.upper() in ('UTF-8', 'ASCII', 'ANSI_X3.4-1968'): # ascii
928
self.assertIs(osutils._walkdirs_fs_utf8,
929
osutils._real_walkdirs_utf8)
931
self.assertIs(osutils._walkdirs_unicode_to_utf8,
932
osutils._real_walkdirs_utf8)
1192
934
def _save_platform_info(self):
1193
self.overrideAttr(win32utils, 'winver')
1194
self.overrideAttr(osutils, '_fs_enc')
1195
self.overrideAttr(osutils, '_selected_dir_reader')
935
cur_winver = win32utils.winver
936
cur_fs_enc = osutils._fs_enc
937
cur_real_walkdirs_utf8 = osutils._real_walkdirs_utf8
939
win32utils.winver = cur_winver
940
osutils._fs_enc = cur_fs_enc
941
osutils._real_walkdirs_utf8 = cur_real_walkdirs_utf8
942
self.addCleanup(restore)
1197
def assertDirReaderIs(self, expected):
944
def assertWalkdirsUtf8Is(self, expected):
1198
945
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
1199
946
# Force it to redetect
1200
osutils._selected_dir_reader = None
947
osutils._real_walkdirs_utf8 = None
1201
948
# Nothing to list, but should still trigger the selection logic
1202
949
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1203
self.assertIsInstance(osutils._selected_dir_reader, expected)
950
self.assertIs(expected, osutils._real_walkdirs_utf8)
1205
952
def test_force_walkdirs_utf8_fs_utf8(self):
1206
self.requireFeature(UTF8DirReaderFeature)
1207
953
self._save_platform_info()
1208
954
win32utils.winver = None # Avoid the win32 detection code
1209
955
osutils._fs_enc = 'UTF-8'
1210
self.assertDirReaderIs(
1211
UTF8DirReaderFeature.module.UTF8DirReader)
956
self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1213
958
def test_force_walkdirs_utf8_fs_ascii(self):
1214
self.requireFeature(UTF8DirReaderFeature)
1215
959
self._save_platform_info()
1216
960
win32utils.winver = None # Avoid the win32 detection code
1217
961
osutils._fs_enc = 'US-ASCII'
1218
self.assertDirReaderIs(
1219
UTF8DirReaderFeature.module.UTF8DirReader)
962
self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1221
964
def test_force_walkdirs_utf8_fs_ANSI(self):
1222
self.requireFeature(UTF8DirReaderFeature)
1223
965
self._save_platform_info()
1224
966
win32utils.winver = None # Avoid the win32 detection code
1225
967
osutils._fs_enc = 'ANSI_X3.4-1968'
1226
self.assertDirReaderIs(
1227
UTF8DirReaderFeature.module.UTF8DirReader)
968
self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1229
970
def test_force_walkdirs_utf8_fs_latin1(self):
1230
971
self._save_platform_info()
1231
972
win32utils.winver = None # Avoid the win32 detection code
1232
973
osutils._fs_enc = 'latin1'
1233
self.assertDirReaderIs(osutils.UnicodeDirReader)
974
self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
1235
976
def test_force_walkdirs_utf8_nt(self):
1236
# Disabled because the thunk of the whole walkdirs api is disabled.
1237
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
977
self.requireFeature(WalkdirsWin32Feature)
1238
978
self._save_platform_info()
1239
979
win32utils.winver = 'Windows NT'
1240
from bzrlib._walkdirs_win32 import Win32ReadDir
1241
self.assertDirReaderIs(Win32ReadDir)
980
from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
981
self.assertWalkdirsUtf8Is(_walkdirs_utf8_win32_find_file)
1243
def test_force_walkdirs_utf8_98(self):
1244
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
983
def test_force_walkdirs_utf8_nt(self):
984
self.requireFeature(WalkdirsWin32Feature)
1245
985
self._save_platform_info()
1246
986
win32utils.winver = 'Windows 98'
1247
self.assertDirReaderIs(osutils.UnicodeDirReader)
987
self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
1249
989
def test_unicode_walkdirs(self):
1250
990
"""Walkdirs should always return unicode paths."""
1251
self.requireFeature(features.UnicodeFilenameFeature)
1252
991
name0 = u'0file-\xb6'
1253
992
name1 = u'1dir-\u062c\u0648'
1254
993
name2 = u'2file-\u0633'
1669
1416
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1670
1417
self.assertEqual('foo', old)
1671
1418
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1672
self.assertFalse('BZR_TEST_ENV_VAR' in os.environ)
1675
class TestSizeShaFile(tests.TestCaseInTempDir):
1677
def test_sha_empty(self):
1678
self.build_tree_contents([('foo', '')])
1679
expected_sha = osutils.sha_string('')
1681
self.addCleanup(f.close)
1682
size, sha = osutils.size_sha_file(f)
1683
self.assertEqual(0, size)
1684
self.assertEqual(expected_sha, sha)
1686
def test_sha_mixed_endings(self):
1687
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1688
self.build_tree_contents([('foo', text)])
1689
expected_sha = osutils.sha_string(text)
1690
f = open('foo', 'rb')
1691
self.addCleanup(f.close)
1692
size, sha = osutils.size_sha_file(f)
1693
self.assertEqual(38, size)
1694
self.assertEqual(expected_sha, sha)
1697
class TestShaFileByName(tests.TestCaseInTempDir):
1699
def test_sha_empty(self):
1700
self.build_tree_contents([('foo', '')])
1701
expected_sha = osutils.sha_string('')
1702
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1704
def test_sha_mixed_endings(self):
1705
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1706
self.build_tree_contents([('foo', text)])
1707
expected_sha = osutils.sha_string(text)
1708
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1711
class TestResourceLoading(tests.TestCaseInTempDir):
1419
self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1422
class TestLocalTimeOffset(TestCase):
1424
def test_local_time_offset(self):
1425
"""Test that local_time_offset() returns a sane value."""
1426
offset = osutils.local_time_offset()
1427
self.assertTrue(isinstance(offset, int))
1428
# Test that the offset is no more than a eighteen hours in
1430
# Time zone handling is system specific, so it is difficult to
1431
# do more specific tests, but a value outside of this range is
1433
eighteen_hours = 18 * 3600
1434
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1436
def test_local_time_offset_with_timestamp(self):
1437
"""Test that local_time_offset() works with a timestamp."""
1438
offset = osutils.local_time_offset(1000000000.1234567)
1439
self.assertTrue(isinstance(offset, int))
1440
eighteen_hours = 18 * 3600
1441
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1444
class TestShaFileByName(TestCaseInTempDir):
1446
def test_sha_empty(self):
1447
self.build_tree_contents([('foo', '')])
1448
expected_sha = osutils.sha_string('')
1449
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1451
def test_sha_mixed_endings(self):
1452
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1453
self.build_tree_contents([('foo', text)])
1454
expected_sha = osutils.sha_string(text)
1455
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1459
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1461
# This program is free software; you can redistribute it and/or modify
1462
# it under the terms of the GNU General Public License as published by
1463
# the Free Software Foundation; either version 2 of the License, or
1464
# (at your option) any later version.
1466
# This program is distributed in the hope that it will be useful,
1467
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1468
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
1469
# GNU General Public License for more details.
1471
# You should have received a copy of the GNU General Public License
1472
# along with this program; if not, write to the Free Software
1473
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
1476
# NOTE: If update these, please also update the help for global-options in
1477
# bzrlib/help_topics/__init__.py
1480
"""Set of flags that enable different debug behaviour.
1482
These are set with eg ``-Dlock`` on the bzr command line.
1486
* auth - show authentication sections used
1487
* error - show stack traces for all top level exceptions
1488
* evil - capture call sites that do expensive or badly-scaling operations.
1489
* fetch - trace history copying between repositories
1490
* graph - trace graph traversal information
1491
* hashcache - log every time a working file is read to determine its hash
1492
* hooks - trace hook execution
1493
* hpss - trace smart protocol requests and responses
1494
* http - trace http connections, requests and responses
1495
* index - trace major index operations
1496
* knit - trace knit operations
1497
* lock - trace when lockdir locks are taken or released
1498
* merge - emit information for debugging merges
1499
* pack - emit information about pack operations
1505
class TestResourceLoading(TestCaseInTempDir):
1713
1507
def test_resource_string(self):
1714
1508
# test resource in bzrlib
1715
1509
text = osutils.resource_string('bzrlib', 'debug.py')
1716
self.assertContainsRe(text, "debug_flags = set()")
1510
self.assertEquals(_debug_text, text)
1717
1511
# test resource under bzrlib
1718
1512
text = osutils.resource_string('bzrlib.ui', 'text.py')
1719
1513
self.assertContainsRe(text, "class TextUIFactory")
1723
1517
# test unknown resource
1724
1518
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1727
class TestReCompile(tests.TestCase):
1729
def _deprecated_re_compile_checked(self, *args, **kwargs):
1730
return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
1731
osutils.re_compile_checked, *args, **kwargs)
1733
def test_re_compile_checked(self):
1734
r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
1735
self.assertTrue(r.match('aaaa'))
1736
self.assertTrue(r.match('aAaA'))
1738
def test_re_compile_checked_error(self):
1739
# like https://bugs.launchpad.net/bzr/+bug/251352
1741
# Due to possible test isolation error, re.compile is not lazy at
1742
# this point. We re-install lazy compile.
1743
lazy_regex.install_lazy_compile()
1744
err = self.assertRaises(
1745
errors.BzrCommandError,
1746
self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
1748
'Invalid regular expression in test case: '
1749
'"*" nothing to repeat',
1753
class TestDirReader(tests.TestCaseInTempDir):
1755
scenarios = dir_reader_scenarios()
1758
_dir_reader_class = None
1759
_native_to_unicode = None
1762
tests.TestCaseInTempDir.setUp(self)
1763
self.overrideAttr(osutils,
1764
'_selected_dir_reader', self._dir_reader_class())
1766
def _get_ascii_tree(self):
1774
expected_dirblocks = [
1776
[('0file', '0file', 'file'),
1777
('1dir', '1dir', 'directory'),
1778
('2file', '2file', 'file'),
1781
(('1dir', './1dir'),
1782
[('1dir/0file', '0file', 'file'),
1783
('1dir/1dir', '1dir', 'directory'),
1786
(('1dir/1dir', './1dir/1dir'),
1791
return tree, expected_dirblocks
1793
def test_walk_cur_dir(self):
1794
tree, expected_dirblocks = self._get_ascii_tree()
1795
self.build_tree(tree)
1796
result = list(osutils._walkdirs_utf8('.'))
1797
# Filter out stat and abspath
1798
self.assertEqual(expected_dirblocks,
1799
[(dirinfo, [line[0:3] for line in block])
1800
for dirinfo, block in result])
1802
def test_walk_sub_dir(self):
1803
tree, expected_dirblocks = self._get_ascii_tree()
1804
self.build_tree(tree)
1805
# you can search a subdir only, with a supplied prefix.
1806
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1807
# Filter out stat and abspath
1808
self.assertEqual(expected_dirblocks[1:],
1809
[(dirinfo, [line[0:3] for line in block])
1810
for dirinfo, block in result])
1812
def _get_unicode_tree(self):
1813
name0u = u'0file-\xb6'
1814
name1u = u'1dir-\u062c\u0648'
1815
name2u = u'2file-\u0633'
1819
name1u + '/' + name0u,
1820
name1u + '/' + name1u + '/',
1823
name0 = name0u.encode('UTF-8')
1824
name1 = name1u.encode('UTF-8')
1825
name2 = name2u.encode('UTF-8')
1826
expected_dirblocks = [
1828
[(name0, name0, 'file', './' + name0u),
1829
(name1, name1, 'directory', './' + name1u),
1830
(name2, name2, 'file', './' + name2u),
1833
((name1, './' + name1u),
1834
[(name1 + '/' + name0, name0, 'file', './' + name1u
1836
(name1 + '/' + name1, name1, 'directory', './' + name1u
1840
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1845
return tree, expected_dirblocks
1847
def _filter_out(self, raw_dirblocks):
1848
"""Filter out a walkdirs_utf8 result.
1850
stat field is removed, all native paths are converted to unicode
1852
filtered_dirblocks = []
1853
for dirinfo, block in raw_dirblocks:
1854
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1857
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1858
filtered_dirblocks.append((dirinfo, details))
1859
return filtered_dirblocks
1861
def test_walk_unicode_tree(self):
1862
self.requireFeature(features.UnicodeFilenameFeature)
1863
tree, expected_dirblocks = self._get_unicode_tree()
1864
self.build_tree(tree)
1865
result = list(osutils._walkdirs_utf8('.'))
1866
self.assertEqual(expected_dirblocks, self._filter_out(result))
1868
def test_symlink(self):
1869
self.requireFeature(features.SymlinkFeature)
1870
self.requireFeature(features.UnicodeFilenameFeature)
1871
target = u'target\N{Euro Sign}'
1872
link_name = u'l\N{Euro Sign}nk'
1873
os.symlink(target, link_name)
1874
target_utf8 = target.encode('UTF-8')
1875
link_name_utf8 = link_name.encode('UTF-8')
1876
expected_dirblocks = [
1878
[(link_name_utf8, link_name_utf8,
1879
'symlink', './' + link_name),],
1881
result = list(osutils._walkdirs_utf8('.'))
1882
self.assertEqual(expected_dirblocks, self._filter_out(result))
1885
class TestReadLink(tests.TestCaseInTempDir):
1886
"""Exposes os.readlink() problems and the osutils solution.
1888
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1889
unicode string will be returned if a unicode string is passed.
1891
But prior python versions failed to properly encode the passed unicode
1894
_test_needs_features = [features.SymlinkFeature, features.UnicodeFilenameFeature]
1897
super(tests.TestCaseInTempDir, self).setUp()
1898
self.link = u'l\N{Euro Sign}ink'
1899
self.target = u'targe\N{Euro Sign}t'
1900
os.symlink(self.target, self.link)
1902
def test_os_readlink_link_encoding(self):
1903
self.assertEquals(self.target, os.readlink(self.link))
1905
def test_os_readlink_link_decoding(self):
1906
self.assertEquals(self.target.encode(osutils._fs_enc),
1907
os.readlink(self.link.encode(osutils._fs_enc)))
1910
class TestConcurrency(tests.TestCase):
1913
super(TestConcurrency, self).setUp()
1914
self.overrideAttr(osutils, '_cached_local_concurrency')
1916
def test_local_concurrency(self):
1917
concurrency = osutils.local_concurrency()
1918
self.assertIsInstance(concurrency, int)
1920
def test_local_concurrency_environment_variable(self):
1921
self.overrideEnv('BZR_CONCURRENCY', '2')
1922
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1923
self.overrideEnv('BZR_CONCURRENCY', '3')
1924
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1925
self.overrideEnv('BZR_CONCURRENCY', 'foo')
1926
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1928
def test_option_concurrency(self):
1929
self.overrideEnv('BZR_CONCURRENCY', '1')
1930
self.run_bzr('rocks --concurrency 42')
1931
# Command line overrides environment variable
1932
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1933
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1936
class TestFailedToLoadExtension(tests.TestCase):
1938
def _try_loading(self):
1940
import bzrlib._fictional_extension_py
1941
except ImportError, e:
1942
osutils.failed_to_load_extension(e)
1946
super(TestFailedToLoadExtension, self).setUp()
1947
self.overrideAttr(osutils, '_extension_load_failures', [])
1949
def test_failure_to_load(self):
1951
self.assertLength(1, osutils._extension_load_failures)
1952
self.assertEquals(osutils._extension_load_failures[0],
1953
"No module named _fictional_extension_py")
1955
def test_report_extension_load_failures_no_warning(self):
1956
self.assertTrue(self._try_loading())
1957
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
1958
# it used to give a Python warning; it no longer does
1959
self.assertLength(0, warnings)
1961
def test_report_extension_load_failures_message(self):
1963
trace.push_log_file(log)
1964
self.assertTrue(self._try_loading())
1965
osutils.report_extension_load_failures()
1966
self.assertContainsRe(
1968
r"bzr: warning: some compiled extensions could not be loaded; "
1969
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1973
class TestTerminalWidth(tests.TestCase):
1976
tests.TestCase.setUp(self)
1977
self._orig_terminal_size_state = osutils._terminal_size_state
1978
self._orig_first_terminal_size = osutils._first_terminal_size
1979
self.addCleanup(self.restore_osutils_globals)
1980
osutils._terminal_size_state = 'no_data'
1981
osutils._first_terminal_size = None
1983
def restore_osutils_globals(self):
1984
osutils._terminal_size_state = self._orig_terminal_size_state
1985
osutils._first_terminal_size = self._orig_first_terminal_size
1987
def replace_stdout(self, new):
1988
self.overrideAttr(sys, 'stdout', new)
1990
def replace__terminal_size(self, new):
1991
self.overrideAttr(osutils, '_terminal_size', new)
1993
def set_fake_tty(self):
1995
class I_am_a_tty(object):
1999
self.replace_stdout(I_am_a_tty())
2001
def test_default_values(self):
2002
self.assertEqual(80, osutils.default_terminal_width)
2004
def test_defaults_to_BZR_COLUMNS(self):
2005
# BZR_COLUMNS is set by the test framework
2006
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
2007
self.overrideEnv('BZR_COLUMNS', '12')
2008
self.assertEqual(12, osutils.terminal_width())
2010
def test_BZR_COLUMNS_0_no_limit(self):
2011
self.overrideEnv('BZR_COLUMNS', '0')
2012
self.assertEqual(None, osutils.terminal_width())
2014
def test_falls_back_to_COLUMNS(self):
2015
self.overrideEnv('BZR_COLUMNS', None)
2016
self.assertNotEqual('42', os.environ['COLUMNS'])
2018
self.overrideEnv('COLUMNS', '42')
2019
self.assertEqual(42, osutils.terminal_width())
2021
def test_tty_default_without_columns(self):
2022
self.overrideEnv('BZR_COLUMNS', None)
2023
self.overrideEnv('COLUMNS', None)
2025
def terminal_size(w, h):
2029
# We need to override the osutils definition as it depends on the
2030
# running environment that we can't control (PQM running without a
2031
# controlling terminal is one example).
2032
self.replace__terminal_size(terminal_size)
2033
self.assertEqual(42, osutils.terminal_width())
2035
def test_non_tty_default_without_columns(self):
2036
self.overrideEnv('BZR_COLUMNS', None)
2037
self.overrideEnv('COLUMNS', None)
2038
self.replace_stdout(None)
2039
self.assertEqual(None, osutils.terminal_width())
2041
def test_no_TIOCGWINSZ(self):
2042
self.requireFeature(term_ios_feature)
2043
termios = term_ios_feature.module
2044
# bug 63539 is about a termios without TIOCGWINSZ attribute
2046
orig = termios.TIOCGWINSZ
2047
except AttributeError:
2048
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
2051
self.overrideAttr(termios, 'TIOCGWINSZ')
2052
del termios.TIOCGWINSZ
2053
self.overrideEnv('BZR_COLUMNS', None)
2054
self.overrideEnv('COLUMNS', None)
2055
# Whatever the result is, if we don't raise an exception, it's ok.
2056
osutils.terminal_width()
2059
class TestCreationOps(tests.TestCaseInTempDir):
2060
_test_needs_features = [features.chown_feature]
2063
tests.TestCaseInTempDir.setUp(self)
2064
self.overrideAttr(os, 'chown', self._dummy_chown)
2066
# params set by call to _dummy_chown
2067
self.path = self.uid = self.gid = None
2069
def _dummy_chown(self, path, uid, gid):
2070
self.path, self.uid, self.gid = path, uid, gid
2072
def test_copy_ownership_from_path(self):
2073
"""copy_ownership_from_path test with specified src."""
2075
f = open('test_file', 'wt')
2076
osutils.copy_ownership_from_path('test_file', ownsrc)
2079
self.assertEquals(self.path, 'test_file')
2080
self.assertEquals(self.uid, s.st_uid)
2081
self.assertEquals(self.gid, s.st_gid)
2083
def test_copy_ownership_nonesrc(self):
2084
"""copy_ownership_from_path test with src=None."""
2085
f = open('test_file', 'wt')
2086
# should use parent dir for permissions
2087
osutils.copy_ownership_from_path('test_file')
2090
self.assertEquals(self.path, 'test_file')
2091
self.assertEquals(self.uid, s.st_uid)
2092
self.assertEquals(self.gid, s.st_gid)
2095
class TestGetuserUnicode(tests.TestCase):
2097
def test_ascii_user(self):
2098
self.overrideEnv('LOGNAME', 'jrandom')
2099
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2101
def test_unicode_user(self):
2102
ue = osutils.get_user_encoding()
2103
uni_val, env_val = tests.probe_unicode_in_user_encoding()
2105
raise tests.TestSkipped(
2106
'Cannot find a unicode character that works in encoding %s'
2107
% (osutils.get_user_encoding(),))
2108
uni_username = u'jrandom' + uni_val
2109
encoded_username = uni_username.encode(ue)
2110
self.overrideEnv('LOGNAME', encoded_username)
2111
self.assertEqual(uni_username, osutils.getuser_unicode())
2112
self.overrideEnv('LOGNAME', u'jrandom\xb6'.encode(ue))
2113
self.assertEqual(u'jrandom\xb6', osutils.getuser_unicode())
2115
def test_no_username_bug_660174(self):
2116
self.requireFeature(features.win32_feature)
2117
for name in ('LOGNAME', 'USER', 'LNAME', 'USERNAME'):
2118
self.overrideEnv(name, None)
2119
self.assertEqual(u'UNKNOWN', osutils.getuser_unicode())
2122
class TestBackupNames(tests.TestCase):
2125
super(TestBackupNames, self).setUp()
2128
def backup_exists(self, name):
2129
return name in self.backups
2131
def available_backup_name(self, name):
2132
backup_name = osutils.available_backup_name(name, self.backup_exists)
2133
self.backups.append(backup_name)
2136
def assertBackupName(self, expected, name):
2137
self.assertEqual(expected, self.available_backup_name(name))
2139
def test_empty(self):
2140
self.assertBackupName('file.~1~', 'file')
2142
def test_existing(self):
2143
self.available_backup_name('file')
2144
self.available_backup_name('file')
2145
self.assertBackupName('file.~3~', 'file')
2146
# Empty slots are found, this is not a strict requirement and may be
2147
# revisited if we test against all implementations.
2148
self.backups.remove('file.~2~')
2149
self.assertBackupName('file.~2~', 'file')
2152
class TestFindExecutableInPath(tests.TestCase):
2154
def test_windows(self):
2155
if sys.platform != 'win32':
2156
raise tests.TestSkipped('test requires win32')
2157
self.assertTrue(osutils.find_executable_on_path('explorer') is not None)
2159
osutils.find_executable_on_path('explorer.exe') is not None)
2161
osutils.find_executable_on_path('EXPLORER.EXE') is not None)
2163
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2164
self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
2166
def test_other(self):
2167
if sys.platform == 'win32':
2168
raise tests.TestSkipped('test requires non-win32')
2169
self.assertTrue(osutils.find_executable_on_path('sh') is not None)
2171
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)