/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Robert Collins
  • Date: 2010-05-06 11:08:10 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506110810-h3j07fh5gmw54s25
Cleaner matcher matching revised unlocking protocol.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
1
 
# Copyright (C) 2005, 2006 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
 
19
from cStringIO import StringIO
19
20
import errno
20
21
import os
 
22
import re
21
23
import socket
22
24
import stat
23
25
import sys
 
26
import time
24
27
 
25
 
import bzrlib
26
28
from bzrlib import (
27
29
    errors,
28
30
    osutils,
 
31
    tests,
 
32
    trace,
29
33
    win32utils,
30
34
    )
31
 
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
32
35
from bzrlib.tests import (
33
 
        StringIOWrapper,
34
 
        TestCase, 
35
 
        TestCaseInTempDir, 
36
 
        TestSkipped,
37
 
        )
38
 
 
39
 
 
40
 
class TestOSUtils(TestCaseInTempDir):
 
36
    features,
 
37
    file_utils,
 
38
    test__walkdirs_win32,
 
39
    )
 
40
 
 
41
 
 
42
class _UTF8DirReaderFeature(tests.Feature):
 
43
 
 
44
    def _probe(self):
 
45
        try:
 
46
            from bzrlib import _readdir_pyx
 
47
            self.reader = _readdir_pyx.UTF8DirReader
 
48
            return True
 
49
        except ImportError:
 
50
            return False
 
51
 
 
52
    def feature_name(self):
 
53
        return 'bzrlib._readdir_pyx'
 
54
 
 
55
UTF8DirReaderFeature = _UTF8DirReaderFeature()
 
56
 
 
57
term_ios_feature = tests.ModuleAvailableFeature('termios')
 
58
 
 
59
 
 
60
def _already_unicode(s):
 
61
    return s
 
62
 
 
63
 
 
64
def _utf8_to_unicode(s):
 
65
    return s.decode('UTF-8')
 
66
 
 
67
 
 
68
def dir_reader_scenarios():
 
69
    # For each dir reader we define:
 
70
 
 
71
    # - native_to_unicode: a function converting the native_abspath as returned
 
72
    #   by DirReader.read_dir to its unicode representation
 
73
 
 
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
 
79
    # available.
 
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)))
 
85
 
 
86
    if test__walkdirs_win32.win32_readdir_feature.available():
 
87
        try:
 
88
            from bzrlib import _walkdirs_win32
 
89
            scenarios.append(
 
90
                ('win32',
 
91
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
 
92
                      _native_to_unicode=_already_unicode)))
 
93
        except ImportError:
 
94
            pass
 
95
    return scenarios
 
96
 
 
97
 
 
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)
 
104
    return suite
 
105
 
 
106
 
 
107
class TestContainsWhitespace(tests.TestCase):
41
108
 
42
109
    def test_contains_whitespace(self):
43
110
        self.failUnless(osutils.contains_whitespace(u' '))
53
120
        self.failIf(osutils.contains_whitespace(u'hellothere'))
54
121
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
55
122
 
 
123
 
 
124
class TestRename(tests.TestCaseInTempDir):
 
125
 
 
126
    def create_file(self, filename, content):
 
127
        f = open(filename, 'wb')
 
128
        try:
 
129
            f.write(content)
 
130
        finally:
 
131
            f.close()
 
132
 
 
133
    def _fancy_rename(self, a, b):
 
134
        osutils.fancy_rename(a, b, rename_func=os.rename,
 
135
                             unlink_func=os.unlink)
 
136
 
56
137
    def test_fancy_rename(self):
57
138
        # This should work everywhere
58
 
        def rename(a, b):
59
 
            osutils.fancy_rename(a, b,
60
 
                    rename_func=os.rename,
61
 
                    unlink_func=os.unlink)
62
 
 
63
 
        open('a', 'wb').write('something in a\n')
64
 
        rename('a', 'b')
 
139
        self.create_file('a', 'something in a\n')
 
140
        self._fancy_rename('a', 'b')
65
141
        self.failIfExists('a')
66
142
        self.failUnlessExists('b')
67
143
        self.check_file_contents('b', 'something in a\n')
68
144
 
69
 
        open('a', 'wb').write('new something in a\n')
70
 
        rename('b', 'a')
 
145
        self.create_file('a', 'new something in a\n')
 
146
        self._fancy_rename('b', 'a')
71
147
 
72
148
        self.check_file_contents('a', 'something in a\n')
73
149
 
 
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')
 
157
 
 
158
    def test_fancy_rename_fails_if_source_and_target_missing(self):
 
159
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
160
                          'missingsource', 'missingtarget')
 
161
 
74
162
    def test_rename(self):
75
163
        # Rename should be semi-atomic on all platforms
76
 
        open('a', 'wb').write('something in a\n')
 
164
        self.create_file('a', 'something in a\n')
77
165
        osutils.rename('a', 'b')
78
166
        self.failIfExists('a')
79
167
        self.failUnlessExists('b')
80
168
        self.check_file_contents('b', 'something in a\n')
81
169
 
82
 
        open('a', 'wb').write('new something in a\n')
 
170
        self.create_file('a', 'new something in a\n')
83
171
        osutils.rename('b', 'a')
84
172
 
85
173
        self.check_file_contents('a', 'something in a\n')
86
174
 
87
175
    # TODO: test fancy_rename using a MemoryTransport
88
176
 
 
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)
 
186
 
 
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')
 
193
 
 
194
 
 
195
class TestRandChars(tests.TestCase):
 
196
 
89
197
    def test_01_rand_chars_empty(self):
90
198
        result = osutils.rand_chars(0)
91
199
        self.assertEqual(result, '')
96
204
        self.assertEqual(type(result), str)
97
205
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
98
206
 
 
207
 
 
208
class TestIsInside(tests.TestCase):
 
