/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-06-29 21:21:43 UTC
  • mto: This revision was merged to the branch mainline in revision 7001.
  • Revision ID: jelmer@jelmer.uk-20180629212143-l5iyocbkjvwva1vm
Fix --list.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005-2016 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
19
 
from cStringIO import StringIO
 
19
from __future__ import absolute_import, division
 
20
 
20
21
import errno
21
22
import os
22
23
import re
 
24
import select
23
25
import socket
24
 
import stat
25
26
import sys
 
27
import tempfile
26
28
import time
27
29
 
28
 
from bzrlib import (
 
30
from .. import (
29
31
    errors,
 
32
    lazy_regex,
30
33
    osutils,
 
34
    symbol_versioning,
31
35
    tests,
32
36
    trace,
33
37
    win32utils,
34
38
    )
35
 
from bzrlib.tests import (
 
39
from ..sixish import (
 
40
    BytesIO,
 
41
    text_type,
 
42
    )
 
43
from . import (
36
44
    features,
37
45
    file_utils,
38
46
    test__walkdirs_win32,
39
47
    )
40
 
 
41
 
 
42
 
class _UTF8DirReaderFeature(tests.Feature):
 
48
from .scenarios import load_tests_apply_scenarios
 
49
 
 
50
 
 
51
class _UTF8DirReaderFeature(features.ModuleAvailableFeature):
43
52
 
44
53
    def _probe(self):
45
54
        try:
46
 
            from bzrlib import _readdir_pyx
 
55
            from .. import _readdir_pyx
 
56
            self._module = _readdir_pyx
47
57
            self.reader = _readdir_pyx.UTF8DirReader
48
58
            return True
49
59
        except ImportError:
50
60
            return False
51
61
 
52
 
    def feature_name(self):
53
 
        return 'bzrlib._readdir_pyx'
54
 
 
55
 
UTF8DirReaderFeature = _UTF8DirReaderFeature()
56
 
 
57
 
term_ios_feature = tests.ModuleAvailableFeature('termios')
 
62
UTF8DirReaderFeature = _UTF8DirReaderFeature('breezy._readdir_pyx')
 
63
 
 
64
term_ios_feature = features.ModuleAvailableFeature('termios')
58
65
 
59
66
 
60
67
def _already_unicode(s):
78
85
    # Some DirReaders are platform specific and even there they may not be
79
86
    # available.
80
87
    if UTF8DirReaderFeature.available():
81
 
        from bzrlib import _readdir_pyx
 
88
        from .. import _readdir_pyx
82
89
        scenarios.append(('utf8',
83
90
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
84
91
                               _native_to_unicode=_utf8_to_unicode)))
85
92
 
86
93
    if test__walkdirs_win32.win32_readdir_feature.available():
87
94
        try:
88
 
            from bzrlib import _walkdirs_win32
 
95
            from .. import _walkdirs_win32
89
96
            scenarios.append(
90
97
                ('win32',
91
98
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
95
102
    return scenarios
96
103
 
97
104
 
98
 
def load_tests(basic_tests, module, loader):
99
 
    suite = loader.suiteClass()
100
 
    dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
101
 
        basic_tests, tests.condition_isinstance(TestDirReader))
102
 
    tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
103
 
    suite.addTest(remaining_tests)
104
 
    return suite
 
105
load_tests = load_tests_apply_scenarios
105
106
 
106
107
 
107
108
class TestContainsWhitespace(tests.TestCase):
108
109
 
109
110
    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'))
 
111
        self.assertTrue(osutils.contains_whitespace(u' '))
 
112
        self.assertTrue(osutils.contains_whitespace(u'hello there'))
 
113
        self.assertTrue(osutils.contains_whitespace(u'hellothere\n'))
 
114
        self.assertTrue(osutils.contains_whitespace(u'hello\nthere'))
 
115
        self.assertTrue(osutils.contains_whitespace(u'hello\rthere'))
 
116
        self.assertTrue(osutils.contains_whitespace(u'hello\tthere'))
116
117
 
117
118
        # \xa0 is "Non-breaking-space" which on some python locales thinks it
118
119
        # 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'))
 
120
        self.assertFalse(osutils.contains_whitespace(u''))
 
121
        self.assertFalse(osutils.contains_whitespace(u'hellothere'))
 
122
        self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
122
123
 
123
124
 
124
125
class TestRename(tests.TestCaseInTempDir):
138
139
        # This should work everywhere
139
140
        self.create_file('a', 'something in a\n')
140
141
        self._fancy_rename('a', 'b')
141
 
        self.failIfExists('a')
142
 
        self.failUnlessExists('b')
 
142
        self.assertPathDoesNotExist('a')
 
143
        self.assertPathExists('b')
143
144
        self.check_file_contents('b', 'something in a\n')
144
145
 
145
146
        self.create_file('a', 'new something in a\n')
152
153
        self.create_file('target', 'data in target\n')
153
154
        self.assertRaises((IOError, OSError), self._fancy_rename,
154
155
                          'missingsource', 'target')
155
 
        self.failUnlessExists('target')
 
156
        self.assertPathExists('target')
156
157
        self.check_file_contents('target', 'data in target\n')
157
158
 
158
159
    def test_fancy_rename_fails_if_source_and_target_missing(self):
163
164
        # Rename should be semi-atomic on all platforms
164
165
        self.create_file('a', 'something in a\n')
165
166
        osutils.rename('a', 'b')
166
 
        self.failIfExists('a')
167
 
        self.failUnlessExists('b')
 
167
        self.assertPathDoesNotExist('a')
 
168
        self.assertPathExists('b')
168
169
        self.check_file_contents('b', 'something in a\n')
169
170
 
170
171
        self.create_file('a', 'new something in a\n')
182
183
        # we can't use failUnlessExists on case-insensitive filesystem
183
184
        # so try to check shape of the tree
184
185
        shape = sorted(os.listdir('.'))
185
 
        self.assertEquals(['A', 'B'], shape)
 
186
        self.assertEqual(['A', 'B'], shape)
186
187
 
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')
 
188
    def test_rename_exception(self):
 
189
        try:
 
190
            osutils.rename('nonexistent_path', 'different_nonexistent_path')
 
191
        except OSError as e:
 
192
            self.assertEqual(e.old_filename, 'nonexistent_path')
 
193
            self.assertEqual(e.new_filename, 'different_nonexistent_path')
 
194
            self.assertTrue('nonexistent_path' in e.strerror)
 
195
            self.assertTrue('different_nonexistent_path' in e.strerror)
193
196
 
194
197
 
195
198
class TestRandChars(tests.TestCase):
222
225
                         (['src'], SRC_FOO_C),
223
226
                         (['src'], 'src'),
224
227
                         ]:
225
 
            self.assert_(osutils.is_inside_any(dirs, fn))
 
228
            self.assertTrue(osutils.is_inside_any(dirs, fn))
226
229
        for dirs, fn in [(['src'], 'srccontrol'),
227
230
                         (['src'], 'srccontrol/foo')]:
228
231
            self.assertFalse(osutils.is_inside_any(dirs, fn))
234
237
                         (['src/bar.c', 'bla/foo.c'], 'src'),
235
238
                         (['src'], 'src'),
236
239
                         ]:
237
 
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
240
            self.assertTrue(osutils.is_inside_or_parent_of_any(dirs, fn))
238
241
 
