~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: John Arbash Meinel
  • Date: 2008-08-25 21:50:11 UTC
  • mfrom: (0.11.3 tools)
  • mto: This revision was merged to the branch mainline in revision 3659.
  • Revision ID: john@arbash-meinel.com-20080825215011-de9esmzgkue3e522
Merge in Lukáš's helper scripts.
Update the packaging documents to describe how to do the releases
using bzr-builddeb to package all distro platforms
simultaneously.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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
23
22
import socket
24
23
import stat
25
24
import sys
26
25
import time
27
26
 
 
27
import bzrlib
28
28
from bzrlib import (
29
29
    errors,
30
30
    osutils,
31
31
    tests,
32
 
    trace,
33
32
    win32utils,
34
33
    )
 
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
        )
35
41
from bzrlib.tests import (
36
 
    file_utils,
37
 
    test__walkdirs_win32,
 
42
        probe_unicode_in_user_encoding,
 
43
        StringIOWrapper,
 
44
        SymlinkFeature,
 
45
        TestCase,
 
46
        TestCaseInTempDir,
 
47
        TestSkipped,
 
48
        )
 
49
from bzrlib.tests.file_utils import (
 
50
    FakeReadFile,
38
51
    )
39
 
 
40
 
 
41
 
class _UTF8DirReaderFeature(tests.Feature):
42
 
 
43
 
    def _probe(self):
44
 
        try:
45
 
            from bzrlib import _readdir_pyx
46
 
            self.reader = _readdir_pyx.UTF8DirReader
47
 
            return True
48
 
        except ImportError:
49
 
            return False
50
 
 
51
 
    def feature_name(self):
52
 
        return 'bzrlib._readdir_pyx'
53
 
 
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):
 
52
from bzrlib.tests.test__walkdirs_win32 import WalkdirsWin32Feature
 
53
 
 
54
 
 
55
class TestOSUtils(TestCaseInTempDir):
107
56
 
108
57
    def test_contains_whitespace(self):
109
58
        self.failUnless(osutils.contains_whitespace(u' '))
119
68
        self.failIf(osutils.contains_whitespace(u'hellothere'))
120
69
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
121
70
 
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
 
 
136
71
    def test_fancy_rename(self):
137
72
        # This should work everywhere
138
 
        self.create_file('a', 'something in a\n')
139
 
        self._fancy_rename('a', 'b')
 
73
        def rename(a, b):
 
74
            osutils.fancy_rename(a, b,
 
75
                    rename_func=os.rename,
 
76
                    unlink_func=os.unlink)
 
77
 
 
78
        open('a', 'wb').write('something in a\n')
 
79
        rename('a', 'b')
140
80
        self.failIfExists('a')
141
81
        self.failUnlessExists('b')
142
82
        self.check_file_contents('b', 'something in a\n')
143
83
 
144
 
        self.create_file('a', 'new something in a\n')
145
 
        self._fancy_rename('b', 'a')
 
84
        open('a', 'wb').write('new something in a\n')
 
85
        rename('b', 'a')
146
86
 
147
87
        self.check_file_contents('a', 'something in a\n')
148
88
 
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
 
 
161
89
    def test_rename(self):
162
90
        # Rename should be semi-atomic on all platforms
163
 
        self.create_file('a', 'something in a\n')
 
91
        open('a', 'wb').write('something in a\n')
164
92
        osutils.rename('a', 'b')
165
93
        self.failIfExists('a')
166
94
        self.failUnlessExists('b')
167
95
        self.check_file_contents('b', 'something in a\n')
168
96
 
169
 
        self.create_file('a', 'new something in a\n')
 
97
        open('a', 'wb').write('new something in a\n')
170
98
        osutils.rename('b', 'a')
171
99
 
172
100
        self.check_file_contents('a', 'something in a\n')
183
111
        shape = sorted(os.listdir('.'))
184
112
        self.assertEquals(['A', 'B'], shape)
185
113
 
186
 
 
187
 
class TestRandChars(tests.TestCase):
188
 
 
189
114
    def test_01_rand_chars_empty(self):
190
115
        result = osutils.rand_chars(0)
191
116
        self.assertEqual(result, '')
196
121
        self.assertEqual(type(result), str)
197
122
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
198
123
 
199
 
 
200
 
class TestIsInside(tests.TestCase):
201
 
 
202
124
    def test_is_inside(self):
203
125
        is_inside = osutils.is_inside
204
126
        self.assertTrue(is_inside('src', 'src/foo.c'))
209
131
        self.assertTrue(is_inside('', 'foo.c'))
210
132
 
211
133
    def test_is_inside_any(self):
212
 
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
 
