/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Robert Collins
  • Date: 2010-05-06 23:41:35 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506234135-yivbzczw1sejxnxc
Lock methods on ``Tree``, ``Branch`` and ``Repository`` are now
expected to return an object which can be used to unlock them. This reduces
duplicate code when using cleanups. The previous 'tokens's returned by
``Branch.lock_write`` and ``Repository.lock_write`` are now attributes
on the result of the lock_write. ``repository.RepositoryWriteLockResult``
and ``branch.BranchWriteLockResult`` document this. (Robert Collins)

``log._get_info_for_log_files`` now takes an add_cleanup callable.
(Robert Collins)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
1
 
# Copyright (C) 2005-2016 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
19
 
from __future__ import absolute_import, division
20
 
 
 
19
from cStringIO import StringIO
21
20
import errno
22
21
import os
23
 
import select
 
22
import re
24
23
import socket
 
24
import stat
25
25
import sys
26
 
import tempfile
27
26
import time
28
27
 
29
 
from .. import (
 
28
from bzrlib import (
30
29
    errors,
31
30
    osutils,
32
31
    tests,
33
32
    trace,
34
33
    win32utils,
35
34
    )
36
 
from ..sixish import (
37
 
    BytesIO,
38
 
    PY3,
39
 
    text_type,
40
 
    )
41
 
from . import (
 
35
from bzrlib.tests import (
42
36
    features,
43
37
    file_utils,
44
38
    test__walkdirs_win32,
45
39
    )
46
 
from .scenarios import load_tests_apply_scenarios
47
 
 
48
 
 
49
 
class _UTF8DirReaderFeature(features.ModuleAvailableFeature):
 
40
 
 
41
 
 
42
class _UTF8DirReaderFeature(tests.Feature):
50
43
 
51
44
    def _probe(self):
52
45
        try:
53
 
            from .. import _readdir_pyx
54
 
            self._module = _readdir_pyx
 
46
            from bzrlib import _readdir_pyx
55
47
            self.reader = _readdir_pyx.UTF8DirReader
56
48
            return True
57
49
        except ImportError:
58
50
            return False
59
51
 
60
 
 
61
 
UTF8DirReaderFeature = _UTF8DirReaderFeature('breezy._readdir_pyx')
62
 
 
63
 
term_ios_feature = features.ModuleAvailableFeature('termios')
 
52
    def feature_name(self):
 
53
        return 'bzrlib._readdir_pyx'
 
54
 
 
55
UTF8DirReaderFeature = _UTF8DirReaderFeature()
 
56
 
 
57
term_ios_feature = tests.ModuleAvailableFeature('termios')
64
58
 
65
59
 
66
60
def _already_unicode(s):
84
78
    # Some DirReaders are platform specific and even there they may not be
85
79
    # available.
86
80
    if UTF8DirReaderFeature.available():
87
 
        from .. import _readdir_pyx
 
81
        from bzrlib import _readdir_pyx
88
82
        scenarios.append(('utf8',
89
83
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
90
84
                               _native_to_unicode=_utf8_to_unicode)))
91
85
 
92
86
    if test__walkdirs_win32.win32_readdir_feature.available():
93
87
        try:
94
 
            from .. import _walkdirs_win32
 
88
            from bzrlib import _walkdirs_win32
95
89
            scenarios.append(
96
90
                ('win32',
97
91
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
101
95
    return scenarios
102
96
 
103
97
 
104
 
load_tests = load_tests_apply_scenarios
 
98
def load_tests(basic_tests, module, loader):
 
99
    suite = loader.suiteClass()
 
100
    dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
 
101
        basic_tests, tests.condition_isinstance(TestDirReader))
 
102
    tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
 
103
    suite.addTest(remaining_tests)
 
104
    return suite
105
105
 
106
106
 
107
107
class TestContainsWhitespace(tests.TestCase):
108
108
 
109
109
    def test_contains_whitespace(self):
110
 
        self.assertTrue(osutils.contains_whitespace(u' '))
111
 
        self.assertTrue(osutils.contains_whitespace(u'hello there'))
112
 
        self.assertTrue(osutils.contains_whitespace(u'hellothere\n'))
113
 
        self.assertTrue(osutils.contains_whitespace(u'hello\nthere'))
114
 
        self.assertTrue(osutils.contains_whitespace(u'hello\rthere'))
115
 
        self.assertTrue(osutils.contains_whitespace(u'hello\tthere'))
 
110
        self.failUnless(osutils.contains_whitespace(u' '))
 
111
        self.failUnless(osutils.contains_whitespace(u'hello there'))
 
112
        self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
 
113
        self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
 
114
        self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
 
115
        self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
116
116
 
117
117
        # \xa0 is "Non-breaking-space" which on some python locales thinks it
118
118
        # is whitespace, but we do not.
119
 
        self.assertFalse(osutils.contains_whitespace(u''))
120
 
        self.assertFalse(osutils.contains_whitespace(u'hellothere'))
121
 
        self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
 
119
        self.failIf(osutils.contains_whitespace(u''))
 
120
        self.failIf(osutils.contains_whitespace(u'hellothere'))
 
121
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
122
122
 
123
123
 
124
124
class TestRename(tests.TestCaseInTempDir):
136
136
 
137
137
    def test_fancy_rename(self):
138
138
        # This should work everywhere
139
 
        self.create_file('a', b'something in a\n')
 
139
        self.create_file('a', 'something in a\n')
140
140
        self._fancy_rename('a', 'b')
141
 
        self.assertPathDoesNotExist('a')
142
 
        self.assertPathExists('b')
143
 
        self.check_file_contents('b', b'something in a\n')
 
141
        self.failIfExists('a')
 
142
        self.failUnlessExists('b')
 
143
        self.check_file_contents('b', 'something in a\n')
144
144
 
145
 
        self.create_file('a', b'new something in a\n')
 
145
        self.create_file('a', 'new something in a\n')
146
146
        self._fancy_rename('b', 'a')
147
147
 
148
 
        self.check_file_contents('a', b'something in a\n')
 
148
        self.check_file_contents('a', 'something in a\n')
149
149
 
150
150
    def test_fancy_rename_fails_source_missing(self):
151
151
        # An exception should be raised, and the target should be left in place
152
 
        self.create_file('target', b'data in target\n')
 
152
        self.create_file('target', 'data in target\n')
153
153
        self.assertRaises((IOError, OSError), self._fancy_rename,
154
154
                          'missingsource', 'target')
155
 
        self.assertPathExists('target')
156
 
        self.check_file_contents('target', b'data in target\n')
 
155
        self.failUnlessExists('target')
 
156
        self.check_file_contents('target', 'data in target\n')
157
157
 
158
158
    def test_fancy_rename_fails_if_source_and_target_missing(self):
159
159
        self.assertRaises((IOError, OSError), self._fancy_rename,
161
161
 
162
162
    def test_rename(self):
163
163
        # Rename should be semi-atomic on all platforms
164
 
        self.create_file('a', b'something in a\n')
 
164
        self.create_file('a', 'something in a\n')
165
165
        osutils.rename('a', 'b')
166
 
        self.assertPathDoesNotExist('a')
167
 
        self.assertPathExists('b')
168
 
        self.check_file_contents('b', b'something in a\n')
 
166
        self.failIfExists('a')
 
167
        self.failUnlessExists('b')
 
168
        self.check_file_contents('b', 'something in a\n')
169
169
 
170
 
        self.create_file('a', b'new something in a\n')
 
170
        self.create_file('a', 'new something in a\n')
171
171
        osutils.rename('b', 'a')
172
172
 
173
 
        self.check_file_contents('a', b'something in a\n')
 
173
        self.check_file_contents('a', 'something in a\n')
174
174
 
175
175
    # TODO: test fancy_rename using a MemoryTransport
176
176
 
182
182
        # we can't use failUnlessExists on case-insensitive filesystem
183
183
        # so try to check shape of the tree
184
184
        shape = sorted(os.listdir('.'))
185
 
        self.assertEqual(['A', 'B'], shape)
 
185
        self.assertEquals(['A', 'B'], shape)
186
186
 
187
 
    def test_rename_exception(self):
188
 
        try:
189
 
            osutils.rename('nonexistent_path', 'different_nonexistent_path')
190
 
        except OSError as e:
191
 
            self.assertEqual(e.old_filename, 'nonexistent_path')
192
 
            self.assertEqual(e.new_filename, 'different_nonexistent_path')
193
 
            self.assertTrue('nonexistent_path' in e.strerror)
194
 
            self.assertTrue('different_nonexistent_path' in e.strerror)
 
187
    def test_rename_error(self):
 
188
        # We wrap os.rename to make it give an error including the filenames
 
189
        # https://bugs.launchpad.net/bzr/+bug/491763
 
190
        err = self.assertRaises(OSError, osutils.rename,
 
191
            'nonexistent', 'target')
 
192
        self.assertContainsString(str(err), 'nonexistent')
195
193
 
196
194
 
197
195
class TestRandChars(tests.TestCase):
224
222
                         (['src'], SRC_FOO_C),
225
223
                         (['src'], 'src'),
226
224
                         ]:
227
 
            self.assertTrue(osutils.is_inside_any(dirs, fn))
 
225
            self.assert_(osutils.is_inside_any(dirs, fn))
228
226
        for dirs, fn in [(['src'], 'srccontrol'),
229
227
                         (['src'], 'srccontrol/foo')]:
230
228
            self.assertFalse(osutils.is_inside_any(dirs, fn))
236
234
                         (['src/bar.c', 'bla/foo.c'], 'src'),
237
235
                         (['src'], 'src'),
238
236
                         ]:
239
 
            self.assertTrue(osutils.is_inside_or_parent_of_any(dirs, fn))
 
237
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
240
238
 
241
239
        for dirs, fn in [(['src'], 'srccontrol'),
242
240
                         (['srccontrol/foo.c'], 'src'),
244
242
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
245
243
 
246
244
 
247
 
class TestLstat(tests.TestCaseInTempDir):
248
 
 
249
 
    def test_lstat_matches_fstat(self):
250
 
        # On Windows, lstat and fstat don't always agree, primarily in the
251
 
        # 'st_ino' and 'st_dev' fields. So we force them to be '0' in our
252
 
        # custom implementation.
253
 
        if sys.platform == 'win32':
254
 
            # We only have special lstat/fstat if we have the extension.
255
 
            # Without it, we may end up re-reading content when we don't have
256
 
            # to, but otherwise it doesn't effect correctness.
257
 
            self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
258
 
        with open('test-file.txt', 'wb') as f:
259
 
            f.write(b'some content\n')
260
 
            f.flush()
261
 
            self.assertEqualStat(osutils.fstat(f.fileno()),
262
 
                                 osutils.lstat('test-file.txt'))
263
 
 
264
 
 
265
245
class TestRmTree(tests.TestCaseInTempDir):
266
246
 
267
247
    def test_rmtree(self):
268
248
        # Check to remove tree with read-only files/dirs
269
249
        os.mkdir('dir')
270
 
        with open('dir/file', 'w') as f:
271
 
            f.write('spam')
 
250
        f = file('dir/file', 'w')
 
251
        f.write('spam')
 
252
        f.close()
272
253
        # would like to also try making the directory readonly, but at the
273
254
        # moment python shutil.rmtree doesn't handle that properly - it would
274
255
        # need to chmod the directory before removing things inside it - deferred
278
259
 
279
260
        osutils.rmtree('dir')
280
261
 
281
 
        self.assertPathDoesNotExist('dir/file')
282
 
        self.assertPathDoesNotExist('dir')
 
262
        self.failIfExists('dir/file')
 
263
        self.failIfExists('dir')
283
264
 
284
265
 
285
266
class TestDeleteAny(tests.TestCaseInTempDir):
298
279
 
299
280
    def test_file_kind(self):
300
281
        self.build_tree(['file', 'dir/'])
301
 
        self.assertEqual('file', osutils.file_kind('file'))
302
 
        self.assertEqual('directory', osutils.file_kind('dir/'))
 
282
        self.assertEquals('file', osutils.file_kind('file'))
 
283
        self.assertEquals('directory', osutils.file_kind('dir/'))
303
284
        if osutils.has_symlinks():
304
285
            os.symlink('symlink', 'symlink')
305
 
            self.assertEqual('symlink', osutils.file_kind('symlink'))
 
286
            self.assertEquals('symlink', osutils.file_kind('symlink'))
306
287
 
307
288
        # TODO: jam 20060529 Test a block device
308
289
        try:
309
290
            os.lstat('/dev/null')
310
 
        except OSError as e:
 
291
        except OSError, e:
311
292
            if e.errno not in (errno.ENOENT,):
312
293
                raise
313
294
        else:
314
 
            self.assertEqual(
315
 
                'chardev',
316
 
                osutils.file_kind(os.path.realpath('/dev/null')))
 
295
            self.assertEquals('chardev', osutils.file_kind('/dev/null'))
317
296
 
318
297
        mkfifo = getattr(os, 'mkfifo', None)
319
298
        if mkfifo:
320
299
            mkfifo('fifo')
321
300
            try:
322
 
                self.assertEqual('fifo', osutils.file_kind('fifo'))
 
301
                self.assertEquals('fifo', osutils.file_kind('fifo'))
323
302
            finally:
324
303
                os.remove('fifo')
325
304
 
328
307
            s = socket.socket(AF_UNIX)
329
308
            s.bind('socket')
330
309
            try:
331
 
                self.assertEqual('socket', osutils.file_kind('socket'))
 
310
                self.assertEquals('socket', osutils.file_kind('socket'))
332
311
            finally:
333
312
                os.remove('socket')
334
313
 
353
332
 
354
333
        orig_umask = osutils.get_umask()
355
334
        self.addCleanup(os.umask, orig_umask)
356
 
        os.umask(0o222)
357
 
        self.assertEqual(0o222, osutils.get_umask())
358
 
        os.umask(0o022)
359
 
        self.assertEqual(0o022, osutils.get_umask())
360
 
        os.umask(0o002)
361
 
        self.assertEqual(0o002, osutils.get_umask())
362
 
        os.umask(0o027)
363
 
        self.assertEqual(0o027, osutils.get_umask())
 
335
        os.umask(0222)
 
336
        self.assertEqual(0222, osutils.get_umask())
 
337
        os.umask(0022)
 
338
        self.assertEqual(0022, osutils.get_umask())
 
339
        os.umask(0002)
 
340
        self.assertEqual(0002, osutils.get_umask())
 
341
        os.umask(0027)
 
342
        self.assertEqual(0027, osutils.get_umask())
364
343
 
365
344
 
366
345
class TestDateTime(tests.TestCase):
402
381
        self.assertFormatedDelta('2 seconds in the future', -2)
403
382
 
404
383
    def test_format_date(self):
405
 
        self.assertRaises(osutils.UnsupportedTimezoneFormat,
406
 
                          osutils.format_date, 0, timezone='foo')
 
384
        self.assertRaises(errors.UnsupportedTimezoneFormat,
 
385
            osutils.format_date, 0, timezone='foo')
407
386
        self.assertIsInstance(osutils.format_date(0), str)
408
 
        self.assertIsInstance(osutils.format_local_date(0), text_type)
 
387
        self.assertIsInstance(osutils.format_local_date(0), unicode)
409
388
        # Testing for the actual value of the local weekday without
410
389
        # duplicating the code from format_date is difficult.
411
390
        # Instead blackbox.test_locale should check for localized
413
392
 
414
393
    def test_format_date_with_offset_in_original_timezone(self):
415
394
        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
416
 
                         osutils.format_date_with_offset_in_original_timezone(0))
 
395
            osutils.format_date_with_offset_in_original_timezone(0))
