← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Ian Clatworthy
  • Date: 2007-08-14 03:59:22 UTC
  • mto: (2733.1.1 ianc-integration)
  • mto: This revision was merged to the branch mainline in revision 2734.
  • Revision ID: ian.clatworthy@internode.on.net-20070814035922-siavg542cwvkf4r5
Fix pretty doc generation so works for all html docs

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006 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
 
from cStringIO import StringIO
20
19
import errno
21
20
import os
22
 
import re
23
21
import socket
24
22
import stat
25
23
import sys
26
 
import time
27
24
 
 
25
import bzrlib
28
26
from bzrlib import (
29
27
    errors,
30
28
    osutils,
31
 
    tests,
32
 
    trace,
33
29
    win32utils,
34
30
    )
 
31
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
35
32
from bzrlib.tests import (
36
 
    file_utils,
37
 
    test__walkdirs_win32,
38
 
    )
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):
 
33
        StringIOWrapper,
 
34
        TestCase, 
 
35
        TestCaseInTempDir, 
 
36
        TestSkipped,
 
37
        )
 
38
 
 
39
 
 
40
class TestOSUtils(TestCaseInTempDir):
107
41
 
108
42
    def test_contains_whitespace(self):
109
43
        self.failUnless(osutils.contains_whitespace(u' '))
119
53
        self.failIf(osutils.contains_whitespace(u'hellothere'))
120
54
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
121
55
 
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
56
    def test_fancy_rename(self):
137
57
        # This should work everywhere
138
 
        self.create_file('a', 'something in a\n')
139
 
        self._fancy_rename('a', 'b')
 
58
        def rename(a, b):
 
59
            osutils.fancy_rename(a, b,
 
60
                    rename_func=os.rename,
 
61
                    unlink_func=os.unlink)
 
62
 
 
63
        open('a', 'wb').write('something in a\n')
 
64
        rename('a', 'b')
140
65
        self.failIfExists('a')
141
66
        self.failUnlessExists('b')
142
67
        self.check_file_contents('b', 'something in a\n')
143
68
 
144
 
        self.create_file('a', 'new something in a\n')
145
 
        self._fancy_rename('b', 'a')
 
69
        open('a', 'wb').write('new something in a\n')
 
70
        rename('b', 'a')
146
71
 
147
72
        self.check_file_contents('a', 'something in a\n')
148
73
 
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
74
    def test_rename(self):
162
75
        # Rename should be semi-atomic on all platforms
163
 
        self.create_file('a', 'something in a\n')
 
76
        open('a', 'wb').write('something in a\n')
164
77
        osutils.rename('a', 'b')
165
78
        self.failIfExists('a')
166
79
        self.failUnlessExists('b')
167
80
        self.check_file_contents('b', 'something in a\n')
168
81
 
169
 
        self.create_file('a', 'new something in a\n')
 
82
        open('a', 'wb').write('new something in a\n')
170
83
        osutils.rename('b', 'a')
171
84
 
172
85
        self.check_file_contents('a', 'something in a\n')
173
86
 
174
87
    # TODO: test fancy_rename using a MemoryTransport
175
88
 
176
 
    def test_rename_change_case(self):
177
 
        # on Windows we should be able to change filename case by rename
178
 
        self.build_tree(['a', 'b/'])
179
 
        osutils.rename('a', 'A')
180
 
        osutils.rename('b', 'B')
181
 
        # we can't use failUnlessExists on case-insensitive filesystem
182
 
        # so try to check shape of the tree
183
 
        shape = sorted(os.listdir('.'))
184
 
        self.assertEquals(['A', 'B'], shape)
185
 
 
186
 
 
187
 
class TestRandChars(tests.TestCase):
188
 
 
189
89
    def test_01_rand_chars_empty(self):
190
90
        result = osutils.rand_chars(0)
191
91
        self.assertEqual(result, '')
196
96
        self.assertEqual(type(result), str)
197
97
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
198
98
 
199
 
 
200
 
class TestIsInside(tests.TestCase):
201
 
 
202
99
    def test_is_inside(self):
203
100
        is_inside = osutils.is_inside
204
101
        self.assertTrue(is_inside('src', 'src/foo.c'))
208
105
        self.assertFalse(is_inside('foo.c', ''))
209
106
        self.assertTrue(is_inside('', 'foo.c'))
210
107
 
211
 
    def test_is_inside_any(self):
212
 
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
213
 
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
214
 
                         (['src'], SRC_FOO_C),
215
 
                         (['src'], 'src'),
216
 
                         ]:
217
 
            self.assert_(osutils.is_inside_any(dirs, fn))