134
        SRC_FOO_C = pathjoin('src', 'foo.c')
213
135
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
214
136
                         (['src'], SRC_FOO_C),
215
137
                         (['src'], 'src'),
216
138
                         ]:
217
 
            self.assert_(osutils.is_inside_any(dirs, fn))
 
139
            self.assert_(is_inside_any(dirs, fn))
218
140
        for dirs, fn in [(['src'], 'srccontrol'),
219
141
                         (['src'], 'srccontrol/foo')]:
220
 
            self.assertFalse(osutils.is_inside_any(dirs, fn))
 
142
            self.assertFalse(is_inside_any(dirs, fn))
221
143
 
222
144
    def test_is_inside_or_parent_of_any(self):
223
145
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
226
148
                         (['src/bar.c', 'bla/foo.c'], 'src'),
227
149
                         (['src'], 'src'),
228
150
                         ]:
229
 
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
230
 
 
 
151
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
 
152
            
231
153
        for dirs, fn in [(['src'], 'srccontrol'),
232
154
                         (['srccontrol/foo.c'], 'src'),
233
155
                         (['src'], 'srccontrol/foo')]:
234
 
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
235
 
 
236
 
 
237
 
class TestRmTree(tests.TestCaseInTempDir):
 
156
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
238
157
 
239
158
    def test_rmtree(self):
240
159
        # Check to remove tree with read-only files/dirs
254
173
        self.failIfExists('dir/file')
255
174
        self.failIfExists('dir')
256
175
 
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
 
 
272
176
    def test_file_kind(self):
273
177
        self.build_tree(['file', 'dir/'])
274
178
        self.assertEquals('file', osutils.file_kind('file'))
276
180
        if osutils.has_symlinks():
277
181
            os.symlink('symlink', 'symlink')
278
182
            self.assertEquals('symlink', osutils.file_kind('symlink'))
279
 
 
 
183
        
280
184
        # TODO: jam 20060529 Test a block device
281
185
        try:
282
186
            os.lstat('/dev/null')
304
208
                os.remove('socket')
305
209
 
306
210
    def test_kind_marker(self):
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):
 
211
        self.assertEqual(osutils.kind_marker('file'), '')
 
212
        self.assertEqual(osutils.kind_marker('directory'), '/')
 
213
        self.assertEqual(osutils.kind_marker('symlink'), '@')
 
214
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
318
215
 
319
216
    def test_get_umask(self):
320
217
        if sys.platform == 'win32':
323
220
            return
324
221
 
325
222
        orig_umask = osutils.get_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):
 
223
        try:
 
224
            os.umask(0222)
 
225
            self.assertEqual(0222, osutils.get_umask())
 
226
            os.umask(0022)
 
227
            self.assertEqual(0022, osutils.get_umask())
 
228
            os.umask(0002)
 
229
            self.assertEqual(0002, osutils.get_umask())
 
230
            os.umask(0027)
 
231
            self.assertEqual(0027, osutils.get_umask())
 
232
        finally:
 
233
            os.umask(orig_umask)
338
234
 
339
235
    def assertFormatedDelta(self, expected, seconds):
340
236
        """Assert osutils.format_delta formats as expected"""
375
271
    def test_format_date(self):
376
272
        self.assertRaises(errors.UnsupportedTimezoneFormat,
377
273
            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):
414
274
 
415
275
    def test_dereference_path(self):
416
 
        self.requireFeature(tests.SymlinkFeature)
 
276
        self.requireFeature(SymlinkFeature)
417
277
        cwd = osutils.realpath('.')
418
278
        os.mkdir('bar')
419
279
        bar_path = osutils.pathjoin(cwd, 'bar')
422
282
        self.assertEqual(bar_path, osutils.realpath('./bar'))
423
283
        os.symlink('bar', 'foo')
424
284
        self.assertEqual(bar_path, osutils.realpath('./foo'))
425
 
 
 
285
        
426
286
        # Does not dereference terminal symlinks
427
287
        foo_path = osutils.pathjoin(cwd, 'foo')
428
288
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
460
320
            osutils.make_readonly('dangling')
461
321
            osutils.make_writable('dangling')
462
322
 
 
323
    def test_kind_marker(self):
 
324
        self.assertEqual("", osutils.kind_marker("file"))
 
325
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
326
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
327
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
 
328
 
463
329
    def test_host_os_dereferences_symlinks(self):
464
330
        osutils.host_os_dereferences_symlinks()
465
331
 
466
332
 
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):
 
333
class TestPumpFile(TestCase):
528
334
    """Test pumpfile method."""
529
 
 
530
335
    def setUp(self):
531
 
        tests.TestCase.setUp(self)