417
396
        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
418
 
                         osutils.format_date_with_offset_in_original_timezone(100000))
 
397
            osutils.format_date_with_offset_in_original_timezone(100000))
419
398
        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
420
 
                         osutils.format_date_with_offset_in_original_timezone(100000, 7200))
 
399
            osutils.format_date_with_offset_in_original_timezone(100000, 7200))
421
400
 
422
401
    def test_local_time_offset(self):
423
402
        """Test that local_time_offset() returns a sane value."""
439
418
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
440
419
 
441
420
 
442
 
class TestFdatasync(tests.TestCaseInTempDir):
443
 
 
444
 
    def do_fdatasync(self):
445
 
        f = tempfile.NamedTemporaryFile()
446
 
        osutils.fdatasync(f.fileno())
447
 
        f.close()
448
 
 
449
 
    @staticmethod
450
 
    def raise_eopnotsupp(*args, **kwargs):
451
 
        raise IOError(errno.EOPNOTSUPP, os.strerror(errno.EOPNOTSUPP))
452
 
 
453
 
    @staticmethod
454
 
    def raise_enotsup(*args, **kwargs):
455
 
        raise IOError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
456
 
 
457
 
    def test_fdatasync_handles_system_function(self):
458
 
        self.overrideAttr(os, "fdatasync")
459
 
        self.do_fdatasync()
460
 
 
461
 
    def test_fdatasync_handles_no_fdatasync_no_fsync(self):
462
 
        self.overrideAttr(os, "fdatasync")
463
 
        self.overrideAttr(os, "fsync")
464
 
        self.do_fdatasync()
465
 
 
466
 
    def test_fdatasync_handles_no_EOPNOTSUPP(self):
467
 
        self.overrideAttr(errno, "EOPNOTSUPP")
468
 
        self.do_fdatasync()
469
 
 
470
 
    def test_fdatasync_catches_ENOTSUP(self):
471
 
        enotsup = getattr(errno, "ENOTSUP", None)
472
 
        if enotsup is None:
473
 
            raise tests.TestNotApplicable("No ENOTSUP on this platform")
474
 
        self.overrideAttr(os, "fdatasync", self.raise_enotsup)
475
 
        self.do_fdatasync()
476
 
 
477
 
    def test_fdatasync_catches_EOPNOTSUPP(self):
478
 
        enotsup = getattr(errno, "EOPNOTSUPP", None)
479
 
        if enotsup is None:
480
 
            raise tests.TestNotApplicable("No EOPNOTSUPP on this platform")
481
 
        self.overrideAttr(os, "fdatasync", self.raise_eopnotsupp)
482
 
        self.do_fdatasync()
483
 
 
484
 
 
485
421
class TestLinks(tests.TestCaseInTempDir):
486
422
 
487
423
    def test_dereference_path(self):
488
 
        self.requireFeature(features.SymlinkFeature)
 
424
        self.requireFeature(tests.SymlinkFeature)
489
425
        cwd = osutils.realpath('.')
490
426
        os.mkdir('bar')
491
427
        bar_path = osutils.pathjoin(cwd, 'bar')
512
448
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
513
449
 
514
450
    def test_changing_access(self):
515
 
        with open('file', 'w') as f:
516
 
            f.write('monkey')
 
451
        f = file('file', 'w')
 
452
        f.write('monkey')
 
453
        f.close()
517
454
 
518
455
        # Make a file readonly
519
456
        osutils.make_readonly('file')
520
457
        mode = os.lstat('file').st_mode
521
 
        self.assertEqual(mode, mode & 0o777555)
 
458
        self.assertEqual(mode, mode & 0777555)
522
459
 
523
460
        # Make a file writable
524
461
        osutils.make_writable('file')
525
462
        mode = os.lstat('file').st_mode
526
 
        self.assertEqual(mode, mode | 0o200)
 
463
        self.assertEqual(mode, mode | 0200)
527
464
 
528
465
        if osutils.has_symlinks():
529
466
            # should not error when handed a symlink
537
474
 
538
475
class TestCanonicalRelPath(tests.TestCaseInTempDir):
539
476
 
540
 
    _test_needs_features = [features.CaseInsCasePresFilenameFeature]
 
477
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
541
478
 
542
479
    def test_canonical_relpath_simple(self):
543
 
        f = open('MixedCaseName', 'w')
 
480
        f = file('MixedCaseName', 'w')
544
481
        f.close()
545
482
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
546
 
        self.assertEqual('work/MixedCaseName', actual)
 
483
        self.failUnlessEqual('work/MixedCaseName', actual)
547
484
 
548
485
    def test_canonical_relpath_missing_tail(self):
549
486
        os.mkdir('MixedCaseParent')
550
487
        actual = osutils.canonical_relpath(self.test_base_dir,
551
488
                                           'mixedcaseparent/nochild')
552
 
        self.assertEqual('work/MixedCaseParent/nochild', actual)
 
489
        self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
553
490
 
554
491
 
555
492
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
599
536
    """Test pumpfile method."""
600
537
 
601
538
    def setUp(self):
602
 
        super(TestPumpFile, self).setUp()
 
539
        tests.TestCase.setUp(self)
603
540
        # create a test datablock
604
541
        self.block_size = 512
605
 
        pattern = b'0123456789ABCDEF'
