/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to breezy/tests/test_osutils.py

  • Committer: Jelmer Vernooij
  • Date: 2020-02-18 01:57:45 UTC
  • mto: This revision was merged to the branch mainline in revision 7493.
  • Revision ID: jelmer@jelmer.uk-20200218015745-q2ss9tsk74h4nh61
drop use of future.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_osutils.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005-2016 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
19
 
from cStringIO import StringIO
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')
 
818
                          osutils.safe_utf8, b'\xbb\xbb')
757
819
 
758
820
 
759
821
class TestSafeRevisionId(tests.TestCase):
760
822
 
761
823
    def test_from_ascii_string(self):
762
824
        # this shouldn't give a warning because it's getting an ascii string
763
 
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
 
825
        self.assertEqual(b'foobar', osutils.safe_revision_id(b'foobar'))
764
826
 
765
827
    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'))
 
828
        self.assertRaises(TypeError,
 
829
                          osutils.safe_revision_id, u'bargam')
773
830
 
774
831
    def test_from_unicode_string_unicode_contents(self):
775
 
        self.assertEqual('bargam\xc2\xae',
776
 
                         osutils.safe_revision_id(u'bargam\xae', warn=False))
 
832
        self.assertRaises(TypeError,
 
833
                          osutils.safe_revision_id, u'bargam\xae')
777
834
 
778
835
    def test_from_utf8_string(self):
779
 
        self.assertEqual('foo\xc2\xae',
780
 
                         osutils.safe_revision_id('foo\xc2\xae'))
 
836
        self.assertEqual(b'foo\xc2\xae',
 
837
                         osutils.safe_revision_id(b'foo\xc2\xae'))
781
838
 
782
839
    def test_none(self):
783
840
        """Currently, None is a valid revision_id"""
787
844
class TestSafeFileId(tests.TestCase):
788
845
 
789
846
    def test_from_ascii_string(self):
790
 
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
 
847
        self.assertEqual(b'foobar', osutils.safe_file_id(b'foobar'))
791
848
 
792
849
    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'))
 
850
        self.assertRaises(TypeError, osutils.safe_file_id, u'bargam')
799
851
 
800
852
    def test_from_unicode_string_unicode_contents(self):
801
 
        self.assertEqual('bargam\xc2\xae',
802
 
                         osutils.safe_file_id(u'bargam\xae', warn=False))
 
853
        self.assertRaises(TypeError,
 
854
                          osutils.safe_file_id, u'bargam\xae')
803
855
 
804
856
    def test_from_utf8_string(self):
805
 
        self.assertEqual('foo\xc2\xae',
806
 
                         osutils.safe_file_id('foo\xc2\xae'))
 
857
        self.assertEqual(b'foo\xc2\xae',
 
858
                         osutils.safe_file_id(b'foo\xc2\xae'))
807
859
 
808
860
    def test_none(self):
809
861
        """Currently, None is a valid revision_id"""
810
862
        self.assertEqual(None, osutils.safe_file_id(None))
811
863
 
812
864
 
 
865
class TestSendAll(tests.TestCase):
 
866
 
 
867
    def test_send_with_disconnected_socket(self):
 
868
        class DisconnectedSocket(object):
 
869
            def __init__(self, err):
 
870
                self.err = err
 
871
 
 
872
            def send(self, content):
 
873
                raise self.err
 
874
 
 
875
            def close(self):
 
876
                pass
 
877
        # All of these should be treated as ConnectionReset
 
878
        errs = []
 
879
        for err_cls in (IOError, socket.error):
 
880
            for errnum in osutils._end_of_stream_errors:
 
881
                errs.append(err_cls(errnum))
 
882
        for err in errs:
 
883
            sock = DisconnectedSocket(err)
 
884
            self.assertRaises(errors.ConnectionReset,
 
885
                              osutils.send_all, sock, b'some more content')
 
886
 
 
887
    def test_send_with_no_progress(self):
 
888
        # See https://bugs.launchpad.net/bzr/+bug/1047309
 
889
        # It seems that paramiko can get into a state where it doesn't error,
 
890
        # but it returns 0 bytes sent for requests over and over again.
 
891
        class NoSendingSocket(object):
 
892
            def __init__(self):
 
893
                self.call_count = 0
 
894
 
 
895
            def send(self, bytes):
 
896
                self.call_count += 1
 
897
                if self.call_count > 100:
 
898
                    # Prevent the test suite from hanging
 
899
                    raise RuntimeError('too many calls')
 
900
                return 0
 
901
        sock = NoSendingSocket()
 
902
        self.assertRaises(errors.ConnectionReset,
 
903
                          osutils.send_all, sock, b'content')
 
904
        self.assertEqual(1, sock.call_count)
 
905
 
 
906
 
 
907
class TestPosixFuncs(tests.TestCase):
 
908
    """Test that the posix version of normpath returns an appropriate path
 
909
       when used with 2 leading slashes."""
 
910
 
 
911
    def test_normpath(self):
 
912
        self.assertEqual('/etc/shadow', osutils._posix_normpath('/etc/shadow'))
 
913
        self.assertEqual(
 
914
            '/etc/shadow', osutils._posix_normpath('//etc/shadow'))
 
915
        self.assertEqual(
 
916
            '/etc/shadow', osutils._posix_normpath('///etc/shadow'))
 
917
 
 
918
 
813
919
class TestWin32Funcs(tests.TestCase):
814
920
    """Test that _win32 versions of os utilities return appropriate paths."""
815
921
 
816
922
    def test_abspath(self):
 
923
        self.requireFeature(features.win32_feature)
817
924
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
818
925
        self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
819
926
        self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
832
939
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
833
940
        self.assertEqual('path/to/foo',
834
941
                         osutils._win32_pathjoin('path/to/', 'foo'))