239
242
        for dirs, fn in [(['src'], 'srccontrol'),
240
243
                         (['srccontrol/foo.c'], 'src'),
242
245
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
243
246
 
244
247
 
 
248
class TestLstat(tests.TestCaseInTempDir):
 
249
 
 
250
    def test_lstat_matches_fstat(self):
 
251
        # On Windows, lstat and fstat don't always agree, primarily in the
 
252
        # 'st_ino' and 'st_dev' fields. So we force them to be '0' in our
 
253
        # custom implementation.
 
254
        if sys.platform == 'win32':
 
255
            # We only have special lstat/fstat if we have the extension.
 
256
            # Without it, we may end up re-reading content when we don't have
 
257
            # to, but otherwise it doesn't effect correctness.
 
258
            self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
259
        f = open('test-file.txt', 'wb')
 
260
        self.addCleanup(f.close)
 
261
        f.write('some content\n')
 
262
        f.flush()
 
263
        self.assertEqualStat(osutils.fstat(f.fileno()),
 
264
                             osutils.lstat('test-file.txt'))
 
265
 
 
266
 
245
267
class TestRmTree(tests.TestCaseInTempDir):
246
268
 
247
269
    def test_rmtree(self):
259
281
 
260
282
        osutils.rmtree('dir')
261
283
 
262
 
        self.failIfExists('dir/file')
263
 
        self.failIfExists('dir')
 
284
        self.assertPathDoesNotExist('dir/file')
 
285
        self.assertPathDoesNotExist('dir')
264
286
 
265
287
 
266
288
class TestDeleteAny(tests.TestCaseInTempDir):
279
301
 
280
302
    def test_file_kind(self):
281
303
        self.build_tree(['file', 'dir/'])
282
 
        self.assertEquals('file', osutils.file_kind('file'))
283
 
        self.assertEquals('directory', osutils.file_kind('dir/'))
 
304
        self.assertEqual('file', osutils.file_kind('file'))
 
305
        self.assertEqual('directory', osutils.file_kind('dir/'))
284
306
        if osutils.has_symlinks():
285
307
            os.symlink('symlink', 'symlink')
286
 
            self.assertEquals('symlink', osutils.file_kind('symlink'))
 
308
            self.assertEqual('symlink', osutils.file_kind('symlink'))
287
309
 
288
310
        # TODO: jam 20060529 Test a block device
289
311
        try:
290
312
            os.lstat('/dev/null')
291
 
        except OSError, e:
 
313
        except OSError as e:
292
314
            if e.errno not in (errno.ENOENT,):
293
315
                raise
294
316
        else:
295
 
            self.assertEquals('chardev', osutils.file_kind('/dev/null'))
 
317
            self.assertEqual('chardev', osutils.file_kind('/dev/null'))
296
318
 
297
319
        mkfifo = getattr(os, 'mkfifo', None)
298
320
        if mkfifo:
299
321
            mkfifo('fifo')
300
322
            try:
301
 
                self.assertEquals('fifo', osutils.file_kind('fifo'))
 
323
                self.assertEqual('fifo', osutils.file_kind('fifo'))
302
324
            finally:
303
325
                os.remove('fifo')
304
326
 
307
329
            s = socket.socket(AF_UNIX)
308
330
            s.bind('socket')
309
331
            try:
310
 
                self.assertEquals('socket', osutils.file_kind('socket'))
 
332
                self.assertEqual('socket', osutils.file_kind('socket'))
311
333
            finally:
312
334
                os.remove('socket')
313
335
 
332
354
 
333
355
        orig_umask = osutils.get_umask()
334
356
        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())
 
357
        os.umask(0o222)
 
358
        self.assertEqual(0o222, osutils.get_umask())
 
359
        os.umask(0o022)
 
360
        self.assertEqual(0o022, osutils.get_umask())
 
361
        os.umask(0o002)
 
362
        self.assertEqual(0o002, osutils.get_umask())
 
363
        os.umask(0o027)
 
364
        self.assertEqual(0o027, osutils.get_umask())
343
365
 
344
366
 
345
367
class TestDateTime(tests.TestCase):
381
403
        self.assertFormatedDelta('2 seconds in the future', -2)
382
404
 
383
405
    def test_format_date(self):