606
 
        self.test_data = pattern * (3 * self.block_size // len(pattern))
 
542
        pattern = '0123456789ABCDEF'
 
543
        self.test_data = pattern * (3 * self.block_size / len(pattern))
607
544
        self.test_data_len = len(self.test_data)
608
545
 
609
546
    def test_bracket_block_size(self):
613
550
        self.assertTrue(self.test_data_len > self.block_size)
614
551
 
615
552
        from_file = file_utils.FakeReadFile(self.test_data)
616
 
        to_file = BytesIO()
 
553
        to_file = StringIO()
617
554
 
618
 
        # read (max // 2) bytes and verify read size wasn't affected
619
 
        num_bytes_to_read = self.block_size // 2
620
 
        osutils.pumpfile(from_file, to_file,
621
 
                         num_bytes_to_read, self.block_size)
 
555
        # read (max / 2) bytes and verify read size wasn't affected
 
556
        num_bytes_to_read = self.block_size / 2
 
557
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
622
558
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
623
559
        self.assertEqual(from_file.get_read_count(), 1)
624
560
 
625
561
        # read (max) bytes and verify read size wasn't affected
626
562
        num_bytes_to_read = self.block_size
627
563
        from_file.reset_read_count()
628
 
        osutils.pumpfile(from_file, to_file,
629
 
                         num_bytes_to_read, self.block_size)
 
564
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
630
565
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
631
566
        self.assertEqual(from_file.get_read_count(), 1)
632
567
 
633
568
        # read (max + 1) bytes and verify read size was limited
634
569
        num_bytes_to_read = self.block_size + 1
635
570
        from_file.reset_read_count()
636
 
        osutils.pumpfile(from_file, to_file,
637
 
                         num_bytes_to_read, self.block_size)
 
571
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
638
572
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
639
573
        self.assertEqual(from_file.get_read_count(), 2)
640
574
 
641
575
        # finish reading the rest of the data
642
576
        num_bytes_to_read = self.test_data_len - to_file.tell()
643
 
        osutils.pumpfile(from_file, to_file,
644
 
                         num_bytes_to_read, self.block_size)
 
577
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
645
578
 
646
579
        # report error if the data wasn't equal (we only report the size due
647
580
        # to the length of the data)
658
591
 
659
592
        # retrieve data in blocks
660
593
        from_file = file_utils.FakeReadFile(self.test_data)
661
 
        to_file = BytesIO()
 
594
        to_file = StringIO()
662
595
        osutils.pumpfile(from_file, to_file, self.test_data_len,
663
596
                         self.block_size)
664
597
 
682
615
 
683
616
        # retrieve data to EOF
684
617
        from_file = file_utils.FakeReadFile(self.test_data)
685
 
        to_file = BytesIO()
 
618
        to_file = StringIO()
686
619
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
687
620
 
688
621
        # verify read size was equal to the maximum read size
702
635
        with this new version."""
703
636
        # retrieve data using default (old) pumpfile method
704
637
        from_file = file_utils.FakeReadFile(self.test_data)
705
 
        to_file = BytesIO()
 
638
        to_file = StringIO()
706
639
        osutils.pumpfile(from_file, to_file)
707
640
 
708
641
        # report error if the data wasn't equal (we only report the size due
714
647
 
715
648
    def test_report_activity(self):
716
649
        activity = []
717
 
 
718
650
        def log_activity(length, direction):
719
651
            activity.append((length, direction))
720
 
        from_file = BytesIO(self.test_data)
721
 
        to_file = BytesIO()
 
652
        from_file = StringIO(self.test_data)
 
653
        to_file = StringIO()
722
654
        osutils.pumpfile(from_file, to_file, buff_size=500,
723
655
                         report_activity=log_activity, direction='read')
724
656
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
725
657
                          (36, 'read')], activity)
726
658
 
727
 
        from_file = BytesIO(self.test_data)
728
 
        to_file = BytesIO()
 
659
        from_file = StringIO(self.test_data)
 
660
        to_file = StringIO()
729
661
        del activity[:]
730
662
        osutils.pumpfile(from_file, to_file, buff_size=500,
731
663
                         report_activity=log_activity, direction='write')
733
665
                          (36, 'write')], activity)
734
666
 
735
667
        # And with a limited amount of data
736
 
        from_file = BytesIO(self.test_data)
737
 
        to_file = BytesIO()
 
668
        from_file = StringIO(self.test_data)
 
669
        to_file = StringIO()
738
670
        del activity[:]
739
671
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
740
672
                         report_activity=log_activity, direction='read')
741
 
        self.assertEqual(
742
 
            [(500, 'read'), (500, 'read'), (28, 'read')], activity)
 
673
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
 
674
 
743
675
 
744
676
 
745
677
class TestPumpStringFile(tests.TestCase):
746
678
 
747
679
    def test_empty(self):
748
 
        output = BytesIO()
749
 
        osutils.pump_string_file(b"", output)
750
 
        self.assertEqual(b"", output.getvalue())
 
680
        output = StringIO()
 
681
        osutils.pump_string_file("", output)
 
682
        self.assertEqual("", output.getvalue())
751
683
 
752
684
    def test_more_than_segment_size(self):
753
 
        output = BytesIO()
754
 
        osutils.pump_string_file(b"123456789", output, 2)
755
 
        self.assertEqual(b"123456789", output.getvalue())
 
685
        output = StringIO()
 
686
        osutils.pump_string_file("123456789", output, 2)
 
687
        self.assertEqual("123456789", output.getvalue())
756
688
 
757
689
    def test_segment_size(self):
758
 
        output = BytesIO()
759
 
        osutils.pump_string_file(b"12", output, 2)
760
 
        self.assertEqual(b"12", output.getvalue())
 
690
        output = StringIO()
 
691
        osutils.pump_string_file("12", output, 2)
 
692
        self.assertEqual("12", output.getvalue())
761
693
 
762
694
    def test_segment_size_multiple(self):
763
 
        output = BytesIO()
764
 
        osutils.pump_string_file(b"1234", output, 2)
765
 
        self.assertEqual(b"1234", output.getvalue())
 
695
        output = StringIO()
 
696
        osutils.pump_string_file("1234", output, 2)
 
697
        self.assertEqual("1234", output.getvalue())
766
698
 
767
699
 
768
700
class TestRelpath(tests.TestCase):
787
719
class TestSafeUnicode(tests.TestCase):
788
720
 
789
721
    def test_from_ascii_string(self):
790
 
        self.assertEqual(u'foobar', osutils.safe_unicode(b'foobar'))
 
722
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
791
723
 
792
724
    def test_from_unicode_string_ascii_contents(self):
793
725
        self.assertEqual(u'bargam', osutils.safe_unicode(u'bargam'))
796
728
        self.assertEqual(u'bargam\xae', osutils.safe_unicode(u'bargam\xae'))
797
729
 
798
730
    def test_from_utf8_string(self):
799
 
        self.assertEqual(u'foo\xae', osutils.safe_unicode(b'foo\xc2\xae'))
 
731
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
800
732
 
801
733
    def test_bad_utf8_string(self):
802
734
        self.assertRaises(errors.BzrBadParameterNotUnicode,
803
735
                          osutils.safe_unicode,
804
 
                          b'\xbb\xbb')
 
736
                          '\xbb\xbb')
805
737
 
806
738
 
807
739
class TestSafeUtf8(tests.TestCase):
808
740
 
809
741
    def test_from_ascii_string(self):
810
 
        f = b'foobar'
811
 
        self.assertEqual(b'foobar', osutils.safe_utf8(f))
 
742
        f = 'foobar'
 
743
        self.assertEqual('foobar', osutils.safe_utf8(f))
812
744
 
813
745
    def test_from_unicode_string_ascii_contents(self):
814
 
        self.assertEqual(b'bargam', osutils.safe_utf8(u'bargam'))
 
746
        self.assertEqual('bargam', osutils.safe_utf8(u'bargam'))
815
747
 
816
748
    def test_from_unicode_string_unicode_contents(self):
817
 
        self.assertEqual(b'bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
 
749
        self.assertEqual('bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
818
750
 
819
751
    def test_from_utf8_string(self):
820
 
        self.assertEqual(b'foo\xc2\xae', osutils.safe_utf8(b'foo\xc2\xae'))
 
752
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
821
753
 
822
754
    def test_bad_utf8_string(self):
823
755
        self.assertRaises(errors.BzrBadParameterNotUnicode,
824
 
                          osutils.safe_utf8, b'\xbb\xbb')
 
756
                          osutils.safe_utf8, '\xbb\xbb')
825
757
 
826
758
 
827
759
class TestSafeRevisionId(tests.TestCase):
828
760
 
829
761
    def test_from_ascii_string(self):
830
762
        # this shouldn't give a warning because it's getting an ascii string
831
 
        self.assertEqual(b'foobar', osutils.safe_revision_id(b'foobar'))
 
763
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
832
764
 
833
765
    def test_from_unicode_string_ascii_contents(self):
834
 
        self.assertRaises(TypeError,
835
 
                          osutils.safe_revision_id, u'bargam')
 
766
        self.assertEqual('bargam',
 
767
                         osutils.safe_revision_id(u'bargam', warn=False))
 
768
 
 
769
    def test_from_unicode_deprecated(self):
 
770
        self.assertEqual('bargam',
 
771
            self.callDeprecated([osutils._revision_id_warning],
 
772
                                osutils.safe_revision_id, u'bargam'))
836
773
 
837
774
    def test_from_unicode_string_unicode_contents(self):
838
 
        self.assertRaises(TypeError,
839
 
                          osutils.safe_revision_id, u'bargam\xae')
 
775
        self.assertEqual('bargam\xc2\xae',
 
776
                         osutils.safe_revision_id(u'bargam\xae', warn=False))
840
777
 
841
778
    def test_from_utf8_string(self):
842
 
        self.assertEqual(b'foo\xc2\xae',
843
 
                         osutils.safe_revision_id(b'foo\xc2\xae'))
 
779
        self.assertEqual('foo\xc2\xae',
 
780
                         osutils.safe_revision_id('foo\xc2\xae'))
844
781
 
845
782
    def test_none(self):
846
783
        """Currently, None is a valid revision_id"""
850
787
class TestSafeFileId(tests.TestCase):
851
788
 
852
789
    def test_from_ascii_string(self):
853
 
        self.assertEqual(b'foobar', osutils.safe_file_id(b'foobar'))
 
790
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
854
791
 
855
792
    def test_from_unicode_string_ascii_contents(self):
856
 
        self.assertRaises(TypeError, osutils.safe_file_id, u'bargam')
 
793
        self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
 
794
 
 
795
    def test_from_unicode_deprecated(self):
 
796
        self.assertEqual('bargam',
 
797
            self.callDeprecated([osutils._file_id_warning],
 
798
                                osutils.safe_file_id, u'bargam'))
857
799
 
858
800
    def test_from_unicode_string_unicode_contents(self):
859
 
        self.assertRaises(TypeError,
860
 
                          osutils.safe_file_id, u'bargam\xae')
 
801
        self.assertEqual('bargam\xc2\xae',
 
802
                         osutils.safe_file_id(u'bargam\xae', warn=False))
861
803
 
862
804
    def test_from_utf8_string(self):
863
 
        self.assertEqual(b'foo\xc2\xae',
864
 
                         osutils.safe_file_id(b'foo\xc2\xae'))
 
805
        self.assertEqual('foo\xc2\xae',
 
806
                         osutils.safe_file_id('foo\xc2\xae'))
865
807
 
866
808
    def test_none(self):
867
809
        """Currently, None is a valid revision_id"""
868
810
        self.assertEqual(None, osutils.safe_file_id(None))
869
811
 
870
812
 
871
 
class TestSendAll(tests.TestCase):
872
 
 
873
 
    def test_send_with_disconnected_socket(self):
874
 
        class DisconnectedSocket(object):
875
 
            def __init__(self, err):
876
 
                self.err = err
877
 
 
878
 
            def send(self, content):
879
 
                raise self.err
880
 
 
881
 
            def close(self):
882
 
                pass
883
 
        # All of these should be treated as ConnectionReset
884
 
        errs = []
885
 
        for err_cls in (IOError, socket.error):
886
 
            for errnum in osutils._end_of_stream_errors:
887
 
                errs.append(err_cls(errnum))
888
 
        for err in errs:
889
 
            sock = DisconnectedSocket(err)
890
 
            self.assertRaises(errors.ConnectionReset,
891
 
                              osutils.send_all, sock, b'some more content')
892
 
 
893
 
    def test_send_with_no_progress(self):
894
 
        # See https://bugs.launchpad.net/bzr/+bug/1047309
895
 
        # It seems that paramiko can get into a state where it doesn't error,
896
 
        # but it returns 0 bytes sent for requests over and over again.
897
 
        class NoSendingSocket(object):
898
 
            def __init__(self):
899
 
                self.call_count = 0
900
 
 
901
 
            def send(self, bytes):
902
 
                self.call_count += 1
903
 
                if self.call_count > 100:
904
 
                    # Prevent the test suite from hanging
905
 
                    raise RuntimeError('too many calls')
906
 
                return 0
907
 
        sock = NoSendingSocket()
908
 
        self.assertRaises(errors.ConnectionReset,
909
 
                          osutils.send_all, sock, b'content')
910
 
        self.assertEqual(1, sock.call_count)
911
 
 
912
 
 
913
 
class TestPosixFuncs(tests.TestCase):
914
 
    """Test that the posix version of normpath returns an appropriate path
915
 
       when used with 2 leading slashes."""
916
 
 
917
 
    def test_normpath(self):
918
 
        self.assertEqual('/etc/shadow', osutils._posix_normpath('/etc/shadow'))
919
 
        self.assertEqual(
920
 
            '/etc/shadow', osutils._posix_normpath('//etc/shadow'))
921
 
        self.assertEqual(
922
 
            '/etc/shadow', osutils._posix_normpath('///etc/shadow'))
923
 
 
924
 
 
925
813
class TestWin32Funcs(tests.TestCase):
926
814
    """Test that _win32 versions of os utilities return appropriate paths."""
927
815
 
928
816
    def test_abspath(self):
929
 
        self.requireFeature(features.win32_feature)
930
817
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
931
818
        self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
932
819
        self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
945
832
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
946
833
        self.assertEqual('path/to/foo',
947
834
                         osutils._win32_pathjoin('path/to/', 'foo'))
948
 
 
949
 
    def test_pathjoin_late_bugfix(self):
950
 
        if sys.version_info < (2, 7, 6):
951
 
            expected = '/foo'
952
 
        else:
953
 
            expected = 'C:/foo'
954
 
        self.assertEqual(expected,
 
835
        self.assertEqual('/foo',
955
836
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
956
 
        self.assertEqual(expected,
 
837
        self.assertEqual('/foo',
957
838
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
958
839
 
959
840
    def test_normpath(self):
964
845
 
965
846
    def test_getcwd(self):
966
847
        cwd = osutils._win32_getcwd()
967
 
        os_cwd = osutils._getcwd()
 
848
        os_cwd = os.getcwdu()
968
849
        self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
969
850
        # win32 is inconsistent whether it returns lower or upper case
970
851
        # and even if it was consistent the user might type the other
978
859
        self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
979
860
        self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
980
861
 
 
862
    def test_win98_abspath(self):
 
863
        # absolute path
 
864
        self.assertEqual('C:/foo', osutils._win98_abspath('C:\\foo'))
 
865
        self.assertEqual('C:/foo', osutils._win98_abspath('C:/foo'))
 
866
        # UNC path
 
867
        self.assertEqual('//HOST/path', osutils._win98_abspath(r'\\HOST\path'))
 
868
        self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
 
869
        # relative path
 
870
        cwd = osutils.getcwd().rstrip('/')
 
871
        drive = osutils._nt_splitdrive(cwd)[0]
 
872
        self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
 
873
        self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
 
874
        # unicode path
 
875
        u = u'\u1234'
 
876
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
 
877
 
981
878
 
982
879
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
983
880
    """Test win32 functions that create files."""
984
881
 
985
882
    def test_getcwd(self):
986
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
883
        self.requireFeature(tests.UnicodeFilenameFeature)
987
884
        os.mkdir(u'mu-\xb5')
988
885
        os.chdir(u'mu-\xb5')
989
886
        # TODO: jam 20060427 This will probably fail on Mac OSX because
994
891
 
995
892
    def test_minimum_path_selection(self):
996
893
        self.assertEqual(set(),
997
 
                         osutils.minimum_path_selection([]))
998
 
        self.assertEqual({'a'},
999
 
                         osutils.minimum_path_selection(['a']))
1000
 
        self.assertEqual({'a', 'b'},
1001
 
                         osutils.minimum_path_selection(['a', 'b']))
1002
 
        self.assertEqual({'a/', 'b'},
1003
 
                         osutils.minimum_path_selection(['a/', 'b']))
1004
 
        self.assertEqual({'a/', 'b'},
1005
 
                         osutils.minimum_path_selection(['a/c', 'a/', 'b']))
1006
 
        self.assertEqual({'a-b', 'a', 'a0b'},
1007
 
                         osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
 
894
            osutils.minimum_path_selection([]))
 
895
        self.assertEqual(set(['a']),
 
896
            osutils.minimum_path_selection(['a']))
 
897
        self.assertEqual(set(['a', 'b']),
 
898
            osutils.minimum_path_selection(['a', 'b']))
 
899
        self.assertEqual(set(['a/', 'b']),
 
900
            osutils.minimum_path_selection(['a/', 'b']))
 
901
        self.assertEqual(set(['a/', 'b']),
 
902
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
 
903
        self.assertEqual(set(['a-b', 'a', 'a0b']),
 
904
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
1008
905
 
1009
906
    def test_mkdtemp(self):
1010
907
        tmpdir = osutils._win32_mkdtemp(dir='.')
1011
908
        self.assertFalse('\\' in tmpdir)
1012
909
 
1013
910
    def test_rename(self):
1014
 
        with open('a', 'wb') as a:
1015
 
            a.write(b'foo\n')
1016
 
        with open('b', 'wb') as b:
1017
 
            b.write(b'baz\n')
 
911
        a = open('a', 'wb')
 
912
        a.write('foo\n')
 
913
        a.close()
 
914
        b = open('b', 'wb')
 
915
        b.write('baz\n')
 
916
        b.close()
1018
917
 
1019
918
        osutils._win32_rename('b', 'a')
1020
 
        self.assertPathExists('a')
1021
 
        self.assertPathDoesNotExist('b')
1022
 
        self.assertFileEqual(b'baz\n', 'a')
 
919
        self.failUnlessExists('a')
 
920
        self.failIfExists('b')
 
921
        self.assertFileEqual('baz\n', 'a')
1023
922
 
1024
923
    def test_rename_missing_file(self):
1025
 
        with open('a', 'wb') as a:
1026
 
            a.write(b'foo\n')
 
924
        a = open('a', 'wb')
 
925
        a.write('foo\n')
 
926
        a.close()
1027
927
 
1028
928
        try:
1029
929
            osutils._win32_rename('b', 'a')
1030
 
        except (IOError, OSError) as e:
 
930
        except (IOError, OSError), e:
1031
931
            self.assertEqual(errno.ENOENT, e.errno)
1032
 
        self.assertFileEqual(b'foo\n', 'a')
 
932
        self.assertFileEqual('foo\n', 'a')
1033
933
 
1034
934
    def test_rename_missing_dir(self):
1035
935
        os.mkdir('a')
1036
936
        try:
1037
937
            osutils._win32_rename('b', 'a')
1038
 
        except (IOError, OSError) as e:
 
938
        except (IOError, OSError), e:
1039
939
            self.assertEqual(errno.ENOENT, e.errno)
1040
940
 
1041
941
    def test_rename_current_dir(self):
1047
947
        # doesn't exist.
1048
948
        try:
1049
949
            osutils._win32_rename('b', '.')
1050
 
        except (IOError, OSError) as e:
 
950
        except (IOError, OSError), e:
1051
951
            self.assertEqual(errno.ENOENT, e.errno)
1052
952
 
1053
953
    def test_splitpath(self):
1058
958
        check(['a', 'b'], 'a/b')
1059
959
        check(['a', 'b'], 'a/./b')
1060
960
        check(['a', '.b'], 'a/.b')
1061
 
        if os.path.sep == '\\':
1062
 
            check(['a', '.b'], 'a\\.b')
1063
 
        else:
1064
 
            check(['a\\.b'], 'a\\.b')
 
961
        check(['a', '.b'], 'a\\.b')
1065
962
 
1066
963
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
1067
964
 
1079
976
    """Test mac special functions that require directories."""
1080
977
 
1081
978
    def test_getcwd(self):
1082
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
979
        self.requireFeature(tests.UnicodeFilenameFeature)
1083
980
        os.mkdir(u'B\xe5gfors')
1084
981
        os.chdir(u'B\xe5gfors')
1085
982
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
1086
983
 
1087
984
    def test_getcwd_nonnorm(self):
1088
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
985
        self.requireFeature(tests.UnicodeFilenameFeature)
1089
986
        # Test that _mac_getcwd() will normalize this path
1090
987
        os.mkdir(u'Ba\u030agfors')
1091
988
        os.chdir(u'Ba\u030agfors')
1095
992
class TestChunksToLines(tests.TestCase):
1096
993
 
1097
994
    def test_smoketest(self):
1098
 
        self.assertEqual([b'foo\n', b'bar\n', b'baz\n'],
1099
 
                         osutils.chunks_to_lines([b'foo\nbar', b'\nbaz\n']))
1100
 
        self.assertEqual([b'foo\n', b'bar\n', b'baz\n'],
1101
 
                         osutils.chunks_to_lines([b'foo\n', b'bar\n', b'baz\n']))
 
995
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
996
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
 
997
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
998
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
1102
999
 
1103
1000
    def test_osutils_binding(self):
1104
 
        from . import test__chunks_to_lines
 
1001
        from bzrlib.tests import test__chunks_to_lines
1105
1002
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
1106
 
            from .._chunks_to_lines_pyx import chunks_to_lines
 
1003
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
1107
1004
        else:
1108
 
            from .._chunks_to_lines_py import chunks_to_lines
 
1005
            from bzrlib._chunks_to_lines_py import chunks_to_lines
1109
1006
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1110
1007
 
1111
1008
 
1118
1015
                         osutils.split_lines(u'foo\nbar\xae\n'))
1119
1016
 
1120
1017
    def test_split_with_carriage_returns(self):
1121
 
        self.assertEqual([b'foo\rbar\n'],
1122
 
                         osutils.split_lines(b'foo\rbar\n'))
 
1018
        self.assertEqual(['foo\rbar\n'],
 
1019
                         osutils.split_lines('foo\rbar\n'))
1123
1020
 
1124
1021
 
1125
1022
class TestWalkDirs(tests.TestCaseInTempDir):
1140
1037
            ]
1141
1038
        self.build_tree(tree)
1142
1039
        expected_dirblocks = [
1143
 
            (('', '.'),
1144
 
             [('0file', '0file', 'file'),
1145
 
              ('1dir', '1dir', 'directory'),
1146
 
              ('2file', '2file', 'file'),
1147
 
              ]
1148
 
             ),
1149
 
            (('1dir', './1dir'),
1150
 
             [('1dir/0file', '0file', 'file'),
1151
 
              ('1dir/1dir', '1dir', 'directory'),
1152
 
              ]
1153
 
             ),
1154
 
            (('1dir/1dir', './1dir/1dir'),
1155
 
             [
1156
 
                ]
1157
 
             ),
 
1040
                (('', '.'),
 
1041
                 [('0file', '0file', 'file'),
 
1042
                  ('1dir', '1dir', 'directory'),
 
1043
                  ('2file', '2file', 'file'),
 
1044
                 ]
 
1045
                ),
 
1046
                (('1dir', './1dir'),
 
1047
                 [('1dir/0file', '0file', 'file'),
 
1048
                  ('1dir/1dir', '1dir', 'directory'),
 
1049
                 ]
 
1050
                ),
 
1051
                (('1dir/1dir', './1dir/1dir'),
 
1052
                 [
 
1053
                 ]
 
1054
                ),
1158
1055
            ]
1159
1056
        result = []
1160
1057
        found_bzrdir = False
1174
1071
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
1175
1072
 
1176
1073
    def test_walkdirs_os_error(self):
1177
 
        # <https://bugs.launchpad.net/bzr/+bug/338653>
 
1074
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
1178
1075
        # Pyrex readdir didn't raise useful messages if it had an error
1179
1076
        # reading the directory
1180
1077
        if sys.platform == 'win32':
1181
1078
            raise tests.TestNotApplicable(
1182
1079
                "readdir IOError not tested on win32")
1183
 
        self.requireFeature(features.not_running_as_root)
1184
1080
        os.mkdir("test-unreadable")
1185
1081
        os.chmod("test-unreadable", 0000)
1186
1082
        # must chmod it back so that it can be removed
1187
 
        self.addCleanup(os.chmod, "test-unreadable", 0o700)
 
1083
        self.addCleanup(os.chmod, "test-unreadable", 0700)
1188
1084
        # The error is not raised until the generator is actually evaluated.
1189
1085
        # (It would be ok if it happened earlier but at the moment it
1190
1086
        # doesn't.)
1191
1087
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1192
 
        self.assertEqual('./test-unreadable', osutils.safe_unicode(e.filename))
1193
 
        self.assertEqual(errno.EACCES, e.errno)
 
1088
        self.assertEquals('./test-unreadable', e.filename)
 
1089
        self.assertEquals(errno.EACCES, e.errno)
1194
1090
        # Ensure the message contains the file name
1195
 
        self.assertContainsRe(str(e), "\\./test-unreadable")
1196
 
 
1197
 
    def test_walkdirs_encoding_error(self):
1198
 
        # <https://bugs.launchpad.net/bzr/+bug/488519>
1199
 
        # walkdirs didn't raise a useful message when the filenames
1200
 
        # are not using the filesystem's encoding
1201
 
 
1202
 
        # require a bytestring based filesystem
1203
 
        self.requireFeature(features.ByteStringNamedFilesystem)
1204
 
 
1205
 
        tree = [
1206
 
            '.bzr',
1207
 
            '0file',
1208
 
            '1dir/',
1209
 
            '1dir/0file',
1210
 
            '1dir/1dir/',
1211
 
            '1file'
1212
 
            ]
1213
 
 
1214
 
        self.build_tree(tree)
1215
 
 
1216
 
        # rename the 1file to a latin-1 filename
1217
 
        os.rename(b"./1file", b"\xe8file")
1218
 
        if b"\xe8file" not in os.listdir("."):
1219
 
            self.skipTest("Lack filesystem that preserves arbitrary bytes")
1220
 
 
1221
 
        self._save_platform_info()
1222
 
        osutils._fs_enc = 'UTF-8'
1223
 
 
1224
 
        # this should raise on error
1225
 
        def attempt():
1226
 
            for dirdetail, dirblock in osutils.walkdirs(b'.'):
1227
 
                pass
1228
 
 
1229
 
        self.assertRaises(errors.BadFilenameEncoding, attempt)
 
1091
        self.assertContainsRe(str(e), "\./test-unreadable")
1230
1092
 
1231
1093
    def test__walkdirs_utf8(self):
1232
1094
        tree = [
1239
1101
            ]
1240
1102
        self.build_tree(tree)
1241
1103
        expected_dirblocks = [
1242
 
            (('', '.'),
1243
 
             [('0file', '0file', 'file'),
1244
 
              ('1dir', '1dir', 'directory'),
1245
 
              ('2file', '2file', 'file'),
1246
 
              ]
1247
 
             ),
1248
 
            (('1dir', './1dir'),
1249
 
             [('1dir/0file', '0file', 'file'),
1250
 
              ('1dir/1dir', '1dir', 'directory'),
1251
 
              ]
1252
 
             ),
1253
 
            (('1dir/1dir', './1dir/1dir'),
1254
 
             [
1255
 
                ]
1256
 
             ),
 
1104
                (('', '.'),
 
1105
                 [('0file', '0file', 'file'),
 
1106
                  ('1dir', '1dir', 'directory'),
 
1107
                  ('2file', '2file', 'file'),
 
1108
                 ]
 
1109
                ),
 
1110
                (('1dir', './1dir'),
 
1111
                 [('1dir/0file', '0file', 'file'),
 
1112
                  ('1dir/1dir', '1dir', 'directory'),
 
1113
                 ]
 
1114
                ),
 
1115
                (('1dir/1dir', './1dir/1dir'),
 
1116
                 [
 
1117
                 ]
 
1118
                ),
1257
1119
            ]
1258
1120
        result = []
1259
1121
        found_bzrdir = False
1260
 
        for dirdetail, dirblock in osutils._walkdirs_utf8(b'.'):
1261
 
            if len(dirblock) and dirblock[0][1] == b'.bzr':
 
1122
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
 
1123
            if len(dirblock) and dirblock[0][1] == '.bzr':
1262
1124
                # this tests the filtering of selected paths
1263
1125
                found_bzrdir = True
1264
1126
                del dirblock[0]
1265
 
            dirdetail = (dirdetail[0].decode('utf-8'),
1266
 
                         osutils.safe_unicode(dirdetail[1]))
1267
 
            dirblock = [
1268
 
                (entry[0].decode('utf-8'), entry[1].decode('utf-8'), entry[2])
1269
 
                for entry in dirblock]
1270
1127
            result.append((dirdetail, dirblock))
1271
1128
 
1272
1129
        self.assertTrue(found_bzrdir)
1288
1145
            dirblock[:] = new_dirblock
1289
1146
 
1290
1147
    def _save_platform_info(self):
 
1148
        self.overrideAttr(win32utils, 'winver')
1291
1149
        self.overrideAttr(osutils, '_fs_enc')
1292
1150
        self.overrideAttr(osutils, '_selected_dir_reader')
1293
1151
 
1294
 
    def assertDirReaderIs(self, expected, top):
 
1152
    def assertDirReaderIs(self, expected):
1295
1153
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1296
1154
        # Force it to redetect
1297
1155
        osutils._selected_dir_reader = None
1298
1156
        # Nothing to list, but should still trigger the selection logic
1299
 
        self.assertEqual([((b'', top), [])], list(osutils._walkdirs_utf8('.')))
 
1157
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1300
1158
        self.assertIsInstance(osutils._selected_dir_reader, expected)
1301
1159
 
1302
1160
    def test_force_walkdirs_utf8_fs_utf8(self):
1303
1161
        self.requireFeature(UTF8DirReaderFeature)
1304
1162
        self._save_platform_info()
1305
 
        osutils._fs_enc = 'utf-8'
1306
 
        self.assertDirReaderIs(UTF8DirReaderFeature.module.UTF8DirReader, b".")
 
1163
        win32utils.winver = None # Avoid the win32 detection code
 
1164
        osutils._fs_enc = 'UTF-8'
 
1165
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1307
1166
 
1308
1167
    def test_force_walkdirs_utf8_fs_ascii(self):
1309
1168
        self.requireFeature(UTF8DirReaderFeature)
1310
1169
        self._save_platform_info()
1311
 
        osutils._fs_enc = 'ascii'
1312
 
        self.assertDirReaderIs(
1313
 
            UTF8DirReaderFeature.module.UTF8DirReader, b".")
 
1170
        win32utils.winver = None # Avoid the win32 detection code
 
1171
        osutils._fs_enc = 'US-ASCII'
 
1172
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
1173
 
 
1174
    def test_force_walkdirs_utf8_fs_ANSI(self):
 
1175
        self.requireFeature(UTF8DirReaderFeature)
 
1176
        self._save_platform_info()
 
1177
        win32utils.winver = None # Avoid the win32 detection code
 
1178
        osutils._fs_enc = 'ANSI_X3.4-1968'
 
1179
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1314
1180
 
1315
1181
    def test_force_walkdirs_utf8_fs_latin1(self):
1316
1182
        self._save_platform_info()
1317
 
        osutils._fs_enc = 'iso-8859-1'
1318
 
        self.assertDirReaderIs(osutils.UnicodeDirReader, ".")
 
1183
        win32utils.winver = None # Avoid the win32 detection code
 
1184
        osutils._fs_enc = 'latin1'
 
1185
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1319
1186
 
1320
1187
    def test_force_walkdirs_utf8_nt(self):
1321
1188
        # Disabled because the thunk of the whole walkdirs api is disabled.
1322
1189
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1323
1190
        self._save_platform_info()
1324
 
        from .._walkdirs_win32 import Win32ReadDir
1325
 
        self.assertDirReaderIs(Win32ReadDir, ".")
 
1191
        win32utils.winver = 'Windows NT'
 
1192
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1193
        self.assertDirReaderIs(Win32ReadDir)
 
1194
 
 
1195
    def test_force_walkdirs_utf8_98(self):
 
1196
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1197
        self._save_platform_info()
 
1198
        win32utils.winver = 'Windows 98'
 
1199
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1326
1200
 
1327
1201
    def test_unicode_walkdirs(self):
1328
1202
        """Walkdirs should always return unicode paths."""
1329
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
1203
        self.requireFeature(tests.UnicodeFilenameFeature)
1330
1204
        name0 = u'0file-\xb6'
1331
1205
        name1 = u'1dir-\u062c\u0648'
1332
1206
        name2 = u'2file-\u0633'
1339
1213
            ]
1340
1214
        self.build_tree(tree)
1341
1215
        expected_dirblocks = [
1342
 
            ((u'', u'.'),
1343
 
             [(name0, name0, 'file', './' + name0),
1344
 
              (name1, name1, 'directory', './' + name1),
1345
 
              (name2, name2, 'file', './' + name2),
1346
 
              ]
1347
 
             ),
1348
 
            ((name1, './' + name1),
1349
 
             [(name1 + '/' + name0, name0, 'file', './' + name1
1350
 
               + '/' + name0),
1351
 
              (name1 + '/' + name1, name1, 'directory', './' + name1
1352
 
               + '/' + name1),
1353
 
              ]
1354
 
             ),
1355
 
            ((name1 + '/' + name1, './' + name1 + '/' + name1),
1356
 
             [
1357
 
                ]
1358
 
             ),
 
1216
                ((u'', u'.'),
 
1217
                 [(name0, name0, 'file', './' + name0),
 
1218
                  (name1, name1, 'directory', './' + name1),
 
1219
                  (name2, name2, 'file', './' + name2),
 
1220
                 ]
 
1221
                ),
 
1222
                ((name1, './' + name1),
 
1223
                 [(name1 + '/' + name0, name0, 'file', './' + name1
 
1224
                                                        + '/' + name0),
 
1225
                  (name1 + '/' + name1, name1, 'directory', './' + name1
 
1226
                                                            + '/' + name1),
 
1227
                 ]
 
1228
                ),
 
1229
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
 
1230
                 [
 
1231
                 ]
 
1232
                ),
1359
1233
            ]
1360
1234
        result = list(osutils.walkdirs('.'))
1361
1235
        self._filter_out_stat(result)
1362
1236
        self.assertEqual(expected_dirblocks, result)
1363
 
        result = list(osutils.walkdirs(u'./' + name1, name1))
 
1237
        result = list(osutils.walkdirs(u'./'+name1, name1))
1364
1238
        self._filter_out_stat(result)
1365
1239
        self.assertEqual(expected_dirblocks[1:], result)
1366
1240
 
1369
1243
 
1370
1244
        The abspath portion might be in unicode or utf-8
1371
1245
        """
1372
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
1246
        self.requireFeature(tests.UnicodeFilenameFeature)
1373
1247
        name0 = u'0file-\xb6'
1374
1248
        name1 = u'1dir-\u062c\u0648'
1375
1249
        name2 = u'2file-\u0633'
1386
1260
        name2 = name2.encode('utf8')
1387
1261
 
1388
1262
        expected_dirblocks = [
1389
 
            ((b'', b'.'),
1390
 
             [(name0, name0, 'file', b'./' + name0),
1391
 
              (name1, name1, 'directory', b'./' + name1),
1392
 
              (name2, name2, 'file', b'./' + name2),
1393
 
              ]
1394
 
             ),
1395
 
            ((name1, b'./' + name1),
1396
 
             [(name1 + b'/' + name0, name0, 'file', b'./' + name1
1397
 
               + b'/' + name0),
1398
 
              (name1 + b'/' + name1, name1, 'directory', b'./' + name1
1399
 
               + b'/' + name1),
1400
 
              ]
1401
 
             ),
1402
 
            ((name1 + b'/' + name1, b'./' + name1 + b'/' + name1),
1403
 
             [
1404
 
                ]
1405
 
             ),
 
1263
                (('', '.'),
 
1264
                 [(name0, name0, 'file', './' + name0),
 
1265
                  (name1, name1, 'directory', './' + name1),
 
1266
                  (name2, name2, 'file', './' + name2),
 
1267
                 ]
 
1268
                ),
 
1269
                ((name1, './' + name1),
 
1270
                 [(name1 + '/' + name0, name0, 'file', './' + name1
 
1271
                                                        + '/' + name0),
 
1272
                  (name1 + '/' + name1, name1, 'directory', './' + name1
 
1273
                                                            + '/' + name1),
 
1274
                 ]
 
1275
                ),
 
1276
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
 
1277
                 [
 
1278
                 ]
 
1279
                ),
1406
1280
            ]