532
336
        # create a test datablock
533
337
        self.block_size = 512
534
338
        pattern = '0123456789ABCDEF'
541
345
        # make sure test data is larger than max read size
542
346
        self.assertTrue(self.test_data_len > self.block_size)
543
347
 
544
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
348
        from_file = FakeReadFile(self.test_data)
545
349
        to_file = StringIO()
546
350
 
547
351
        # read (max / 2) bytes and verify read size wasn't affected
548
352
        num_bytes_to_read = self.block_size / 2
549
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
353
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
550
354
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
551
355
        self.assertEqual(from_file.get_read_count(), 1)
552
356
 
553
357
        # read (max) bytes and verify read size wasn't affected
554
358
        num_bytes_to_read = self.block_size
555
359
        from_file.reset_read_count()
556
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
360
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
557
361
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
558
362
        self.assertEqual(from_file.get_read_count(), 1)
559
363
 
560
364
        # read (max + 1) bytes and verify read size was limited
561
365
        num_bytes_to_read = self.block_size + 1
562
366
        from_file.reset_read_count()
563
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
367
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
564
368
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
565
369
        self.assertEqual(from_file.get_read_count(), 2)
566
370
 
567
371
        # finish reading the rest of the data
568
372
        num_bytes_to_read = self.test_data_len - to_file.tell()
569
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
373
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
570
374
 
571
375
        # report error if the data wasn't equal (we only report the size due
572
376
        # to the length of the data)
582
386
        self.assertTrue(self.test_data_len > self.block_size)
583
387
 
584
388
        # retrieve data in blocks
585
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
389
        from_file = FakeReadFile(self.test_data)
586
390
        to_file = StringIO()
587
 
        osutils.pumpfile(from_file, to_file, self.test_data_len,
588
 
                         self.block_size)
 
391
        pumpfile(from_file, to_file, self.test_data_len, self.block_size)
589
392
 
590
393
        # verify read size was equal to the maximum read size
591
394
        self.assertTrue(from_file.get_max_read_size() > 0)
606
409
        self.assertTrue(self.test_data_len > self.block_size)
607
410
 
608
411
        # retrieve data to EOF
609
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
412
        from_file = FakeReadFile(self.test_data)
610
413
        to_file = StringIO()
611
 
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
 
414
        pumpfile(from_file, to_file, -1, self.block_size)
612
415
 
613
416
        # verify read size was equal to the maximum read size
614
417
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
626
429
        test verifies that any existing usages of pumpfile will not be broken
