~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-03-06 00:05:11 UTC
  • mfrom: (5075.1.1 named-branches-tweak)
  • Revision ID: pqm@pqm.ubuntu.com-20100306000511-w0z9eaes61s10syk
(Jelmer) Document that BzrDirFormat.colocated_branches is a cvar.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007 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
19
from cStringIO import StringIO
20
20
import errno
21
21
import os
 
22
import re
22
23
import socket
23
24
import stat
24
25
import sys
25
26
import time
26
27
 
27
 
import bzrlib
28
28
from bzrlib import (
29
29
    errors,
30
30
    osutils,
31
31
    tests,
 
32
    trace,
32
33
    win32utils,
33
34
    )
34
 
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
35
 
from bzrlib.osutils import (
36
 
        is_inside_any,
37
 
        is_inside_or_parent_of_any,
38
 
        pathjoin,
39
 
        pumpfile,
40
 
        pump_string_file,
41
 
        )
42
35
from bzrlib.tests import (
43
 
        adapt_tests,
44
 
        Feature,
45
 
        probe_unicode_in_user_encoding,
46
 
        split_suite_by_re,
47
 
        StringIOWrapper,
48
 
        SymlinkFeature,
49
 
        TestCase,
50
 
        TestCaseInTempDir,
51
 
        TestScenarioApplier,
52
 
        TestSkipped,
53
 
        )
54
 
from bzrlib.tests.file_utils import (
55
 
    FakeReadFile,
 
36
    file_utils,
 
37
    test__walkdirs_win32,
56
38
    )
57
 
from bzrlib.tests.test__walkdirs_win32 import WalkdirsWin32Feature
58
 
 
59
 
 
60
 
def load_tests(standard_tests, module, loader):
61
 
    """Parameterize readdir tests."""
62
 
    to_adapt, result = split_suite_by_re(standard_tests, "readdir")
63
 
    adapter = TestScenarioApplier()
64
 
    from bzrlib import _readdir_py
65
 
    adapter.scenarios = [('python', {'read_dir': _readdir_py.read_dir})]
66
 
    if ReadDirFeature.available():
67
 
        adapter.scenarios.append(('pyrex',
68
 
            {'read_dir': ReadDirFeature.read_dir}))
69
 
    adapt_tests(to_adapt, adapter, result)
70
 
    return result
71
 
 
72
 
 
73
 
class _ReadDirFeature(Feature):
 
39
 
 
40
 
 
41
class _UTF8DirReaderFeature(tests.Feature):
74
42
 
75
43
    def _probe(self):
76
44
        try:
77
45
            from bzrlib import _readdir_pyx
78
 
            self.read_dir = _readdir_pyx.read_dir
 
46
            self.reader = _readdir_pyx.UTF8DirReader
79
47
            return True
80
48
        except ImportError:
81
49
            return False
83
51
    def feature_name(self):
84
52
        return 'bzrlib._readdir_pyx'
85
53
 
86
 
ReadDirFeature = _ReadDirFeature()
87
 
 
88
 
 
89
 
class TestOSUtils(TestCaseInTempDir):
 
54
UTF8DirReaderFeature = _UTF8DirReaderFeature()
 
55
 
 
56
term_ios_feature = tests.ModuleAvailableFeature('termios')
 
57
 
 
58
 
 
59
def _already_unicode(s):
 
60
    return s
 
61
 
 
62
 
 
63
def _utf8_to_unicode(s):
 
64
    return s.decode('UTF-8')
 
65
 
 
66
 
 
67
def dir_reader_scenarios():
 
68
    # For each dir reader we define:
 
69
 
 
70
    # - native_to_unicode: a function converting the native_abspath as returned
 
71
    #   by DirReader.read_dir to its unicode representation
 
72
 
 
73
    # UnicodeDirReader is the fallback, it should be tested on all platforms.
 
74
    scenarios = [('unicode',
 
75
                  dict(_dir_reader_class=osutils.UnicodeDirReader,
 
76
                       _native_to_unicode=_already_unicode))]
 
77
    # Some DirReaders are platform specific and even there they may not be
 
78
    # available.
 
79
    if UTF8DirReaderFeature.available():
 
80
        from bzrlib import _readdir_pyx
 
81
        scenarios.append(('utf8',
 
82
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
 
83
                               _native_to_unicode=_utf8_to_unicode)))
 
84
 
 
85
    if test__walkdirs_win32.win32_readdir_feature.available():
 
86
        try:
 
87
            from bzrlib import _walkdirs_win32
 
88
            scenarios.append(
 
89
                ('win32',
 
90
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
 
91
                      _native_to_unicode=_already_unicode)))
 
92
        except ImportError:
 
93
            pass
 
94
    return scenarios
 
95
 
 
96
 
 
97
def load_tests(basic_tests, module, loader):
 
98
    suite = loader.suiteClass()
 
99
    dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
 
100
        basic_tests, tests.condition_isinstance(TestDirReader))
 
101
    tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
 
102
    suite.addTest(remaining_tests)
 
103
    return suite
 
104
 
 
105
 
 
106
class TestContainsWhitespace(tests.TestCase):
90
107
 
91
108
    def test_contains_whitespace(self):
92
109
        self.failUnless(osutils.contains_whitespace(u' '))
102
119
        self.failIf(osutils.contains_whitespace(u'hellothere'))
103
120
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
104
121
 
 
122
 
 
123
class TestRename(tests.TestCaseInTempDir):
 
124
 
 
125
    def create_file(self, filename, content):
 
126
        f = open(filename, 'wb')
 
127
        try:
 
128
            f.write(content)
 
129
        finally:
 
130
            f.close()
 
131
 
 
132
    def _fancy_rename(self, a, b):
 
133
        osutils.fancy_rename(a, b, rename_func=os.rename,
 
134
                             unlink_func=os.unlink)
 
135
 
105
136
    def test_fancy_rename(self):
106
137
        # This should work everywhere
107
 
        def rename(a, b):
108
 
            osutils.fancy_rename(a, b,
109
 
                    rename_func=os.rename,
110
 
                    unlink_func=os.unlink)
111
 
 
112
 
        open('a', 'wb').write('something in a\n')
113
 
        rename('a', 'b')
 
138
        self.create_file('a', 'something in a\n')
 
139
        self._fancy_rename('a', 'b')
114
140
        self.failIfExists('a')
115
141
        self.failUnlessExists('b')
116
142
        self.check_file_contents('b', 'something in a\n')
117
143
 
118
 
        open('a', 'wb').write('new something in a\n')
119
 
        rename('b', 'a')
 
144
        self.create_file('a', 'new something in a\n')
 
145
        self._fancy_rename('b', 'a')
120
146
 
121
147
        self.check_file_contents('a', 'something in a\n')
122
148
 
 
149
    def test_fancy_rename_fails_source_missing(self):
 
150
        # An exception should be raised, and the target should be left in place
 
151
        self.create_file('target', 'data in target\n')
 
152
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
153
                          'missingsource', 'target')
 
154
        self.failUnlessExists('target')
 
155
        self.check_file_contents('target', 'data in target\n')
 
156
 
 
157
    def test_fancy_rename_fails_if_source_and_target_missing(self):
 
158
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
159
                          'missingsource', 'missingtarget')
 
160
 
123
161
    def test_rename(self):
124
162
        # Rename should be semi-atomic on all platforms
125
 
        open('a', 'wb').write('something in a\n')
 
163
        self.create_file('a', 'something in a\n')
126
164
        osutils.rename('a', 'b')
127
165
        self.failIfExists('a')
128
166
        self.failUnlessExists('b')
129
167
        self.check_file_contents('b', 'something in a\n')
130
168
 
131
 
        open('a', 'wb').write('new something in a\n')
 
169
        self.create_file('a', 'new something in a\n')
132
170
        osutils.rename('b', 'a')
133
171
 
134
172
        self.check_file_contents('a', 'something in a\n')
145
183
        shape = sorted(os.listdir('.'))
146
184
        self.assertEquals(['A', 'B'], shape)
147
185
 
 
186
 
 
187
class TestRandChars(tests.TestCase):
 
188
 
148
189
    def test_01_rand_chars_empty(self):
149
190
        result = osutils.rand_chars(0)
150
191
        self.assertEqual(result, '')
155
196
        self.assertEqual(type(result), str)
156
197
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
157
198
 
 
199
 
 
200
class TestIsInside(tests.TestCase):
 
201
 
158
202
    def test_is_inside(self):
159
203
        is_inside = osutils.is_inside
160
204
        self.assertTrue(is_inside('src', 'src/foo.c'))
165
209
        self.assertTrue(is_inside('', 'foo.c'))
166
210
 
167
211
    def test_is_inside_any(self):