1407
1281
        result = []
1408
1282
        # For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1409
1283
        # all abspaths are Unicode, and encode them back into utf8.
1410
1284
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1411
 
            self.assertIsInstance(dirdetail[0], bytes)
1412
 
            if isinstance(dirdetail[1], text_type):
 
1285
            self.assertIsInstance(dirdetail[0], str)
 
1286
            if isinstance(dirdetail[1], unicode):
1413
1287
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1414
1288
                dirblock = [list(info) for info in dirblock]
1415
1289
                for info in dirblock:
1416
 
                    self.assertIsInstance(info[4], text_type)
 
1290
                    self.assertIsInstance(info[4], unicode)
1417
1291
                    info[4] = info[4].encode('utf8')
1418
1292
            new_dirblock = []
1419
1293
            for info in dirblock:
1420
 
                self.assertIsInstance(info[0], bytes)
1421
 
                self.assertIsInstance(info[1], bytes)
1422
 
                self.assertIsInstance(info[4], bytes)
 
1294
                self.assertIsInstance(info[0], str)
 
1295
                self.assertIsInstance(info[1], str)
 
1296
                self.assertIsInstance(info[4], str)
1423
1297
                # Remove the stat information
1424
1298
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1425
1299
            result.append((dirdetail, new_dirblock))
