/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: 2018-11-16 23:19:12 UTC
  • mfrom: (7180 work)
  • mto: This revision was merged to the branch mainline in revision 7294.
  • Revision ID: jelmer@jelmer.uk-20181116231912-e043vpq22bdkxa6q
Merge trunk.

Show diffs side-by-side

added added

removed removed

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