209
 
99
210
    def test_is_inside(self):
100
211
        is_inside = osutils.is_inside
101
212
        self.assertTrue(is_inside('src', 'src/foo.c'))
105
216
        self.assertFalse(is_inside('foo.c', ''))
106
217
        self.assertTrue(is_inside('', 'foo.c'))
107
218
 
 
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),
 
223
                         (['src'], 'src'),
 
224
                         ]:
 
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))
 
229
 
 
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'),
 
235
                         (['src'], 'src'),
 
236
                         ]:
 
237
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
238
 
 
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))
 
243
 
 
244
 
 
245
class TestRmTree(tests.TestCaseInTempDir):
 
246
 
108
247
    def test_rmtree(self):
109
248
        # Check to remove tree with read-only files/dirs
110
249
        os.mkdir('dir')
123
262
        self.failIfExists('dir/file')
124
263
        self.failIfExists('dir')
125
264
 
 
265
 
 
266
class TestDeleteAny(tests.TestCaseInTempDir):
 
267
 
 
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')
 
273
 
 
274
        osutils.delete_any('f')
 
275
        osutils.delete_any('d')
 
276
 
 
277
 
 
278
class TestKind(tests.TestCaseInTempDir):
 
279
 
126
280
    def test_file_kind(self):
127
281
        self.build_tree(['file', 'dir/'])
128
282
        self.assertEquals('file', osutils.file_kind('file'))
130
284
        if osutils.has_symlinks():
131
285
            os.symlink('symlink', 'symlink')
132
286
            self.assertEquals('symlink', osutils.file_kind('symlink'))
133
 
        
 
287
 
134
288
        # TODO: jam 20060529 Test a block device
135
289
        try:
136
290
            os.lstat('/dev/null')
158
312
                os.remove('socket')
159
313
 
160
314
    def test_kind_marker(self):
161
 
        self.assertEqual(osutils.kind_marker('file'), '')
162
 
        self.assertEqual(osutils.kind_marker('directory'), '/')
163
 
        self.assertEqual(osutils.kind_marker('symlink'), '@')
164
 
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
 
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"))
 
323
 
 
324
 
 
325
class TestUmask(tests.TestCaseInTempDir):
165
326
 
166
327
    def test_get_umask(self):
167
328
        if sys.platform == 'win32':
170
331
            return
171
332
 
172
333
        orig_umask = osutils.get_umask()
173
 
        try:
174
 
            os.umask(0222)
175
 
            self.assertEqual(0222, osutils.get_umask())
176
 
            os.umask(0022)
177
 
            self.assertEqual(0022, osutils.get_umask())
178
 
            os.umask(0002)
179
 
            self.assertEqual(0002, osutils.get_umask())
180
 
            os.umask(0027)
181
 
            self.assertEqual(0027, osutils.get_umask())
182
 
        finally:
183
 
            os.umask(orig_umask)
 
334
        self.addCleanup(os.umask, orig_umask)
 
335
        os.umask(0222)
 
336
        self.assertEqual(0222, osutils.get_umask())
 
337
        os.umask(0022)
 
338
        self.assertEqual(0022, osutils.get_umask())
 
339
        os.umask(0002)
 
340
        self.assertEqual(0002, osutils.get_umask())
 
341
        os.umask(0027)
 
342
        self.assertEqual(0027, osutils.get_umask())
 
343
 
 
344
 
 
345
class TestDateTime(tests.TestCase):
184
346
 
185
347
    def assertFormatedDelta(self, expected, seconds):
186
348
        """Assert osutils.format_delta formats as expected"""
218
380
        self.assertFormatedDelta('1 second in the future', -1)
219
381
        self.assertFormatedDelta('2 seconds in the future', -2)
220
382
 
 
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.
 
392
 
 
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))
 
400
 
 
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
 
406
        # either direction.
 
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
 
409
        # probably wrong.
 
410
        eighteen_hours = 18 * 3600
 
411
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
412
 
 
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)
 
419
 
 
420
 
 
421
class TestLinks(tests.TestCaseInTempDir):
 
422
 
221
423
    def test_dereference_path(self):
222
 
        if not osutils.has_symlinks():
223
 
            raise TestSkipped('Symlinks are not supported on this platform')
 
424
        self.requireFeature(tests.SymlinkFeature)
224
425
        cwd = osutils.realpath('.')
225
426
        os.mkdir('bar')
226
427
        bar_path = osutils.pathjoin(cwd, 'bar')
229
430
        self.assertEqual(bar_path, osutils.realpath('./bar'))
230
431
        os.symlink('bar', 'foo')
231
432
        self.assertEqual(bar_path, osutils.realpath('./foo'))
232
 
        
 
433
 
233
434
        # Does not dereference terminal symlinks
234
435
        foo_path = osutils.pathjoin(cwd, 'foo')
235
436
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
246
447
        foo_baz_path = osutils.pathjoin(foo_path, 'baz')
247
448
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
248
449
 
249
 
 
250
 
    def test_kind_marker(self):
251
 
        self.assertEqual("", osutils.kind_marker("file"))
252
 
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
253
 
        self.assertEqual("@", osutils.kind_marker("symlink"))
254
 
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
255
 
 
256
 
 
257
 
class TestSafeUnicode(TestCase):
 
450
    def test_changing_access(self):
 
451
        f = file('file', 'w')
 
452
        f.write('monkey')
 
453
        f.close()
 
454
 
 
455
        # Make a file readonly
 
456
        osutils.make_readonly('file')
 
457
        mode = os.lstat('file').st_mode
 
458
        self.assertEqual(mode, mode & 0777555)
 
459
 
 
460
        # Make a file writable
 
461
        osutils.make_writable('file')
 
462
        mode = os.lstat('file').st_mode
 
463
        self.assertEqual(mode, mode | 0200)
 