218
 
        for dirs, fn in [(['src'], 'srccontrol'),
219
 
                         (['src'], 'srccontrol/foo')]:
220
 
            self.assertFalse(osutils.is_inside_any(dirs, fn))
221
 
 
222
 
    def test_is_inside_or_parent_of_any(self):
223
 
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
224
 
                         (['src'], 'src/foo.c'),
225
 
                         (['src/bar.c'], 'src'),
226
 
                         (['src/bar.c', 'bla/foo.c'], 'src'),
227
 
                         (['src'], 'src'),
228
 
                         ]:
229
 
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
230
 
 
231
 
        for dirs, fn in [(['src'], 'srccontrol'),
232
 
                         (['srccontrol/foo.c'], 'src'),
233
 
                         (['src'], 'srccontrol/foo')]:
234
 
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
235
 
 
236
 
 
237
 
class TestRmTree(tests.TestCaseInTempDir):
238
 
 
239
108
    def test_rmtree(self):
240
109
        # Check to remove tree with read-only files/dirs
241
110
        os.mkdir('dir')
254
123
        self.failIfExists('dir/file')
255
124
        self.failIfExists('dir')
256
125
 
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
126
    def test_file_kind(self):
273
127
        self.build_tree(['file', 'dir/'])
274
128
        self.assertEquals('file', osutils.file_kind('file'))
276
130
        if osutils.has_symlinks():
277
131
            os.symlink('symlink', 'symlink')
278
132
            self.assertEquals('symlink', osutils.file_kind('symlink'))
279
 
 
 
133
        
280
134
        # TODO: jam 20060529 Test a block device
281
135
        try:
282
136
            os.lstat('/dev/null')
304
158
                os.remove('socket')
305
159
 
306
160
    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):
 
161
        self.assertEqual(osutils.kind_marker('file'), '')
 
162
        self.assertEqual(osutils.kind_marker('directory'), '/')
 
163
        self.assertEqual(osutils.kind_marker('symlink'), '@')
 
164
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
318
165
 
319
166
    def test_get_umask(self):
320
167
        if sys.platform == 'win32':
323
170
            return
324
171
 
325
172
        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):
 
173
        try:
 
174
            os.umask(0222)
 
175
            self.assertEqual(0222, osutils.get_umask())
 
176
            os.umask(0022)
 
177
            self.assertEqual(0022, osutils.get_umask())
 
178
            os.umask(0002)
 
179
            self.assertEqual(0002, osutils.get_umask())
 
180
            os.umask(0027)
 
181
            self.assertEqual(0027, osutils.get_umask())
 
182
        finally:
 
183
            os.umask(orig_umask)
338
184
 
339
185
    def assertFormatedDelta(self, expected, seconds):
340
186
        """Assert osutils.format_delta formats as expected"""
372
218
        self.assertFormatedDelta('1 second in the future', -1)
373
219
        self.assertFormatedDelta('2 seconds in the future', -2)
374
220
 
375
 
    def test_format_date(self):
376
 
        self.assertRaises(errors.UnsupportedTimezoneFormat,
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):
414
 
 
415
221
    def test_dereference_path(self):
416
 
        self.requireFeature(tests.SymlinkFeature)
 
222
        if not osutils.has_symlinks():
 
223
            raise TestSkipped('Symlinks are not supported on this platform')
417
224
        cwd = osutils.realpath('.')
418
225
        os.mkdir('bar')
419
226
        bar_path = osutils.pathjoin(cwd, 'bar')
422
229
        self.assertEqual(bar_path, osutils.realpath('./bar'))
423
230
        os.symlink('bar', 'foo')
424
231
        self.assertEqual(bar_path, osutils.realpath('./foo'))
425
 
 
 
232
        
426
233
        # Does not dereference terminal symlinks
427
234
        foo_path = osutils.pathjoin(cwd, 'foo')
428
235
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
439
246
        foo_baz_path = osutils.pathjoin(foo_path, 'baz')
440
247
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
441
248
 
 
249
 
442
250
    def test_changing_access(self):
443
251
        f = file('file', 'w')
444
252
        f.write('monkey')
446
254
 
447
255
        # Make a file readonly
448
256
        osutils.make_readonly('file')
449
 
        mode = os.lstat('file').st_mode
 
257
        mode = osutils.lstat('file').st_mode
450
258
        self.assertEqual(mode, mode & 0777555)
451
259
 
452
260
        # Make a file writable
453
261
        osutils.make_writable('file')
454
 
        mode = os.lstat('file').st_mode
 
262
        mode = osutils.lstat('file').st_mode
455
263
        self.assertEqual(mode, mode | 0200)
456
264
 
457
265
        if osutils.has_symlinks():