835
 
        self.assertEqual('/foo',
 
942
 
 
943
    def test_pathjoin_late_bugfix(self):
 
944
        expected = 'C:/foo'
 
945
        self.assertEqual(expected,
836
946
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
837
 
        self.assertEqual('/foo',
 
947
        self.assertEqual(expected,
838
948
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
839
949
 
840
950
    def test_normpath(self):
845
955
 
846
956
    def test_getcwd(self):
847
957
        cwd = osutils._win32_getcwd()
848
 
        os_cwd = os.getcwdu()
 
958
        os_cwd = osutils._getcwd()
849
959
        self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
850
960
        # win32 is inconsistent whether it returns lower or upper case
851
961
        # and even if it was consistent the user might type the other
859
969
        self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
860
970
        self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
861
971
 
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
972
 
879
973
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
880
974
    """Test win32 functions that create files."""
881
975
 
882
976
    def test_getcwd(self):
883
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
977
        self.requireFeature(features.UnicodeFilenameFeature)
884
978
        os.mkdir(u'mu-\xb5')
885
979
        os.chdir(u'mu-\xb5')
886
980
        # TODO: jam 20060427 This will probably fail on Mac OSX because
891
985
 
892
986
    def test_minimum_path_selection(self):
893
987
        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']))
 
988
                         osutils.minimum_path_selection([]))
 
989
        self.assertEqual({'a'},
 
990
                         osutils.minimum_path_selection(['a']))
 
991
        self.assertEqual({'a', 'b'},
 
992
                         osutils.minimum_path_selection(['a', 'b']))
 
993
        self.assertEqual({'a/', 'b'},
 
994
                         osutils.minimum_path_selection(['a/', 'b']))
 
995
        self.assertEqual({'a/', 'b'},
 
996
                         osutils.minimum_path_selection(['a/c', 'a/', 'b']))
 
997
        self.assertEqual({'a-b', 'a', 'a0b'},
 
998
                         osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
905
999
 
906
1000
    def test_mkdtemp(self):
907
1001
        tmpdir = osutils._win32_mkdtemp(dir='.')
908
1002
        self.assertFalse('\\' in tmpdir)
909
1003
 
910
1004
    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()
 
1005
        with open('a', 'wb') as a:
 
1006
            a.write(b'foo\n')
 
1007
        with open('b', 'wb') as b:
 
1008
            b.write(b'baz\n')
917
1009
 
918
1010
        osutils._win32_rename('b', 'a')
919
 
        self.failUnlessExists('a')
920
 
        self.failIfExists('b')
921
 
        self.assertFileEqual('baz\n', 'a')
 
1011
        self.assertPathExists('a')
 
1012
        self.assertPathDoesNotExist('b')
 
1013
        self.assertFileEqual(b'baz\n', 'a')
922
1014
 
923
1015
    def test_rename_missing_file(self):
924
 
        a = open('a', 'wb')
925
 
        a.write('foo\n')
926
 
        a.close()
 
1016
        with open('a', 'wb') as a:
 
1017
            a.write(b'foo\n')
927
1018
 
928
1019
        try:
929
1020
            osutils._win32_rename('b', 'a')
930
 
        except (IOError, OSError), e:
 
1021
        except (IOError, OSError) as e:
931
1022
            self.assertEqual(errno.ENOENT, e.errno)
932
 
        self.assertFileEqual('foo\n', 'a')
 
1023
        self.assertFileEqual(b'foo\n', 'a')
933
1024
 
934
1025
    def test_rename_missing_dir(self):
935
1026
        os.mkdir('a')
936
1027
        try:
937
1028
            osutils._win32_rename('b', 'a')
938
 
        except (IOError, OSError), e:
 
1029
        except (IOError, OSError) as e:
939
1030
            self.assertEqual(errno.ENOENT, e.errno)
940
1031
 
941
1032
    def test_rename_current_dir(self):
947
1038
        # doesn't exist.
948
1039
        try:
949
1040
            osutils._win32_rename('b', '.')
950
 
        except (IOError, OSError), e:
 
1041
        except (IOError, OSError) as e:
951
1042
            self.assertEqual(errno.ENOENT, e.errno)
952
1043
 
953
1044
    def test_splitpath(self):
958
1049
        check(['a', 'b'], 'a/b')
959
1050
        check(['a', 'b'], 'a/./b')
960
1051
        check(['a', '.b'], 'a/.b')
961
 
        check(['a', '.b'], 'a\\.b')
 
1052
        if os.path.sep == '\\':
 
1053
            check(['a', '.b'], 'a\\.b')
 
1054
        else:
 
1055
            check(['a\\.b'], 'a\\.b')
962
1056
 
963
1057
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
964
1058
 
976
1070
    """Test mac special functions that require directories."""
977
1071
 
978
1072
    def test_getcwd(self):
979
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1073
        self.requireFeature(features.UnicodeFilenameFeature)
980
1074
        os.mkdir(u'B\xe5gfors')
981
1075
        os.chdir(u'B\xe5gfors')
982
1076
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
983
1077
 
984
1078
    def test_getcwd_nonnorm(self):
985
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1079
        self.requireFeature(features.UnicodeFilenameFeature)
986
1080
        # Test that _mac_getcwd() will normalize this path
987
1081
        os.mkdir(u'Ba\u030agfors')
988
1082
        os.chdir(u'Ba\u030agfors')
992
1086
class TestChunksToLines(tests.TestCase):
993
1087
 
994
1088
    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']))
 
1089
        self.assertEqual([b'foo\n', b'bar\n', b'baz\n'],
 
1090
                         osutils.chunks_to_lines([b'foo\nbar', b'\nbaz\n']))
 
1091
        self.assertEqual([b'foo\n', b'bar\n', b'baz\n'],
 
1092
                         osutils.chunks_to_lines([b'foo\n', b'bar\n', b'baz\n']))
999
1093
 
1000
1094
    def test_osutils_binding(self):
1001
 
        from bzrlib.tests import test__chunks_to_lines
 
1095
        from . import test__chunks_to_lines
1002
1096
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
1003
 
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
 
1097
            from .._chunks_to_lines_pyx import chunks_to_lines
1004
1098
        else:
1005
 
            from bzrlib._chunks_to_lines_py import chunks_to_lines
 
1099
            from .._chunks_to_lines_py import chunks_to_lines
1006
1100
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1007
1101
 
1008
1102
 
1015
1109
                         osutils.split_lines(u'foo\nbar\xae\n'))
1016
1110
 
1017
1111
    def test_split_with_carriage_returns(self):
1018
 
        self.assertEqual(['foo\rbar\n'],
1019
 
                         osutils.split_lines('foo\rbar\n'))
 
1112
        self.assertEqual([b'foo\rbar\n'],
 
1113
                         osutils.split_lines(b'foo\rbar\n'))
1020
1114
 
1021
1115
 
1022
1116
class TestWalkDirs(tests.TestCaseInTempDir):
1037
1131
            ]
1038
1132
        self.build_tree(tree)
1039
1133
        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
 
                ),
 
1134
            (('', '.'),
 
1135
             [('0file', '0file', 'file'),
 
1136
              ('1dir', '1dir', 'directory'),
 
1137
              ('2file', '2file', 'file'),
 
1138
              ]
 
1139
             ),
 
1140
            (('1dir', './1dir'),
 
1141
             [('1dir/0file', '0file', 'file'),
 
1142
              ('1dir/1dir', '1dir', 'directory'),
 
1143
              ]
 
1144
             ),
 
1145
            (('1dir/1dir', './1dir/1dir'),
 
1146
             [
 
1147
                ]
 
1148
             ),
1055
1149
            ]
1056
1150
        result = []
1057
1151
        found_bzrdir = False
1071
1165
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
1072
1166
 
1073
1167
    def test_walkdirs_os_error(self):
1074
 
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
 
1168
        # <https://bugs.launchpad.net/bzr/+bug/338653>
1075
1169
        # Pyrex readdir didn't raise useful messages if it had an error
1076
1170
        # reading the directory
1077
1171
        if sys.platform == 'win32':
1078
1172
            raise tests.TestNotApplicable(
1079
1173
                "readdir IOError not tested on win32")
 
1174
        self.requireFeature(features.not_running_as_root)
1080
1175
        os.mkdir("test-unreadable")
1081
1176
        os.chmod("test-unreadable", 0000)
1082
1177
        # must chmod it back so that it can be removed
1083
 
        self.addCleanup(os.chmod, "test-unreadable", 0700)
 
1178
        self.addCleanup(os.chmod, "test-unreadable", 0o700)
1084
1179
        # The error is not raised until the generator is actually evaluated.
1085
1180
        # (It would be ok if it happened earlier but at the moment it
1086
1181
        # doesn't.)
1087
1182
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1088
 
        self.assertEquals('./test-unreadable', e.filename)
1089
 
        self.assertEquals(errno.EACCES, e.errno)
 
1183
        self.assertEqual('./test-unreadable', osutils.safe_unicode(e.filename))
 
1184
        self.assertEqual(errno.EACCES, e.errno)
1090
1185
        # Ensure the message contains the file name
1091
 
        self.assertContainsRe(str(e), "\./test-unreadable")
 
1186
        self.assertContainsRe(str(e), "\\./test-unreadable")
 
1187
 
 
1188
    def test_walkdirs_encoding_error(self):
 
1189
        # <https://bugs.launchpad.net/bzr/+bug/488519>
 
1190
        # walkdirs didn't raise a useful message when the filenames
 
1191
        # are not using the filesystem's encoding
 
1192
 
 
1193
        # require a bytestring based filesystem
 
1194
        self.requireFeature(features.ByteStringNamedFilesystem)
 
1195
 
 
1196
        tree = [
 
1197
            '.bzr',
 
1198
            '0file',
 
1199
            '1dir/',
 
1200
            '1dir/0file',
 
1201
            '1dir/1dir/',
 
1202
            '1file'
 
1203
            ]
 
1204
 
 
1205
        self.build_tree(tree)
 
