1
# Copyright (C) 2005-2010 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
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
17
"""Tests for the osutils wrapper."""
19
from cStringIO import StringIO
35
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)
137
def test_fancy_rename(self):
138
# This should work everywhere
139
self.create_file('a', 'something in a\n')
140
self._fancy_rename('a', 'b')
141
self.failIfExists('a')
142
self.failUnlessExists('b')
143
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')
148
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
def test_rename(self):
163
# Rename should be semi-atomic on all platforms
164
self.create_file('a', 'something in a\n')
165
osutils.rename('a', 'b')
166
self.failIfExists('a')
167
self.failUnlessExists('b')
168
self.check_file_contents('b', 'something in a\n')
170
self.create_file('a', 'new something in a\n')
171
osutils.rename('b', 'a')
173
self.check_file_contents('a', 'something in a\n')
175
# 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
def test_01_rand_chars_empty(self):
198
result = osutils.rand_chars(0)
199
self.assertEqual(result, '')
201
def test_02_rand_chars_100(self):
202
result = osutils.rand_chars(100)
203
self.assertEqual(len(result), 100)
204
self.assertEqual(type(result), str)
205
self.assertContainsRe(result, r'^[a-z0-9]{100}$')
208
class TestIsInside(tests.TestCase):
210
def test_is_inside(self):
211
is_inside = osutils.is_inside
212
self.assertTrue(is_inside('src', 'src/foo.c'))
213
self.assertFalse(is_inside('src', 'srccontrol'))
214
self.assertTrue(is_inside('src', 'src/a/a/a/foo.c'))
215
self.assertTrue(is_inside('foo.c', 'foo.c'))
216
self.assertFalse(is_inside('foo.c', ''))
217
self.assertTrue(is_inside('', 'foo.c'))
219
def test_is_inside_any(self):
220
SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
221
for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
222
(['src'], SRC_FOO_C),
225
self.assert_(osutils.is_inside_any(dirs, fn))
226
for dirs, fn in [(['src'], 'srccontrol'),
227
(['src'], 'srccontrol/foo')]:
228
self.assertFalse(osutils.is_inside_any(dirs, fn))
230
def test_is_inside_or_parent_of_any(self):
231
for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
232
(['src'], 'src/foo.c'),
233
(['src/bar.c'], 'src'),
234
(['src/bar.c', 'bla/foo.c'], 'src'),
237
self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
239
for dirs, fn in [(['src'], 'srccontrol'),
240
(['srccontrol/foo.c'], 'src'),
241
(['src'], 'srccontrol/foo')]:
242
self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
245
class TestRmTree(tests.TestCaseInTempDir):
247
def test_rmtree(self):
248
# Check to remove tree with read-only files/dirs
250
f = file('dir/file', 'w')
253
# would like to also try making the directory readonly, but at the
254
# moment python shutil.rmtree doesn't handle that properly - it would
255
# need to chmod the directory before removing things inside it - deferred
256
# for now -- mbp 20060505
257
# osutils.make_readonly('dir')
258
osutils.make_readonly('dir/file')
260
osutils.rmtree('dir')
262
self.failIfExists('dir/file')
263
self.failIfExists('dir')
266
class TestDeleteAny(tests.TestCaseInTempDir):
268
def test_delete_any_readonly(self):
269
# from <https://bugs.launchpad.net/bzr/+bug/218206>
270
self.build_tree(['d/', 'f'])
271
osutils.make_readonly('d')
272
osutils.make_readonly('f')
274
osutils.delete_any('f')
275
osutils.delete_any('d')
278
class TestKind(tests.TestCaseInTempDir):
280
def test_file_kind(self):
281
self.build_tree(['file', 'dir/'])
282
self.assertEquals('file', osutils.file_kind('file'))
283
self.assertEquals('directory', osutils.file_kind('dir/'))
284
if osutils.has_symlinks():
285
os.symlink('symlink', 'symlink')
286
self.assertEquals('symlink', osutils.file_kind('symlink'))
288
# TODO: jam 20060529 Test a block device
290
os.lstat('/dev/null')
292
if e.errno not in (errno.ENOENT,):
295
self.assertEquals('chardev', osutils.file_kind('/dev/null'))
297
mkfifo = getattr(os, 'mkfifo', None)
301
self.assertEquals('fifo', osutils.file_kind('fifo'))
305
AF_UNIX = getattr(socket, 'AF_UNIX', None)
307
s = socket.socket(AF_UNIX)
310
self.assertEquals('socket', osutils.file_kind('socket'))
314
def test_kind_marker(self):
315
self.assertEqual("", osutils.kind_marker("file"))
316
self.assertEqual("/", osutils.kind_marker('directory'))
317
self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
318
self.assertEqual("@", osutils.kind_marker("symlink"))
319
self.assertEqual("+", osutils.kind_marker("tree-reference"))
320
self.assertEqual("", osutils.kind_marker("fifo"))
321
self.assertEqual("", osutils.kind_marker("socket"))
322
self.assertEqual("", osutils.kind_marker("unknown"))
325
class TestUmask(tests.TestCaseInTempDir):
327
def test_get_umask(self):
328
if sys.platform == 'win32':
329
# umask always returns '0', no way to set it
330
self.assertEqual(0, osutils.get_umask())
333
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):
721
def test_from_ascii_string(self):
722
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
724
def test_from_unicode_string_ascii_contents(self):
725
self.assertEqual(u'bargam', osutils.safe_unicode(u'bargam'))
727
def test_from_unicode_string_unicode_contents(self):
728
self.assertEqual(u'bargam\xae', osutils.safe_unicode(u'bargam\xae'))
730
def test_from_utf8_string(self):
731
self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
733
def test_bad_utf8_string(self):
734
self.assertRaises(errors.BzrBadParameterNotUnicode,
735
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."""
816
def test_abspath(self):
817
self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
818
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
def test_realpath(self):
823
self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
824
self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
826
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'))
840
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'))
846
def test_getcwd(self):
847
cwd = osutils._win32_getcwd()
848
os_cwd = os.getcwdu()
849
self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
850
# win32 is inconsistent whether it returns lower or upper case
851
# and even if it was consistent the user might type the other
852
# so we force it to uppercase
853
# running python.exe under cmd.exe return capital C:\\
854
# running win32 python inside a cygwin shell returns lowercase
855
self.assertEqual(os_cwd[0].upper(), cwd[0])
857
def test_fixdrive(self):
858
self.assertEqual('H:/foo', osutils._win32_fixdrive('h:/foo'))
859
self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
860
self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
862
def test_win98_abspath(self):
864
self.assertEqual('C:/foo', osutils._win98_abspath('C:\\foo'))
865
self.assertEqual('C:/foo', osutils._win98_abspath('C:/foo'))
867
self.assertEqual('//HOST/path', osutils._win98_abspath(r'\\HOST\path'))
868
self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
870
cwd = osutils.getcwd().rstrip('/')
871
drive = osutils._nt_splitdrive(cwd)[0]
872
self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
873
self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
876
self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
879
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
880
"""Test win32 functions that create files."""
882
def test_getcwd(self):
883
self.requireFeature(tests.UnicodeFilenameFeature)
886
# TODO: jam 20060427 This will probably fail on Mac OSX because
887
# it will change the normalization of B\xe5gfors
888
# Consider using a different unicode character, or make
889
# osutils.getcwd() renormalize the path.
890
self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
892
def test_minimum_path_selection(self):
893
self.assertEqual(set(),
894
osutils.minimum_path_selection([]))
895
self.assertEqual(set(['a']),
896
osutils.minimum_path_selection(['a']))
897
self.assertEqual(set(['a', 'b']),
898
osutils.minimum_path_selection(['a', 'b']))
899
self.assertEqual(set(['a/', 'b']),
900
osutils.minimum_path_selection(['a/', 'b']))
901
self.assertEqual(set(['a/', 'b']),
902
osutils.minimum_path_selection(['a/c', 'a/', 'b']))
903
self.assertEqual(set(['a-b', 'a', 'a0b']),
904
osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
906
def test_mkdtemp(self):
907
tmpdir = osutils._win32_mkdtemp(dir='.')
908
self.assertFalse('\\' in tmpdir)
910
def test_rename(self):
918
osutils._win32_rename('b', 'a')
919
self.failUnlessExists('a')
920
self.failIfExists('b')
921
self.assertFileEqual('baz\n', 'a')
923
def test_rename_missing_file(self):
929
osutils._win32_rename('b', 'a')
930
except (IOError, OSError), e:
931
self.assertEqual(errno.ENOENT, e.errno)
932
self.assertFileEqual('foo\n', 'a')
934
def test_rename_missing_dir(self):
937
osutils._win32_rename('b', 'a')
938
except (IOError, OSError), e:
939
self.assertEqual(errno.ENOENT, e.errno)
941
def test_rename_current_dir(self):
944
# You can't rename the working directory
945
# doing rename non-existant . usually
946
# just raises ENOENT, since non-existant
949
osutils._win32_rename('b', '.')
950
except (IOError, OSError), e:
951
self.assertEqual(errno.ENOENT, e.errno)
953
def test_splitpath(self):
954
def check(expected, path):
955
self.assertEqual(expected, osutils.splitpath(path))
958
check(['a', 'b'], 'a/b')
959
check(['a', 'b'], 'a/./b')
960
check(['a', '.b'], 'a/.b')
961
check(['a', '.b'], 'a\\.b')
963
self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
966
class TestParentDirectories(tests.TestCaseInTempDir):
967
"""Test osutils.parent_directories()"""
969
def test_parent_directories(self):
970
self.assertEqual([], osutils.parent_directories('a'))
971
self.assertEqual(['a'], osutils.parent_directories('a/b'))
972
self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
975
class TestMacFuncsDirs(tests.TestCaseInTempDir):
976
"""Test mac special functions that require directories."""
978
def test_getcwd(self):
979
self.requireFeature(tests.UnicodeFilenameFeature)
980
os.mkdir(u'B\xe5gfors')
981
os.chdir(u'B\xe5gfors')
982
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
984
def test_getcwd_nonnorm(self):
985
self.requireFeature(tests.UnicodeFilenameFeature)
986
# Test that _mac_getcwd() will normalize this path
987
os.mkdir(u'Ba\u030agfors')
988
os.chdir(u'Ba\u030agfors')
989
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
992
class TestChunksToLines(tests.TestCase):
994
def test_smoketest(self):
995
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
996
osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
997
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
998
osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
1000
def test_osutils_binding(self):
1001
from bzrlib.tests import test__chunks_to_lines
1002
if test__chunks_to_lines.compiled_chunkstolines_feature.available():
1003
from bzrlib._chunks_to_lines_pyx import chunks_to_lines
1005
from bzrlib._chunks_to_lines_py import chunks_to_lines
1006
self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1009
class TestSplitLines(tests.TestCase):
1011
def test_split_unicode(self):
1012
self.assertEqual([u'foo\n', u'bar\xae'],
1013
osutils.split_lines(u'foo\nbar\xae'))
1014
self.assertEqual([u'foo\n', u'bar\xae\n'],
1015
osutils.split_lines(u'foo\nbar\xae\n'))
1017
def test_split_with_carriage_returns(self):
1018
self.assertEqual(['foo\rbar\n'],
1019
osutils.split_lines('foo\rbar\n'))
1022
class TestWalkDirs(tests.TestCaseInTempDir):
1024
def assertExpectedBlocks(self, expected, result):
1025
self.assertEqual(expected,
1026
[(dirinfo, [line[0:3] for line in block])
1027
for dirinfo, block in result])
1029
def test_walkdirs(self):
1038
self.build_tree(tree)
1039
expected_dirblocks = [
1041
[('0file', '0file', 'file'),
1042
('1dir', '1dir', 'directory'),
1043
('2file', '2file', 'file'),
1046
(('1dir', './1dir'),
1047
[('1dir/0file', '0file', 'file'),
1048
('1dir/1dir', '1dir', 'directory'),
1051
(('1dir/1dir', './1dir/1dir'),
1057
found_bzrdir = False
1058
for dirdetail, dirblock in osutils.walkdirs('.'):
1059
if len(dirblock) and dirblock[0][1] == '.bzr':
1060
# this tests the filtering of selected paths
1063
result.append((dirdetail, dirblock))
1065
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])
1447
def assertPathCompare(self, path_less, path_greater):
1448
"""check that path_less and path_greater compare correctly."""
1449
self.assertEqual(0, osutils.compare_paths_prefix_order(
1450
path_less, path_less))
1451
self.assertEqual(0, osutils.compare_paths_prefix_order(
1452
path_greater, path_greater))
1453
self.assertEqual(-1, osutils.compare_paths_prefix_order(
1454
path_less, path_greater))
1455
self.assertEqual(1, osutils.compare_paths_prefix_order(
1456
path_greater, path_less))
1458
def test_compare_paths_prefix_order(self):
1459
# root before all else
1460
self.assertPathCompare("/", "/a")
1461
# alpha within a dir
1462
self.assertPathCompare("/a", "/b")
1463
self.assertPathCompare("/b", "/z")
1464
# high dirs before lower.
1465
self.assertPathCompare("/z", "/a/a")
1466
# except if the deeper dir should be output first
1467
self.assertPathCompare("/a/b/c", "/d/g")
1468
# lexical betwen dirs of the same height
1469
self.assertPathCompare("/a/z", "/z/z")
1470
self.assertPathCompare("/a/c/z", "/a/d/e")
1472
# this should also be consistent for no leading / paths
1473
# root before all else
1474
self.assertPathCompare("", "a")
1475
# alpha within a dir
1476
self.assertPathCompare("a", "b")
1477
self.assertPathCompare("b", "z")
1478
# high dirs before lower.
1479
self.assertPathCompare("z", "a/a")
1480
# except if the deeper dir should be output first
1481
self.assertPathCompare("a/b/c", "d/g")
1482
# lexical betwen dirs of the same height
1483
self.assertPathCompare("a/z", "z/z")
1484
self.assertPathCompare("a/c/z", "a/d/e")
1486
def test_path_prefix_sorting(self):
1487
"""Doing a sort on path prefix should match our sample data."""
1502
dir_sorted_paths = [
1518
sorted(original_paths, key=osutils.path_prefix_key))
1519
# using the comparison routine shoudl work too:
1522
sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1525
class TestCopyTree(tests.TestCaseInTempDir):
1527
def test_copy_basic_tree(self):
1528
self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1529
osutils.copy_tree('source', 'target')
1530
self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
1531
self.assertEqual(['c'], os.listdir('target/b'))
1533
def test_copy_tree_target_exists(self):
1534
self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c',
1536
osutils.copy_tree('source', 'target')
1537
self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
1538
self.assertEqual(['c'], os.listdir('target/b'))
1540
def test_copy_tree_symlinks(self):
1541
self.requireFeature(tests.SymlinkFeature)
1542
self.build_tree(['source/'])
1543
os.symlink('a/generic/path', 'source/lnk')
1544
osutils.copy_tree('source', 'target')
1545
self.assertEqual(['lnk'], os.listdir('target'))
1546
self.assertEqual('a/generic/path', os.readlink('target/lnk'))
1548
def test_copy_tree_handlers(self):
1549
processed_files = []
1550
processed_links = []
1551
def file_handler(from_path, to_path):
1552
processed_files.append(('f', from_path, to_path))
1553
def dir_handler(from_path, to_path):
1554
processed_files.append(('d', from_path, to_path))
1555
def link_handler(from_path, to_path):
1556
processed_links.append((from_path, to_path))
1557
handlers = {'file':file_handler,
1558
'directory':dir_handler,
1559
'symlink':link_handler,
1562
self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1563
if osutils.has_symlinks():
1564
os.symlink('a/generic/path', 'source/lnk')
1565
osutils.copy_tree('source', 'target', handlers=handlers)
1567
self.assertEqual([('d', 'source', 'target'),
1568
('f', 'source/a', 'target/a'),
1569
('d', 'source/b', 'target/b'),
1570
('f', 'source/b/c', 'target/b/c'),
1572
self.failIfExists('target')
1573
if osutils.has_symlinks():
1574
self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1577
class TestSetUnsetEnv(tests.TestCase):
1578
"""Test updating the environment"""
1581
super(TestSetUnsetEnv, self).setUp()
1583
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
1584
'Environment was not cleaned up properly.'
1585
' Variable BZR_TEST_ENV_VAR should not exist.')
1587
if 'BZR_TEST_ENV_VAR' in os.environ:
1588
del os.environ['BZR_TEST_ENV_VAR']
1589
self.addCleanup(cleanup)
1592
"""Test that we can set an env variable"""
1593
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1594
self.assertEqual(None, old)
1595
self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
1597
def test_double_set(self):
1598
"""Test that we get the old value out"""
1599
osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1600
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
1601
self.assertEqual('foo', old)
1602
self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
1604
def test_unicode(self):
1605
"""Environment can only contain plain strings
1607
So Unicode strings must be encoded.
1609
uni_val, env_val = tests.probe_unicode_in_user_encoding()
1611
raise tests.TestSkipped(
1612
'Cannot find a unicode character that works in encoding %s'
1613
% (osutils.get_user_encoding(),))
1615
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1616
self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1618
def test_unset(self):
1619
"""Test that passing None will remove the env var"""
1620
osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1621
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1622
self.assertEqual('foo', old)
1623
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1624
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