168
 
        SRC_FOO_C = pathjoin('src', 'foo.c')
 
212
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
169
213
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
170
214
                         (['src'], SRC_FOO_C),
171
215
                         (['src'], 'src'),
172
216
                         ]:
173
 
            self.assert_(is_inside_any(dirs, fn))
 
217
            self.assert_(osutils.is_inside_any(dirs, fn))
174
218
        for dirs, fn in [(['src'], 'srccontrol'),
175
219
                         (['src'], 'srccontrol/foo')]:
176
 
            self.assertFalse(is_inside_any(dirs, fn))
 
220
            self.assertFalse(osutils.is_inside_any(dirs, fn))
177
221
 
178
222
    def test_is_inside_or_parent_of_any(self):
179
223
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
182
226
                         (['src/bar.c', 'bla/foo.c'], 'src'),
183
227
                         (['src'], 'src'),
184
228
                         ]:
185
 
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
186
 
            
 
229
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
230
 
187
231
        for dirs, fn in [(['src'], 'srccontrol'),
188
232
                         (['srccontrol/foo.c'], 'src'),
189
233
                         (['src'], 'srccontrol/foo')]:
190
 
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
 
234
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
 
235
 
 
236
 
 
237
class TestRmTree(tests.TestCaseInTempDir):
191
238
 
192
239
    def test_rmtree(self):
193
240
        # Check to remove tree with read-only files/dirs
207
254
        self.failIfExists('dir/file')
208
255
        self.failIfExists('dir')
209
256
 
 
257
 
 
258
class TestDeleteAny(tests.TestCaseInTempDir):
 
259
 
 
260
    def test_delete_any_readonly(self):
 
261
        # from <https://bugs.launchpad.net/bzr/+bug/218206>
 
262
        self.build_tree(['d/', 'f'])
 
263
        osutils.make_readonly('d')
 
264
        osutils.make_readonly('f')
 
265
 
 
266
        osutils.delete_any('f')
 
267
        osutils.delete_any('d')
 
268
 
 
269
 
 
270
class TestKind(tests.TestCaseInTempDir):
 
271
 
210
272
    def test_file_kind(self):
211
273
        self.build_tree(['file', 'dir/'])
212
274
        self.assertEquals('file', osutils.file_kind('file'))
214
276
        if osutils.has_symlinks():
215
277
            os.symlink('symlink', 'symlink')
216
278
            self.assertEquals('symlink', osutils.file_kind('symlink'))
217
 
        
 
279
 
218
280
        # TODO: jam 20060529 Test a block device
219
281
        try:
220
282
            os.lstat('/dev/null')
242
304
                os.remove('socket')
243
305
 
244
306
    def test_kind_marker(self):
245
 
        self.assertEqual(osutils.kind_marker('file'), '')
246
 
        self.assertEqual(osutils.kind_marker('directory'), '/')
247
 
        self.assertEqual(osutils.kind_marker('symlink'), '@')
248
 
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
 
307
        self.assertEqual("", osutils.kind_marker("file"))
 
308
        self.assertEqual("/", osutils.kind_marker('directory'))
 
309
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
310
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
311
        self.assertEqual("+", osutils.kind_marker("tree-reference"))
 
312
        self.assertEqual("", osutils.kind_marker("fifo"))
 
313
        self.assertEqual("", osutils.kind_marker("socket"))
 
314
        self.assertEqual("", osutils.kind_marker("unknown"))
 
315
 
 
316
 
 
317
class TestUmask(tests.TestCaseInTempDir):
249
318
 
250
319
    def test_get_umask(self):
251
320
        if sys.platform == 'win32':
254
323
            return
255
324
 
256
325
        orig_umask = osutils.get_umask()
257
 
        try:
258
 
            os.umask(0222)
259
 
            self.assertEqual(0222, osutils.get_umask())
260
 
            os.umask(0022)
261
 
            self.assertEqual(0022, osutils.get_umask())
262
 
            os.umask(0002)
263
 
            self.assertEqual(0002, osutils.get_umask())
264
 
            os.umask(0027)
265
 
            self.assertEqual(0027, osutils.get_umask())
266
 
        finally:
267
 
            os.umask(orig_umask)
 
326
        self.addCleanup(os.umask, orig_umask)
 
327
        os.umask(0222)
 
328
        self.assertEqual(0222, osutils.get_umask())
 
329
        os.umask(0022)
 
330
        self.assertEqual(0022, osutils.get_umask())
 
331
        os.umask(0002)
 
332
        self.assertEqual(0002, osutils.get_umask())
 
333
        os.umask(0027)
 
334
        self.assertEqual(0027, osutils.get_umask())
 
335
 
 
336
 
 
337
class TestDateTime(tests.TestCase):
268
338
 
269
339
    def assertFormatedDelta(self, expected, seconds):
270
340
        """Assert osutils.format_delta formats as expected"""
305
375
    def test_format_date(self):
306
376
        self.assertRaises(errors.UnsupportedTimezoneFormat,
307
377
            osutils.format_date, 0, timezone='foo')
 
378
        self.assertIsInstance(osutils.format_date(0), str)
 
379
        self.assertIsInstance(osutils.format_local_date(0), unicode)
 
380
        # Testing for the actual value of the local weekday without
 
381
        # duplicating the code from format_date is difficult.
 
382
        # Instead blackbox.test_locale should check for localized
 
383
        # dates once they do occur in output strings.
 
384
 
 
385
    def test_format_date_with_offset_in_original_timezone(self):
 
386
        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
 
387
            osutils.format_date_with_offset_in_original_timezone(0))
 
388
        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
 
389
            osutils.format_date_with_offset_in_original_timezone(100000))
 
390
        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
 
391
            osutils.format_date_with_offset_in_original_timezone(100000, 7200))
 
392
 
 
393
    def test_local_time_offset(self):
 
394
        """Test that local_time_offset() returns a sane value."""
 
395
        offset = osutils.local_time_offset()
 
396
        self.assertTrue(isinstance(offset, int))
 
397
        # Test that the offset is no more than a eighteen hours in
 
398
        # either direction.
 
399
        # Time zone handling is system specific, so it is difficult to
 
400
        # do more specific tests, but a value outside of this range is
 
401
        # probably wrong.
 
402
        eighteen_hours = 18 * 3600
 
403
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
404
 
 
405
    def test_local_time_offset_with_timestamp(self):
 
406
        """Test that local_time_offset() works with a timestamp."""
 
407
        offset = osutils.local_time_offset(1000000000.1234567)
 
408
        self.assertTrue(isinstance(offset, int))
 
409
        eighteen_hours = 18 * 3600
 
410
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
411
 
 
412
 
 
413
class TestLinks(tests.TestCaseInTempDir):
308
414
 
309
415
    def test_dereference_path(self):
310
 
        self.requireFeature(SymlinkFeature)
 
416
        self.requireFeature(tests.SymlinkFeature)
311
417
        cwd = osutils.realpath('.')
312
418
        os.mkdir('bar')
313
419
        bar_path = osutils.pathjoin(cwd, 'bar')
316
422
        self.assertEqual(bar_path, osutils.realpath('./bar'))
317
423
        os.symlink('bar', 'foo')
318
424
        self.assertEqual(bar_path, osutils.realpath('./foo'))
319
 
        
 
425
 
320
426
        # Does not dereference terminal symlinks
321
427
        foo_path = osutils.pathjoin(cwd, 'foo')
322
428
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
354
460
            osutils.make_readonly('dangling')
355
461
            osutils.make_writable('dangling')
356
462
 
357
 
    def test_kind_marker(self):
358
 
        self.assertEqual("", osutils.kind_marker("file"))
359
 
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
360
 
        self.assertEqual("@", osutils.kind_marker("symlink"))
361
 
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
362
 
 
363
463
    def test_host_os_dereferences_symlinks(self):
364
464
        osutils.host_os_dereferences_symlinks()
365
465
 
366
466
 
367
 
class TestPumpFile(TestCase):
 
467
class TestCanonicalRelPath(tests.TestCaseInTempDir):
 
468
 
 
469
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
 
470
 
 
471
    def test_canonical_relpath_simple(self):
 
472
        f = file('MixedCaseName', 'w')
 
473
        f.close()
 
474
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
 
475
        self.failUnlessEqual('work/MixedCaseName', actual)
 
476
 
 
477
    def test_canonical_relpath_missing_tail(self):
 
478
        os.mkdir('MixedCaseParent')
 
479
        actual = osutils.canonical_relpath(self.test_base_dir,
 
480
                                           'mixedcaseparent/nochild')
 
481
        self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
 
482
 
 
483
 
 
484
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
 