627
430
        with this new version."""
628
431
        # retrieve data using default (old) pumpfile method
629
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
432
        from_file = FakeReadFile(self.test_data)
630
433
        to_file = StringIO()
631
 
        osutils.pumpfile(from_file, to_file)
 
434
        pumpfile(from_file, to_file)
632
435
 
633
436
        # report error if the data wasn't equal (we only report the size due
634
437
        # to the length of the data)
637
440
            message = "Data not equal.  Expected %d bytes, received %d."
638
441
            self.fail(message % (len(response_data), self.test_data_len))
639
442
 
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):
670
 
 
671
 
    def test_empty(self):
672
 
        output = StringIO()
673
 
        osutils.pump_string_file("", output)
674
 
        self.assertEqual("", output.getvalue())
675
 
 
676
 
    def test_more_than_segment_size(self):
677
 
        output = StringIO()
678
 
        osutils.pump_string_file("123456789", output, 2)
679
 
        self.assertEqual("123456789", output.getvalue())
680
 
 
681
 
    def test_segment_size(self):
682
 
        output = StringIO()
683
 
        osutils.pump_string_file("12", output, 2)
684
 
        self.assertEqual("12", output.getvalue())
685
 
 
686
 
    def test_segment_size_multiple(self):
687
 
        output = StringIO()
688
 
        osutils.pump_string_file("1234", output, 2)
689
 
        self.assertEqual("1234", output.getvalue())
690
 
 
691
 
 
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):
 
443
class TestSafeUnicode(TestCase):
712
444
 
713
445
    def test_from_ascii_string(self):
714
446
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
723
455
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
724
456
 
725
457
    def test_bad_utf8_string(self):
726
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
458
        self.assertRaises(BzrBadParameterNotUnicode,
727
459
                          osutils.safe_unicode,
728
460
                          '\xbb\xbb')
729
461
 
730
462
 
731
 
class TestSafeUtf8(tests.TestCase):
 
463
class TestSafeUtf8(TestCase):
732
464
 
733
465
    def test_from_ascii_string(self):
734
466
        f = 'foobar'
744
476
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
745
477
 
746
478
    def test_bad_utf8_string(self):
747
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
479
        self.assertRaises(BzrBadParameterNotUnicode,
748
480
                          osutils.safe_utf8, '\xbb\xbb')
749
481
 
750
482
 
751
 
class TestSafeRevisionId(tests.TestCase):
 
483
class TestSafeRevisionId(TestCase):
752
484
 
753
485
    def test_from_ascii_string(self):
754
486
        # this shouldn't give a warning because it's getting an ascii string
776
508
        self.assertEqual(None, osutils.safe_revision_id(None))
777
509
 
778
510
 
779
 
class TestSafeFileId(tests.TestCase):
 
511
class TestSafeFileId(TestCase):
780
512
 
781
513
    def test_from_ascii_string(self):
782
514
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
802
534
        self.assertEqual(None, osutils.safe_file_id(None))
803
535
 
804
536
 
805
 
class TestWin32Funcs(tests.TestCase):
806
 
    """Test that _win32 versions of os utilities return appropriate paths."""
 
537
class TestWin32Funcs(TestCase):
 
538
    """Test that the _win32 versions of os utilities return appropriate paths."""
807
539
 
808
540
    def test_abspath(self):
809
541
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
816
548
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
817
549
 
818
550
    def test_pathjoin(self):
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'))
 
551
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
 
552
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
553
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
 
554
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
 
555
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
556
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
831
557
 
832
558
    def test_normpath(self):
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'))
 
559
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
560
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
837
561
 
838
562
    def test_getcwd(self):
839
563
        cwd = osutils._win32_getcwd()
868
592
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
869
593
 
870
594
 
871
 
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
 
595
class TestWin32FuncsDirs(TestCaseInTempDir):
872
596
    """Test win32 functions that create files."""
 
597
    
 
598
    def test_getcwd(self):
 
599
        if win32utils.winver == 'Windows 98':
 
600
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
 
601
        # Make sure getcwd can handle unicode filenames
 
602
        try:
 
603
            os.mkdir(u'mu-\xb5')
 
604
        except UnicodeError:
 
605
            raise TestSkipped("Unable to create Unicode filename")
873
606
 
874
 
    def test_getcwd(self):
875
 
        self.requireFeature(tests.UnicodeFilenameFeature)
876
 
        os.mkdir(u'mu-\xb5')
877
607
        os.chdir(u'mu-\xb5')
878
608
        # TODO: jam 20060427 This will probably fail on Mac OSX because
879
609
        #       it will change the normalization of B\xe5gfors
884
614
    def test_minimum_path_selection(self):
885
615
        self.assertEqual(set(),
886
616
            osutils.minimum_path_selection([]))
887
 
        self.assertEqual(set(['a']),
888
 
            osutils.minimum_path_selection(['a']))
889
617
        self.assertEqual(set(['a', 'b']),
890
618
            osutils.minimum_path_selection(['a', 'b']))
891
619
        self.assertEqual(set(['a/', 'b']),
892
620
            osutils.minimum_path_selection(['a/', 'b']))
893
621
        self.assertEqual(set(['a/', 'b']),
894
622
            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']))
897
623
 
898
624
    def test_mkdtemp(self):
899
625
        tmpdir = osutils._win32_mkdtemp(dir='.')
955
681
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
956
682
 
957
683
 
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):
 
684
class TestMacFuncsDirs(TestCaseInTempDir):
968
685
    """Test mac special functions that require directories."""
969
686
 
970
687
    def test_getcwd(self):
971
 
        self.requireFeature(tests.UnicodeFilenameFeature)
972
 
        os.mkdir(u'B\xe5gfors')
 
688
        # On Mac, this will actually create Ba\u030agfors
 
689
        # but chdir will still work, because it accepts both paths
 
690
        try:
 
691
            os.mkdir(u'B\xe5gfors')
 
692
        except UnicodeError:
 
693
            raise TestSkipped("Unable to create Unicode filename")
 
694
 
973
695
        os.chdir(u'B\xe5gfors')
974
696
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
975
697
 
976
698
    def test_getcwd_nonnorm(self):
977
 
        self.requireFeature(tests.UnicodeFilenameFeature)
978
699
        # Test that _mac_getcwd() will normalize this path
979
 
        os.mkdir(u'Ba\u030agfors')
 
700
        try:
 
701
            os.mkdir(u'Ba\u030agfors')
 
702
        except UnicodeError:
 
703
            raise TestSkipped("Unable to create Unicode filename")
 
704
 
980
705
        os.chdir(u'Ba\u030agfors')
981
706
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
982
707
 
983
708
 
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):
 
709
class TestSplitLines(TestCase):
1002
710
 
1003
711
    def test_split_unicode(self):
1004
712
        self.assertEqual([u'foo\n', u'bar\xae'],
1011
719
                         osutils.split_lines('foo\rbar\n'))
1012
720
 
1013
721
 
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])
 
722
class TestWalkDirs(TestCaseInTempDir):
1020
723
 
1021
724
    def test_walkdirs(self):
1022
725
        tree = [
1055
758
            result.append((dirdetail, dirblock))
1056
759
 
1057
760
        self.assertTrue(found_bzrdir)
1058
 
        self.assertExpectedBlocks(expected_dirblocks, result)
 
761
        self.assertEqual(expected_dirblocks,
 
762
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1059
763
        # you can search a subdir only, with a supplied prefix.
1060
764
        result = []
1061
765
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1062
766
            result.append(dirblock)
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")
 
767
        self.assertEqual(expected_dirblocks[1:],
 
768
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1084
769
 
1085
770
    def test__walkdirs_utf8(self):
1086
771
        tree = [
1119
804
            result.append((dirdetail, dirblock))
1120
805
 
1121
806
        self.assertTrue(found_bzrdir)
1122
 
        self.assertExpectedBlocks(expected_dirblocks, result)
1123
 
 
 
807
        self.assertEqual(expected_dirblocks,
 
808
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1124
809
        # you can search a subdir only, with a supplied prefix.
1125
810
        result = []
1126
811
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1127
812
            result.append(dirblock)
1128
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
813
        self.assertEqual(expected_dirblocks[1:],
 
814
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1129
815
 
1130
816
    def _filter_out_stat(self, result):
1131
817
        """Filter out the stat value from the walkdirs result"""
1136
822
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1137
823
            dirblock[:] = new_dirblock
1138
824
 
 
825
    def test__walkdirs_utf8_selection(self):
 
826
        # Just trigger the function once, to make sure it has selected a real
 
827
        # implementation.
 
828
        list(osutils._walkdirs_utf8('.'))
 
829
        if WalkdirsWin32Feature.available():
 
830
            # If the compiled form is available, make sure it is used
 
831
            from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
832
            self.assertIs(_walkdirs_utf8_win32_find_file,
 
833
                          osutils._real_walkdirs_utf8)
 
834
        elif sys.platform == 'win32':
 
835
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
 
836
                          osutils._real_walkdirs_utf8)
 
837
        elif osutils._fs_enc.upper() in ('UTF-8', 'ASCII', 'ANSI_X3.4-1968'): # ascii
 
838
            self.assertIs(osutils._walkdirs_fs_utf8,
 
839
                          osutils._real_walkdirs_utf8)
 
840
        else:
 
841
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
 
842
                          osutils._real_walkdirs_utf8)
 
843
 
1139
844
    def _save_platform_info(self):
1140
 
        self.overrideAttr(win32utils, 'winver')
1141
 
        self.overrideAttr(osutils, '_fs_enc')
1142
 
        self.overrideAttr(osutils, '_selected_dir_reader')
 
845
        cur_winver = win32utils.winver
 
846
        cur_fs_enc = osutils._fs_enc
 
847
        cur_real_walkdirs_utf8 = osutils._real_walkdirs_utf8
 
848
        def restore():
 
849
            win32utils.winver = cur_winver
 
850
            osutils._fs_enc = cur_fs_enc
 
851
            osutils._real_walkdirs_utf8 = cur_real_walkdirs_utf8
 
852
        self.addCleanup(restore)
1143
853
 
1144
 
    def assertDirReaderIs(self, expected):
 
854
    def assertWalkdirsUtf8Is(self, expected):
1145
855
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1146
856
        # Force it to redetect
1147
 
        osutils._selected_dir_reader = None
 
857
        osutils._real_walkdirs_utf8 = None
1148
858
        # Nothing to list, but should still trigger the selection logic
1149
859
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1150
 
        self.assertIsInstance(osutils._selected_dir_reader, expected)
 
860
        self.assertIs(expected, osutils._real_walkdirs_utf8)
1151
861
 
1152
862
    def test_force_walkdirs_utf8_fs_utf8(self):
1153
 
        self.requireFeature(UTF8DirReaderFeature)
1154
863
        self._save_platform_info()
1155
864
        win32utils.winver = None # Avoid the win32 detection code
1156
865
        osutils._fs_enc = 'UTF-8'
1157
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
866
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1158
867
 
1159
868
    def test_force_walkdirs_utf8_fs_ascii(self):
1160
 
        self.requireFeature(UTF8DirReaderFeature)
1161
869
        self._save_platform_info()
1162
870
        win32utils.winver = None # Avoid the win32 detection code
1163
871
        osutils._fs_enc = 'US-ASCII'
1164
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
872
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1165
873
 
1166
874
    def test_force_walkdirs_utf8_fs_ANSI(self):
1167
 
        self.requireFeature(UTF8DirReaderFeature)
1168
875
        self._save_platform_info()
1169
876
        win32utils.winver = None # Avoid the win32 detection code
1170
877
        osutils._fs_enc = 'ANSI_X3.4-1968'
1171
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
878
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
1172
879
 
1173
880
    def test_force_walkdirs_utf8_fs_latin1(self):
1174
881
        self._save_platform_info()
1175
882
        win32utils.winver = None # Avoid the win32 detection code
1176
883
        osutils._fs_enc = 'latin1'
1177
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
884
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
1178
885
 
1179
886
    def test_force_walkdirs_utf8_nt(self):
1180
 
        # Disabled because the thunk of the whole walkdirs api is disabled.
1181
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
887
        self.requireFeature(WalkdirsWin32Feature)
1182
888
        self._save_platform_info()
1183
889
        win32utils.winver = 'Windows NT'
1184
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1185
 
        self.assertDirReaderIs(Win32ReadDir)
 
890
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
891
        self.assertWalkdirsUtf8Is(_walkdirs_utf8_win32_find_file)
1186
892
 
1187
 
    def test_force_walkdirs_utf8_98(self):
1188
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
893
    def test_force_walkdirs_utf8_nt(self):
 
894
        self.requireFeature(WalkdirsWin32Feature)
1189
895
        self._save_platform_info()
1190
896
        win32utils.winver = 'Windows 98'
1191
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
897
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
1192
898
 
1193
899
    def test_unicode_walkdirs(self):
1194
900
        """Walkdirs should always return unicode paths."""
1195
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1196
901
        name0 = u'0file-\xb6'
1197
902
        name1 = u'1dir-\u062c\u0648'
1198
903
        name2 = u'2file-\u0633'
1203
908
            name1 + '/' + name1 + '/',
1204
909
            name2,
1205
910
            ]
1206
 
        self.build_tree(tree)
 
911
        try:
 
912
            self.build_tree(tree)
 
913
        except UnicodeError:
 
914
            raise TestSkipped('Could not represent Unicode chars'
 
915
                              ' in current encoding.')
1207
916
        expected_dirblocks = [
1208
917
                ((u'', u'.'),
1209
918
                 [(name0, name0, 'file', './' + name0),
1235
944
 
1236
945
        The abspath portion might be in unicode or utf-8
1237
946
        """