384
 
        self.assertRaises(errors.UnsupportedTimezoneFormat,
 
406
        self.assertRaises(osutils.UnsupportedTimezoneFormat,
385
407
            osutils.format_date, 0, timezone='foo')
386
408
        self.assertIsInstance(osutils.format_date(0), str)
387
 
        self.assertIsInstance(osutils.format_local_date(0), unicode)
 
409
        self.assertIsInstance(osutils.format_local_date(0), text_type)
388
410
        # Testing for the actual value of the local weekday without
389
411
        # duplicating the code from format_date is difficult.
390
412
        # Instead blackbox.test_locale should check for localized
418
440
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
419
441
 
420
442
 
 
443
class TestFdatasync(tests.TestCaseInTempDir):
 
444
 
 
445
    def do_fdatasync(self):
 
446
        f = tempfile.NamedTemporaryFile()
 
447
        osutils.fdatasync(f.fileno())
 
448
        f.close()
 
449
 
 
450
    @staticmethod
 
451
    def raise_eopnotsupp(*args, **kwargs):
 
452
        raise IOError(errno.EOPNOTSUPP, os.strerror(errno.EOPNOTSUPP))
 
453
 
 
454
    @staticmethod
 
455
    def raise_enotsup(*args, **kwargs):
 
456
        raise IOError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
 
457
 
 
458
    def test_fdatasync_handles_system_function(self):
 
459
        self.overrideAttr(os, "fdatasync")
 
460
        self.do_fdatasync()
 
461
 
 
462
    def test_fdatasync_handles_no_fdatasync_no_fsync(self):
 
463
        self.overrideAttr(os, "fdatasync")
 
464
        self.overrideAttr(os, "fsync")
 
465
        self.do_fdatasync()
 
466
 
 
467
    def test_fdatasync_handles_no_EOPNOTSUPP(self):
 
468
        self.overrideAttr(errno, "EOPNOTSUPP")
 
469
        self.do_fdatasync()
 
470
 
 
471
    def test_fdatasync_catches_ENOTSUP(self):
 
472
        enotsup = getattr(errno, "ENOTSUP", None)
 
473
        if enotsup is None:
 
474
            raise tests.TestNotApplicable("No ENOTSUP on this platform")
 
475
        self.overrideAttr(os, "fdatasync", self.raise_enotsup)
 
476
        self.do_fdatasync()
 
477
 
 
478
    def test_fdatasync_catches_EOPNOTSUPP(self):
 
479
        enotsup = getattr(errno, "EOPNOTSUPP", None)
 
480
        if enotsup is None:
 
481
            raise tests.TestNotApplicable("No EOPNOTSUPP on this platform")
 
482
        self.overrideAttr(os, "fdatasync", self.raise_eopnotsupp)
 
483
        self.do_fdatasync()
 
484
 
 
485
 
421
486
class TestLinks(tests.TestCaseInTempDir):
422
487
 
423
488
    def test_dereference_path(self):
424
 
        self.requireFeature(tests.SymlinkFeature)
 
489
        self.requireFeature(features.SymlinkFeature)
425
490
        cwd = osutils.realpath('.')
426
491
        os.mkdir('bar')
427
492
        bar_path = osutils.pathjoin(cwd, 'bar')
455
520
        # Make a file readonly
456
521
        osutils.make_readonly('file')
457
522
        mode = os.lstat('file').st_mode
458
 
        self.assertEqual(mode, mode & 0777555)
 
523
        self.assertEqual(mode, mode & 0o777555)
459
524
 
460
525
        # Make a file writable
461
526
        osutils.make_writable('file')
462
527
        mode = os.lstat('file').st_mode
463
 
        self.assertEqual(mode, mode | 0200)
 
528
        self.assertEqual(mode, mode | 0o200)
464
529
 
465
530
        if osutils.has_symlinks():
466
531
            # should not error when handed a symlink
474
539
 
475
540
class TestCanonicalRelPath(tests.TestCaseInTempDir):
476
541
 
477
 
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
 
542
    _test_needs_features = [features.CaseInsCasePresFilenameFeature]
478
543
 
479
544
    def test_canonical_relpath_simple(self):
480
545
        f = file('MixedCaseName', 'w')
481
546
        f.close()
482
547
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
483
 
        self.failUnlessEqual('work/MixedCaseName', actual)
 
548
        self.assertEqual('work/MixedCaseName', actual)
484
549
 
485
550
    def test_canonical_relpath_missing_tail(self):
486
551
        os.mkdir('MixedCaseParent')
487
552
        actual = osutils.canonical_relpath(self.test_base_dir,
488
553
                                           'mixedcaseparent/nochild')
489
 
        self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
 
554
        self.assertEqual('work/MixedCaseParent/nochild', actual)
490
555
 
491
556
 
492
557
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
536
601
    """Test pumpfile method."""
537
602
 
538
603
    def setUp(self):
539
 
        tests.TestCase.setUp(self)
 
604
        super(TestPumpFile, self).setUp()
540
605
        # create a test datablock
541
606
        self.block_size = 512
542
 
        pattern = '0123456789ABCDEF'
543
 
        self.test_data = pattern * (3 * self.block_size / len(pattern))
 
607
        pattern = b'0123456789ABCDEF'
 
608
        self.test_data = pattern * (3 * self.block_size // len(pattern))
544
609
        self.test_data_len = len(self.test_data)
545
610
 
546
611
    def test_bracket_block_size(self):
550
615
        self.assertTrue(self.test_data_len > self.block_size)
551
616
 
552
617
        from_file = file_utils.FakeReadFile(self.test_data)
553
 
        to_file = StringIO()
 
618
        to_file = BytesIO()
554
619
 
555
 
        # read (max / 2) bytes and verify read size wasn't affected
556
 
        num_bytes_to_read = self.block_size / 2
 
620
        # read (max // 2) bytes and verify read size wasn't affected
 
621
        num_bytes_to_read = self.block_size // 2
557
622
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
558
623
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
559
624
        self.assertEqual(from_file.get_read_count(), 1)
591
656
 
592
657
        # retrieve data in blocks
593
658
        from_file = file_utils.FakeReadFile(self.test_data)
594
 
        to_file = StringIO()
 
659
        to_file = BytesIO()
595
660
        osutils.pumpfile(from_file, to_file, self.test_data_len,
596
661
                         self.block_size)
597
662
 
615
680
 
616
681
        # retrieve data to EOF
617
682
        from_file = file_utils.FakeReadFile(self.test_data)
618
 
        to_file = StringIO()
 
683
        to_file = BytesIO()
619
684
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
620
685
 
621
686
        # verify read size was equal to the maximum read size
635
700
        with this new version."""
636
701
        # retrieve data using default (old) pumpfile method
637
702
        from_file = file_utils.FakeReadFile(self.test_data)
638
 
        to_file = StringIO()
 
703
        to_file = BytesIO()
639
704
        osutils.pumpfile(from_file, to_file)
640
705
 
641
706
        # report error if the data wasn't equal (we only report the size due
649
714
        activity = []
650
715
        def log_activity(length, direction):
651
716
            activity.append((length, direction))
652
 
        from_file = StringIO(self.test_data)
653
 
        to_file = StringIO()
 
717
        from_file = BytesIO(self.test_data)
 
718
        to_file = BytesIO()
654
719
        osutils.pumpfile(from_file, to_file, buff_size=500,
655
720
                         report_activity=log_activity, direction='read')
656
721
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
657
722
                          (36, 'read')], activity)
658
723
 
659
 
        from_file = StringIO(self.test_data)
660
 
        to_file = StringIO()
 
724
        from_file = BytesIO(self.test_data)
 
725
        to_file = BytesIO()
661
726
        del activity[:]
662
727
        osutils.pumpfile(from_file, to_file, buff_size=500,
663
728
                         report_activity=log_activity, direction='write')
665
730
                          (36, 'write')], activity)
666
731
 
667
732
        # And with a limited amount of data
668
 
        from_file = StringIO(self.test_data)
669
 
        to_file = StringIO()
 
733
        from_file = BytesIO(self.test_data)
 
734
        to_file = BytesIO()
670
735
        del activity[:]
671
736
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
672
737
                         report_activity=log_activity, direction='read')
677
742
class TestPumpStringFile(tests.TestCase):
678
743
 
679
744
    def test_empty(self):
680
 
        output = StringIO()
681
 
        osutils.pump_string_file("", output)
682
 
        self.assertEqual("", output.getvalue())
 
745
        output = BytesIO()
 
746
        osutils.pump_string_file(b"", output)
 
747
        self.assertEqual(b"", output.getvalue())
683
748
 
684
749
    def test_more_than_segment_size(self):
685
 
        output = StringIO()
686
 
        osutils.pump_string_file("123456789", output, 2)
687
 
        self.assertEqual("123456789", output.getvalue())
 
750
        output = BytesIO()
 
751
        osutils.pump_string_file(b"123456789", output, 2)
 
752
        self.assertEqual(b"123456789", output.getvalue())
688
753
 
689
754
    def test_segment_size(self):
690
 
        output = StringIO()
691
 
        osutils.pump_string_file("12", output, 2)
692
 
        self.assertEqual("12", output.getvalue())
 
755
        output = BytesIO()
 
756
        osutils.pump_string_file(b"12", output, 2)
 
757
        self.assertEqual(b"12", output.getvalue())
693
758
 
694
759
    def test_segment_size_multiple(self):
695
 
        output = StringIO()
696
 
        osutils.pump_string_file("1234", output, 2)
697
 
        self.assertEqual("1234", output.getvalue())
 
760
        output = BytesIO()
 
761
        osutils.pump_string_file(b"1234", output, 2)
 
762
        self.assertEqual(b"1234", output.getvalue())
698
763
 
699
764
 
700
765
class TestRelpath(tests.TestCase):
763
828
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
764
829
 
765
830
    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'))
 
831
        self.assertRaises(TypeError,
 
832
                          osutils.safe_revision_id, u'bargam')
773
833
 
774
834
    def test_from_unicode_string_unicode_contents(self):
775
 
        self.assertEqual('bargam\xc2\xae',
776
 
                         osutils.safe_revision_id(u'bargam\xae', warn=False))
 
835
        self.assertRaises(TypeError,
 
836
                         osutils.safe_revision_id, u'bargam\xae')
777
837
 
778
838
    def test_from_utf8_string(self):
779
839
        self.assertEqual('foo\xc2\xae',
790
850
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
791
851
 
792
852
    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'))
 
853
        self.assertRaises(TypeError, osutils.safe_file_id, u'bargam')
799
854
 
800
855
    def test_from_unicode_string_unicode_contents(self):
801
 
        self.assertEqual('bargam\xc2\xae',
802
 
                         osutils.safe_file_id(u'bargam\xae', warn=False))
 
856
        self.assertRaises(TypeError,
 
857
                          osutils.safe_file_id, u'bargam\xae')
803
858
 
804
859
    def test_from_utf8_string(self):
805
860
        self.assertEqual('foo\xc2\xae',
810
865
        self.assertEqual(None, osutils.safe_file_id(None))
811
866
 
812
867
 
 
868
class TestSendAll(tests.TestCase):
 
869
 
 
870
    def test_send_with_disconnected_socket(self):
 
871
        class DisconnectedSocket(object):
 
872
            def __init__(self, err):
 
873
                self.err = err
 
874
            def send(self, content):
 
875
                raise self.err
 
876
            def close(self):
 
877
                pass
 
878
        # All of these should be treated as ConnectionReset
 
879
        errs = []
 
880
        for err_cls in (IOError, socket.error):
 
881
            for errnum in osutils._end_of_stream_errors:
 
882
                errs.append(err_cls(errnum))
 
883
        for err in errs:
 
884
            sock = DisconnectedSocket(err)
 
885
            self.assertRaises(errors.ConnectionReset,
 
886
                osutils.send_all, sock, b'some more content')
 
887
 
 
888
    def test_send_with_no_progress(self):
 
889
        # See https://bugs.launchpad.net/bzr/+bug/1047309
 
890
        # It seems that paramiko can get into a state where it doesn't error,
 
891
        # but it returns 0 bytes sent for requests over and over again.
 
892
        class NoSendingSocket(object):
 
893
            def __init__(self):
 
894
                self.call_count = 0
 
895
            def send(self, bytes):
 
896
                self.call_count += 1
 
897
                if self.call_count > 100:
 
898
                    # Prevent the test suite from hanging
 
899
                    raise RuntimeError('too many calls')
 
900
                return 0
 
901
        sock = NoSendingSocket()
 
902
        self.assertRaises(errors.ConnectionReset,
 
903
                          osutils.send_all, sock, b'content')
 
904
        self.assertEqual(1, sock.call_count)
 
905
 
 
906
 
 
907
class TestPosixFuncs(tests.TestCase):
 
908
    """Test that the posix version of normpath returns an appropriate path
 
909
       when used with 2 leading slashes."""
 
910
 
 
911
    def test_normpath(self):
 
912
        self.assertEqual('/etc/shadow', osutils._posix_normpath('/etc/shadow'))
 
913
        self.assertEqual('/etc/shadow', osutils._posix_normpath('//etc/shadow'))
 
914
        self.assertEqual('/etc/shadow', osutils._posix_normpath('///etc/shadow'))
 
915
 
 
916
 
813
917
class TestWin32Funcs(tests.TestCase):
814
918
    """Test that _win32 versions of os utilities return appropriate paths."""
815
919
 
816
920
    def test_abspath(self):
 
921
        self.requireFeature(features.win32_feature)
817
922
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
818
923
        self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
819
924
        self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
832
937
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
833
938
        self.assertEqual('path/to/foo',
834
939
                         osutils._win32_pathjoin('path/to/', 'foo'))
835
 
        self.assertEqual('/foo',
 
940
 
 
941
    def test_pathjoin_late_bugfix(self):
 
942
        if sys.version_info < (2, 7, 6):
 
943
            expected = '/foo'
 
944
        else:
 
945
            expected = 'C:/foo'
 
946
        self.assertEqual(expected,
836
947
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
837
 
        self.assertEqual('/foo',
 
948
        self.assertEqual(expected,
838
949
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
839
950
 
840
951
    def test_normpath(self):
845
956
 
846
957
    def test_getcwd(self):
847
958
        cwd = osutils._win32_getcwd()
848
 
        os_cwd = os.getcwdu()
 
959
        os_cwd = osutils._getcwd()
849
960
        self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
850
961
        # win32 is inconsistent whether it returns lower or upper case
851
962
        # and even if it was consistent the user might type the other
859
970
        self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
860
971
        self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
861
972
 
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
973
 
879
974
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
880
975
    """Test win32 functions that create files."""
881
976
 
882
977
    def test_getcwd(self):
883
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
978
        self.requireFeature(features.UnicodeFilenameFeature)
884
979
        os.mkdir(u'mu-\xb5')
885
980
        os.chdir(u'mu-\xb5')
886
981
        # TODO: jam 20060427 This will probably fail on Mac OSX because
892
987
    def test_minimum_path_selection(self):
893
988
        self.assertEqual(set(),
894
989
            osutils.minimum_path_selection([]))
895
 
        self.assertEqual(set(['a']),
 
990
        self.assertEqual({'a'},
896
991
            osutils.minimum_path_selection(['a']))
897
 
        self.assertEqual(set(['a', 'b']),
 
992
        self.assertEqual({'a', 'b'},
898
993
            osutils.minimum_path_selection(['a', 'b']))
899
 
        self.assertEqual(set(['a/', 'b']),
 
994
        self.assertEqual({'a/', 'b'},
900
995
            osutils.minimum_path_selection(['a/', 'b']))
901
 
        self.assertEqual(set(['a/', 'b']),
 
996
        self.assertEqual({'a/', 'b'},
902
997
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
903
 
        self.assertEqual(set(['a-b', 'a', 'a0b']),
 
998
        self.assertEqual({'a-b', 'a', 'a0b'},
904
999
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
905
1000
 
906
1001
    def test_mkdtemp(self):
916
1011
        b.close()
917
1012
 
918
1013
        osutils._win32_rename('b', 'a')
919
 
        self.failUnlessExists('a')
920
 
        self.failIfExists('b')
 
1014
        self.assertPathExists('a')
 
1015
        self.assertPathDoesNotExist('b')
921
1016
        self.assertFileEqual('baz\n', 'a')
922
1017
 
923
1018
    def test_rename_missing_file(self):
927
1022
 
928
1023
        try:
929
1024
            osutils._win32_rename('b', 'a')
930
 
        except (IOError, OSError), e:
 
1025
        except (IOError, OSError) as e:
931
1026
            self.assertEqual(errno.ENOENT, e.errno)
932
1027
        self.assertFileEqual('foo\n', 'a')
933
1028
 
935
1030
        os.mkdir('a')
936
1031
        try:
937
1032
            osutils._win32_rename('b', 'a')
938
 
        except (IOError, OSError), e:
 
1033
        except (IOError, OSError) as e:
939
1034
            self.assertEqual(errno.ENOENT, e.errno)
940
1035
 
941
1036
    def test_rename_current_dir(self):
947
1042
        # doesn't exist.
948
1043
        try:
949
1044
            osutils._win32_rename('b', '.')
950
 
        except (IOError, OSError), e:
 
1045
        except (IOError, OSError) as e:
951
1046
            self.assertEqual(errno.ENOENT, e.errno)
952
1047
 
953
1048
    def test_splitpath(self):
976
1071
    """Test mac special functions that require directories."""
977
1072
 
978
1073
    def test_getcwd(self):
979
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1074
        self.requireFeature(features.UnicodeFilenameFeature)
980
1075
        os.mkdir(u'B\xe5gfors')
981
1076
        os.chdir(u'B\xe5gfors')
982
1077
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
983
1078
 
984
1079
    def test_getcwd_nonnorm(self):
985
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1080
        self.requireFeature(features.UnicodeFilenameFeature)
986
1081
        # Test that _mac_getcwd() will normalize this path
987
1082
        os.mkdir(u'Ba\u030agfors')
988
1083
        os.chdir(u'Ba\u030agfors')
998
1093
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
999
1094
 
1000
1095
    def test_osutils_binding(self):
1001
 
        from bzrlib.tests import test__chunks_to_lines
 
1096
        from . import test__chunks_to_lines
1002
1097
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
1003
 
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
 
1098
            from .._chunks_to_lines_pyx import chunks_to_lines
1004
1099
        else:
1005
 
            from bzrlib._chunks_to_lines_py import chunks_to_lines
 
1100
            from .._chunks_to_lines_py import chunks_to_lines
1006
1101
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1007
1102
 
1008
1103
 
1071
1166
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
1072
1167
 
1073
1168
    def test_walkdirs_os_error(self):
1074
 
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
 
1169
        # <https://bugs.launchpad.net/bzr/+bug/338653>
1075
1170
        # Pyrex readdir didn't raise useful messages if it had an error
1076
1171
        # reading the directory
1077
1172
        if sys.platform == 'win32':
1078
1173
            raise tests.TestNotApplicable(
1079
1174
                "readdir IOError not tested on win32")
 
1175
        self.requireFeature(features.not_running_as_root)
1080
1176
        os.mkdir("test-unreadable")
1081
1177
        os.chmod("test-unreadable", 0000)
1082
1178
        # must chmod it back so that it can be removed
1083
 
        self.addCleanup(os.chmod, "test-unreadable", 0700)
 
1179
        self.addCleanup(os.chmod, "test-unreadable", 0o700)
1084
1180
        # The error is not raised until the generator is actually evaluated.
1085
1181
        # (It would be ok if it happened earlier but at the moment it
1086
1182
        # doesn't.)
1087
1183
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1088
 
        self.assertEquals('./test-unreadable', e.filename)
1089
 
        self.assertEquals(errno.EACCES, e.errno)
 
1184
        self.assertEqual('./test-unreadable', e.filename)
 
1185
        self.assertEqual(errno.EACCES, e.errno)
1090
1186
        # Ensure the message contains the file name
1091
 
        self.assertContainsRe(str(e), "\./test-unreadable")
 
1187
        self.assertContainsRe(str(e), "\\./test-unreadable")
 
1188
 
 
1189
 
 
1190
    def test_walkdirs_encoding_error(self):
 
1191
        # <https://bugs.launchpad.net/bzr/+bug/488519>
 
1192
        # walkdirs didn't raise a useful message when the filenames
 
1193
        # are not using the filesystem's encoding
 
1194
 
 
1195
        # require a bytestring based filesystem
 
1196
        self.requireFeature(features.ByteStringNamedFilesystem)
 
1197
 
 
1198
        tree = [
 
1199
            '.bzr',
 
1200
            '0file',
 
1201
            '1dir/',
 
1202
            '1dir/0file',
 
1203
            '1dir/1dir/',
 
1204
            '1file'
 
1205
            ]
 
1206
 
 
1207
        self.build_tree(tree)
 
1208
 
 
1209
        # rename the 1file to a latin-1 filename
 
1210
        os.rename("./1file", "\xe8file")
 
1211
        if "\xe8file" not in os.listdir("."):
 
1212
            self.skipTest("Lack filesystem that preserves arbitrary bytes")
 
1213
 
 
1214
        self._save_platform_info()
 
1215
        osutils._fs_enc = 'UTF-8'
 
1216
 
 
1217
        # this should raise on error
 
1218
        def attempt():
 
1219
            for dirdetail, dirblock in osutils.walkdirs('.'):
 
1220
                pass
 
1221
 
 
1222
        self.assertRaises(errors.BadFilenameEncoding, attempt)
1092
1223
 
1093
1224
    def test__walkdirs_utf8(self):
1094
1225
        tree = [
1145
1276
            dirblock[:] = new_dirblock
1146
1277
 
1147
1278
    def _save_platform_info(self):
1148
 
        self.overrideAttr(win32utils, 'winver')
1149
1279
        self.overrideAttr(osutils, '_fs_enc')
1150
1280
        self.overrideAttr(osutils, '_selected_dir_reader')
1151
1281
 
1160
1290
    def test_force_walkdirs_utf8_fs_utf8(self):
1161
1291
        self.requireFeature(UTF8DirReaderFeature)
1162
1292
        self._save_platform_info()
1163
 
        win32utils.winver = None # Avoid the win32 detection code
1164
 
        osutils._fs_enc = 'UTF-8'
1165
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
1293
        osutils._fs_enc = 'utf-8'
 
1294
        self.assertDirReaderIs(
 
1295
            UTF8DirReaderFeature.module.UTF8DirReader)
1166
1296
 
1167
1297
    def test_force_walkdirs_utf8_fs_ascii(self):
1168
1298
        self.requireFeature(UTF8DirReaderFeature)
1169
1299
        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)
 
1300
        osutils._fs_enc = 'ascii'
 
1301
        self.assertDirReaderIs(
 
1302
            UTF8DirReaderFeature.module.UTF8DirReader)
1180
1303
 
1181
1304
    def test_force_walkdirs_utf8_fs_latin1(self):
1182
1305
        self._save_platform_info()
1183
 
        win32utils.winver = None # Avoid the win32 detection code
1184
 
        osutils._fs_enc = 'latin1'
 
1306
        osutils._fs_enc = 'iso-8859-1'
1185
1307
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1186
1308
 
1187
1309
    def test_force_walkdirs_utf8_nt(self):
1188
1310
        # Disabled because the thunk of the whole walkdirs api is disabled.
1189
1311
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1190
1312
        self._save_platform_info()
1191
 
        win32utils.winver = 'Windows NT'
1192
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1313
        from .._walkdirs_win32 import Win32ReadDir
1193
1314
        self.assertDirReaderIs(Win32ReadDir)
1194
1315
 
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)
1200
 
 
1201
1316
    def test_unicode_walkdirs(self):
1202
1317
        """Walkdirs should always return unicode paths."""
1203
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1318
        self.requireFeature(features.UnicodeFilenameFeature)
1204
1319
        name0 = u'0file-\xb6'
1205
1320
        name1 = u'1dir-\u062c\u0648'
1206
1321
        name2 = u'2file-\u0633'
1243
1358
 
1244
1359
        The abspath portion might be in unicode or utf-8
1245
1360
        """
1246
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1361
        self.requireFeature(features.UnicodeFilenameFeature)
1247
1362
        name0 = u'0file-\xb6'
1248
1363
        name1 = u'1dir-\u062c\u0648'
1249
1364
        name2 = u'2file-\u0633'
1283
1398
        # all abspaths are Unicode, and encode them back into utf8.
1284
1399
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1285
1400
            self.assertIsInstance(dirdetail[0], str)
1286
 
            if isinstance(dirdetail[1], unicode):
 
1401
            if isinstance(dirdetail[1], text_type):
1287
1402
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1288
1403
                dirblock = [list(info) for info in dirblock]
1289
1404
                for info in dirblock:
1290
 
                    self.assertIsInstance(info[4], unicode)
 
1405
                    self.assertIsInstance(info[4], text_type)
1291
1406
                    info[4] = info[4].encode('utf8')
1292
1407
            new_dirblock = []
1293
1408
            for info in dirblock:
1304
1419
 
1305
1420
        The abspath portion should be in unicode
1306
1421
        """
1307
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1422
        self.requireFeature(features.UnicodeFilenameFeature)
1308
1423
        # Use the unicode reader. TODO: split into driver-and-driven unit
1309
1424
        # tests.
1310
1425
        self._save_platform_info()
1351
1466
 
1352
1467
    def test__walkdirs_utf8_win32readdir(self):
1353
1468
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1354
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1355
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1469
        self.requireFeature(features.UnicodeFilenameFeature)
 
1470
        from .._walkdirs_win32 import Win32ReadDir
1356
1471
        self._save_platform_info()
1357
1472
        osutils._selected_dir_reader = Win32ReadDir()
1358
1473
        name0u = u'0file-\xb6'
1408
1523
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1409
1524
        """make sure our Stat values are valid"""
1410
1525
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1411
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1412
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1526
        self.requireFeature(features.UnicodeFilenameFeature)
 
1527
        from .._walkdirs_win32 import Win32ReadDir
1413
1528
        name0u = u'0file-\xb6'
1414
1529
        name0 = name0u.encode('utf8')
1415
1530
        self.build_tree([name0u])
1432
1547
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1433
1548
        """make sure our Stat values are valid"""
1434
1549
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1435
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1436
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1550
        self.requireFeature(features.UnicodeFilenameFeature)
 
1551
        from .._walkdirs_win32 import Win32ReadDir
1437
1552
        name0u = u'0dir-\u062c\u0648'
1438
1553
        name0 = name0u.encode('utf8')
1439
1554
        self.build_tree([name0u + '/'])
1538
1653
        self.assertEqual(['c'], os.listdir('target/b'))
1539
1654
 
1540
1655
    def test_copy_tree_symlinks(self):
1541
 
        self.requireFeature(tests.SymlinkFeature)
 
1656
        self.requireFeature(features.SymlinkFeature)
1542
1657
        self.build_tree(['source/'])
1543
1658
        os.symlink('a/generic/path', 'source/lnk')
1544
1659
        osutils.copy_tree('source', 'target')
1554
1669
            processed_files.append(('d', from_path, to_path))
1555
1670
        def link_handler(from_path, to_path):
1556
1671
            processed_links.append((from_path, to_path))
1557
 
        handlers = {'file':file_handler,
1558
 
                    'directory':dir_handler,
1559
 
                    'symlink':link_handler,
 
1672
        handlers = {'file': file_handler,
 
1673
                    'directory': dir_handler,
 
1674
                    'symlink': link_handler,
1560
1675
                   }
1561
1676
 
1562
1677
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1569
1684
                          ('d', 'source/b', 'target/b'),
1570
1685
                          ('f', 'source/b/c', 'target/b/c'),
1571
1686
                         ], processed_files)
1572
 
        self.failIfExists('target')
 
1687
        self.assertPathDoesNotExist('target')
1573
1688
        if osutils.has_symlinks():
1574
1689
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1575
1690
 
1580
1695
    def setUp(self):
1581
1696
        super(TestSetUnsetEnv, self).setUp()
1582
1697
 
1583
 
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
 
1698
        self.assertEqual(None, os.environ.get('BRZ_TEST_ENV_VAR'),
1584
1699
                         'Environment was not cleaned up properly.'
1585
 
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
 
1700
                         ' Variable BRZ_TEST_ENV_VAR should not exist.')
1586
1701
        def cleanup():
1587
 
            if 'BZR_TEST_ENV_VAR' in os.environ:
1588
 
                del os.environ['BZR_TEST_ENV_VAR']
 
1702
            if 'BRZ_TEST_ENV_VAR' in os.environ:
 
1703
                del os.environ['BRZ_TEST_ENV_VAR']
1589
1704
        self.addCleanup(cleanup)
1590
1705
 
1591
1706
    def test_set(self):
1592
1707
        """Test that we can set an env variable"""
1593
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1708
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
1594
1709
        self.assertEqual(None, old)
1595
 
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
 
1710
        self.assertEqual('foo', os.environ.get('BRZ_TEST_ENV_VAR'))
1596
1711
 
1597
1712
    def test_double_set(self):
1598
1713
        """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')
 
1714
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
 
1715
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'bar')
1601
1716
        self.assertEqual('foo', old)
1602
 
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
 
1717
        self.assertEqual('bar', os.environ.get('BRZ_TEST_ENV_VAR'))
1603
1718
 
1604
1719
    def test_unicode(self):
1605
1720
        """Environment can only contain plain strings
1612
1727
                'Cannot find a unicode character that works in encoding %s'
1613
1728
                % (osutils.get_user_encoding(),))
1614
1729
 
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'))
 
1730
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', uni_val)
 
1731
        self.assertEqual(env_val, os.environ.get('BRZ_TEST_ENV_VAR'))
1617
1732
 
1618
1733
    def test_unset(self):
1619
1734
        """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)
 