485
 
 
486
    def assertRelpath(self, expected, base, path):
 
487
        actual = osutils._cicp_canonical_relpath(base, path)
 
488
        self.assertEqual(expected, actual)
 
489
 
 
490
    def test_simple(self):
 
491
        self.build_tree(['MixedCaseName'])
 
492
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
493
        self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
 
494
 
 
495
    def test_subdir_missing_tail(self):
 
496
        self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
 
497
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
498
        self.assertRelpath('MixedCaseParent/a_child', base,
 
499
                           'MixedCaseParent/a_child')
 
500
        self.assertRelpath('MixedCaseParent/a_child', base,
 
501
                           'MixedCaseParent/A_Child')
 
502
        self.assertRelpath('MixedCaseParent/not_child', base,
 
503
                           'MixedCaseParent/not_child')
 
504
 
 
505
    def test_at_root_slash(self):
 
506
        # We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
 
507
        # check...
 
508
        if osutils.MIN_ABS_PATHLENGTH > 1:
 
509
            raise tests.TestSkipped('relpath requires %d chars'
 
510
                                    % osutils.MIN_ABS_PATHLENGTH)
 
511
        self.assertRelpath('foo', '/', '/foo')
 
512
 
 
513
    def test_at_root_drive(self):
 
514
        if sys.platform != 'win32':
 
515
            raise tests.TestNotApplicable('we can only test drive-letter relative'
 
516
                                          ' paths on Windows where we have drive'
 
517
                                          ' letters.')
 
518
        # see bug #322807
 
519
        # The specific issue is that when at the root of a drive, 'abspath'
 
520
        # returns "C:/" or just "/". However, the code assumes that abspath
 
521
        # always returns something like "C:/foo" or "/foo" (no trailing slash).
 
522
        self.assertRelpath('foo', 'C:/', 'C:/foo')
 
523
        self.assertRelpath('foo', 'X:/', 'X:/foo')
 
524
        self.assertRelpath('foo', 'X:/', 'X://foo')
 
525
 
 
526
 
 
527
class TestPumpFile(tests.TestCase):
368
528
    """Test pumpfile method."""
 
529
 
369
530
    def setUp(self):
 
531
        tests.TestCase.setUp(self)
370
532
        # create a test datablock
371
533
        self.block_size = 512
372
534
        pattern = '0123456789ABCDEF'
379
541
        # make sure test data is larger than max read size
380
542
        self.assertTrue(self.test_data_len > self.block_size)
381
543
 
382
 
        from_file = FakeReadFile(self.test_data)
 
544
        from_file = file_utils.FakeReadFile(self.test_data)
383
545
        to_file = StringIO()
384
546
 
385
547
        # read (max / 2) bytes and verify read size wasn't affected
386
548
        num_bytes_to_read = self.block_size / 2
387
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
549
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
388
550
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
389
551
        self.assertEqual(from_file.get_read_count(), 1)
390
552
 
391
553
        # read (max) bytes and verify read size wasn't affected
392
554
        num_bytes_to_read = self.block_size
393
555
        from_file.reset_read_count()
394
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
556
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
395
557
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
396
558
        self.assertEqual(from_file.get_read_count(), 1)
397
559
 
398
560
        # read (max + 1) bytes and verify read size was limited
399
561
        num_bytes_to_read = self.block_size + 1
400
562
        from_file.reset_read_count()
401
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
563
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
402
564
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
403
565
        self.assertEqual(from_file.get_read_count(), 2)
404
566
 
405
567
        # finish reading the rest of the data
406
568
        num_bytes_to_read = self.test_data_len - to_file.tell()
407
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
569
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
408
570
 
409
571
        # report error if the data wasn't equal (we only report the size due
410
572
        # to the length of the data)
420
582
        self.assertTrue(self.test_data_len > self.block_size)
421
583
 
422
584
        # retrieve data in blocks
423
 
        from_file = FakeReadFile(self.test_data)
 
585
        from_file = file_utils.FakeReadFile(self.test_data)
424
586
        to_file = StringIO()
425
 
        pumpfile(from_file, to_file, self.test_data_len, self.block_size)
 
587
        osutils.pumpfile(from_file, to_file, self.test_data_len,
 
588
                         self.block_size)
426
589
 
427
590
        # verify read size was equal to the maximum read size
428
591
        self.assertTrue(from_file.get_max_read_size() > 0)
443
606
        self.assertTrue(self.test_data_len > self.block_size)
444
607
 
445
608
        # retrieve data to EOF
446
 
        from_file = FakeReadFile(self.test_data)
 
609
        from_file = file_utils.FakeReadFile(self.test_data)
447
610
        to_file = StringIO()
448
 
        pumpfile(from_file, to_file, -1, self.block_size)
 
611
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
449
612
 
450
613
        # verify read size was equal to the maximum read size
451
614
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
463
626
        test verifies that any existing usages of pumpfile will not be broken