1238
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1239
947
        name0 = u'0file-\xb6'
1240
948
        name1 = u'1dir-\u062c\u0648'
1241
949
        name2 = u'2file-\u0633'
1246
954
            name1 + '/' + name1 + '/',
1247
955
            name2,
1248
956
            ]
1249
 
        self.build_tree(tree)
 
957
        try:
 
958
            self.build_tree(tree)
 
959
        except UnicodeError:
 
960
            raise TestSkipped('Could not represent Unicode chars'
 
961
                              ' in current encoding.')
1250
962
        name0 = name0.encode('utf8')
1251
963
        name1 = name1.encode('utf8')
1252
964
        name2 = name2.encode('utf8')
1291
1003
            result.append((dirdetail, new_dirblock))
1292
1004
        self.assertEqual(expected_dirblocks, result)
1293
1005
 
1294
 
    def test__walkdirs_utf8_with_unicode_fs(self):
1295
 
        """UnicodeDirReader should be a safe fallback everywhere
 
1006
    def test_unicode__walkdirs_unicode_to_utf8(self):
 
1007
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
1296
1008
 
1297
1009
        The abspath portion should be in unicode
1298
1010
        """
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()
1304
1011
        name0u = u'0file-\xb6'
1305
1012
        name1u = u'1dir-\u062c\u0648'