1430
1304
 
1431
1305
        The abspath portion should be in unicode
1432
1306
        """
1433
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
1307
        self.requireFeature(tests.UnicodeFilenameFeature)
1434
1308
        # Use the unicode reader. TODO: split into driver-and-driven unit
1435
1309
        # tests.
1436
1310
        self._save_platform_info()
1453
1327
        # All of the abspaths should be in unicode, all of the relative paths
1454
1328
        # should be in utf8
1455
1329
        expected_dirblocks = [
1456
 
            ((b'', '.'),
1457
 
             [(name0, name0, 'file', './' + name0u),
1458
 
              (name1, name1, 'directory', './' + name1u),
1459
 
              (name2, name2, 'file', './' + name2u),
1460
 
              ]
1461
 
             ),
1462
 
            ((name1, './' + name1u),
1463
 
             [(name1 + b'/' + name0, name0, 'file', './' + name1u
1464
 
               + '/' + name0u),
1465
 
              (name1 + b'/' + name1, name1, 'directory', './' + name1u
1466
 
               + '/' + name1u),
1467
 
              ]
1468
 
             ),
1469
 
            ((name1 + b'/' + name1, './' + name1u + '/' + name1u),
1470
 
             [
1471
 
                ]
1472
 
             ),
 
1330
                (('', '.'),
 
1331
                 [(name0, name0, 'file', './' + name0u),
 
1332
                  (name1, name1, 'directory', './' + name1u),
 
1333
                  (name2, name2, 'file', './' + name2u),
 
1334
                 ]
 
1335
                ),
 
1336
                ((name1, './' + name1u),
 
1337
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1338
                                                        + '/' + name0u),
 
1339
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1340
                                                            + '/' + name1u),
 
1341
                 ]
 
1342
                ),
 
1343
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1344
                 [
 
1345
                 ]
 
1346
                ),
1473
1347
            ]
1474
1348
        result = list(osutils._walkdirs_utf8('.'))
1475
1349
        self._filter_out_stat(result)
1477
1351
 
1478
1352
    def test__walkdirs_utf8_win32readdir(self):
1479
1353
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1480
 
        self.requireFeature(features.UnicodeFilenameFeature)
1481
 
        from .._walkdirs_win32 import Win32ReadDir
 
1354
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1355
        from bzrlib._walkdirs_win32 import Win32ReadDir
1482
1356
        self._save_platform_info()
1483
1357
        osutils._selected_dir_reader = Win32ReadDir()
1484
1358
        name0u = u'0file-\xb6'
1499
1373
        # All of the abspaths should be in unicode, all of the relative paths
1500
1374
        # should be in utf8
1501
1375
        expected_dirblocks = [
1502
 
            (('', '.'),
1503
 
             [(name0, name0, 'file', './' + name0u),
1504
 
              (name1, name1, 'directory', './' + name1u),
1505
 
              (name2, name2, 'file', './' + name2u),
1506
 
              ]
1507
 
             ),
1508
 
            ((name1, './' + name1u),
1509
 
             [(name1 + '/' + name0, name0, 'file', './' + name1u
1510
 
               + '/' + name0u),
1511
 
              (name1 + '/' + name1, name1, 'directory', './' + name1u
1512
 
               + '/' + name1u),
1513
 
              ]
1514
 
             ),
1515
 
            ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1516
 
             [
1517
 
                ]
1518
 
             ),
 
1376
                (('', '.'),
 
1377
                 [(name0, name0, 'file', './' + name0u),
 
1378
                  (name1, name1, 'directory', './' + name1u),
 
1379
                  (name2, name2, 'file', './' + name2u),
 
1380
                 ]
 
1381
                ),
 
1382
                ((name1, './' + name1u),
 
1383
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1384
                                                        + '/' + name0u),
 
1385
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1386
                                                            + '/' + name1u),
 
1387
                 ]
 
1388
                ),
 
1389
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1390
                 [
 
1391
                 ]
 
1392
                ),
1519
1393
            ]
1520
1394
        result = list(osutils._walkdirs_utf8(u'.'))
1521
1395
        self._filter_out_stat(result)
1534
1408
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1535
1409
        """make sure our Stat values are valid"""
1536
1410
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1537
 
        self.requireFeature(features.UnicodeFilenameFeature)
1538
 
        from .._walkdirs_win32 import Win32ReadDir
 
1411
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1412
        from bzrlib._walkdirs_win32 import Win32ReadDir
1539
1413
        name0u = u'0file-\xb6'
1540
1414
        name0 = name0u.encode('utf8')
1541
1415
        self.build_tree([name0u])
1542
1416
        # I hate to sleep() here, but I'm trying to make the ctime different
1543
1417
        # from the mtime
1544
1418
        time.sleep(2)
1545
 
        with open(name0u, 'ab') as f:
1546
 
            f.write(b'just a small update')
 
1419
        f = open(name0u, 'ab')
 
1420
        try:
 
1421
            f.write('just a small update')
 
1422
        finally:
 
1423
            f.close()
1547
1424
 
1548
1425
        result = Win32ReadDir().read_dir('', u'.')
1549
1426
        entry = result[0]
1555
1432
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1556
1433
        """make sure our Stat values are valid"""
1557
1434
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1558
 
        self.requireFeature(features.UnicodeFilenameFeature)
1559
 
        from .._walkdirs_win32 import Win32ReadDir
 
1435
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1436
        from bzrlib._walkdirs_win32 import Win32ReadDir
1560
1437
        name0u = u'0dir-\u062c\u0648'
1561
1438
        name0 = name0u.encode('utf8')
1562
1439
        self.build_tree([name0u + '/'])
1642
1519
        # using the comparison routine shoudl work too:
1643
1520
        self.assertEqual(
1644
1521
            dir_sorted_paths,
1645
 
            sorted(original_paths, key=osutils.path_prefix_key))
 
1522
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1646
1523
 
1647
1524
 
1648
1525
class TestCopyTree(tests.TestCaseInTempDir):
1661
1538
        self.assertEqual(['c'], os.listdir('target/b'))
1662
1539
 
1663
1540
    def test_copy_tree_symlinks(self):
1664
 
        self.requireFeature(features.SymlinkFeature)
 
1541
        self.requireFeature(tests.SymlinkFeature)
1665
1542
        self.build_tree(['source/'])
1666
1543
        os.symlink('a/generic/path', 'source/lnk')
1667
1544
        osutils.copy_tree('source', 'target')
1671
1548
    def test_copy_tree_handlers(self):
1672
1549
        processed_files = []
1673
1550
        processed_links = []
1674
 
 
1675
1551
        def file_handler(from_path, to_path):
1676
1552
            processed_files.append(('f', from_path, to_path))
1677
 
 
1678
1553
        def dir_handler(from_path, to_path):
1679
1554
            processed_files.append(('d', from_path, to_path))
1680
 
 
1681
1555
        def link_handler(from_path, to_path):
1682
1556
            processed_links.append((from_path, to_path))
1683
 
        handlers = {'file': file_handler,
1684
 
                    'directory': dir_handler,
1685
 
                    'symlink': link_handler,
1686
 
                    }
 
1557
        handlers = {'file':file_handler,
 
1558
                    'directory':dir_handler,
 
1559
                    'symlink':link_handler,
 
1560
                   }
1687
1561
 
1688
1562
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1689
1563
        if osutils.has_symlinks():
1694
1568
                          ('f', 'source/a', 'target/a'),
1695
1569
                          ('d', 'source/b', 'target/b'),
1696
1570
                          ('f', 'source/b/c', 'target/b/c'),
1697
 
                          ], processed_files)
1698
 
        self.assertPathDoesNotExist('target')
 
1571
                         ], processed_files)
 
1572
        self.failIfExists('target')
1699
1573
        if osutils.has_symlinks():
1700
1574
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1701
1575
 
1706
1580
    def setUp(self):
1707
1581
        super(TestSetUnsetEnv, self).setUp()
1708
1582
 
1709
 
        self.assertEqual(None, os.environ.get('BRZ_TEST_ENV_VAR'),
 
1583
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
1710
1584
                         'Environment was not cleaned up properly.'
1711
 
                         ' Variable BRZ_TEST_ENV_VAR should not exist.')
1712
 
 
 
1585
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
1713
1586
        def cleanup():
1714
 
            if 'BRZ_TEST_ENV_VAR' in os.environ:
1715
 
                del os.environ['BRZ_TEST_ENV_VAR']
 
1587
            if 'BZR_TEST_ENV_VAR' in os.environ:
 
1588
                del os.environ['BZR_TEST_ENV_VAR']
1716
1589
        self.addCleanup(cleanup)
1717
1590
 
1718
1591
    def test_set(self):
1719
1592
        """Test that we can set an env variable"""
1720
 
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
 
1593
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1721
1594
        self.assertEqual(None, old)
1722
 
        self.assertEqual('foo', os.environ.get('BRZ_TEST_ENV_VAR'))
 
1595
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
1723
1596
 
1724
1597
    def test_double_set(self):
1725
1598
        """Test that we get the old value out"""
1726
 
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
1727
 
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'bar')
 
1599
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1600
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
1728
1601
        self.assertEqual('foo', old)
1729
 
        self.assertEqual('bar', os.environ.get('BRZ_TEST_ENV_VAR'))
 
1602
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
1730
1603
 
1731
1604
    def test_unicode(self):
1732
1605
        """Environment can only contain plain strings