464
627
        with this new version."""
465
628
        # retrieve data using default (old) pumpfile method
466
 
        from_file = FakeReadFile(self.test_data)
 
629
        from_file = file_utils.FakeReadFile(self.test_data)
467
630
        to_file = StringIO()
468
 
        pumpfile(from_file, to_file)
 
631
        osutils.pumpfile(from_file, to_file)
469
632
 
470
633
        # report error if the data wasn't equal (we only report the size due
471
634
        # to the length of the data)
474
637
            message = "Data not equal.  Expected %d bytes, received %d."
475
638
            self.fail(message % (len(response_data), self.test_data_len))
476
639
 
477
 
 
478
 
class TestPumpStringFile(TestCase):
 
640
    def test_report_activity(self):
 
641
        activity = []
 
642
        def log_activity(length, direction):
 
643
            activity.append((length, direction))
 
644
        from_file = StringIO(self.test_data)
 
645
        to_file = StringIO()
 
646
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
647
                         report_activity=log_activity, direction='read')
 
648
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
 
649
                          (36, 'read')], activity)
 
650
 
 
651
        from_file = StringIO(self.test_data)
 
652
        to_file = StringIO()
 
653
        del activity[:]
 
654
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
655
                         report_activity=log_activity, direction='write')
 
656
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
 
657
                          (36, 'write')], activity)
 
658
 
 
659
        # And with a limited amount of data
 
660
        from_file = StringIO(self.test_data)
 
661
        to_file = StringIO()
 
662
        del activity[:]
 
663
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
 
664
                         report_activity=log_activity, direction='read')
 
665
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
 
666
 
 
667
 
 
668
 
 
669
class TestPumpStringFile(tests.TestCase):
479
670
 
480
671
    def test_empty(self):
481
672
        output = StringIO()
482
 
        pump_string_file("", output)
 
673
        osutils.pump_string_file("", output)
483
674
        self.assertEqual("", output.getvalue())
484
675
 
485
676
    def test_more_than_segment_size(self):
486
677
        output = StringIO()
487
 
        pump_string_file("123456789", output, 2)
 
678
        osutils.pump_string_file("123456789", output, 2)
488
679
        self.assertEqual("123456789", output.getvalue())
489
680
 
490
681
    def test_segment_size(self):
491
682
        output = StringIO()
492
 
        pump_string_file("12", output, 2)
 
683
        osutils.pump_string_file("12", output, 2)
493
684
        self.assertEqual("12", output.getvalue())
494
685
 
495
686
    def test_segment_size_multiple(self):
496
687
        output = StringIO()
497
 
        pump_string_file("1234", output, 2)
 
688
        osutils.pump_string_file("1234", output, 2)
498
689
        self.assertEqual("1234", output.getvalue())
499
690
 
500
691
 
501
 
class TestSafeUnicode(TestCase):
 
692
class TestRelpath(tests.TestCase):
 
693
 
 
694
    def test_simple_relpath(self):
 
695
        cwd = osutils.getcwd()
 
696
        subdir = cwd + '/subdir'
 
697
        self.assertEqual('subdir', osutils.relpath(cwd, subdir))
 
698
 
 
699
    def test_deep_relpath(self):
 
700
        cwd = osutils.getcwd()
 
701
        subdir = cwd + '/sub/subsubdir'
 
702
        self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
 
703
 
 
704
    def test_not_relative(self):
 
705
        self.assertRaises(errors.PathNotChild,
 
706
                          osutils.relpath, 'C:/path', 'H:/path')
 
707
        self.assertRaises(errors.PathNotChild,
 
708
                          osutils.relpath, 'C:/', 'H:/path')
 
709
 
 
710
 
 
711
class TestSafeUnicode(tests.TestCase):
502
712
 
503
713
    def test_from_ascii_string(self):
504
714
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
513
723
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
514
724
 
515
725
    def test_bad_utf8_string(self):
516
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
726
        self.assertRaises(errors.BzrBadParameterNotUnicode,
517
727
                          osutils.safe_unicode,
518
728
                          '\xbb\xbb')
519
729
 
520
730
 
521
 
class TestSafeUtf8(TestCase):
 
731
class TestSafeUtf8(tests.TestCase):
522
732
 
523
733
    def test_from_ascii_string(self):
524
734
        f = 'foobar'
534
744
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
535
745
 
536
746
    def test_bad_utf8_string(self):
537
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
747
        self.assertRaises(errors.BzrBadParameterNotUnicode,
538
748
                          osutils.safe_utf8, '\xbb\xbb')
539
749
 
540
750
 
541
 
class TestSafeRevisionId(TestCase):
 
751
class TestSafeRevisionId(tests.TestCase):
542
752
 
543
753
    def test_from_ascii_string(self):
544
754
        # this shouldn't give a warning because it's getting an ascii string
566
776
        self.assertEqual(None, osutils.safe_revision_id(None))
567
777
 
568
778
 
569
 
class TestSafeFileId(TestCase):
 
779
class TestSafeFileId(tests.TestCase):
570
780
 
571
781
    def test_from_ascii_string(self):
572
782
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
592
802
        self.assertEqual(None, osutils.safe_file_id(None))
593
803
 
594
804
 
595
 
class TestWin32Funcs(TestCase):
596
 
    """Test that the _win32 versions of os utilities return appropriate paths."""
 
805
class TestWin32Funcs(tests.TestCase):
 
806
    """Test that _win32 versions of os utilities return appropriate paths."""
597
807
 
598
808
    def test_abspath(self):
599
809
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
606
816
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
607
817
 
608
818
    def test_pathjoin(self):
609
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
610
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
611
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
612
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
613
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
614
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
819
        self.assertEqual('path/to/foo',
 
820
                         osutils._win32_pathjoin('path', 'to', 'foo'))
 
821
        self.assertEqual('C:/foo',
 
822
                         osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
823
        self.assertEqual('C:/foo',
 
824
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
 
825
        self.assertEqual('path/to/foo',
 
826
                         osutils._win32_pathjoin('path/to/', 'foo'))
 
827
        self.assertEqual('/foo',
 
828
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
829
        self.assertEqual('/foo',
 
830
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
615
831
 
616
832
    def test_normpath(self):
617
 
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
618
 
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
 
833
        self.assertEqual('path/to/foo',
 
834
                         osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
835
        self.assertEqual('path/to/foo',
 
836
                         osutils._win32_normpath('path//from/../to/./foo'))
619
837
 
620
838
    def test_getcwd(self):
621
839
        cwd = osutils._win32_getcwd()
650
868
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
651
869
 
652
870
 
653
 
class TestWin32FuncsDirs(TestCaseInTempDir):
 
871
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
654
872
    """Test win32 functions that create files."""
655
 
    
 
873
 
656
874
    def test_getcwd(self):
657
 
        if win32utils.winver == 'Windows 98':
658
 
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
659
 
        # Make sure getcwd can handle unicode filenames
660
 
        try:
661
 
            os.mkdir(u'mu-\xb5')
662
 
        except UnicodeError:
663
 
            raise TestSkipped("Unable to create Unicode filename")
664
 
 
 
875
        self.requireFeature(tests.UnicodeFilenameFeature)
 
876
        os.mkdir(u'mu-\xb5')
665
877
        os.chdir(u'mu-\xb5')
666
878
        # TODO: jam 20060427 This will probably fail on Mac OSX because
667
879
        #       it will change the normalization of B\xe5gfors
672
884
    def test_minimum_path_selection(self):
673
885
        self.assertEqual(set(),
674
886
            osutils.minimum_path_selection([]))
 
887
        self.assertEqual(set(['a']),
 
888
            osutils.minimum_path_selection(['a']))
675
889
        self.assertEqual(set(['a', 'b']),
676
890
            osutils.minimum_path_selection(['a', 'b']))
677
891
        self.assertEqual(set(['a/', 'b']),
678
892
            osutils.minimum_path_selection(['a/', 'b']))
679
893
        self.assertEqual(set(['a/', 'b']),
680
894
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
 
895
        self.assertEqual(set(['a-b', 'a', 'a0b']),
 
896
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
681
897
 
682
898
    def test_mkdtemp(self):
683
899
        tmpdir = osutils._win32_mkdtemp(dir='.')
739
955
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
740
956
 
741
957
 
742
 
class TestMacFuncsDirs(TestCaseInTempDir):
 
958
class TestParentDirectories(tests.TestCaseInTempDir):
 
959
    """Test osutils.parent_directories()"""
 
960
 
 
961
    def test_parent_directories(self):
 
962
        self.assertEqual([], osutils.parent_directories('a'))
 
963
        self.assertEqual(['a'], osutils.parent_directories('a/b'))
 
964
        self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
 
965
 
 
966
 
 
967
class TestMacFuncsDirs(tests.TestCaseInTempDir):
743
968
    """Test mac special functions that require directories."""
744
969
 
745
970
    def test_getcwd(self):
746
 
        # On Mac, this will actually create Ba\u030agfors
747
 
        # but chdir will still work, because it accepts both paths
748
 
        try:
749
 
            os.mkdir(u'B\xe5gfors')
750
 
        except UnicodeError:
751
 
            raise TestSkipped("Unable to create Unicode filename")
752
 
 
 
971
        self.requireFeature(tests.UnicodeFilenameFeature)
 
972
        os.mkdir(u'B\xe5gfors')
753
973
        os.chdir(u'B\xe5gfors')
754
974
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
755
975
 
756
976
    def test_getcwd_nonnorm(self):
 
977
        self.requireFeature(tests.UnicodeFilenameFeature)
757
978
        # Test that _mac_getcwd() will normalize this path
758
 
        try:
759
 
            os.mkdir(u'Ba\u030agfors')
760
 
        except UnicodeError:
761
 
            raise TestSkipped("Unable to create Unicode filename")
762
 
 
 
979
        os.mkdir(u'Ba\u030agfors')
763
980
        os.chdir(u'Ba\u030agfors')
764
981
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
765
982
 
766
983
 
767
 
class TestSplitLines(TestCase):
 
984
class TestChunksToLines(tests.TestCase):
 
985
 
 
986
    def test_smoketest(self):
 
987
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
988
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
 
989
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
990
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
 
991
 
 
992
    def test_osutils_binding(self):
 
993
        from bzrlib.tests import test__chunks_to_lines
 
994
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
 
995
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
 
996
        else:
 
997
            from bzrlib._chunks_to_lines_py import chunks_to_lines
 
998
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
 
999
 
 
1000
 
 
1001
class TestSplitLines(tests.TestCase):
768
1002
 
769
1003
    def test_split_unicode(self):
770
1004
        self.assertEqual([u'foo\n', u'bar\xae'],
777
1011
                         osutils.split_lines('foo\rbar\n'))
778
1012
 
779
1013
 
780
 
class TestWalkDirs(TestCaseInTempDir):
781
 
 
782
 
    def test_readdir(self):
783
 
        tree = [
784
 
            '.bzr/',
785
 
            '0file',
786
 
            '1dir/',
787
 
            '1dir/0file',
788
 
            '1dir/1dir/',
789
 
            '2file'
790
 
            ]
791
 
        self.build_tree(tree)
792
 
        expected_names = ['.bzr', '0file', '1dir', '2file']
793
 
        # read_dir returns pairs, which form a table with either None in all
794
 
        # the first columns, or a sort key to get best on-disk-read order, 
795
 
        # and the disk path name in utf-8 encoding in the second column.
796
 
        read_result = self.read_dir('.')
797
 
        # The second column is always the names, and every name except "." and
798
 
        # ".." should be present.
799
 
        names = sorted([row[1] for row in read_result])
800
 
        self.assertEqual(expected_names, names)
801
 
        expected_sort_key = None
802
 
        if read_result[0][0] is None:
803
 
            # No sort key returned - all keys must None
804
 
            operator = self.assertEqual
805
 
        else:
806
 
            # A sort key in the first row implies sort keys in the other rows.
807
 
            operator = self.assertNotEqual
808
 
        for row in read_result:
809
 
            operator(None, row[0])
810
 
 
811
 
    def test_compiled_extension_exists(self):
812
 
        self.requireFeature(ReadDirFeature)
813
 
        
 
1014
class TestWalkDirs(tests.TestCaseInTempDir):
 
1015
 
 
1016
    def assertExpectedBlocks(self, expected, result):
 
1017
        self.assertEqual(expected,
 
1018
                         [(dirinfo, [line[0:3] for line in block])
 
1019
                          for dirinfo, block in result])
 
1020
 
814
1021
    def test_walkdirs(self):
815
1022
        tree = [
816
1023
            '.bzr',
848
1055
            result.append((dirdetail, dirblock))
849
1056
 
850
1057
        self.assertTrue(found_bzrdir)
851
 
        self.assertEqual(expected_dirblocks,
852
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1058
        self.assertExpectedBlocks(expected_dirblocks, result)
853
1059
        # you can search a subdir only, with a supplied prefix.
854
1060
        result = []
855
1061
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
856
1062
            result.append(dirblock)
857
 
        self.assertEqual(expected_dirblocks[1:],
858
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1063
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
1064
 
 
1065
    def test_walkdirs_os_error(self):
 
1066
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
 
1067
        # Pyrex readdir didn't raise useful messages if it had an error
 
1068
        # reading the directory
 
1069
        if sys.platform == 'win32':
 
1070
            raise tests.TestNotApplicable(
 
1071
                "readdir IOError not tested on win32")
 
1072
        os.mkdir("test-unreadable")
 
1073
        os.chmod("test-unreadable", 0000)
 
1074
        # must chmod it back so that it can be removed
 
1075
        self.addCleanup(os.chmod, "test-unreadable", 0700)
 
1076
        # The error is not raised until the generator is actually evaluated.
 
1077
        # (It would be ok if it happened earlier but at the moment it
 
1078
        # doesn't.)
 
1079
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
 
1080
        self.assertEquals('./test-unreadable', e.filename)
 
1081
        self.assertEquals(errno.EACCES, e.errno)
 
1082
        # Ensure the message contains the file name
 
1083
        self.assertContainsRe(str(e), "\./test-unreadable")
859
1084
 
860
1085
    def test__walkdirs_utf8(self):
861
1086
        tree = [
894
1119
            result.append((dirdetail, dirblock))
895
1120
 
896
1121
        self.assertTrue(found_bzrdir)
897
 
        self.assertEqual(expected_dirblocks,
898
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1122
        self.assertExpectedBlocks(expected_dirblocks, result)
 
1123
 
899
1124
        # you can search a subdir only, with a supplied prefix.
900
1125
        result = []
901
1126
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
902
1127
            result.append(dirblock)
903
 
        self.assertEqual(expected_dirblocks[1:],
904
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1128
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
905
1129
 
906
1130
    def _filter_out_stat(self, result):
907
1131
        """Filter out the stat value from the walkdirs result"""
912
1136
                new_dirblock.append((info[0], info[1], info[2], info[4]))
913
1137
            dirblock[:] = new_dirblock
914
1138
 
915
 
    def test__walkdirs_utf8_selection(self):
916
 
        # Just trigger the function once, to make sure it has selected a real
917
 
        # implementation.
918
 
        list(osutils._walkdirs_utf8('.'))
919
 
        if WalkdirsWin32Feature.available():
920
 
            # If the compiled form is available, make sure it is used
921
 
            from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
922
 
            self.assertIs(_walkdirs_utf8_win32_find_file,
923
 
                          osutils._real_walkdirs_utf8)
924
 
        elif sys.platform == 'win32':
925
 
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
926
 
                          osutils._real_walkdirs_utf8)
927
 
        elif osutils._fs_enc.upper() in ('UTF-8', 'ASCII', 'ANSI_X3.4-1968'): # ascii
928
 
            self.assertIs(osutils._walkdirs_fs_utf8,
929
 
                          osutils._real_walkdirs_utf8)
930
 
        else:
931
 
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
932
 
                          osutils._real_walkdirs_utf8)
933
 
 
934
1139
    def _save_platform_info(self):
935
 
        cur_winver = win32utils.winver
936
 
        cur_fs_enc = osutils._fs_enc
937
 
        cur_real_walkdirs_utf8 = osutils._real_walkdirs_utf8
938
 
        def restore():
939
 
            win32utils.winver = cur_winver
940
 
            osutils._fs_enc = cur_fs_enc
941
 
            osutils._real_walkdirs_utf8 = cur_real_walkdirs_utf8
942
 
        self.addCleanup(restore)
 
1140
        self.overrideAttr(win32utils, 'winver')
 
1141
        self.overrideAttr(osutils, '_fs_enc')
 
1142
        self.overrideAttr(osutils, '_selected_dir_reader')
943
1143
 
944
 
    def assertWalkdirsUtf8Is(self, expected):
 
1144
    def assertDirReaderIs(self, expected):
945
1145
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
946
1146
        # Force it to redetect
947
 
        osutils._real_walkdirs_utf8 = None
 
1147
        osutils._selected_dir_reader = None
948
1148
        # Nothing to list, but should still trigger the selection logic
949
1149
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
950
 
        self.assertIs(expected, osutils._real_walkdirs_utf8)
 
1150
        self.assertIsInstance(osutils._selected_dir_reader, expected)
951
1151
 
952
1152
    def test_force_walkdirs_utf8_fs_utf8(self):
 
1153
        self.requireFeature(UTF8DirReaderFeature)
953
1154
        self._save_platform_info()
954
1155
        win32utils.winver = None # Avoid the win32 detection code
955
1156
        osutils._fs_enc = 'UTF-8'
956
 
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
1157
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
957
1158
 
958
1159
    def test_force_walkdirs_utf8_fs_ascii(self):
 
1160
        self.requireFeature(UTF8DirReaderFeature)
959
1161
        self._save_platform_info()
960
1162
        win32utils.winver = None # Avoid the win32 detection code
961
1163
        osutils._fs_enc = 'US-ASCII'
962
 
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
1164
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
963
1165
 
964
1166
    def test_force_walkdirs_utf8_fs_ANSI(self):
 
1167
        self.requireFeature(UTF8DirReaderFeature)
965
1168
        self._save_platform_info()
966
1169
        win32utils.winver = None # Avoid the win32 detection code
967
1170
        osutils._fs_enc = 'ANSI_X3.4-1968'
968
 
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
1171
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
969
1172
 
970
1173
    def test_force_walkdirs_utf8_fs_latin1(self):
971
1174
        self._save_platform_info()
972
1175
        win32utils.winver = None # Avoid the win32 detection code
973
1176
        osutils._fs_enc = 'latin1'
974
 
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
 
1177
        self.assertDirReaderIs(osutils.UnicodeDirReader)
975
1178
 
976
1179
    def test_force_walkdirs_utf8_nt(self):
977
 
        self.requireFeature(WalkdirsWin32Feature)
 
1180
        # Disabled because the thunk of the whole walkdirs api is disabled.
 
1181
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
978
1182
        self._save_platform_info()
979
1183
        win32utils.winver = 'Windows NT'
980
 
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
981
 
        self.assertWalkdirsUtf8Is(_walkdirs_utf8_win32_find_file)
 
1184
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1185
        self.assertDirReaderIs(Win32ReadDir)
982
1186
 
983
 
    def test_force_walkdirs_utf8_nt(self):
984
 
        self.requireFeature(WalkdirsWin32Feature)
 
1187
    def test_force_walkdirs_utf8_98(self):
 
1188
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
985
1189
        self._save_platform_info()
986
1190
        win32utils.winver = 'Windows 98'
987
 
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
 
1191
        self.assertDirReaderIs(osutils.UnicodeDirReader)
988
1192
 
989
1193
    def test_unicode_walkdirs(self):
990
1194
        """Walkdirs should always return unicode paths."""
 
1195
        self.requireFeature(tests.UnicodeFilenameFeature)
991
1196
        name0 = u'0file-\xb6'
992
1197
        name1 = u'1dir-\u062c\u0648'
993
1198
        name2 = u'2file-\u0633'
998
1203
            name1 + '/' + name1 + '/',
999
1204
            name2,
1000
1205
            ]
1001
 
        try:
1002
 
            self.build_tree(tree)
1003
 
        except UnicodeError:
1004
 
            raise TestSkipped('Could not represent Unicode chars'
1005
 
                              ' in current encoding.')
 
1206
        self.build_tree(tree)
1006
1207
        expected_dirblocks = [
1007
1208
                ((u'', u'.'),
1008
1209
                 [(name0, name0, 'file', './' + name0),
1034
1235
 
1035
1236
        The abspath portion might be in unicode or utf-8
1036
1237
        """
 