1206
 
 
1207
        # rename the 1file to a latin-1 filename
 
1208
        os.rename(b"./1file", b"\xe8file")
 
1209
        if b"\xe8file" not in os.listdir("."):
 
1210
            self.skipTest("Lack filesystem that preserves arbitrary bytes")
 
1211
 
 
1212
        self._save_platform_info()
 
1213
        osutils._fs_enc = 'UTF-8'
 
1214
 
 
1215
        # this should raise on error
 
1216
        def attempt():
 
1217
            for dirdetail, dirblock in osutils.walkdirs(b'.'):
 
1218
                pass
 
1219
 
 
1220
        self.assertRaises(errors.BadFilenameEncoding, attempt)
1092
1221
 
1093
1222
    def test__walkdirs_utf8(self):
1094
1223
        tree = [
1101
1230
            ]
1102
1231
        self.build_tree(tree)
1103
1232
        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
 
                ),
 
1233
            (('', '.'),
 
1234
             [('0file', '0file', 'file'),
 
1235
              ('1dir', '1dir', 'directory'),
 
1236
              ('2file', '2file', 'file'),
 
1237
              ]
 
1238
             ),
 
1239
            (('1dir', './1dir'),
 
1240
             [('1dir/0file', '0file', 'file'),
 
1241
              ('1dir/1dir', '1dir', 'directory'),
 
1242
              ]
 
1243
             ),
 
1244
            (('1dir/1dir', './1dir/1dir'),
 
1245
             [
 
1246
                ]
 
1247
             ),
1119
1248
            ]
1120
1249
        result = []
1121
1250
        found_bzrdir = False
1122
 
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1123
 
            if len(dirblock) and dirblock[0][1] == '.bzr':
 
1251
        for dirdetail, dirblock in osutils._walkdirs_utf8(b'.'):
 
1252
            if len(dirblock) and dirblock[0][1] == b'.bzr':
1124
1253
                # this tests the filtering of selected paths
1125
1254
                found_bzrdir = True
1126
1255
                del dirblock[0]
 
1256
            dirdetail = (dirdetail[0].decode('utf-8'),
 
1257
                         osutils.safe_unicode(dirdetail[1]))
 
1258
            dirblock = [
 
1259
                (entry[0].decode('utf-8'), entry[1].decode('utf-8'), entry[2])
 
1260
                for entry in dirblock]
1127
1261
            result.append((dirdetail, dirblock))
1128
1262
 
1129
1263
        self.assertTrue(found_bzrdir)
1145
1279
            dirblock[:] = new_dirblock
1146
1280
 
1147
1281
    def _save_platform_info(self):
1148
 
        self.overrideAttr(win32utils, 'winver')
1149
1282
        self.overrideAttr(osutils, '_fs_enc')
1150
1283
        self.overrideAttr(osutils, '_selected_dir_reader')
1151
1284
 
1152
 
    def assertDirReaderIs(self, expected):
 
1285
    def assertDirReaderIs(self, expected, top):
1153
1286
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1154
1287
        # Force it to redetect
1155
1288
        osutils._selected_dir_reader = None
1156
1289
        # Nothing to list, but should still trigger the selection logic
1157
 
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
 
1290
        self.assertEqual([((b'', top), [])], list(osutils._walkdirs_utf8('.')))
1158
1291
        self.assertIsInstance(osutils._selected_dir_reader, expected)
1159
1292
 
1160
1293
    def test_force_walkdirs_utf8_fs_utf8(self):
1161
1294
        self.requireFeature(UTF8DirReaderFeature)
1162
1295
        self._save_platform_info()
1163
 
        win32utils.winver = None # Avoid the win32 detection code
1164
 
        osutils._fs_enc = 'UTF-8'
1165
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
1296
        osutils._fs_enc = 'utf-8'
 
1297
        self.assertDirReaderIs(UTF8DirReaderFeature.module.UTF8DirReader, b".")
1166
1298
 
1167
1299
    def test_force_walkdirs_utf8_fs_ascii(self):
1168
1300
        self.requireFeature(UTF8DirReaderFeature)
1169
1301
        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)
 
1302
        osutils._fs_enc = 'ascii'
 
1303
        self.assertDirReaderIs(
 
1304
            UTF8DirReaderFeature.module.UTF8DirReader, b".")
1180
1305
 
1181
1306
    def test_force_walkdirs_utf8_fs_latin1(self):
1182
1307
        self._save_platform_info()
1183
 
        win32utils.winver = None # Avoid the win32 detection code
1184
 
        osutils._fs_enc = 'latin1'
1185
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1308
        osutils._fs_enc = 'iso-8859-1'
 
1309
        self.assertDirReaderIs(osutils.UnicodeDirReader, ".")
1186
1310
 
1187
1311
    def test_force_walkdirs_utf8_nt(self):
1188
1312
        # Disabled because the thunk of the whole walkdirs api is disabled.
1189
1313
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1190
1314
        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)
 
1315
        from .._walkdirs_win32 import Win32ReadDir
 
1316
        self.assertDirReaderIs(Win32ReadDir, ".")
1200
1317
 
1201
1318
    def test_unicode_walkdirs(self):
1202
1319
        """Walkdirs should always return unicode paths."""
1203
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1320
        self.requireFeature(features.UnicodeFilenameFeature)
1204
1321
        name0 = u'0file-\xb6'
1205
1322
        name1 = u'1dir-\u062c\u0648'
1206
1323
        name2 = u'2file-\u0633'
1213
1330
            ]
1214
1331
        self.build_tree(tree)
1215
1332
        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
 
                ),
 
1333
            ((u'', u'.'),
 
1334
             [(name0, name0, 'file', './' + name0),
 
1335
              (name1, name1, 'directory', './' + name1),
 
1336
              (name2, name2, 'file', './' + name2),
 
1337
              ]
 
1338
             ),
 
1339
            ((name1, './' + name1),
 
1340
             [(name1 + '/' + name0, name0, 'file', './' + name1
 
1341
               + '/' + name0),
 
1342
              (name1 + '/' + name1, name1, 'directory', './' + name1
 
1343
               + '/' + name1),
 
1344
              ]
 
1345
             ),
 
1346
            ((name1 + '/' + name1, './' + name1 + '/' + name1),
 
1347
             [
 
1348
                ]
 
1349
             ),
1233
1350
            ]
1234
1351
        result = list(osutils.walkdirs('.'))
1235
1352
        self._filter_out_stat(result)
1236
1353
        self.assertEqual(expected_dirblocks, result)
1237
 
        result = list(osutils.walkdirs(u'./'+name1, name1))
 
1354
        result = list(osutils.walkdirs(u'./' + name1, name1))
1238
1355
        self._filter_out_stat(result)
1239
1356
        self.assertEqual(expected_dirblocks[1:], result)
1240
1357
 
1243
1360
 
1244
1361
        The abspath portion might be in unicode or utf-8
