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
26
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
27
import bzrlib.osutils as osutils
35
28
from bzrlib.tests import (
42
class _UTF8DirReaderFeature(tests.Feature):
46
from bzrlib import _readdir_pyx
47
self.reader = _readdir_pyx.UTF8DirReader
52
def feature_name(self):
53
return 'bzrlib._readdir_pyx'
55
UTF8DirReaderFeature = _UTF8DirReaderFeature()
57
term_ios_feature = tests.ModuleAvailableFeature('termios')
60
def _already_unicode(s):
64
def _utf8_to_unicode(s):
65
return s.decode('UTF-8')
68
def dir_reader_scenarios():
69
# For each dir reader we define:
71
# - native_to_unicode: a function converting the native_abspath as returned
72
# by DirReader.read_dir to its unicode representation
74
# UnicodeDirReader is the fallback, it should be tested on all platforms.
75
scenarios = [('unicode',
76
dict(_dir_reader_class=osutils.UnicodeDirReader,
77
_native_to_unicode=_already_unicode))]
78
# Some DirReaders are platform specific and even there they may not be
80
if UTF8DirReaderFeature.available():
81
from bzrlib import _readdir_pyx
82
scenarios.append(('utf8',
83
dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
84
_native_to_unicode=_utf8_to_unicode)))
86
if test__walkdirs_win32.win32_readdir_feature.available():
88
from bzrlib import _walkdirs_win32
91
dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
92
_native_to_unicode=_already_unicode)))
98
def load_tests(basic_tests, module, loader):
99
suite = loader.suiteClass()
100
dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
101
basic_tests, tests.condition_isinstance(TestDirReader))
102
tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
103
suite.addTest(remaining_tests)
107
class TestContainsWhitespace(tests.TestCase):
109
def test_contains_whitespace(self):
110
self.failUnless(osutils.contains_whitespace(u' '))
111
self.failUnless(osutils.contains_whitespace(u'hello there'))
112
self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
113
self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
114
self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
115
self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
117
# \xa0 is "Non-breaking-space" which on some python locales thinks it
118
# is whitespace, but we do not.
119
self.failIf(osutils.contains_whitespace(u''))
120
self.failIf(osutils.contains_whitespace(u'hellothere'))
121
self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
124
class TestRename(tests.TestCaseInTempDir):
126
def create_file(self, filename, content):
127
f = open(filename, 'wb')
133
def _fancy_rename(self, a, b):
134
osutils.fancy_rename(a, b, rename_func=os.rename,
135
unlink_func=os.unlink)
36
class TestOSUtils(TestCaseInTempDir):
137
38
def test_fancy_rename(self):
138
39
# This should work everywhere
139
self.create_file('a', 'something in a\n')
140
self._fancy_rename('a', 'b')
41
osutils.fancy_rename(a, b,
42
rename_func=os.rename,
43
unlink_func=os.unlink)
45
open('a', 'wb').write('something in a\n')
141
47
self.failIfExists('a')
142
48
self.failUnlessExists('b')
143
49
self.check_file_contents('b', 'something in a\n')
145
self.create_file('a', 'new something in a\n')
146
self._fancy_rename('b', 'a')
51
open('a', 'wb').write('new something in a\n')
148
54
self.check_file_contents('a', 'something in a\n')
150
def test_fancy_rename_fails_source_missing(self):
151
# An exception should be raised, and the target should be left in place
152
self.create_file('target', 'data in target\n')
153
self.assertRaises((IOError, OSError), self._fancy_rename,
154
'missingsource', 'target')
155
self.failUnlessExists('target')
156
self.check_file_contents('target', 'data in target\n')
158
def test_fancy_rename_fails_if_source_and_target_missing(self):
159
self.assertRaises((IOError, OSError), self._fancy_rename,
160
'missingsource', 'missingtarget')
162
56
def test_rename(self):
163
57
# Rename should be semi-atomic on all platforms
164
self.create_file('a', 'something in a\n')
58
open('a', 'wb').write('something in a\n')
165
59
osutils.rename('a', 'b')
166
60
self.failIfExists('a')
167
61
self.failUnlessExists('b')
168
62
self.check_file_contents('b', 'something in a\n')
170
self.create_file('a', 'new something in a\n')
64
open('a', 'wb').write('new something in a\n')
171
65
osutils.rename('b', 'a')
173
67
self.check_file_contents('a', 'something in a\n')
175
69
# TODO: test fancy_rename using a MemoryTransport
177
def test_rename_change_case(self):
178
# on Windows we should be able to change filename case by rename
179
self.build_tree(['a', 'b/'])
180
osutils.rename('a', 'A')
181
osutils.rename('b', 'B')
182
# we can't use failUnlessExists on case-insensitive filesystem
183
# so try to check shape of the tree
184
shape = sorted(os.listdir('.'))
185
self.assertEquals(['A', 'B'], shape)
187
def test_rename_error(self):
188
# We wrap os.rename to make it give an error including the filenames
189
# https://bugs.launchpad.net/bzr/+bug/491763
190
err = self.assertRaises(OSError, osutils.rename,
191
'nonexistent', 'target')
192
self.assertContainsString(str(err), 'nonexistent')
195
class TestRandChars(tests.TestCase):
197
71
def test_01_rand_chars_empty(self):
198
72
result = osutils.rand_chars(0)
199
73
self.assertEqual(result, '')
333
140
orig_umask = osutils.get_umask()
334
self.addCleanup(os.umask, orig_umask)
336
self.assertEqual(0222, osutils.get_umask())
338
self.assertEqual(0022, osutils.get_umask())
340
self.assertEqual(0002, osutils.get_umask())
342
self.assertEqual(0027, osutils.get_umask())
345
class TestDateTime(tests.TestCase):
347
def assertFormatedDelta(self, expected, seconds):
348
"""Assert osutils.format_delta formats as expected"""
349
actual = osutils.format_delta(seconds)
350
self.assertEqual(expected, actual)
352
def test_format_delta(self):
353
self.assertFormatedDelta('0 seconds ago', 0)
354
self.assertFormatedDelta('1 second ago', 1)
355
self.assertFormatedDelta('10 seconds ago', 10)
356
self.assertFormatedDelta('59 seconds ago', 59)
357
self.assertFormatedDelta('89 seconds ago', 89)
358
self.assertFormatedDelta('1 minute, 30 seconds ago', 90)
359
self.assertFormatedDelta('3 minutes, 0 seconds ago', 180)
360
self.assertFormatedDelta('3 minutes, 1 second ago', 181)
361
self.assertFormatedDelta('10 minutes, 15 seconds ago', 615)
362
self.assertFormatedDelta('30 minutes, 59 seconds ago', 1859)
363
self.assertFormatedDelta('31 minutes, 0 seconds ago', 1860)
364
self.assertFormatedDelta('60 minutes, 0 seconds ago', 3600)
365
self.assertFormatedDelta('89 minutes, 59 seconds ago', 5399)
366
self.assertFormatedDelta('1 hour, 30 minutes ago', 5400)
367
self.assertFormatedDelta('2 hours, 30 minutes ago', 9017)
368
self.assertFormatedDelta('10 hours, 0 minutes ago', 36000)
369
self.assertFormatedDelta('24 hours, 0 minutes ago', 86400)
370
self.assertFormatedDelta('35 hours, 59 minutes ago', 129599)
371
self.assertFormatedDelta('36 hours, 0 minutes ago', 129600)
372
self.assertFormatedDelta('36 hours, 0 minutes ago', 129601)
373
self.assertFormatedDelta('36 hours, 1 minute ago', 129660)
374
self.assertFormatedDelta('36 hours, 1 minute ago', 129661)
375
self.assertFormatedDelta('84 hours, 10 minutes ago', 303002)
377
# We handle when time steps the wrong direction because computers
378
# don't have synchronized clocks.
379
self.assertFormatedDelta('84 hours, 10 minutes in the future', -303002)
380
self.assertFormatedDelta('1 second in the future', -1)
381
self.assertFormatedDelta('2 seconds in the future', -2)
383
def test_format_date(self):
384
self.assertRaises(errors.UnsupportedTimezoneFormat,
385
osutils.format_date, 0, timezone='foo')
386
self.assertIsInstance(osutils.format_date(0), str)
387
self.assertIsInstance(osutils.format_local_date(0), unicode)
388
# Testing for the actual value of the local weekday without
389
# duplicating the code from format_date is difficult.
390
# Instead blackbox.test_locale should check for localized
391
# dates once they do occur in output strings.
393
def test_format_date_with_offset_in_original_timezone(self):
394
self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
395
osutils.format_date_with_offset_in_original_timezone(0))
396
self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
397
osutils.format_date_with_offset_in_original_timezone(100000))
398
self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
399
osutils.format_date_with_offset_in_original_timezone(100000, 7200))
401
def test_local_time_offset(self):
402
"""Test that local_time_offset() returns a sane value."""
403
offset = osutils.local_time_offset()
404
self.assertTrue(isinstance(offset, int))
405
# Test that the offset is no more than a eighteen hours in
407
# Time zone handling is system specific, so it is difficult to
408
# do more specific tests, but a value outside of this range is
410
eighteen_hours = 18 * 3600
411
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
413
def test_local_time_offset_with_timestamp(self):
414
"""Test that local_time_offset() works with a timestamp."""
415
offset = osutils.local_time_offset(1000000000.1234567)
416
self.assertTrue(isinstance(offset, int))
417
eighteen_hours = 18 * 3600
418
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
421
class TestLinks(tests.TestCaseInTempDir):
423
def test_dereference_path(self):
424
self.requireFeature(tests.SymlinkFeature)
425
cwd = osutils.realpath('.')
427
bar_path = osutils.pathjoin(cwd, 'bar')
428
# Using './' to avoid bug #1213894 (first path component not
429
# dereferenced) in Python 2.4.1 and earlier
430
self.assertEqual(bar_path, osutils.realpath('./bar'))
431
os.symlink('bar', 'foo')
432
self.assertEqual(bar_path, osutils.realpath('./foo'))
434
# Does not dereference terminal symlinks
435
foo_path = osutils.pathjoin(cwd, 'foo')
436
self.assertEqual(foo_path, osutils.dereference_path('./foo'))
438
# Dereferences parent symlinks
440
baz_path = osutils.pathjoin(bar_path, 'baz')
441
self.assertEqual(baz_path, osutils.dereference_path('./foo/baz'))
443
# Dereferences parent symlinks that are the first path element
444
self.assertEqual(baz_path, osutils.dereference_path('foo/baz'))
446
# Dereferences parent symlinks in absolute paths
447
foo_baz_path = osutils.pathjoin(foo_path, 'baz')
448
self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
450
def test_changing_access(self):
451
f = file('file', 'w')
455
# Make a file readonly
456
osutils.make_readonly('file')
457
mode = os.lstat('file').st_mode
458
self.assertEqual(mode, mode & 0777555)
460
# Make a file writable
461
osutils.make_writable('file')
462
mode = os.lstat('file').st_mode
463
self.assertEqual(mode, mode | 0200)
465
if osutils.has_symlinks():
466
# should not error when handed a symlink
467
os.symlink('nonexistent', 'dangling')
468
osutils.make_readonly('dangling')
469
osutils.make_writable('dangling')
471
def test_host_os_dereferences_symlinks(self):
472
osutils.host_os_dereferences_symlinks()
475
class TestCanonicalRelPath(tests.TestCaseInTempDir):
477
_test_needs_features = [tests.CaseInsCasePresFilenameFeature]
479
def test_canonical_relpath_simple(self):
480
f = file('MixedCaseName', 'w')
482
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
483
self.failUnlessEqual('work/MixedCaseName', actual)
485
def test_canonical_relpath_missing_tail(self):
486
os.mkdir('MixedCaseParent')
487
actual = osutils.canonical_relpath(self.test_base_dir,
488
'mixedcaseparent/nochild')
489
self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
492
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
494
def assertRelpath(self, expected, base, path):
495
actual = osutils._cicp_canonical_relpath(base, path)
496
self.assertEqual(expected, actual)
498
def test_simple(self):
499
self.build_tree(['MixedCaseName'])
500
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
501
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
503
def test_subdir_missing_tail(self):
504
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
505
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
506
self.assertRelpath('MixedCaseParent/a_child', base,
507
'MixedCaseParent/a_child')
508
self.assertRelpath('MixedCaseParent/a_child', base,
509
'MixedCaseParent/A_Child')
510
self.assertRelpath('MixedCaseParent/not_child', base,
511
'MixedCaseParent/not_child')
513
def test_at_root_slash(self):
514
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
516
if osutils.MIN_ABS_PATHLENGTH > 1:
517
raise tests.TestSkipped('relpath requires %d chars'
518
% osutils.MIN_ABS_PATHLENGTH)
519
self.assertRelpath('foo', '/', '/foo')
521
def test_at_root_drive(self):
522
if sys.platform != 'win32':
523
raise tests.TestNotApplicable('we can only test drive-letter relative'
524
' paths on Windows where we have drive'
527
# The specific issue is that when at the root of a drive, 'abspath'
528
# returns "C:/" or just "/". However, the code assumes that abspath
529
# always returns something like "C:/foo" or "/foo" (no trailing slash).
530
self.assertRelpath('foo', 'C:/', 'C:/foo')
531
self.assertRelpath('foo', 'X:/', 'X:/foo')
532
self.assertRelpath('foo', 'X:/', 'X://foo')
535
class TestPumpFile(tests.TestCase):
536
"""Test pumpfile method."""
539
tests.TestCase.setUp(self)
540
# create a test datablock
541
self.block_size = 512
542
pattern = '0123456789ABCDEF'
543
self.test_data = pattern * (3 * self.block_size / len(pattern))
544
self.test_data_len = len(self.test_data)
546
def test_bracket_block_size(self):
547
"""Read data in blocks with the requested read size bracketing the
549
# make sure test data is larger than max read size
550
self.assertTrue(self.test_data_len > self.block_size)
552
from_file = file_utils.FakeReadFile(self.test_data)
555
# read (max / 2) bytes and verify read size wasn't affected
556
num_bytes_to_read = self.block_size / 2
557
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
558
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
559
self.assertEqual(from_file.get_read_count(), 1)
561
# read (max) bytes and verify read size wasn't affected
562
num_bytes_to_read = self.block_size
563
from_file.reset_read_count()
564
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
565
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
566
self.assertEqual(from_file.get_read_count(), 1)
568
# read (max + 1) bytes and verify read size was limited
569
num_bytes_to_read = self.block_size + 1
570
from_file.reset_read_count()
571
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
572
self.assertEqual(from_file.get_max_read_size(), self.block_size)
573
self.assertEqual(from_file.get_read_count(), 2)
575
# finish reading the rest of the data
576
num_bytes_to_read = self.test_data_len - to_file.tell()
577
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
579
# report error if the data wasn't equal (we only report the size due
580
# to the length of the data)
581
response_data = to_file.getvalue()
582
if response_data != self.test_data:
583
message = "Data not equal. Expected %d bytes, received %d."
584
self.fail(message % (len(response_data), self.test_data_len))
586
def test_specified_size(self):
587
"""Request a transfer larger than the maximum block size and verify
588
that the maximum read doesn't exceed the block_size."""
589
# make sure test data is larger than max read size
590
self.assertTrue(self.test_data_len > self.block_size)
592
# retrieve data in blocks
593
from_file = file_utils.FakeReadFile(self.test_data)
595
osutils.pumpfile(from_file, to_file, self.test_data_len,
598
# verify read size was equal to the maximum read size
599
self.assertTrue(from_file.get_max_read_size() > 0)
600
self.assertEqual(from_file.get_max_read_size(), self.block_size)
601
self.assertEqual(from_file.get_read_count(), 3)
603
# report error if the data wasn't equal (we only report the size due
604
# to the length of the data)
605
response_data = to_file.getvalue()
606
if response_data != self.test_data:
607
message = "Data not equal. Expected %d bytes, received %d."
608
self.fail(message % (len(response_data), self.test_data_len))
610
def test_to_eof(self):
611
"""Read to end-of-file and verify that the reads are not larger than
612
the maximum read size."""
613
# make sure test data is larger than max read size
614
self.assertTrue(self.test_data_len > self.block_size)
616
# retrieve data to EOF
617
from_file = file_utils.FakeReadFile(self.test_data)
619
osutils.pumpfile(from_file, to_file, -1, self.block_size)
621
# verify read size was equal to the maximum read size
622
self.assertEqual(from_file.get_max_read_size(), self.block_size)
623
self.assertEqual(from_file.get_read_count(), 4)
625
# report error if the data wasn't equal (we only report the size due
626
# to the length of the data)
627
response_data = to_file.getvalue()
628
if response_data != self.test_data:
629
message = "Data not equal. Expected %d bytes, received %d."
630
self.fail(message % (len(response_data), self.test_data_len))
632
def test_defaults(self):
633
"""Verifies that the default arguments will read to EOF -- this
634
test verifies that any existing usages of pumpfile will not be broken
635
with this new version."""
636
# retrieve data using default (old) pumpfile method
637
from_file = file_utils.FakeReadFile(self.test_data)
639
osutils.pumpfile(from_file, to_file)
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_report_activity(self):
650
def log_activity(length, direction):
651
activity.append((length, direction))
652
from_file = StringIO(self.test_data)
654
osutils.pumpfile(from_file, to_file, buff_size=500,
655
report_activity=log_activity, direction='read')
656
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
657
(36, 'read')], activity)
659
from_file = StringIO(self.test_data)
662
osutils.pumpfile(from_file, to_file, buff_size=500,
663
report_activity=log_activity, direction='write')
664
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
665
(36, 'write')], activity)
667
# And with a limited amount of data
668
from_file = StringIO(self.test_data)
671
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
672
report_activity=log_activity, direction='read')
673
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
677
class TestPumpStringFile(tests.TestCase):
679
def test_empty(self):
681
osutils.pump_string_file("", output)
682
self.assertEqual("", output.getvalue())
684
def test_more_than_segment_size(self):
686
osutils.pump_string_file("123456789", output, 2)
687
self.assertEqual("123456789", output.getvalue())
689
def test_segment_size(self):
691
osutils.pump_string_file("12", output, 2)
692
self.assertEqual("12", output.getvalue())
694
def test_segment_size_multiple(self):
696
osutils.pump_string_file("1234", output, 2)
697
self.assertEqual("1234", output.getvalue())
700
class TestRelpath(tests.TestCase):
702
def test_simple_relpath(self):
703
cwd = osutils.getcwd()
704
subdir = cwd + '/subdir'
705
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
707
def test_deep_relpath(self):
708
cwd = osutils.getcwd()
709
subdir = cwd + '/sub/subsubdir'
710
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
712
def test_not_relative(self):
713
self.assertRaises(errors.PathNotChild,
714
osutils.relpath, 'C:/path', 'H:/path')
715
self.assertRaises(errors.PathNotChild,
716
osutils.relpath, 'C:/', 'H:/path')
719
class TestSafeUnicode(tests.TestCase):
143
self.assertEqual(0222, osutils.get_umask())
145
self.assertEqual(0022, osutils.get_umask())
147
self.assertEqual(0002, osutils.get_umask())
149
self.assertEqual(0027, osutils.get_umask())
154
class TestSafeUnicode(TestCase):
721
156
def test_from_ascii_string(self):
722
157
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
731
166
self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
733
168
def test_bad_utf8_string(self):
734
self.assertRaises(errors.BzrBadParameterNotUnicode,
169
self.assertRaises(BzrBadParameterNotUnicode,
735
170
osutils.safe_unicode,
739
class TestSafeUtf8(tests.TestCase):
741
def test_from_ascii_string(self):
743
self.assertEqual('foobar', osutils.safe_utf8(f))
745
def test_from_unicode_string_ascii_contents(self):
746
self.assertEqual('bargam', osutils.safe_utf8(u'bargam'))
748
def test_from_unicode_string_unicode_contents(self):
749
self.assertEqual('bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
751
def test_from_utf8_string(self):
752
self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
754
def test_bad_utf8_string(self):
755
self.assertRaises(errors.BzrBadParameterNotUnicode,
756
osutils.safe_utf8, '\xbb\xbb')
759
class TestSafeRevisionId(tests.TestCase):
761
def test_from_ascii_string(self):
762
# this shouldn't give a warning because it's getting an ascii string
763
self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
765
def test_from_unicode_string_ascii_contents(self):
766
self.assertEqual('bargam',
767
osutils.safe_revision_id(u'bargam', warn=False))
769
def test_from_unicode_deprecated(self):
770
self.assertEqual('bargam',
771
self.callDeprecated([osutils._revision_id_warning],
772
osutils.safe_revision_id, u'bargam'))
774
def test_from_unicode_string_unicode_contents(self):
775
self.assertEqual('bargam\xc2\xae',
776
osutils.safe_revision_id(u'bargam\xae', warn=False))
778
def test_from_utf8_string(self):
779
self.assertEqual('foo\xc2\xae',
780
osutils.safe_revision_id('foo\xc2\xae'))
783
"""Currently, None is a valid revision_id"""
784
self.assertEqual(None, osutils.safe_revision_id(None))
787
class TestSafeFileId(tests.TestCase):
789
def test_from_ascii_string(self):
790
self.assertEqual('foobar', osutils.safe_file_id('foobar'))
792
def test_from_unicode_string_ascii_contents(self):
793
self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
795
def test_from_unicode_deprecated(self):
796
self.assertEqual('bargam',
797
self.callDeprecated([osutils._file_id_warning],
798
osutils.safe_file_id, u'bargam'))
800
def test_from_unicode_string_unicode_contents(self):
801
self.assertEqual('bargam\xc2\xae',
802
osutils.safe_file_id(u'bargam\xae', warn=False))
804
def test_from_utf8_string(self):
805
self.assertEqual('foo\xc2\xae',
806
osutils.safe_file_id('foo\xc2\xae'))
809
"""Currently, None is a valid revision_id"""
810
self.assertEqual(None, osutils.safe_file_id(None))
813
class TestWin32Funcs(tests.TestCase):
814
"""Test that _win32 versions of os utilities return appropriate paths."""
174
class TestWin32Funcs(TestCase):
175
"""Test that the _win32 versions of os utilities return appropriate paths."""
816
177
def test_abspath(self):
817
178
self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
818
179
self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
819
self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
820
self.assertEqual('//HOST/path', osutils._win32_abspath('//HOST/path'))
822
181
def test_realpath(self):
823
182
self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
824
183
self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
826
185
def test_pathjoin(self):
827
self.assertEqual('path/to/foo',
828
osutils._win32_pathjoin('path', 'to', 'foo'))
829
self.assertEqual('C:/foo',
830
osutils._win32_pathjoin('path\\to', 'C:\\foo'))
831
self.assertEqual('C:/foo',
832
osutils._win32_pathjoin('path/to', 'C:/foo'))
833
self.assertEqual('path/to/foo',
834
osutils._win32_pathjoin('path/to/', 'foo'))
835
self.assertEqual('/foo',
836
osutils._win32_pathjoin('C:/path/to/', '/foo'))
837
self.assertEqual('/foo',
838
osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
186
self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
187
self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
188
self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
189
self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
190
self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
191
self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
840
193
def test_normpath(self):
841
self.assertEqual('path/to/foo',
842
osutils._win32_normpath(r'path\\from\..\to\.\foo'))
843
self.assertEqual('path/to/foo',
844
osutils._win32_normpath('path//from/../to/./foo'))
194
self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
195
self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
846
197
def test_getcwd(self):
847
198
cwd = osutils._win32_getcwd()
1063
352
result.append((dirdetail, dirblock))
1065
354
self.assertTrue(found_bzrdir)
1066
self.assertExpectedBlocks(expected_dirblocks, result)
1067
# you can search a subdir only, with a supplied prefix.
1069
for dirblock in osutils.walkdirs('./1dir', '1dir'):
1070
result.append(dirblock)
1071
self.assertExpectedBlocks(expected_dirblocks[1:], result)
1073
def test_walkdirs_os_error(self):
1074
# <https://bugs.edge.launchpad.net/bzr/+bug/338653>
1075
# Pyrex readdir didn't raise useful messages if it had an error
1076
# reading the directory
1077
if sys.platform == 'win32':
1078
raise tests.TestNotApplicable(
1079
"readdir IOError not tested on win32")
1080
os.mkdir("test-unreadable")
1081
os.chmod("test-unreadable", 0000)
1082
# must chmod it back so that it can be removed
1083
self.addCleanup(os.chmod, "test-unreadable", 0700)
1084
# The error is not raised until the generator is actually evaluated.
1085
# (It would be ok if it happened earlier but at the moment it
1087
e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1088
self.assertEquals('./test-unreadable', e.filename)
1089
self.assertEquals(errno.EACCES, e.errno)
1090
# Ensure the message contains the file name
1091
self.assertContainsRe(str(e), "\./test-unreadable")
1093
def test__walkdirs_utf8(self):
1102
self.build_tree(tree)
1103
expected_dirblocks = [
1105
[('0file', '0file', 'file'),
1106
('1dir', '1dir', 'directory'),
1107
('2file', '2file', 'file'),
1110
(('1dir', './1dir'),
1111
[('1dir/0file', '0file', 'file'),
1112
('1dir/1dir', '1dir', 'directory'),
1115
(('1dir/1dir', './1dir/1dir'),
1121
found_bzrdir = False
1122
for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1123
if len(dirblock) and dirblock[0][1] == '.bzr':
1124
# this tests the filtering of selected paths
1127
result.append((dirdetail, dirblock))
1129
self.assertTrue(found_bzrdir)
1130
self.assertExpectedBlocks(expected_dirblocks, result)
1132
# you can search a subdir only, with a supplied prefix.
1134
for dirblock in osutils.walkdirs('./1dir', '1dir'):
1135
result.append(dirblock)
1136
self.assertExpectedBlocks(expected_dirblocks[1:], result)
1138
def _filter_out_stat(self, result):
1139
"""Filter out the stat value from the walkdirs result"""
1140
for dirdetail, dirblock in result:
1142
for info in dirblock:
1143
# Ignore info[3] which is the stat
1144
new_dirblock.append((info[0], info[1], info[2], info[4]))
1145
dirblock[:] = new_dirblock
1147
def _save_platform_info(self):
1148
self.overrideAttr(win32utils, 'winver')
1149
self.overrideAttr(osutils, '_fs_enc')
1150
self.overrideAttr(osutils, '_selected_dir_reader')
1152
def assertDirReaderIs(self, expected):
1153
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
1154
# Force it to redetect
1155
osutils._selected_dir_reader = None
1156
# Nothing to list, but should still trigger the selection logic
1157
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1158
self.assertIsInstance(osutils._selected_dir_reader, expected)
1160
def test_force_walkdirs_utf8_fs_utf8(self):
1161
self.requireFeature(UTF8DirReaderFeature)
1162
self._save_platform_info()
1163
win32utils.winver = None # Avoid the win32 detection code
1164
osutils._fs_enc = 'UTF-8'
1165
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1167
def test_force_walkdirs_utf8_fs_ascii(self):
1168
self.requireFeature(UTF8DirReaderFeature)
1169
self._save_platform_info()
1170
win32utils.winver = None # Avoid the win32 detection code
1171
osutils._fs_enc = 'US-ASCII'
1172
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1174
def test_force_walkdirs_utf8_fs_ANSI(self):
1175
self.requireFeature(UTF8DirReaderFeature)
1176
self._save_platform_info()
1177
win32utils.winver = None # Avoid the win32 detection code
1178
osutils._fs_enc = 'ANSI_X3.4-1968'
1179
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1181
def test_force_walkdirs_utf8_fs_latin1(self):
1182
self._save_platform_info()
1183
win32utils.winver = None # Avoid the win32 detection code
1184
osutils._fs_enc = 'latin1'
1185
self.assertDirReaderIs(osutils.UnicodeDirReader)
1187
def test_force_walkdirs_utf8_nt(self):
1188
# Disabled because the thunk of the whole walkdirs api is disabled.
1189
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1190
self._save_platform_info()
1191
win32utils.winver = 'Windows NT'
1192
from bzrlib._walkdirs_win32 import Win32ReadDir
1193
self.assertDirReaderIs(Win32ReadDir)
1195
def test_force_walkdirs_utf8_98(self):
1196
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1197
self._save_platform_info()
1198
win32utils.winver = 'Windows 98'
1199
self.assertDirReaderIs(osutils.UnicodeDirReader)
1201
def test_unicode_walkdirs(self):
1202
"""Walkdirs should always return unicode paths."""
1203
self.requireFeature(tests.UnicodeFilenameFeature)
1204
name0 = u'0file-\xb6'
1205
name1 = u'1dir-\u062c\u0648'
1206
name2 = u'2file-\u0633'
1210
name1 + '/' + name0,
1211
name1 + '/' + name1 + '/',
1214
self.build_tree(tree)
1215
expected_dirblocks = [
1217
[(name0, name0, 'file', './' + name0),
1218
(name1, name1, 'directory', './' + name1),
1219
(name2, name2, 'file', './' + name2),
1222
((name1, './' + name1),
1223
[(name1 + '/' + name0, name0, 'file', './' + name1
1225
(name1 + '/' + name1, name1, 'directory', './' + name1
1229
((name1 + '/' + name1, './' + name1 + '/' + name1),
1234
result = list(osutils.walkdirs('.'))
1235
self._filter_out_stat(result)
1236
self.assertEqual(expected_dirblocks, result)
1237
result = list(osutils.walkdirs(u'./'+name1, name1))
1238
self._filter_out_stat(result)
1239
self.assertEqual(expected_dirblocks[1:], result)
1241
def test_unicode__walkdirs_utf8(self):
1242
"""Walkdirs_utf8 should always return utf8 paths.
1244
The abspath portion might be in unicode or utf-8
1246
self.requireFeature(tests.UnicodeFilenameFeature)
1247
name0 = u'0file-\xb6'
1248
name1 = u'1dir-\u062c\u0648'
1249
name2 = u'2file-\u0633'
1253
name1 + '/' + name0,
1254
name1 + '/' + name1 + '/',
1257
self.build_tree(tree)
1258
name0 = name0.encode('utf8')
1259
name1 = name1.encode('utf8')
1260
name2 = name2.encode('utf8')
1262
expected_dirblocks = [
1264
[(name0, name0, 'file', './' + name0),
1265
(name1, name1, 'directory', './' + name1),
1266
(name2, name2, 'file', './' + name2),
1269
((name1, './' + name1),
1270
[(name1 + '/' + name0, name0, 'file', './' + name1
1272
(name1 + '/' + name1, name1, 'directory', './' + name1
1276
((name1 + '/' + name1, './' + name1 + '/' + name1),
1282
# For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1283
# all abspaths are Unicode, and encode them back into utf8.
1284
for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1285
self.assertIsInstance(dirdetail[0], str)
1286
if isinstance(dirdetail[1], unicode):
1287
dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1288
dirblock = [list(info) for info in dirblock]
1289
for info in dirblock:
1290
self.assertIsInstance(info[4], unicode)
1291
info[4] = info[4].encode('utf8')
1293
for info in dirblock:
1294
self.assertIsInstance(info[0], str)
1295
self.assertIsInstance(info[1], str)
1296
self.assertIsInstance(info[4], str)
1297
# Remove the stat information
1298
new_dirblock.append((info[0], info[1], info[2], info[4]))
1299
result.append((dirdetail, new_dirblock))
1300
self.assertEqual(expected_dirblocks, result)
1302
def test__walkdirs_utf8_with_unicode_fs(self):
1303
"""UnicodeDirReader should be a safe fallback everywhere
1305
The abspath portion should be in unicode
1307
self.requireFeature(tests.UnicodeFilenameFeature)
1308
# Use the unicode reader. TODO: split into driver-and-driven unit
1310
self._save_platform_info()
1311
osutils._selected_dir_reader = osutils.UnicodeDirReader()
1312
name0u = u'0file-\xb6'
1313
name1u = u'1dir-\u062c\u0648'
1314
name2u = u'2file-\u0633'
1318
name1u + '/' + name0u,
1319
name1u + '/' + name1u + '/',
1322
self.build_tree(tree)
1323
name0 = name0u.encode('utf8')
1324
name1 = name1u.encode('utf8')
1325
name2 = name2u.encode('utf8')
1327
# All of the abspaths should be in unicode, all of the relative paths
1329
expected_dirblocks = [
1331
[(name0, name0, 'file', './' + name0u),
1332
(name1, name1, 'directory', './' + name1u),
1333
(name2, name2, 'file', './' + name2u),
1336
((name1, './' + name1u),
1337
[(name1 + '/' + name0, name0, 'file', './' + name1u
1339
(name1 + '/' + name1, name1, 'directory', './' + name1u
1343
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1348
result = list(osutils._walkdirs_utf8('.'))
1349
self._filter_out_stat(result)
1350
self.assertEqual(expected_dirblocks, result)
1352
def test__walkdirs_utf8_win32readdir(self):
1353
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1354
self.requireFeature(tests.UnicodeFilenameFeature)
1355
from bzrlib._walkdirs_win32 import Win32ReadDir
1356
self._save_platform_info()
1357
osutils._selected_dir_reader = Win32ReadDir()
1358
name0u = u'0file-\xb6'
1359
name1u = u'1dir-\u062c\u0648'
1360
name2u = u'2file-\u0633'
1364
name1u + '/' + name0u,
1365
name1u + '/' + name1u + '/',
1368
self.build_tree(tree)
1369
name0 = name0u.encode('utf8')
1370
name1 = name1u.encode('utf8')
1371
name2 = name2u.encode('utf8')
1373
# All of the abspaths should be in unicode, all of the relative paths
1375
expected_dirblocks = [
1377
[(name0, name0, 'file', './' + name0u),
1378
(name1, name1, 'directory', './' + name1u),
1379
(name2, name2, 'file', './' + name2u),
1382
((name1, './' + name1u),
1383
[(name1 + '/' + name0, name0, 'file', './' + name1u
1385
(name1 + '/' + name1, name1, 'directory', './' + name1u
1389
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1394
result = list(osutils._walkdirs_utf8(u'.'))
1395
self._filter_out_stat(result)
1396
self.assertEqual(expected_dirblocks, result)
1398
def assertStatIsCorrect(self, path, win32stat):
1399
os_stat = os.stat(path)
1400
self.assertEqual(os_stat.st_size, win32stat.st_size)
1401
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1402
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1403
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1404
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1405
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1406
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1408
def test__walkdirs_utf_win32_find_file_stat_file(self):
1409
"""make sure our Stat values are valid"""
1410
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1411
self.requireFeature(tests.UnicodeFilenameFeature)
1412
from bzrlib._walkdirs_win32 import Win32ReadDir
1413
name0u = u'0file-\xb6'
1414
name0 = name0u.encode('utf8')
1415
self.build_tree([name0u])
1416
# I hate to sleep() here, but I'm trying to make the ctime different
1419
f = open(name0u, 'ab')
1421
f.write('just a small update')
1425
result = Win32ReadDir().read_dir('', u'.')
1427
self.assertEqual((name0, name0, 'file'), entry[:3])
1428
self.assertEqual(u'./' + name0u, entry[4])
1429
self.assertStatIsCorrect(entry[4], entry[3])
1430
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1432
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1433
"""make sure our Stat values are valid"""
1434
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1435
self.requireFeature(tests.UnicodeFilenameFeature)
1436
from bzrlib._walkdirs_win32 import Win32ReadDir
1437
name0u = u'0dir-\u062c\u0648'
1438
name0 = name0u.encode('utf8')
1439
self.build_tree([name0u + '/'])
1441
result = Win32ReadDir().read_dir('', u'.')
1443
self.assertEqual((name0, name0, 'directory'), entry[:3])
1444
self.assertEqual(u'./' + name0u, entry[4])
1445
self.assertStatIsCorrect(entry[4], entry[3])
355
self.assertEqual(expected_dirblocks,
356
[(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
357
# you can search a subdir only, with a supplied prefix.
359
for dirblock in osutils.walkdirs('./1dir', '1dir'):
360
result.append(dirblock)
361
self.assertEqual(expected_dirblocks[1:],
362
[(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1447
364
def assertPathCompare(self, path_less, path_greater):
1448
365
"""check that path_less and path_greater compare correctly."""
1623
589
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1624
590
self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1627
class TestSizeShaFile(tests.TestCaseInTempDir):
1629
def test_sha_empty(self):
1630
self.build_tree_contents([('foo', '')])
1631
expected_sha = osutils.sha_string('')
1633
self.addCleanup(f.close)
1634
size, sha = osutils.size_sha_file(f)
1635
self.assertEqual(0, size)
1636
self.assertEqual(expected_sha, sha)
1638
def test_sha_mixed_endings(self):
1639
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1640
self.build_tree_contents([('foo', text)])
1641
expected_sha = osutils.sha_string(text)
1642
f = open('foo', 'rb')
1643
self.addCleanup(f.close)
1644
size, sha = osutils.size_sha_file(f)
1645
self.assertEqual(38, size)
1646
self.assertEqual(expected_sha, sha)
1649
class TestShaFileByName(tests.TestCaseInTempDir):
1651
def test_sha_empty(self):
1652
self.build_tree_contents([('foo', '')])
1653
expected_sha = osutils.sha_string('')
1654
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1656
def test_sha_mixed_endings(self):
1657
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1658
self.build_tree_contents([('foo', text)])
1659
expected_sha = osutils.sha_string(text)
1660
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1663
class TestResourceLoading(tests.TestCaseInTempDir):
1665
def test_resource_string(self):
1666
# test resource in bzrlib
1667
text = osutils.resource_string('bzrlib', 'debug.py')
1668
self.assertContainsRe(text, "debug_flags = set()")
1669
# test resource under bzrlib
1670
text = osutils.resource_string('bzrlib.ui', 'text.py')
1671
self.assertContainsRe(text, "class TextUIFactory")
1672
# test unsupported package
1673
self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1675
# test unknown resource
1676
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1679
class TestReCompile(tests.TestCase):
1681
def test_re_compile_checked(self):
1682
r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1683
self.assertTrue(r.match('aaaa'))
1684
self.assertTrue(r.match('aAaA'))
1686
def test_re_compile_checked_error(self):
1687
# like https://bugs.launchpad.net/bzr/+bug/251352
1688
err = self.assertRaises(
1689
errors.BzrCommandError,
1690
osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1692
"Invalid regular expression in test case: '*': "
1693
"nothing to repeat",
1697
class TestDirReader(tests.TestCaseInTempDir):
1700
_dir_reader_class = None
1701
_native_to_unicode = None
1704
tests.TestCaseInTempDir.setUp(self)
1705
self.overrideAttr(osutils,
1706
'_selected_dir_reader', self._dir_reader_class())
1708
def _get_ascii_tree(self):
1716
expected_dirblocks = [
1718
[('0file', '0file', 'file'),
1719
('1dir', '1dir', 'directory'),
1720
('2file', '2file', 'file'),
1723
(('1dir', './1dir'),
1724
[('1dir/0file', '0file', 'file'),
1725
('1dir/1dir', '1dir', 'directory'),
1728
(('1dir/1dir', './1dir/1dir'),
1733
return tree, expected_dirblocks
1735
def test_walk_cur_dir(self):
1736
tree, expected_dirblocks = self._get_ascii_tree()
1737
self.build_tree(tree)
1738
result = list(osutils._walkdirs_utf8('.'))
1739
# Filter out stat and abspath
1740
self.assertEqual(expected_dirblocks,
1741
[(dirinfo, [line[0:3] for line in block])
1742
for dirinfo, block in result])
1744
def test_walk_sub_dir(self):
1745
tree, expected_dirblocks = self._get_ascii_tree()
1746
self.build_tree(tree)
1747
# you can search a subdir only, with a supplied prefix.
1748
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1749
# Filter out stat and abspath
1750
self.assertEqual(expected_dirblocks[1:],
1751
[(dirinfo, [line[0:3] for line in block])
1752
for dirinfo, block in result])
1754
def _get_unicode_tree(self):
1755
name0u = u'0file-\xb6'
1756
name1u = u'1dir-\u062c\u0648'
1757
name2u = u'2file-\u0633'
1761
name1u + '/' + name0u,
1762
name1u + '/' + name1u + '/',
1765
name0 = name0u.encode('UTF-8')
1766
name1 = name1u.encode('UTF-8')
1767
name2 = name2u.encode('UTF-8')
1768
expected_dirblocks = [
1770
[(name0, name0, 'file', './' + name0u),
1771
(name1, name1, 'directory', './' + name1u),
1772
(name2, name2, 'file', './' + name2u),
1775
((name1, './' + name1u),
1776
[(name1 + '/' + name0, name0, 'file', './' + name1u
1778
(name1 + '/' + name1, name1, 'directory', './' + name1u
1782
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1787
return tree, expected_dirblocks
1789
def _filter_out(self, raw_dirblocks):
1790
"""Filter out a walkdirs_utf8 result.
1792
stat field is removed, all native paths are converted to unicode
1794
filtered_dirblocks = []
1795
for dirinfo, block in raw_dirblocks:
1796
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1799
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1800
filtered_dirblocks.append((dirinfo, details))
1801
return filtered_dirblocks
1803
def test_walk_unicode_tree(self):
1804
self.requireFeature(tests.UnicodeFilenameFeature)
1805
tree, expected_dirblocks = self._get_unicode_tree()
1806
self.build_tree(tree)
1807
result = list(osutils._walkdirs_utf8('.'))
1808
self.assertEqual(expected_dirblocks, self._filter_out(result))
1810
def test_symlink(self):
1811
self.requireFeature(tests.SymlinkFeature)
1812
self.requireFeature(tests.UnicodeFilenameFeature)
1813
target = u'target\N{Euro Sign}'
1814
link_name = u'l\N{Euro Sign}nk'
1815
os.symlink(target, link_name)
1816
target_utf8 = target.encode('UTF-8')
1817
link_name_utf8 = link_name.encode('UTF-8')
1818
expected_dirblocks = [
1820
[(link_name_utf8, link_name_utf8,
1821
'symlink', './' + link_name),],
1823
result = list(osutils._walkdirs_utf8('.'))
1824
self.assertEqual(expected_dirblocks, self._filter_out(result))
1827
class TestReadLink(tests.TestCaseInTempDir):
1828
"""Exposes os.readlink() problems and the osutils solution.
1830
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1831
unicode string will be returned if a unicode string is passed.
1833
But prior python versions failed to properly encode the passed unicode
1836
_test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1839
super(tests.TestCaseInTempDir, self).setUp()
1840
self.link = u'l\N{Euro Sign}ink'
1841
self.target = u'targe\N{Euro Sign}t'
1842
os.symlink(self.target, self.link)
1844
def test_os_readlink_link_encoding(self):
1845
if sys.version_info < (2, 6):
1846
self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1848
self.assertEquals(self.target, os.readlink(self.link))
1850
def test_os_readlink_link_decoding(self):
1851
self.assertEquals(self.target.encode(osutils._fs_enc),
1852
os.readlink(self.link.encode(osutils._fs_enc)))
1855
class TestConcurrency(tests.TestCase):
1858
super(TestConcurrency, self).setUp()
1859
self.overrideAttr(osutils, '_cached_local_concurrency')
1861
def test_local_concurrency(self):
1862
concurrency = osutils.local_concurrency()
1863
self.assertIsInstance(concurrency, int)
1865
def test_local_concurrency_environment_variable(self):
1866
os.environ['BZR_CONCURRENCY'] = '2'
1867
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1868
os.environ['BZR_CONCURRENCY'] = '3'
1869
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1870
os.environ['BZR_CONCURRENCY'] = 'foo'
1871
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1873
def test_option_concurrency(self):
1874
os.environ['BZR_CONCURRENCY'] = '1'
1875
self.run_bzr('rocks --concurrency 42')
1876
# Command line overrides envrionment variable
1877
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1878
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1881
class TestFailedToLoadExtension(tests.TestCase):
1883
def _try_loading(self):
1885
import bzrlib._fictional_extension_py
1886
except ImportError, e:
1887
osutils.failed_to_load_extension(e)
1891
super(TestFailedToLoadExtension, self).setUp()
1892
self.overrideAttr(osutils, '_extension_load_failures', [])
1894
def test_failure_to_load(self):
1896
self.assertLength(1, osutils._extension_load_failures)
1897
self.assertEquals(osutils._extension_load_failures[0],
1898
"No module named _fictional_extension_py")
1900
def test_report_extension_load_failures_no_warning(self):
1901
self.assertTrue(self._try_loading())
1902
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
1903
# it used to give a Python warning; it no longer does
1904
self.assertLength(0, warnings)
1906
def test_report_extension_load_failures_message(self):
1908
trace.push_log_file(log)
1909
self.assertTrue(self._try_loading())
1910
osutils.report_extension_load_failures()
1911
self.assertContainsRe(
1913
r"bzr: warning: some compiled extensions could not be loaded; "
1914
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1918
class TestTerminalWidth(tests.TestCase):
1920
def replace_stdout(self, new):
1921
self.overrideAttr(sys, 'stdout', new)
1923
def replace__terminal_size(self, new):
1924
self.overrideAttr(osutils, '_terminal_size', new)
1926
def set_fake_tty(self):
1928
class I_am_a_tty(object):
1932
self.replace_stdout(I_am_a_tty())
1934
def test_default_values(self):
1935
self.assertEqual(80, osutils.default_terminal_width)
1937
def test_defaults_to_BZR_COLUMNS(self):
1938
# BZR_COLUMNS is set by the test framework
1939
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
1940
os.environ['BZR_COLUMNS'] = '12'
1941
self.assertEqual(12, osutils.terminal_width())
1943
def test_falls_back_to_COLUMNS(self):
1944
del os.environ['BZR_COLUMNS']
1945
self.assertNotEqual('42', os.environ['COLUMNS'])
1947
os.environ['COLUMNS'] = '42'
1948
self.assertEqual(42, osutils.terminal_width())
1950
def test_tty_default_without_columns(self):
1951
del os.environ['BZR_COLUMNS']
1952
del os.environ['COLUMNS']
1954
def terminal_size(w, h):
1958
# We need to override the osutils definition as it depends on the
1959
# running environment that we can't control (PQM running without a
1960
# controlling terminal is one example).
1961
self.replace__terminal_size(terminal_size)
1962
self.assertEqual(42, osutils.terminal_width())
1964
def test_non_tty_default_without_columns(self):
1965
del os.environ['BZR_COLUMNS']
1966
del os.environ['COLUMNS']
1967
self.replace_stdout(None)
1968
self.assertEqual(None, osutils.terminal_width())
1970
def test_no_TIOCGWINSZ(self):
1971
self.requireFeature(term_ios_feature)
1972
termios = term_ios_feature.module
1973
# bug 63539 is about a termios without TIOCGWINSZ attribute
1975
orig = termios.TIOCGWINSZ
1976
except AttributeError:
1977
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
1980
self.overrideAttr(termios, 'TIOCGWINSZ')
1981
del termios.TIOCGWINSZ
1982
del os.environ['BZR_COLUMNS']
1983
del os.environ['COLUMNS']
1984
# Whatever the result is, if we don't raise an exception, it's ok.
1985
osutils.terminal_width()
1987
class TestCreationOps(tests.TestCaseInTempDir):
1988
_test_needs_features = [features.chown_feature]
1991
tests.TestCaseInTempDir.setUp(self)
1992
self.overrideAttr(os, 'chown', self._dummy_chown)
1994
# params set by call to _dummy_chown
1995
self.path = self.uid = self.gid = None
1997
def _dummy_chown(self, path, uid, gid):
1998
self.path, self.uid, self.gid = path, uid, gid
2000
def test_copy_ownership_from_path(self):
2001
"""copy_ownership_from_path test with specified src."""
2003
f = open('test_file', 'wt')
2004
osutils.copy_ownership_from_path('test_file', ownsrc)
2007
self.assertEquals(self.path, 'test_file')
2008
self.assertEquals(self.uid, s.st_uid)
2009
self.assertEquals(self.gid, s.st_gid)
2011
def test_copy_ownership_nonesrc(self):
2012
"""copy_ownership_from_path test with src=None."""
2013
f = open('test_file', 'wt')
2014
# should use parent dir for permissions
2015
osutils.copy_ownership_from_path('test_file')
2018
self.assertEquals(self.path, 'test_file')
2019
self.assertEquals(self.uid, s.st_uid)
2020
self.assertEquals(self.gid, s.st_gid)
added
removed
bzrlib/tests/test_osutils.py
