/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 breezy/tests/test_osutils.py

  • Committer: Jelmer Vernooij
  • Date: 2019-06-15 19:09:08 UTC
  • mto: This revision was merged to the branch mainline in revision 7339.
  • Revision ID: jelmer@jelmer.uk-20190615190908-gyjn0ye90vu2lhim
https://launchpad.net/bugs/237844
Print sensible error message when an invalid argument is specified for an option.

Show diffs side-by-side

added added

removed removed

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