1306
1013
        name2u = u'2file-\u0633'
1311
1018
            name1u + '/' + name1u + '/',
1312
1019
            name2u,
1313
1020
            ]
1314
 
        self.build_tree(tree)
 
1021
        try:
 
1022
            self.build_tree(tree)
 
1023
        except UnicodeError:
 
1024
            raise TestSkipped('Could not represent Unicode chars'
 
1025
                              ' in current encoding.')
1315
1026
        name0 = name0u.encode('utf8')
1316
1027
        name1 = name1u.encode('utf8')
1317
1028
        name2 = name2u.encode('utf8')
1337
1048
                 ]
1338
1049
                ),
1339
1050
            ]
1340
 
        result = list(osutils._walkdirs_utf8('.'))
 
1051
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
1341
1052
        self._filter_out_stat(result)
1342
1053
        self.assertEqual(expected_dirblocks, result)
1343
1054
 
1344
 
    def test__walkdirs_utf8_win32readdir(self):
1345
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1055
    def test__walkdirs_utf_win32_find_file(self):
 
1056
        self.requireFeature(WalkdirsWin32Feature)
1346
1057
        self.requireFeature(tests.UnicodeFilenameFeature)
1347
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1348
 
        self._save_platform_info()
1349
 
        osutils._selected_dir_reader = Win32ReadDir()
 