1245
1362
        """
1246
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1363
        self.requireFeature(features.UnicodeFilenameFeature)
1247
1364
        name0 = u'0file-\xb6'
1248
1365
        name1 = u'1dir-\u062c\u0648'
1249
1366
        name2 = u'2file-\u0633'
1260
1377
        name2 = name2.encode('utf8')
1261
1378
 
1262
1379
        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
 
                ),
 
1380
            ((b'', b'.'),
 
1381
             [(name0, name0, 'file', b'./' + name0),
 
1382
              (name1, name1, 'directory', b'./' + name1),
 
1383
              (name2, name2, 'file', b'./' + name2),
 
1384
              ]
 
1385
             ),
 
1386
            ((name1, b'./' + name1),
 
1387
             [(name1 + b'/' + name0, name0, 'file', b'./' + name1
 
1388
               + b'/' + name0),
 
1389
              (name1 + b'/' + name1, name1, 'directory', b'./' + name1
 
1390
               + b'/' + name1),
 
1391
              ]
 
1392
             ),
 
1393
            ((name1 + b'/' + name1, b'./' + name1 + b'/' + name1),
 
1394
             [
 
1395
                ]
 
1396
             ),
1280
1397
            ]
1281
1398
        result = []
1282
1399
        # For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1283
1400
        # all abspaths are Unicode, and encode them back into utf8.
1284
1401
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1285
 
            self.assertIsInstance(dirdetail[0], str)
1286
 
            if isinstance(dirdetail[1], unicode):
 
1402
            self.assertIsInstance(dirdetail[0], bytes)
 
1403
            if isinstance(dirdetail[1], str):
1287
1404
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1288
1405
                dirblock = [list(info) for info in dirblock]
1289
1406
                for info in dirblock:
1290
 
                    self.assertIsInstance(info[4], unicode)
 
1407
                    self.assertIsInstance(info[4], str)
1291
1408
                    info[4] = info[4].encode('utf8')
1292
1409
            new_dirblock = []
1293
1410
            for info in dirblock:
1294
 
                self.assertIsInstance(info[0], str)
1295
 
                self.assertIsInstance(info[1], str)
1296
 
                self.assertIsInstance(info[4], str)
 
1411
                self.assertIsInstance(info[0], bytes)
 
1412
                self.assertIsInstance(info[1], bytes)
 
1413
                self.assertIsInstance(info[4], bytes)
1297
1414
                # Remove the stat information
1298
1415
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1299
1416
            result.append((dirdetail, new_dirblock))
1304
1421
 
1305
1422
        The abspath portion should be in unicode
1306
1423
        """
1307
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1424
        self.requireFeature(features.UnicodeFilenameFeature)
1308
1425
        # Use the unicode reader. TODO: split into driver-and-driven unit
1309
1426
        # tests.
1310
1427
        self._save_platform_info()
1327
1444
        # All of the abspaths should be in unicode, all of the relative paths
1328
1445
        # should be in utf8
1329
1446
        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
 
                ),
 
1447
            ((b'', '.'),
 
1448
             [(name0, name0, 'file', './' + name0u),
 
1449
              (name1, name1, 'directory', './' + name1u),
 
1450
              (name2, name2, 'file', './' + name2u),
 
1451
              ]
 
1452
             ),
 
1453
            ((name1, './' + name1u),
 
1454
             [(name1 + b'/' + name0, name0, 'file', './' + name1u
 
1455
               + '/' + name0u),
 
1456
              (name1 + b'/' + name1, name1, 'directory', './' + name1u
 
1457
               + '/' + name1u),
 
1458
              ]
 
1459
             ),
 
1460
            ((name1 + b'/' + name1, './' + name1u + '/' + name1u),
 
1461
             [
 
1462
                ]
 
1463
             ),
1347
1464
            ]
1348
1465
        result = list(osutils._walkdirs_utf8('.'))
1349
1466
        self._filter_out_stat(result)
1351
1468
 
1352
1469
    def test__walkdirs_utf8_win32readdir(self):
1353
1470
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1354
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1355
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1471
        self.requireFeature(features.UnicodeFilenameFeature)
 
1472
        from .._walkdirs_win32 import Win32ReadDir
1356
1473
        self._save_platform_info()
1357
1474
        osutils._selected_dir_reader = Win32ReadDir()
1358
1475
        name0u = u'0file-\xb6'
1373
1490
        # All of the abspaths should be in unicode, all of the relative paths
1374
1491
        # should be in utf8
1375
1492
        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
 
                ),
 
1493
            (('', '.'),
 
1494
             [(name0, name0, 'file', './' + name0u),
 
1495
              (name1, name1, 'directory', './' + name1u),
 
1496
              (name2, name2, 'file', './' + name2u),
 
1497
              ]
 
1498
             ),
 
1499
            ((name1, './' + name1u),
 
1500
             [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1501
               + '/' + name0u),
 
1502
              (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1503
               + '/' + name1u),
 
1504
              ]
 
1505
             ),
 
1506
            ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1507
             [
 
1508
                ]
 
1509
             ),
1393
1510
            ]
1394
1511
        result = list(osutils._walkdirs_utf8(u'.'))
1395
1512
        self._filter_out_stat(result)
1408
1525
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1409
1526
        """make sure our Stat values are valid"""
1410
1527
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1411
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1412
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1528
        self.requireFeature(features.UnicodeFilenameFeature)
 
1529
        from .._walkdirs_win32 import Win32ReadDir
1413
1530
        name0u = u'0file-\xb6'
1414
1531
        name0 = name0u.encode('utf8')
1415
1532
        self.build_tree([name0u])
1416
1533
        # I hate to sleep() here, but I'm trying to make the ctime different
1417
1534
        # from the mtime
1418
1535
        time.sleep(2)
1419
 
        f = open(name0u, 'ab')
1420
 
        try:
1421
 
            f.write('just a small update')
1422
 
        finally:
1423
 
            f.close()
 
1536
        with open(name0u, 'ab') as f:
 
1537
            f.write(b'just a small update')
1424
1538
 
1425
1539
        result = Win32ReadDir().read_dir('', u'.')
1426
1540
        entry = result[0]
1432
1546
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1433
1547
        """make sure our Stat values are valid"""
1434
1548
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1435
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1436
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1549
        self.requireFeature(features.UnicodeFilenameFeature)
 
1550
        from .._walkdirs_win32 import Win32ReadDir
1437
1551
        name0u = u'0dir-\u062c\u0648'
1438
1552
        name0 = name0u.encode('utf8')
1439
1553
        self.build_tree([name0u + '/'])
1519
1633
        # using the comparison routine shoudl work too:
1520
1634
        self.assertEqual(
1521
1635
            dir_sorted_paths,
1522
 
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
 
1636
            sorted(original_paths, key=osutils.path_prefix_key))
1523
1637
 
1524
1638
 
1525
1639
class TestCopyTree(tests.TestCaseInTempDir):
1538
1652
        self.assertEqual(['c'], os.listdir('target/b'))
1539
1653
 
1540
1654
    def test_copy_tree_symlinks(self):
1541
 
        self.requireFeature(tests.SymlinkFeature)
 
1655
        self.requireFeature(features.SymlinkFeature)
1542
1656
        self.build_tree(['source/'])
1543
1657
        os.symlink('a/generic/path', 'source/lnk')
1544
1658
        osutils.copy_tree('source', 'target')
1548
1662
    def test_copy_tree_handlers(self):
1549
1663
        processed_files = []
1550
1664
        processed_links = []
 
1665
 
1551
1666
        def file_handler(from_path, to_path):
1552
1667
            processed_files.append(('f', from_path, to_path))
 
1668
 
1553
1669
        def dir_handler(from_path, to_path):
1554
1670
            processed_files.append(('d', from_path, to_path))
 
1671
 
1555
1672
        def link_handler(from_path, to_path):
1556
1673
            processed_links.append((from_path, to_path))
1557
 
        handlers = {'file':file_handler,
1558
 
                    'directory':dir_handler,
1559
 
                    'symlink':link_handler,
1560
 
                   }
 
1674
        handlers = {'file': file_handler,
 
1675
                    'directory': dir_handler,
 
1676
                    'symlink': link_handler,
 
1677
                    }
1561
1678
 
1562
1679
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1563
1680
        if osutils.has_symlinks():
1568
1685
                          ('f', 'source/a', 'target/a'),
1569
1686
                          ('d', 'source/b', 'target/b'),
1570
1687
                          ('f', 'source/b/c', 'target/b/c'),
1571
 
                         ], processed_files)
1572
 
        self.failIfExists('target')
 
1688
                          ], processed_files)
 
1689
        self.assertPathDoesNotExist('target')
1573
1690
        if osutils.has_symlinks():
1574
1691
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1575
1692
 
1580
1697
    def setUp(self):
1581
1698
        super(TestSetUnsetEnv, self).setUp()