1735
        osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', 'foo')
 
1736
        old = osutils.set_or_unset_env('BRZ_TEST_ENV_VAR', None)
1622
1737
        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)
 
1738
        self.assertEqual(None, os.environ.get('BRZ_TEST_ENV_VAR'))
 
1739
        self.assertFalse('BRZ_TEST_ENV_VAR' in os.environ)
1625
1740
 
1626
1741
 
1627
1742
class TestSizeShaFile(tests.TestCaseInTempDir):
1628
1743
 
1629
1744
    def test_sha_empty(self):
1630
 
        self.build_tree_contents([('foo', '')])
 
1745
        self.build_tree_contents([('foo', b'')])
1631
1746
        expected_sha = osutils.sha_string('')
1632
1747
        f = open('foo')
1633
1748
        self.addCleanup(f.close)
1636
1751
        self.assertEqual(expected_sha, sha)
1637
1752
 
1638
1753
    def test_sha_mixed_endings(self):
1639
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1754
        text = b'test\r\nwith\nall\rpossible line endings\r\n'
1640
1755
        self.build_tree_contents([('foo', text)])
1641
1756
        expected_sha = osutils.sha_string(text)
1642
1757
        f = open('foo', 'rb')
1649
1764
class TestShaFileByName(tests.TestCaseInTempDir):
1650
1765
 