464
 
 
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')
 
470
 
 
471
    def test_host_os_dereferences_symlinks(self):
 
472
        osutils.host_os_dereferences_symlinks()
 
473
 
 
474
 
 
475
class TestCanonicalRelPath(tests.TestCaseInTempDir):
 
476
 
 
477
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
 
478
 
 
479
    def test_canonical_relpath_simple(self):
 
480
        f = file('MixedCaseName', 'w')
 
481
        f.close()
 
482
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
 
483
        self.failUnlessEqual('work/MixedCaseName', actual)
 
484
 
 
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)
 
490
 
 
491
 
 
492
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
 
493
 
 
494
    def assertRelpath(self, expected, base, path):
 
495
        actual = osutils._cicp_canonical_relpath(base, path)
 
496
        self.assertEqual(expected, actual)
 
497
 
 
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')
 
502
 
 
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')
 
512
 
 
513
    def test_at_root_slash(self):
 
514
        # We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
 
515
        # check...
 
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')
 
520
 
 
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'
 
525
                                          ' letters.')
 
526
        # see bug #322807
 
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')
 
533
 
 
534
 
 
535
class TestPumpFile(tests.TestCase):
 
536
    """Test pumpfile method."""
 
537
 
 
538
    def setUp(self):
 
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)
 
545
 
 
546
    def test_bracket_block_size(self):
 
547
        """Read data in blocks with the requested read size bracketing the
 
548
        block size."""
 
549
        # make sure test data is larger than max read size
 
550
        self.assertTrue(self.test_data_len > self.block_size)
 
551
 
 
552
        from_file = file_utils.FakeReadFile(self.test_data)
 
553
        to_file = StringIO()
 
554
 
 
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)
 
560
 
 
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)
 
567
 
 
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)
 
574
 
 
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)
 
578
 
 
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))
 
585
 
 
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)
 
591
 
 
592
        # retrieve data in blocks
 
593
        from_file = file_utils.FakeReadFile(self.test_data)
 
594
        to_file = StringIO()
 
595
        osutils.pumpfile(from_file, to_file, self.test_data_len,
 
596
                         self.block_size)
 
597
 
 
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)
 
602
 
 
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))
 
609
 
 
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)
 
615
 
 
616
        # retrieve data to EOF
 
617
        from_file = file_utils.FakeReadFile(self.test_data)
 
618
        to_file = StringIO()
 
619
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
 
620
 
 
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)
 
624
 
 
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))
 
631
 
 
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)
 
638
        to_file = StringIO()
 
639
        osutils.pumpfile(from_file, to_file)
 
640
 
 
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))
 
647
 
 
648
    def test_report_activity(self):
 
649
        activity = []
 
650
        def log_activity(length, direction):
 
651
            activity.append((length, direction))
 
652
        from_file = StringIO(self.test_data)
 
653
        to_file = StringIO()
 
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)
 
658
 
 
659
        from_file = StringIO(self.test_data)
 
660
        to_file = StringIO()
 
661
        del activity[:]
 
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)
 
666
 
 
667
        # And with a limited amount of data
 
668
        from_file = StringIO(self.test_data)
 
669
        to_file = StringIO()
 
670
        del activity[:]
 
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)
 
674
 
 
675
 
 
676
 
 
677
class TestPumpStringFile(tests.TestCase):
 
678
 
 
679
    def test_empty(self):
 
680
        output = StringIO()
 
681
        osutils.pump_string_file("", output)
 
682
        self.assertEqual("", output.getvalue())
 
683
 
 
684
    def test_more_than_segment_size(self):
 
685
        output = StringIO()
 
686
        osutils.pump_string_file("123456789", output, 2)
 
687
        self.assertEqual("123456789", output.getvalue())
 
688
 
 
689
    def test_segment_size(self):
 
690
        output = StringIO()
 
691
        osutils.pump_string_file("12", output, 2)
 
692
        self.assertEqual("12", output.getvalue())
 
693
 
 
694
    def test_segment_size_multiple(self):
 
695
        output = StringIO()
 
696
        osutils.pump_string_file("1234", output, 2)
 
697
        self.assertEqual("1234", output.getvalue())
 
698
 
 
699
 
 
700
class TestRelpath(tests.TestCase):
 
701
 
 
702
    def test_simple_relpath(self):
 
703
        cwd = osutils.getcwd()
 
704
        subdir = cwd + '/subdir'
 
705
        self.assertEqual('subdir', osutils.relpath(cwd, subdir))
 
706
 
 
707
    def test_deep_relpath(self):
 
708
        cwd = osutils.getcwd()
 
709
        subdir = cwd + '/sub/subsubdir'
 
710
        self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
 
711
 
 
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')
 
717
 
 
718
 
 
719
class TestSafeUnicode(tests.TestCase):
258
720
 
259
721
    def test_from_ascii_string(self):
260
722
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
269
731
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
270
732
 
271
733
    def test_bad_utf8_string(self):
272
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
734
        self.assertRaises(errors.BzrBadParameterNotUnicode,
273
735
                          osutils.safe_unicode,
274
736
                          '\xbb\xbb')
275
737
 
276
738
 
277
 
class TestSafeUtf8(TestCase):
 
739
class TestSafeUtf8(tests.TestCase):
278
740
 
279
741
    def test_from_ascii_string(self):
280
742
        f = 'foobar'
290
752
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
291
753
 
292
754
    def test_bad_utf8_string(self):
293
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
755
        self.assertRaises(errors.BzrBadParameterNotUnicode,
294
756
                          osutils.safe_utf8, '\xbb\xbb')
295
757
 
296
758
 
297
 
class TestSafeRevisionId(TestCase):
 
759
class TestSafeRevisionId(tests.TestCase):
298
760
 
299
761
    def test_from_ascii_string(self):
 