1582
1699
 
1583
 
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
 
1700
        self.assertEqual(None, os.environ.get('BRZ_TEST_ENV_VAR'),
1584
1701
                         'Environment was not cleaned up properly.'
1585
 
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
 
1702
                         ' Variable BRZ_TEST_ENV_VAR should not exist.')
 
1703
 
1586
1704
        def cleanup():
1587
 
            if 'BZR_TEST_ENV_VAR' in os.environ:
1588
 
                del os.environ['BZR_TEST_ENV_VAR']
 
1705
            if 'BRZ_TEST_ENV_VAR' in os.environ:
 
1706
                del os.environ['BRZ_TEST_ENV_VAR']
1589
1707
        self.addCleanup(cleanup)
1590
1708
 
1591
1709
    def test_set(self):
1592
1710
        """Test that we can set an env variable"""
1593
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1711
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
1594
1712
        self.assertEqual(None, old)
1595
 
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
 
1713
        self.assertEqual('foo', os.environ.get('BRZ_TEST_ENV_VAR'))
1596
1714
 
1597
1715
    def test_double_set(self):
1598
1716
        """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')
 
1717
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
 
1718
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'bar')
1601
1719
        self.assertEqual('foo', old)
1602
 
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
 
1720
        self.assertEqual('bar', os.environ.get('BRZ_TEST_ENV_VAR'))
1603
1721
 
1604
1722
    def test_unicode(self):
1605
1723
        """Environment can only contain plain strings
1612
1730
                'Cannot find a unicode character that works in encoding %s'
1613
1731
                % (osutils.get_user_encoding(),))
1614
1732
 
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'))
 
1733
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', uni_val)
 
1734
        self.assertEqual(uni_val, os.environ.get('BRZ_TEST_ENV_VAR'))
1617
1735
 
1618
1736
    def test_unset(self):
1619
1737
        """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)
 
1738
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
 
1739
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', None)
1622
1740
        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)
 
1741
        self.assertEqual(None, os.environ.get('BRZ_TEST_ENV_VAR'))
 
1742
        self.assertNotIn('BRZ_TEST_ENV_VAR', os.environ)
1625
1743
 
1626
1744
 
1627
1745
class TestSizeShaFile(tests.TestCaseInTempDir):
1628
1746
 
1629
1747
    def test_sha_empty(self):
1630
 
        self.build_tree_contents([('foo', '')])
1631
 
        expected_sha = osutils.sha_string('')
 
1748
        self.build_tree_contents([('foo', b'')])
 
1749
        expected_sha = osutils.sha_string(b'')
1632
1750
        f = open('foo')
1633
1751
        self.addCleanup(f.close)
1634
1752
        size, sha = osutils.size_sha_file(f)
1636
1754
        self.assertEqual(expected_sha, sha)
1637
1755
 
1638
1756
    def test_sha_mixed_endings(self):
1639
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1757
        text = b'test\r\nwith\nall\rpossible line endings\r\n'
1640
1758
        self.build_tree_contents([('foo', text)])
1641
1759
        expected_sha = osutils.sha_string(text)
1642
1760
        f = open('foo', 'rb')
1649
1767
class TestShaFileByName(tests.TestCaseInTempDir):
1650
1768
 
1651
1769
    def test_sha_empty(self):
1652
 
        self.build_tree_contents([('foo', '')])
1653
 
        expected_sha = osutils.sha_string('')
 
1770
        self.build_tree_contents([('foo', b'')])
 
1771
        expected_sha = osutils.sha_string(b'')
1654
1772
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1655
1773
 
1656
1774
    def test_sha_mixed_endings(self):
1657
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1775
        text = b'test\r\nwith\nall\rpossible line endings\r\n'
1658
1776
        self.build_tree_contents([('foo', text)])
1659
1777
        expected_sha = osutils.sha_string(text)
1660
1778
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1663
1781
class TestResourceLoading(tests.TestCaseInTempDir):
1664
1782
 
1665
1783
    def test_resource_string(self):
1666
 
        # test resource in bzrlib
1667
 
        text = osutils.resource_string('bzrlib', 'debug.py')
 
1784
        # test resource in breezy
 
1785
        text = osutils.resource_string('breezy', 'debug.py')
1668
1786
        self.assertContainsRe(text, "debug_flags = set()")
1669
 
        # test resource under bzrlib
1670
 
        text = osutils.resource_string('bzrlib.ui', 'text.py')
 
1787
        # test resource under breezy
 
1788
        text = osutils.resource_string('breezy.ui', 'text.py')
1671
1789
        self.assertContainsRe(text, "class TextUIFactory")
1672
1790
        # test unsupported package
1673
1791
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1674
 
            'yyy.xx')
 
1792
                          'yyy.xx')
1675
1793
        # 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))
 
1794
        self.assertRaises(IOError, osutils.resource_string, 'breezy', 'yyy.xx')
1695
1795
 
1696
1796
 
1697
1797
class TestDirReader(tests.TestCaseInTempDir):
1698
1798
 
 
1799
    scenarios = dir_reader_scenarios()
 
1800
 
1699
1801
    # Set by load_tests
1700
1802
    _dir_reader_class = None
1701
1803
    _native_to_unicode = None
1702
1804
 
1703
1805
    def setUp(self):
1704
 
        tests.TestCaseInTempDir.setUp(self)
 
1806
        super(TestDirReader, self).setUp()
1705
1807
        self.overrideAttr(osutils,
1706
1808
                          '_selected_dir_reader', self._dir_reader_class())
1707
1809
 
1714
1816
            '2file'
1715
1817
            ]
1716
1818
        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
 
                ),
 
1819
            ((b'', '.'),
 
1820
             [(b'0file', b'0file', 'file', './0file'),
 
1821
              (b'1dir', b'1dir', 'directory', './1dir'),
 
1822
              (b'2file', b'2file', 'file', './2file'),
 
1823
              ]
 
1824
             ),
 
1825
            ((b'1dir', './1dir'),
 
1826
             [(b'1dir/0file', b'0file', 'file', './1dir/0file'),
 
1827
              (b'1dir/1dir', b'1dir', 'directory', './1dir/1dir'),
 
1828
              ]
 
1829
             ),
 
1830
            ((b'1dir/1dir', './1dir/1dir'),
 
1831
             [
 
1832
                ]
 
1833
             ),
1732
1834
            ]
1733
1835
        return tree, expected_dirblocks
1734
1836
 
1738
1840
        result = list(osutils._walkdirs_utf8('.'))
1739
1841
        # Filter out stat and abspath
1740
1842
        self.assertEqual(expected_dirblocks,
1741
 
                         [(dirinfo, [line[0:3] for line in block])
1742
 
                          for dirinfo, block in result])
 
1843
                         self._filter_out(result))
1743
1844
 
1744
1845
    def test_walk_sub_dir(self):
1745
1846
        tree, expected_dirblocks = self._get_ascii_tree()
1746
1847
        self.build_tree(tree)
1747
1848
        # you can search a subdir only, with a supplied prefix.
1748
 
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
 
1849
        result = list(osutils._walkdirs_utf8(b'./1dir', b'1dir'))
1749
1850
        # Filter out stat and abspath
1750
1851
        self.assertEqual(expected_dirblocks[1:],
1751
 
                         [(dirinfo, [line[0:3] for line in block])
1752
 
                          for dirinfo, block in result])
 
1852
                         self._filter_out(result))
1753
1853
 
1754
1854
    def _get_unicode_tree(self):
1755
1855
        name0u = u'0file-\xb6'
1766
1866
        name1 = name1u.encode('UTF-8')
1767
1867
        name2 = name2u.encode('UTF-8')
1768
1868
        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
 
                ),
 
1869
            ((b'', '.'),
 
1870
             [(name0, name0, 'file', './' + name0u),
 
1871
              (name1, name1, 'directory', './' + name1u),
 
1872
              (name2, name2, 'file', './' + name2u),
 
1873
              ]
 
1874
             ),
 
1875
            ((name1, './' + name1u),
 
1876
             [(name1 + b'/' + name0, name0, 'file', './' + name1u
 
1877
               + '/' + name0u),
 
1878
              (name1 + b'/' + name1, name1, 'directory', './' + name1u
 
1879
               + '/' + name1u),
 
1880
              ]
 
1881
             ),
 
1882
            ((name1 + b'/' + name1, './' + name1u + '/' + name1u),
 
1883
             [
 
1884
                ]
 
1885
             ),
1786
1886
            ]
1787
1887
        return tree, expected_dirblocks
1788
1888
 
1796
1896
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1797
1897
            details = []
1798
1898
            for line in block:
1799
 
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
 
1899
                details.append(
 
1900
                    line[0:3] + (self._native_to_unicode(line[4]), ))
1800
1901
            filtered_dirblocks.append((dirinfo, details))
1801
1902
        return filtered_dirblocks
1802
1903
 
1803
1904
    def test_walk_unicode_tree(self):
1804
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1905
        self.requireFeature(features.UnicodeFilenameFeature)
1805
1906
        tree, expected_dirblocks = self._get_unicode_tree()
1806
1907
        self.build_tree(tree)
1807
1908
        result = list(osutils._walkdirs_utf8('.'))
1808
1909
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1809
1910
 
1810
1911
    def test_symlink(self):
1811
 
        self.requireFeature(tests.SymlinkFeature)
1812
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1912
        self.requireFeature(features.SymlinkFeature)
 
1913
        self.requireFeature(features.UnicodeFilenameFeature)
1813
1914
        target = u'target\N{Euro Sign}'
1814
1915
        link_name = u'l\N{Euro Sign}nk'
1815
1916
        os.symlink(target, link_name)
1816
 
        target_utf8 = target.encode('UTF-8')
1817
1917
        link_name_utf8 = link_name.encode('UTF-8')
1818
1918
        expected_dirblocks = [
1819
 
                (('', '.'),
1820
 
                 [(link_name_utf8, link_name_utf8,
1821
 
                   'symlink', './' + link_name),],
1822
 
                 )]
 
1919
            ((b'', '.'),
 
1920
             [(link_name_utf8, link_name_utf8,
 
1921
               'symlink', './' + link_name), ],
 
1922
             )]
1823
1923
        result = list(osutils._walkdirs_utf8('.'))
1824
1924
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1825
1925
 
1833
1933
    But prior python versions failed to properly encode the passed unicode
1834
1934
    string.
1835
1935
    """
