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)
188
class TestRandChars(tests.TestCase):
190
def test_01_rand_chars_empty(self):
191
result = osutils.rand_chars(0)
192
self.assertEqual(result, '')
194
def test_02_rand_chars_100(self):
195
result = osutils.rand_chars(100)
196
self.assertEqual(len(result), 100)
197
self.assertEqual(type(result), str)
198
self.assertContainsRe(result, r'^[a-z0-9]{100}$')
201
class TestIsInside(tests.TestCase):
203
def test_is_inside(self):
204
is_inside = osutils.is_inside
205
self.assertTrue(is_inside('src', 'src/foo.c'))
206
self.assertFalse(is_inside('src', 'srccontrol'))
207
self.assertTrue(is_inside('src', 'src/a/a/a/foo.c'))
208
self.assertTrue(is_inside('foo.c', 'foo.c'))
209
self.assertFalse(is_inside('foo.c', ''))
210
self.assertTrue(is_inside('', 'foo.c'))
212
def test_is_inside_any(self):
213
SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
214
for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
215
(['src'], SRC_FOO_C),
218
self.assert_(osutils.is_inside_any(dirs, fn))
219
for dirs, fn in [(['src'], 'srccontrol'),
220
(['src'], 'srccontrol/foo')]:
221
self.assertFalse(osutils.is_inside_any(dirs, fn))
223
def test_is_inside_or_parent_of_any(self):
224
for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
225
(['src'], 'src/foo.c'),
226
(['src/bar.c'], 'src'),
227
(['src/bar.c', 'bla/foo.c'], 'src'),
230
self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
232
for dirs, fn in [(['src'], 'srccontrol'),
233
(['srccontrol/foo.c'], 'src'),
234
(['src'], 'srccontrol/foo')]:
235
self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
238
class TestRmTree(tests.TestCaseInTempDir):
240
def test_rmtree(self):
241
# Check to remove tree with read-only files/dirs
243
f = file('dir/file', 'w')
246
# would like to also try making the directory readonly, but at the
247
# moment python shutil.rmtree doesn't handle that properly - it would
248
# need to chmod the directory before removing things inside it - deferred
249
# for now -- mbp 20060505
250
# osutils.make_readonly('dir')
251
osutils.make_readonly('dir/file')
253
osutils.rmtree('dir')
255
self.failIfExists('dir/file')
256
self.failIfExists('dir')
259
class TestDeleteAny(tests.TestCaseInTempDir):
261
def test_delete_any_readonly(self):
262
# from <https://bugs.launchpad.net/bzr/+bug/218206>
263
self.build_tree(['d/', 'f'])
264
osutils.make_readonly('d')
265
osutils.make_readonly('f')
267
osutils.delete_any('f')
268
osutils.delete_any('d')
271
class TestKind(tests.TestCaseInTempDir):
273
def test_file_kind(self):
274
self.build_tree(['file', 'dir/'])
275
self.assertEquals('file', osutils.file_kind('file'))
276
self.assertEquals('directory', osutils.file_kind('dir/'))
277
if osutils.has_symlinks():
278
os.symlink('symlink', 'symlink')
279
self.assertEquals('symlink', osutils.file_kind('symlink'))
281
# TODO: jam 20060529 Test a block device
283
os.lstat('/dev/null')
285
if e.errno not in (errno.ENOENT,):
288
self.assertEquals('chardev', osutils.file_kind('/dev/null'))
290
mkfifo = getattr(os, 'mkfifo', None)
294
self.assertEquals('fifo', osutils.file_kind('fifo'))
298
AF_UNIX = getattr(socket, 'AF_UNIX', None)
300
s = socket.socket(AF_UNIX)
303
self.assertEquals('socket', osutils.file_kind('socket'))
307
def test_kind_marker(self):
308
self.assertEqual("", osutils.kind_marker("file"))
309
self.assertEqual("/", osutils.kind_marker('directory'))
310
self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
311
self.assertEqual("@", osutils.kind_marker("symlink"))
312
self.assertEqual("+", osutils.kind_marker("tree-reference"))
313
self.assertEqual("", osutils.kind_marker("fifo"))
314
self.assertEqual("", osutils.kind_marker("socket"))
315
self.assertEqual("", osutils.kind_marker("unknown"))
318
class TestUmask(tests.TestCaseInTempDir):
320
def test_get_umask(self):
321
if sys.platform == 'win32':
322
# umask always returns '0', no way to set it
323
self.assertEqual(0, osutils.get_umask())
326
orig_umask = osutils.get_umask()
327
self.addCleanup(os.umask, orig_umask)
329
self.assertEqual(0222, osutils.get_umask())
331
self.assertEqual(0022, osutils.get_umask())
333
self.assertEqual(0002, osutils.get_umask())
335
self.assertEqual(0027, osutils.get_umask())
338
class TestDateTime(tests.TestCase):
340
def assertFormatedDelta(self, expected, seconds):
341
"""Assert osutils.format_delta formats as expected"""
342
actual = osutils.format_delta(seconds)
343
self.assertEqual(expected, actual)
345
def test_format_delta(self):
346
self.assertFormatedDelta('0 seconds ago', 0)
347
self.assertFormatedDelta('1 second ago', 1)
348
self.assertFormatedDelta('10 seconds ago', 10)
349
self.assertFormatedDelta('59 seconds ago', 59)
350
self.assertFormatedDelta('89 seconds ago', 89)
351
self.assertFormatedDelta('1 minute, 30 seconds ago', 90)
352
self.assertFormatedDelta('3 minutes, 0 seconds ago', 180)
353
self.assertFormatedDelta('3 minutes, 1 second ago', 181)
354
self.assertFormatedDelta('10 minutes, 15 seconds ago', 615)
355
self.assertFormatedDelta('30 minutes, 59 seconds ago', 1859)
356
self.assertFormatedDelta('31 minutes, 0 seconds ago', 1860)
357
self.assertFormatedDelta('60 minutes, 0 seconds ago', 3600)
358
self.assertFormatedDelta('89 minutes, 59 seconds ago', 5399)
359
self.assertFormatedDelta('1 hour, 30 minutes ago', 5400)
360
self.assertFormatedDelta('2 hours, 30 minutes ago', 9017)
361
self.assertFormatedDelta('10 hours, 0 minutes ago', 36000)
362
self.assertFormatedDelta('24 hours, 0 minutes ago', 86400)
363
self.assertFormatedDelta('35 hours, 59 minutes ago', 129599)
364
self.assertFormatedDelta('36 hours, 0 minutes ago', 129600)
365
self.assertFormatedDelta('36 hours, 0 minutes ago', 129601)
366
self.assertFormatedDelta('36 hours, 1 minute ago', 129660)
367
self.assertFormatedDelta('36 hours, 1 minute ago', 129661)
368
self.assertFormatedDelta('84 hours, 10 minutes ago', 303002)
370
# We handle when time steps the wrong direction because computers
371
# don't have synchronized clocks.
372
self.assertFormatedDelta('84 hours, 10 minutes in the future', -303002)
373
self.assertFormatedDelta('1 second in the future', -1)
374
self.assertFormatedDelta('2 seconds in the future', -2)
376
def test_format_date(self):
377
self.assertRaises(errors.UnsupportedTimezoneFormat,
378
osutils.format_date, 0, timezone='foo')
379
self.assertIsInstance(osutils.format_date(0), str)
380
self.assertIsInstance(osutils.format_local_date(0), unicode)
381
# Testing for the actual value of the local weekday without
382
# duplicating the code from format_date is difficult.
383
# Instead blackbox.test_locale should check for localized
384
# dates once they do occur in output strings.
386
def test_format_date_with_offset_in_original_timezone(self):
387
self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
388
osutils.format_date_with_offset_in_original_timezone(0))
389
self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
390
osutils.format_date_with_offset_in_original_timezone(100000))
391
self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
392
osutils.format_date_with_offset_in_original_timezone(100000, 7200))
394
def test_local_time_offset(self):
395
"""Test that local_time_offset() returns a sane value."""
396
offset = osutils.local_time_offset()
397
self.assertTrue(isinstance(offset, int))
398
# Test that the offset is no more than a eighteen hours in
400
# Time zone handling is system specific, so it is difficult to
401
# do more specific tests, but a value outside of this range is
403
eighteen_hours = 18 * 3600
404
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
406
def test_local_time_offset_with_timestamp(self):
407
"""Test that local_time_offset() works with a timestamp."""
408
offset = osutils.local_time_offset(1000000000.1234567)
409
self.assertTrue(isinstance(offset, int))
410
eighteen_hours = 18 * 3600
411
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
414
class TestLinks(tests.TestCaseInTempDir):
416
def test_dereference_path(self):
417
self.requireFeature(tests.SymlinkFeature)
418
cwd = osutils.realpath('.')
420
bar_path = osutils.pathjoin(cwd, 'bar')
421
# Using './' to avoid bug #1213894 (first path component not
422
# dereferenced) in Python 2.4.1 and earlier
423
self.assertEqual(bar_path, osutils.realpath('./bar'))
424
os.symlink('bar', 'foo')
425
self.assertEqual(bar_path, osutils.realpath('./foo'))
427
# Does not dereference terminal symlinks
428
foo_path = osutils.pathjoin(cwd, 'foo')
429
self.assertEqual(foo_path, osutils.dereference_path('./foo'))
431
# Dereferences parent symlinks
433
baz_path = osutils.pathjoin(bar_path, 'baz')
434
self.assertEqual(baz_path, osutils.dereference_path('./foo/baz'))
436
# Dereferences parent symlinks that are the first path element
437
self.assertEqual(baz_path, osutils.dereference_path('foo/baz'))
439
# Dereferences parent symlinks in absolute paths
440
foo_baz_path = osutils.pathjoin(foo_path, 'baz')
441
self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
443
def test_changing_access(self):
444
f = file('file', 'w')
448
# Make a file readonly
449
osutils.make_readonly('file')
450
mode = os.lstat('file').st_mode
451
self.assertEqual(mode, mode & 0777555)
453
# Make a file writable
454
osutils.make_writable('file')
455
mode = os.lstat('file').st_mode
456
self.assertEqual(mode, mode | 0200)
458
if osutils.has_symlinks():
459
# should not error when handed a symlink
460
os.symlink('nonexistent', 'dangling')
461
osutils.make_readonly('dangling')
462
osutils.make_writable('dangling')
464
def test_host_os_dereferences_symlinks(self):
465
osutils.host_os_dereferences_symlinks()
468
class TestCanonicalRelPath(tests.TestCaseInTempDir):
470
_test_needs_features = [tests.CaseInsCasePresFilenameFeature]
472
def test_canonical_relpath_simple(self):
473
f = file('MixedCaseName', 'w')
475
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
476
self.failUnlessEqual('work/MixedCaseName', actual)
478
def test_canonical_relpath_missing_tail(self):
479
os.mkdir('MixedCaseParent')
480
actual = osutils.canonical_relpath(self.test_base_dir,
481
'mixedcaseparent/nochild')
482
self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
485
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
487
def assertRelpath(self, expected, base, path):
488
actual = osutils._cicp_canonical_relpath(base, path)
489
self.assertEqual(expected, actual)
491
def test_simple(self):
492
self.build_tree(['MixedCaseName'])
493
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
494
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
496
def test_subdir_missing_tail(self):
497
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
498
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
499
self.assertRelpath('MixedCaseParent/a_child', base,
500
'MixedCaseParent/a_child')
501
self.assertRelpath('MixedCaseParent/a_child', base,
502
'MixedCaseParent/A_Child')
503
self.assertRelpath('MixedCaseParent/not_child', base,
504
'MixedCaseParent/not_child')
506
def test_at_root_slash(self):
507
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
509
if osutils.MIN_ABS_PATHLENGTH > 1:
510
raise tests.TestSkipped('relpath requires %d chars'
511
% osutils.MIN_ABS_PATHLENGTH)
512
self.assertRelpath('foo', '/', '/foo')
514
def test_at_root_drive(self):
515
if sys.platform != 'win32':
516
raise tests.TestNotApplicable('we can only test drive-letter relative'
517
' paths on Windows where we have drive'
520
# The specific issue is that when at the root of a drive, 'abspath'
521
# returns "C:/" or just "/". However, the code assumes that abspath
522
# always returns something like "C:/foo" or "/foo" (no trailing slash).
523
self.assertRelpath('foo', 'C:/', 'C:/foo')
524
self.assertRelpath('foo', 'X:/', 'X:/foo')
525
self.assertRelpath('foo', 'X:/', 'X://foo')
528
class TestPumpFile(tests.TestCase):
529
"""Test pumpfile method."""
532
tests.TestCase.setUp(self)
533
# create a test datablock
534
self.block_size = 512
535
pattern = '0123456789ABCDEF'
536
self.test_data = pattern * (3 * self.block_size / len(pattern))
537
self.test_data_len = len(self.test_data)
539
def test_bracket_block_size(self):
540
"""Read data in blocks with the requested read size bracketing the
542
# make sure test data is larger than max read size
543
self.assertTrue(self.test_data_len > self.block_size)
545
from_file = file_utils.FakeReadFile(self.test_data)
548
# read (max / 2) bytes and verify read size wasn't affected
549
num_bytes_to_read = self.block_size / 2
550
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
551
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
552
self.assertEqual(from_file.get_read_count(), 1)
554
# read (max) bytes and verify read size wasn't affected
555
num_bytes_to_read = self.block_size
556
from_file.reset_read_count()
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 + 1) bytes and verify read size was limited
562
num_bytes_to_read = self.block_size + 1
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(), self.block_size)
566
self.assertEqual(from_file.get_read_count(), 2)
568
# finish reading the rest of the data
569
num_bytes_to_read = self.test_data_len - to_file.tell()
570
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
572
# report error if the data wasn't equal (we only report the size due
573
# to the length of the data)
574
response_data = to_file.getvalue()
575
if response_data != self.test_data:
576
message = "Data not equal. Expected %d bytes, received %d."
577
self.fail(message % (len(response_data), self.test_data_len))
579
def test_specified_size(self):
580
"""Request a transfer larger than the maximum block size and verify
581
that the maximum read doesn't exceed the block_size."""
582
# make sure test data is larger than max read size
583
self.assertTrue(self.test_data_len > self.block_size)
585
# retrieve data in blocks
586
from_file = file_utils.FakeReadFile(self.test_data)
588
osutils.pumpfile(from_file, to_file, self.test_data_len,
591
# verify read size was equal to the maximum read size
592
self.assertTrue(from_file.get_max_read_size() > 0)
593
self.assertEqual(from_file.get_max_read_size(), self.block_size)
594
self.assertEqual(from_file.get_read_count(), 3)
596
# report error if the data wasn't equal (we only report the size due
597
# to the length of the data)
598
response_data = to_file.getvalue()
599
if response_data != self.test_data:
600
message = "Data not equal. Expected %d bytes, received %d."
601
self.fail(message % (len(response_data), self.test_data_len))
603
def test_to_eof(self):
604
"""Read to end-of-file and verify that the reads are not larger than
605
the maximum read size."""
606
# make sure test data is larger than max read size
607
self.assertTrue(self.test_data_len > self.block_size)
609
# retrieve data to EOF
610
from_file = file_utils.FakeReadFile(self.test_data)
612
osutils.pumpfile(from_file, to_file, -1, self.block_size)
614
# verify read size was equal to the maximum read size
615
self.assertEqual(from_file.get_max_read_size(), self.block_size)
616
self.assertEqual(from_file.get_read_count(), 4)
618
# report error if the data wasn't equal (we only report the size due
619
# to the length of the data)
620
response_data = to_file.getvalue()
621
if response_data != self.test_data:
622
message = "Data not equal. Expected %d bytes, received %d."
623
self.fail(message % (len(response_data), self.test_data_len))
625
def test_defaults(self):
626
"""Verifies that the default arguments will read to EOF -- this
627
test verifies that any existing usages of pumpfile will not be broken
628
with this new version."""
629
# retrieve data using default (old) pumpfile method
630
from_file = file_utils.FakeReadFile(self.test_data)
632
osutils.pumpfile(from_file, to_file)
634
# report error if the data wasn't equal (we only report the size due
635
# to the length of the data)
636
response_data = to_file.getvalue()
637
if response_data != self.test_data:
638
message = "Data not equal. Expected %d bytes, received %d."
639
self.fail(message % (len(response_data), self.test_data_len))
641
def test_report_activity(self):
643
def log_activity(length, direction):
644
activity.append((length, direction))
645
from_file = StringIO(self.test_data)
647
osutils.pumpfile(from_file, to_file, buff_size=500,
648
report_activity=log_activity, direction='read')
649
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
650
(36, 'read')], activity)
652
from_file = StringIO(self.test_data)
655
osutils.pumpfile(from_file, to_file, buff_size=500,
656
report_activity=log_activity, direction='write')
657
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
658
(36, 'write')], activity)
660
# And with a limited amount of data
661
from_file = StringIO(self.test_data)
664
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
665
report_activity=log_activity, direction='read')
666
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
670
class TestPumpStringFile(tests.TestCase):
672
def test_empty(self):
674
osutils.pump_string_file("", output)
675
self.assertEqual("", output.getvalue())
677
def test_more_than_segment_size(self):
679
osutils.pump_string_file("123456789", output, 2)
680
self.assertEqual("123456789", output.getvalue())
682
def test_segment_size(self):
684
osutils.pump_string_file("12", output, 2)
685
self.assertEqual("12", output.getvalue())
687
def test_segment_size_multiple(self):
689
osutils.pump_string_file("1234", output, 2)
690
self.assertEqual("1234", output.getvalue())
693
class TestRelpath(tests.TestCase):
695
def test_simple_relpath(self):
696
cwd = osutils.getcwd()
697
subdir = cwd + '/subdir'
698
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
700
def test_deep_relpath(self):
701
cwd = osutils.getcwd()
702
subdir = cwd + '/sub/subsubdir'
703
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
705
def test_not_relative(self):
706
self.assertRaises(errors.PathNotChild,
707
osutils.relpath, 'C:/path', 'H:/path')
708
self.assertRaises(errors.PathNotChild,
709
osutils.relpath, 'C:/', 'H:/path')
712
class TestSafeUnicode(tests.TestCase):
714
def test_from_ascii_string(self):
715
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
717
def test_from_unicode_string_ascii_contents(self):
718
self.assertEqual(u'bargam', osutils.safe_unicode(u'bargam'))
720
def test_from_unicode_string_unicode_contents(self):
721
self.assertEqual(u'bargam\xae', osutils.safe_unicode(u'bargam\xae'))
723
def test_from_utf8_string(self):
724
self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
726
def test_bad_utf8_string(self):
727
self.assertRaises(errors.BzrBadParameterNotUnicode,
728
osutils.safe_unicode,
732
class TestSafeUtf8(tests.TestCase):
734
def test_from_ascii_string(self):
736
self.assertEqual('foobar', osutils.safe_utf8(f))
738
def test_from_unicode_string_ascii_contents(self):
739
self.assertEqual('bargam', osutils.safe_utf8(u'bargam'))
741
def test_from_unicode_string_unicode_contents(self):
742
self.assertEqual('bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
744
def test_from_utf8_string(self):
745
self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
747
def test_bad_utf8_string(self):
748
self.assertRaises(errors.BzrBadParameterNotUnicode,
749
osutils.safe_utf8, '\xbb\xbb')
752
class TestSafeRevisionId(tests.TestCase):
754
def test_from_ascii_string(self):
755
# this shouldn't give a warning because it's getting an ascii string
756
self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
758
def test_from_unicode_string_ascii_contents(self):
759
self.assertEqual('bargam',
760
osutils.safe_revision_id(u'bargam', warn=False))
762
def test_from_unicode_deprecated(self):
763
self.assertEqual('bargam',
764
self.callDeprecated([osutils._revision_id_warning],
765
osutils.safe_revision_id, u'bargam'))
767
def test_from_unicode_string_unicode_contents(self):
768
self.assertEqual('bargam\xc2\xae',
769
osutils.safe_revision_id(u'bargam\xae', warn=False))
771
def test_from_utf8_string(self):
772
self.assertEqual('foo\xc2\xae',
773
osutils.safe_revision_id('foo\xc2\xae'))
776
"""Currently, None is a valid revision_id"""
777
self.assertEqual(None, osutils.safe_revision_id(None))
780
class TestSafeFileId(tests.TestCase):
782
def test_from_ascii_string(self):
783
self.assertEqual('foobar', osutils.safe_file_id('foobar'))
785
def test_from_unicode_string_ascii_contents(self):
786
self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
788
def test_from_unicode_deprecated(self):
789
self.assertEqual('bargam',
790
self.callDeprecated([osutils._file_id_warning],
791
osutils.safe_file_id, u'bargam'))
793
def test_from_unicode_string_unicode_contents(self):
794
self.assertEqual('bargam\xc2\xae',
795
osutils.safe_file_id(u'bargam\xae', warn=False))
797
def test_from_utf8_string(self):
798
self.assertEqual('foo\xc2\xae',
799
osutils.safe_file_id('foo\xc2\xae'))
802
"""Currently, None is a valid revision_id"""
803
self.assertEqual(None, osutils.safe_file_id(None))
806
class TestWin32Funcs(tests.TestCase):
807
"""Test that _win32 versions of os utilities return appropriate paths."""
809
def test_abspath(self):
810
self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
811
self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
812
self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
813
self.assertEqual('//HOST/path', osutils._win32_abspath('//HOST/path'))
815
def test_realpath(self):
816
self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
817
self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
819
def test_pathjoin(self):
820
self.assertEqual('path/to/foo',
821
osutils._win32_pathjoin('path', 'to', 'foo'))
822
self.assertEqual('C:/foo',
823
osutils._win32_pathjoin('path\\to', 'C:\\foo'))
824
self.assertEqual('C:/foo',
825
osutils._win32_pathjoin('path/to', 'C:/foo'))
826
self.assertEqual('path/to/foo',
827
osutils._win32_pathjoin('path/to/', 'foo'))
828
self.assertEqual('/foo',
829
osutils._win32_pathjoin('C:/path/to/', '/foo'))
830
self.assertEqual('/foo',
831
osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
833
def test_normpath(self):
834
self.assertEqual('path/to/foo',
835
osutils._win32_normpath(r'path\\from\..\to\.\foo'))
836
self.assertEqual('path/to/foo',
837
osutils._win32_normpath('path//from/../to/./foo'))
839
def test_getcwd(self):
840
cwd = osutils._win32_getcwd()
841
os_cwd = os.getcwdu()
842
self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
843
# win32 is inconsistent whether it returns lower or upper case
844
# and even if it was consistent the user might type the other
845
# so we force it to uppercase
846
# running python.exe under cmd.exe return capital C:\\
847
# running win32 python inside a cygwin shell returns lowercase
848
self.assertEqual(os_cwd[0].upper(), cwd[0])
850
def test_fixdrive(self):
851
self.assertEqual('H:/foo', osutils._win32_fixdrive('h:/foo'))
852
self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
853
self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
855
def test_win98_abspath(self):
857
self.assertEqual('C:/foo', osutils._win98_abspath('C:\\foo'))
858
self.assertEqual('C:/foo', osutils._win98_abspath('C:/foo'))
860
self.assertEqual('//HOST/path', osutils._win98_abspath(r'\\HOST\path'))
861
self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
863
cwd = osutils.getcwd().rstrip('/')
864
drive = osutils._nt_splitdrive(cwd)[0]
865
self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
866
self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
869
self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
872
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
873
"""Test win32 functions that create files."""
875
def test_getcwd(self):
876
self.requireFeature(tests.UnicodeFilenameFeature)
879
# TODO: jam 20060427 This will probably fail on Mac OSX because
880
# it will change the normalization of B\xe5gfors
881
# Consider using a different unicode character, or make
882
# osutils.getcwd() renormalize the path.
883
self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
885
def test_minimum_path_selection(self):
886
self.assertEqual(set(),
887
osutils.minimum_path_selection([]))
888
self.assertEqual(set(['a']),
889
osutils.minimum_path_selection(['a']))
890
self.assertEqual(set(['a', 'b']),
891
osutils.minimum_path_selection(['a', 'b']))
892
self.assertEqual(set(['a/', 'b']),
893
osutils.minimum_path_selection(['a/', 'b']))
894
self.assertEqual(set(['a/', 'b']),
895
osutils.minimum_path_selection(['a/c', 'a/', 'b']))
896
self.assertEqual(set(['a-b', 'a', 'a0b']),
897
osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
899
def test_mkdtemp(self):
900
tmpdir = osutils._win32_mkdtemp(dir='.')
901
self.assertFalse('\\' in tmpdir)
903
def test_rename(self):
911
osutils._win32_rename('b', 'a')
912
self.failUnlessExists('a')
913
self.failIfExists('b')
914
self.assertFileEqual('baz\n', 'a')
916
def test_rename_missing_file(self):
922
osutils._win32_rename('b', 'a')
923
except (IOError, OSError), e:
924
self.assertEqual(errno.ENOENT, e.errno)
925
self.assertFileEqual('foo\n', 'a')
927
def test_rename_missing_dir(self):
930
osutils._win32_rename('b', 'a')
931
except (IOError, OSError), e:
932
self.assertEqual(errno.ENOENT, e.errno)
934
def test_rename_current_dir(self):
937
# You can't rename the working directory
938
# doing rename non-existant . usually
939
# just raises ENOENT, since non-existant
942
osutils._win32_rename('b', '.')
943
except (IOError, OSError), e:
944
self.assertEqual(errno.ENOENT, e.errno)
946
def test_splitpath(self):
947
def check(expected, path):
948
self.assertEqual(expected, osutils.splitpath(path))
951
check(['a', 'b'], 'a/b')
952
check(['a', 'b'], 'a/./b')
953
check(['a', '.b'], 'a/.b')
954
check(['a', '.b'], 'a\\.b')
956
self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
959
class TestParentDirectories(tests.TestCaseInTempDir):
960
"""Test osutils.parent_directories()"""
962
def test_parent_directories(self):
963
self.assertEqual([], osutils.parent_directories('a'))
964
self.assertEqual(['a'], osutils.parent_directories('a/b'))
965
self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
968
class TestMacFuncsDirs(tests.TestCaseInTempDir):
969
"""Test mac special functions that require directories."""
971
def test_getcwd(self):
972
self.requireFeature(tests.UnicodeFilenameFeature)
973
os.mkdir(u'B\xe5gfors')
974
os.chdir(u'B\xe5gfors')
975
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
977
def test_getcwd_nonnorm(self):
978
self.requireFeature(tests.UnicodeFilenameFeature)
979
# Test that _mac_getcwd() will normalize this path
980
os.mkdir(u'Ba\u030agfors')
981
os.chdir(u'Ba\u030agfors')
982
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
985
class TestChunksToLines(tests.TestCase):
987
def test_smoketest(self):
988
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
989
osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
990
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
991
osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
993
def test_osutils_binding(self):
994
from bzrlib.tests import test__chunks_to_lines
995
if test__chunks_to_lines.compiled_chunkstolines_feature.available():
996
from bzrlib._chunks_to_lines_pyx import chunks_to_lines
998
from bzrlib._chunks_to_lines_py import chunks_to_lines
999
self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1002
class TestSplitLines(tests.TestCase):
1004
def test_split_unicode(self):
1005
self.assertEqual([u'foo\n', u'bar\xae'],
1006
osutils.split_lines(u'foo\nbar\xae'))
1007
self.assertEqual([u'foo\n', u'bar\xae\n'],
1008
osutils.split_lines(u'foo\nbar\xae\n'))
1010
def test_split_with_carriage_returns(self):
1011
self.assertEqual(['foo\rbar\n'],
1012
osutils.split_lines('foo\rbar\n'))
1015
class TestWalkDirs(tests.TestCaseInTempDir):
1017
def assertExpectedBlocks(self, expected, result):
1018
self.assertEqual(expected,
1019
[(dirinfo, [line[0:3] for line in block])
1020
for dirinfo, block in result])
1022
def test_walkdirs(self):
1031
self.build_tree(tree)
1032
expected_dirblocks = [
1034
[('0file', '0file', 'file'),
1035
('1dir', '1dir', 'directory'),
1036
('2file', '2file', 'file'),
1039
(('1dir', './1dir'),
1040
[('1dir/0file', '0file', 'file'),
1041
('1dir/1dir', '1dir', 'directory'),
1044
(('1dir/1dir', './1dir/1dir'),
1050
found_bzrdir = False
1051
for dirdetail, dirblock in osutils.walkdirs('.'):
1052
if len(dirblock) and dirblock[0][1] == '.bzr':
1053
# this tests the filtering of selected paths
1056
result.append((dirdetail, dirblock))
1058
self.assertTrue(found_bzrdir)
1059
self.assertExpectedBlocks(expected_dirblocks, result)
1060
# you can search a subdir only, with a supplied prefix.
1062
for dirblock in osutils.walkdirs('./1dir', '1dir'):
1063
result.append(dirblock)
1064
self.assertExpectedBlocks(expected_dirblocks[1:], result)
1066
def test_walkdirs_os_error(self):
1067
# <https://bugs.edge.launchpad.net/bzr/+bug/338653>
1068
# Pyrex readdir didn't raise useful messages if it had an error
1069
# reading the directory
1070
if sys.platform == 'win32':
1071
raise tests.TestNotApplicable(
1072
"readdir IOError not tested on win32")
1073
os.mkdir("test-unreadable")
1074
os.chmod("test-unreadable", 0000)
1075
# must chmod it back so that it can be removed
1076
self.addCleanup(os.chmod, "test-unreadable", 0700)
1077
# The error is not raised until the generator is actually evaluated.
1078
# (It would be ok if it happened earlier but at the moment it
1080
e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1081
self.assertEquals('./test-unreadable', e.filename)
1082
self.assertEquals(errno.EACCES, e.errno)
1083
# Ensure the message contains the file name
1084
self.assertContainsRe(str(e), "\./test-unreadable")
1086
def test__walkdirs_utf8(self):
1095
self.build_tree(tree)
1096
expected_dirblocks = [
1098
[('0file', '0file', 'file'),
1099
('1dir', '1dir', 'directory'),
1100
('2file', '2file', 'file'),
1103
(('1dir', './1dir'),
1104
[('1dir/0file', '0file', 'file'),
1105
('1dir/1dir', '1dir', 'directory'),
1108
(('1dir/1dir', './1dir/1dir'),
1114
found_bzrdir = False
1115
for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1116
if len(dirblock) and dirblock[0][1] == '.bzr':
1117
# this tests the filtering of selected paths
1120
result.append((dirdetail, dirblock))
1122
self.assertTrue(found_bzrdir)
1123
self.assertExpectedBlocks(expected_dirblocks, result)
1125
# you can search a subdir only, with a supplied prefix.
1127
for dirblock in osutils.walkdirs('./1dir', '1dir'):
1128
result.append(dirblock)
1129
self.assertExpectedBlocks(expected_dirblocks[1:], result)
1131
def _filter_out_stat(self, result):
1132
"""Filter out the stat value from the walkdirs result"""
1133
for dirdetail, dirblock in result:
1135
for info in dirblock:
1136
# Ignore info[3] which is the stat
1137
new_dirblock.append((info[0], info[1], info[2], info[4]))
1138
dirblock[:] = new_dirblock
1140
def _save_platform_info(self):
1141
self.overrideAttr(win32utils, 'winver')
1142
self.overrideAttr(osutils, '_fs_enc')
1143
self.overrideAttr(osutils, '_selected_dir_reader')
1145
def assertDirReaderIs(self, expected):
1146
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
1147
# Force it to redetect
1148
osutils._selected_dir_reader = None
1149
# Nothing to list, but should still trigger the selection logic
1150
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1151
self.assertIsInstance(osutils._selected_dir_reader, expected)
1153
def test_force_walkdirs_utf8_fs_utf8(self):
1154
self.requireFeature(UTF8DirReaderFeature)
1155
self._save_platform_info()
1156
win32utils.winver = None # Avoid the win32 detection code
1157
osutils._fs_enc = 'UTF-8'
1158
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1160
def test_force_walkdirs_utf8_fs_ascii(self):
1161
self.requireFeature(UTF8DirReaderFeature)
1162
self._save_platform_info()
1163
win32utils.winver = None # Avoid the win32 detection code
1164
osutils._fs_enc = 'US-ASCII'
1165
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1167
def test_force_walkdirs_utf8_fs_ANSI(self):
1168
self.requireFeature(UTF8DirReaderFeature)
1169
self._save_platform_info()
1170
win32utils.winver = None # Avoid the win32 detection code
1171
osutils._fs_enc = 'ANSI_X3.4-1968'
1172
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1174
def test_force_walkdirs_utf8_fs_latin1(self):
1175
self._save_platform_info()
1176
win32utils.winver = None # Avoid the win32 detection code
1177
osutils._fs_enc = 'latin1'
1178
self.assertDirReaderIs(osutils.UnicodeDirReader)
1180
def test_force_walkdirs_utf8_nt(self):
1181
# Disabled because the thunk of the whole walkdirs api is disabled.
1182
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1183
self._save_platform_info()
1184
win32utils.winver = 'Windows NT'
1185
from bzrlib._walkdirs_win32 import Win32ReadDir
1186
self.assertDirReaderIs(Win32ReadDir)
1188
def test_force_walkdirs_utf8_98(self):
1189
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1190
self._save_platform_info()
1191
win32utils.winver = 'Windows 98'
1192
self.assertDirReaderIs(osutils.UnicodeDirReader)
1194
def test_unicode_walkdirs(self):
1195
"""Walkdirs should always return unicode paths."""
1196
self.requireFeature(tests.UnicodeFilenameFeature)
1197
name0 = u'0file-\xb6'
1198
name1 = u'1dir-\u062c\u0648'
1199
name2 = u'2file-\u0633'
1203
name1 + '/' + name0,
1204
name1 + '/' + name1 + '/',
1207
self.build_tree(tree)
1208
expected_dirblocks = [
1210
[(name0, name0, 'file', './' + name0),
1211
(name1, name1, 'directory', './' + name1),
1212
(name2, name2, 'file', './' + name2),
1215
((name1, './' + name1),
1216
[(name1 + '/' + name0, name0, 'file', './' + name1
1218
(name1 + '/' + name1, name1, 'directory', './' + name1
1222
((name1 + '/' + name1, './' + name1 + '/' + name1),
1227
result = list(osutils.walkdirs('.'))
1228
self._filter_out_stat(result)
1229
self.assertEqual(expected_dirblocks, result)
1230
result = list(osutils.walkdirs(u'./'+name1, name1))
1231
self._filter_out_stat(result)
1232
self.assertEqual(expected_dirblocks[1:], result)
1234
def test_unicode__walkdirs_utf8(self):
1235
"""Walkdirs_utf8 should always return utf8 paths.
1237
The abspath portion might be in unicode or utf-8
1239
self.requireFeature(tests.UnicodeFilenameFeature)
1240
name0 = u'0file-\xb6'
1241
name1 = u'1dir-\u062c\u0648'
1242
name2 = u'2file-\u0633'
1246
name1 + '/' + name0,
1247
name1 + '/' + name1 + '/',
1250
self.build_tree(tree)
1251
name0 = name0.encode('utf8')
1252
name1 = name1.encode('utf8')
1253
name2 = name2.encode('utf8')
1255
expected_dirblocks = [
1257
[(name0, name0, 'file', './' + name0),
1258
(name1, name1, 'directory', './' + name1),
1259
(name2, name2, 'file', './' + name2),
1262
((name1, './' + name1),
1263
[(name1 + '/' + name0, name0, 'file', './' + name1
1265
(name1 + '/' + name1, name1, 'directory', './' + name1
1269
((name1 + '/' + name1, './' + name1 + '/' + name1),
1275
# For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1276
# all abspaths are Unicode, and encode them back into utf8.
1277
for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1278
self.assertIsInstance(dirdetail[0], str)
1279
if isinstance(dirdetail[1], unicode):
1280
dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1281
dirblock = [list(info) for info in dirblock]
1282
for info in dirblock:
1283
self.assertIsInstance(info[4], unicode)
1284
info[4] = info[4].encode('utf8')
1286
for info in dirblock:
1287
self.assertIsInstance(info[0], str)
1288
self.assertIsInstance(info[1], str)
1289
self.assertIsInstance(info[4], str)
1290
# Remove the stat information
1291
new_dirblock.append((info[0], info[1], info[2], info[4]))
1292
result.append((dirdetail, new_dirblock))
1293
self.assertEqual(expected_dirblocks, result)
1295
def test__walkdirs_utf8_with_unicode_fs(self):
1296
"""UnicodeDirReader should be a safe fallback everywhere
1298
The abspath portion should be in unicode
1300
self.requireFeature(tests.UnicodeFilenameFeature)
1301
# Use the unicode reader. TODO: split into driver-and-driven unit
1303
self._save_platform_info()
1304
osutils._selected_dir_reader = osutils.UnicodeDirReader()
1305
name0u = u'0file-\xb6'
1306
name1u = u'1dir-\u062c\u0648'
1307
name2u = u'2file-\u0633'
1311
name1u + '/' + name0u,
1312
name1u + '/' + name1u + '/',
1315
self.build_tree(tree)
1316
name0 = name0u.encode('utf8')
1317
name1 = name1u.encode('utf8')
1318
name2 = name2u.encode('utf8')
1320
# All of the abspaths should be in unicode, all of the relative paths
1322
expected_dirblocks = [
1324
[(name0, name0, 'file', './' + name0u),
1325
(name1, name1, 'directory', './' + name1u),
1326
(name2, name2, 'file', './' + name2u),
1329
((name1, './' + name1u),
1330
[(name1 + '/' + name0, name0, 'file', './' + name1u
1332
(name1 + '/' + name1, name1, 'directory', './' + name1u
1336
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1341
result = list(osutils._walkdirs_utf8('.'))
1342
self._filter_out_stat(result)
1343
self.assertEqual(expected_dirblocks, result)
1345
def test__walkdirs_utf8_win32readdir(self):
1346
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1347
self.requireFeature(tests.UnicodeFilenameFeature)
1348
from bzrlib._walkdirs_win32 import Win32ReadDir
1349
self._save_platform_info()
1350
osutils._selected_dir_reader = Win32ReadDir()
1351
name0u = u'0file-\xb6'
1352
name1u = u'1dir-\u062c\u0648'
1353
name2u = u'2file-\u0633'
1357
name1u + '/' + name0u,
1358
name1u + '/' + name1u + '/',
1361
self.build_tree(tree)
1362
name0 = name0u.encode('utf8')
1363
name1 = name1u.encode('utf8')
1364
name2 = name2u.encode('utf8')
1366
# All of the abspaths should be in unicode, all of the relative paths
1368
expected_dirblocks = [
1370
[(name0, name0, 'file', './' + name0u),
1371
(name1, name1, 'directory', './' + name1u),
1372
(name2, name2, 'file', './' + name2u),
1375
((name1, './' + name1u),
1376
[(name1 + '/' + name0, name0, 'file', './' + name1u
1378
(name1 + '/' + name1, name1, 'directory', './' + name1u
1382
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1387
result = list(osutils._walkdirs_utf8(u'.'))
1388
self._filter_out_stat(result)
1389
self.assertEqual(expected_dirblocks, result)
1391
def assertStatIsCorrect(self, path, win32stat):
1392
os_stat = os.stat(path)
1393
self.assertEqual(os_stat.st_size, win32stat.st_size)
1394
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1395
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1396
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1397
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1398
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1399
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1401
def test__walkdirs_utf_win32_find_file_stat_file(self):
1402
"""make sure our Stat values are valid"""
1403
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1404
self.requireFeature(tests.UnicodeFilenameFeature)
1405
from bzrlib._walkdirs_win32 import Win32ReadDir
1406
name0u = u'0file-\xb6'
1407
name0 = name0u.encode('utf8')
1408
self.build_tree([name0u])
1409
# I hate to sleep() here, but I'm trying to make the ctime different
1412
f = open(name0u, 'ab')
1414
f.write('just a small update')
1418
result = Win32ReadDir().read_dir('', u'.')
1420
self.assertEqual((name0, name0, 'file'), entry[:3])
1421
self.assertEqual(u'./' + name0u, entry[4])
1422
self.assertStatIsCorrect(entry[4], entry[3])
1423
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1425
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1426
"""make sure our Stat values are valid"""
1427
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1428
self.requireFeature(tests.UnicodeFilenameFeature)
1429
from bzrlib._walkdirs_win32 import Win32ReadDir
1430
name0u = u'0dir-\u062c\u0648'
1431
name0 = name0u.encode('utf8')
1432
self.build_tree([name0u + '/'])
1434
result = Win32ReadDir().read_dir('', u'.')
1436
self.assertEqual((name0, name0, 'directory'), entry[:3])
1437
self.assertEqual(u'./' + name0u, entry[4])
1438
self.assertStatIsCorrect(entry[4], entry[3])
1440
def assertPathCompare(self, path_less, path_greater):
1441
"""check that path_less and path_greater compare correctly."""
1442
self.assertEqual(0, osutils.compare_paths_prefix_order(
1443
path_less, path_less))
1444
self.assertEqual(0, osutils.compare_paths_prefix_order(
1445
path_greater, path_greater))
1446
self.assertEqual(-1, osutils.compare_paths_prefix_order(
1447
path_less, path_greater))
1448
self.assertEqual(1, osutils.compare_paths_prefix_order(
1449
path_greater, path_less))
1451
def test_compare_paths_prefix_order(self):
1452
# root before all else
1453
self.assertPathCompare("/", "/a")
1454
# alpha within a dir
1455
self.assertPathCompare("/a", "/b")
1456
self.assertPathCompare("/b", "/z")
1457
# high dirs before lower.
1458
self.assertPathCompare("/z", "/a/a")
1459
# except if the deeper dir should be output first
1460
self.assertPathCompare("/a/b/c", "/d/g")
1461
# lexical betwen dirs of the same height
1462
self.assertPathCompare("/a/z", "/z/z")
1463
self.assertPathCompare("/a/c/z", "/a/d/e")
1465
# this should also be consistent for no leading / paths
1466
# root before all else
1467
self.assertPathCompare("", "a")
1468
# alpha within a dir
1469
self.assertPathCompare("a", "b")
1470
self.assertPathCompare("b", "z")
1471
# high dirs before lower.
1472
self.assertPathCompare("z", "a/a")
1473
# except if the deeper dir should be output first
1474
self.assertPathCompare("a/b/c", "d/g")
1475
# lexical betwen dirs of the same height
1476
self.assertPathCompare("a/z", "z/z")
1477
self.assertPathCompare("a/c/z", "a/d/e")
1479
def test_path_prefix_sorting(self):
1480
"""Doing a sort on path prefix should match our sample data."""
1495
dir_sorted_paths = [
1511
sorted(original_paths, key=osutils.path_prefix_key))
1512
# using the comparison routine shoudl work too:
1515
sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1518
class TestCopyTree(tests.TestCaseInTempDir):
1520
def test_copy_basic_tree(self):
1521
self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1522
osutils.copy_tree('source', 'target')
1523
self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
1524
self.assertEqual(['c'], os.listdir('target/b'))
1526
def test_copy_tree_target_exists(self):
1527
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_symlinks(self):
1534
self.requireFeature(tests.SymlinkFeature)
1535
self.build_tree(['source/'])
1536
os.symlink('a/generic/path', 'source/lnk')
1537
osutils.copy_tree('source', 'target')
1538
self.assertEqual(['lnk'], os.listdir('target'))
1539
self.assertEqual('a/generic/path', os.readlink('target/lnk'))
1541
def test_copy_tree_handlers(self):
1542
processed_files = []
1543
processed_links = []
1544
def file_handler(from_path, to_path):
1545
processed_files.append(('f', from_path, to_path))
1546
def dir_handler(from_path, to_path):
1547
processed_files.append(('d', from_path, to_path))
1548
def link_handler(from_path, to_path):
1549
processed_links.append((from_path, to_path))
1550
handlers = {'file':file_handler,
1551
'directory':dir_handler,
1552
'symlink':link_handler,
1555
self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1556
if osutils.has_symlinks():
1557
os.symlink('a/generic/path', 'source/lnk')
1558
osutils.copy_tree('source', 'target', handlers=handlers)
1560
self.assertEqual([('d', 'source', 'target'),
1561
('f', 'source/a', 'target/a'),
1562
('d', 'source/b', 'target/b'),
1563
('f', 'source/b/c', 'target/b/c'),
1565
self.failIfExists('target')
1566
if osutils.has_symlinks():
1567
self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1570
class TestSetUnsetEnv(tests.TestCase):
1571
"""Test updating the environment"""
1574
super(TestSetUnsetEnv, self).setUp()
1576
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
1577
'Environment was not cleaned up properly.'
1578
' Variable BZR_TEST_ENV_VAR should not exist.')
1580
if 'BZR_TEST_ENV_VAR' in os.environ:
1581
del os.environ['BZR_TEST_ENV_VAR']
1582
self.addCleanup(cleanup)
1585
"""Test that we can set an env variable"""
1586
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1587
self.assertEqual(None, old)
1588
self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
1590
def test_double_set(self):
1591
"""Test that we get the old value out"""
1592
osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1593
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
1594
self.assertEqual('foo', old)
1595
self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
1597
def test_unicode(self):
1598
"""Environment can only contain plain strings
1600
So Unicode strings must be encoded.
1602
uni_val, env_val = tests.probe_unicode_in_user_encoding()
1604
raise tests.TestSkipped(
1605
'Cannot find a unicode character that works in encoding %s'
1606
% (osutils.get_user_encoding(),))
1608
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1609
self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1611
def test_unset(self):
1612
"""Test that passing None will remove the env var"""
1613
osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1614
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1615
self.assertEqual('foo', old)
1616
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1617
self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1620
class TestSizeShaFile(tests.TestCaseInTempDir):
1622
def test_sha_empty(self):
1623
self.build_tree_contents([('foo', '')])
1624
expected_sha = osutils.sha_string('')
1626
self.addCleanup(f.close)
1627
size, sha = osutils.size_sha_file(f)
1628
self.assertEqual(0, size)
1629
self.assertEqual(expected_sha, sha)
1631
def test_sha_mixed_endings(self):
1632
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1633
self.build_tree_contents([('foo', text)])
1634
expected_sha = osutils.sha_string(text)
1635
f = open('foo', 'rb')
1636
self.addCleanup(f.close)
1637
size, sha = osutils.size_sha_file(f)
1638
self.assertEqual(38, size)
1639
self.assertEqual(expected_sha, sha)
1642
class TestShaFileByName(tests.TestCaseInTempDir):
1644
def test_sha_empty(self):
1645
self.build_tree_contents([('foo', '')])
1646
expected_sha = osutils.sha_string('')
1647
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1649
def test_sha_mixed_endings(self):
1650
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1651
self.build_tree_contents([('foo', text)])
1652
expected_sha = osutils.sha_string(text)
1653
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1656
class TestResourceLoading(tests.TestCaseInTempDir):
1658
def test_resource_string(self):
1659
# test resource in bzrlib
1660
text = osutils.resource_string('bzrlib', 'debug.py')
1661
self.assertContainsRe(text, "debug_flags = set()")
1662
# test resource under bzrlib
1663
text = osutils.resource_string('bzrlib.ui', 'text.py')
1664
self.assertContainsRe(text, "class TextUIFactory")
1665
# test unsupported package
1666
self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1668
# test unknown resource
1669
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1672
class TestReCompile(tests.TestCase):
1674
def test_re_compile_checked(self):
1675
r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1676
self.assertTrue(r.match('aaaa'))
1677
self.assertTrue(r.match('aAaA'))
1679
def test_re_compile_checked_error(self):
1680
# like https://bugs.launchpad.net/bzr/+bug/251352
1681
err = self.assertRaises(
1682
errors.BzrCommandError,
1683
osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1685
"Invalid regular expression in test case: '*': "
1686
"nothing to repeat",
1690
class TestDirReader(tests.TestCaseInTempDir):
1693
_dir_reader_class = None
1694
_native_to_unicode = None
1697
tests.TestCaseInTempDir.setUp(self)
1698
self.overrideAttr(osutils,
1699
'_selected_dir_reader', self._dir_reader_class())
1701
def _get_ascii_tree(self):
1709
expected_dirblocks = [
1711
[('0file', '0file', 'file'),
1712
('1dir', '1dir', 'directory'),
1713
('2file', '2file', 'file'),
1716
(('1dir', './1dir'),
1717
[('1dir/0file', '0file', 'file'),
1718
('1dir/1dir', '1dir', 'directory'),
1721
(('1dir/1dir', './1dir/1dir'),
1726
return tree, expected_dirblocks
1728
def test_walk_cur_dir(self):
1729
tree, expected_dirblocks = self._get_ascii_tree()
1730
self.build_tree(tree)
1731
result = list(osutils._walkdirs_utf8('.'))
1732
# Filter out stat and abspath
1733
self.assertEqual(expected_dirblocks,
1734
[(dirinfo, [line[0:3] for line in block])
1735
for dirinfo, block in result])
1737
def test_walk_sub_dir(self):
1738
tree, expected_dirblocks = self._get_ascii_tree()
1739
self.build_tree(tree)
1740
# you can search a subdir only, with a supplied prefix.
1741
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1742
# Filter out stat and abspath
1743
self.assertEqual(expected_dirblocks[1:],
1744
[(dirinfo, [line[0:3] for line in block])
1745
for dirinfo, block in result])
1747
def _get_unicode_tree(self):
1748
name0u = u'0file-\xb6'
1749
name1u = u'1dir-\u062c\u0648'
1750
name2u = u'2file-\u0633'
1754
name1u + '/' + name0u,
1755
name1u + '/' + name1u + '/',
1758
name0 = name0u.encode('UTF-8')
1759
name1 = name1u.encode('UTF-8')
1760
name2 = name2u.encode('UTF-8')
1761
expected_dirblocks = [
1763
[(name0, name0, 'file', './' + name0u),
1764
(name1, name1, 'directory', './' + name1u),
1765
(name2, name2, 'file', './' + name2u),
1768
((name1, './' + name1u),
1769
[(name1 + '/' + name0, name0, 'file', './' + name1u
1771
(name1 + '/' + name1, name1, 'directory', './' + name1u
1775
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1780
return tree, expected_dirblocks
1782
def _filter_out(self, raw_dirblocks):
1783
"""Filter out a walkdirs_utf8 result.
1785
stat field is removed, all native paths are converted to unicode
1787
filtered_dirblocks = []
1788
for dirinfo, block in raw_dirblocks:
1789
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1792
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1793
filtered_dirblocks.append((dirinfo, details))
1794
return filtered_dirblocks
1796
def test_walk_unicode_tree(self):
1797
self.requireFeature(tests.UnicodeFilenameFeature)
1798
tree, expected_dirblocks = self._get_unicode_tree()
1799
self.build_tree(tree)
1800
result = list(osutils._walkdirs_utf8('.'))
1801
self.assertEqual(expected_dirblocks, self._filter_out(result))
1803
def test_symlink(self):
1804
self.requireFeature(tests.SymlinkFeature)
1805
self.requireFeature(tests.UnicodeFilenameFeature)
1806
target = u'target\N{Euro Sign}'
1807
link_name = u'l\N{Euro Sign}nk'
1808
os.symlink(target, link_name)
1809
target_utf8 = target.encode('UTF-8')
1810
link_name_utf8 = link_name.encode('UTF-8')
1811
expected_dirblocks = [
1813
[(link_name_utf8, link_name_utf8,
1814
'symlink', './' + link_name),],
1816
result = list(osutils._walkdirs_utf8('.'))
1817
self.assertEqual(expected_dirblocks, self._filter_out(result))
1820
class TestReadLink(tests.TestCaseInTempDir):
1821
"""Exposes os.readlink() problems and the osutils solution.
1823
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1824
unicode string will be returned if a unicode string is passed.
1826
But prior python versions failed to properly encode the passed unicode
1829
_test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1832
super(tests.TestCaseInTempDir, self).setUp()
1833
self.link = u'l\N{Euro Sign}ink'
1834
self.target = u'targe\N{Euro Sign}t'
1835
os.symlink(self.target, self.link)
1837
def test_os_readlink_link_encoding(self):
1838
if sys.version_info < (2, 6):
1839
self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1841
self.assertEquals(self.target, os.readlink(self.link))
1843
def test_os_readlink_link_decoding(self):
1844
self.assertEquals(self.target.encode(osutils._fs_enc),
1845
os.readlink(self.link.encode(osutils._fs_enc)))
1848
class TestConcurrency(tests.TestCase):
1851
super(TestConcurrency, self).setUp()
1852
self.overrideAttr(osutils, '_cached_local_concurrency')
1854
def test_local_concurrency(self):
1855
concurrency = osutils.local_concurrency()
1856
self.assertIsInstance(concurrency, int)
1858
def test_local_concurrency_environment_variable(self):
1859
os.environ['BZR_CONCURRENCY'] = '2'
1860
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1861
os.environ['BZR_CONCURRENCY'] = '3'
1862
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1863
os.environ['BZR_CONCURRENCY'] = 'foo'
1864
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1866
def test_option_concurrency(self):
1867
os.environ['BZR_CONCURRENCY'] = '1'
1868
self.run_bzr('rocks --concurrency 42')
1869
# Command line overrides envrionment variable
1870
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1871
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1874
class TestFailedToLoadExtension(tests.TestCase):
1876
def _try_loading(self):
1878
import bzrlib._fictional_extension_py
1879
except ImportError, e:
1880
osutils.failed_to_load_extension(e)
1884
super(TestFailedToLoadExtension, self).setUp()
1885
self.overrideAttr(osutils, '_extension_load_failures', [])
1887
def test_failure_to_load(self):
1889
self.assertLength(1, osutils._extension_load_failures)
1890
self.assertEquals(osutils._extension_load_failures[0],
1891
"No module named _fictional_extension_py")
1893
def test_report_extension_load_failures_no_warning(self):
1894
self.assertTrue(self._try_loading())
1895
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
1896
# it used to give a Python warning; it no longer does
1897
self.assertLength(0, warnings)
1899
def test_report_extension_load_failures_message(self):
1901
trace.push_log_file(log)
1902
self.assertTrue(self._try_loading())
1903
osutils.report_extension_load_failures()
1904
self.assertContainsRe(
1906
r"bzr: warning: some compiled extensions could not be loaded; "
1907
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1911
class TestTerminalWidth(tests.TestCase):
1913
def replace_stdout(self, new):
1914
self.overrideAttr(sys, 'stdout', new)
1916
def replace__terminal_size(self, new):
1917
self.overrideAttr(osutils, '_terminal_size', new)
1919
def set_fake_tty(self):
1921
class I_am_a_tty(object):
1925
self.replace_stdout(I_am_a_tty())
1927
def test_default_values(self):
1928
self.assertEqual(80, osutils.default_terminal_width)
1930
def test_defaults_to_BZR_COLUMNS(self):
1931
# BZR_COLUMNS is set by the test framework
1932
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
1933
os.environ['BZR_COLUMNS'] = '12'
1934
self.assertEqual(12, osutils.terminal_width())
1936
def test_falls_back_to_COLUMNS(self):
1937
del os.environ['BZR_COLUMNS']
1938
self.assertNotEqual('42', os.environ['COLUMNS'])
1940
os.environ['COLUMNS'] = '42'
1941
self.assertEqual(42, osutils.terminal_width())
1943
def test_tty_default_without_columns(self):
1944
del os.environ['BZR_COLUMNS']
1945
del os.environ['COLUMNS']
1947
def terminal_size(w, h):
1951
# We need to override the osutils definition as it depends on the
1952
# running environment that we can't control (PQM running without a
1953
# controlling terminal is one example).
1954
self.replace__terminal_size(terminal_size)
1955
self.assertEqual(42, osutils.terminal_width())
1957
def test_non_tty_default_without_columns(self):
1958
del os.environ['BZR_COLUMNS']
1959
del os.environ['COLUMNS']
1960
self.replace_stdout(None)
1961
self.assertEqual(None, osutils.terminal_width())
1963
def test_no_TIOCGWINSZ(self):
1964
self.requireFeature(term_ios_feature)
1965
termios = term_ios_feature.module
1966
# bug 63539 is about a termios without TIOCGWINSZ attribute
1968
orig = termios.TIOCGWINSZ
1969
except AttributeError:
1970
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
1973
self.overrideAttr(termios, 'TIOCGWINSZ')
1974
del termios.TIOCGWINSZ
1975
del os.environ['BZR_COLUMNS']
1976
del os.environ['COLUMNS']
1977
# Whatever the result is, if we don't raise an exception, it's ok.
1978
osutils.terminal_width()
1980
class TestCreationOps(tests.TestCaseInTempDir):
1981
_test_needs_features = [features.chown_feature]
1984
tests.TestCaseInTempDir.setUp(self)
1985
self.overrideAttr(os, 'chown', self._dummy_chown)
1987
# params set by call to _dummy_chown
1988
self.path = self.uid = self.gid = None
1990
def _dummy_chown(self, path, uid, gid):
1991
self.path, self.uid, self.gid = path, uid, gid
1993
def test_mkdir_with_ownership_chown(self):
1994
"""Ensure that osutils.mkdir_with_ownership chowns correctly with ownership_src.
1997
osutils.mkdir_with_ownership('foo', ownsrc)
2000
self.assertEquals(self.path, 'foo')
2001
self.assertEquals(self.uid, s.st_uid)
2002
self.assertEquals(self.gid, s.st_gid)
2004
def test_open_with_ownership_chown(self):
2005
"""Ensure that osutils.open_with_ownership chowns correctly with ownership_src.
2008
f = osutils.open_with_ownership('foo', 'w', ownership_src=ownsrc)
2010
# do a test write and close
2015
self.assertEquals(self.path, 'foo')
2016
self.assertEquals(self.uid, s.st_uid)
2017
self.assertEquals(self.gid, s.st_gid)
added
removed
bzrlib/tests/test_osutils.py