1651
1766
    def test_sha_empty(self):
1652
 
        self.build_tree_contents([('foo', '')])
 
1767
        self.build_tree_contents([('foo', b'')])
1653
1768
        expected_sha = osutils.sha_string('')
1654
1769
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1655
1770
 
1656
1771
    def test_sha_mixed_endings(self):
1657
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1772
        text = b'test\r\nwith\nall\rpossible line endings\r\n'
1658
1773
        self.build_tree_contents([('foo', text)])
1659
1774
        expected_sha = osutils.sha_string(text)
1660
1775
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1663
1778
class TestResourceLoading(tests.TestCaseInTempDir):
1664
1779
 
1665
1780
    def test_resource_string(self):
1666
 
        # test resource in bzrlib
1667
 
        text = osutils.resource_string('bzrlib', 'debug.py')
 
1781
        # test resource in breezy
 
1782
        text = osutils.resource_string('breezy', 'debug.py')
1668
1783
        self.assertContainsRe(text, "debug_flags = set()")
1669
 
        # test resource under bzrlib
1670
 
        text = osutils.resource_string('bzrlib.ui', 'text.py')
 
1784
        # test resource under breezy
 
1785
        text = osutils.resource_string('breezy.ui', 'text.py')
1671
1786
        self.assertContainsRe(text, "class TextUIFactory")