1836
 
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
 
1936
    _test_needs_features = [features.SymlinkFeature,
 
1937
                            features.UnicodeFilenameFeature]
1837
1938
 
1838
1939
    def setUp(self):
1839
1940
        super(tests.TestCaseInTempDir, self).setUp()
1842
1943
        os.symlink(self.target, self.link)
1843
1944
 
1844
1945
    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))
 
1946
        self.assertEqual(self.target, os.readlink(self.link))
1849
1947
 
1850
1948
    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)))
 
1949
        self.assertEqual(self.target.encode(osutils._fs_enc),
 
1950
                         os.readlink(self.link.encode(osutils._fs_enc)))
1853
1951
 
1854
1952
 
1855
1953
class TestConcurrency(tests.TestCase):
1863
1961
        self.assertIsInstance(concurrency, int)
1864
1962
 
1865
1963
    def test_local_concurrency_environment_variable(self):
1866
 
        os.environ['BZR_CONCURRENCY'] = '2'
 
1964
        self.overrideEnv('BRZ_CONCURRENCY', '2')
1867
1965
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1868
 
        os.environ['BZR_CONCURRENCY'] = '3'
 
1966
        self.overrideEnv('BRZ_CONCURRENCY', '3')
1869
1967
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1870
 
        os.environ['BZR_CONCURRENCY'] = 'foo'
 
1968
        self.overrideEnv('BRZ_CONCURRENCY', 'foo')
1871
1969
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1872
1970
 
1873
1971
    def test_option_concurrency(self):
1874
 
        os.environ['BZR_CONCURRENCY'] = '1'
 
1972
        self.overrideEnv('BRZ_CONCURRENCY', '1')
1875
1973
        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))
 
1974
        # Command line overrides environment variable
 
1975
        self.assertEqual('42', os.environ['BRZ_CONCURRENCY'])
 
1976
        self.assertEqual(42, osutils.local_concurrency(use_cache=False))
1879
1977
 
1880
1978
 
1881
1979
class TestFailedToLoadExtension(tests.TestCase):
1882
1980
 
1883
1981
    def _try_loading(self):
1884
1982
        try:
1885
 
            import bzrlib._fictional_extension_py
1886
 
        except ImportError, e:
 
1983
            import breezy._fictional_extension_py  # noqa: F401
 
1984
        except ImportError as e:
1887
1985
            osutils.failed_to_load_extension(e)
1888
1986
            return True
1889
1987
 
1894
1992
    def test_failure_to_load(self):
1895
1993
        self._try_loading()
1896
1994
        self.assertLength(1, osutils._extension_load_failures)
1897
 
        self.assertEquals(osutils._extension_load_failures[0],
1898
 
            "No module named _fictional_extension_py")
 
1995
        self.assertEqual(
 
1996
            osutils._extension_load_failures[0],
 
1997
            "No module named 'breezy._fictional_extension_py'")
1899
1998
 
1900
1999
    def test_report_extension_load_failures_no_warning(self):
1901
2000
        self.assertTrue(self._try_loading())
1902
 
        warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
 
2001
        warnings, result = self.callCatchWarnings(
 
2002
            osutils.report_extension_load_failures)
1903
2003
        # it used to give a Python warning; it no longer does
1904
2004
        self.assertLength(0, warnings)
1905
2005
 
1906
2006
    def test_report_extension_load_failures_message(self):
1907
 
        log = StringIO()
 
2007
        log = BytesIO()
1908
2008
        trace.push_log_file(log)
1909
2009
        self.assertTrue(self._try_loading())
1910
2010
        osutils.report_extension_load_failures()
1911
2011
        self.assertContainsRe(
1912
2012
            log.getvalue(),
1913
 
            r"bzr: warning: some compiled extensions could not be loaded; "
1914
 
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
 
2013
            br"brz: warning: some compiled extensions could not be loaded; "
 
2014
            b"see ``brz help missing-extensions``\n"
1915
2015
            )
1916
2016
 
1917
2017
 
1918
2018
class TestTerminalWidth(tests.TestCase):
1919
2019
 
 
2020
    def setUp(self):
 
2021
        super(TestTerminalWidth, self).setUp()
 
2022
        self._orig_terminal_size_state = osutils._terminal_size_state
 
2023
        self._orig_first_terminal_size = osutils._first_terminal_size
 
2024
        self.addCleanup(self.restore_osutils_globals)
 
2025
        osutils._terminal_size_state = 'no_data'
 
2026
        osutils._first_terminal_size = None
 
2027
 
 
2028
    def restore_osutils_globals(self):
 
2029
        osutils._terminal_size_state = self._orig_terminal_size_state
 
2030
        osutils._first_terminal_size = self._orig_first_terminal_size
 
2031
 
1920
2032
    def replace_stdout(self, new):
1921
2033
        self.overrideAttr(sys, 'stdout', new)
1922
2034
 
1934
2046
    def test_default_values(self):
1935
2047
        self.assertEqual(80, osutils.default_terminal_width)
1936
2048
 
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'
 
2049
    def test_defaults_to_BRZ_COLUMNS(self):
 
2050
        # BRZ_COLUMNS is set by the test framework
 
2051
        self.assertNotEqual('12', os.environ['BRZ_COLUMNS'])
 
2052
        self.overrideEnv('BRZ_COLUMNS', '12')
1941
2053
        self.assertEqual(12, osutils.terminal_width())