1739
1612
                'Cannot find a unicode character that works in encoding %s'
1740
1613
                % (osutils.get_user_encoding(),))
1741
1614
 
1742
 
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', uni_val)
1743
 
        if PY3:
1744
 
            self.assertEqual(uni_val, os.environ.get('BRZ_TEST_ENV_VAR'))
1745
 
        else:
1746
 
            self.assertEqual(env_val, os.environ.get('BRZ_TEST_ENV_VAR'))
 
1615
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
 
1616
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1747
1617
 
1748
1618
    def test_unset(self):
1749
1619
        """Test that passing None will remove the env var"""
1750
 
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
1751
 
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', None)
 
1620
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1621
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1752
1622
        self.assertEqual('foo', old)
1753
 
        self.assertEqual(None, os.environ.get('BRZ_TEST_ENV_VAR'))
1754
 
        self.assertNotIn('BRZ_TEST_ENV_VAR', os.environ)
 
1623
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
 
1624
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1755
1625
 
1756
1626
 
1757
1627
class TestSizeShaFile(tests.TestCaseInTempDir):
1758
1628
 
1759
1629
    def test_sha_empty(self):
1760
 
        self.build_tree_contents([('foo', b'')])
1761
 
        expected_sha = osutils.sha_string(b'')
 
1630
        self.build_tree_contents([('foo', '')])
 
1631
        expected_sha = osutils.sha_string('')
1762
1632
        f = open('foo')
1763
1633
        self.addCleanup(f.close)
1764
1634
        size, sha = osutils.size_sha_file(f)
1766
1636
        self.assertEqual(expected_sha, sha)
1767
1637
 
1768
1638
    def test_sha_mixed_endings(self):
1769
 
        text = b'test\r\nwith\nall\rpossible line endings\r\n'
 
1639
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1770
1640
        self.build_tree_contents([('foo', text)])
1771
1641
        expected_sha = osutils.sha_string(text)
1772
1642
        f = open('foo', 'rb')
1779
1649
class TestShaFileByName(tests.TestCaseInTempDir):
1780
1650
 
1781
1651
    def test_sha_empty(self):
1782
 
        self.build_tree_contents([('foo', b'')])
1783
 
        expected_sha = osutils.sha_string(b'')
 
1652
        self.build_tree_contents([('foo', '')])
 
1653
        expected_sha = osutils.sha_string('')
1784
1654
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1785
1655
 
1786
1656
    def test_sha_mixed_endings(self):
1787
 
        text = b'test\r\nwith\nall\rpossible line endings\r\n'
 
1657
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1788
1658
        self.build_tree_contents([('foo', text)])
1789
1659
        expected_sha = osutils.sha_string(text)
1790
1660
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1793
1663
class TestResourceLoading(tests.TestCaseInTempDir):
1794
1664
 
1795
1665
    def test_resource_string(self):
1796
 
        # test resource in breezy
1797
 
        text = osutils.resource_string('breezy', 'debug.py')
 
1666
        # test resource in bzrlib
 
1667
        text = osutils.resource_string('bzrlib', 'debug.py')
1798
1668
        self.assertContainsRe(text, "debug_flags = set()")
1799
 
        # test resource under breezy
1800
 
        text = osutils.resource_string('breezy.ui', 'text.py')
 
1669
        # test resource under bzrlib
 
1670
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1801
1671
        self.assertContainsRe(text, "class TextUIFactory")
1802
1672
        # test unsupported package
1803
1673
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1804
 
                          'yyy.xx')
 
1674
            'yyy.xx')
1805
1675
        # test unknown resource
1806
 
        self.assertRaises(IOError, osutils.resource_string, 'breezy', 'yyy.xx')
 
1676
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
 
1677
 
 
1678
 
 
1679
class TestReCompile(tests.TestCase):
 
1680
 
 
1681
    def test_re_compile_checked(self):
 
1682
        r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
 
1683
        self.assertTrue(r.match('aaaa'))
 
1684
        self.assertTrue(r.match('aAaA'))
 
1685
 
 
1686
    def test_re_compile_checked_error(self):
 
1687
        # like https://bugs.launchpad.net/bzr/+bug/251352
 
1688
        err = self.assertRaises(
 
1689
            errors.BzrCommandError,
 
1690
            osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
 
1691
        self.assertEqual(
 
1692
            "Invalid regular expression in test case: '*': "
 
1693
            "nothing to repeat",
 
1694
            str(err))
1807
1695
 
1808
1696
 
1809
1697
class TestDirReader(tests.TestCaseInTempDir):
1810
1698
 
1811
 
    scenarios = dir_reader_scenarios()
1812
 
 
1813
1699
    # Set by load_tests
1814
1700
    _dir_reader_class = None
1815
1701
    _native_to_unicode = None
1816
1702
 
1817
1703
    def setUp(self):
1818
 
        super(TestDirReader, self).setUp()
 
1704
        tests.TestCaseInTempDir.setUp(self)
1819
1705
        self.overrideAttr(osutils,
1820
1706
                          '_selected_dir_reader', self._dir_reader_class())
1821
1707
 
1828
1714
            '2file'
1829
1715
            ]
1830
1716
        expected_dirblocks = [
1831
 
            ((b'', '.'),
1832
 
             [(b'0file', b'0file', 'file', './0file'),
1833
 
              (b'1dir', b'1dir', 'directory', './1dir'),
1834
 
              (b'2file', b'2file', 'file', './2file'),
1835
 
              ]
1836
 
             ),
1837
 
            ((b'1dir', './1dir'),
1838
 
             [(b'1dir/0file', b'0file', 'file', './1dir/0file'),
1839
 
              (b'1dir/1dir', b'1dir', 'directory', './1dir/1dir'),
1840
 
              ]
1841
 
             ),
1842
 
            ((b'1dir/1dir', './1dir/1dir'),
1843
 
             [
1844
 
                ]
1845
 
             ),
 
1717
                (('', '.'),
 
1718
                 [('0file', '0file', 'file'),
 
1719
                  ('1dir', '1dir', 'directory'),
 
1720
                  ('2file', '2file', 'file'),
 
1721
                 ]
 
1722
                ),
 
1723
                (('1dir', './1dir'),
 
1724
                 [('1dir/0file', '0file', 'file'),
 
1725
                  ('1dir/1dir', '1dir', 'directory'),
 
1726
                 ]
 
1727
                ),
 
1728
                (('1dir/1dir', './1dir/1dir'),
 
1729
                 [
 
1730
                 ]
 
1731
                ),
1846
1732
            ]
1847
1733
        return tree, expected_dirblocks
1848
1734
 
1852
1738
        result = list(osutils._walkdirs_utf8('.'))
1853
1739
        # Filter out stat and abspath
1854
1740
        self.assertEqual(expected_dirblocks,
1855
 
                         self._filter_out(result))
 
1741
                         [(dirinfo, [line[0:3] for line in block])
 
1742
                          for dirinfo, block in result])
1856
1743
 
1857
1744
    def test_walk_sub_dir(self):
1858
1745
        tree, expected_dirblocks = self._get_ascii_tree()
1859
1746
        self.build_tree(tree)
1860
1747
        # you can search a subdir only, with a supplied prefix.
1861
 
        result = list(osutils._walkdirs_utf8(b'./1dir', b'1dir'))
 
1748
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1862
1749
        # Filter out stat and abspath
1863
1750
        self.assertEqual(expected_dirblocks[1:],
1864
 
                         self._filter_out(result))
 
1751
                         [(dirinfo, [line[0:3] for line in block])
 
1752
                          for dirinfo, block in result])
1865
1753
 
1866
1754
    def _get_unicode_tree(self):
1867
1755
        name0u = u'0file-\xb6'
1878
1766
        name1 = name1u.encode('UTF-8')
1879
1767
        name2 = name2u.encode('UTF-8')