460
268
            osutils.make_readonly('dangling')
461
269
            osutils.make_writable('dangling')
462
270
 
463
 
    def test_host_os_dereferences_symlinks(self):
464
 
        osutils.host_os_dereferences_symlinks()
465
 
 
466
 
 
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):
528
 
    """Test pumpfile method."""
529
 
 
530
 
    def setUp(self):
531
 
        tests.TestCase.setUp(self)
532
 
        # create a test datablock
533
 
        self.block_size = 512
534
 
        pattern = '0123456789ABCDEF'
535
 
        self.test_data = pattern * (3 * self.block_size / len(pattern))
536
 
        self.test_data_len = len(self.test_data)
537
 
 
538
 
    def test_bracket_block_size(self):
539
 
        """Read data in blocks with the requested read size bracketing the
540
 
        block size."""
541
 
        # make sure test data is larger than max read size
542
 
        self.assertTrue(self.test_data_len > self.block_size)
543
 
 
544
 
        from_file = file_utils.FakeReadFile(self.test_data)
545
 
        to_file = StringIO()
546
 
 
547
 
        # read (max / 2) bytes and verify read size wasn't affected
548
 
        num_bytes_to_read = self.block_size / 2
549
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
550
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
551
 
        self.assertEqual(from_file.get_read_count(), 1)
552
 
 
553
 
        # read (max) bytes and verify read size wasn't affected
554
 
        num_bytes_to_read = self.block_size
555
 
        from_file.reset_read_count()
556
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
557
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
558
 
        self.assertEqual(from_file.get_read_count(), 1)
559
 
 
560
 
        # read (max + 1) bytes and verify read size was limited
561
 
        num_bytes_to_read = self.block_size + 1
562
 
        from_file.reset_read_count()
563
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
564
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
565
 
        self.assertEqual(from_file.get_read_count(), 2)
566
 
 
567
 
        # finish reading the rest of the data
568
 
        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)
570
 
 
571
 
        # report error if the data wasn't equal (we only report the size due
572
 
        # to the length of the data)
573
 
        response_data = to_file.getvalue()
574
 
        if response_data != self.test_data:
575
 
            message = "Data not equal.  Expected %d bytes, received %d."
576
 
            self.fail(message % (len(response_data), self.test_data_len))
577
 
 
578
 
    def test_specified_size(self):
579
 
        """Request a transfer larger than the maximum block size and verify
580
 
        that the maximum read doesn't exceed the block_size."""
581
 
        # make sure test data is larger than max read size
582
 
        self.assertTrue(self.test_data_len > self.block_size)
583
 
 
584
 
        # retrieve data in blocks
585
 
        from_file = file_utils.FakeReadFile(self.test_data)
586
 
        to_file = StringIO()
587
 
        osutils.pumpfile(from_file, to_file, self.test_data_len,
588
 
                         self.block_size)
589
 
 
590
 
        # verify read size was equal to the maximum read size
591
 
        self.assertTrue(from_file.get_max_read_size() > 0)
592
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
593
 
        self.assertEqual(from_file.get_read_count(), 3)
594
 
 
595
 
        # report error if the data wasn't equal (we only report the size due
596
 
        # to the length of the data)
597
 
        response_data = to_file.getvalue()
598
 
        if response_data != self.test_data:
599
 
            message = "Data not equal.  Expected %d bytes, received %d."
600
 
            self.fail(message % (len(response_data), self.test_data_len))
601
 
 
602
 
    def test_to_eof(self):
603
 
        """Read to end-of-file and verify that the reads are not larger than
604
 
        the maximum read size."""
605
 
        # make sure test data is larger than max read size
606
 
        self.assertTrue(self.test_data_len > self.block_size)
607
 
 
608
 
        # retrieve data to EOF
609
 
        from_file = file_utils.FakeReadFile(self.test_data)
610
 
        to_file = StringIO()
611
 
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
612
 
 
613
 
        # verify read size was equal to the maximum read size
614
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
615
 
        self.assertEqual(from_file.get_read_count(), 4)
616
 
 
617
 
        # report error if the data wasn't equal (we only report the size due
618
 
        # to the length of the data)
619
 
        response_data = to_file.getvalue()
620
 
        if response_data != self.test_data:
621
 
            message = "Data not equal.  Expected %d bytes, received %d."
622
 
            self.fail(message % (len(response_data), self.test_data_len))
623
 
 
624
 
    def test_defaults(self):
625
 
        """Verifies that the default arguments will read to EOF -- this
626
 
        test verifies that any existing usages of pumpfile will not be broken
627
 
        with this new version."""
628
 
        # retrieve data using default (old) pumpfile method
