/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: Breezy landing bot
  • Author(s): Jelmer Vernooij
  • Date: 2020-08-23 01:15:41 UTC
  • mfrom: (7520.1.4 merge-3.1)
  • Revision ID: breezy.the.bot@gmail.com-20200823011541-nv0oh7nzaganx2qy
Merge lp:brz/3.1.

Merged from https://code.launchpad.net/~jelmer/brz/merge-3.1/+merge/389690

Show diffs side-by-side

added added

removed removed

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