762
        # this shouldn't give a warning because it's getting an ascii string
300
763
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
301
764
 
302
765
    def test_from_unicode_string_ascii_contents(self):
321
784
        self.assertEqual(None, osutils.safe_revision_id(None))
322
785
 
323
786
 
324
 
class TestSafeFileId(TestCase):
 
787
class TestSafeFileId(tests.TestCase):
325
788
 
326
789
    def test_from_ascii_string(self):
327
790
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
347
810
        self.assertEqual(None, osutils.safe_file_id(None))
348
811
 
349
812
 
350
 
class TestWin32Funcs(TestCase):
351
 
    """Test that the _win32 versions of os utilities return appropriate paths."""
 
813
class TestWin32Funcs(tests.TestCase):
 
814
    """Test that _win32 versions of os utilities return appropriate paths."""
352
815
 
353
816
    def test_abspath(self):
354
817
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
361
824
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
362
825
 
363
826
    def test_pathjoin(self):
364
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
365
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
366
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
367
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
368
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
369
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
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'))
370
839
 
371
840
    def test_normpath(self):
372
 
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
373
 
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
 
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'))
374
845
 
375
846
    def test_getcwd(self):
376
847
        cwd = osutils._win32_getcwd()
405
876
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
406
877
 
407
878
 
408
 
class TestWin32FuncsDirs(TestCaseInTempDir):
 
879
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
409
880
    """Test win32 functions that create files."""
410
 
    
 
881
 
411
882
    def test_getcwd(self):
412
 
        if win32utils.winver == 'Windows 98':
413
 
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
414
 
        # Make sure getcwd can handle unicode filenames
415
 
        try:
416
 
            os.mkdir(u'mu-\xb5')
417
 
        except UnicodeError:
418
 
            raise TestSkipped("Unable to create Unicode filename")
419
 
 
 
883
        self.requireFeature(tests.UnicodeFilenameFeature)
 
884
        os.mkdir(u'mu-\xb5')
420
885
        os.chdir(u'mu-\xb5')
421
886
        # TODO: jam 20060427 This will probably fail on Mac OSX because
422
887
        #       it will change the normalization of B\xe5gfors
424
889
        #       osutils.getcwd() renormalize the path.
425
890
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
426
891
 
 
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']))
 
905
 
427
906
    def test_mkdtemp(self):
428
907
        tmpdir = osutils._win32_mkdtemp(dir='.')
429
908
        self.assertFalse('\\' in tmpdir)
484
963
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
485
964
 
486
965
 
487
 
class TestMacFuncsDirs(TestCaseInTempDir):
 
966
class TestParentDirectories(tests.TestCaseInTempDir):
 
967
    """Test osutils.parent_directories()"""
 
968
 
 
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'))
 
973
 
 
974
 
 
975
class TestMacFuncsDirs(tests.TestCaseInTempDir):
488
976
    """Test mac special functions that require directories."""
489
977
 
490
978
    def test_getcwd(self):
491
 
        # On Mac, this will actually create Ba\u030agfors
492
 
        # but chdir will still work, because it accepts both paths
493
 
        try:
494
 
            os.mkdir(u'B\xe5gfors')
495
 
        except UnicodeError:
496
 
            raise TestSkipped("Unable to create Unicode filename")
497
 
 
 
979
        self.requireFeature(tests.UnicodeFilenameFeature)
 
980
        os.mkdir(u'B\xe5gfors')
498
981
        os.chdir(u'B\xe5gfors')
499
982
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
500
983
 
501
984
    def test_getcwd_nonnorm(self):
 
985
        self.requireFeature(tests.UnicodeFilenameFeature)
502
986
        # Test that _mac_getcwd() will normalize this path
503
 
        try:
504
 
            os.mkdir(u'Ba\u030agfors')
505
 
        except UnicodeError:
506
 
            raise TestSkipped("Unable to create Unicode filename")
507
 
 
 
987
        os.mkdir(u'Ba\u030agfors')
508
988
        os.chdir(u'Ba\u030agfors')
509
989
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
510
990
 
511
991
 
512
 
class TestSplitLines(TestCase):
 
992
class TestChunksToLines(tests.TestCase):
 
993
 
 
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']))
 
999
 
 
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
 
1004
        else:
 
1005
            from bzrlib._chunks_to_lines_py import chunks_to_lines
 
1006
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
 
1007
 
 
1008
 
 
1009
class TestSplitLines(tests.TestCase):
513
1010
 
514
1011
    def test_split_unicode(self):
515
1012
        self.assertEqual([u'foo\n', u'bar\xae'],
522
1019
                         osutils.split_lines('foo\rbar\n'))
523
1020
 
524
1021
 
525
 
class TestWalkDirs(TestCaseInTempDir):
 
1022
class TestWalkDirs(tests.TestCaseInTempDir):
 
1023
 
 
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])
526
1028
 
527
1029
    def test_walkdirs(self):