1058
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
1350
1059
        name0u = u'0file-\xb6'
1351
1060
        name1u = u'1dir-\u062c\u0648'
1352
1061
        name2u = u'2file-\u0633'
1383
1092
                 ]
1384
1093
                ),
1385
1094
            ]
1386
 
        result = list(osutils._walkdirs_utf8(u'.'))
 
1095
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
1387
1096
        self._filter_out_stat(result)
1388
1097
        self.assertEqual(expected_dirblocks, result)
1389
1098
 
1399
1108
 
1400
1109
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1401
1110
        """make sure our Stat values are valid"""
1402
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1111
        self.requireFeature(WalkdirsWin32Feature)
1403
1112
        self.requireFeature(tests.UnicodeFilenameFeature)
1404
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1113
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
1405
1114
        name0u = u'0file-\xb6'
1406
1115
        name0 = name0u.encode('utf8')
1407
1116
        self.build_tree([name0u])
1414
1123
        finally:
1415
1124
            f.close()
1416
1125
 
1417
 
        result = Win32ReadDir().read_dir('', u'.')
1418
 
        entry = result[0]
 
1126
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1127
        entry = result[0][1][0]
1419
1128
        self.assertEqual((name0, name0, 'file'), entry[:3])
1420
1129
        self.assertEqual(u'./' + name0u, entry[4])
1421
1130
        self.assertStatIsCorrect(entry[4], entry[3])
1423
1132
 
1424
1133
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1425
1134
        """make sure our Stat values are valid"""
1426
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1135
        self.requireFeature(WalkdirsWin32Feature)
1427
1136
        self.requireFeature(tests.UnicodeFilenameFeature)
1428
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1137
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
1429
1138
        name0u = u'0dir-\u062c\u0648'
1430
1139
        name0 = name0u.encode('utf8')
1431
1140
        self.build_tree([name0u + '/'])
1432
1141
 
1433
 
        result = Win32ReadDir().read_dir('', u'.')
1434
 
        entry = result[0]
 
1142
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1143
        entry = result[0][1][0]
1435
1144
        self.assertEqual((name0, name0, 'directory'), entry[:3])
1436
1145
        self.assertEqual(u'./' + name0u, entry[4])
1437
1146
        self.assertStatIsCorrect(entry[4], entry[3])
1514
1223
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1515
1224
 
1516
1225
 
1517
 
class TestCopyTree(tests.TestCaseInTempDir):
1518
 
 
 
1226
class TestCopyTree(TestCaseInTempDir):
 
1227
    
1519
1228
    def test_copy_basic_tree(self):
1520
1229
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1521
1230
        osutils.copy_tree('source', 'target')
1530
1239
        self.assertEqual(['c'], os.listdir('target/b'))
1531
1240
 
1532
1241
    def test_copy_tree_symlinks(self):
1533
 
        self.requireFeature(tests.SymlinkFeature)
 
1242
        self.requireFeature(SymlinkFeature)
1534
1243
        self.build_tree(['source/'])
1535
1244
        os.symlink('a/generic/path', 'source/lnk')
1536
1245
        osutils.copy_tree('source', 'target')
1566
1275
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1567
1276
 
1568
1277
 
1569
 
class TestSetUnsetEnv(tests.TestCase):
 
1278
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
 
1279
# [bialix] 2006/12/26
 
1280
 
 
1281
 
 
1282
class TestSetUnsetEnv(TestCase):
1570
1283
    """Test updating the environment"""
1571
1284
 
1572
1285
    def setUp(self):
1578
1291
        def cleanup():
1579
1292
            if 'BZR_TEST_ENV_VAR' in os.environ:
1580
1293
                del os.environ['BZR_TEST_ENV_VAR']
 
1294
 
1581
1295
        self.addCleanup(cleanup)
1582
1296
 
1583
1297
    def test_set(self):
1595
1309
 
1596
1310
    def test_unicode(self):
1597
1311
        """Environment can only contain plain strings
1598
 
 
 
1312
        
1599
1313
        So Unicode strings must be encoded.
1600
1314
        """