1880
1768
        expected_dirblocks = [
1881
 
            ((b'', '.'),
1882
 
             [(name0, name0, 'file', './' + name0u),
1883
 
              (name1, name1, 'directory', './' + name1u),
1884
 
              (name2, name2, 'file', './' + name2u),
1885
 
              ]
1886
 
             ),
1887
 
            ((name1, './' + name1u),
1888
 
             [(name1 + b'/' + name0, name0, 'file', './' + name1u
1889
 
               + '/' + name0u),
1890
 
              (name1 + b'/' + name1, name1, 'directory', './' + name1u
1891
 
               + '/' + name1u),
1892
 
              ]
1893
 
             ),
1894
 
            ((name1 + b'/' + name1, './' + name1u + '/' + name1u),
1895
 
             [
1896
 
                ]
1897
 
             ),
 
1769
                (('', '.'),
 
1770
                 [(name0, name0, 'file', './' + name0u),
 
1771
                  (name1, name1, 'directory', './' + name1u),
 
1772
                  (name2, name2, 'file', './' + name2u),
 
1773
                 ]
 
1774
                ),
 
1775
                ((name1, './' + name1u),
 
1776
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1777
                                                        + '/' + name0u),
 
1778
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1779
                                                            + '/' + name1u),
 
1780
                 ]
 
1781
                ),
 
1782
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1783
                 [
 
1784
                 ]
 
1785
                ),
1898
1786
            ]
1899
1787
        return tree, expected_dirblocks
1900
1788
 
1908
1796
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1909
1797
            details = []
1910
1798
            for line in block:
1911
 
                details.append(
1912
 
                    line[0:3] + (self._native_to_unicode(line[4]), ))
 
1799
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1913
1800
            filtered_dirblocks.append((dirinfo, details))
1914
1801
        return filtered_dirblocks
1915
1802
 
1916
1803
    def test_walk_unicode_tree(self):
1917
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
1804
        self.requireFeature(tests.UnicodeFilenameFeature)
1918
1805
        tree, expected_dirblocks = self._get_unicode_tree()
1919
1806
        self.build_tree(tree)
1920
1807
        result = list(osutils._walkdirs_utf8('.'))
1921
1808
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1922
1809
 
1923
1810
    def test_symlink(self):
1924
 
        self.requireFeature(features.SymlinkFeature)
1925
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
1811
        self.requireFeature(tests.SymlinkFeature)
 
1812
        self.requireFeature(tests.UnicodeFilenameFeature)
1926
1813
        target = u'target\N{Euro Sign}'
1927
1814
        link_name = u'l\N{Euro Sign}nk'
1928
1815
        os.symlink(target, link_name)
 
1816
        target_utf8 = target.encode('UTF-8')
1929
1817
        link_name_utf8 = link_name.encode('UTF-8')
1930
1818
        expected_dirblocks = [
1931
 
            ((b'', '.'),
1932
 
             [(link_name_utf8, link_name_utf8,
1933
 
               'symlink', './' + link_name), ],
1934
 
             )]
 
1819
                (('', '.'),
 
1820
                 [(link_name_utf8, link_name_utf8,
 
1821
                   'symlink', './' + link_name),],
 
1822
                 )]
1935
1823
        result = list(osutils._walkdirs_utf8('.'))
1936
1824
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1937
1825
 
1945
1833
    But prior python versions failed to properly encode the passed unicode
1946
1834
    string.