528
1030
        tree = [
561
1063
            result.append((dirdetail, dirblock))
562
1064
 
563
1065
        self.assertTrue(found_bzrdir)
564
 
        self.assertEqual(expected_dirblocks,
565
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1066
        self.assertExpectedBlocks(expected_dirblocks, result)
566
1067
        # you can search a subdir only, with a supplied prefix.
567
1068
        result = []
568
1069
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
569
1070
            result.append(dirblock)
570
 
        self.assertEqual(expected_dirblocks[1:],
571
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1071
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
1072
 
 
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
 
1086
        # doesn't.)
 
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")
572
1092
 
573
1093
    def test__walkdirs_utf8(self):
574
1094
        tree = [
607
1127
            result.append((dirdetail, dirblock))
608
1128
 
609
1129
        self.assertTrue(found_bzrdir)
610
 
        self.assertEqual(expected_dirblocks,
611
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1130
        self.assertExpectedBlocks(expected_dirblocks, result)
 
1131
 
612
1132
        # you can search a subdir only, with a supplied prefix.
613
1133
        result = []
614
1134
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
615
1135
            result.append(dirblock)
616
 
        self.assertEqual(expected_dirblocks[1:],
617
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1136
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
618
1137
 
619
1138
    def _filter_out_stat(self, result):
620
1139
        """Filter out the stat value from the walkdirs result"""
625
1144
                new_dirblock.append((info[0], info[1], info[2], info[4]))
626
1145
            dirblock[:] = new_dirblock
627
1146
 
 
1147
    def _save_platform_info(self):
 
1148
        self.overrideAttr(win32utils, 'winver')
 
1149
        self.overrideAttr(osutils, '_fs_enc')
 
1150
        self.overrideAttr(osutils, '_selected_dir_reader')
 
1151
 
 
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)
 
1159
 
 
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)
 
1166
 
 
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)
 
1173
 
 
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)
 
1180
 
 
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)
 
1186
 
 
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)
 
1194
 
 
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)
 
1200
 
628
1201
    def test_unicode_walkdirs(self):
629
1202
        """Walkdirs should always return unicode paths."""
 
1203
        self.requireFeature(tests.UnicodeFilenameFeature)
630
1204
        name0 = u'0file-\xb6'
631
1205
        name1 = u'1dir-\u062c\u0648'
632
1206
        name2 = u'2file-\u0633'
637
1211
            name1 + '/' + name1 + '/',
638
1212
            name2,
639
1213
            ]
640
 
        try:
641
 
            self.build_tree(tree)
642
 
        except UnicodeError:
643
 
            raise TestSkipped('Could not represent Unicode chars'
644
 
                              ' in current encoding.')
 
1214
        self.build_tree(tree)