1601
 
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
1315
        uni_val, env_val = probe_unicode_in_user_encoding()
1602
1316
        if uni_val is None:
1603
 
            raise tests.TestSkipped(
1604
 
                'Cannot find a unicode character that works in encoding %s'
1605
 
                % (osutils.get_user_encoding(),))
 
1317
            raise TestSkipped('Cannot find a unicode character that works in'
 
1318
                              ' encoding %s' % (bzrlib.user_encoding,))
1606
1319
 
1607
1320
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1608
1321
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1616
1329
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1617
1330
 
1618
1331
 
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):
 
1332
class TestLocalTimeOffset(TestCase):
 
1333
 
 
1334
    def test_local_time_offset(self):
 
1335
        """Test that local_time_offset() returns a sane value."""
 
1336
        offset = osutils.local_time_offset()
 
1337
        self.assertTrue(isinstance(offset, int))
 
1338
        # Test that the offset is no more than a eighteen hours in
 
1339
        # either direction.
 
1340
        # Time zone handling is system specific, so it is difficult to
 
1341
        # do more specific tests, but a value outside of this range is
 
1342
        # probably wrong.
 
1343
        eighteen_hours = 18 * 3600
 
1344
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1345
 
 
1346
    def test_local_time_offset_with_timestamp(self):
 
1347
        """Test that local_time_offset() works with a timestamp."""
 
1348
        offset = osutils.local_time_offset(1000000000.1234567)
 
1349
        self.assertTrue(isinstance(offset, int))
 
1350
        eighteen_hours = 18 * 3600
 
1351
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1352
 
 
1353
 
 
1354
class TestShaFileByName(TestCaseInTempDir):
 
1355
 
 
1356
    def test_sha_empty(self):
 
1357
        self.build_tree_contents([('foo', '')])
 
1358
        expected_sha = osutils.sha_string('')
 
1359
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1360
 
 
1361
    def test_sha_mixed_endings(self):
 
1362
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1363
        self.build_tree_contents([('foo', text)])
 
1364
        expected_sha = osutils.sha_string(text)
 
1365
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1366
 
 
1367
 
 
1368
_debug_text = \
 
1369
r'''# Copyright (C) 2005, 2006 Canonical Ltd
 
1370
#
 
1371
# This program is free software; you can redistribute it and/or modify
 
1372
# it under the terms of the GNU General Public License as published by
 
1373
# the Free Software Foundation; either version 2 of the License, or
 
1374
# (at your option) any later version.
 
1375
#
 
1376
# This program is distributed in the hope that it will be useful,
 
1377
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
1378
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
1379
# GNU General Public License for more details.
 
1380
#
 
1381
# You should have received a copy of the GNU General Public License
 
1382
# along with this program; if not, write to the Free Software
 
1383
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
1384
 
 
1385
 
 
1386
# NOTE: If update these, please also update the help for global-options in
 
1387
#       bzrlib/help_topics/__init__.py
 
1388
 
 
1389
debug_flags = set()
 
1390
"""Set of flags that enable different debug behaviour.
 
1391
 
 
1392
These are set with eg ``-Dlock`` on the bzr command line.
 
1393
 
 
1394
Options include:
 
1395
 
 
1396
 * auth - show authentication sections used
 
1397
 * error - show stack traces for all top level exceptions
 
1398
 * evil - capture call sites that do expensive or badly-scaling operations.
 
1399
 * fetch - trace history copying between repositories
 
1400
 * graph - trace graph traversal information
 
1401
 * hashcache - log every time a working file is read to determine its hash
 
1402
 * hooks - trace hook execution
 
1403
 * hpss - trace smart protocol requests and responses
 
1404
 * http - trace http connections, requests and responses
 
1405
 * index - trace major index operations
 
1406
 * knit - trace knit operations
 
1407
 * lock - trace when lockdir locks are taken or released
 
1408
 * merge - emit information for debugging merges
 
1409
 * pack - emit information about pack operations
 
1410
 
 
1411
"""
 
1412
'''
 
1413
 
 
1414
 
 
1415
class TestResourceLoading(TestCaseInTempDir):
1656
1416
 
1657
1417
    def test_resource_string(self):
1658
1418
        # test resource in bzrlib
1659
1419
        text = osutils.resource_string('bzrlib', 'debug.py')
1660
 
        self.assertContainsRe(text, "debug_flags = set()")
 
1420
        self.assertEquals(_debug_text, text)
1661
1421
        # test resource under bzrlib
1662
1422
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1663
1423
        self.assertContainsRe(text, "class TextUIFactory")
1666
1426
            'yyy.xx')
1667
1427
        # test unknown resource
1668
1428
        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()