629
 
        from_file = file_utils.FakeReadFile(self.test_data)
630
 
        to_file = StringIO()
631
 
        osutils.pumpfile(from_file, to_file)
632
 
 
633
 
        # report error if the data wasn't equal (we only report the size due
634
 
        # to the length of the data)
635
 
        response_data = to_file.getvalue()
636
 
        if response_data != self.test_data:
637
 
            message = "Data not equal.  Expected %d bytes, received %d."
638
 
            self.fail(message % (len(response_data), self.test_data_len))
639
 
 
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):
 
271
 
 
272
    def test_kind_marker(self):
 
273
        self.assertEqual("", osutils.kind_marker("file"))
 
274
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
275
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
276
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
 
277
 
 
278
 
 
279
class TestSafeUnicode(TestCase):
712
280
 
713
281
    def test_from_ascii_string(self):
714
282
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
723
291
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
724
292
 
725
293
    def test_bad_utf8_string(self):
726
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
294
        self.assertRaises(BzrBadParameterNotUnicode,
727
295
                          osutils.safe_unicode,
728
296
                          '\xbb\xbb')
729
297
 
730
298
 
731
 
class TestSafeUtf8(tests.TestCase):
 
299
class TestSafeUtf8(TestCase):
732
300
 
733
301
    def test_from_ascii_string(self):
734
302
        f = 'foobar'
744
312
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
745
313
 
746
314
    def test_bad_utf8_string(self):
747
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
315
        self.assertRaises(BzrBadParameterNotUnicode,
748
316
                          osutils.safe_utf8, '\xbb\xbb')
749
317
 
750
318
 
751
 
class TestSafeRevisionId(tests.TestCase):
 
319
class TestSafeRevisionId(TestCase):
752
320
 
753
321
    def test_from_ascii_string(self):
754
 
        # this shouldn't give a warning because it's getting an ascii string
755
322
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
756
323
 
757
324
    def test_from_unicode_string_ascii_contents(self):
776
343
        self.assertEqual(None, osutils.safe_revision_id(None))
777
344
 
778
345
 
779
 
class TestSafeFileId(tests.TestCase):
 
346
class TestSafeFileId(TestCase):
780
347
 
781
348
    def test_from_ascii_string(self):
782
349
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
802
369
        self.assertEqual(None, osutils.safe_file_id(None))
803
370
 
804
371
 
805
 
class TestWin32Funcs(tests.TestCase):
806
 
    """Test that _win32 versions of os utilities return appropriate paths."""
 
372
class TestWin32Funcs(TestCase):
 
373
    """Test that the _win32 versions of os utilities return appropriate paths."""
807
374
 
808
375
    def test_abspath(self):
809
376
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
816
383
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
817
384
 
818
385
    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'))
 
386
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
 
387
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
388
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
 
389
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
 
390
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
391
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
831
392
 
832
393
    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'))
 
394
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
395
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
837
396
 
838
397
    def test_getcwd(self):
839
398
        cwd = osutils._win32_getcwd()
868
427
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
869
428
 
870
429
 
871
 
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
 
430
class TestWin32FuncsDirs(TestCaseInTempDir):
872
431
    """Test win32 functions that create files."""
 
432
    
 
433
    def test_getcwd(self):
 
434
        if win32utils.winver == 'Windows 98':
 
435
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
 
436
        # Make sure getcwd can handle unicode filenames
 
437
        try:
 
438
            os.mkdir(u'mu-\xb5')
 
439
        except UnicodeError:
 
440
            raise TestSkipped("Unable to create Unicode filename")
873
441
 
874
 
    def test_getcwd(self):
875
 
        self.requireFeature(tests.UnicodeFilenameFeature)
876
 
        os.mkdir(u'mu-\xb5')
877
442
        os.chdir(u'mu-\xb5')
878
443
        # TODO: jam 20060427 This will probably fail on Mac OSX because
879
444
        #       it will change the normalization of B\xe5gfors
881
446
        #       osutils.getcwd() renormalize the path.
882
447
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
883
448
 
884
 
    def test_minimum_path_selection(self):
885
 
        self.assertEqual(set(),
886
 
            osutils.minimum_path_selection([]))
887
 
        self.assertEqual(set(['a']),
888
 
            osutils.minimum_path_selection(['a']))
889
 
        self.assertEqual(set(['a', 'b']),
890
 
            osutils.minimum_path_selection(['a', 'b']))
891
 
        self.assertEqual(set(['a/', 'b']),
892
 
            osutils.minimum_path_selection(['a/', 'b']))