1672
1787
        # test unsupported package
1673
1788
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1674
1789
            'yyy.xx')
1675
1790
        # 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))
 
1791
        self.assertRaises(IOError, osutils.resource_string, 'breezy', 'yyy.xx')
1695
1792
 
1696
1793
 
1697
1794
class TestDirReader(tests.TestCaseInTempDir):
1698
1795
 
 
1796
    scenarios = dir_reader_scenarios()
 
1797
 
1699
1798
    # Set by load_tests
1700
1799
    _dir_reader_class = None
1701
1800
    _native_to_unicode = None
1702
1801
 
1703
1802
    def setUp(self):
1704
 
        tests.TestCaseInTempDir.setUp(self)
 
1803
        super(TestDirReader, self).setUp()
1705
1804
        self.overrideAttr(osutils,
1706
1805
                          '_selected_dir_reader', self._dir_reader_class())
1707
1806
 
1801
1900
        return filtered_dirblocks
1802
1901
 
1803
1902
    def test_walk_unicode_tree(self):
1804
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1903
        self.requireFeature(features.UnicodeFilenameFeature)
1805
1904
        tree, expected_dirblocks = self._get_unicode_tree()
1806
1905
        self.build_tree(tree)
1807
1906
        result = list(osutils._walkdirs_utf8('.'))
1808
1907
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1809
1908
 
1810
1909
    def test_symlink(self):
1811
 
        self.requireFeature(tests.SymlinkFeature)
1812
 
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1910
        self.requireFeature(features.SymlinkFeature)
 
1911
        self.requireFeature(features.UnicodeFilenameFeature)
1813
1912
        target = u'target\N{Euro Sign}'
1814
1913
        link_name = u'l\N{Euro Sign}nk'
1815
1914
        os.symlink(target, link_name)
1833
1932
    But prior python versions failed to properly encode the passed unicode
1834
1933
    string.
