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
19
from cStringIO import StringIO
28
28
from bzrlib import (
34
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
35
from bzrlib.osutils import (
37
is_inside_or_parent_of_any,
35
41
from bzrlib.tests import (
42
probe_unicode_in_user_encoding,
49
from bzrlib.tests.file_utils import (
41
class _UTF8DirReaderFeature(tests.Feature):
45
from bzrlib import _readdir_pyx
46
self.reader = _readdir_pyx.UTF8DirReader
51
def feature_name(self):
52
return 'bzrlib._readdir_pyx'
54
UTF8DirReaderFeature = _UTF8DirReaderFeature()
56
term_ios_feature = tests.ModuleAvailableFeature('termios')
59
def _already_unicode(s):
63
def _utf8_to_unicode(s):
64
return s.decode('UTF-8')
67
def dir_reader_scenarios():
68
# For each dir reader we define:
70
# - native_to_unicode: a function converting the native_abspath as returned
71
# by DirReader.read_dir to its unicode representation
73
# UnicodeDirReader is the fallback, it should be tested on all platforms.
74
scenarios = [('unicode',
75
dict(_dir_reader_class=osutils.UnicodeDirReader,
76
_native_to_unicode=_already_unicode))]
77
# Some DirReaders are platform specific and even there they may not be
79
if UTF8DirReaderFeature.available():
80
from bzrlib import _readdir_pyx
81
scenarios.append(('utf8',
82
dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
83
_native_to_unicode=_utf8_to_unicode)))
85
if test__walkdirs_win32.win32_readdir_feature.available():
87
from bzrlib import _walkdirs_win32
90
dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
91
_native_to_unicode=_already_unicode)))
97
def load_tests(basic_tests, module, loader):
98
suite = loader.suiteClass()
99
dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
100
basic_tests, tests.condition_isinstance(TestDirReader))
101
tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
102
suite.addTest(remaining_tests)
106
class TestContainsWhitespace(tests.TestCase):
52
from bzrlib.tests.test__walkdirs_win32 import WalkdirsWin32Feature
55
class TestOSUtils(TestCaseInTempDir):
108
57
def test_contains_whitespace(self):
109
58
self.failUnless(osutils.contains_whitespace(u' '))
119
68
self.failIf(osutils.contains_whitespace(u'hellothere'))
120
69
self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
123
class TestRename(tests.TestCaseInTempDir):
125
def create_file(self, filename, content):
126
f = open(filename, 'wb')
132
def _fancy_rename(self, a, b):
133
osutils.fancy_rename(a, b, rename_func=os.rename,
134
unlink_func=os.unlink)
136
71
def test_fancy_rename(self):
137
72
# This should work everywhere
138
self.create_file('a', 'something in a\n')
139
self._fancy_rename('a', 'b')
74
osutils.fancy_rename(a, b,
75
rename_func=os.rename,
76
unlink_func=os.unlink)
78
open('a', 'wb').write('something in a\n')
140
80
self.failIfExists('a')
141
81
self.failUnlessExists('b')
142
82
self.check_file_contents('b', 'something in a\n')
144
self.create_file('a', 'new something in a\n')
145
self._fancy_rename('b', 'a')
84
open('a', 'wb').write('new something in a\n')
147
87
self.check_file_contents('a', 'something in a\n')
149
def test_fancy_rename_fails_source_missing(self):
150
# An exception should be raised, and the target should be left in place
151
self.create_file('target', 'data in target\n')
152
self.assertRaises((IOError, OSError), self._fancy_rename,
153
'missingsource', 'target')
154
self.failUnlessExists('target')
155
self.check_file_contents('target', 'data in target\n')
157
def test_fancy_rename_fails_if_source_and_target_missing(self):
158
self.assertRaises((IOError, OSError), self._fancy_rename,
159
'missingsource', 'missingtarget')
161
89
def test_rename(self):
162
90
# Rename should be semi-atomic on all platforms
163
self.create_file('a', 'something in a\n')
91
open('a', 'wb').write('something in a\n')
164
92
osutils.rename('a', 'b')
165
93
self.failIfExists('a')
166
94
self.failUnlessExists('b')
167
95
self.check_file_contents('b', 'something in a\n')
169
self.create_file('a', 'new something in a\n')
97
open('a', 'wb').write('new something in a\n')
170
98
osutils.rename('b', 'a')
172
100
self.check_file_contents('a', 'something in a\n')
375
271
def test_format_date(self):
376
272
self.assertRaises(errors.UnsupportedTimezoneFormat,
377
273
osutils.format_date, 0, timezone='foo')
378
self.assertIsInstance(osutils.format_date(0), str)
379
self.assertIsInstance(osutils.format_local_date(0), unicode)
380
# Testing for the actual value of the local weekday without
381
# duplicating the code from format_date is difficult.
382
# Instead blackbox.test_locale should check for localized
383
# dates once they do occur in output strings.
385
def test_format_date_with_offset_in_original_timezone(self):
386
self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
387
osutils.format_date_with_offset_in_original_timezone(0))
388
self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
389
osutils.format_date_with_offset_in_original_timezone(100000))
390
self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
391
osutils.format_date_with_offset_in_original_timezone(100000, 7200))
393
def test_local_time_offset(self):
394
"""Test that local_time_offset() returns a sane value."""
395
offset = osutils.local_time_offset()
396
self.assertTrue(isinstance(offset, int))
397
# Test that the offset is no more than a eighteen hours in
399
# Time zone handling is system specific, so it is difficult to
400
# do more specific tests, but a value outside of this range is
402
eighteen_hours = 18 * 3600
403
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
405
def test_local_time_offset_with_timestamp(self):
406
"""Test that local_time_offset() works with a timestamp."""
407
offset = osutils.local_time_offset(1000000000.1234567)
408
self.assertTrue(isinstance(offset, int))
409
eighteen_hours = 18 * 3600
410
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
413
class TestLinks(tests.TestCaseInTempDir):
415
275
def test_dereference_path(self):
416
self.requireFeature(tests.SymlinkFeature)
276
self.requireFeature(SymlinkFeature)
417
277
cwd = osutils.realpath('.')
419
279
bar_path = osutils.pathjoin(cwd, 'bar')
460
320
osutils.make_readonly('dangling')
461
321
osutils.make_writable('dangling')
323
def test_kind_marker(self):
324
self.assertEqual("", osutils.kind_marker("file"))
325
self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
326
self.assertEqual("@", osutils.kind_marker("symlink"))
327
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
463
329
def test_host_os_dereferences_symlinks(self):
464
330
osutils.host_os_dereferences_symlinks()
467
class TestCanonicalRelPath(tests.TestCaseInTempDir):
469
_test_needs_features = [tests.CaseInsCasePresFilenameFeature]
471
def test_canonical_relpath_simple(self):
472
f = file('MixedCaseName', 'w')
474
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
475
self.failUnlessEqual('work/MixedCaseName', actual)
477
def test_canonical_relpath_missing_tail(self):
478
os.mkdir('MixedCaseParent')
479
actual = osutils.canonical_relpath(self.test_base_dir,
480
'mixedcaseparent/nochild')
481
self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
484
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
486
def assertRelpath(self, expected, base, path):
487
actual = osutils._cicp_canonical_relpath(base, path)
488
self.assertEqual(expected, actual)
490
def test_simple(self):
491
self.build_tree(['MixedCaseName'])
492
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
493
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
495
def test_subdir_missing_tail(self):
496
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
497
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
498
self.assertRelpath('MixedCaseParent/a_child', base,
499
'MixedCaseParent/a_child')
500
self.assertRelpath('MixedCaseParent/a_child', base,
501
'MixedCaseParent/A_Child')
502
self.assertRelpath('MixedCaseParent/not_child', base,
503
'MixedCaseParent/not_child')
505
def test_at_root_slash(self):
506
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
508
if osutils.MIN_ABS_PATHLENGTH > 1:
509
raise tests.TestSkipped('relpath requires %d chars'
510
% osutils.MIN_ABS_PATHLENGTH)
511
self.assertRelpath('foo', '/', '/foo')
513
def test_at_root_drive(self):
514
if sys.platform != 'win32':
515
raise tests.TestNotApplicable('we can only test drive-letter relative'
516
' paths on Windows where we have drive'
519
# The specific issue is that when at the root of a drive, 'abspath'
520
# returns "C:/" or just "/". However, the code assumes that abspath
521
# always returns something like "C:/foo" or "/foo" (no trailing slash).
522
self.assertRelpath('foo', 'C:/', 'C:/foo')
523
self.assertRelpath('foo', 'X:/', 'X:/foo')
524
self.assertRelpath('foo', 'X:/', 'X://foo')
527
class TestPumpFile(tests.TestCase):
333
class TestPumpFile(TestCase):
528
334
"""Test pumpfile method."""
531
tests.TestCase.setUp(self)
532
336
# create a test datablock
533
337
self.block_size = 512
534
338
pattern = '0123456789ABCDEF'
541
345
# make sure test data is larger than max read size
542
346
self.assertTrue(self.test_data_len > self.block_size)
544
from_file = file_utils.FakeReadFile(self.test_data)
348
from_file = FakeReadFile(self.test_data)
545
349
to_file = StringIO()
547
351
# read (max / 2) bytes and verify read size wasn't affected
548
352
num_bytes_to_read = self.block_size / 2
549
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
353
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
550
354
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
551
355
self.assertEqual(from_file.get_read_count(), 1)
553
357
# read (max) bytes and verify read size wasn't affected
554
358
num_bytes_to_read = self.block_size
555
359
from_file.reset_read_count()
556
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
360
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
557
361
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
558
362
self.assertEqual(from_file.get_read_count(), 1)
560
364
# read (max + 1) bytes and verify read size was limited
561
365
num_bytes_to_read = self.block_size + 1
562
366
from_file.reset_read_count()
563
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
367
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
564
368
self.assertEqual(from_file.get_max_read_size(), self.block_size)
565
369
self.assertEqual(from_file.get_read_count(), 2)
567
371
# finish reading the rest of the data
568
372
num_bytes_to_read = self.test_data_len - to_file.tell()
569
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
373
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
571
375
# report error if the data wasn't equal (we only report the size due
572
376
# to the length of the data)
637
440
message = "Data not equal. Expected %d bytes, received %d."
638
441
self.fail(message % (len(response_data), self.test_data_len))
640
def test_report_activity(self):
642
def log_activity(length, direction):
643
activity.append((length, direction))
644
from_file = StringIO(self.test_data)
646
osutils.pumpfile(from_file, to_file, buff_size=500,
647
report_activity=log_activity, direction='read')
648
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
649
(36, 'read')], activity)
651
from_file = StringIO(self.test_data)
654
osutils.pumpfile(from_file, to_file, buff_size=500,
655
report_activity=log_activity, direction='write')
656
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
657
(36, 'write')], activity)
659
# And with a limited amount of data
660
from_file = StringIO(self.test_data)
663
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
664
report_activity=log_activity, direction='read')
665
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
669
class TestPumpStringFile(tests.TestCase):
671
def test_empty(self):
673
osutils.pump_string_file("", output)
674
self.assertEqual("", output.getvalue())
676
def test_more_than_segment_size(self):
678
osutils.pump_string_file("123456789", output, 2)
679
self.assertEqual("123456789", output.getvalue())
681
def test_segment_size(self):
683
osutils.pump_string_file("12", output, 2)
684
self.assertEqual("12", output.getvalue())
686
def test_segment_size_multiple(self):
688
osutils.pump_string_file("1234", output, 2)
689
self.assertEqual("1234", output.getvalue())
692
class TestRelpath(tests.TestCase):
694
def test_simple_relpath(self):
695
cwd = osutils.getcwd()
696
subdir = cwd + '/subdir'
697
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
699
def test_deep_relpath(self):
700
cwd = osutils.getcwd()
701
subdir = cwd + '/sub/subsubdir'
702
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
704
def test_not_relative(self):
705
self.assertRaises(errors.PathNotChild,
706
osutils.relpath, 'C:/path', 'H:/path')
707
self.assertRaises(errors.PathNotChild,
708
osutils.relpath, 'C:/', 'H:/path')
711
class TestSafeUnicode(tests.TestCase):
443
class TestSafeUnicode(TestCase):
713
445
def test_from_ascii_string(self):
714
446
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
955
681
self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
958
class TestParentDirectories(tests.TestCaseInTempDir):
959
"""Test osutils.parent_directories()"""
961
def test_parent_directories(self):
962
self.assertEqual([], osutils.parent_directories('a'))
963
self.assertEqual(['a'], osutils.parent_directories('a/b'))
964
self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
967
class TestMacFuncsDirs(tests.TestCaseInTempDir):
684
class TestMacFuncsDirs(TestCaseInTempDir):
968
685
"""Test mac special functions that require directories."""
970
687
def test_getcwd(self):
971
self.requireFeature(tests.UnicodeFilenameFeature)
972
os.mkdir(u'B\xe5gfors')
688
# On Mac, this will actually create Ba\u030agfors
689
# but chdir will still work, because it accepts both paths
691
os.mkdir(u'B\xe5gfors')
693
raise TestSkipped("Unable to create Unicode filename")
973
695
os.chdir(u'B\xe5gfors')
974
696
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
976
698
def test_getcwd_nonnorm(self):
977
self.requireFeature(tests.UnicodeFilenameFeature)
978
699
# Test that _mac_getcwd() will normalize this path
979
os.mkdir(u'Ba\u030agfors')
701
os.mkdir(u'Ba\u030agfors')
703
raise TestSkipped("Unable to create Unicode filename")
980
705
os.chdir(u'Ba\u030agfors')
981
706
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
984
class TestChunksToLines(tests.TestCase):
986
def test_smoketest(self):
987
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
988
osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
989
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
990
osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
992
def test_osutils_binding(self):
993
from bzrlib.tests import test__chunks_to_lines
994
if test__chunks_to_lines.compiled_chunkstolines_feature.available():
995
from bzrlib._chunks_to_lines_pyx import chunks_to_lines
997
from bzrlib._chunks_to_lines_py import chunks_to_lines
998
self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1001
class TestSplitLines(tests.TestCase):
709
class TestSplitLines(TestCase):
1003
711
def test_split_unicode(self):
1004
712
self.assertEqual([u'foo\n', u'bar\xae'],
1136
822
new_dirblock.append((info[0], info[1], info[2], info[4]))
1137
823
dirblock[:] = new_dirblock
825
def test__walkdirs_utf8_selection(self):
826
# Just trigger the function once, to make sure it has selected a real
828
list(osutils._walkdirs_utf8('.'))
829
if WalkdirsWin32Feature.available():
830
# If the compiled form is available, make sure it is used
831
from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
832
self.assertIs(_walkdirs_utf8_win32_find_file,
833
osutils._real_walkdirs_utf8)
834
elif sys.platform == 'win32':
835
self.assertIs(osutils._walkdirs_unicode_to_utf8,
836
osutils._real_walkdirs_utf8)
837
elif osutils._fs_enc.upper() in ('UTF-8', 'ASCII', 'ANSI_X3.4-1968'): # ascii
838
self.assertIs(osutils._walkdirs_fs_utf8,
839
osutils._real_walkdirs_utf8)
841
self.assertIs(osutils._walkdirs_unicode_to_utf8,
842
osutils._real_walkdirs_utf8)
1139
844
def _save_platform_info(self):
1140
self.overrideAttr(win32utils, 'winver')
1141
self.overrideAttr(osutils, '_fs_enc')
1142
self.overrideAttr(osutils, '_selected_dir_reader')
845
cur_winver = win32utils.winver
846
cur_fs_enc = osutils._fs_enc
847
cur_real_walkdirs_utf8 = osutils._real_walkdirs_utf8
849
win32utils.winver = cur_winver
850
osutils._fs_enc = cur_fs_enc
851
osutils._real_walkdirs_utf8 = cur_real_walkdirs_utf8
852
self.addCleanup(restore)
1144
def assertDirReaderIs(self, expected):
854
def assertWalkdirsUtf8Is(self, expected):
1145
855
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
1146
856
# Force it to redetect
1147
osutils._selected_dir_reader = None
857
osutils._real_walkdirs_utf8 = None
1148
858
# Nothing to list, but should still trigger the selection logic
1149
859
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1150
self.assertIsInstance(osutils._selected_dir_reader, expected)
860
self.assertIs(expected, osutils._real_walkdirs_utf8)
1152
862
def test_force_walkdirs_utf8_fs_utf8(self):
1153
self.requireFeature(UTF8DirReaderFeature)
1154
863
self._save_platform_info()
1155
864
win32utils.winver = None # Avoid the win32 detection code
1156
865
osutils._fs_enc = 'UTF-8'
1157
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
866
self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1159
868
def test_force_walkdirs_utf8_fs_ascii(self):
1160
self.requireFeature(UTF8DirReaderFeature)
1161
869
self._save_platform_info()
1162
870
win32utils.winver = None # Avoid the win32 detection code
1163
871
osutils._fs_enc = 'US-ASCII'
1164
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
872
self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1166
874
def test_force_walkdirs_utf8_fs_ANSI(self):
1167
self.requireFeature(UTF8DirReaderFeature)
1168
875
self._save_platform_info()
1169
876
win32utils.winver = None # Avoid the win32 detection code
1170
877
osutils._fs_enc = 'ANSI_X3.4-1968'
1171
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
878
self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1173
880
def test_force_walkdirs_utf8_fs_latin1(self):
1174
881
self._save_platform_info()
1175
882
win32utils.winver = None # Avoid the win32 detection code
1176
883
osutils._fs_enc = 'latin1'
1177
self.assertDirReaderIs(osutils.UnicodeDirReader)
884
self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
1179
886
def test_force_walkdirs_utf8_nt(self):
1180
# Disabled because the thunk of the whole walkdirs api is disabled.
1181
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
887
self.requireFeature(WalkdirsWin32Feature)
1182
888
self._save_platform_info()
1183
889
win32utils.winver = 'Windows NT'
1184
from bzrlib._walkdirs_win32 import Win32ReadDir
1185
self.assertDirReaderIs(Win32ReadDir)
890
from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
891
self.assertWalkdirsUtf8Is(_walkdirs_utf8_win32_find_file)
1187
def test_force_walkdirs_utf8_98(self):
1188
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
893
def test_force_walkdirs_utf8_nt(self):
894
self.requireFeature(WalkdirsWin32Feature)
1189
895
self._save_platform_info()
1190
896
win32utils.winver = 'Windows 98'
1191
self.assertDirReaderIs(osutils.UnicodeDirReader)
897
self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
1193
899
def test_unicode_walkdirs(self):
1194
900
"""Walkdirs should always return unicode paths."""
1195
self.requireFeature(tests.UnicodeFilenameFeature)
1196
901
name0 = u'0file-\xb6'
1197
902
name1 = u'1dir-\u062c\u0648'
1198
903
name2 = u'2file-\u0633'
1616
1329
self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1619
class TestSizeShaFile(tests.TestCaseInTempDir):
1621
def test_sha_empty(self):
1622
self.build_tree_contents([('foo', '')])
1623
expected_sha = osutils.sha_string('')
1625
self.addCleanup(f.close)
1626
size, sha = osutils.size_sha_file(f)
1627
self.assertEqual(0, size)
1628
self.assertEqual(expected_sha, sha)
1630
def test_sha_mixed_endings(self):
1631
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1632
self.build_tree_contents([('foo', text)])
1633
expected_sha = osutils.sha_string(text)
1634
f = open('foo', 'rb')
1635
self.addCleanup(f.close)
1636
size, sha = osutils.size_sha_file(f)
1637
self.assertEqual(38, size)
1638
self.assertEqual(expected_sha, sha)
1641
class TestShaFileByName(tests.TestCaseInTempDir):
1643
def test_sha_empty(self):
1644
self.build_tree_contents([('foo', '')])
1645
expected_sha = osutils.sha_string('')
1646
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1648
def test_sha_mixed_endings(self):
1649
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1650
self.build_tree_contents([('foo', text)])
1651
expected_sha = osutils.sha_string(text)
1652
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1655
class TestResourceLoading(tests.TestCaseInTempDir):
1332
class TestLocalTimeOffset(TestCase):
1334
def test_local_time_offset(self):
1335
"""Test that local_time_offset() returns a sane value."""
1336
offset = osutils.local_time_offset()
1337
self.assertTrue(isinstance(offset, int))
1338
# Test that the offset is no more than a eighteen hours in
1340
# Time zone handling is system specific, so it is difficult to
1341
# do more specific tests, but a value outside of this range is
1343
eighteen_hours = 18 * 3600
1344
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1346
def test_local_time_offset_with_timestamp(self):
1347
"""Test that local_time_offset() works with a timestamp."""
1348
offset = osutils.local_time_offset(1000000000.1234567)
1349
self.assertTrue(isinstance(offset, int))
1350
eighteen_hours = 18 * 3600
1351
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1354
class TestShaFileByName(TestCaseInTempDir):
1356
def test_sha_empty(self):
1357
self.build_tree_contents([('foo', '')])
1358
expected_sha = osutils.sha_string('')
1359
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1361
def test_sha_mixed_endings(self):
1362
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1363
self.build_tree_contents([('foo', text)])
1364
expected_sha = osutils.sha_string(text)
1365
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1369
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1371
# This program is free software; you can redistribute it and/or modify
1372
# it under the terms of the GNU General Public License as published by
1373
# the Free Software Foundation; either version 2 of the License, or
1374
# (at your option) any later version.
1376
# This program is distributed in the hope that it will be useful,
1377
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1378
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
1379
# GNU General Public License for more details.
1381
# You should have received a copy of the GNU General Public License
1382
# along with this program; if not, write to the Free Software
1383
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
1386
# NOTE: If update these, please also update the help for global-options in
1387
# bzrlib/help_topics/__init__.py
1390
"""Set of flags that enable different debug behaviour.
1392
These are set with eg ``-Dlock`` on the bzr command line.
1396
* auth - show authentication sections used
1397
* error - show stack traces for all top level exceptions
1398
* evil - capture call sites that do expensive or badly-scaling operations.
1399
* fetch - trace history copying between repositories
1400
* graph - trace graph traversal information
1401
* hashcache - log every time a working file is read to determine its hash
1402
* hooks - trace hook execution
1403
* hpss - trace smart protocol requests and responses
1404
* http - trace http connections, requests and responses
1405
* index - trace major index operations
1406
* knit - trace knit operations
1407
* lock - trace when lockdir locks are taken or released
1408
* merge - emit information for debugging merges
1409
* pack - emit information about pack operations
1415
class TestResourceLoading(TestCaseInTempDir):
1657
1417
def test_resource_string(self):
1658
1418
# test resource in bzrlib
1659
1419
text = osutils.resource_string('bzrlib', 'debug.py')
1660
self.assertContainsRe(text, "debug_flags = set()")
1420
self.assertEquals(_debug_text, text)
1661
1421
# test resource under bzrlib
1662
1422
text = osutils.resource_string('bzrlib.ui', 'text.py')
1663
1423
self.assertContainsRe(text, "class TextUIFactory")
1667
1427
# test unknown resource
1668
1428
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1671
class TestReCompile(tests.TestCase):
1673
def test_re_compile_checked(self):
1674
r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1675
self.assertTrue(r.match('aaaa'))
1676
self.assertTrue(r.match('aAaA'))
1678
def test_re_compile_checked_error(self):
1679
# like https://bugs.launchpad.net/bzr/+bug/251352
1680
err = self.assertRaises(
1681
errors.BzrCommandError,
1682
osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1684
"Invalid regular expression in test case: '*': "
1685
"nothing to repeat",
1689
class TestDirReader(tests.TestCaseInTempDir):
1692
_dir_reader_class = None
1693
_native_to_unicode = None
1696
tests.TestCaseInTempDir.setUp(self)
1697
self.overrideAttr(osutils,
1698
'_selected_dir_reader', self._dir_reader_class())
1700
def _get_ascii_tree(self):
1708
expected_dirblocks = [
1710
[('0file', '0file', 'file'),
1711
('1dir', '1dir', 'directory'),
1712
('2file', '2file', 'file'),
1715
(('1dir', './1dir'),
1716
[('1dir/0file', '0file', 'file'),
1717
('1dir/1dir', '1dir', 'directory'),
1720
(('1dir/1dir', './1dir/1dir'),
1725
return tree, expected_dirblocks
1727
def test_walk_cur_dir(self):
1728
tree, expected_dirblocks = self._get_ascii_tree()
1729
self.build_tree(tree)
1730
result = list(osutils._walkdirs_utf8('.'))
1731
# Filter out stat and abspath
1732
self.assertEqual(expected_dirblocks,
1733
[(dirinfo, [line[0:3] for line in block])
1734
for dirinfo, block in result])
1736
def test_walk_sub_dir(self):
1737
tree, expected_dirblocks = self._get_ascii_tree()
1738
self.build_tree(tree)
1739
# you can search a subdir only, with a supplied prefix.
1740
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1741
# Filter out stat and abspath
1742
self.assertEqual(expected_dirblocks[1:],
1743
[(dirinfo, [line[0:3] for line in block])
1744
for dirinfo, block in result])
1746
def _get_unicode_tree(self):
1747
name0u = u'0file-\xb6'
1748
name1u = u'1dir-\u062c\u0648'
1749
name2u = u'2file-\u0633'
1753
name1u + '/' + name0u,
1754
name1u + '/' + name1u + '/',
1757
name0 = name0u.encode('UTF-8')
1758
name1 = name1u.encode('UTF-8')
1759
name2 = name2u.encode('UTF-8')
1760
expected_dirblocks = [
1762
[(name0, name0, 'file', './' + name0u),
1763
(name1, name1, 'directory', './' + name1u),
1764
(name2, name2, 'file', './' + name2u),
1767
((name1, './' + name1u),
1768
[(name1 + '/' + name0, name0, 'file', './' + name1u
1770
(name1 + '/' + name1, name1, 'directory', './' + name1u
1774
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1779
return tree, expected_dirblocks
1781
def _filter_out(self, raw_dirblocks):
1782
"""Filter out a walkdirs_utf8 result.
1784
stat field is removed, all native paths are converted to unicode
1786
filtered_dirblocks = []
1787
for dirinfo, block in raw_dirblocks:
1788
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1791
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1792
filtered_dirblocks.append((dirinfo, details))
1793
return filtered_dirblocks
1795
def test_walk_unicode_tree(self):
1796
self.requireFeature(tests.UnicodeFilenameFeature)
1797
tree, expected_dirblocks = self._get_unicode_tree()
1798
self.build_tree(tree)
1799
result = list(osutils._walkdirs_utf8('.'))
1800
self.assertEqual(expected_dirblocks, self._filter_out(result))
1802
def test_symlink(self):
1803
self.requireFeature(tests.SymlinkFeature)
1804
self.requireFeature(tests.UnicodeFilenameFeature)
1805
target = u'target\N{Euro Sign}'
1806
link_name = u'l\N{Euro Sign}nk'
1807
os.symlink(target, link_name)
1808
target_utf8 = target.encode('UTF-8')
1809
link_name_utf8 = link_name.encode('UTF-8')
1810
expected_dirblocks = [
1812
[(link_name_utf8, link_name_utf8,
1813
'symlink', './' + link_name),],
1815
result = list(osutils._walkdirs_utf8('.'))
1816
self.assertEqual(expected_dirblocks, self._filter_out(result))
1819
class TestReadLink(tests.TestCaseInTempDir):
1820
"""Exposes os.readlink() problems and the osutils solution.
1822
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1823
unicode string will be returned if a unicode string is passed.
1825
But prior python versions failed to properly encode the passed unicode
1828
_test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1831
super(tests.TestCaseInTempDir, self).setUp()
1832
self.link = u'l\N{Euro Sign}ink'
1833
self.target = u'targe\N{Euro Sign}t'
1834
os.symlink(self.target, self.link)
1836
def test_os_readlink_link_encoding(self):
1837
if sys.version_info < (2, 6):
1838
self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1840
self.assertEquals(self.target, os.readlink(self.link))
1842
def test_os_readlink_link_decoding(self):
1843
self.assertEquals(self.target.encode(osutils._fs_enc),
1844
os.readlink(self.link.encode(osutils._fs_enc)))
1847
class TestConcurrency(tests.TestCase):
1850
super(TestConcurrency, self).setUp()
1851
self.overrideAttr(osutils, '_cached_local_concurrency')
1853
def test_local_concurrency(self):
1854
concurrency = osutils.local_concurrency()
1855
self.assertIsInstance(concurrency, int)
1857
def test_local_concurrency_environment_variable(self):
1858
os.environ['BZR_CONCURRENCY'] = '2'
1859
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1860
os.environ['BZR_CONCURRENCY'] = '3'
1861
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1862
os.environ['BZR_CONCURRENCY'] = 'foo'
1863
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1865
def test_option_concurrency(self):
1866
os.environ['BZR_CONCURRENCY'] = '1'
1867
self.run_bzr('rocks --concurrency 42')
1868
# Command line overrides envrionment variable
1869
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1870
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1873
class TestFailedToLoadExtension(tests.TestCase):
1875
def _try_loading(self):
1877
import bzrlib._fictional_extension_py
1878
except ImportError, e:
1879
osutils.failed_to_load_extension(e)
1883
super(TestFailedToLoadExtension, self).setUp()
1884
self.overrideAttr(osutils, '_extension_load_failures', [])
1886
def test_failure_to_load(self):
1888
self.assertLength(1, osutils._extension_load_failures)
1889
self.assertEquals(osutils._extension_load_failures[0],
1890
"No module named _fictional_extension_py")
1892
def test_report_extension_load_failures_no_warning(self):
1893
self.assertTrue(self._try_loading())
1894
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
1895
# it used to give a Python warning; it no longer does
1896
self.assertLength(0, warnings)
1898
def test_report_extension_load_failures_message(self):
1900
trace.push_log_file(log)
1901
self.assertTrue(self._try_loading())
1902
osutils.report_extension_load_failures()
1903
self.assertContainsRe(
1905
r"bzr: warning: some compiled extensions could not be loaded; "
1906
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1910
class TestTerminalWidth(tests.TestCase):
1912
def replace_stdout(self, new):
1913
self.overrideAttr(sys, 'stdout', new)
1915
def replace__terminal_size(self, new):
1916
self.overrideAttr(osutils, '_terminal_size', new)
1918
def set_fake_tty(self):
1920
class I_am_a_tty(object):
1924
self.replace_stdout(I_am_a_tty())
1926
def test_default_values(self):
1927
self.assertEqual(80, osutils.default_terminal_width)
1929
def test_defaults_to_BZR_COLUMNS(self):
1930
# BZR_COLUMNS is set by the test framework
1931
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
1932
os.environ['BZR_COLUMNS'] = '12'
1933
self.assertEqual(12, osutils.terminal_width())
1935
def test_falls_back_to_COLUMNS(self):
1936
del os.environ['BZR_COLUMNS']
1937
self.assertNotEqual('42', os.environ['COLUMNS'])
1939
os.environ['COLUMNS'] = '42'
1940
self.assertEqual(42, osutils.terminal_width())
1942
def test_tty_default_without_columns(self):
1943
del os.environ['BZR_COLUMNS']
1944
del os.environ['COLUMNS']
1946
def terminal_size(w, h):
1950
# We need to override the osutils definition as it depends on the
1951
# running environment that we can't control (PQM running without a
1952
# controlling terminal is one example).
1953
self.replace__terminal_size(terminal_size)
1954
self.assertEqual(42, osutils.terminal_width())
1956
def test_non_tty_default_without_columns(self):
1957
del os.environ['BZR_COLUMNS']
1958
del os.environ['COLUMNS']
1959
self.replace_stdout(None)
1960
self.assertEqual(None, osutils.terminal_width())
1962
def test_no_TIOCGWINSZ(self):
1963
self.requireFeature(term_ios_feature)
1964
termios = term_ios_feature.module
1965
# bug 63539 is about a termios without TIOCGWINSZ attribute
1967
orig = termios.TIOCGWINSZ
1968
except AttributeError:
1969
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
1972
self.overrideAttr(termios, 'TIOCGWINSZ')
1973
del termios.TIOCGWINSZ
1974
del os.environ['BZR_COLUMNS']
1975
del os.environ['COLUMNS']
1976
# Whatever the result is, if we don't raise an exception, it's ok.
1977
osutils.terminal_width()