645
1215
        expected_dirblocks = [
646
1216
                ((u'', u'.'),
647
1217
                 [(name0, name0, 'file', './' + name0),
673
1243
 
674
1244
        The abspath portion might be in unicode or utf-8
675
1245
        """
 
1246
        self.requireFeature(tests.UnicodeFilenameFeature)
676
1247
        name0 = u'0file-\xb6'
677
1248
        name1 = u'1dir-\u062c\u0648'
678
1249
        name2 = u'2file-\u0633'
683
1254
            name1 + '/' + name1 + '/',
684
1255
            name2,
685
1256
            ]
686
 
        try:
687
 
            self.build_tree(tree)
688
 
        except UnicodeError:
689
 
            raise TestSkipped('Could not represent Unicode chars'
690
 
                              ' in current encoding.')
 
1257
        self.build_tree(tree)
691
1258
        name0 = name0.encode('utf8')
692
1259
        name1 = name1.encode('utf8')
693
1260
        name2 = name2.encode('utf8')
732
1299
            result.append((dirdetail, new_dirblock))
733
1300
        self.assertEqual(expected_dirblocks, result)
734
1301
 
735
 
    def test_unicode__walkdirs_unicode_to_utf8(self):
736
 
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
 
1302
    def test__walkdirs_utf8_with_unicode_fs(self):
 
1303
        """UnicodeDirReader should be a safe fallback everywhere
737
1304
 
738
1305
        The abspath portion should be in unicode
739
1306
        """
740
 
        name0u = u'0file-\xb6'
741
 
        name1u = u'1dir-\u062c\u0648'
742
 
        name2u = u'2file-\u0633'
743
 
        tree = [
744
 
            name0u,
745
 
            name1u + '/',
746
 
            name1u + '/' + name0u,
747
 
            name1u + '/' + name1u + '/',
748
 
            name2u,
749
 
            ]
 
1307
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1308
        # Use the unicode reader. TODO: split into driver-and-driven unit
 
1309
        # tests.
 
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'
 
1315
        tree = [
 
1316
            name0u,
 
1317
            name1u + '/',
 
1318
            name1u + '/' + name0u,
 
1319
            name1u + '/' + name1u + '/',
 
1320
            name2u,
 
1321
            ]
 
1322
        self.build_tree(tree)
 
1323
        name0 = name0u.encode('utf8')
 
1324
        name1 = name1u.encode('utf8')
 
1325
        name2 = name2u.encode('utf8')
 
1326
 
 
1327
        # All of the abspaths should be in unicode, all of the relative paths
 
1328
        # should be in utf8
 
1329
        expected_dirblocks = [
 
1330
                (('', '.'),
 
1331
                 [(name0, name0, 'file', './' + name0u),
 
1332
                  (name1, name1, 'directory', './' + name1u),
 
1333
                  (name2, name2, 'file', './' + name2u),
 
1334
                 ]
 
1335
                ),
 
1336
                ((name1, './' + name1u),
 
1337
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1338
                                                        + '/' + name0u),
 
1339
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1340
                                                            + '/' + name1u),
 
1341
                 ]
 
1342
                ),
 
1343
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1344
                 [
 
1345
                 ]
 
1346
                ),
 
1347
            ]
 
1348
        result = list(osutils._walkdirs_utf8('.'))
 
1349
        self._filter_out_stat(result)
 
1350
        self.assertEqual(expected_dirblocks, result)
 
1351
 
 
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'
 
1361
        tree = [
 
1362
            name0u,
 
1363
            name1u + '/',
 
1364
            name1u + '/' + name0u,
 
1365
            name1u + '/' + name1u + '/',
 
1366
            name2u,
 
1367
            ]
 
1368
        self.build_tree(tree)
 
1369
        name0 = name0u.encode('utf8')
 
1370
        name1 = name1u.encode('utf8')
 
1371
        name2 = name2u.encode('utf8')
 
1372
 
 
1373
        # All of the abspaths should be in unicode, all of the relative paths
 
1374
        # should be in utf8
 
1375
        expected_dirblocks = [
 
1376
                (('', '.'),
 
1377
                 [(name0, name0, 'file', './' + name0u),
 
1378
                  (name1, name1, 'directory', './' + name1u),
 
1379
                  (name2, name2, 'file', './' + name2u),
 
1380
                 ]
 
1381
                ),
 
1382
                ((name1, './' + name1u),
 
1383
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1384
                                                        + '/' + name0u),
 
1385
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1386
                                                            + '/' + name1u),
 
1387
                 ]
 
1388
                ),
 
1389
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1390
                 [
 
1391
                 ]
 
1392
                ),
 
1393
            ]
 
1394
        result = list(osutils._walkdirs_utf8(u'.'))
 
1395
        self._filter_out_stat(result)
 
1396
        self.assertEqual(expected_dirblocks, result)
 
1397
 
 
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)
 
1407
 
 
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
 
1417
        # from the mtime
 
1418
        time.sleep(2)
 
1419
        f = open(name0u, 'ab')
750
1420
        try:
751
 
            self.build_tree(tree)
752
 
        except UnicodeError:
753
 
            raise TestSkipped('Could not represent Unicode chars'
754
 
                              ' in current encoding.')
 
1421
            f.write('just a small update')
 
1422
        finally:
 
1423
            f.close()
 
1424
 
 
1425
        result = Win32ReadDir().read_dir('', u'.')
 
1426
        entry = result[0]
 
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)
 
1431
 
 
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'
755
1438
        name0 = name0u.encode('utf8')
756
 
        name1 = name1u.encode('utf8')
757
 
        name2 = name2u.encode('utf8')
 
1439
        self.build_tree([name0u + '/'])
758
1440
 
759
 
        # All of the abspaths should be in unicode, all of the relative paths
760
 
        # should be in utf8
761
 
        expected_dirblocks = [
762
 
                (('', '.'),
763
 
                 [(name0, name0, 'file', './' + name0u),
764
 
                  (name1, name1, 'directory', './' + name1u),
765
 
                  (name2, name2, 'file', './' + name2u),
766
 
                 ]
767
 
                ),
768
 
                ((name1, './' + name1u),
769
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
770
 
                                                        + '/' + name0u),
771
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
772
 
                                                            + '/' + name1u),
773
 
                 ]
774
 
                ),
775
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
776
 
                 [
777
 
                 ]
778
 
                ),
779
 
            ]
780
 
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
781
 
        self._filter_out_stat(result)
782
 
        self.assertEqual(expected_dirblocks, result)
 
1441
        result = Win32ReadDir().read_dir('', u'.')
 
1442
        entry = result[0]
 
1443
        self.assertEqual((name0, name0, 'directory'), entry[:3])
 
1444
        self.assertEqual(u'./' + name0u, entry[4])
 
1445
        self.assertStatIsCorrect(entry[4], entry[3])
783
1446
 
784
1447
    def assertPathCompare(self, path_less, path_greater):
785
1448
        """check that path_less and path_greater compare correctly."""
859
1522
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
860
1523
 
861
1524
 
862
 
class TestCopyTree(TestCaseInTempDir):
863
 
    
 
1525
class TestCopyTree(tests.TestCaseInTempDir):
 
1526
 
864
1527
    def test_copy_basic_tree(self):
865
1528
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
866
1529
        osutils.copy_tree('source', 'target')
875
1538
        self.assertEqual(['c'], os.listdir('target/b'))
876
1539
 
877
1540
    def test_copy_tree_symlinks(self):
878
 
        if not osutils.has_symlinks():
879
 
            return
 
1541
        self.requireFeature(tests.SymlinkFeature)
880
1542
        self.build_tree(['source/'])
881
1543
        os.symlink('a/generic/path', 'source/lnk')
882
1544
        osutils.copy_tree('source', 'target')
912
1574
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
913
1575
 
914
1576
 
915
 
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
916
 
# [bialix] 2006/12/26
917
 
 
918
 
 
919
 
class TestSetUnsetEnv(TestCase):
 
1577
class TestSetUnsetEnv(tests.TestCase):
920
1578
    """Test updating the environment"""
921
1579
 
922
1580
    def setUp(self):
928
1586
        def cleanup():
929
1587
            if 'BZR_TEST_ENV_VAR' in os.environ:
930
1588
                del os.environ['BZR_TEST_ENV_VAR']
931
 
 
932
1589
        self.addCleanup(cleanup)
933
1590
 
934
1591
    def test_set(self):
946
1603
 
947
1604
    def test_unicode(self):
948
1605
        """Environment can only contain plain strings
949
 
        
 
1606
 
950
1607
        So Unicode strings must be encoded.
951
1608
        """
952
 
        # Try a few different characters, to see if we can get
953
 
        # one that will be valid in the user_encoding
954
 
        possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
955
 
        for uni_val in possible_vals:
956
 
            try:
957
 
                env_val = uni_val.encode(bzrlib.user_encoding)
958
 
            except UnicodeEncodeError:
959
 
                # Try a different character
960
 
                pass
961
 
            else:
962
 
                break
963
 
        else:
964
 
            raise TestSkipped('Cannot find a unicode character that works in'
965
 
                              ' encoding %s' % (bzrlib.user_encoding,))
 
1609
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
1610
        if uni_val is None:
 
1611
            raise tests.TestSkipped(
 
1612
                'Cannot find a unicode character that works in encoding %s'
 
1613
                % (osutils.get_user_encoding(),))
966
1614
 
967
1615
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
968
1616
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
976
1624
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
977
1625
 
978
1626
 
979
 
class TestLocalTimeOffset(TestCase):
980
 
 
981
 
    def test_local_time_offset(self):
982
 
        """Test that local_time_offset() returns a sane value."""
983
 
        offset = osutils.local_time_offset()
984
 
        self.assertTrue(isinstance(offset, int))
985
 
        # Test that the offset is no more than a eighteen hours in
986
 
        # either direction.
987
 
        # Time zone handling is system specific, so it is difficult to
988
 
        # do more specific tests, but a value outside of this range is
989
 
        # probably wrong.
990
 
        eighteen_hours = 18 * 3600
991
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
992
 
 
993
 
    def test_local_time_offset_with_timestamp(self):
994
 
        """Test that local_time_offset() works with a timestamp."""
995
 
        offset = osutils.local_time_offset(1000000000.1234567)
996
 
        self.assertTrue(isinstance(offset, int))
997
 
        eighteen_hours = 18 * 3600
998
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1627
class TestSizeShaFile(tests.TestCaseInTempDir):
 
1628
 
 
1629
    def test_sha_empty(self):
 
1630
        self.build_tree_contents([('foo', '')])
 
1631
        expected_sha = osutils.sha_string('')
 
1632
        f = open('foo')
 
1633
        self.addCleanup(f.close)
 
1634
        size, sha = osutils.size_sha_file(f)
 
1635
        self.assertEqual(0, size)
 
1636
        self.assertEqual(expected_sha, sha)
 
1637
 
 
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)
 
1647
 
 
1648
 
 
1649
class TestShaFileByName(tests.TestCaseInTempDir):
 
1650
 
 
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'))
 
1655
 
 
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'))
 
1661
 
 
1662
 
 
1663
class TestResourceLoading(tests.TestCaseInTempDir):
 
1664
 
 
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',
 
1674
            'yyy.xx')
 
1675
        # test unknown resource
 
1676
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
 
1677
 
 
1678
 
 
1679
class TestReCompile(tests.TestCase):
 
1680
 
 
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'))
 
1685
 
 
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')
 
1691
        self.assertEqual(
 
1692
            "Invalid regular expression in test case: '*': "
 
1693
            "nothing to repeat",
 
1694
            str(err))
 
1695
 
 
1696
 
 
1697
class TestDirReader(tests.TestCaseInTempDir):
 
1698
 
 
1699
    # Set by load_tests
 
1700
    _dir_reader_class = None
 
1701
    _native_to_unicode = None
 
1702
 
 
1703
    def setUp(self):
 
1704
        tests.TestCaseInTempDir.setUp(self)
 
1705
        self.overrideAttr(osutils,
 
1706
                          '_selected_dir_reader', self._dir_reader_class())
 
1707
 
 
1708
    def _get_ascii_tree(self):
 
1709
        tree = [
 
1710
            '0file',
 
1711
            '1dir/',
 
1712
            '1dir/0file',
 
1713
            '1dir/1dir/',
 
1714
            '2file'
 
1715
            ]
 
1716
        expected_dirblocks = [
 
1717
                (('', '.'),
 
1718
                 [('0file', '0file', 'file'),
 
1719
                  ('1dir', '1dir', 'directory'),
 
1720
                  ('2file', '2file', 'file'),
 
1721
                 ]
 
1722
                ),
 
1723
                (('1dir', './1dir'),
 
1724
                 [('1dir/0file', '0file', 'file'),
 
1725
                  ('1dir/1dir', '1dir', 'directory'),
 
1726
                 ]
 
1727
                ),
 
1728
                (('1dir/1dir', './1dir/1dir'),
 
1729
                 [
 
1730
                 ]
 
1731
                ),
 
1732
            ]
 
1733
        return tree, expected_dirblocks
 
1734
 
 
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])
 
1743
 
 
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])
 
1753
 
 
1754
    def _get_unicode_tree(self):
 
1755
        name0u = u'0file-\xb6'
 
1756
        name1u = u'1dir-\u062c\u0648'
 
1757
        name2u = u'2file-\u0633'
 
1758
        tree = [
 
1759
            name0u,
 
1760
            name1u + '/',
 
1761
            name1u + '/' + name0u,
 
1762
            name1u + '/' + name1u + '/',
 
1763
            name2u,
 
1764
            ]
 
1765
        name0 = name0u.encode('UTF-8')
 
1766
        name1 = name1u.encode('UTF-8')
 
1767
        name2 = name2u.encode('UTF-8')
 
1768
        expected_dirblocks = [
 
1769
                (('', '.'),
 
1770
                 [(name0, name0, 'file', './' + name0u),
 
1771
                  (name1, name1, 'directory', './' + name1u),
 
1772
                  (name2, name2, 'file', './' + name2u),
 
1773
                 ]
 
1774
                ),
 
1775
                ((name1, './' + name1u),
 
1776
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1777
                                                        + '/' + name0u),
 
1778
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1779
                                                            + '/' + name1u),
 
1780
                 ]
 
1781
                ),
 
1782
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1783
                 [
 
1784
                 ]
 
1785
                ),
 
1786
            ]
 
1787
        return tree, expected_dirblocks
 
1788
 
 
1789
    def _filter_out(self, raw_dirblocks):
 
1790
        """Filter out a walkdirs_utf8 result.
 
1791
 
 
1792
        stat field is removed, all native paths are converted to unicode
 
1793
        """
 
1794
        filtered_dirblocks = []
 
1795
        for dirinfo, block in raw_dirblocks:
 
1796
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
 
1797
            details = []
 
1798
            for line in block:
 
1799
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
 
1800
            filtered_dirblocks.append((dirinfo, details))
 
1801
        return filtered_dirblocks
 
1802
 
 
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))
 
1809
 
 
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 = [
 
1819
                (('', '.'),
 
1820
                 [(link_name_utf8, link_name_utf8,
 
1821
                   'symlink', './' + link_name),],
 
1822
                 )]
 
1823
        result = list(osutils._walkdirs_utf8('.'))
 
1824
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1825
 
 
1826
 
 
1827
class TestReadLink(tests.TestCaseInTempDir):
 
1828
    """Exposes os.readlink() problems and the osutils solution.
 
1829
 
 
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.
 
1832
 
 
1833
    But prior python versions failed to properly encode the passed unicode
 
1834
    string.
 
1835
    """
 
1836
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
 
1837
 
 
1838
    def setUp(self):
 
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)
 
1843
 
 
1844
    def test_os_readlink_link_encoding(self):
 
1845
        if sys.version_info < (2, 6):
 
1846
            self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
 
1847
        else:
 
1848
            self.assertEquals(self.target,  os.readlink(self.link))
 
1849
 
 
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)))
 
1853
 
 
1854
 
 
1855
class TestConcurrency(tests.TestCase):
 
1856
 
 
1857
    def setUp(self):
 
1858
        super(TestConcurrency, self).setUp()
 
1859
        self.overrideAttr(osutils, '_cached_local_concurrency')
 
1860
 
 
1861
    def test_local_concurrency(self):
 
1862
        concurrency = osutils.local_concurrency()
 
1863
        self.assertIsInstance(concurrency, int)
 
1864
 
 
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))
 
1872
 
 
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))
 
1879
 
 
1880
 
 
1881
class TestFailedToLoadExtension(tests.TestCase):
 
1882
 
 
1883
    def _try_loading(self):
 
1884
        try:
 
1885
            import bzrlib._fictional_extension_py
 
1886
        except ImportError, e:
 
1887
            osutils.failed_to_load_extension(e)
 
1888
            return True
 
1889
 
 
1890
    def setUp(self):
 
1891
        super(TestFailedToLoadExtension, self).setUp()
 
1892
        self.overrideAttr(osutils, '_extension_load_failures', [])
 
1893
 
 
1894
    def test_failure_to_load(self):
 
1895
        self._try_loading()
 
1896
        self.assertLength(1, osutils._extension_load_failures)
 
1897
        self.assertEquals(osutils._extension_load_failures[0],
 
1898
            "No module named _fictional_extension_py")
 
1899
 
 
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)
 
1905
 
 
1906
    def test_report_extension_load_failures_message(self):
 
1907
        log = StringIO()
 
1908
        trace.push_log_file(log)
 
1909
        self.assertTrue(self._try_loading())
 
1910
        osutils.report_extension_load_failures()
 
1911
        self.assertContainsRe(
 
1912
            log.getvalue(),
 
1913
            r"bzr: warning: some compiled extensions could not be loaded; "
 
1914
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
 
1915
            )
 
1916
 
 
1917
 
 
1918
class TestTerminalWidth(tests.TestCase):
 
1919
 
 
1920
    def replace_stdout(self, new):
 
1921
        self.overrideAttr(sys, 'stdout', new)
 
1922
 
 
1923
    def replace__terminal_size(self, new):
 
1924
        self.overrideAttr(osutils, '_terminal_size', new)
 
1925
 
 
1926
    def set_fake_tty(self):
 
1927
 
 
1928
        class I_am_a_tty(object):
 
1929
            def isatty(self):
 
1930
                return True
 
1931
 
 
1932
        self.replace_stdout(I_am_a_tty())
 
1933
 
 
1934
    def test_default_values(self):
 
1935
        self.assertEqual(80, osutils.default_terminal_width)
 
1936
 
 
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())
 
1942
 
 
1943
    def test_falls_back_to_COLUMNS(self):
 
1944
        del os.environ['BZR_COLUMNS']
 
1945
        self.assertNotEqual('42', os.environ['COLUMNS'])
 
1946
        self.set_fake_tty()
 
1947
        os.environ['COLUMNS'] = '42'
 
1948
        self.assertEqual(42, osutils.terminal_width())
 
1949
 
 
1950
    def test_tty_default_without_columns(self):
 
1951
        del os.environ['BZR_COLUMNS']
 
1952
        del os.environ['COLUMNS']
 
1953
 
 
1954
        def terminal_size(w, h):
 
1955
            return 42, 42
 
1956
 
 
1957
        self.set_fake_tty()
 
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())
 
1963
 
 
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())
 
1969
 
 
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
 
1974
        try:
 
1975
            orig = termios.TIOCGWINSZ
 
1976
        except AttributeError:
 
1977
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
 
1978
            pass
 
1979
        else:
 
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()
 
1986
 
 
1987
class TestCreationOps(tests.TestCaseInTempDir):
 
1988
    _test_needs_features = [features.chown_feature]
 
1989
 
 
1990
    def setUp(self):
 
1991
        tests.TestCaseInTempDir.setUp(self)
 
1992
        self.overrideAttr(os, 'chown', self._dummy_chown)
 
1993
 
 
1994
        # params set by call to _dummy_chown
 
1995
        self.path = self.uid = self.gid = None
 
1996
 
 
1997
    def _dummy_chown(self, path, uid, gid):
 
1998
        self.path, self.uid, self.gid = path, uid, gid
 
1999
 
 
2000
    def test_copy_ownership_from_path(self):
 
2001
        """copy_ownership_from_path test with specified src."""
 
2002
        ownsrc = '/'
 
2003
        f = open('test_file', 'wt')
 
2004
        osutils.copy_ownership_from_path('test_file', ownsrc)
 
2005
 
 
2006
        s = os.stat(ownsrc)
 
2007
        self.assertEquals(self.path, 'test_file')
 
2008
        self.assertEquals(self.uid, s.st_uid)
 
2009
        self.assertEquals(self.gid, s.st_gid)
 
2010
 
 
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')
 
2016
 
 
2017
        s = os.stat('..')
 
2018
        self.assertEquals(self.path, 'test_file')
 
2019
        self.assertEquals(self.uid, s.st_uid)
 
2020
        self.assertEquals(self.gid, s.st_gid)