1835
1934
    """
1836
 
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
 
1935
    _test_needs_features = [features.SymlinkFeature, features.UnicodeFilenameFeature]
1837
1936
 
1838
1937
    def setUp(self):
1839
1938
        super(tests.TestCaseInTempDir, self).setUp()
1842
1941
        os.symlink(self.target, self.link)
1843
1942
 
1844
1943
    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))
 
1944
        self.assertEqual(self.target,  os.readlink(self.link))
1849
1945
 
1850
1946
    def test_os_readlink_link_decoding(self):
1851
 
        self.assertEquals(self.target.encode(osutils._fs_enc),
 
1947
        self.assertEqual(self.target.encode(osutils._fs_enc),
1852
1948
                          os.readlink(self.link.encode(osutils._fs_enc)))
1853
1949
 
1854
1950
 
1863
1959
        self.assertIsInstance(concurrency, int)
1864
1960
 
1865
1961
    def test_local_concurrency_environment_variable(self):
1866
 
        os.environ['BZR_CONCURRENCY'] = '2'
 
1962
        self.overrideEnv('BRZ_CONCURRENCY', '2')
1867
1963
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1868
 
        os.environ['BZR_CONCURRENCY'] = '3'
 
1964
        self.overrideEnv('BRZ_CONCURRENCY', '3')
1869
1965
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1870
 
        os.environ['BZR_CONCURRENCY'] = 'foo'
 
1966
        self.overrideEnv('BRZ_CONCURRENCY', 'foo')
1871
1967
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1872
1968
 
1873
1969
    def test_option_concurrency(self):
1874
 
        os.environ['BZR_CONCURRENCY'] = '1'
 
1970
        self.overrideEnv('BRZ_CONCURRENCY', '1')
1875
1971
        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))
 
1972
        # Command line overrides environment variable
 
1973
        self.assertEqual('42', os.environ['BRZ_CONCURRENCY'])
 
1974
        self.assertEqual(42, osutils.local_concurrency(use_cache=False))
1879
1975
 
1880
1976
 
1881
1977
class TestFailedToLoadExtension(tests.TestCase):
1882
1978
 
1883
1979
    def _try_loading(self):
1884
1980
        try:
1885
 
            import bzrlib._fictional_extension_py
1886
 
        except ImportError, e:
 
1981
            import breezy._fictional_extension_py
 
1982
        except ImportError as e:
1887
1983
            osutils.failed_to_load_extension(e)
1888
1984
            return True
1889
1985
 
1894
1990
    def test_failure_to_load(self):
1895
1991
        self._try_loading()
1896
1992
        self.assertLength(1, osutils._extension_load_failures)
1897
 
        self.assertEquals(osutils._extension_load_failures[0],
 
1993
        self.assertEqual(osutils._extension_load_failures[0],
1898
1994
            "No module named _fictional_extension_py")
1899
1995
 
1900
1996
    def test_report_extension_load_failures_no_warning(self):
1904
2000
        self.assertLength(0, warnings)
1905
2001
 
1906
2002
    def test_report_extension_load_failures_message(self):
1907
 
        log = StringIO()
 
2003
        log = BytesIO()
1908
2004
        trace.push_log_file(log)
1909
2005
        self.assertTrue(self._try_loading())
1910
2006
        osutils.report_extension_load_failures()
1911
2007
        self.assertContainsRe(
1912
2008
            log.getvalue(),
1913
 
            r"bzr: warning: some compiled extensions could not be loaded; "
1914
 
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
 
2009
            r"brz: warning: some compiled extensions could not be loaded; "
 
2010
            "see ``brz help missing-extensions``\n"
1915
2011
            )
1916
2012
 
1917
2013
 
1918
2014
class TestTerminalWidth(tests.TestCase):
1919
2015
 
 
2016
    def setUp(self):
 
2017
        super(TestTerminalWidth, self).setUp()
 
2018
        self._orig_terminal_size_state = osutils._terminal_size_state
 
2019
        self._orig_first_terminal_size = osutils._first_terminal_size
 
2020
        self.addCleanup(self.restore_osutils_globals)
 
2021
        osutils._terminal_size_state = 'no_data'
 
2022
        osutils._first_terminal_size = None
 
2023
 
 
2024
    def restore_osutils_globals(self):
 
2025
        osutils._terminal_size_state = self._orig_terminal_size_state
 
2026
        osutils._first_terminal_size = self._orig_first_terminal_size
 
2027
 
1920
2028
    def replace_stdout(self, new):
1921
2029
        self.overrideAttr(sys, 'stdout', new)
1922
2030
 
1934
2042
    def test_default_values(self):
1935
2043
        self.assertEqual(80, osutils.default_terminal_width)
1936
2044
 
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'
 
2045
    def test_defaults_to_BRZ_COLUMNS(self):
 
2046
        # BRZ_COLUMNS is set by the test framework
 
2047
        self.assertNotEqual('12', os.environ['BRZ_COLUMNS'])
 
2048
        self.overrideEnv('BRZ_COLUMNS', '12')
1941
2049
        self.assertEqual(12, osutils.terminal_width())
1942
2050
 
 
2051
    def test_BRZ_COLUMNS_0_no_limit(self):
 
2052
        self.overrideEnv('BRZ_COLUMNS', '0')
 
2053
        self.assertEqual(None, osutils.terminal_width())
 
2054
 
1943
2055
    def test_falls_back_to_COLUMNS(self):
1944
 
        del os.environ['BZR_COLUMNS']
 
2056
        self.overrideEnv('BRZ_COLUMNS', None)
1945
2057
        self.assertNotEqual('42', os.environ['COLUMNS'])
1946
2058
        self.set_fake_tty()
1947
 
        os.environ['COLUMNS'] = '42'
 
2059
        self.overrideEnv('COLUMNS', '42')
1948
2060
        self.assertEqual(42, osutils.terminal_width())
1949
2061
 
1950
2062
    def test_tty_default_without_columns(self):
1951
 
        del os.environ['BZR_COLUMNS']
1952
 
        del os.environ['COLUMNS']
 
2063
        self.overrideEnv('BRZ_COLUMNS', None)
 
2064
        self.overrideEnv('COLUMNS', None)
1953
2065
 
1954
2066
        def terminal_size(w, h):
1955
2067
            return 42, 42
1962
2074
        self.assertEqual(42, osutils.terminal_width())
1963
2075
 
1964
2076
    def test_non_tty_default_without_columns(self):
1965
 
        del os.environ['BZR_COLUMNS']
1966
 
        del os.environ['COLUMNS']
 
2077
        self.overrideEnv('BRZ_COLUMNS', None)
 
2078
        self.overrideEnv('COLUMNS', None)
1967
2079
        self.replace_stdout(None)
1968
2080
        self.assertEqual(None, osutils.terminal_width())
1969
2081
 
1979
2091
        else:
1980
2092
            self.overrideAttr(termios, 'TIOCGWINSZ')
1981
2093
            del termios.TIOCGWINSZ
1982
 
        del os.environ['BZR_COLUMNS']
1983
 
        del os.environ['COLUMNS']
 
2094
        self.overrideEnv('BRZ_COLUMNS', None)
 
2095
        self.overrideEnv('COLUMNS', None)
1984
2096
        # Whatever the result is, if we don't raise an exception, it's ok.
1985
2097
        osutils.terminal_width()
1986
2098
 
 
2099
 
1987
2100
class TestCreationOps(tests.TestCaseInTempDir):
1988
2101
    _test_needs_features = [features.chown_feature]
1989
2102
 
1990
2103
    def setUp(self):
1991
 
        tests.TestCaseInTempDir.setUp(self)
 
2104
        super(TestCreationOps, self).setUp()
1992
2105
        self.overrideAttr(os, 'chown', self._dummy_chown)
1993
2106
 
1994
2107
        # params set by call to _dummy_chown
2004
2117
        osutils.copy_ownership_from_path('test_file', ownsrc)
2005
2118
 
2006
2119
        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)
 
2120
        self.assertEqual(self.path, 'test_file')
 
2121
        self.assertEqual(self.uid, s.st_uid)
 
2122
        self.assertEqual(self.gid, s.st_gid)
2010
2123
 
2011
2124
    def test_copy_ownership_nonesrc(self):
2012
2125
        """copy_ownership_from_path test with src=None."""
2015
2128
        osutils.copy_ownership_from_path('test_file')
2016
2129
 
2017
2130
        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)
 
2131
        self.assertEqual(self.path, 'test_file')
 
2132
        self.assertEqual(self.uid, s.st_uid)
 
2133
        self.assertEqual(self.gid, s.st_gid)
 
2134
 
 
2135
 
 
2136
class TestPathFromEnviron(tests.TestCase):
 
2137
 
 
2138
    def test_is_unicode(self):
 
2139
        self.overrideEnv('BRZ_TEST_PATH', './anywhere at all/')
 
2140
        path = osutils.path_from_environ('BRZ_TEST_PATH')
 
2141
        self.assertIsInstance(path, text_type)
 
2142
        self.assertEqual(u'./anywhere at all/', path)
 
2143
 
 
2144
    def test_posix_path_env_ascii(self):
 
2145
        self.overrideEnv('BRZ_TEST_PATH', '/tmp')
 
2146
        home = osutils._posix_path_from_environ('BRZ_TEST_PATH')
 
2147
        self.assertIsInstance(home, text_type)
 
2148
        self.assertEqual(u'/tmp', home)
 
2149
 
 
2150
    def test_posix_path_env_unicode(self):
 
2151
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2152
        self.overrideEnv('BRZ_TEST_PATH', '/home/\xa7test')
 
2153
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2154
        self.assertEqual(u'/home/\xa7test',
 
2155
            osutils._posix_path_from_environ('BRZ_TEST_PATH'))
 
2156
        osutils._fs_enc = "iso8859-5"
 
2157
        self.assertEqual(u'/home/\u0407test',
 
2158
            osutils._posix_path_from_environ('BRZ_TEST_PATH'))
 
2159
        osutils._fs_enc = "utf-8"
 
2160
        self.assertRaises(errors.BadFilenameEncoding,
 
2161
            osutils._posix_path_from_environ, 'BRZ_TEST_PATH')
 
2162
 
 
2163
 
 
2164
class TestGetHomeDir(tests.TestCase):
 
2165
 
 
2166
    def test_is_unicode(self):
 
2167
        home = osutils._get_home_dir()
 
2168
        self.assertIsInstance(home, text_type)
 
2169
 
 
2170
    def test_posix_homeless(self):
 
2171
        self.overrideEnv('HOME', None)
 
2172
        home = osutils._get_home_dir()
 
2173
        self.assertIsInstance(home, text_type)
 
2174
 
 
2175
    def test_posix_home_ascii(self):
 
2176
        self.overrideEnv('HOME', '/home/test')
 
2177
        home = osutils._posix_get_home_dir()
 
2178
        self.assertIsInstance(home, text_type)
 
2179
        self.assertEqual(u'/home/test', home)
 
2180
 
 
2181
    def test_posix_home_unicode(self):
 
2182
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2183
        self.overrideEnv('HOME', '/home/\xa7test')
 
2184
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2185
        self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
 
2186
        osutils._fs_enc = "iso8859-5"
 
2187
        self.assertEqual(u'/home/\u0407test', osutils._posix_get_home_dir())
 
2188
        osutils._fs_enc = "utf-8"
 
2189
        self.assertRaises(errors.BadFilenameEncoding,
 
2190
            osutils._posix_get_home_dir)
 
2191
 
 
2192
 
 
2193
class TestGetuserUnicode(tests.TestCase):
 
2194
 
 
2195
    def test_is_unicode(self):
 
2196
        user = osutils.getuser_unicode()
 
2197
        self.assertIsInstance(user, text_type)
 
2198
 
 
2199
    def envvar_to_override(self):
 
2200
        if sys.platform == "win32":
 
2201
            # Disable use of platform calls on windows so envvar is used
 
2202
            self.overrideAttr(win32utils, 'has_ctypes', False)
 
2203
            return 'USERNAME' # only variable used on windows
 
2204
        return 'LOGNAME' # first variable checked by getpass.getuser()
 
2205
 
 
2206
    def test_ascii_user(self):
 
2207
        self.overrideEnv(self.envvar_to_override(), 'jrandom')
 
2208
        self.assertEqual(u'jrandom', osutils.getuser_unicode())
 
2209
 
 
2210
    def test_unicode_user(self):
 
2211
        ue = osutils.get_user_encoding()
 
2212
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
2213
        if uni_val is None:
 
2214
            raise tests.TestSkipped(
 
2215
                'Cannot find a unicode character that works in encoding %s'
 
2216
                % (osutils.get_user_encoding(),))
 
2217
        uni_username = u'jrandom' + uni_val
 
2218
        encoded_username = uni_username.encode(ue)
 
2219
        self.overrideEnv(self.envvar_to_override(), encoded_username)
 
2220
        self.assertEqual(uni_username, osutils.getuser_unicode())
 
2221
 
 
2222
 
 
2223
class TestBackupNames(tests.TestCase):
 
2224
 
 
2225
    def setUp(self):
 
2226
        super(TestBackupNames, self).setUp()
 
2227
        self.backups = []
 
2228
 
 
2229
    def backup_exists(self, name):
 
2230
        return name in self.backups
 
2231
 
 
2232
    def available_backup_name(self, name):
 
2233
        backup_name = osutils.available_backup_name(name, self.backup_exists)
 
2234
        self.backups.append(backup_name)
 
2235
        return backup_name
 
2236
 
 
2237
    def assertBackupName(self, expected, name):
 
2238
        self.assertEqual(expected, self.available_backup_name(name))
 
2239
 
 
2240
    def test_empty(self):
 
2241
        self.assertBackupName('file.~1~', 'file')
 
2242
 
 
2243
    def test_existing(self):
 
2244
        self.available_backup_name('file')
 
2245
        self.available_backup_name('file')
 
2246
        self.assertBackupName('file.~3~', 'file')
 
2247
        # Empty slots are found, this is not a strict requirement and may be
 
2248
        # revisited if we test against all implementations.
 
2249
        self.backups.remove('file.~2~')
 
2250
        self.assertBackupName('file.~2~', 'file')
 
2251
 
 
2252
 
 
2253
class TestFindExecutableInPath(tests.TestCase):
 
2254
 
 
2255
    def test_windows(self):
 
2256
        if sys.platform != 'win32':
 
2257
            raise tests.TestSkipped('test requires win32')
 
2258
        self.assertTrue(osutils.find_executable_on_path('explorer') is not None)
 
2259
        self.assertTrue(
 
2260
            osutils.find_executable_on_path('explorer.exe') is not None)
 
2261
        self.assertTrue(
 
2262
            osutils.find_executable_on_path('EXPLORER.EXE') is not None)
 
2263
        self.assertTrue(
 
2264
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2265
        self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
 
2266
        
 
2267
    def test_windows_app_path(self):
 
2268
        if sys.platform != 'win32':
 
2269
            raise tests.TestSkipped('test requires win32')
 
2270
        # Override PATH env var so that exe can only be found on App Path
 
2271
        self.overrideEnv('PATH', '')
 
2272
        # Internt Explorer is always registered in the App Path
 
2273
        self.assertTrue(osutils.find_executable_on_path('iexplore') is not None)
 
2274
 
 
2275
    def test_other(self):
 
2276
        if sys.platform == 'win32':
 
2277
            raise tests.TestSkipped('test requires non-win32')
 
2278
        self.assertTrue(osutils.find_executable_on_path('sh') is not None)
 
2279
        self.assertTrue(
 
2280
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2281
 
 
2282
 
 
2283
class TestEnvironmentErrors(tests.TestCase):
 
2284
    """Test handling of environmental errors"""
 
2285
 
 
2286
    def test_is_oserror(self):
 
2287
        self.assertTrue(osutils.is_environment_error(
 
2288
            OSError(errno.EINVAL, "Invalid parameter")))
 
2289
 
 
2290
    def test_is_ioerror(self):
 
2291
        self.assertTrue(osutils.is_environment_error(
 
2292
            IOError(errno.EINVAL, "Invalid parameter")))
 
2293
 
 
2294
    def test_is_socket_error(self):
 
2295
        self.assertTrue(osutils.is_environment_error(
 
2296
            socket.error(errno.EINVAL, "Invalid parameter")))
 
2297
 
 
2298
    def test_is_select_error(self):
 
2299
        self.assertTrue(osutils.is_environment_error(
 
2300
            select.error(errno.EINVAL, "Invalid parameter")))
 
2301
 
 
2302
    def test_is_pywintypes_error(self):
 
2303
        self.requireFeature(features.pywintypes)
 
2304
        import pywintypes
 
2305
        self.assertTrue(osutils.is_environment_error(
 
2306
            pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))