893
 
        self.assertEqual(set(['a/', 'b']),
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']))
897
 
 
898
449
    def test_mkdtemp(self):
899
450
        tmpdir = osutils._win32_mkdtemp(dir='.')
900
451
        self.assertFalse('\\' in tmpdir)
955
506
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
956
507
 
957
508
 
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):
 
509
class TestMacFuncsDirs(TestCaseInTempDir):
968
510
    """Test mac special functions that require directories."""
969
511
 
970
512
    def test_getcwd(self):
971
 
        self.requireFeature(tests.UnicodeFilenameFeature)
972
 
        os.mkdir(u'B\xe5gfors')
 
513
        # On Mac, this will actually create Ba\u030agfors
 
514
        # but chdir will still work, because it accepts both paths
 
515
        try:
 
516
            os.mkdir(u'B\xe5gfors')
 
517
        except UnicodeError:
 
518
            raise TestSkipped("Unable to create Unicode filename")
 
519
 
973
520
        os.chdir(u'B\xe5gfors')
974
521
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
975
522
 
976
523
    def test_getcwd_nonnorm(self):
977
 
        self.requireFeature(tests.UnicodeFilenameFeature)
978
524
        # Test that _mac_getcwd() will normalize this path
979
 
        os.mkdir(u'Ba\u030agfors')
 
525
        try:
 
526
            os.mkdir(u'Ba\u030agfors')
 
527
        except UnicodeError:
 
528
            raise TestSkipped("Unable to create Unicode filename")
 
529
 
980
530
        os.chdir(u'Ba\u030agfors')
981
531
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
982
532
 
983
533
 
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):
 
534
class TestSplitLines(TestCase):
1002
535
 
1003
536
    def test_split_unicode(self):
1004
537
        self.assertEqual([u'foo\n', u'bar\xae'],
1011
544
                         osutils.split_lines('foo\rbar\n'))
1012
545
 
1013
546
 
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])
 
547
class TestWalkDirs(TestCaseInTempDir):
1020
548
 
1021
549
    def test_walkdirs(self):
