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
29
26
from bzrlib import (
30
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
38
31
from bzrlib.tests import (
43
from bzrlib.tests.scenarios import load_tests_apply_scenarios
46
class _UTF8DirReaderFeature(features.Feature):
50
from bzrlib import _readdir_pyx
51
self.reader = _readdir_pyx.UTF8DirReader
56
def feature_name(self):
57
return 'bzrlib._readdir_pyx'
59
UTF8DirReaderFeature = features.ModuleAvailableFeature('bzrlib._readdir_pyx')
61
term_ios_feature = features.ModuleAvailableFeature('termios')
64
def _already_unicode(s):
68
def _utf8_to_unicode(s):
69
return s.decode('UTF-8')
72
def dir_reader_scenarios():
73
# For each dir reader we define:
75
# - native_to_unicode: a function converting the native_abspath as returned
76
# by DirReader.read_dir to its unicode representation
78
# UnicodeDirReader is the fallback, it should be tested on all platforms.
79
scenarios = [('unicode',
80
dict(_dir_reader_class=osutils.UnicodeDirReader,
81
_native_to_unicode=_already_unicode))]
82
# Some DirReaders are platform specific and even there they may not be
84
if UTF8DirReaderFeature.available():
85
from bzrlib import _readdir_pyx
86
scenarios.append(('utf8',
87
dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
88
_native_to_unicode=_utf8_to_unicode)))
90
if test__walkdirs_win32.win32_readdir_feature.available():
92
from bzrlib import _walkdirs_win32
95
dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
96
_native_to_unicode=_already_unicode)))
102
load_tests = load_tests_apply_scenarios
105
class TestContainsWhitespace(tests.TestCase):
107
def test_contains_whitespace(self):
108
self.assertTrue(osutils.contains_whitespace(u' '))
109
self.assertTrue(osutils.contains_whitespace(u'hello there'))
110
self.assertTrue(osutils.contains_whitespace(u'hellothere\n'))
111
self.assertTrue(osutils.contains_whitespace(u'hello\nthere'))
112
self.assertTrue(osutils.contains_whitespace(u'hello\rthere'))
113
self.assertTrue(osutils.contains_whitespace(u'hello\tthere'))
115
# \xa0 is "Non-breaking-space" which on some python locales thinks it
116
# is whitespace, but we do not.
117
self.assertFalse(osutils.contains_whitespace(u''))
118
self.assertFalse(osutils.contains_whitespace(u'hellothere'))
119
self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
122
class TestRename(tests.TestCaseInTempDir):
124
def create_file(self, filename, content):
125
f = open(filename, 'wb')
131
def _fancy_rename(self, a, b):
132
osutils.fancy_rename(a, b, rename_func=os.rename,
133
unlink_func=os.unlink)
39
class TestOSUtils(TestCaseInTempDir):
135
41
def test_fancy_rename(self):
136
42
# This should work everywhere
137
self.create_file('a', 'something in a\n')
138
self._fancy_rename('a', 'b')
139
self.assertPathDoesNotExist('a')
140
self.assertPathExists('b')
44
osutils.fancy_rename(a, b,
45
rename_func=os.rename,
46
unlink_func=os.unlink)
48
open('a', 'wb').write('something in a\n')
50
self.failIfExists('a')
51
self.failUnlessExists('b')
141
52
self.check_file_contents('b', 'something in a\n')
143
self.create_file('a', 'new something in a\n')
144
self._fancy_rename('b', 'a')
54
open('a', 'wb').write('new something in a\n')
146
57
self.check_file_contents('a', 'something in a\n')
148
def test_fancy_rename_fails_source_missing(self):
149
# An exception should be raised, and the target should be left in place
150
self.create_file('target', 'data in target\n')
151
self.assertRaises((IOError, OSError), self._fancy_rename,
152
'missingsource', 'target')
153
self.assertPathExists('target')
154
self.check_file_contents('target', 'data in target\n')
156
def test_fancy_rename_fails_if_source_and_target_missing(self):
157
self.assertRaises((IOError, OSError), self._fancy_rename,
158
'missingsource', 'missingtarget')
160
59
def test_rename(self):
161
60
# Rename should be semi-atomic on all platforms
162
self.create_file('a', 'something in a\n')
61
open('a', 'wb').write('something in a\n')
163
62
osutils.rename('a', 'b')
164
self.assertPathDoesNotExist('a')
165
self.assertPathExists('b')
63
self.failIfExists('a')
64
self.failUnlessExists('b')
166
65
self.check_file_contents('b', 'something in a\n')
168
self.create_file('a', 'new something in a\n')
67
open('a', 'wb').write('new something in a\n')
169
68
osutils.rename('b', 'a')
171
70
self.check_file_contents('a', 'something in a\n')
173
72
# TODO: test fancy_rename using a MemoryTransport
175
def test_rename_change_case(self):
176
# on Windows we should be able to change filename case by rename
177
self.build_tree(['a', 'b/'])
178
osutils.rename('a', 'A')
179
osutils.rename('b', 'B')
180
# we can't use failUnlessExists on case-insensitive filesystem
181
# so try to check shape of the tree
182
shape = sorted(os.listdir('.'))
183
self.assertEqual(['A', 'B'], shape)
185
def test_rename_exception(self):
187
osutils.rename('nonexistent_path', 'different_nonexistent_path')
189
self.assertEqual(e.old_filename, 'nonexistent_path')
190
self.assertEqual(e.new_filename, 'different_nonexistent_path')
191
self.assertTrue('nonexistent_path' in e.strerror)
192
self.assertTrue('different_nonexistent_path' in e.strerror)
195
class TestRandChars(tests.TestCase):
197
74
def test_01_rand_chars_empty(self):
198
75
result = osutils.rand_chars(0)
199
76
self.assertEqual(result, '')
399
197
self.assertFormatedDelta('1 second in the future', -1)
400
198
self.assertFormatedDelta('2 seconds in the future', -2)
402
def test_format_date(self):
403
self.assertRaises(errors.UnsupportedTimezoneFormat,
404
osutils.format_date, 0, timezone='foo')
405
self.assertIsInstance(osutils.format_date(0), str)
406
self.assertIsInstance(osutils.format_local_date(0), unicode)
407
# Testing for the actual value of the local weekday without
408
# duplicating the code from format_date is difficult.
409
# Instead blackbox.test_locale should check for localized
410
# dates once they do occur in output strings.
412
def test_format_date_with_offset_in_original_timezone(self):
413
self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
414
osutils.format_date_with_offset_in_original_timezone(0))
415
self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
416
osutils.format_date_with_offset_in_original_timezone(100000))
417
self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
418
osutils.format_date_with_offset_in_original_timezone(100000, 7200))
420
def test_local_time_offset(self):
421
"""Test that local_time_offset() returns a sane value."""
422
offset = osutils.local_time_offset()
423
self.assertTrue(isinstance(offset, int))
424
# Test that the offset is no more than a eighteen hours in
426
# Time zone handling is system specific, so it is difficult to
427
# do more specific tests, but a value outside of this range is
429
eighteen_hours = 18 * 3600
430
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
432
def test_local_time_offset_with_timestamp(self):
433
"""Test that local_time_offset() works with a timestamp."""
434
offset = osutils.local_time_offset(1000000000.1234567)
435
self.assertTrue(isinstance(offset, int))
436
eighteen_hours = 18 * 3600
437
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
440
class TestFdatasync(tests.TestCaseInTempDir):
442
def do_fdatasync(self):
443
f = tempfile.NamedTemporaryFile()
444
osutils.fdatasync(f.fileno())
448
def raise_eopnotsupp(*args, **kwargs):
449
raise IOError(errno.EOPNOTSUPP, os.strerror(errno.EOPNOTSUPP))
452
def raise_enotsup(*args, **kwargs):
453
raise IOError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
455
def test_fdatasync_handles_system_function(self):
456
self.overrideAttr(os, "fdatasync")
459
def test_fdatasync_handles_no_fdatasync_no_fsync(self):
460
self.overrideAttr(os, "fdatasync")
461
self.overrideAttr(os, "fsync")
464
def test_fdatasync_handles_no_EOPNOTSUPP(self):
465
self.overrideAttr(errno, "EOPNOTSUPP")
468
def test_fdatasync_catches_ENOTSUP(self):
469
enotsup = getattr(errno, "ENOTSUP", None)
471
raise tests.TestNotApplicable("No ENOTSUP on this platform")
472
self.overrideAttr(os, "fdatasync", self.raise_enotsup)
475
def test_fdatasync_catches_EOPNOTSUPP(self):
476
enotsup = getattr(errno, "EOPNOTSUPP", None)
478
raise tests.TestNotApplicable("No EOPNOTSUPP on this platform")
479
self.overrideAttr(os, "fdatasync", self.raise_eopnotsupp)
483
class TestLinks(tests.TestCaseInTempDir):
485
def test_dereference_path(self):
486
self.requireFeature(features.SymlinkFeature)
487
cwd = osutils.realpath('.')
489
bar_path = osutils.pathjoin(cwd, 'bar')
490
# Using './' to avoid bug #1213894 (first path component not
491
# dereferenced) in Python 2.4.1 and earlier
492
self.assertEqual(bar_path, osutils.realpath('./bar'))
493
os.symlink('bar', 'foo')
494
self.assertEqual(bar_path, osutils.realpath('./foo'))
496
# Does not dereference terminal symlinks
497
foo_path = osutils.pathjoin(cwd, 'foo')
498
self.assertEqual(foo_path, osutils.dereference_path('./foo'))
500
# Dereferences parent symlinks
502
baz_path = osutils.pathjoin(bar_path, 'baz')
503
self.assertEqual(baz_path, osutils.dereference_path('./foo/baz'))
505
# Dereferences parent symlinks that are the first path element
506
self.assertEqual(baz_path, osutils.dereference_path('foo/baz'))
508
# Dereferences parent symlinks in absolute paths
509
foo_baz_path = osutils.pathjoin(foo_path, 'baz')
510
self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
512
def test_changing_access(self):
513
f = file('file', 'w')
517
# Make a file readonly
518
osutils.make_readonly('file')
519
mode = os.lstat('file').st_mode
520
self.assertEqual(mode, mode & 0777555)
522
# Make a file writable
523
osutils.make_writable('file')
524
mode = os.lstat('file').st_mode
525
self.assertEqual(mode, mode | 0200)
527
if osutils.has_symlinks():
528
# should not error when handed a symlink
529
os.symlink('nonexistent', 'dangling')
530
osutils.make_readonly('dangling')
531
osutils.make_writable('dangling')
533
def test_host_os_dereferences_symlinks(self):
534
osutils.host_os_dereferences_symlinks()
537
class TestCanonicalRelPath(tests.TestCaseInTempDir):
539
_test_needs_features = [features.CaseInsCasePresFilenameFeature]
541
def test_canonical_relpath_simple(self):
542
f = file('MixedCaseName', 'w')
544
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
545
self.assertEqual('work/MixedCaseName', actual)
547
def test_canonical_relpath_missing_tail(self):
548
os.mkdir('MixedCaseParent')
549
actual = osutils.canonical_relpath(self.test_base_dir,
550
'mixedcaseparent/nochild')
551
self.assertEqual('work/MixedCaseParent/nochild', actual)
554
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
556
def assertRelpath(self, expected, base, path):
557
actual = osutils._cicp_canonical_relpath(base, path)
558
self.assertEqual(expected, actual)
560
def test_simple(self):
561
self.build_tree(['MixedCaseName'])
562
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
563
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
565
def test_subdir_missing_tail(self):
566
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
567
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
568
self.assertRelpath('MixedCaseParent/a_child', base,
569
'MixedCaseParent/a_child')
570
self.assertRelpath('MixedCaseParent/a_child', base,
571
'MixedCaseParent/A_Child')
572
self.assertRelpath('MixedCaseParent/not_child', base,
573
'MixedCaseParent/not_child')
575
def test_at_root_slash(self):
576
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
578
if osutils.MIN_ABS_PATHLENGTH > 1:
579
raise tests.TestSkipped('relpath requires %d chars'
580
% osutils.MIN_ABS_PATHLENGTH)
581
self.assertRelpath('foo', '/', '/foo')
583
def test_at_root_drive(self):
584
if sys.platform != 'win32':
585
raise tests.TestNotApplicable('we can only test drive-letter relative'
586
' paths on Windows where we have drive'
589
# The specific issue is that when at the root of a drive, 'abspath'
590
# returns "C:/" or just "/". However, the code assumes that abspath
591
# always returns something like "C:/foo" or "/foo" (no trailing slash).
592
self.assertRelpath('foo', 'C:/', 'C:/foo')
593
self.assertRelpath('foo', 'X:/', 'X:/foo')
594
self.assertRelpath('foo', 'X:/', 'X://foo')
597
class TestPumpFile(tests.TestCase):
598
"""Test pumpfile method."""
601
super(TestPumpFile, self).setUp()
602
# create a test datablock
603
self.block_size = 512
604
pattern = '0123456789ABCDEF'
605
self.test_data = pattern * (3 * self.block_size / len(pattern))
606
self.test_data_len = len(self.test_data)
608
def test_bracket_block_size(self):
609
"""Read data in blocks with the requested read size bracketing the
611
# make sure test data is larger than max read size
612
self.assertTrue(self.test_data_len > self.block_size)
614
from_file = file_utils.FakeReadFile(self.test_data)
617
# read (max / 2) bytes and verify read size wasn't affected
618
num_bytes_to_read = self.block_size / 2
619
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
620
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
621
self.assertEqual(from_file.get_read_count(), 1)
623
# read (max) bytes and verify read size wasn't affected
624
num_bytes_to_read = self.block_size
625
from_file.reset_read_count()
626
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
627
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
628
self.assertEqual(from_file.get_read_count(), 1)
630
# read (max + 1) bytes and verify read size was limited
631
num_bytes_to_read = self.block_size + 1
632
from_file.reset_read_count()
633
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
634
self.assertEqual(from_file.get_max_read_size(), self.block_size)
635
self.assertEqual(from_file.get_read_count(), 2)
637
# finish reading the rest of the data
638
num_bytes_to_read = self.test_data_len - to_file.tell()
639
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
641
# report error if the data wasn't equal (we only report the size due
642
# to the length of the data)
643
response_data = to_file.getvalue()
644
if response_data != self.test_data:
645
message = "Data not equal. Expected %d bytes, received %d."
646
self.fail(message % (len(response_data), self.test_data_len))
648
def test_specified_size(self):
649
"""Request a transfer larger than the maximum block size and verify
650
that the maximum read doesn't exceed the block_size."""
651
# make sure test data is larger than max read size
652
self.assertTrue(self.test_data_len > self.block_size)
654
# retrieve data in blocks
655
from_file = file_utils.FakeReadFile(self.test_data)
657
osutils.pumpfile(from_file, to_file, self.test_data_len,
660
# verify read size was equal to the maximum read size
661
self.assertTrue(from_file.get_max_read_size() > 0)
662
self.assertEqual(from_file.get_max_read_size(), self.block_size)
663
self.assertEqual(from_file.get_read_count(), 3)
665
# report error if the data wasn't equal (we only report the size due
666
# to the length of the data)
667
response_data = to_file.getvalue()
668
if response_data != self.test_data:
669
message = "Data not equal. Expected %d bytes, received %d."
670
self.fail(message % (len(response_data), self.test_data_len))
672
def test_to_eof(self):
673
"""Read to end-of-file and verify that the reads are not larger than
674
the maximum read size."""
675
# make sure test data is larger than max read size
676
self.assertTrue(self.test_data_len > self.block_size)
678
# retrieve data to EOF
679
from_file = file_utils.FakeReadFile(self.test_data)
681
osutils.pumpfile(from_file, to_file, -1, self.block_size)
683
# verify read size was equal to the maximum read size
684
self.assertEqual(from_file.get_max_read_size(), self.block_size)
685
self.assertEqual(from_file.get_read_count(), 4)
687
# report error if the data wasn't equal (we only report the size due
688
# to the length of the data)
689
response_data = to_file.getvalue()
690
if response_data != self.test_data:
691
message = "Data not equal. Expected %d bytes, received %d."
692
self.fail(message % (len(response_data), self.test_data_len))
694
def test_defaults(self):
695
"""Verifies that the default arguments will read to EOF -- this
696
test verifies that any existing usages of pumpfile will not be broken
697
with this new version."""
698
# retrieve data using default (old) pumpfile method
699
from_file = file_utils.FakeReadFile(self.test_data)
701
osutils.pumpfile(from_file, to_file)
703
# report error if the data wasn't equal (we only report the size due
704
# to the length of the data)
705
response_data = to_file.getvalue()
706
if response_data != self.test_data:
707
message = "Data not equal. Expected %d bytes, received %d."
708
self.fail(message % (len(response_data), self.test_data_len))
710
def test_report_activity(self):
712
def log_activity(length, direction):
713
activity.append((length, direction))
714
from_file = StringIO(self.test_data)
716
osutils.pumpfile(from_file, to_file, buff_size=500,
717
report_activity=log_activity, direction='read')
718
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
719
(36, 'read')], activity)
721
from_file = StringIO(self.test_data)
724
osutils.pumpfile(from_file, to_file, buff_size=500,
725
report_activity=log_activity, direction='write')
726
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
727
(36, 'write')], activity)
729
# And with a limited amount of data
730
from_file = StringIO(self.test_data)
733
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
734
report_activity=log_activity, direction='read')
735
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
739
class TestPumpStringFile(tests.TestCase):
741
def test_empty(self):
743
osutils.pump_string_file("", output)
744
self.assertEqual("", output.getvalue())
746
def test_more_than_segment_size(self):
748
osutils.pump_string_file("123456789", output, 2)
749
self.assertEqual("123456789", output.getvalue())
751
def test_segment_size(self):
753
osutils.pump_string_file("12", output, 2)
754
self.assertEqual("12", output.getvalue())
756
def test_segment_size_multiple(self):
758
osutils.pump_string_file("1234", output, 2)
759
self.assertEqual("1234", output.getvalue())
762
class TestRelpath(tests.TestCase):
764
def test_simple_relpath(self):
765
cwd = osutils.getcwd()
766
subdir = cwd + '/subdir'
767
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
769
def test_deep_relpath(self):
770
cwd = osutils.getcwd()
771
subdir = cwd + '/sub/subsubdir'
772
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
774
def test_not_relative(self):
775
self.assertRaises(errors.PathNotChild,
776
osutils.relpath, 'C:/path', 'H:/path')
777
self.assertRaises(errors.PathNotChild,
778
osutils.relpath, 'C:/', 'H:/path')
781
class TestSafeUnicode(tests.TestCase):
201
class TestSafeUnicode(TestCase):
783
203
def test_from_ascii_string(self):
784
204
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
793
213
self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
795
215
def test_bad_utf8_string(self):
796
self.assertRaises(errors.BzrBadParameterNotUnicode,
216
self.assertRaises(BzrBadParameterNotUnicode,
797
217
osutils.safe_unicode,
801
class TestSafeUtf8(tests.TestCase):
803
def test_from_ascii_string(self):
805
self.assertEqual('foobar', osutils.safe_utf8(f))
807
def test_from_unicode_string_ascii_contents(self):
808
self.assertEqual('bargam', osutils.safe_utf8(u'bargam'))
810
def test_from_unicode_string_unicode_contents(self):
811
self.assertEqual('bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
813
def test_from_utf8_string(self):
814
self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
816
def test_bad_utf8_string(self):
817
self.assertRaises(errors.BzrBadParameterNotUnicode,
818
osutils.safe_utf8, '\xbb\xbb')
821
class TestSafeRevisionId(tests.TestCase):
823
def test_from_ascii_string(self):
824
# this shouldn't give a warning because it's getting an ascii string
825
self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
827
def test_from_unicode_string_ascii_contents(self):
828
self.assertEqual('bargam',
829
osutils.safe_revision_id(u'bargam', warn=False))
831
def test_from_unicode_deprecated(self):
832
self.assertEqual('bargam',
833
self.callDeprecated([osutils._revision_id_warning],
834
osutils.safe_revision_id, u'bargam'))
836
def test_from_unicode_string_unicode_contents(self):
837
self.assertEqual('bargam\xc2\xae',
838
osutils.safe_revision_id(u'bargam\xae', warn=False))
840
def test_from_utf8_string(self):
841
self.assertEqual('foo\xc2\xae',
842
osutils.safe_revision_id('foo\xc2\xae'))
845
"""Currently, None is a valid revision_id"""
846
self.assertEqual(None, osutils.safe_revision_id(None))
849
class TestSafeFileId(tests.TestCase):
851
def test_from_ascii_string(self):
852
self.assertEqual('foobar', osutils.safe_file_id('foobar'))
854
def test_from_unicode_string_ascii_contents(self):
855
self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
857
def test_from_unicode_deprecated(self):
858
self.assertEqual('bargam',
859
self.callDeprecated([osutils._file_id_warning],
860
osutils.safe_file_id, u'bargam'))
862
def test_from_unicode_string_unicode_contents(self):
863
self.assertEqual('bargam\xc2\xae',
864
osutils.safe_file_id(u'bargam\xae', warn=False))
866
def test_from_utf8_string(self):
867
self.assertEqual('foo\xc2\xae',
868
osutils.safe_file_id('foo\xc2\xae'))
871
"""Currently, None is a valid revision_id"""
872
self.assertEqual(None, osutils.safe_file_id(None))
875
class TestSendAll(tests.TestCase):
877
def test_send_with_disconnected_socket(self):
878
class DisconnectedSocket(object):
879
def __init__(self, err):
881
def send(self, content):
885
# All of these should be treated as ConnectionReset
887
for err_cls in (IOError, socket.error):
888
for errnum in osutils._end_of_stream_errors:
889
errs.append(err_cls(errnum))
891
sock = DisconnectedSocket(err)
892
self.assertRaises(errors.ConnectionReset,
893
osutils.send_all, sock, 'some more content')
895
def test_send_with_no_progress(self):
896
# See https://bugs.launchpad.net/bzr/+bug/1047309
897
# It seems that paramiko can get into a state where it doesn't error,
898
# but it returns 0 bytes sent for requests over and over again.
899
class NoSendingSocket(object):
902
def send(self, bytes):
904
if self.call_count > 100:
905
# Prevent the test suite from hanging
906
raise RuntimeError('too many calls')
908
sock = NoSendingSocket()
909
self.assertRaises(errors.ConnectionReset,
910
osutils.send_all, sock, 'content')
911
self.assertEqual(1, sock.call_count)
914
class TestPosixFuncs(tests.TestCase):
915
"""Test that the posix version of normpath returns an appropriate path
916
when used with 2 leading slashes."""
918
def test_normpath(self):
919
self.assertEqual('/etc/shadow', osutils._posix_normpath('/etc/shadow'))
920
self.assertEqual('/etc/shadow', osutils._posix_normpath('//etc/shadow'))
921
self.assertEqual('/etc/shadow', osutils._posix_normpath('///etc/shadow'))
924
class TestWin32Funcs(tests.TestCase):
925
"""Test that _win32 versions of os utilities return appropriate paths."""
221
class TestWin32Funcs(TestCase):
222
"""Test that the _win32 versions of os utilities return appropriate paths."""
927
224
def test_abspath(self):
928
self.requireFeature(features.win32_feature)
929
225
self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
930
226
self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
931
self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
932
self.assertEqual('//HOST/path', osutils._win32_abspath('//HOST/path'))
934
228
def test_realpath(self):
935
229
self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
936
230
self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
938
232
def test_pathjoin(self):
939
self.assertEqual('path/to/foo',
940
osutils._win32_pathjoin('path', 'to', 'foo'))
941
self.assertEqual('C:/foo',
942
osutils._win32_pathjoin('path\\to', 'C:\\foo'))
943
self.assertEqual('C:/foo',
944
osutils._win32_pathjoin('path/to', 'C:/foo'))
945
self.assertEqual('path/to/foo',
946
osutils._win32_pathjoin('path/to/', 'foo'))
948
def test_pathjoin_late_bugfix(self):
949
if sys.version_info < (2, 7, 6):
953
self.assertEqual(expected,
954
osutils._win32_pathjoin('C:/path/to/', '/foo'))
955
self.assertEqual(expected,
956
osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
233
self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
234
self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
235
self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
236
self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
237
self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
238
self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
958
240
def test_normpath(self):
959
self.assertEqual('path/to/foo',
960
osutils._win32_normpath(r'path\\from\..\to\.\foo'))
961
self.assertEqual('path/to/foo',
962
osutils._win32_normpath('path//from/../to/./foo'))
241
self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
242
self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
964
244
def test_getcwd(self):
965
245
cwd = osutils._win32_getcwd()
1182
412
result.append((dirdetail, dirblock))
1184
414
self.assertTrue(found_bzrdir)
1185
self.assertExpectedBlocks(expected_dirblocks, result)
1186
# you can search a subdir only, with a supplied prefix.
1188
for dirblock in osutils.walkdirs('./1dir', '1dir'):
1189
result.append(dirblock)
1190
self.assertExpectedBlocks(expected_dirblocks[1:], result)
1192
def test_walkdirs_os_error(self):
1193
# <https://bugs.launchpad.net/bzr/+bug/338653>
1194
# Pyrex readdir didn't raise useful messages if it had an error
1195
# reading the directory
1196
if sys.platform == 'win32':
1197
raise tests.TestNotApplicable(
1198
"readdir IOError not tested on win32")
1199
self.requireFeature(features.not_running_as_root)
1200
os.mkdir("test-unreadable")
1201
os.chmod("test-unreadable", 0000)
1202
# must chmod it back so that it can be removed
1203
self.addCleanup(os.chmod, "test-unreadable", 0700)
1204
# The error is not raised until the generator is actually evaluated.
1205
# (It would be ok if it happened earlier but at the moment it
1207
e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1208
self.assertEqual('./test-unreadable', e.filename)
1209
self.assertEqual(errno.EACCES, e.errno)
1210
# Ensure the message contains the file name
1211
self.assertContainsRe(str(e), "\./test-unreadable")
1214
def test_walkdirs_encoding_error(self):
1215
# <https://bugs.launchpad.net/bzr/+bug/488519>
1216
# walkdirs didn't raise a useful message when the filenames
1217
# are not using the filesystem's encoding
1219
# require a bytestring based filesystem
1220
self.requireFeature(features.ByteStringNamedFilesystem)
1231
self.build_tree(tree)
1233
# rename the 1file to a latin-1 filename
1234
os.rename("./1file", "\xe8file")
1235
if "\xe8file" not in os.listdir("."):
1236
self.skip("Lack filesystem that preserves arbitrary bytes")
1238
self._save_platform_info()
1239
win32utils.winver = None # Avoid the win32 detection code
1240
osutils._fs_enc = 'UTF-8'
1242
# this should raise on error
1244
for dirdetail, dirblock in osutils.walkdirs('.'):
1247
self.assertRaises(errors.BadFilenameEncoding, attempt)
1249
def test__walkdirs_utf8(self):
1258
self.build_tree(tree)
1259
expected_dirblocks = [
1261
[('0file', '0file', 'file'),
1262
('1dir', '1dir', 'directory'),
1263
('2file', '2file', 'file'),
1266
(('1dir', './1dir'),
1267
[('1dir/0file', '0file', 'file'),
1268
('1dir/1dir', '1dir', 'directory'),
1271
(('1dir/1dir', './1dir/1dir'),
1277
found_bzrdir = False
1278
for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1279
if len(dirblock) and dirblock[0][1] == '.bzr':
1280
# this tests the filtering of selected paths
1283
result.append((dirdetail, dirblock))
1285
self.assertTrue(found_bzrdir)
1286
self.assertExpectedBlocks(expected_dirblocks, result)
1288
# you can search a subdir only, with a supplied prefix.
1290
for dirblock in osutils.walkdirs('./1dir', '1dir'):
1291
result.append(dirblock)
1292
self.assertExpectedBlocks(expected_dirblocks[1:], result)
1294
def _filter_out_stat(self, result):
1295
"""Filter out the stat value from the walkdirs result"""
1296
for dirdetail, dirblock in result:
1298
for info in dirblock:
1299
# Ignore info[3] which is the stat
1300
new_dirblock.append((info[0], info[1], info[2], info[4]))
1301
dirblock[:] = new_dirblock
1303
def _save_platform_info(self):
1304
self.overrideAttr(win32utils, 'winver')
1305
self.overrideAttr(osutils, '_fs_enc')
1306
self.overrideAttr(osutils, '_selected_dir_reader')
1308
def assertDirReaderIs(self, expected):
1309
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
1310
# Force it to redetect
1311
osutils._selected_dir_reader = None
1312
# Nothing to list, but should still trigger the selection logic
1313
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1314
self.assertIsInstance(osutils._selected_dir_reader, expected)
1316
def test_force_walkdirs_utf8_fs_utf8(self):
1317
self.requireFeature(UTF8DirReaderFeature)
1318
self._save_platform_info()
1319
win32utils.winver = None # Avoid the win32 detection code
1320
osutils._fs_enc = 'utf-8'
1321
self.assertDirReaderIs(
1322
UTF8DirReaderFeature.module.UTF8DirReader)
1324
def test_force_walkdirs_utf8_fs_ascii(self):
1325
self.requireFeature(UTF8DirReaderFeature)
1326
self._save_platform_info()
1327
win32utils.winver = None # Avoid the win32 detection code
1328
osutils._fs_enc = 'ascii'
1329
self.assertDirReaderIs(
1330
UTF8DirReaderFeature.module.UTF8DirReader)
1332
def test_force_walkdirs_utf8_fs_latin1(self):
1333
self._save_platform_info()
1334
win32utils.winver = None # Avoid the win32 detection code
1335
osutils._fs_enc = 'iso-8859-1'
1336
self.assertDirReaderIs(osutils.UnicodeDirReader)
1338
def test_force_walkdirs_utf8_nt(self):
1339
# Disabled because the thunk of the whole walkdirs api is disabled.
1340
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1341
self._save_platform_info()
1342
win32utils.winver = 'Windows NT'
1343
from bzrlib._walkdirs_win32 import Win32ReadDir
1344
self.assertDirReaderIs(Win32ReadDir)
1346
def test_force_walkdirs_utf8_98(self):
1347
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1348
self._save_platform_info()
1349
win32utils.winver = 'Windows 98'
1350
self.assertDirReaderIs(osutils.UnicodeDirReader)
1352
def test_unicode_walkdirs(self):
1353
"""Walkdirs should always return unicode paths."""
1354
self.requireFeature(features.UnicodeFilenameFeature)
1355
name0 = u'0file-\xb6'
1356
name1 = u'1dir-\u062c\u0648'
1357
name2 = u'2file-\u0633'
1361
name1 + '/' + name0,
1362
name1 + '/' + name1 + '/',
1365
self.build_tree(tree)
1366
expected_dirblocks = [
1368
[(name0, name0, 'file', './' + name0),
1369
(name1, name1, 'directory', './' + name1),
1370
(name2, name2, 'file', './' + name2),
1373
((name1, './' + name1),
1374
[(name1 + '/' + name0, name0, 'file', './' + name1
1376
(name1 + '/' + name1, name1, 'directory', './' + name1
1380
((name1 + '/' + name1, './' + name1 + '/' + name1),
1385
result = list(osutils.walkdirs('.'))
1386
self._filter_out_stat(result)
1387
self.assertEqual(expected_dirblocks, result)
1388
result = list(osutils.walkdirs(u'./'+name1, name1))
1389
self._filter_out_stat(result)
1390
self.assertEqual(expected_dirblocks[1:], result)
1392
def test_unicode__walkdirs_utf8(self):
1393
"""Walkdirs_utf8 should always return utf8 paths.
1395
The abspath portion might be in unicode or utf-8
1397
self.requireFeature(features.UnicodeFilenameFeature)
1398
name0 = u'0file-\xb6'
1399
name1 = u'1dir-\u062c\u0648'
1400
name2 = u'2file-\u0633'
1404
name1 + '/' + name0,
1405
name1 + '/' + name1 + '/',
1408
self.build_tree(tree)
1409
name0 = name0.encode('utf8')
1410
name1 = name1.encode('utf8')
1411
name2 = name2.encode('utf8')
1413
expected_dirblocks = [
1415
[(name0, name0, 'file', './' + name0),
1416
(name1, name1, 'directory', './' + name1),
1417
(name2, name2, 'file', './' + name2),
1420
((name1, './' + name1),
1421
[(name1 + '/' + name0, name0, 'file', './' + name1
1423
(name1 + '/' + name1, name1, 'directory', './' + name1
1427
((name1 + '/' + name1, './' + name1 + '/' + name1),
1433
# For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1434
# all abspaths are Unicode, and encode them back into utf8.
1435
for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1436
self.assertIsInstance(dirdetail[0], str)
1437
if isinstance(dirdetail[1], unicode):
1438
dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1439
dirblock = [list(info) for info in dirblock]
1440
for info in dirblock:
1441
self.assertIsInstance(info[4], unicode)
1442
info[4] = info[4].encode('utf8')
1444
for info in dirblock:
1445
self.assertIsInstance(info[0], str)
1446
self.assertIsInstance(info[1], str)
1447
self.assertIsInstance(info[4], str)
1448
# Remove the stat information
1449
new_dirblock.append((info[0], info[1], info[2], info[4]))
1450
result.append((dirdetail, new_dirblock))
1451
self.assertEqual(expected_dirblocks, result)
1453
def test__walkdirs_utf8_with_unicode_fs(self):
1454
"""UnicodeDirReader should be a safe fallback everywhere
1456
The abspath portion should be in unicode
1458
self.requireFeature(features.UnicodeFilenameFeature)
1459
# Use the unicode reader. TODO: split into driver-and-driven unit
1461
self._save_platform_info()
1462
osutils._selected_dir_reader = osutils.UnicodeDirReader()
1463
name0u = u'0file-\xb6'
1464
name1u = u'1dir-\u062c\u0648'
1465
name2u = u'2file-\u0633'
1469
name1u + '/' + name0u,
1470
name1u + '/' + name1u + '/',
1473
self.build_tree(tree)
1474
name0 = name0u.encode('utf8')
1475
name1 = name1u.encode('utf8')
1476
name2 = name2u.encode('utf8')
1478
# All of the abspaths should be in unicode, all of the relative paths
1480
expected_dirblocks = [
1482
[(name0, name0, 'file', './' + name0u),
1483
(name1, name1, 'directory', './' + name1u),
1484
(name2, name2, 'file', './' + name2u),
1487
((name1, './' + name1u),
1488
[(name1 + '/' + name0, name0, 'file', './' + name1u
1490
(name1 + '/' + name1, name1, 'directory', './' + name1u
1494
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1499
result = list(osutils._walkdirs_utf8('.'))
1500
self._filter_out_stat(result)
1501
self.assertEqual(expected_dirblocks, result)
1503
def test__walkdirs_utf8_win32readdir(self):
1504
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1505
self.requireFeature(features.UnicodeFilenameFeature)
1506
from bzrlib._walkdirs_win32 import Win32ReadDir
1507
self._save_platform_info()
1508
osutils._selected_dir_reader = Win32ReadDir()
1509
name0u = u'0file-\xb6'
1510
name1u = u'1dir-\u062c\u0648'
1511
name2u = u'2file-\u0633'
1515
name1u + '/' + name0u,
1516
name1u + '/' + name1u + '/',
1519
self.build_tree(tree)
1520
name0 = name0u.encode('utf8')
1521
name1 = name1u.encode('utf8')
1522
name2 = name2u.encode('utf8')
1524
# All of the abspaths should be in unicode, all of the relative paths
1526
expected_dirblocks = [
1528
[(name0, name0, 'file', './' + name0u),
1529
(name1, name1, 'directory', './' + name1u),
1530
(name2, name2, 'file', './' + name2u),
1533
((name1, './' + name1u),
1534
[(name1 + '/' + name0, name0, 'file', './' + name1u
1536
(name1 + '/' + name1, name1, 'directory', './' + name1u
1540
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1545
result = list(osutils._walkdirs_utf8(u'.'))
1546
self._filter_out_stat(result)
1547
self.assertEqual(expected_dirblocks, result)
1549
def assertStatIsCorrect(self, path, win32stat):
1550
os_stat = os.stat(path)
1551
self.assertEqual(os_stat.st_size, win32stat.st_size)
1552
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1553
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1554
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1555
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1556
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1557
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1559
def test__walkdirs_utf_win32_find_file_stat_file(self):
1560
"""make sure our Stat values are valid"""
1561
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1562
self.requireFeature(features.UnicodeFilenameFeature)
1563
from bzrlib._walkdirs_win32 import Win32ReadDir
1564
name0u = u'0file-\xb6'
1565
name0 = name0u.encode('utf8')
1566
self.build_tree([name0u])
1567
# I hate to sleep() here, but I'm trying to make the ctime different
1570
f = open(name0u, 'ab')
1572
f.write('just a small update')
1576
result = Win32ReadDir().read_dir('', u'.')
1578
self.assertEqual((name0, name0, 'file'), entry[:3])
1579
self.assertEqual(u'./' + name0u, entry[4])
1580
self.assertStatIsCorrect(entry[4], entry[3])
1581
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1583
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1584
"""make sure our Stat values are valid"""
1585
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1586
self.requireFeature(features.UnicodeFilenameFeature)
1587
from bzrlib._walkdirs_win32 import Win32ReadDir
1588
name0u = u'0dir-\u062c\u0648'
1589
name0 = name0u.encode('utf8')
1590
self.build_tree([name0u + '/'])
1592
result = Win32ReadDir().read_dir('', u'.')
1594
self.assertEqual((name0, name0, 'directory'), entry[:3])
1595
self.assertEqual(u'./' + name0u, entry[4])
1596
self.assertStatIsCorrect(entry[4], entry[3])
415
self.assertEqual(expected_dirblocks,
416
[(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
417
# you can search a subdir only, with a supplied prefix.
419
for dirblock in osutils.walkdirs('./1dir', '1dir'):
420
result.append(dirblock)
421
self.assertEqual(expected_dirblocks[1:],
422
[(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1598
424
def assertPathCompare(self, path_less, path_greater):
1599
425
"""check that path_less and path_greater compare correctly."""
1772
647
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1773
648
self.assertEqual('foo', old)
1774
649
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1775
self.assertFalse('BZR_TEST_ENV_VAR' in os.environ)
1778
class TestSizeShaFile(tests.TestCaseInTempDir):
1780
def test_sha_empty(self):
1781
self.build_tree_contents([('foo', '')])
1782
expected_sha = osutils.sha_string('')
1784
self.addCleanup(f.close)
1785
size, sha = osutils.size_sha_file(f)
1786
self.assertEqual(0, size)
1787
self.assertEqual(expected_sha, sha)
1789
def test_sha_mixed_endings(self):
1790
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1791
self.build_tree_contents([('foo', text)])
1792
expected_sha = osutils.sha_string(text)
1793
f = open('foo', 'rb')
1794
self.addCleanup(f.close)
1795
size, sha = osutils.size_sha_file(f)
1796
self.assertEqual(38, size)
1797
self.assertEqual(expected_sha, sha)
1800
class TestShaFileByName(tests.TestCaseInTempDir):
1802
def test_sha_empty(self):
1803
self.build_tree_contents([('foo', '')])
1804
expected_sha = osutils.sha_string('')
1805
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1807
def test_sha_mixed_endings(self):
1808
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1809
self.build_tree_contents([('foo', text)])
1810
expected_sha = osutils.sha_string(text)
1811
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1814
class TestResourceLoading(tests.TestCaseInTempDir):
1816
def test_resource_string(self):
1817
# test resource in bzrlib
1818
text = osutils.resource_string('bzrlib', 'debug.py')
1819
self.assertContainsRe(text, "debug_flags = set()")
1820
# test resource under bzrlib
1821
text = osutils.resource_string('bzrlib.ui', 'text.py')
1822
self.assertContainsRe(text, "class TextUIFactory")
1823
# test unsupported package
1824
self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1826
# test unknown resource
1827
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1830
class TestReCompile(tests.TestCase):
1832
def _deprecated_re_compile_checked(self, *args, **kwargs):
1833
return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
1834
osutils.re_compile_checked, *args, **kwargs)
1836
def test_re_compile_checked(self):
1837
r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
1838
self.assertTrue(r.match('aaaa'))
1839
self.assertTrue(r.match('aAaA'))
1841
def test_re_compile_checked_error(self):
1842
# like https://bugs.launchpad.net/bzr/+bug/251352
1844
# Due to possible test isolation error, re.compile is not lazy at
1845
# this point. We re-install lazy compile.
1846
lazy_regex.install_lazy_compile()
1847
err = self.assertRaises(
1848
errors.BzrCommandError,
1849
self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
1851
'Invalid regular expression in test case: '
1852
'"*" nothing to repeat',
1856
class TestDirReader(tests.TestCaseInTempDir):
1858
scenarios = dir_reader_scenarios()
1861
_dir_reader_class = None
1862
_native_to_unicode = None
1865
super(TestDirReader, self).setUp()
1866
self.overrideAttr(osutils,
1867
'_selected_dir_reader', self._dir_reader_class())
1869
def _get_ascii_tree(self):
1877
expected_dirblocks = [
1879
[('0file', '0file', 'file'),
1880
('1dir', '1dir', 'directory'),
1881
('2file', '2file', 'file'),
1884
(('1dir', './1dir'),
1885
[('1dir/0file', '0file', 'file'),
1886
('1dir/1dir', '1dir', 'directory'),
1889
(('1dir/1dir', './1dir/1dir'),
1894
return tree, expected_dirblocks
1896
def test_walk_cur_dir(self):
1897
tree, expected_dirblocks = self._get_ascii_tree()
1898
self.build_tree(tree)
1899
result = list(osutils._walkdirs_utf8('.'))
1900
# Filter out stat and abspath
1901
self.assertEqual(expected_dirblocks,
1902
[(dirinfo, [line[0:3] for line in block])
1903
for dirinfo, block in result])
1905
def test_walk_sub_dir(self):
1906
tree, expected_dirblocks = self._get_ascii_tree()
1907
self.build_tree(tree)
1908
# you can search a subdir only, with a supplied prefix.
1909
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1910
# Filter out stat and abspath
1911
self.assertEqual(expected_dirblocks[1:],
1912
[(dirinfo, [line[0:3] for line in block])
1913
for dirinfo, block in result])
1915
def _get_unicode_tree(self):
1916
name0u = u'0file-\xb6'
1917
name1u = u'1dir-\u062c\u0648'
1918
name2u = u'2file-\u0633'
1922
name1u + '/' + name0u,
1923
name1u + '/' + name1u + '/',
1926
name0 = name0u.encode('UTF-8')
1927
name1 = name1u.encode('UTF-8')
1928
name2 = name2u.encode('UTF-8')
1929
expected_dirblocks = [
1931
[(name0, name0, 'file', './' + name0u),
1932
(name1, name1, 'directory', './' + name1u),
1933
(name2, name2, 'file', './' + name2u),
1936
((name1, './' + name1u),
1937
[(name1 + '/' + name0, name0, 'file', './' + name1u
1939
(name1 + '/' + name1, name1, 'directory', './' + name1u
1943
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1948
return tree, expected_dirblocks
1950
def _filter_out(self, raw_dirblocks):
1951
"""Filter out a walkdirs_utf8 result.
1953
stat field is removed, all native paths are converted to unicode
1955
filtered_dirblocks = []
1956
for dirinfo, block in raw_dirblocks:
1957
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1960
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1961
filtered_dirblocks.append((dirinfo, details))
1962
return filtered_dirblocks
1964
def test_walk_unicode_tree(self):
1965
self.requireFeature(features.UnicodeFilenameFeature)
1966
tree, expected_dirblocks = self._get_unicode_tree()
1967
self.build_tree(tree)
1968
result = list(osutils._walkdirs_utf8('.'))
1969
self.assertEqual(expected_dirblocks, self._filter_out(result))
1971
def test_symlink(self):
1972
self.requireFeature(features.SymlinkFeature)
1973
self.requireFeature(features.UnicodeFilenameFeature)
1974
target = u'target\N{Euro Sign}'
1975
link_name = u'l\N{Euro Sign}nk'
1976
os.symlink(target, link_name)
1977
target_utf8 = target.encode('UTF-8')
1978
link_name_utf8 = link_name.encode('UTF-8')
1979
expected_dirblocks = [
1981
[(link_name_utf8, link_name_utf8,
1982
'symlink', './' + link_name),],
1984
result = list(osutils._walkdirs_utf8('.'))
1985
self.assertEqual(expected_dirblocks, self._filter_out(result))
1988
class TestReadLink(tests.TestCaseInTempDir):
1989
"""Exposes os.readlink() problems and the osutils solution.
1991
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1992
unicode string will be returned if a unicode string is passed.
1994
But prior python versions failed to properly encode the passed unicode
1997
_test_needs_features = [features.SymlinkFeature, features.UnicodeFilenameFeature]
2000
super(tests.TestCaseInTempDir, self).setUp()
2001
self.link = u'l\N{Euro Sign}ink'
2002
self.target = u'targe\N{Euro Sign}t'
2003
os.symlink(self.target, self.link)
2005
def test_os_readlink_link_encoding(self):
2006
self.assertEqual(self.target, os.readlink(self.link))
2008
def test_os_readlink_link_decoding(self):
2009
self.assertEqual(self.target.encode(osutils._fs_enc),
2010
os.readlink(self.link.encode(osutils._fs_enc)))
2013
class TestConcurrency(tests.TestCase):
2016
super(TestConcurrency, self).setUp()
2017
self.overrideAttr(osutils, '_cached_local_concurrency')
2019
def test_local_concurrency(self):
2020
concurrency = osutils.local_concurrency()
2021
self.assertIsInstance(concurrency, int)
2023
def test_local_concurrency_environment_variable(self):
2024
self.overrideEnv('BZR_CONCURRENCY', '2')
2025
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
2026
self.overrideEnv('BZR_CONCURRENCY', '3')
2027
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
2028
self.overrideEnv('BZR_CONCURRENCY', 'foo')
2029
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
2031
def test_option_concurrency(self):
2032
self.overrideEnv('BZR_CONCURRENCY', '1')
2033
self.run_bzr('rocks --concurrency 42')
2034
# Command line overrides environment variable
2035
self.assertEqual('42', os.environ['BZR_CONCURRENCY'])
2036
self.assertEqual(42, osutils.local_concurrency(use_cache=False))
2039
class TestFailedToLoadExtension(tests.TestCase):
2041
def _try_loading(self):
2043
import bzrlib._fictional_extension_py
2044
except ImportError, e:
2045
osutils.failed_to_load_extension(e)
2049
super(TestFailedToLoadExtension, self).setUp()
2050
self.overrideAttr(osutils, '_extension_load_failures', [])
2052
def test_failure_to_load(self):
2054
self.assertLength(1, osutils._extension_load_failures)
2055
self.assertEqual(osutils._extension_load_failures[0],
2056
"No module named _fictional_extension_py")
2058
def test_report_extension_load_failures_no_warning(self):
2059
self.assertTrue(self._try_loading())
2060
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
2061
# it used to give a Python warning; it no longer does
2062
self.assertLength(0, warnings)
2064
def test_report_extension_load_failures_message(self):
2066
trace.push_log_file(log)
2067
self.assertTrue(self._try_loading())
2068
osutils.report_extension_load_failures()
2069
self.assertContainsRe(
2071
r"bzr: warning: some compiled extensions could not be loaded; "
2072
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
2076
class TestTerminalWidth(tests.TestCase):
2079
super(TestTerminalWidth, self).setUp()
2080
self._orig_terminal_size_state = osutils._terminal_size_state
2081
self._orig_first_terminal_size = osutils._first_terminal_size
2082
self.addCleanup(self.restore_osutils_globals)
2083
osutils._terminal_size_state = 'no_data'
2084
osutils._first_terminal_size = None
2086
def restore_osutils_globals(self):
2087
osutils._terminal_size_state = self._orig_terminal_size_state
2088
osutils._first_terminal_size = self._orig_first_terminal_size
2090
def replace_stdout(self, new):
2091
self.overrideAttr(sys, 'stdout', new)
2093
def replace__terminal_size(self, new):
2094
self.overrideAttr(osutils, '_terminal_size', new)
2096
def set_fake_tty(self):
2098
class I_am_a_tty(object):
2102
self.replace_stdout(I_am_a_tty())
2104
def test_default_values(self):
2105
self.assertEqual(80, osutils.default_terminal_width)
2107
def test_defaults_to_BZR_COLUMNS(self):
2108
# BZR_COLUMNS is set by the test framework
2109
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
2110
self.overrideEnv('BZR_COLUMNS', '12')
2111
self.assertEqual(12, osutils.terminal_width())
2113
def test_BZR_COLUMNS_0_no_limit(self):
2114
self.overrideEnv('BZR_COLUMNS', '0')
2115
self.assertEqual(None, osutils.terminal_width())
2117
def test_falls_back_to_COLUMNS(self):
2118
self.overrideEnv('BZR_COLUMNS', None)
2119
self.assertNotEqual('42', os.environ['COLUMNS'])
2121
self.overrideEnv('COLUMNS', '42')
2122
self.assertEqual(42, osutils.terminal_width())
2124
def test_tty_default_without_columns(self):
2125
self.overrideEnv('BZR_COLUMNS', None)
2126
self.overrideEnv('COLUMNS', None)
2128
def terminal_size(w, h):
2132
# We need to override the osutils definition as it depends on the
2133
# running environment that we can't control (PQM running without a
2134
# controlling terminal is one example).
2135
self.replace__terminal_size(terminal_size)
2136
self.assertEqual(42, osutils.terminal_width())
2138
def test_non_tty_default_without_columns(self):
2139
self.overrideEnv('BZR_COLUMNS', None)
2140
self.overrideEnv('COLUMNS', None)
2141
self.replace_stdout(None)
2142
self.assertEqual(None, osutils.terminal_width())
2144
def test_no_TIOCGWINSZ(self):
2145
self.requireFeature(term_ios_feature)
2146
termios = term_ios_feature.module
2147
# bug 63539 is about a termios without TIOCGWINSZ attribute
2149
orig = termios.TIOCGWINSZ
2150
except AttributeError:
2151
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
2154
self.overrideAttr(termios, 'TIOCGWINSZ')
2155
del termios.TIOCGWINSZ
2156
self.overrideEnv('BZR_COLUMNS', None)
2157
self.overrideEnv('COLUMNS', None)
2158
# Whatever the result is, if we don't raise an exception, it's ok.
2159
osutils.terminal_width()
2162
class TestCreationOps(tests.TestCaseInTempDir):
2163
_test_needs_features = [features.chown_feature]
2166
super(TestCreationOps, self).setUp()
2167
self.overrideAttr(os, 'chown', self._dummy_chown)
2169
# params set by call to _dummy_chown
2170
self.path = self.uid = self.gid = None
2172
def _dummy_chown(self, path, uid, gid):
2173
self.path, self.uid, self.gid = path, uid, gid
2175
def test_copy_ownership_from_path(self):
2176
"""copy_ownership_from_path test with specified src."""
2178
f = open('test_file', 'wt')
2179
osutils.copy_ownership_from_path('test_file', ownsrc)
2182
self.assertEqual(self.path, 'test_file')
2183
self.assertEqual(self.uid, s.st_uid)
2184
self.assertEqual(self.gid, s.st_gid)
2186
def test_copy_ownership_nonesrc(self):
2187
"""copy_ownership_from_path test with src=None."""
2188
f = open('test_file', 'wt')
2189
# should use parent dir for permissions
2190
osutils.copy_ownership_from_path('test_file')
2193
self.assertEqual(self.path, 'test_file')
2194
self.assertEqual(self.uid, s.st_uid)
2195
self.assertEqual(self.gid, s.st_gid)
2198
class TestPathFromEnviron(tests.TestCase):
2200
def test_is_unicode(self):
2201
self.overrideEnv('BZR_TEST_PATH', './anywhere at all/')
2202
path = osutils.path_from_environ('BZR_TEST_PATH')
2203
self.assertIsInstance(path, unicode)
2204
self.assertEqual(u'./anywhere at all/', path)
2206
def test_posix_path_env_ascii(self):
2207
self.overrideEnv('BZR_TEST_PATH', '/tmp')
2208
home = osutils._posix_path_from_environ('BZR_TEST_PATH')
2209
self.assertIsInstance(home, unicode)
2210
self.assertEqual(u'/tmp', home)
2212
def test_posix_path_env_unicode(self):
2213
self.requireFeature(features.ByteStringNamedFilesystem)
2214
self.overrideEnv('BZR_TEST_PATH', '/home/\xa7test')
2215
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2216
self.assertEqual(u'/home/\xa7test',
2217
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2218
osutils._fs_enc = "iso8859-5"
2219
self.assertEqual(u'/home/\u0407test',
2220
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2221
osutils._fs_enc = "utf-8"
2222
self.assertRaises(errors.BadFilenameEncoding,
2223
osutils._posix_path_from_environ, 'BZR_TEST_PATH')
2226
class TestGetHomeDir(tests.TestCase):
2228
def test_is_unicode(self):
2229
home = osutils._get_home_dir()
2230
self.assertIsInstance(home, unicode)
2232
def test_posix_homeless(self):
2233
self.overrideEnv('HOME', None)
2234
home = osutils._get_home_dir()
2235
self.assertIsInstance(home, unicode)
2237
def test_posix_home_ascii(self):
2238
self.overrideEnv('HOME', '/home/test')
2239
home = osutils._posix_get_home_dir()
2240
self.assertIsInstance(home, unicode)
2241
self.assertEqual(u'/home/test', home)
2243
def test_posix_home_unicode(self):
2244
self.requireFeature(features.ByteStringNamedFilesystem)
2245
self.overrideEnv('HOME', '/home/\xa7test')
2246
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2247
self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
2248
osutils._fs_enc = "iso8859-5"
2249
self.assertEqual(u'/home/\u0407test', osutils._posix_get_home_dir())
2250
osutils._fs_enc = "utf-8"
2251
self.assertRaises(errors.BadFilenameEncoding,
2252
osutils._posix_get_home_dir)
2255
class TestGetuserUnicode(tests.TestCase):
2257
def test_is_unicode(self):
2258
user = osutils.getuser_unicode()
2259
self.assertIsInstance(user, unicode)
2261
def envvar_to_override(self):
2262
if sys.platform == "win32":
2263
# Disable use of platform calls on windows so envvar is used
2264
self.overrideAttr(win32utils, 'has_ctypes', False)
2265
return 'USERNAME' # only variable used on windows
2266
return 'LOGNAME' # first variable checked by getpass.getuser()
2268
def test_ascii_user(self):
2269
self.overrideEnv(self.envvar_to_override(), 'jrandom')
2270
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2272
def test_unicode_user(self):
2273
ue = osutils.get_user_encoding()
2274
uni_val, env_val = tests.probe_unicode_in_user_encoding()
2276
raise tests.TestSkipped(
2277
'Cannot find a unicode character that works in encoding %s'
2278
% (osutils.get_user_encoding(),))
2279
uni_username = u'jrandom' + uni_val
2280
encoded_username = uni_username.encode(ue)
2281
self.overrideEnv(self.envvar_to_override(), encoded_username)
2282
self.assertEqual(uni_username, osutils.getuser_unicode())
2285
class TestBackupNames(tests.TestCase):
2288
super(TestBackupNames, self).setUp()
2291
def backup_exists(self, name):
2292
return name in self.backups
2294
def available_backup_name(self, name):
2295
backup_name = osutils.available_backup_name(name, self.backup_exists)
2296
self.backups.append(backup_name)
2299
def assertBackupName(self, expected, name):
2300
self.assertEqual(expected, self.available_backup_name(name))
2302
def test_empty(self):
2303
self.assertBackupName('file.~1~', 'file')
2305
def test_existing(self):
2306
self.available_backup_name('file')
2307
self.available_backup_name('file')
2308
self.assertBackupName('file.~3~', 'file')
2309
# Empty slots are found, this is not a strict requirement and may be
2310
# revisited if we test against all implementations.
2311
self.backups.remove('file.~2~')
2312
self.assertBackupName('file.~2~', 'file')
2315
class TestFindExecutableInPath(tests.TestCase):
2317
def test_windows(self):
2318
if sys.platform != 'win32':
2319
raise tests.TestSkipped('test requires win32')
2320
self.assertTrue(osutils.find_executable_on_path('explorer') is not None)
2322
osutils.find_executable_on_path('explorer.exe') is not None)
2324
osutils.find_executable_on_path('EXPLORER.EXE') is not None)
2326
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2327
self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
2329
def test_windows_app_path(self):
2330
if sys.platform != 'win32':
2331
raise tests.TestSkipped('test requires win32')
2332
# Override PATH env var so that exe can only be found on App Path
2333
self.overrideEnv('PATH', '')
2334
# Internt Explorer is always registered in the App Path
2335
self.assertTrue(osutils.find_executable_on_path('iexplore') is not None)
2337
def test_other(self):
2338
if sys.platform == 'win32':
2339
raise tests.TestSkipped('test requires non-win32')
2340
self.assertTrue(osutils.find_executable_on_path('sh') is not None)
2342
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2345
class TestEnvironmentErrors(tests.TestCase):
2346
"""Test handling of environmental errors"""
2348
def test_is_oserror(self):
2349
self.assertTrue(osutils.is_environment_error(
2350
OSError(errno.EINVAL, "Invalid parameter")))
2352
def test_is_ioerror(self):
2353
self.assertTrue(osutils.is_environment_error(
2354
IOError(errno.EINVAL, "Invalid parameter")))
2356
def test_is_socket_error(self):
2357
self.assertTrue(osutils.is_environment_error(
2358
socket.error(errno.EINVAL, "Invalid parameter")))
2360
def test_is_select_error(self):
2361
self.assertTrue(osutils.is_environment_error(
2362
select.error(errno.EINVAL, "Invalid parameter")))
2364
def test_is_pywintypes_error(self):
2365
self.requireFeature(features.pywintypes)
2367
self.assertTrue(osutils.is_environment_error(
2368
pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))
650
self.failIf('BZR_TEST_ENV_VAR' in os.environ)
added
removed
bzrlib/tests/test_osutils.py