1947
1835
    """
1948
 
    _test_needs_features = [features.SymlinkFeature,
1949
 
                            features.UnicodeFilenameFeature]
 
1836
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1950
1837
 
1951
1838
    def setUp(self):
1952
1839
        super(tests.TestCaseInTempDir, self).setUp()
1955
1842
        os.symlink(self.target, self.link)
1956
1843
 
1957
1844
    def test_os_readlink_link_encoding(self):
1958
 
        self.assertEqual(self.target, os.readlink(self.link))
 
1845
        if sys.version_info < (2, 6):
 
1846
            self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
 
1847
        else:
 
1848
            self.assertEquals(self.target,  os.readlink(self.link))
1959
1849
 
1960
1850
    def test_os_readlink_link_decoding(self):
1961
 
        self.assertEqual(self.target.encode(osutils._fs_enc),
1962
 
                         os.readlink(self.link.encode(osutils._fs_enc)))
 
1851
        self.assertEquals(self.target.encode(osutils._fs_enc),
 
1852
                          os.readlink(self.link.encode(osutils._fs_enc)))
1963
1853
 
1964
1854
 
1965
1855
class TestConcurrency(tests.TestCase):
1973
1863
        self.assertIsInstance(concurrency, int)
1974
1864
 
1975
1865
    def test_local_concurrency_environment_variable(self):
1976
 
        self.overrideEnv('BRZ_CONCURRENCY', '2')
 
1866
        os.environ['BZR_CONCURRENCY'] = '2'
1977
1867
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1978
 
        self.overrideEnv('BRZ_CONCURRENCY', '3')
 
1868
        os.environ['BZR_CONCURRENCY'] = '3'
1979
1869
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1980
 
        self.overrideEnv('BRZ_CONCURRENCY', 'foo')
 
1870
        os.environ['BZR_CONCURRENCY'] = 'foo'
1981
1871
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1982
1872
 
1983
1873
    def test_option_concurrency(self):
1984
 
        self.overrideEnv('BRZ_CONCURRENCY', '1')
 
1874
        os.environ['BZR_CONCURRENCY'] = '1'
1985
1875
        self.run_bzr('rocks --concurrency 42')
1986
 
        # Command line overrides environment variable
1987
 
        self.assertEqual('42', os.environ['BRZ_CONCURRENCY'])
1988
 
        self.assertEqual(42, osutils.local_concurrency(use_cache=False))
 
1876
        # Command line overrides envrionment variable
 
1877
        self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
 
1878
        self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1989
1879
 
1990
1880
 
1991
1881
class TestFailedToLoadExtension(tests.TestCase):
1992
1882
 
1993
1883
    def _try_loading(self):
1994
1884
        try:
1995
 
            import breezy._fictional_extension_py  # noqa: F401
1996
 
        except ImportError as e:
 
1885
            import bzrlib._fictional_extension_py
 
1886
        except ImportError, e:
1997
1887
            osutils.failed_to_load_extension(e)
1998
1888
            return True
1999
1889
 
2004
1894
    def test_failure_to_load(self):
2005
1895
        self._try_loading()
2006
1896
        self.assertLength(1, osutils._extension_load_failures)
2007
 
        if PY3:
2008
 
            self.assertEqual(
2009
 
                osutils._extension_load_failures[0],
2010
 
                "No module named 'breezy._fictional_extension_py'")
2011
 
        else:
2012
 
            self.assertEqual(osutils._extension_load_failures[0],
2013
 
                             "No module named _fictional_extension_py")
 
1897
        self.assertEquals(osutils._extension_load_failures[0],
 
1898
            "No module named _fictional_extension_py")
2014
1899
 
2015
1900
    def test_report_extension_load_failures_no_warning(self):
2016
1901
        self.assertTrue(self._try_loading())
2017
 
        warnings, result = self.callCatchWarnings(
2018
 
            osutils.report_extension_load_failures)
 
1902
        warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
2019
1903
        # it used to give a Python warning; it no longer does
2020
1904
        self.assertLength(0, warnings)
2021
1905
 
2022
1906
    def test_report_extension_load_failures_message(self):
2023
 
        log = BytesIO()
 
1907
        log = StringIO()
2024
1908
        trace.push_log_file(log)
2025
1909
        self.assertTrue(self._try_loading())
2026
1910
        osutils.report_extension_load_failures()
2027
1911
        self.assertContainsRe(
2028
1912
            log.getvalue(),
2029
 
            br"brz: warning: some compiled extensions could not be loaded; "
2030
 
            b"see ``brz help missing-extensions``\n"
 
1913
            r"bzr: warning: some compiled extensions could not be loaded; "
 
1914
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
2031
1915
            )
2032
1916
 
2033
1917
 
2034
1918
class TestTerminalWidth(tests.TestCase):
2035
1919
 
2036
 
    def setUp(self):
2037
 
        super(TestTerminalWidth, self).setUp()
2038
 
        self._orig_terminal_size_state = osutils._terminal_size_state
2039
 
        self._orig_first_terminal_size = osutils._first_terminal_size
2040
 
        self.addCleanup(self.restore_osutils_globals)
2041
 
        osutils._terminal_size_state = 'no_data'
2042
 
        osutils._first_terminal_size = None
2043
 
 
2044
 
    def restore_osutils_globals(self):
2045
 
        osutils._terminal_size_state = self._orig_terminal_size_state
2046
 
        osutils._first_terminal_size = self._orig_first_terminal_size
2047
 
 
2048
1920
    def replace_stdout(self, new):
2049
1921
        self.overrideAttr(sys, 'stdout', new)
2050
1922
 
2062
1934
    def test_default_values(self):
2063
1935
        self.assertEqual(80, osutils.default_terminal_width)
2064
1936
 
2065
 
    def test_defaults_to_BRZ_COLUMNS(self):
2066
 
        # BRZ_COLUMNS is set by the test framework
2067
 
        self.assertNotEqual('12', os.environ['BRZ_COLUMNS'])
2068
 
        self.overrideEnv('BRZ_COLUMNS', '12')
 
1937
    def test_defaults_to_BZR_COLUMNS(self):
 
1938
        # BZR_COLUMNS is set by the test framework
 
1939
        self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
 
1940
        os.environ['BZR_COLUMNS'] = '12'
2069
1941
        self.assertEqual(12, osutils.terminal_width())
2070
1942
 
2071
 
    def test_BRZ_COLUMNS_0_no_limit(self):
2072
 
        self.overrideEnv('BRZ_COLUMNS', '0')
2073
 
        self.assertEqual(None, osutils.terminal_width())
2074
 
 
2075
1943
    def test_falls_back_to_COLUMNS(self):
2076
 
        self.overrideEnv('BRZ_COLUMNS', None)
 
1944
        del os.environ['BZR_COLUMNS']
2077
1945
        self.assertNotEqual('42', os.environ['COLUMNS'])
2078
1946
        self.set_fake_tty()
2079
 
        self.overrideEnv('COLUMNS', '42')
 
1947
        os.environ['COLUMNS'] = '42'
2080
1948
        self.assertEqual(42, osutils.terminal_width())
2081
1949
 
2082
1950
    def test_tty_default_without_columns(self):
2083
 
        self.overrideEnv('BRZ_COLUMNS', None)
2084
 
        self.overrideEnv('COLUMNS', None)
 
1951
        del os.environ['BZR_COLUMNS']
 
1952
        del os.environ['COLUMNS']
2085
1953
 
2086
1954
        def terminal_size(w, h):
2087
1955
            return 42, 42
2094
1962
        self.assertEqual(42, osutils.terminal_width())
2095
1963
 
2096
1964
    def test_non_tty_default_without_columns(self):
2097
 
        self.overrideEnv('BRZ_COLUMNS', None)
2098
 
        self.overrideEnv('COLUMNS', None)
 
1965
        del os.environ['BZR_COLUMNS']
 
1966
        del os.environ['COLUMNS']
2099
1967
        self.replace_stdout(None)
2100
1968
        self.assertEqual(None, osutils.terminal_width())
2101
1969
 
2104
1972
        termios = term_ios_feature.module
2105
1973
        # bug 63539 is about a termios without TIOCGWINSZ attribute
2106
1974
        try:
2107
 
            termios.TIOCGWINSZ
 
1975
            orig = termios.TIOCGWINSZ
2108
1976
        except AttributeError:
2109
1977
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
2110
1978
            pass
2111
1979
        else:
2112
1980
            self.overrideAttr(termios, 'TIOCGWINSZ')
2113
1981
            del termios.TIOCGWINSZ
2114
 
        self.overrideEnv('BRZ_COLUMNS', None)
2115
 
        self.overrideEnv('COLUMNS', None)
 
1982
        del os.environ['BZR_COLUMNS']
 
1983
        del os.environ['COLUMNS']
2116
1984
        # Whatever the result is, if we don't raise an exception, it's ok.
2117
1985
        osutils.terminal_width()
2118
1986
 
2119
 
 
2120
1987
class TestCreationOps(tests.TestCaseInTempDir):
2121
1988
    _test_needs_features = [features.chown_feature]
2122
1989
 
2123
1990
    def setUp(self):
2124
 
        super(TestCreationOps, self).setUp()
 
1991
        tests.TestCaseInTempDir.setUp(self)
2125
1992
        self.overrideAttr(os, 'chown', self._dummy_chown)
2126
1993
 
2127
1994
        # params set by call to _dummy_chown
2133
2000
    def test_copy_ownership_from_path(self):
2134
2001
        """copy_ownership_from_path test with specified src."""
2135
2002
        ownsrc = '/'
2136
 
        open('test_file', 'wt').close()
 
2003
        f = open('test_file', 'wt')
2137
2004
        osutils.copy_ownership_from_path('test_file', ownsrc)
2138
2005
 
2139
2006
        s = os.stat(ownsrc)
2140
 
        self.assertEqual(self.path, 'test_file')
2141
 
        self.assertEqual(self.uid, s.st_uid)
2142
 
        self.assertEqual(self.gid, s.st_gid)
 
2007
        self.assertEquals(self.path, 'test_file')
 
2008
        self.assertEquals(self.uid, s.st_uid)
 
2009
        self.assertEquals(self.gid, s.st_gid)
2143
2010
 
2144
2011
    def test_copy_ownership_nonesrc(self):
2145
2012
        """copy_ownership_from_path test with src=None."""
2146
 
        open('test_file', 'wt').close()
 
2013
        f = open('test_file', 'wt')
2147
2014
        # should use parent dir for permissions
2148
2015
        osutils.copy_ownership_from_path('test_file')
2149
2016
 
2150
2017
        s = os.stat('..')
2151
 
        self.assertEqual(self.path, 'test_file')
2152
 
        self.assertEqual(self.uid, s.st_uid)
2153
 
        self.assertEqual(self.gid, s.st_gid)
2154
 
 
2155
 
 
2156
 
class TestPathFromEnviron(tests.TestCase):
2157
 
 
2158
 
    def test_is_unicode(self):
2159
 
        self.overrideEnv('BRZ_TEST_PATH', './anywhere at all/')
2160
 
        path = osutils.path_from_environ('BRZ_TEST_PATH')
2161
 
        self.assertIsInstance(path, text_type)
2162
 
        self.assertEqual(u'./anywhere at all/', path)
2163
 
 
2164
 
    def test_posix_path_env_ascii(self):
2165
 
        self.overrideEnv('BRZ_TEST_PATH', '/tmp')
2166
 
        home = osutils._posix_path_from_environ('BRZ_TEST_PATH')
2167
 
        self.assertIsInstance(home, text_type)
2168
 
        self.assertEqual(u'/tmp', home)
2169
 
 
2170
 
    def test_posix_path_env_unicode(self):
2171
 
        self.requireFeature(features.ByteStringNamedFilesystem)
2172
 
        self.overrideEnv('BRZ_TEST_PATH', '/home/\xa7test')
2173
 
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2174
 
        self.assertEqual(u'/home/\xa7test',
2175
 
                         osutils._posix_path_from_environ('BRZ_TEST_PATH'))
2176
 
        osutils._fs_enc = "iso8859-5"
2177
 
        if PY3:
2178
 
            # In Python 3, os.environ returns unicode.
2179
 
            self.assertEqual(u'/home/\xa7test',
2180
 
                             osutils._posix_path_from_environ('BRZ_TEST_PATH'))
2181
 
        else:
2182
 
            self.assertEqual(u'/home/\u0407test',
2183
 
                             osutils._posix_path_from_environ('BRZ_TEST_PATH'))
2184
 
            osutils._fs_enc = "utf-8"
2185
 
            self.assertRaises(
2186
 
                errors.BadFilenameEncoding,
2187
 
                osutils._posix_path_from_environ, 'BRZ_TEST_PATH')
2188
 
 
2189
 
 
2190
 
class TestGetHomeDir(tests.TestCase):
2191
 
 
2192
 
    def test_is_unicode(self):
2193
 
        home = osutils._get_home_dir()
2194
 
        self.assertIsInstance(home, text_type)
2195
 
 
2196
 
    def test_posix_homeless(self):
2197
 
        self.overrideEnv('HOME', None)
2198
 
        home = osutils._get_home_dir()
2199
 
        self.assertIsInstance(home, text_type)
2200
 
 
2201
 
    def test_posix_home_ascii(self):
2202
 
        self.overrideEnv('HOME', '/home/test')
2203
 
        home = osutils._posix_get_home_dir()
2204
 
        self.assertIsInstance(home, text_type)
2205
 
        self.assertEqual(u'/home/test', home)
2206
 
 
2207
 
    def test_posix_home_unicode(self):
2208
 
        self.requireFeature(features.ByteStringNamedFilesystem)
2209
 
        self.overrideEnv('HOME', '/home/\xa7test')
2210
 
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2211
 
        self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
2212
 
        osutils._fs_enc = "iso8859-5"
2213
 
        if PY3:
2214
 
            # In python 3, os.environ returns unicode
2215
 
            self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
2216
 
        else:
2217
 
            self.assertEqual(u'/home/\u0407test',
2218
 
                             osutils._posix_get_home_dir())
2219
 
            osutils._fs_enc = "utf-8"
2220
 
            self.assertRaises(errors.BadFilenameEncoding,
2221
 
                              osutils._posix_get_home_dir)
2222
 
 
2223
 
 
2224
 
class TestGetuserUnicode(tests.TestCase):
2225
 
 
2226
 
    def test_is_unicode(self):
2227
 
        user = osutils.getuser_unicode()
2228
 
        self.assertIsInstance(user, text_type)
2229
 
 
2230
 
    def envvar_to_override(self):
2231
 
        if sys.platform == "win32":
2232
 
            # Disable use of platform calls on windows so envvar is used
2233
 
            self.overrideAttr(win32utils, 'has_ctypes', False)
2234
 
            return 'USERNAME'  # only variable used on windows
2235
 
        return 'LOGNAME'  # first variable checked by getpass.getuser()
2236
 
 
2237
 
    def test_ascii_user(self):
2238
 
        self.overrideEnv(self.envvar_to_override(), 'jrandom')
2239
 
        self.assertEqual(u'jrandom', osutils.getuser_unicode())
2240
 
 
2241
 
    def test_unicode_user(self):
2242
 
        ue = osutils.get_user_encoding()
2243
 
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
2244
 
        if uni_val is None:
2245
 
            raise tests.TestSkipped(
2246
 
                'Cannot find a unicode character that works in encoding %s'
2247
 
                % (osutils.get_user_encoding(),))
2248
 
        uni_username = u'jrandom' + uni_val
2249
 
        encoded_username = uni_username.encode(ue)
2250
 
        if PY3:
2251
 
            self.overrideEnv(self.envvar_to_override(), uni_username)
2252
 
        else:
2253
 
            self.overrideEnv(self.envvar_to_override(), encoded_username)
2254
 
        self.assertEqual(uni_username, osutils.getuser_unicode())
2255
 
 
2256
 
 
2257
 
class TestBackupNames(tests.TestCase):
2258
 
 
2259
 
    def setUp(self):
2260
 
        super(TestBackupNames, self).setUp()
2261
 
        self.backups = []
2262
 
 
2263
 
    def backup_exists(self, name):
2264
 
        return name in self.backups
2265
 
 
2266
 
    def available_backup_name(self, name):
2267
 
        backup_name = osutils.available_backup_name(name, self.backup_exists)
2268
 
        self.backups.append(backup_name)
2269
 
        return backup_name
2270
 
 
2271
 
    def assertBackupName(self, expected, name):
2272
 
        self.assertEqual(expected, self.available_backup_name(name))
2273
 
 
2274
 
    def test_empty(self):
2275
 
        self.assertBackupName('file.~1~', 'file')
2276
 
 
2277
 
    def test_existing(self):
2278
 
        self.available_backup_name('file')
2279
 
        self.available_backup_name('file')
2280
 
        self.assertBackupName('file.~3~', 'file')
2281
 
        # Empty slots are found, this is not a strict requirement and may be
2282
 
        # revisited if we test against all implementations.
2283
 
        self.backups.remove('file.~2~')
2284
 
        self.assertBackupName('file.~2~', 'file')
2285
 
 
2286
 
 
2287
 
class TestFindExecutableInPath(tests.TestCase):
2288
 
 
2289
 
    def test_windows(self):
2290
 
        if sys.platform != 'win32':
2291
 
            raise tests.TestSkipped('test requires win32')
2292
 
        self.assertTrue(osutils.find_executable_on_path(
2293
 
            'explorer') is not None)
2294
 
        self.assertTrue(
2295
 
            osutils.find_executable_on_path('explorer.exe') is not None)
2296
 
        self.assertTrue(
2297
 
            osutils.find_executable_on_path('EXPLORER.EXE') is not None)
2298
 
        self.assertTrue(
2299
 
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2300
 
        self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
2301
 
 
2302
 
    def test_windows_app_path(self):
2303
 
        if sys.platform != 'win32':
2304
 
            raise tests.TestSkipped('test requires win32')
2305
 
        # Override PATH env var so that exe can only be found on App Path
2306
 
        self.overrideEnv('PATH', '')
2307
 
        # Internt Explorer is always registered in the App Path
2308
 
        self.assertTrue(osutils.find_executable_on_path(
2309
 
            'iexplore') is not None)
2310
 
 
2311
 
    def test_other(self):
2312
 
        if sys.platform == 'win32':
2313
 
            raise tests.TestSkipped('test requires non-win32')
2314
 
        self.assertTrue(osutils.find_executable_on_path('sh') is not None)
2315
 
        self.assertTrue(
2316
 
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2317
 
 
2318
 
 
2319
 
class TestEnvironmentErrors(tests.TestCase):
2320
 
    """Test handling of environmental errors"""
2321
 
 
2322
 
    def test_is_oserror(self):
2323
 
        self.assertTrue(osutils.is_environment_error(
2324
 
            OSError(errno.EINVAL, "Invalid parameter")))
2325
 
 
2326
 
    def test_is_ioerror(self):
2327
 
        self.assertTrue(osutils.is_environment_error(
2328
 
            IOError(errno.EINVAL, "Invalid parameter")))
2329
 
 
2330
 
    def test_is_socket_error(self):
2331
 
        self.assertTrue(osutils.is_environment_error(
2332
 
            socket.error(errno.EINVAL, "Invalid parameter")))
2333
 
 
2334
 
    def test_is_select_error(self):
2335
 
        self.assertTrue(osutils.is_environment_error(
2336
 
            select.error(errno.EINVAL, "Invalid parameter")))
2337
 
 
2338
 
    def test_is_pywintypes_error(self):
2339
 
        self.requireFeature(features.pywintypes)
2340
 
        import pywintypes
2341
 
        self.assertTrue(osutils.is_environment_error(
2342
 
            pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))
2343
 
 
2344
 
 
2345
 
class SupportsExecutableTests(tests.TestCaseInTempDir):
2346
 
 
2347
 
    def test_returns_bool(self):
2348
 
        self.assertIsInstance(osutils.supports_executable(self.test_dir), bool)
2349
 
 
2350
 
 
2351
 
class SupportsSymlinksTests(tests.TestCaseInTempDir):
2352
 
 
2353
 
    def test_returns_bool(self):
2354
 
        self.assertIsInstance(osutils.supports_symlinks(self.test_dir), bool)
2355
 
 
2356
 
 
2357
 
class MtabReader(tests.TestCaseInTempDir):
2358
 
 
2359
 
    def test_read_mtab(self):
2360
 
        self.build_tree_contents([('mtab', """\
2361
 
/dev/mapper/blah--vg-root / ext4 rw,relatime,errors=remount-ro 0 0
2362
 
/dev/mapper/blah--vg-home /home vfat rw,relatime 0 0
2363
 
# comment
2364
 
 
2365
 
iminvalid
2366
 
""")])
2367
 
        self.assertEqual(
2368
 
            list(osutils.read_mtab('mtab')),
2369
 
            [(b'/', 'ext4'),
2370
 
             (b'/home', 'vfat')])
2371
 
 
2372
 
 
2373
 
class GetFsTypeTests(tests.TestCaseInTempDir):
2374
 
 
2375
 
    def test_returns_string_or_none(self):
2376
 
        ret = osutils.get_fs_type(self.test_dir)
2377
 
        self.assertTrue(isinstance(ret, text_type) or ret is None)
2378
 
 
2379
 
    def test_returns_most_specific(self):
2380
 
        self.overrideAttr(
2381
 
            osutils, '_FILESYSTEM_FINDER',
2382
 
            osutils.FilesystemFinder(
2383
 
                [(b'/', 'ext4'), (b'/home', 'vfat'),
2384
 
                 (b'/home/jelmer', 'ext2')]))
2385
 
        self.assertEqual(osutils.get_fs_type(b'/home/jelmer/blah'), 'ext2')
2386
 
        self.assertEqual(osutils.get_fs_type('/home/jelmer/blah'), 'ext2')
2387
 
        self.assertEqual(osutils.get_fs_type(b'/home/jelmer'), 'ext2')
2388
 
        self.assertEqual(osutils.get_fs_type(b'/home/martin'), 'vfat')
2389
 
        self.assertEqual(osutils.get_fs_type(b'/home'), 'vfat')
2390
 
        self.assertEqual(osutils.get_fs_type(b'/other'), 'ext4')
2391
 
 
2392
 
    def test_returns_none(self):
2393
 
        self.overrideAttr(
2394
 
            osutils, '_FILESYSTEM_FINDER',
2395
 
            osutils.FilesystemFinder([]))
2396
 
        self.assertIs(osutils.get_fs_type('/home/jelmer/blah'), None)
2397
 
        self.assertIs(osutils.get_fs_type(b'/home/jelmer/blah'), None)
2398
 
        self.assertIs(osutils.get_fs_type('/home/jelmer'), None)
 
2018
        self.assertEquals(self.path, 'test_file')
 
2019
        self.assertEquals(self.uid, s.st_uid)
 
2020
        self.assertEquals(self.gid, s.st_gid)