1238
        self.requireFeature(tests.UnicodeFilenameFeature)
1037
1239
        name0 = u'0file-\xb6'
1038
1240
        name1 = u'1dir-\u062c\u0648'
1039
1241
        name2 = u'2file-\u0633'
1044
1246
            name1 + '/' + name1 + '/',
1045
1247
            name2,
1046
1248
            ]
1047
 
        try:
1048
 
            self.build_tree(tree)
1049
 
        except UnicodeError:
1050
 
            raise TestSkipped('Could not represent Unicode chars'
1051
 
                              ' in current encoding.')
 
1249
        self.build_tree(tree)
1052
1250
        name0 = name0.encode('utf8')
1053
1251
        name1 = name1.encode('utf8')
1054
1252
        name2 = name2.encode('utf8')
1093
1291
            result.append((dirdetail, new_dirblock))
1094
1292
        self.assertEqual(expected_dirblocks, result)
1095
1293
 
1096
 
    def test_unicode__walkdirs_unicode_to_utf8(self):
1097
 
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
 
1294
    def test__walkdirs_utf8_with_unicode_fs(self):
 
1295
        """UnicodeDirReader should be a safe fallback everywhere
1098
1296
 
1099
1297
        The abspath portion should be in unicode
1100
1298
        """
 
1299
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1300
        # Use the unicode reader. TODO: split into driver-and-driven unit
 