1942
2054
 
 
2055
    def test_BRZ_COLUMNS_0_no_limit(self):
 
2056
        self.overrideEnv('BRZ_COLUMNS', '0')
 
2057
        self.assertEqual(None, osutils.terminal_width())
 
2058
 
1943
2059
    def test_falls_back_to_COLUMNS(self):
1944
 
        del os.environ['BZR_COLUMNS']
 
2060
        self.overrideEnv('BRZ_COLUMNS', None)
1945
2061
        self.assertNotEqual('42', os.environ['COLUMNS'])
1946
2062
        self.set_fake_tty()
1947
 
        os.environ['COLUMNS'] = '42'
 
2063
        self.overrideEnv('COLUMNS', '42')
1948
2064
        self.assertEqual(42, osutils.terminal_width())
1949
2065
 
1950
2066
    def test_tty_default_without_columns(self):
1951
 
        del os.environ['BZR_COLUMNS']
1952
 
        del os.environ['COLUMNS']
 
2067
        self.overrideEnv('BRZ_COLUMNS', None)
 
2068
        self.overrideEnv('COLUMNS', None)
1953
2069
 
1954
2070
        def terminal_size(w, h):
1955
2071
            return 42, 42
1962
2078
        self.assertEqual(42, osutils.terminal_width())
1963
2079
 
1964
2080
    def test_non_tty_default_without_columns(self):
1965
 
        del os.environ['BZR_COLUMNS']
1966
 
        del os.environ['COLUMNS']
 
2081
        self.overrideEnv('BRZ_COLUMNS', None)
 
2082
        self.overrideEnv('COLUMNS', None)
1967
2083
        self.replace_stdout(None)
1968
2084
        self.assertEqual(None, osutils.terminal_width())
1969
2085
 
1972
2088
        termios = term_ios_feature.module
1973
2089
        # bug 63539 is about a termios without TIOCGWINSZ attribute
1974
2090
        try:
1975
 
            orig = termios.TIOCGWINSZ
 
2091
            termios.TIOCGWINSZ
1976
2092
        except AttributeError:
1977
2093
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
1978
2094
            pass
1979
2095
        else:
1980
2096
            self.overrideAttr(termios, 'TIOCGWINSZ')
1981
2097
            del termios.TIOCGWINSZ
1982
 
        del os.environ['BZR_COLUMNS']
1983
 
        del os.environ['COLUMNS']
 
2098
        self.overrideEnv('BRZ_COLUMNS', None)
 
2099
        self.overrideEnv('COLUMNS', None)
1984
2100
        # Whatever the result is, if we don't raise an exception, it's ok.
1985
2101
        osutils.terminal_width()
1986
2102
 
 
2103
 
1987
2104
class TestCreationOps(tests.TestCaseInTempDir):
1988
2105
    _test_needs_features = [features.chown_feature]
1989
2106
 
1990
2107
    def setUp(self):
1991
 
        tests.TestCaseInTempDir.setUp(self)
 
2108
        super(TestCreationOps, self).setUp()
1992
2109
        self.overrideAttr(os, 'chown', self._dummy_chown)
1993
2110
 
1994
2111
        # params set by call to _dummy_chown
2000
2117
    def test_copy_ownership_from_path(self):
2001
2118
        """copy_ownership_from_path test with specified src."""
2002
2119
        ownsrc = '/'
2003
 
        f = open('test_file', 'wt')
 
2120
        open('test_file', 'wt').close()
2004
2121
        osutils.copy_ownership_from_path('test_file', ownsrc)
2005
2122
 
2006
2123
        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)
 
2124
        self.assertEqual(self.path, 'test_file')
 
2125
        self.assertEqual(self.uid, s.st_uid)
 
2126
        self.assertEqual(self.gid, s.st_gid)
2010
2127
 
2011
2128
    def test_copy_ownership_nonesrc(self):
2012
2129
        """copy_ownership_from_path test with src=None."""
2013
 
        f = open('test_file', 'wt')
 
2130
        open('test_file', 'wt').close()
2014
2131
        # should use parent dir for permissions
2015
2132
        osutils.copy_ownership_from_path('test_file')
2016
2133
 
2017
2134
        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)
 
2135
        self.assertEqual(self.path, 'test_file')
 
2136
        self.assertEqual(self.uid, s.st_uid)
 
2137
        self.assertEqual(self.gid, s.st_gid)
 
2138
 
 
2139
 
 
2140
class TestPathFromEnviron(tests.TestCase):
 
2141
 
 
2142
    def test_is_unicode(self):
 
2143
        self.overrideEnv('BRZ_TEST_PATH', './anywhere at all/')
 
2144
        path = osutils.path_from_environ('BRZ_TEST_PATH')
 
2145
        self.assertIsInstance(path, str)
 
2146
        self.assertEqual(u'./anywhere at all/', path)
 
2147
 
 
2148
    def test_posix_path_env_ascii(self):
 
2149
        self.overrideEnv('BRZ_TEST_PATH', '/tmp')
 
2150
        home = osutils._posix_path_from_environ('BRZ_TEST_PATH')
 
2151
        self.assertIsInstance(home, str)
 
2152
        self.assertEqual(u'/tmp', home)
 
2153
 
 
2154
    def test_posix_path_env_unicode(self):
 
2155
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2156
        self.overrideEnv('BRZ_TEST_PATH', '/home/\xa7test')
 
2157
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2158
        self.assertEqual(u'/home/\xa7test',
 
2159
                         osutils._posix_path_from_environ('BRZ_TEST_PATH'))
 
2160
        osutils._fs_enc = "iso8859-5"
 
2161
        # In Python 3, os.environ returns unicode.
 
2162
        self.assertEqual(u'/home/\xa7test',
 
2163
                         osutils._posix_path_from_environ('BRZ_TEST_PATH'))
 
2164
 
 
2165
 
 
2166
class TestGetHomeDir(tests.TestCase):
 
2167
 
 
2168
    def test_is_unicode(self):
 
2169
        home = osutils._get_home_dir()
 
2170
        self.assertIsInstance(home, str)
 
2171
 
 
2172
    def test_posix_homeless(self):
 
2173
        self.overrideEnv('HOME', None)
 
2174
        home = osutils._get_home_dir()
 
2175
        self.assertIsInstance(home, str)
 
2176
 
 
2177
    def test_posix_home_ascii(self):
 
2178
        self.overrideEnv('HOME', '/home/test')
 
2179
        home = osutils._posix_get_home_dir()
 
2180
        self.assertIsInstance(home, str)
 
2181
        self.assertEqual(u'/home/test', home)
 
2182
 
 
2183
    def test_posix_home_unicode(self):
 
2184
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2185
        self.overrideEnv('HOME', '/home/\xa7test')
 
2186
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2187
        self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
 
2188
        osutils._fs_enc = "iso8859-5"
 
2189
        # In python 3, os.environ returns unicode
 
2190
        self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
 
2191
 
2021
2192
 
2022
2193
class TestGetuserUnicode(tests.TestCase):
2023
2194
 
 
2195
    def test_is_unicode(self):
 
2196
        user = osutils.getuser_unicode()
 
2197
        self.assertIsInstance(user, str)
 
2198
 
 
2199
    def envvar_to_override(self):
 
2200
        if sys.platform == "win32":
 
2201
            # Disable use of platform calls on windows so envvar is used
 
2202
            self.overrideAttr(win32utils, 'has_ctypes', False)
 
2203
            return 'USERNAME'  # only variable used on windows
 
2204
        return 'LOGNAME'  # first variable checked by getpass.getuser()
 
2205
 
2024
2206
    def test_ascii_user(self):
2025
 
        osutils.set_or_unset_env('LOGNAME', 'jrandom')
 
2207
        self.overrideEnv(self.envvar_to_override(), 'jrandom')
2026
2208
        self.assertEqual(u'jrandom', osutils.getuser_unicode())