1022
550
        tree = [
1055
583
            result.append((dirdetail, dirblock))
1056
584
 
1057
585
        self.assertTrue(found_bzrdir)
1058
 
        self.assertExpectedBlocks(expected_dirblocks, result)
 
586
        self.assertEqual(expected_dirblocks,
 
587
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1059
588
        # you can search a subdir only, with a supplied prefix.
1060
589
        result = []
1061
590
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1062
591
            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")
 
592
        self.assertEqual(expected_dirblocks[1:],
 
593
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1084
594
 
1085
595
    def test__walkdirs_utf8(self):
1086
596
        tree = [
1119
629
            result.append((dirdetail, dirblock))
1120
630
 
1121
631
        self.assertTrue(found_bzrdir)
1122
 
        self.assertExpectedBlocks(expected_dirblocks, result)
1123
 
 
 
632
        self.assertEqual(expected_dirblocks,
 
633
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1124
634
        # you can search a subdir only, with a supplied prefix.
1125
635
        result = []
1126
636
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1127
637
            result.append(dirblock)
1128
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
638
        self.assertEqual(expected_dirblocks[1:],
 
639
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1129
640
 
1130
641
    def _filter_out_stat(self, result):
1131
642
        """Filter out the stat value from the walkdirs result"""
1136
647
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1137
648
            dirblock[:] = new_dirblock
1138
649
 
1139
 
    def _save_platform_info(self):
1140
 
        self.overrideAttr(win32utils, 'winver')
1141
 
        self.overrideAttr(osutils, '_fs_enc')
1142
 
        self.overrideAttr(osutils, '_selected_dir_reader')
1143
 
 
1144
 
    def assertDirReaderIs(self, expected):
1145
 
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1146
 
        # Force it to redetect
1147
 
        osutils._selected_dir_reader = None
1148
 
        # Nothing to list, but should still trigger the selection logic
1149
 
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1150
 
        self.assertIsInstance(osutils._selected_dir_reader, expected)
1151
 
 
1152
 
    def test_force_walkdirs_utf8_fs_utf8(self):
1153
 
        self.requireFeature(UTF8DirReaderFeature)
1154
 
        self._save_platform_info()
1155
 
        win32utils.winver = None # Avoid the win32 detection code
1156
 
        osutils._fs_enc = 'UTF-8'
1157
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1158
 
 
1159
 
    def test_force_walkdirs_utf8_fs_ascii(self):
1160
 
        self.requireFeature(UTF8DirReaderFeature)
1161
 
        self._save_platform_info()
1162
 
        win32utils.winver = None # Avoid the win32 detection code
1163
 
        osutils._fs_enc = 'US-ASCII'
1164
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1165
 
 
1166
 
    def test_force_walkdirs_utf8_fs_ANSI(self):
1167
 
        self.requireFeature(UTF8DirReaderFeature)
1168
 
        self._save_platform_info()
1169
 
        win32utils.winver = None # Avoid the win32 detection code
1170
 
        osutils._fs_enc = 'ANSI_X3.4-1968'
1171
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1172
 
 
1173
 
    def test_force_walkdirs_utf8_fs_latin1(self):
1174
 
        self._save_platform_info()
1175
 
        win32utils.winver = None # Avoid the win32 detection code
1176
 
        osutils._fs_enc = 'latin1'
1177
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1178
 
 
1179
 
    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)
1182
 
        self._save_platform_info()
1183
 
        win32utils.winver = 'Windows NT'
1184
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1185
 
        self.assertDirReaderIs(Win32ReadDir)
1186
 
 
1187
 
    def test_force_walkdirs_utf8_98(self):
1188
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1189
 
        self._save_platform_info()
1190
 
        win32utils.winver = 'Windows 98'
1191
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1192
 
 
1193
650
    def test_unicode_walkdirs(self):
1194
651
        """Walkdirs should always return unicode paths."""
1195
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1196
652
        name0 = u'0file-\xb6'
1197
653
        name1 = u'1dir-\u062c\u0648'
1198
654
        name2 = u'2file-\u0633'
1203
659
            name1 + '/' + name1 + '/',
1204
660
            name2,
1205
661
            ]
1206
 
        self.build_tree(tree)
 
662
        try:
 
663
            self.build_tree(tree)
 
664
        except UnicodeError:
 
665
            raise TestSkipped('Could not represent Unicode chars'
 
666
                              ' in current encoding.')
1207
667
        expected_dirblocks = [
1208
668
                ((u'', u'.'),
1209
669
                 [(name0, name0, 'file', './' + name0),
1235
695
 
1236
696
        The abspath portion might be in unicode or utf-8
1237
697
        """
1238
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1239
698
        name0 = u'0file-\xb6'
1240
699
        name1 = u'1dir-\u062c\u0648'
1241
700
        name2 = u'2file-\u0633'
1246
705
            name1 + '/' + name1 + '/',
1247
706
            name2,
1248
707
            ]
1249
 
        self.build_tree(tree)
 
708
        try:
 
709
            self.build_tree(tree)
 
710
        except UnicodeError:
 
711
            raise TestSkipped('Could not represent Unicode chars'
 
712
                              ' in current encoding.')
1250
713
        name0 = name0.encode('utf8')
1251
714
        name1 = name1.encode('utf8')
1252
715
        name2 = name2.encode('utf8')
1291
754
            result.append((dirdetail, new_dirblock))
1292
755
        self.assertEqual(expected_dirblocks, result)
1293
756
 
1294
 
    def test__walkdirs_utf8_with_unicode_fs(self):
1295
 
        """UnicodeDirReader should be a safe fallback everywhere
 
757
    def test_unicode__walkdirs_unicode_to_utf8(self):
 
758
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
1296
759
 
1297
760
        The abspath portion should be in unicode
1298
761
        """
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
 
        name0u = u'0file-\xb6'
1305
 
        name1u = u'1dir-\u062c\u0648'
1306
 
        name2u = u'2file-\u0633'
1307
 
        tree = [
1308
 
            name0u,
1309
 
            name1u + '/',
1310
 
            name1u + '/' + name0u,
1311
 
            name1u + '/' + name1u + '/',
1312
 
            name2u,
1313
 
            ]
1314
 
        self.build_tree(tree)
1315
 
        name0 = name0u.encode('utf8')
1316
 
        name1 = name1u.encode('utf8')
1317
 
        name2 = name2u.encode('utf8')
1318
 
 
1319
 
        # All of the abspaths should be in unicode, all of the relative paths
1320
 
        # should be in utf8
1321
 
        expected_dirblocks = [
1322
 
                (('', '.'),
1323
 
                 [(name0, name0, 'file', './' + name0u),
1324
 
                  (name1, name1, 'directory', './' + name1u),
1325
 
                  (name2, name2, 'file', './' + name2u),
1326
 
                 ]
1327
 
                ),
1328
 
                ((name1, './' + name1u),
1329
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1330
 
                                                        + '/' + name0u),
1331
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1332
 
                                                            + '/' + name1u),
1333
 
                 ]
1334
 
                ),
1335
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1336
 
                 [
1337
 
                 ]
1338
 
                ),
1339
 
            ]
1340
 
        result = list(osutils._walkdirs_utf8('.'))
1341
 
        self._filter_out_stat(result)
1342
 
        self.assertEqual(expected_dirblocks, result)
1343
 
 
1344
 
    def test__walkdirs_utf8_win32readdir(self):
1345
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1346
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1347
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1348
 
        self._save_platform_info()
1349
 
        osutils._selected_dir_reader = Win32ReadDir()
1350
 
        name0u = u'0file-\xb6'
1351
 
        name1u = u'1dir-\u062c\u0648'
1352
 
        name2u = u'2file-\u0633'
1353
 
        tree = [
1354
 
            name0u,
1355
 
            name1u + '/',
1356
 
            name1u + '/' + name0u,
1357
 
            name1u + '/' + name1u + '/',
1358
 
            name2u,
1359
 
            ]
1360
 
        self.build_tree(tree)
1361
 
        name0 = name0u.encode('utf8')
1362
 
        name1 = name1u.encode('utf8')
1363
 
        name2 = name2u.encode('utf8')
1364
 
 
1365
 
        # All of the abspaths should be in unicode, all of the relative paths
1366
 
        # should be in utf8
1367
 
        expected_dirblocks = [
1368
 
                (('', '.'),
1369
 
                 [(name0, name0, 'file', './' + name0u),
1370
 
                  (name1, name1, 'directory', './' + name1u),
1371
 
                  (name2, name2, 'file', './' + name2u),
1372
 
                 ]
1373
 
                ),
1374
 
                ((name1, './' + name1u),
1375
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1376
 
                                                        + '/' + name0u),
1377
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1378
 
                                                            + '/' + name1u),
1379
 
                 ]
1380
 
                ),
1381
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1382
 
                 [
1383
 
                 ]
1384
 
                ),
1385
 
            ]
1386
 
        result = list(osutils._walkdirs_utf8(u'.'))
1387
 
        self._filter_out_stat(result)
1388
 
        self.assertEqual(expected_dirblocks, result)
1389
 
 
1390
 
    def assertStatIsCorrect(self, path, win32stat):
1391
 
        os_stat = os.stat(path)
1392
 
        self.assertEqual(os_stat.st_size, win32stat.st_size)
1393
 
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1394
 
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1395
 
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1396
 
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1397
 
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1398
 
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1399
 
 
1400
 
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1401
 
        """make sure our Stat values are valid"""
1402
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1403
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1404
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1405
 
        name0u = u'0file-\xb6'
1406
 
        name0 = name0u.encode('utf8')
1407
 
        self.build_tree([name0u])
1408
 
        # I hate to sleep() here, but I'm trying to make the ctime different
1409
 
        # from the mtime
1410
 
        time.sleep(2)
1411
 
        f = open(name0u, 'ab')
 
762
        name0u = u'0file-\xb6'
 
763
        name1u = u'1dir-\u062c\u0648'
 
764
        name2u = u'2file-\u0633'
 
765
        tree = [
 
766
            name0u,
 
767
            name1u + '/',
 
768
            name1u + '/' + name0u,
 
769
            name1u + '/' + name1u + '/',
 
770
            name2u,
 
771
            ]
1412
772
        try:
1413
 
            f.write('just a small update')
1414
 
        finally:
1415
 
            f.close()
1416
 
 
1417
 
        result = Win32ReadDir().read_dir('', u'.')
1418
 
        entry = result[0]
1419
 
        self.assertEqual((name0, name0, 'file'), entry[:3])
1420
 
        self.assertEqual(u'./' + name0u, entry[4])
1421
 
        self.assertStatIsCorrect(entry[4], entry[3])
1422
 
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1423
 
 
1424
 
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1425
 
        """make sure our Stat values are valid"""
1426
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1427
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1428
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1429
 
        name0u = u'0dir-\u062c\u0648'
 
773
            self.build_tree(tree)
 
774
        except UnicodeError:
 
775
            raise TestSkipped('Could not represent Unicode chars'
 
776
                              ' in current encoding.')
1430
777
        name0 = name0u.encode('utf8')
1431
 
        self.build_tree([name0u + '/'])
 
778
        name1 = name1u.encode('utf8')
 
779
        name2 = name2u.encode('utf8')
1432
780
 
1433
 
        result = Win32ReadDir().read_dir('', u'.')
1434
 
        entry = result[0]
1435
 
        self.assertEqual((name0, name0, 'directory'), entry[:3])
1436
 
        self.assertEqual(u'./' + name0u, entry[4])
1437
 
        self.assertStatIsCorrect(entry[4], entry[3])
 
781
        # All of the abspaths should be in unicode, all of the relative paths
 
782
        # should be in utf8
 
783
        expected_dirblocks = [
 
784
                (('', '.'),
 
785
                 [(name0, name0, 'file', './' + name0u),
 
786
                  (name1, name1, 'directory', './' + name1u),
 
787
                  (name2, name2, 'file', './' + name2u),
 
788
                 ]
 
789
                ),
 
790
                ((name1, './' + name1u),
 
791
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
792
                                                        + '/' + name0u),
 
793
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
794
                                                            + '/' + name1u),
 
795
                 ]
 
796
                ),
 
797
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
798
                 [
 
799
                 ]
 
800
                ),
 
801
            ]
 
802
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
 
803
        self._filter_out_stat(result)
 
804
        self.assertEqual(expected_dirblocks, result)
1438
805
 
1439
806
    def assertPathCompare(self, path_less, path_greater):
1440
807
        """check that path_less and path_greater compare correctly."""
1514
881
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1515
882
 
1516
883
 
1517
 
class TestCopyTree(tests.TestCaseInTempDir):
1518
 
 
 
884
class TestCopyTree(TestCaseInTempDir):
 
885
    
1519
886
    def test_copy_basic_tree(self):
1520
887
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1521
888
        osutils.copy_tree('source', 'target')
1530
897
        self.assertEqual(['c'], os.listdir('target/b'))
1531
898
 
1532
899
    def test_copy_tree_symlinks(self):
1533
 
        self.requireFeature(tests.SymlinkFeature)
 
900
        if not osutils.has_symlinks():
 
901
            return
1534
902
        self.build_tree(['source/'])
1535
903
        os.symlink('a/generic/path', 'source/lnk')
1536
904
        osutils.copy_tree('source', 'target')
1566
934
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1567
935
 
1568
936
 
1569
 
class TestSetUnsetEnv(tests.TestCase):
 
937
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
 
938
# [bialix] 2006/12/26
 
939
 
 
940
 
 
941
class TestSetUnsetEnv(TestCase):
1570
942
    """Test updating the environment"""
1571
943
 
1572
944
    def setUp(self):
1578
950
        def cleanup():
1579
951
            if 'BZR_TEST_ENV_VAR' in os.environ:
1580
952
                del os.environ['BZR_TEST_ENV_VAR']
 
953
 
1581
954
        self.addCleanup(cleanup)
1582
955
 
1583
956
    def test_set(self):
1595
968
 
1596
969
    def test_unicode(self):
1597
970
        """Environment can only contain plain strings
1598
 
 
 
971
        
1599
972
        So Unicode strings must be encoded.
1600
973
        """
1601
 
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
1602
 
        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(),))
 
974
        # Try a few different characters, to see if we can get
 
975
        # one that will be valid in the user_encoding
 
976
        possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
 
977
        for uni_val in possible_vals:
 
978
            try:
 
979
                env_val = uni_val.encode(bzrlib.user_encoding)
 
980
            except UnicodeEncodeError:
 
981
                # Try a different character
 
982
                pass
 
983
            else:
 
984
                break
 
985
        else:
 
986
            raise TestSkipped('Cannot find a unicode character that works in'
 
987
                              ' encoding %s' % (bzrlib.user_encoding,))
1606
988
 
1607
989
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1608
990
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1616
998
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1617
999
 
1618
1000
 
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):
1656
 
 
1657
 
    def test_resource_string(self):
1658
 
        # test resource in bzrlib
1659
 
        text = osutils.resource_string('bzrlib', 'debug.py')
1660
 
        self.assertContainsRe(text, "debug_flags = set()")
1661
 
        # test resource under bzrlib
1662
 
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1663
 
        self.assertContainsRe(text, "class TextUIFactory")
1664
 
        # test unsupported package
1665
 
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1666
 
            'yyy.xx')
1667
 
        # test unknown resource
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()
 
1001
class TestLocalTimeOffset(TestCase):
 
1002
 
 
1003
    def test_local_time_offset(self):
 
1004
        """Test that local_time_offset() returns a sane value."""
 
1005
        offset = osutils.local_time_offset()
 
1006
        self.assertTrue(isinstance(offset, int))
 
1007
        # Test that the offset is no more than a eighteen hours in
 
1008
        # either direction.
 
1009
        # Time zone handling is system specific, so it is difficult to
 
1010
        # do more specific tests, but a value outside of this range is
 
1011
        # probably wrong.
 
1012
        eighteen_hours = 18 * 3600
 
1013
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1014
 
 
1015
    def test_local_time_offset_with_timestamp(self):
 
1016
        """Test that local_time_offset() works with a timestamp."""
 
1017
        offset = osutils.local_time_offset(1000000000.1234567)
 
1018
        self.assertTrue(isinstance(offset, int))
 
1019
        eighteen_hours = 18 * 3600
 
1020
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)