1301
        # tests.
 
1302
        self._save_platform_info()
 
1303
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
1101
1304
        name0u = u'0file-\xb6'
1102
1305
        name1u = u'1dir-\u062c\u0648'
1103
1306
        name2u = u'2file-\u0633'
1108
1311
            name1u + '/' + name1u + '/',
1109
1312
            name2u,
1110
1313
            ]
1111
 
        try:
1112
 
            self.build_tree(tree)
1113
 
        except UnicodeError:
1114
 
            raise TestSkipped('Could not represent Unicode chars'
1115
 
                              ' in current encoding.')
 
1314
        self.build_tree(tree)
1116
1315
        name0 = name0u.encode('utf8')
1117
1316
        name1 = name1u.encode('utf8')
1118
1317
        name2 = name2u.encode('utf8')
1138
1337
                 ]
1139
1338
                ),
1140
1339
            ]
1141
 
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
 
1340
        result = list(osutils._walkdirs_utf8('.'))
1142
1341
        self._filter_out_stat(result)
1143
1342
        self.assertEqual(expected_dirblocks, result)
1144
1343
 
1145
 
    def test__walkdirs_utf_win32_find_file(self):
1146
 
        self.requireFeature(WalkdirsWin32Feature)
 
1344
    def test__walkdirs_utf8_win32readdir(self):
 
1345
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1147
1346
        self.requireFeature(tests.UnicodeFilenameFeature)
1148
 
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1347
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1348
        self._save_platform_info()
 
1349
        osutils._selected_dir_reader = Win32ReadDir()
1149
1350
        name0u = u'0file-\xb6'
1150
1351
        name1u = u'1dir-\u062c\u0648'
1151
1352
        name2u = u'2file-\u0633'
1182
1383
                 ]
1183
1384
                ),
1184
1385
            ]
1185
 
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1386
        result = list(osutils._walkdirs_utf8(u'.'))
1186
1387
        self._filter_out_stat(result)
1187
1388
        self.assertEqual(expected_dirblocks, result)
1188
1389
 
1198
1399
 
1199
1400
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1200
1401
        """make sure our Stat values are valid"""
1201
 
        self.requireFeature(WalkdirsWin32Feature)
 
1402
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1202
1403
        self.requireFeature(tests.UnicodeFilenameFeature)
1203
 
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1404
        from bzrlib._walkdirs_win32 import Win32ReadDir
1204
1405
        name0u = u'0file-\xb6'
1205
1406
        name0 = name0u.encode('utf8')
1206
1407
        self.build_tree([name0u])
1213
1414
        finally:
1214
1415
            f.close()
1215
1416
 
1216
 
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
1217
 
        entry = result[0][1][0]
 
1417
        result = Win32ReadDir().read_dir('', u'.')
 
1418
        entry = result[0]
1218
1419
        self.assertEqual((name0, name0, 'file'), entry[:3])
1219
1420
        self.assertEqual(u'./' + name0u, entry[4])
1220
1421
        self.assertStatIsCorrect(entry[4], entry[3])
1222
1423
 
1223
1424
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1224
1425
        """make sure our Stat values are valid"""
1225
 
        self.requireFeature(WalkdirsWin32Feature)
 
1426
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1226
1427
        self.requireFeature(tests.UnicodeFilenameFeature)
1227
 
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1428
        from bzrlib._walkdirs_win32 import Win32ReadDir
1228
1429
        name0u = u'0dir-\u062c\u0648'
1229
1430
        name0 = name0u.encode('utf8')
1230
1431
        self.build_tree([name0u + '/'])
1231
1432
 
1232
 
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
1233
 
        entry = result[0][1][0]
 
1433
        result = Win32ReadDir().read_dir('', u'.')
 
1434
        entry = result[0]
1234
1435
        self.assertEqual((name0, name0, 'directory'), entry[:3])
1235
1436
        self.assertEqual(u'./' + name0u, entry[4])
1236
1437
        self.assertStatIsCorrect(entry[4], entry[3])
1313
1514
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1314
1515
 
1315
1516
 
1316
 
class TestCopyTree(TestCaseInTempDir):
1317
 
    
 
1517
class TestCopyTree(tests.TestCaseInTempDir):
 
1518
 
1318
1519
    def test_copy_basic_tree(self):
1319
1520
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1320
1521
        osutils.copy_tree('source', 'target')
1329
1530
        self.assertEqual(['c'], os.listdir('target/b'))
1330
1531
 
1331
1532
    def test_copy_tree_symlinks(self):
1332
 
        self.requireFeature(SymlinkFeature)
 
1533
        self.requireFeature(tests.SymlinkFeature)
1333
1534
        self.build_tree(['source/'])
1334
1535
        os.symlink('a/generic/path', 'source/lnk')
1335
1536
        osutils.copy_tree('source', 'target')
1365
1566
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1366
1567
 
1367
1568
 
1368
 
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
1369
 
# [bialix] 2006/12/26
1370
 
 
1371
 
 
1372
 
class TestSetUnsetEnv(TestCase):
 
1569
class TestSetUnsetEnv(tests.TestCase):
1373
1570
    """Test updating the environment"""
1374
1571
 
1375
1572
    def setUp(self):
1381
1578
        def cleanup():
1382
1579
            if 'BZR_TEST_ENV_VAR' in os.environ:
1383
1580
                del os.environ['BZR_TEST_ENV_VAR']
1384
 
 
1385
1581
        self.addCleanup(cleanup)
1386
1582
 
1387
1583
    def test_set(self):
1399
1595
 
1400
1596
    def test_unicode(self):
1401
1597
        """Environment can only contain plain strings
1402
 
        
 
1598
 
1403
1599
        So Unicode strings must be encoded.
1404
1600
        """
1405
 
        uni_val, env_val = probe_unicode_in_user_encoding()
 
1601
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
1406
1602
        if uni_val is None:
1407
 
            raise TestSkipped('Cannot find a unicode character that works in'
1408
 
                              ' encoding %s' % (bzrlib.user_encoding,))
 
1603
            raise tests.TestSkipped(
 
1604
                'Cannot find a unicode character that works in encoding %s'
 
1605
                % (osutils.get_user_encoding(),))
1409
1606
 
1410
1607
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1411
1608
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1419
1616
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1420
1617
 
1421
1618
 
1422
 
class TestLocalTimeOffset(TestCase):
1423
 
 
1424
 
    def test_local_time_offset(self):
1425
 
        """Test that local_time_offset() returns a sane value."""
1426
 
        offset = osutils.local_time_offset()
1427
 
        self.assertTrue(isinstance(offset, int))
1428
 
        # Test that the offset is no more than a eighteen hours in
1429
 
        # either direction.
1430
 
        # Time zone handling is system specific, so it is difficult to
1431
 
        # do more specific tests, but a value outside of this range is
1432
 
        # probably wrong.
1433
 
        eighteen_hours = 18 * 3600
1434
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1435
 
 
1436
 
    def test_local_time_offset_with_timestamp(self):
1437
 
        """Test that local_time_offset() works with a timestamp."""
1438
 
        offset = osutils.local_time_offset(1000000000.1234567)
1439
 
        self.assertTrue(isinstance(offset, int))
1440
 
        eighteen_hours = 18 * 3600
1441
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1442
 
 
1443
 
 
1444
 
class TestShaFileByName(TestCaseInTempDir):
1445
 
 
1446
 
    def test_sha_empty(self):
1447
 
        self.build_tree_contents([('foo', '')])
1448
 
        expected_sha = osutils.sha_string('')
1449
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1450
 
 
1451
 
    def test_sha_mixed_endings(self):
1452
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1453
 
        self.build_tree_contents([('foo', text)])
1454
 
        expected_sha = osutils.sha_string(text)
1455
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1456
 
 
1457
 
 
1458
 