2027
2209
 
2028
2210
    def test_unicode_user(self):
2029
2211
        ue = osutils.get_user_encoding()
2030
 
        osutils.set_or_unset_env('LOGNAME', u'jrandom\xb6'.encode(ue))
2031
 
        self.assertEqual(u'jrandom\xb6', osutils.getuser_unicode())
 
2212
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
2213
        if uni_val is None:
 
2214
            raise tests.TestSkipped(
 
2215
                'Cannot find a unicode character that works in encoding %s'
 
2216
                % (osutils.get_user_encoding(),))
 
2217
        uni_username = u'jrandom' + uni_val
 
2218
        encoded_username = uni_username.encode(ue)
 
2219
        self.overrideEnv(self.envvar_to_override(), uni_username)
 
2220
        self.assertEqual(uni_username, osutils.getuser_unicode())
 
2221
 
 
2222
 
 
2223
class TestBackupNames(tests.TestCase):
 
2224
 
 
2225
    def setUp(self):
 
2226
        super(TestBackupNames, self).setUp()
 
2227
        self.backups = []
 
2228
 
 
2229
    def backup_exists(self, name):
 
2230
        return name in self.backups
 
2231
 
 
2232
    def available_backup_name(self, name):
 
2233
        backup_name = osutils.available_backup_name(name, self.backup_exists)
 
2234
        self.backups.append(backup_name)
 
2235
        return backup_name
 
2236
 
 
2237
    def assertBackupName(self, expected, name):
 
2238
        self.assertEqual(expected, self.available_backup_name(name))
 
2239
 
 
2240
    def test_empty(self):
 
2241
        self.assertBackupName('file.~1~', 'file')
 
2242
 
 
2243
    def test_existing(self):
 
2244
        self.available_backup_name('file')
 
2245
        self.available_backup_name('file')
 
2246
        self.assertBackupName('file.~3~', 'file')
 
2247
        # Empty slots are found, this is not a strict requirement and may be
 
2248
        # revisited if we test against all implementations.
 
2249
        self.backups.remove('file.~2~')
 
2250
        self.assertBackupName('file.~2~', 'file')
 
2251
 
 
2252
 
 
2253
class TestFindExecutableInPath(tests.TestCase):
 
2254
 
 
2255
    def test_windows(self):
 
2256
        if sys.platform != 'win32':
 
2257
            raise tests.TestSkipped('test requires win32')
 
2258
        self.assertTrue(osutils.find_executable_on_path(
 
2259
            'explorer') is not None)
 
2260
        self.assertTrue(
 
2261
            osutils.find_executable_on_path('explorer.exe') is not None)
 
2262
        self.assertTrue(
 
2263
            osutils.find_executable_on_path('EXPLORER.EXE') is not None)
 
2264
        self.assertTrue(
 
2265
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2266
        self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
 
2267
 
 
2268
    def test_windows_app_path(self):
 
2269
        if sys.platform != 'win32':
 
2270
            raise tests.TestSkipped('test requires win32')
 
2271
        # Override PATH env var so that exe can only be found on App Path
 
2272
        self.overrideEnv('PATH', '')
 
2273
        # Internt Explorer is always registered in the App Path
 
2274
        self.assertTrue(osutils.find_executable_on_path(
 
2275
            'iexplore') is not None)
 
2276
 
 
2277
    def test_other(self):
 
2278
        if sys.platform == 'win32':
 
2279
            raise tests.TestSkipped('test requires non-win32')
 
2280
        self.assertTrue(osutils.find_executable_on_path('sh') is not None)
 
2281
        self.assertTrue(
 
2282
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2283
 
 
2284
 
 
2285
class TestEnvironmentErrors(tests.TestCase):
 
2286
    """Test handling of environmental errors"""
 
2287
 
 
2288
    def test_is_oserror(self):
 
2289
        self.assertTrue(osutils.is_environment_error(
 
2290
            OSError(errno.EINVAL, "Invalid parameter")))
 
2291
 
 
2292
    def test_is_ioerror(self):
 
2293
        self.assertTrue(osutils.is_environment_error(
 
2294
            IOError(errno.EINVAL, "Invalid parameter")))
 
2295
 
 
2296
    def test_is_socket_error(self):
 
2297
        self.assertTrue(osutils.is_environment_error(
 
2298
            socket.error(errno.EINVAL, "Invalid parameter")))
 
2299
 
 
2300
    def test_is_select_error(self):
 
2301
        self.assertTrue(osutils.is_environment_error(
 
2302
            select.error(errno.EINVAL, "Invalid parameter")))
 
2303
 
 
2304
    def test_is_pywintypes_error(self):
 
2305
        self.requireFeature(features.pywintypes)
 
2306
        import pywintypes
 
2307
        self.assertTrue(osutils.is_environment_error(
 
2308
            pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))
 
2309
 
 
2310
 
 
2311
class SupportsExecutableTests(tests.TestCaseInTempDir):
 
2312
 
 
2313
    def test_returns_bool(self):
 
2314
        self.assertIsInstance(osutils.supports_executable(self.test_dir), bool)
 
2315
 
 
2316
 
 
2317
class SupportsSymlinksTests(tests.TestCaseInTempDir):
 
2318
 
 
2319
    def test_returns_bool(self):
 
2320
        self.assertIsInstance(osutils.supports_symlinks(self.test_dir), bool)
 
2321
 
 
2322
 
 
2323
class MtabReader(tests.TestCaseInTempDir):
 
2324
 
 
2325
    def test_read_mtab(self):
 
2326
        self.build_tree_contents([('mtab', """\
 
2327
/dev/mapper/blah--vg-root / ext4 rw,relatime,errors=remount-ro 0 0
 
2328
/dev/mapper/blah--vg-home /home vfat rw,relatime 0 0
 
2329
# comment
 
2330
 
 
2331
iminvalid
 
2332
""")])
 
2333
        self.assertEqual(
 
2334
            list(osutils.read_mtab('mtab')),
 
2335
            [(b'/', 'ext4'),
 
2336
             (b'/home', 'vfat')])
 
2337
 
 
2338
 
 
2339
class GetFsTypeTests(tests.TestCaseInTempDir):
 
2340
 
 
2341
    def test_returns_string_or_none(self):
 
2342
        ret = osutils.get_fs_type(self.test_dir)
 
2343
        self.assertTrue(isinstance(ret, str) or ret is None)
 
2344
 
 
2345
    def test_returns_most_specific(self):
 
2346
        self.overrideAttr(
 
2347
            osutils, '_FILESYSTEM_FINDER',
 
2348
            osutils.FilesystemFinder(
 
2349
                [(b'/', 'ext4'), (b'/home', 'vfat'),
 
2350
                 (b'/home/jelmer', 'ext2')]))
 
2351
        self.assertEqual(osutils.get_fs_type(b'/home/jelmer/blah'), 'ext2')
 
2352
        self.assertEqual(osutils.get_fs_type('/home/jelmer/blah'), 'ext2')
 
2353
        self.assertEqual(osutils.get_fs_type(b'/home/jelmer'), 'ext2')
 
2354
        self.assertEqual(osutils.get_fs_type(b'/home/martin'), 'vfat')
 
2355
        self.assertEqual(osutils.get_fs_type(b'/home'), 'vfat')
 
2356
        self.assertEqual(osutils.get_fs_type(b'/other'), 'ext4')
 
2357
 
 
2358
    def test_returns_none(self):
 
2359
        self.overrideAttr(
 
2360
            osutils, '_FILESYSTEM_FINDER',
 
2361
            osutils.FilesystemFinder([]))
 
2362
        self.assertIs(osutils.get_fs_type('/home/jelmer/blah'), None)
 
2363
        self.assertIs(osutils.get_fs_type(b'/home/jelmer/blah'), None)
 
2364
        self.assertIs(osutils.get_fs_type('/home/jelmer'), None)