_debug_text = \
1459
 
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1460
 
#
1461
 
# This program is free software; you can redistribute it and/or modify
1462
 
# it under the terms of the GNU General Public License as published by
1463
 
# the Free Software Foundation; either version 2 of the License, or
1464
 
# (at your option) any later version.
1465
 
#
1466
 
# This program is distributed in the hope that it will be useful,
1467
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1468
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
1469
 
# GNU General Public License for more details.
1470
 
#
1471
 
# You should have received a copy of the GNU General Public License
1472
 
# along with this program; if not, write to the Free Software
1473
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
1474
 
 
1475
 
 
1476
 
# NOTE: If update these, please also update the help for global-options in
1477
 
#       bzrlib/help_topics/__init__.py
1478
 
 
1479
 
debug_flags = set()
1480
 
"""Set of flags that enable different debug behaviour.
1481
 
 
1482
 
These are set with eg ``-Dlock`` on the bzr command line.
1483
 
 
1484
 
Options include:
1485
 
 
1486
 
 * auth - show authentication sections used
1487
 
 * error - show stack traces for all top level exceptions
1488
 
 * evil - capture call sites that do expensive or badly-scaling operations.
1489
 
 * fetch - trace history copying between repositories
1490
 
 * graph - trace graph traversal information
1491
 
 * hashcache - log every time a working file is read to determine its hash
1492
 
 * hooks - trace hook execution
1493
 
 * hpss - trace smart protocol requests and responses
1494
 
 * http - trace http connections, requests and responses
1495
 
 * index - trace major index operations
1496
 
 * knit - trace knit operations
1497
 
 * lock - trace when lockdir locks are taken or released
1498
 
 * merge - emit information for debugging merges
1499
 
 * pack - emit information about pack operations
1500
 
 
1501
 
"""
1502
 
'''
1503
 
 
1504
 
 
1505
 
class TestResourceLoading(TestCaseInTempDir):
 
1619
class TestSizeShaFile(tests.TestCaseInTempDir):
 
1620
 
 
1621
    def test_sha_empty(self):
 
1622
        self.build_tree_contents([('foo', '')])
 
1623
        expected_sha = osutils.sha_string('')
 
1624
        f = open('foo')
 
1625
        self.addCleanup(f.close)
 
1626
        size, sha = osutils.size_sha_file(f)
 
1627
        self.assertEqual(0, size)
 
1628
        self.assertEqual(expected_sha, sha)
 
1629
 
 
1630
    def test_sha_mixed_endings(self):
 
1631
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1632
        self.build_tree_contents([('foo', text)])
 
1633
        expected_sha = osutils.sha_string(text)
 
1634
        f = open('foo', 'rb')
 
1635
        self.addCleanup(f.close)
 
1636
        size, sha = osutils.size_sha_file(f)
 
1637
        self.assertEqual(38, size)
 
1638
        self.assertEqual(expected_sha, sha)
 
1639
 
 
1640
 
 
1641
class TestShaFileByName(tests.TestCaseInTempDir):
 
1642
 
 
1643
    def test_sha_empty(self):
 
1644
        self.build_tree_contents([('foo', '')])
 
1645
        expected_sha = osutils.sha_string('')
 
1646
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1647
 
 
1648
    def test_sha_mixed_endings(self):
 
1649
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1650
        self.build_tree_contents([('foo', text)])
 
1651
        expected_sha = osutils.sha_string(text)
 
1652
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1653
 
 
1654
 
 
1655
class TestResourceLoading(tests.TestCaseInTempDir):
1506
1656
 
1507
1657
    def test_resource_string(self):
1508
1658
        # test resource in bzrlib
1509
1659
        text = osutils.resource_string('bzrlib', 'debug.py')
1510
 
        self.assertEquals(_debug_text, text)
 
1660
        self.assertContainsRe(text, "debug_flags = set()")
1511
1661
        # test resource under bzrlib
1512
1662
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1513
1663
        self.assertContainsRe(text, "class TextUIFactory")
1516
1666
            'yyy.xx')
1517
1667
        # test unknown resource
1518
1668
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
 
1669
 
 
1670
 
 
1671
class TestReCompile(tests.TestCase):
 
1672
 
 
1673
    def test_re_compile_checked(self):
 
1674
        r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
 
1675
        self.assertTrue(r.match('aaaa'))
 
1676
        self.assertTrue(r.match('aAaA'))
 
1677
 
 
1678
    def test_re_compile_checked_error(self):
 
1679
        # like https://bugs.launchpad.net/bzr/+bug/251352
 
1680
        err = self.assertRaises(
 
1681
            errors.BzrCommandError,
 
1682
            osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
 
1683
        self.assertEqual(
 
1684
            "Invalid regular expression in test case: '*': "
 
1685
            "nothing to repeat",
 
1686
            str(err))
 
1687
 
 
1688
 
 
1689
class TestDirReader(tests.TestCaseInTempDir):
 
1690
 
 
1691
    # Set by load_tests
 
1692
    _dir_reader_class = None
 
1693
    _native_to_unicode = None
 
1694
 
 
1695
    def setUp(self):
 
1696
        tests.TestCaseInTempDir.setUp(self)
 
1697
        self.overrideAttr(osutils,
 
1698
                          '_selected_dir_reader', self._dir_reader_class())
 
1699
 
 
1700
    def _get_ascii_tree(self):
 
1701
        tree = [
 
1702
            '0file',
 
1703
            '1dir/',
 
1704
            '1dir/0file',
 
1705
            '1dir/1dir/',
 
1706
            '2file'
 
1707
            ]
 
1708
        expected_dirblocks = [
 
1709
                (('', '.'),
 
1710
                 [('0file', '0file', 'file'),
 
1711
                  ('1dir', '1dir', 'directory'),
 
1712
                  ('2file', '2file', 'file'),
 
1713
                 ]
 
1714
                ),
 
1715
                (('1dir', './1dir'),
 
1716
                 [('1dir/0file', '0file', 'file'),
 
1717
                  ('1dir/1dir', '1dir', 'directory'),
 
1718
                 ]
 
1719
                ),
 
1720
                (('1dir/1dir', './1dir/1dir'),
 
1721
                 [
 
1722
                 ]
 
1723
                ),
 
1724
            ]
 
1725
        return tree, expected_dirblocks
 
1726
 
 
1727
    def test_walk_cur_dir(self):
 
1728
        tree, expected_dirblocks = self._get_ascii_tree()
 
1729
        self.build_tree(tree)
 
1730
        result = list(osutils._walkdirs_utf8('.'))
 
1731
        # Filter out stat and abspath
 
1732
        self.assertEqual(expected_dirblocks,
 
1733
                         [(dirinfo, [line[0:3] for line in block])
 
1734
                          for dirinfo, block in result])
 
1735
 
 
1736
    def test_walk_sub_dir(self):
 
1737
        tree, expected_dirblocks = self._get_ascii_tree()
 
1738
        self.build_tree(tree)
 
1739
        # you can search a subdir only, with a supplied prefix.
 
1740
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
 
1741
        # Filter out stat and abspath
 
1742
        self.assertEqual(expected_dirblocks[1:],
 
1743
                         [(dirinfo, [line[0:3] for line in block])
 
1744
                          for dirinfo, block in result])
 
1745
 
 
1746
    def _get_unicode_tree(self):
 
1747
        name0u = u'0file-\xb6'
 
1748
        name1u = u'1dir-\u062c\u0648'
 
1749
        name2u = u'2file-\u0633'
 
1750
        tree = [
 
1751
            name0u,
 
1752
            name1u + '/',
 
1753
            name1u + '/' + name0u,
 
1754
            name1u + '/' + name1u + '/',
 
1755
            name2u,
 
1756
            ]
 
1757
        name0 = name0u.encode('UTF-8')
 
1758
        name1 = name1u.encode('UTF-8')
 
1759
        name2 = name2u.encode('UTF-8')
 
1760
        expected_dirblocks = [
 
1761
                (('', '.'),
 
1762
                 [(name0, name0, 'file', './' + name0u),
 
1763
                  (name1, name1, 'directory', './' + name1u),
 
1764
                  (name2, name2, 'file', './' + name2u),
 
1765
                 ]
 
1766
                ),
 
1767
                ((name1, './' + name1u),
 
1768
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1769
                                                        + '/' + name0u),
 
1770
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1771
                                                            + '/' + name1u),
 
1772
                 ]
 
1773
                ),
 
1774
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1775
                 [
 
1776
                 ]
 
1777
                ),
 
1778
            ]
 
1779
        return tree, expected_dirblocks
 
1780
 
 
1781
    def _filter_out(self, raw_dirblocks):
 
1782
        """Filter out a walkdirs_utf8 result.
 
1783
 
 
1784
        stat field is removed, all native paths are converted to unicode
 
1785
        """
 
1786
        filtered_dirblocks = []
 
1787
        for dirinfo, block in raw_dirblocks:
 
1788
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
 
1789
            details = []
 
1790
            for line in block:
 
1791
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
 
1792
            filtered_dirblocks.append((dirinfo, details))
 
1793
        return filtered_dirblocks
 
1794
 
 
1795
    def test_walk_unicode_tree(self):
 
1796
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1797
        tree, expected_dirblocks = self._get_unicode_tree()
 
1798
        self.build_tree(tree)
 
1799
        result = list(osutils._walkdirs_utf8('.'))
 
1800
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1801
 
 
1802
    def test_symlink(self):
 
1803
        self.requireFeature(tests.SymlinkFeature)
 
1804
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1805
        target = u'target\N{Euro Sign}'
 
1806
        link_name = u'l\N{Euro Sign}nk'
 
1807
        os.symlink(target, link_name)
 
1808
        target_utf8 = target.encode('UTF-8')
 
1809
        link_name_utf8 = link_name.encode('UTF-8')
 
1810
        expected_dirblocks = [
 
1811
                (('', '.'),
 
1812
                 [(link_name_utf8, link_name_utf8,
 
1813
                   'symlink', './' + link_name),],
 
1814
                 )]
 
1815
        result = list(osutils._walkdirs_utf8('.'))
 
1816
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1817
 
 
1818
 
 
1819
class TestReadLink(tests.TestCaseInTempDir):
 
1820
    """Exposes os.readlink() problems and the osutils solution.
 
1821
 
 
1822
    The only guarantee offered by os.readlink(), starting with 2.6, is that a
 
1823
    unicode string will be returned if a unicode string is passed.
 
1824
 
 
1825
    But prior python versions failed to properly encode the passed unicode
 
1826
    string.
 
1827
    """
 
1828
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
 
1829
 
 
1830
    def setUp(self):
 
1831
        super(tests.TestCaseInTempDir, self).setUp()
 
1832
        self.link = u'l\N{Euro Sign}ink'
 
1833
        self.target = u'targe\N{Euro Sign}t'
 
1834
        os.symlink(self.target, self.link)
 
1835
 
 
1836
    def test_os_readlink_link_encoding(self):
 
1837
        if sys.version_info < (2, 6):
 
1838
            self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
 
1839
        else:
 
1840
            self.assertEquals(self.target,  os.readlink(self.link))
 
1841
 
 
1842
    def test_os_readlink_link_decoding(self):
 
1843
        self.assertEquals(self.target.encode(osutils._fs_enc),
 
1844
                          os.readlink(self.link.encode(osutils._fs_enc)))
 
1845
 
 
1846
 
 
1847
class TestConcurrency(tests.TestCase):
 
1848
 
 
1849
    def setUp(self):
 
1850
        super(TestConcurrency, self).setUp()
 
1851
        self.overrideAttr(osutils, '_cached_local_concurrency')
 
1852
 
 
1853
    def test_local_concurrency(self):
 
1854
        concurrency = osutils.local_concurrency()
 
1855
        self.assertIsInstance(concurrency, int)
 
1856
 
 
1857
    def test_local_concurrency_environment_variable(self):
 
1858
        os.environ['BZR_CONCURRENCY'] = '2'
 
1859
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
 
1860
        os.environ['BZR_CONCURRENCY'] = '3'
 
1861
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
 
1862
        os.environ['BZR_CONCURRENCY'] = 'foo'
 
1863
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
 
1864
 
 
1865
    def test_option_concurrency(self):
 
1866
        os.environ['BZR_CONCURRENCY'] = '1'
 
1867
        self.run_bzr('rocks --concurrency 42')
 
1868
        # Command line overrides envrionment variable
 
1869
        self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
 
1870
        self.assertEquals(42, osutils.local_concurrency(use_cache=False))
 
1871
 
 
1872
 
 
1873
class TestFailedToLoadExtension(tests.TestCase):
 
1874
 
 
1875
    def _try_loading(self):
 
1876
        try:
 
1877
            import bzrlib._fictional_extension_py
 
1878
        except ImportError, e:
 
1879
            osutils.failed_to_load_extension(e)
 
1880
            return True
 
1881
 
 
1882
    def setUp(self):
 
1883
        super(TestFailedToLoadExtension, self).setUp()
 
1884
        self.overrideAttr(osutils, '_extension_load_failures', [])
 
1885
 
 
1886
    def test_failure_to_load(self):
 
1887
        self._try_loading()
 
1888
        self.assertLength(1, osutils._extension_load_failures)
 
1889
        self.assertEquals(osutils._extension_load_failures[0],
 
1890
            "No module named _fictional_extension_py")
 
1891
 
 
1892
    def test_report_extension_load_failures_no_warning(self):
 
1893
        self.assertTrue(self._try_loading())
 
1894
        warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
 
1895
        # it used to give a Python warning; it no longer does
 
1896
        self.assertLength(0, warnings)
 
1897
 
 
1898
    def test_report_extension_load_failures_message(self):
 
1899
        log = StringIO()
 
1900
        trace.push_log_file(log)
 
1901
        self.assertTrue(self._try_loading())
 
1902
        osutils.report_extension_load_failures()
 
1903
        self.assertContainsRe(
 
1904
            log.getvalue(),
 
1905
            r"bzr: warning: some compiled extensions could not be loaded; "
 
1906
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
 
1907
            )
 
1908
 
 
1909
 
 
1910
class TestTerminalWidth(tests.TestCase):
 
1911
 
 
1912
    def replace_stdout(self, new):
 
1913
        self.overrideAttr(sys, 'stdout', new)
 
1914
 
 
1915
    def replace__terminal_size(self, new):
 
1916
        self.overrideAttr(osutils, '_terminal_size', new)
 
1917
 
 
1918
    def set_fake_tty(self):
 
1919
 
 
1920
        class I_am_a_tty(object):
 
1921
            def isatty(self):
 
1922
                return True
 
1923
 
 
1924
        self.replace_stdout(I_am_a_tty())
 
1925
 
 
1926
    def test_default_values(self):
 
1927
        self.assertEqual(80, osutils.default_terminal_width)
 
1928
 
 
1929
    def test_defaults_to_BZR_COLUMNS(self):
 
1930
        # BZR_COLUMNS is set by the test framework
 
1931
        self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
 
1932
        os.environ['BZR_COLUMNS'] = '12'
 
1933
        self.assertEqual(12, osutils.terminal_width())
 
1934
 
 
1935
    def test_falls_back_to_COLUMNS(self):
 
1936
        del os.environ['BZR_COLUMNS']
 
1937
        self.assertNotEqual('42', os.environ['COLUMNS'])
 
1938
        self.set_fake_tty()
 
1939
        os.environ['COLUMNS'] = '42'
 
1940
        self.assertEqual(42, osutils.terminal_width())
 
1941
 
 
1942
    def test_tty_default_without_columns(self):
 
1943
        del os.environ['BZR_COLUMNS']
 
1944
        del os.environ['COLUMNS']
 
1945
 
 
1946
        def terminal_size(w, h):
 
1947
            return 42, 42
 
1948
 
 
1949
        self.set_fake_tty()
 
1950
        # We need to override the osutils definition as it depends on the
 
1951
        # running environment that we can't control (PQM running without a
 
1952
        # controlling terminal is one example).
 
1953
        self.replace__terminal_size(terminal_size)
 
1954
        self.assertEqual(42, osutils.terminal_width())
 
1955
 
 
1956
    def test_non_tty_default_without_columns(self):
 
1957
        del os.environ['BZR_COLUMNS']
 
1958
        del os.environ['COLUMNS']
 
1959
        self.replace_stdout(None)
 
1960
        self.assertEqual(None, osutils.terminal_width())
 
1961
 
 
1962
    def test_no_TIOCGWINSZ(self):
 
1963
        self.requireFeature(term_ios_feature)
 
1964
        termios = term_ios_feature.module
 
1965
        # bug 63539 is about a termios without TIOCGWINSZ attribute
 
1966
        try:
 
1967
            orig = termios.TIOCGWINSZ
 
1968
        except AttributeError:
 
1969
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
 
1970
            pass
 
1971
        else:
 
1972
            self.overrideAttr(termios, 'TIOCGWINSZ')
 
1973
            del termios.TIOCGWINSZ
 
1974
        del os.environ['BZR_COLUMNS']
 
1975
        del os.environ['COLUMNS']
 
1976
        # Whatever the result is, if we don't raise an exception, it's ok.
 
1977
        osutils.terminal_width()