~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Aaron Bentley
  • Date: 2012-07-19 16:57:16 UTC
  • mto: This revision was merged to the branch mainline in revision 6540.
  • Revision ID: aaron@aaronbentley.com-20120719165716-b4iupzkb17b9l9wx
Avoid branch write lock to preserve VFS call count.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2005-2011 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
 
19
from cStringIO import StringIO
19
20
import errno
20
21
import os
 
22
import re
 
23
import select
21
24
import socket
22
 
import stat
23
25
import sys
 
26
import time
24
27
 
25
 
import bzrlib
26
28
from bzrlib import (
27
29
    errors,
 
30
    lazy_regex,
28
31
    osutils,
 
32
    symbol_versioning,
 
33
    tests,
 
34
    trace,
29
35
    win32utils,
30
36
    )
31
 
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
32
 
from bzrlib.osutils import (
33
 
        is_inside_any,
34
 
        is_inside_or_parent_of_any,
35
 
        pathjoin,
36
 
        )
37
37
from bzrlib.tests import (
38
 
        probe_unicode_in_user_encoding,
39
 
        StringIOWrapper,
40
 
        SymlinkFeature,
41
 
        TestCase,
42
 
        TestCaseInTempDir,
43
 
        TestSkipped,
44
 
        )
45
 
 
46
 
 
47
 
class TestOSUtils(TestCaseInTempDir):
 
38
    features,
 
39
    file_utils,
 
40
    test__walkdirs_win32,
 
41
    )
 
42
from bzrlib.tests.scenarios import load_tests_apply_scenarios
 
43
 
 
44
 
 
45
class _UTF8DirReaderFeature(features.Feature):
 
46
 
 
47
    def _probe(self):
 
48
        try:
 
49
            from bzrlib import _readdir_pyx
 
50
            self.reader = _readdir_pyx.UTF8DirReader
 
51
            return True
 
52
        except ImportError:
 
53
            return False
 
54
 
 
55
    def feature_name(self):
 
56
        return 'bzrlib._readdir_pyx'
 
57
 
 
58
UTF8DirReaderFeature = features.ModuleAvailableFeature('bzrlib._readdir_pyx')
 
59
 
 
60
term_ios_feature = features.ModuleAvailableFeature('termios')
 
61
 
 
62
 
 
63
def _already_unicode(s):
 
64
    return s
 
65
 
 
66
 
 
67
def _utf8_to_unicode(s):
 
68
    return s.decode('UTF-8')
 
69
 
 
70
 
 
71
def dir_reader_scenarios():
 
72
    # For each dir reader we define:
 
73
 
 
74
    # - native_to_unicode: a function converting the native_abspath as returned
 
75
    #   by DirReader.read_dir to its unicode representation
 
76
 
 
77
    # UnicodeDirReader is the fallback, it should be tested on all platforms.
 
78
    scenarios = [('unicode',
 
79
                  dict(_dir_reader_class=osutils.UnicodeDirReader,
 
80
                       _native_to_unicode=_already_unicode))]
 
81
    # Some DirReaders are platform specific and even there they may not be
 
82
    # available.
 
83
    if UTF8DirReaderFeature.available():
 
84
        from bzrlib import _readdir_pyx
 
85
        scenarios.append(('utf8',
 
86
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
 
87
                               _native_to_unicode=_utf8_to_unicode)))
 
88
 
 
89
    if test__walkdirs_win32.win32_readdir_feature.available():
 
90
        try:
 
91
            from bzrlib import _walkdirs_win32
 
92
            scenarios.append(
 
93
                ('win32',
 
94
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
 
95
                      _native_to_unicode=_already_unicode)))
 
96
        except ImportError:
 
97
            pass
 
98
    return scenarios
 
99
 
 
100
 
 
101
load_tests = load_tests_apply_scenarios
 
102
 
 
103
 
 
104
class TestContainsWhitespace(tests.TestCase):
48
105
 
49
106
    def test_contains_whitespace(self):
50
 
        self.failUnless(osutils.contains_whitespace(u' '))
51
 
        self.failUnless(osutils.contains_whitespace(u'hello there'))
52
 
        self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
53
 
        self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
54
 
        self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
55
 
        self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
 
107
        self.assertTrue(osutils.contains_whitespace(u' '))
 
108
        self.assertTrue(osutils.contains_whitespace(u'hello there'))
 
109
        self.assertTrue(osutils.contains_whitespace(u'hellothere\n'))
 
110
        self.assertTrue(osutils.contains_whitespace(u'hello\nthere'))
 
111
        self.assertTrue(osutils.contains_whitespace(u'hello\rthere'))
 
112
        self.assertTrue(osutils.contains_whitespace(u'hello\tthere'))
56
113
 
57
114
        # \xa0 is "Non-breaking-space" which on some python locales thinks it
58
115
        # is whitespace, but we do not.
59
 
        self.failIf(osutils.contains_whitespace(u''))
60
 
        self.failIf(osutils.contains_whitespace(u'hellothere'))
61
 
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
 
116
        self.assertFalse(osutils.contains_whitespace(u''))
 
117
        self.assertFalse(osutils.contains_whitespace(u'hellothere'))
 
118
        self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
 
119
 
 
120
 
 
121
class TestRename(tests.TestCaseInTempDir):
 
122
 
 
123
    def create_file(self, filename, content):
 
124
        f = open(filename, 'wb')
 
125
        try:
 
126
            f.write(content)
 
127
        finally:
 
128
            f.close()
 
129
 
 
130
    def _fancy_rename(self, a, b):
 
131
        osutils.fancy_rename(a, b, rename_func=os.rename,
 
132
                             unlink_func=os.unlink)
62
133
 
63
134
    def test_fancy_rename(self):
64
135
        # This should work everywhere
65
 
        def rename(a, b):
66
 
            osutils.fancy_rename(a, b,
67
 
                    rename_func=os.rename,
68
 
                    unlink_func=os.unlink)
69
 
 
70
 
        open('a', 'wb').write('something in a\n')
71
 
        rename('a', 'b')
72
 
        self.failIfExists('a')
73
 
        self.failUnlessExists('b')
 
136
        self.create_file('a', 'something in a\n')
 
137
        self._fancy_rename('a', 'b')
 
138
        self.assertPathDoesNotExist('a')
 
139
        self.assertPathExists('b')
74
140
        self.check_file_contents('b', 'something in a\n')
75
141
 
76
 
        open('a', 'wb').write('new something in a\n')
77
 
        rename('b', 'a')
 
142
        self.create_file('a', 'new something in a\n')
 
143
        self._fancy_rename('b', 'a')
78
144
 
79
145
        self.check_file_contents('a', 'something in a\n')
80
146
 
 
147
    def test_fancy_rename_fails_source_missing(self):
 
148
        # An exception should be raised, and the target should be left in place
 
149
        self.create_file('target', 'data in target\n')
 
150
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
151
                          'missingsource', 'target')
 
152
        self.assertPathExists('target')
 
153
        self.check_file_contents('target', 'data in target\n')
 
154
 
 
155
    def test_fancy_rename_fails_if_source_and_target_missing(self):
 
156
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
157
                          'missingsource', 'missingtarget')
 
158
 
81
159
    def test_rename(self):
82
160
        # Rename should be semi-atomic on all platforms
83
 
        open('a', 'wb').write('something in a\n')
 
161
        self.create_file('a', 'something in a\n')
84
162
        osutils.rename('a', 'b')
85
 
        self.failIfExists('a')
86
 
        self.failUnlessExists('b')
 
163
        self.assertPathDoesNotExist('a')
 
164
        self.assertPathExists('b')
87
165
        self.check_file_contents('b', 'something in a\n')
88
166
 
89
 
        open('a', 'wb').write('new something in a\n')
 
167
        self.create_file('a', 'new something in a\n')
90
168
        osutils.rename('b', 'a')
91
169
 
92
170
        self.check_file_contents('a', 'something in a\n')
103
181
        shape = sorted(os.listdir('.'))
104
182
        self.assertEquals(['A', 'B'], shape)
105
183
 
 
184
    def test_rename_exception(self):
 
185
        try:
 
186
            osutils.rename('nonexistent_path', 'different_nonexistent_path')
 
187
        except OSError, e:
 
188
            self.assertEqual(e.old_filename, 'nonexistent_path')
 
189
            self.assertEqual(e.new_filename, 'different_nonexistent_path')
 
190
            self.assertTrue('nonexistent_path' in e.strerror)
 
191
            self.assertTrue('different_nonexistent_path' in e.strerror)
 
192
 
 
193
 
 
194
class TestRandChars(tests.TestCase):
 
195
 
106
196
    def test_01_rand_chars_empty(self):
107
197
        result = osutils.rand_chars(0)
108
198
        self.assertEqual(result, '')
113
203
        self.assertEqual(type(result), str)
114
204
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
115
205
 
 
206
 
 
207
class TestIsInside(tests.TestCase):
 
208
 
116
209
    def test_is_inside(self):
117
210
        is_inside = osutils.is_inside
118
211
        self.assertTrue(is_inside('src', 'src/foo.c'))
123
216
        self.assertTrue(is_inside('', 'foo.c'))
124
217
 
125
218
    def test_is_inside_any(self):
126
 
        SRC_FOO_C = pathjoin('src', 'foo.c')
 
219
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
127
220
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
128
221
                         (['src'], SRC_FOO_C),
129
222
                         (['src'], 'src'),
130
223
                         ]:
131
 
            self.assert_(is_inside_any(dirs, fn))
 
224
            self.assert_(osutils.is_inside_any(dirs, fn))
132
225
        for dirs, fn in [(['src'], 'srccontrol'),
133
226
                         (['src'], 'srccontrol/foo')]:
134
 
            self.assertFalse(is_inside_any(dirs, fn))
 
227
            self.assertFalse(osutils.is_inside_any(dirs, fn))
135
228
 
136
229
    def test_is_inside_or_parent_of_any(self):
137
230
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
140
233
                         (['src/bar.c', 'bla/foo.c'], 'src'),
141
234
                         (['src'], 'src'),
142
235
                         ]:
143
 
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
144
 
            
 
236
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
237
 
145
238
        for dirs, fn in [(['src'], 'srccontrol'),
146
239
                         (['srccontrol/foo.c'], 'src'),
147
240
                         (['src'], 'srccontrol/foo')]:
148
 
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
 
241
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
 
242
 
 
243
 
 
244
class TestLstat(tests.TestCaseInTempDir):
 
245
 
 
246
    def test_lstat_matches_fstat(self):
 
247
        # On Windows, lstat and fstat don't always agree, primarily in the
 
248
        # 'st_ino' and 'st_dev' fields. So we force them to be '0' in our
 
249
        # custom implementation.
 
250
        if sys.platform == 'win32':
 
251
            # We only have special lstat/fstat if we have the extension.
 
252
            # Without it, we may end up re-reading content when we don't have
 
253
            # to, but otherwise it doesn't effect correctness.
 
254
            self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
255
        f = open('test-file.txt', 'wb')
 
256
        self.addCleanup(f.close)
 
257
        f.write('some content\n')
 
258
        f.flush()
 
259
        self.assertEqualStat(osutils.fstat(f.fileno()),
 
260
                             osutils.lstat('test-file.txt'))
 
261
 
 
262
 
 
263
class TestRmTree(tests.TestCaseInTempDir):
149
264
 
150
265
    def test_rmtree(self):
151
266
        # Check to remove tree with read-only files/dirs
162
277
 
163
278
        osutils.rmtree('dir')
164
279
 
165
 
        self.failIfExists('dir/file')
166
 
        self.failIfExists('dir')
 
280
        self.assertPathDoesNotExist('dir/file')
 
281
        self.assertPathDoesNotExist('dir')
 
282
 
 
283
 
 
284
class TestDeleteAny(tests.TestCaseInTempDir):
 
285
 
 
286
    def test_delete_any_readonly(self):
 
287
        # from <https://bugs.launchpad.net/bzr/+bug/218206>
 
288
        self.build_tree(['d/', 'f'])
 
289
        osutils.make_readonly('d')
 
290
        osutils.make_readonly('f')
 
291
 
 
292
        osutils.delete_any('f')
 
293
        osutils.delete_any('d')
 
294
 
 
295
 
 
296
class TestKind(tests.TestCaseInTempDir):
167
297
 
168
298
    def test_file_kind(self):
169
299
        self.build_tree(['file', 'dir/'])
172
302
        if osutils.has_symlinks():
173
303
            os.symlink('symlink', 'symlink')
174
304
            self.assertEquals('symlink', osutils.file_kind('symlink'))
175
 
        
 
305
 
176
306
        # TODO: jam 20060529 Test a block device
177
307
        try:
178
308
            os.lstat('/dev/null')
200
330
                os.remove('socket')
201
331
 
202
332
    def test_kind_marker(self):
203
 
        self.assertEqual(osutils.kind_marker('file'), '')
204
 
        self.assertEqual(osutils.kind_marker('directory'), '/')
205
 
        self.assertEqual(osutils.kind_marker('symlink'), '@')
206
 
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
 
333
        self.assertEqual("", osutils.kind_marker("file"))
 
334
        self.assertEqual("/", osutils.kind_marker('directory'))
 
335
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
336
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
337
        self.assertEqual("+", osutils.kind_marker("tree-reference"))
 
338
        self.assertEqual("", osutils.kind_marker("fifo"))
 
339
        self.assertEqual("", osutils.kind_marker("socket"))
 
340
        self.assertEqual("", osutils.kind_marker("unknown"))
 
341
 
 
342
 
 
343
class TestUmask(tests.TestCaseInTempDir):
207
344
 
208
345
    def test_get_umask(self):
209
346
        if sys.platform == 'win32':
212
349
            return
213
350
 
214
351
        orig_umask = osutils.get_umask()
215
 
        try:
216
 
            os.umask(0222)
217
 
            self.assertEqual(0222, osutils.get_umask())
218
 
            os.umask(0022)
219
 
            self.assertEqual(0022, osutils.get_umask())
220
 
            os.umask(0002)
221
 
            self.assertEqual(0002, osutils.get_umask())
222
 
            os.umask(0027)
223
 
            self.assertEqual(0027, osutils.get_umask())
224
 
        finally:
225
 
            os.umask(orig_umask)
 
352
        self.addCleanup(os.umask, orig_umask)
 
353
        os.umask(0222)
 
354
        self.assertEqual(0222, osutils.get_umask())
 
355
        os.umask(0022)
 
356
        self.assertEqual(0022, osutils.get_umask())
 
357
        os.umask(0002)
 
358
        self.assertEqual(0002, osutils.get_umask())
 
359
        os.umask(0027)
 
360
        self.assertEqual(0027, osutils.get_umask())
 
361
 
 
362
 
 
363
class TestDateTime(tests.TestCase):
226
364
 
227
365
    def assertFormatedDelta(self, expected, seconds):
228
366
        """Assert osutils.format_delta formats as expected"""
263
401
    def test_format_date(self):
264
402
        self.assertRaises(errors.UnsupportedTimezoneFormat,
265
403
            osutils.format_date, 0, timezone='foo')
 
404
        self.assertIsInstance(osutils.format_date(0), str)
 
405
        self.assertIsInstance(osutils.format_local_date(0), unicode)
 
406
        # Testing for the actual value of the local weekday without
 
407
        # duplicating the code from format_date is difficult.
 
408
        # Instead blackbox.test_locale should check for localized
 
409
        # dates once they do occur in output strings.
 
410
 
 
411
    def test_format_date_with_offset_in_original_timezone(self):
 
412
        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
 
413
            osutils.format_date_with_offset_in_original_timezone(0))
 
414
        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
 
415
            osutils.format_date_with_offset_in_original_timezone(100000))
 
416
        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
 
417
            osutils.format_date_with_offset_in_original_timezone(100000, 7200))
 
418
 
 
419
    def test_local_time_offset(self):
 
420
        """Test that local_time_offset() returns a sane value."""
 
421
        offset = osutils.local_time_offset()
 
422
        self.assertTrue(isinstance(offset, int))
 
423
        # Test that the offset is no more than a eighteen hours in
 
424
        # either direction.
 
425
        # Time zone handling is system specific, so it is difficult to
 
426
        # do more specific tests, but a value outside of this range is
 
427
        # probably wrong.
 
428
        eighteen_hours = 18 * 3600
 
429
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
430
 
 
431
    def test_local_time_offset_with_timestamp(self):
 
432
        """Test that local_time_offset() works with a timestamp."""
 
433
        offset = osutils.local_time_offset(1000000000.1234567)
 
434
        self.assertTrue(isinstance(offset, int))
 
435
        eighteen_hours = 18 * 3600
 
436
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
437
 
 
438
 
 
439
class TestLinks(tests.TestCaseInTempDir):
266
440
 
267
441
    def test_dereference_path(self):
268
 
        self.requireFeature(SymlinkFeature)
 
442
        self.requireFeature(features.SymlinkFeature)
269
443
        cwd = osutils.realpath('.')
270
444
        os.mkdir('bar')
271
445
        bar_path = osutils.pathjoin(cwd, 'bar')
274
448
        self.assertEqual(bar_path, osutils.realpath('./bar'))
275
449
        os.symlink('bar', 'foo')
276
450
        self.assertEqual(bar_path, osutils.realpath('./foo'))
277
 
        
 
451
 
278
452
        # Does not dereference terminal symlinks
279
453
        foo_path = osutils.pathjoin(cwd, 'foo')
280
454
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
312
486
            osutils.make_readonly('dangling')
313
487
            osutils.make_writable('dangling')
314
488
 
315
 
    def test_kind_marker(self):
316
 
        self.assertEqual("", osutils.kind_marker("file"))
317
 
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
318
 
        self.assertEqual("@", osutils.kind_marker("symlink"))
319
 
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
320
 
 
321
 
 
322
 
class TestSafeUnicode(TestCase):
 
489
    def test_host_os_dereferences_symlinks(self):
 
490
        osutils.host_os_dereferences_symlinks()
 
491
 
 
492
 
 
493
class TestCanonicalRelPath(tests.TestCaseInTempDir):
 
494
 
 
495
    _test_needs_features = [features.CaseInsCasePresFilenameFeature]
 
496
 
 
497
    def test_canonical_relpath_simple(self):
 
498
        f = file('MixedCaseName', 'w')
 
499
        f.close()
 
500
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
 
501
        self.assertEqual('work/MixedCaseName', actual)
 
502
 
 
503
    def test_canonical_relpath_missing_tail(self):
 
504
        os.mkdir('MixedCaseParent')
 
505
        actual = osutils.canonical_relpath(self.test_base_dir,
 
506
                                           'mixedcaseparent/nochild')
 
507
        self.assertEqual('work/MixedCaseParent/nochild', actual)
 
508
 
 
509
 
 
510
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
 
511
 
 
512
    def assertRelpath(self, expected, base, path):
 
513
        actual = osutils._cicp_canonical_relpath(base, path)
 
514
        self.assertEqual(expected, actual)
 
515
 
 
516
    def test_simple(self):
 
517
        self.build_tree(['MixedCaseName'])
 
518
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
519
        self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
 
520
 
 
521
    def test_subdir_missing_tail(self):
 
522
        self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
 
523
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
524
        self.assertRelpath('MixedCaseParent/a_child', base,
 
525
                           'MixedCaseParent/a_child')
 
526
        self.assertRelpath('MixedCaseParent/a_child', base,
 
527
                           'MixedCaseParent/A_Child')
 
528
        self.assertRelpath('MixedCaseParent/not_child', base,
 
529
                           'MixedCaseParent/not_child')
 
530
 
 
531
    def test_at_root_slash(self):
 
532
        # We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
 
533
        # check...
 
534
        if osutils.MIN_ABS_PATHLENGTH > 1:
 
535
            raise tests.TestSkipped('relpath requires %d chars'
 
536
                                    % osutils.MIN_ABS_PATHLENGTH)
 
537
        self.assertRelpath('foo', '/', '/foo')
 
538
 
 
539
    def test_at_root_drive(self):
 
540
        if sys.platform != 'win32':
 
541
            raise tests.TestNotApplicable('we can only test drive-letter relative'
 
542
                                          ' paths on Windows where we have drive'
 
543
                                          ' letters.')
 
544
        # see bug #322807
 
545
        # The specific issue is that when at the root of a drive, 'abspath'
 
546
        # returns "C:/" or just "/". However, the code assumes that abspath
 
547
        # always returns something like "C:/foo" or "/foo" (no trailing slash).
 
548
        self.assertRelpath('foo', 'C:/', 'C:/foo')
 
549
        self.assertRelpath('foo', 'X:/', 'X:/foo')
 
550
        self.assertRelpath('foo', 'X:/', 'X://foo')
 
551
 
 
552
 
 
553
class TestPumpFile(tests.TestCase):
 
554
    """Test pumpfile method."""
 
555
 
 
556
    def setUp(self):
 
557
        tests.TestCase.setUp(self)
 
558
        # create a test datablock
 
559
        self.block_size = 512
 
560
        pattern = '0123456789ABCDEF'
 
561
        self.test_data = pattern * (3 * self.block_size / len(pattern))
 
562
        self.test_data_len = len(self.test_data)
 
563
 
 
564
    def test_bracket_block_size(self):
 
565
        """Read data in blocks with the requested read size bracketing the
 
566
        block size."""
 
567
        # make sure test data is larger than max read size
 
568
        self.assertTrue(self.test_data_len > self.block_size)
 
569
 
 
570
        from_file = file_utils.FakeReadFile(self.test_data)
 
571
        to_file = StringIO()
 
572
 
 
573
        # read (max / 2) bytes and verify read size wasn't affected
 
574
        num_bytes_to_read = self.block_size / 2
 
575
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
576
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
577
        self.assertEqual(from_file.get_read_count(), 1)
 
578
 
 
579
        # read (max) bytes and verify read size wasn't affected
 
580
        num_bytes_to_read = self.block_size
 
581
        from_file.reset_read_count()
 
582
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
583
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
584
        self.assertEqual(from_file.get_read_count(), 1)
 
585
 
 
586
        # read (max + 1) bytes and verify read size was limited
 
587
        num_bytes_to_read = self.block_size + 1
 
588
        from_file.reset_read_count()
 
589
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
590
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
591
        self.assertEqual(from_file.get_read_count(), 2)
 
592
 
 
593
        # finish reading the rest of the data
 
594
        num_bytes_to_read = self.test_data_len - to_file.tell()
 
595
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
596
 
 
597
        # report error if the data wasn't equal (we only report the size due
 
598
        # to the length of the data)
 
599
        response_data = to_file.getvalue()
 
600
        if response_data != self.test_data:
 
601
            message = "Data not equal.  Expected %d bytes, received %d."
 
602
            self.fail(message % (len(response_data), self.test_data_len))
 
603
 
 
604
    def test_specified_size(self):
 
605
        """Request a transfer larger than the maximum block size and verify
 
606
        that the maximum read doesn't exceed the block_size."""
 
607
        # make sure test data is larger than max read size
 
608
        self.assertTrue(self.test_data_len > self.block_size)
 
609
 
 
610
        # retrieve data in blocks
 
611
        from_file = file_utils.FakeReadFile(self.test_data)
 
612
        to_file = StringIO()
 
613
        osutils.pumpfile(from_file, to_file, self.test_data_len,
 
614
                         self.block_size)
 
615
 
 
616
        # verify read size was equal to the maximum read size
 
617
        self.assertTrue(from_file.get_max_read_size() > 0)
 
618
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
619
        self.assertEqual(from_file.get_read_count(), 3)
 
620
 
 
621
        # report error if the data wasn't equal (we only report the size due
 
622
        # to the length of the data)
 
623
        response_data = to_file.getvalue()
 
624
        if response_data != self.test_data:
 
625
            message = "Data not equal.  Expected %d bytes, received %d."
 
626
            self.fail(message % (len(response_data), self.test_data_len))
 
627
 
 
628
    def test_to_eof(self):
 
629
        """Read to end-of-file and verify that the reads are not larger than
 
630
        the maximum read size."""
 
631
        # make sure test data is larger than max read size
 
632
        self.assertTrue(self.test_data_len > self.block_size)
 
633
 
 
634
        # retrieve data to EOF
 
635
        from_file = file_utils.FakeReadFile(self.test_data)
 
636
        to_file = StringIO()
 
637
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
 
638
 
 
639
        # verify read size was equal to the maximum read size
 
640
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
641
        self.assertEqual(from_file.get_read_count(), 4)
 
642
 
 
643
        # report error if the data wasn't equal (we only report the size due
 
644
        # to the length of the data)
 
645
        response_data = to_file.getvalue()
 
646
        if response_data != self.test_data:
 
647
            message = "Data not equal.  Expected %d bytes, received %d."
 
648
            self.fail(message % (len(response_data), self.test_data_len))
 
649
 
 
650
    def test_defaults(self):
 
651
        """Verifies that the default arguments will read to EOF -- this
 
652
        test verifies that any existing usages of pumpfile will not be broken
 
653
        with this new version."""
 
654
        # retrieve data using default (old) pumpfile method
 
655
        from_file = file_utils.FakeReadFile(self.test_data)
 
656
        to_file = StringIO()
 
657
        osutils.pumpfile(from_file, to_file)
 
658
 
 
659
        # report error if the data wasn't equal (we only report the size due
 
660
        # to the length of the data)
 
661
        response_data = to_file.getvalue()
 
662
        if response_data != self.test_data:
 
663
            message = "Data not equal.  Expected %d bytes, received %d."
 
664
            self.fail(message % (len(response_data), self.test_data_len))
 
665
 
 
666
    def test_report_activity(self):
 
667
        activity = []
 
668
        def log_activity(length, direction):
 
669
            activity.append((length, direction))
 
670
        from_file = StringIO(self.test_data)
 
671
        to_file = StringIO()
 
672
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
673
                         report_activity=log_activity, direction='read')
 
674
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
 
675
                          (36, 'read')], activity)
 
676
 
 
677
        from_file = StringIO(self.test_data)
 
678
        to_file = StringIO()
 
679
        del activity[:]
 
680
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
681
                         report_activity=log_activity, direction='write')
 
682
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
 
683
                          (36, 'write')], activity)
 
684
 
 
685
        # And with a limited amount of data
 
686
        from_file = StringIO(self.test_data)
 
687
        to_file = StringIO()
 
688
        del activity[:]
 
689
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
 
690
                         report_activity=log_activity, direction='read')
 
691
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
 
692
 
 
693
 
 
694
 
 
695
class TestPumpStringFile(tests.TestCase):
 
696
 
 
697
    def test_empty(self):
 
698
        output = StringIO()
 
699
        osutils.pump_string_file("", output)
 
700
        self.assertEqual("", output.getvalue())
 
701
 
 
702
    def test_more_than_segment_size(self):
 
703
        output = StringIO()
 
704
        osutils.pump_string_file("123456789", output, 2)
 
705
        self.assertEqual("123456789", output.getvalue())
 
706
 
 
707
    def test_segment_size(self):
 
708
        output = StringIO()
 
709
        osutils.pump_string_file("12", output, 2)
 
710
        self.assertEqual("12", output.getvalue())
 
711
 
 
712
    def test_segment_size_multiple(self):
 
713
        output = StringIO()
 
714
        osutils.pump_string_file("1234", output, 2)
 
715
        self.assertEqual("1234", output.getvalue())
 
716
 
 
717
 
 
718
class TestRelpath(tests.TestCase):
 
719
 
 
720
    def test_simple_relpath(self):
 
721
        cwd = osutils.getcwd()
 
722
        subdir = cwd + '/subdir'
 
723
        self.assertEqual('subdir', osutils.relpath(cwd, subdir))
 
724
 
 
725
    def test_deep_relpath(self):
 
726
        cwd = osutils.getcwd()
 
727
        subdir = cwd + '/sub/subsubdir'
 
728
        self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
 
729
 
 
730
    def test_not_relative(self):
 
731
        self.assertRaises(errors.PathNotChild,
 
732
                          osutils.relpath, 'C:/path', 'H:/path')
 
733
        self.assertRaises(errors.PathNotChild,
 
734
                          osutils.relpath, 'C:/', 'H:/path')
 
735
 
 
736
 
 
737
class TestSafeUnicode(tests.TestCase):
323
738
 
324
739
    def test_from_ascii_string(self):
325
740
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
334
749
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
335
750
 
336
751
    def test_bad_utf8_string(self):
337
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
752
        self.assertRaises(errors.BzrBadParameterNotUnicode,
338
753
                          osutils.safe_unicode,
339
754
                          '\xbb\xbb')
340
755
 
341
756
 
342
 
class TestSafeUtf8(TestCase):
 
757
class TestSafeUtf8(tests.TestCase):
343
758
 
344
759
    def test_from_ascii_string(self):
345
760
        f = 'foobar'
355
770
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
356
771
 
357
772
    def test_bad_utf8_string(self):
358
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
773
        self.assertRaises(errors.BzrBadParameterNotUnicode,
359
774
                          osutils.safe_utf8, '\xbb\xbb')
360
775
 
361
776
 
362
 
class TestSafeRevisionId(TestCase):
 
777
class TestSafeRevisionId(tests.TestCase):
363
778
 
364
779
    def test_from_ascii_string(self):
365
780
        # this shouldn't give a warning because it's getting an ascii string
387
802
        self.assertEqual(None, osutils.safe_revision_id(None))
388
803
 
389
804
 
390
 
class TestSafeFileId(TestCase):
 
805
class TestSafeFileId(tests.TestCase):
391
806
 
392
807
    def test_from_ascii_string(self):
393
808
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
413
828
        self.assertEqual(None, osutils.safe_file_id(None))
414
829
 
415
830
 
416
 
class TestWin32Funcs(TestCase):
417
 
    """Test that the _win32 versions of os utilities return appropriate paths."""
 
831
class TestPosixFuncs(tests.TestCase):
 
832
    """Test that the posix version of normpath returns an appropriate path
 
833
       when used with 2 leading slashes."""
 
834
 
 
835
    def test_normpath(self):
 
836
        self.assertEqual('/etc/shadow', osutils._posix_normpath('/etc/shadow'))
 
837
        self.assertEqual('/etc/shadow', osutils._posix_normpath('//etc/shadow'))
 
838
        self.assertEqual('/etc/shadow', osutils._posix_normpath('///etc/shadow'))
 
839
 
 
840
 
 
841
class TestWin32Funcs(tests.TestCase):
 
842
    """Test that _win32 versions of os utilities return appropriate paths."""
418
843
 
419
844
    def test_abspath(self):
420
845
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
427
852
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
428
853
 
429
854
    def test_pathjoin(self):
430
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
431
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
432
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
433
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
434
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
435
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
855
        self.assertEqual('path/to/foo',
 
856
                         osutils._win32_pathjoin('path', 'to', 'foo'))
 
857
        self.assertEqual('C:/foo',
 
858
                         osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
859
        self.assertEqual('C:/foo',
 
860
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
 
861
        self.assertEqual('path/to/foo',
 
862
                         osutils._win32_pathjoin('path/to/', 'foo'))
 
863
        self.assertEqual('/foo',
 
864
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
865
        self.assertEqual('/foo',
 
866
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
436
867
 
437
868
    def test_normpath(self):
438
 
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
439
 
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
 
869
        self.assertEqual('path/to/foo',
 
870
                         osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
871
        self.assertEqual('path/to/foo',
 
872
                         osutils._win32_normpath('path//from/../to/./foo'))
440
873
 
441
874
    def test_getcwd(self):
442
875
        cwd = osutils._win32_getcwd()
463
896
        self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
464
897
        # relative path
465
898
        cwd = osutils.getcwd().rstrip('/')
466
 
        drive = osutils._nt_splitdrive(cwd)[0]
 
899
        drive = osutils.ntpath.splitdrive(cwd)[0]
467
900
        self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
468
901
        self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
469
902
        # unicode path
471
904
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
472
905
 
473
906
 
474
 
class TestWin32FuncsDirs(TestCaseInTempDir):
 
907
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
475
908
    """Test win32 functions that create files."""
476
 
    
 
909
 
477
910
    def test_getcwd(self):
478
 
        if win32utils.winver == 'Windows 98':
479
 
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
480
 
        # Make sure getcwd can handle unicode filenames
481
 
        try:
482
 
            os.mkdir(u'mu-\xb5')
483
 
        except UnicodeError:
484
 
            raise TestSkipped("Unable to create Unicode filename")
485
 
 
 
911
        self.requireFeature(features.UnicodeFilenameFeature)
 
912
        os.mkdir(u'mu-\xb5')
486
913
        os.chdir(u'mu-\xb5')
487
914
        # TODO: jam 20060427 This will probably fail on Mac OSX because
488
915
        #       it will change the normalization of B\xe5gfors
493
920
    def test_minimum_path_selection(self):
494
921
        self.assertEqual(set(),
495
922
            osutils.minimum_path_selection([]))
 
923
        self.assertEqual(set(['a']),
 
924
            osutils.minimum_path_selection(['a']))
496
925
        self.assertEqual(set(['a', 'b']),
497
926
            osutils.minimum_path_selection(['a', 'b']))
498
927
        self.assertEqual(set(['a/', 'b']),
499
928
            osutils.minimum_path_selection(['a/', 'b']))
500
929
        self.assertEqual(set(['a/', 'b']),
501
930
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
 
931
        self.assertEqual(set(['a-b', 'a', 'a0b']),
 
932
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
502
933
 
503
934
    def test_mkdtemp(self):
504
935
        tmpdir = osutils._win32_mkdtemp(dir='.')
513
944
        b.close()
514
945
 
515
946
        osutils._win32_rename('b', 'a')
516
 
        self.failUnlessExists('a')
517
 
        self.failIfExists('b')
 
947
        self.assertPathExists('a')
 
948
        self.assertPathDoesNotExist('b')
518
949
        self.assertFileEqual('baz\n', 'a')
519
950
 
520
951
    def test_rename_missing_file(self):
560
991
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
561
992
 
562
993
 
563
 
class TestMacFuncsDirs(TestCaseInTempDir):
 
994
class TestParentDirectories(tests.TestCaseInTempDir):
 
995
    """Test osutils.parent_directories()"""
 
996
 
 
997
    def test_parent_directories(self):
 
998
        self.assertEqual([], osutils.parent_directories('a'))
 
999
        self.assertEqual(['a'], osutils.parent_directories('a/b'))
 
1000
        self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
 
1001
 
 
1002
 
 
1003
class TestMacFuncsDirs(tests.TestCaseInTempDir):
564
1004
    """Test mac special functions that require directories."""
565
1005
 
566
1006
    def test_getcwd(self):
567
 
        # On Mac, this will actually create Ba\u030agfors
568
 
        # but chdir will still work, because it accepts both paths
569
 
        try:
570
 
            os.mkdir(u'B\xe5gfors')
571
 
        except UnicodeError:
572
 
            raise TestSkipped("Unable to create Unicode filename")
573
 
 
 
1007
        self.requireFeature(features.UnicodeFilenameFeature)
 
1008
        os.mkdir(u'B\xe5gfors')
574
1009
        os.chdir(u'B\xe5gfors')
575
1010
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
576
1011
 
577
1012
    def test_getcwd_nonnorm(self):
 
1013
        self.requireFeature(features.UnicodeFilenameFeature)
578
1014
        # Test that _mac_getcwd() will normalize this path
579
 
        try:
580
 
            os.mkdir(u'Ba\u030agfors')
581
 
        except UnicodeError:
582
 
            raise TestSkipped("Unable to create Unicode filename")
583
 
 
 
1015
        os.mkdir(u'Ba\u030agfors')
584
1016
        os.chdir(u'Ba\u030agfors')
585
1017
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
586
1018
 
587
1019
 
588
 
class TestSplitLines(TestCase):
 
1020
class TestChunksToLines(tests.TestCase):
 
1021
 
 
1022
    def test_smoketest(self):
 
1023
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
1024
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
 
1025
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
1026
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
 
1027
 
 
1028
    def test_osutils_binding(self):
 
1029
        from bzrlib.tests import test__chunks_to_lines
 
1030
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
 
1031
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
 
1032
        else:
 
1033
            from bzrlib._chunks_to_lines_py import chunks_to_lines
 
1034
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
 
1035
 
 
1036
 
 
1037
class TestSplitLines(tests.TestCase):
589
1038
 
590
1039
    def test_split_unicode(self):
591
1040
        self.assertEqual([u'foo\n', u'bar\xae'],
598
1047
                         osutils.split_lines('foo\rbar\n'))
599
1048
 
600
1049
 
601
 
class TestWalkDirs(TestCaseInTempDir):
 
1050
class TestWalkDirs(tests.TestCaseInTempDir):
 
1051
 
 
1052
    def assertExpectedBlocks(self, expected, result):
 
1053
        self.assertEqual(expected,
 
1054
                         [(dirinfo, [line[0:3] for line in block])
 
1055
                          for dirinfo, block in result])
602
1056
 
603
1057
    def test_walkdirs(self):
604
1058
        tree = [
637
1091
            result.append((dirdetail, dirblock))
638
1092
 
639
1093
        self.assertTrue(found_bzrdir)
640
 
        self.assertEqual(expected_dirblocks,
641
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1094
        self.assertExpectedBlocks(expected_dirblocks, result)
642
1095
        # you can search a subdir only, with a supplied prefix.
643
1096
        result = []
644
1097
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
645
1098
            result.append(dirblock)
646
 
        self.assertEqual(expected_dirblocks[1:],
647
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1099
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
1100
 
 
1101
    def test_walkdirs_os_error(self):
 
1102
        # <https://bugs.launchpad.net/bzr/+bug/338653>
 
1103
        # Pyrex readdir didn't raise useful messages if it had an error
 
1104
        # reading the directory
 
1105
        if sys.platform == 'win32':
 
1106
            raise tests.TestNotApplicable(
 
1107
                "readdir IOError not tested on win32")
 
1108
        self.requireFeature(features.not_running_as_root)
 
1109
        os.mkdir("test-unreadable")
 
1110
        os.chmod("test-unreadable", 0000)
 
1111
        # must chmod it back so that it can be removed
 
1112
        self.addCleanup(os.chmod, "test-unreadable", 0700)
 
1113
        # The error is not raised until the generator is actually evaluated.
 
1114
        # (It would be ok if it happened earlier but at the moment it
 
1115
        # doesn't.)
 
1116
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
 
1117
        self.assertEquals('./test-unreadable', e.filename)
 
1118
        self.assertEquals(errno.EACCES, e.errno)
 
1119
        # Ensure the message contains the file name
 
1120
        self.assertContainsRe(str(e), "\./test-unreadable")
 
1121
 
 
1122
 
 
1123
    def test_walkdirs_encoding_error(self):
 
1124
        # <https://bugs.launchpad.net/bzr/+bug/488519>
 
1125
        # walkdirs didn't raise a useful message when the filenames
 
1126
        # are not using the filesystem's encoding
 
1127
 
 
1128
        # require a bytestring based filesystem
 
1129
        self.requireFeature(features.ByteStringNamedFilesystem)
 
1130
 
 
1131
        tree = [
 
1132
            '.bzr',
 
1133
            '0file',
 
1134
            '1dir/',
 
1135
            '1dir/0file',
 
1136
            '1dir/1dir/',
 
1137
            '1file'
 
1138
            ]
 
1139
 
 
1140
        self.build_tree(tree)
 
1141
 
 
1142
        # rename the 1file to a latin-1 filename
 
1143
        os.rename("./1file", "\xe8file")
 
1144
        if "\xe8file" not in os.listdir("."):
 
1145
            self.skip("Lack filesystem that preserves arbitrary bytes")
 
1146
 
 
1147
        self._save_platform_info()
 
1148
        win32utils.winver = None # Avoid the win32 detection code
 
1149
        osutils._fs_enc = 'UTF-8'
 
1150
 
 
1151
        # this should raise on error
 
1152
        def attempt():
 
1153
            for dirdetail, dirblock in osutils.walkdirs('.'):
 
1154
                pass
 
1155
 
 
1156
        self.assertRaises(errors.BadFilenameEncoding, attempt)
648
1157
 
649
1158
    def test__walkdirs_utf8(self):
650
1159
        tree = [
683
1192
            result.append((dirdetail, dirblock))
684
1193
 
685
1194
        self.assertTrue(found_bzrdir)
686
 
        self.assertEqual(expected_dirblocks,
687
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1195
        self.assertExpectedBlocks(expected_dirblocks, result)
 
1196
 
688
1197
        # you can search a subdir only, with a supplied prefix.
689
1198
        result = []
690
1199
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
691
1200
            result.append(dirblock)
692
 
        self.assertEqual(expected_dirblocks[1:],
693
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1201
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
694
1202
 
695
1203
    def _filter_out_stat(self, result):
696
1204
        """Filter out the stat value from the walkdirs result"""
701
1209
                new_dirblock.append((info[0], info[1], info[2], info[4]))
702
1210
            dirblock[:] = new_dirblock
703
1211
 
 
1212
    def _save_platform_info(self):
 
1213
        self.overrideAttr(win32utils, 'winver')
 
1214
        self.overrideAttr(osutils, '_fs_enc')
 
1215
        self.overrideAttr(osutils, '_selected_dir_reader')
 
1216
 
 
1217
    def assertDirReaderIs(self, expected):
 
1218
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
 
1219
        # Force it to redetect
 
1220
        osutils._selected_dir_reader = None
 
1221
        # Nothing to list, but should still trigger the selection logic
 
1222
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
 
1223
        self.assertIsInstance(osutils._selected_dir_reader, expected)
 
1224
 
 
1225
    def test_force_walkdirs_utf8_fs_utf8(self):
 
1226
        self.requireFeature(UTF8DirReaderFeature)
 
1227
        self._save_platform_info()
 
1228
        win32utils.winver = None # Avoid the win32 detection code
 
1229
        osutils._fs_enc = 'utf-8'
 
1230
        self.assertDirReaderIs(
 
1231
            UTF8DirReaderFeature.module.UTF8DirReader)
 
1232
 
 
1233
    def test_force_walkdirs_utf8_fs_ascii(self):
 
1234
        self.requireFeature(UTF8DirReaderFeature)
 
1235
        self._save_platform_info()
 
1236
        win32utils.winver = None # Avoid the win32 detection code
 
1237
        osutils._fs_enc = 'ascii'
 
1238
        self.assertDirReaderIs(
 
1239
            UTF8DirReaderFeature.module.UTF8DirReader)
 
1240
 
 
1241
    def test_force_walkdirs_utf8_fs_latin1(self):
 
1242
        self._save_platform_info()
 
1243
        win32utils.winver = None # Avoid the win32 detection code
 
1244
        osutils._fs_enc = 'iso-8859-1'
 
1245
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1246
 
 
1247
    def test_force_walkdirs_utf8_nt(self):
 
1248
        # Disabled because the thunk of the whole walkdirs api is disabled.
 
1249
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1250
        self._save_platform_info()
 
1251
        win32utils.winver = 'Windows NT'
 
1252
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1253
        self.assertDirReaderIs(Win32ReadDir)
 
1254
 
 
1255
    def test_force_walkdirs_utf8_98(self):
 
1256
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1257
        self._save_platform_info()
 
1258
        win32utils.winver = 'Windows 98'
 
1259
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1260
 
704
1261
    def test_unicode_walkdirs(self):
705
1262
        """Walkdirs should always return unicode paths."""
 
1263
        self.requireFeature(features.UnicodeFilenameFeature)
706
1264
        name0 = u'0file-\xb6'
707
1265
        name1 = u'1dir-\u062c\u0648'
708
1266
        name2 = u'2file-\u0633'
713
1271
            name1 + '/' + name1 + '/',
714
1272
            name2,
715
1273
            ]
716
 
        try:
717
 
            self.build_tree(tree)
718
 
        except UnicodeError:
719
 
            raise TestSkipped('Could not represent Unicode chars'
720
 
                              ' in current encoding.')
 
1274
        self.build_tree(tree)
721
1275
        expected_dirblocks = [
722
1276
                ((u'', u'.'),
723
1277
                 [(name0, name0, 'file', './' + name0),
749
1303
 
750
1304
        The abspath portion might be in unicode or utf-8
751
1305
        """
 
1306
        self.requireFeature(features.UnicodeFilenameFeature)
752
1307
        name0 = u'0file-\xb6'
753
1308
        name1 = u'1dir-\u062c\u0648'
754
1309
        name2 = u'2file-\u0633'
759
1314
            name1 + '/' + name1 + '/',
760
1315
            name2,
761
1316
            ]
762
 
        try:
763
 
            self.build_tree(tree)
764
 
        except UnicodeError:
765
 
            raise TestSkipped('Could not represent Unicode chars'
766
 
                              ' in current encoding.')
 
1317
        self.build_tree(tree)
767
1318
        name0 = name0.encode('utf8')
768
1319
        name1 = name1.encode('utf8')
769
1320
        name2 = name2.encode('utf8')
808
1359
            result.append((dirdetail, new_dirblock))
809
1360
        self.assertEqual(expected_dirblocks, result)
810
1361
 
811
 
    def test_unicode__walkdirs_unicode_to_utf8(self):
812
 
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
 
1362
    def test__walkdirs_utf8_with_unicode_fs(self):
 
1363
        """UnicodeDirReader should be a safe fallback everywhere
813
1364
 
814
1365
        The abspath portion should be in unicode
815
1366
        """
816
 
        name0u = u'0file-\xb6'
817
 
        name1u = u'1dir-\u062c\u0648'
818
 
        name2u = u'2file-\u0633'
819
 
        tree = [
820
 
            name0u,
821
 
            name1u + '/',
822
 
            name1u + '/' + name0u,
823
 
            name1u + '/' + name1u + '/',
824
 
            name2u,
825
 
            ]
 
1367
        self.requireFeature(features.UnicodeFilenameFeature)
 
1368
        # Use the unicode reader. TODO: split into driver-and-driven unit
 
1369
        # tests.
 
1370
        self._save_platform_info()
 
1371
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
 
1372
        name0u = u'0file-\xb6'
 
1373
        name1u = u'1dir-\u062c\u0648'
 
1374
        name2u = u'2file-\u0633'
 
1375
        tree = [
 
1376
            name0u,
 
1377
            name1u + '/',
 
1378
            name1u + '/' + name0u,
 
1379
            name1u + '/' + name1u + '/',
 
1380
            name2u,
 
1381
            ]
 
1382
        self.build_tree(tree)
 
1383
        name0 = name0u.encode('utf8')
 
1384
        name1 = name1u.encode('utf8')
 
1385
        name2 = name2u.encode('utf8')
 
1386
 
 
1387
        # All of the abspaths should be in unicode, all of the relative paths
 
1388
        # should be in utf8
 
1389
        expected_dirblocks = [
 
1390
                (('', '.'),
 
1391
                 [(name0, name0, 'file', './' + name0u),
 
1392
                  (name1, name1, 'directory', './' + name1u),
 
1393
                  (name2, name2, 'file', './' + name2u),
 
1394
                 ]
 
1395
                ),
 
1396
                ((name1, './' + name1u),
 
1397
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1398
                                                        + '/' + name0u),
 
1399
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1400
                                                            + '/' + name1u),
 
1401
                 ]
 
1402
                ),
 
1403
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1404
                 [
 
1405
                 ]
 
1406
                ),
 
1407
            ]
 
1408
        result = list(osutils._walkdirs_utf8('.'))
 
1409
        self._filter_out_stat(result)
 
1410
        self.assertEqual(expected_dirblocks, result)
 
1411
 
 
1412
    def test__walkdirs_utf8_win32readdir(self):
 
1413
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1414
        self.requireFeature(features.UnicodeFilenameFeature)
 
1415
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1416
        self._save_platform_info()
 
1417
        osutils._selected_dir_reader = Win32ReadDir()
 
1418
        name0u = u'0file-\xb6'
 
1419
        name1u = u'1dir-\u062c\u0648'
 
1420
        name2u = u'2file-\u0633'
 
1421
        tree = [
 
1422
            name0u,
 
1423
            name1u + '/',
 
1424
            name1u + '/' + name0u,
 
1425
            name1u + '/' + name1u + '/',
 
1426
            name2u,
 
1427
            ]
 
1428
        self.build_tree(tree)
 
1429
        name0 = name0u.encode('utf8')
 
1430
        name1 = name1u.encode('utf8')
 
1431
        name2 = name2u.encode('utf8')
 
1432
 
 
1433
        # All of the abspaths should be in unicode, all of the relative paths
 
1434
        # should be in utf8
 
1435
        expected_dirblocks = [
 
1436
                (('', '.'),
 
1437
                 [(name0, name0, 'file', './' + name0u),
 
1438
                  (name1, name1, 'directory', './' + name1u),
 
1439
                  (name2, name2, 'file', './' + name2u),
 
1440
                 ]
 
1441
                ),
 
1442
                ((name1, './' + name1u),
 
1443
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1444
                                                        + '/' + name0u),
 
1445
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1446
                                                            + '/' + name1u),
 
1447
                 ]
 
1448
                ),
 
1449
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1450
                 [
 
1451
                 ]
 
1452
                ),
 
1453
            ]
 
1454
        result = list(osutils._walkdirs_utf8(u'.'))
 
1455
        self._filter_out_stat(result)
 
1456
        self.assertEqual(expected_dirblocks, result)
 
1457
 
 
1458
    def assertStatIsCorrect(self, path, win32stat):
 
1459
        os_stat = os.stat(path)
 
1460
        self.assertEqual(os_stat.st_size, win32stat.st_size)
 
1461
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
 
1462
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
 
1463
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
 
1464
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
 
1465
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
 
1466
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
 
1467
 
 
1468
    def test__walkdirs_utf_win32_find_file_stat_file(self):
 
1469
        """make sure our Stat values are valid"""
 
1470
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1471
        self.requireFeature(features.UnicodeFilenameFeature)
 
1472
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1473
        name0u = u'0file-\xb6'
 
1474
        name0 = name0u.encode('utf8')
 
1475
        self.build_tree([name0u])
 
1476
        # I hate to sleep() here, but I'm trying to make the ctime different
 
1477
        # from the mtime
 
1478
        time.sleep(2)
 
1479
        f = open(name0u, 'ab')
826
1480
        try:
827
 
            self.build_tree(tree)
828
 
        except UnicodeError:
829
 
            raise TestSkipped('Could not represent Unicode chars'
830
 
                              ' in current encoding.')
 
1481
            f.write('just a small update')
 
1482
        finally:
 
1483
            f.close()
 
1484
 
 
1485
        result = Win32ReadDir().read_dir('', u'.')
 
1486
        entry = result[0]
 
1487
        self.assertEqual((name0, name0, 'file'), entry[:3])
 
1488
        self.assertEqual(u'./' + name0u, entry[4])
 
1489
        self.assertStatIsCorrect(entry[4], entry[3])
 
1490
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
 
1491
 
 
1492
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
 
1493
        """make sure our Stat values are valid"""
 
1494
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1495
        self.requireFeature(features.UnicodeFilenameFeature)
 
1496
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1497
        name0u = u'0dir-\u062c\u0648'
831
1498
        name0 = name0u.encode('utf8')
832
 
        name1 = name1u.encode('utf8')
833
 
        name2 = name2u.encode('utf8')
 
1499
        self.build_tree([name0u + '/'])
834
1500
 
835
 
        # All of the abspaths should be in unicode, all of the relative paths
836
 
        # should be in utf8
837
 
        expected_dirblocks = [
838
 
                (('', '.'),
839
 
                 [(name0, name0, 'file', './' + name0u),
840
 
                  (name1, name1, 'directory', './' + name1u),
841
 
                  (name2, name2, 'file', './' + name2u),
842
 
                 ]
843
 
                ),
844
 
                ((name1, './' + name1u),
845
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
846
 
                                                        + '/' + name0u),
847
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
848
 
                                                            + '/' + name1u),
849
 
                 ]
850
 
                ),
851
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
852
 
                 [
853
 
                 ]
854
 
                ),
855
 
            ]
856
 
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
857
 
        self._filter_out_stat(result)
858
 
        self.assertEqual(expected_dirblocks, result)
 
1501
        result = Win32ReadDir().read_dir('', u'.')
 
1502
        entry = result[0]
 
1503
        self.assertEqual((name0, name0, 'directory'), entry[:3])
 
1504
        self.assertEqual(u'./' + name0u, entry[4])
 
1505
        self.assertStatIsCorrect(entry[4], entry[3])
859
1506
 
860
1507
    def assertPathCompare(self, path_less, path_greater):
861
1508
        """check that path_less and path_greater compare correctly."""
935
1582
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
936
1583
 
937
1584
 
938
 
class TestCopyTree(TestCaseInTempDir):
939
 
    
 
1585
class TestCopyTree(tests.TestCaseInTempDir):
 
1586
 
940
1587
    def test_copy_basic_tree(self):
941
1588
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
942
1589
        osutils.copy_tree('source', 'target')
951
1598
        self.assertEqual(['c'], os.listdir('target/b'))
952
1599
 
953
1600
    def test_copy_tree_symlinks(self):
954
 
        self.requireFeature(SymlinkFeature)
 
1601
        self.requireFeature(features.SymlinkFeature)
955
1602
        self.build_tree(['source/'])
956
1603
        os.symlink('a/generic/path', 'source/lnk')
957
1604
        osutils.copy_tree('source', 'target')
982
1629
                          ('d', 'source/b', 'target/b'),
983
1630
                          ('f', 'source/b/c', 'target/b/c'),
984
1631
                         ], processed_files)
985
 
        self.failIfExists('target')
 
1632
        self.assertPathDoesNotExist('target')
986
1633
        if osutils.has_symlinks():
987
1634
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
988
1635
 
989
1636
 
990
 
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
991
 
# [bialix] 2006/12/26
992
 
 
993
 
 
994
 
class TestSetUnsetEnv(TestCase):
 
1637
class TestSetUnsetEnv(tests.TestCase):
995
1638
    """Test updating the environment"""
996
1639
 
997
1640
    def setUp(self):
1003
1646
        def cleanup():
1004
1647
            if 'BZR_TEST_ENV_VAR' in os.environ:
1005
1648
                del os.environ['BZR_TEST_ENV_VAR']
1006
 
 
1007
1649
        self.addCleanup(cleanup)
1008
1650
 
1009
1651
    def test_set(self):
1021
1663
 
1022
1664
    def test_unicode(self):
1023
1665
        """Environment can only contain plain strings
1024
 
        
 
1666
 
1025
1667
        So Unicode strings must be encoded.
1026
1668
        """
1027
 
        uni_val, env_val = probe_unicode_in_user_encoding()
 
1669
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
1028
1670
        if uni_val is None:
1029
 
            raise TestSkipped('Cannot find a unicode character that works in'
1030
 
                              ' encoding %s' % (bzrlib.user_encoding,))
 
1671
            raise tests.TestSkipped(
 
1672
                'Cannot find a unicode character that works in encoding %s'
 
1673
                % (osutils.get_user_encoding(),))
1031
1674
 
1032
1675
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1033
1676
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1038
1681
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1039
1682
        self.assertEqual('foo', old)
1040
1683
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1041
 
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1042
 
 
1043
 
 
1044
 
class TestLocalTimeOffset(TestCase):
1045
 
 
1046
 
    def test_local_time_offset(self):
1047
 
        """Test that local_time_offset() returns a sane value."""
1048
 
        offset = osutils.local_time_offset()
1049
 
        self.assertTrue(isinstance(offset, int))
1050
 
        # Test that the offset is no more than a eighteen hours in
1051
 
        # either direction.
1052
 
        # Time zone handling is system specific, so it is difficult to
1053
 
        # do more specific tests, but a value outside of this range is
1054
 
        # probably wrong.
1055
 
        eighteen_hours = 18 * 3600
1056
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1057
 
 
1058
 
    def test_local_time_offset_with_timestamp(self):
1059
 
        """Test that local_time_offset() works with a timestamp."""
1060
 
        offset = osutils.local_time_offset(1000000000.1234567)
1061
 
        self.assertTrue(isinstance(offset, int))
1062
 
        eighteen_hours = 18 * 3600
1063
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1064
 
 
1065
 
 
1066
 
class TestShaFileByName(TestCaseInTempDir):
1067
 
 
1068
 
    def test_sha_empty(self):
1069
 
        self.build_tree_contents([('foo', '')])
1070
 
        expected_sha = osutils.sha_string('')
1071
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1072
 
 
1073
 
    def test_sha_mixed_endings(self):
1074
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1075
 
        self.build_tree_contents([('foo', text)])
1076
 
        expected_sha = osutils.sha_string(text)
1077
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1078
 
 
1079
 
 
1080
 
_debug_text = \
1081
 
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1082
 
#
1083
 
# This program is free software; you can redistribute it and/or modify
1084
 
# it under the terms of the GNU General Public License as published by
1085
 
# the Free Software Foundation; either version 2 of the License, or
1086
 
# (at your option) any later version.
1087
 
#
1088
 
# This program is distributed in the hope that it will be useful,
1089
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1090
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
1091
 
# GNU General Public License for more details.
1092
 
#
1093
 
# You should have received a copy of the GNU General Public License
1094
 
# along with this program; if not, write to the Free Software
1095
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
1096
 
 
1097
 
 
1098
 
# NOTE: If update these, please also update the help for global-options in
1099
 
#       bzrlib/help_topics/__init__.py
1100
 
 
1101
 
debug_flags = set()
1102
 
"""Set of flags that enable different debug behaviour.
1103
 
 
1104
 
These are set with eg ``-Dlock`` on the bzr command line.
1105
 
 
1106
 
Options include:
1107
 
 
1108
 
 * auth - show authentication sections used
1109
 
 * error - show stack traces for all top level exceptions
1110
 
 * evil - capture call sites that do expensive or badly-scaling operations.
1111
 
 * fetch - trace history copying between repositories
1112
 
 * graph - trace graph traversal information
1113
 
 * hashcache - log every time a working file is read to determine its hash
1114
 
 * hooks - trace hook execution
1115
 
 * hpss - trace smart protocol requests and responses
1116
 
 * http - trace http connections, requests and responses
1117
 
 * index - trace major index operations
1118
 
 * knit - trace knit operations
1119
 
 * lock - trace when lockdir locks are taken or released
1120
 
 * merge - emit information for debugging merges
1121
 
 * pack - emit information about pack operations
1122
 
 
1123
 
"""
1124
 
'''
1125
 
 
1126
 
 
1127
 
class TestResourceLoading(TestCaseInTempDir):
 
1684
        self.assertFalse('BZR_TEST_ENV_VAR' in os.environ)
 
1685
 
 
1686
 
 
1687
class TestSizeShaFile(tests.TestCaseInTempDir):
 
1688
 
 
1689
    def test_sha_empty(self):
 
1690
        self.build_tree_contents([('foo', '')])
 
1691
        expected_sha = osutils.sha_string('')
 
1692
        f = open('foo')
 
1693
        self.addCleanup(f.close)
 
1694
        size, sha = osutils.size_sha_file(f)
 
1695
        self.assertEqual(0, size)
 
1696
        self.assertEqual(expected_sha, sha)
 
1697
 
 
1698
    def test_sha_mixed_endings(self):
 
1699
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1700
        self.build_tree_contents([('foo', text)])
 
1701
        expected_sha = osutils.sha_string(text)
 
1702
        f = open('foo', 'rb')
 
1703
        self.addCleanup(f.close)
 
1704
        size, sha = osutils.size_sha_file(f)
 
1705
        self.assertEqual(38, size)
 
1706
        self.assertEqual(expected_sha, sha)
 
1707
 
 
1708
 
 
1709
class TestShaFileByName(tests.TestCaseInTempDir):
 
1710
 
 
1711
    def test_sha_empty(self):
 
1712
        self.build_tree_contents([('foo', '')])
 
1713
        expected_sha = osutils.sha_string('')
 
1714
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1715
 
 
1716
    def test_sha_mixed_endings(self):
 
1717
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1718
        self.build_tree_contents([('foo', text)])
 
1719
        expected_sha = osutils.sha_string(text)
 
1720
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1721
 
 
1722
 
 
1723
class TestResourceLoading(tests.TestCaseInTempDir):
1128
1724
 
1129
1725
    def test_resource_string(self):
1130
1726
        # test resource in bzrlib
1131
1727
        text = osutils.resource_string('bzrlib', 'debug.py')
1132
 
        self.assertEquals(_debug_text, text)
 
1728
        self.assertContainsRe(text, "debug_flags = set()")
1133
1729
        # test resource under bzrlib
1134
1730
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1135
1731
        self.assertContainsRe(text, "class TextUIFactory")
1138
1734
            'yyy.xx')
1139
1735
        # test unknown resource
1140
1736
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
 
1737
 
 
1738
 
 
1739
class TestReCompile(tests.TestCase):
 
1740
 
 
1741
    def _deprecated_re_compile_checked(self, *args, **kwargs):
 
1742
        return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
 
1743
            osutils.re_compile_checked, *args, **kwargs)
 
1744
 
 
1745
    def test_re_compile_checked(self):
 
1746
        r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
 
1747
        self.assertTrue(r.match('aaaa'))
 
1748
        self.assertTrue(r.match('aAaA'))
 
1749
 
 
1750
    def test_re_compile_checked_error(self):
 
1751
        # like https://bugs.launchpad.net/bzr/+bug/251352
 
1752
 
 
1753
        # Due to possible test isolation error, re.compile is not lazy at
 
1754
        # this point. We re-install lazy compile.
 
1755
        lazy_regex.install_lazy_compile()
 
1756
        err = self.assertRaises(
 
1757
            errors.BzrCommandError,
 
1758
            self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
 
1759
        self.assertEqual(
 
1760
            'Invalid regular expression in test case: '
 
1761
            '"*" nothing to repeat',
 
1762
            str(err))
 
1763
 
 
1764
 
 
1765
class TestDirReader(tests.TestCaseInTempDir):
 
1766
 
 
1767
    scenarios = dir_reader_scenarios()
 
1768
 
 
1769
    # Set by load_tests
 
1770
    _dir_reader_class = None
 
1771
    _native_to_unicode = None
 
1772
 
 
1773
    def setUp(self):
 
1774
        tests.TestCaseInTempDir.setUp(self)
 
1775
        self.overrideAttr(osutils,
 
1776
                          '_selected_dir_reader', self._dir_reader_class())
 
1777
 
 
1778
    def _get_ascii_tree(self):
 
1779
        tree = [
 
1780
            '0file',
 
1781
            '1dir/',
 
1782
            '1dir/0file',
 
1783
            '1dir/1dir/',
 
1784
            '2file'
 
1785
            ]
 
1786
        expected_dirblocks = [
 
1787
                (('', '.'),
 
1788
                 [('0file', '0file', 'file'),
 
1789
                  ('1dir', '1dir', 'directory'),
 
1790
                  ('2file', '2file', 'file'),
 
1791
                 ]
 
1792
                ),
 
1793
                (('1dir', './1dir'),
 
1794
                 [('1dir/0file', '0file', 'file'),
 
1795
                  ('1dir/1dir', '1dir', 'directory'),
 
1796
                 ]
 
1797
                ),
 
1798
                (('1dir/1dir', './1dir/1dir'),
 
1799
                 [
 
1800
                 ]
 
1801
                ),
 
1802
            ]
 
1803
        return tree, expected_dirblocks
 
1804
 
 
1805
    def test_walk_cur_dir(self):
 
1806
        tree, expected_dirblocks = self._get_ascii_tree()
 
1807
        self.build_tree(tree)
 
1808
        result = list(osutils._walkdirs_utf8('.'))
 
1809
        # Filter out stat and abspath
 
1810
        self.assertEqual(expected_dirblocks,
 
1811
                         [(dirinfo, [line[0:3] for line in block])
 
1812
                          for dirinfo, block in result])
 
1813
 
 
1814
    def test_walk_sub_dir(self):
 
1815
        tree, expected_dirblocks = self._get_ascii_tree()
 
1816
        self.build_tree(tree)
 
1817
        # you can search a subdir only, with a supplied prefix.
 
1818
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
 
1819
        # Filter out stat and abspath
 
1820
        self.assertEqual(expected_dirblocks[1:],
 
1821
                         [(dirinfo, [line[0:3] for line in block])
 
1822
                          for dirinfo, block in result])
 
1823
 
 
1824
    def _get_unicode_tree(self):
 
1825
        name0u = u'0file-\xb6'
 
1826
        name1u = u'1dir-\u062c\u0648'
 
1827
        name2u = u'2file-\u0633'
 
1828
        tree = [
 
1829
            name0u,
 
1830
            name1u + '/',
 
1831
            name1u + '/' + name0u,
 
1832
            name1u + '/' + name1u + '/',
 
1833
            name2u,
 
1834
            ]
 
1835
        name0 = name0u.encode('UTF-8')
 
1836
        name1 = name1u.encode('UTF-8')
 
1837
        name2 = name2u.encode('UTF-8')
 
1838
        expected_dirblocks = [
 
1839
                (('', '.'),
 
1840
                 [(name0, name0, 'file', './' + name0u),
 
1841
                  (name1, name1, 'directory', './' + name1u),
 
1842
                  (name2, name2, 'file', './' + name2u),
 
1843
                 ]
 
1844
                ),
 
1845
                ((name1, './' + name1u),
 
1846
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1847
                                                        + '/' + name0u),
 
1848
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1849
                                                            + '/' + name1u),
 
1850
                 ]
 
1851
                ),
 
1852
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1853
                 [
 
1854
                 ]
 
1855
                ),
 
1856
            ]
 
1857
        return tree, expected_dirblocks
 
1858
 
 
1859
    def _filter_out(self, raw_dirblocks):
 
1860
        """Filter out a walkdirs_utf8 result.
 
1861
 
 
1862
        stat field is removed, all native paths are converted to unicode
 
1863
        """
 
1864
        filtered_dirblocks = []
 
1865
        for dirinfo, block in raw_dirblocks:
 
1866
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
 
1867
            details = []
 
1868
            for line in block:
 
1869
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
 
1870
            filtered_dirblocks.append((dirinfo, details))
 
1871
        return filtered_dirblocks
 
1872
 
 
1873
    def test_walk_unicode_tree(self):
 
1874
        self.requireFeature(features.UnicodeFilenameFeature)
 
1875
        tree, expected_dirblocks = self._get_unicode_tree()
 
1876
        self.build_tree(tree)
 
1877
        result = list(osutils._walkdirs_utf8('.'))
 
1878
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1879
 
 
1880
    def test_symlink(self):
 
1881
        self.requireFeature(features.SymlinkFeature)
 
1882
        self.requireFeature(features.UnicodeFilenameFeature)
 
1883
        target = u'target\N{Euro Sign}'
 
1884
        link_name = u'l\N{Euro Sign}nk'
 
1885
        os.symlink(target, link_name)
 
1886
        target_utf8 = target.encode('UTF-8')
 
1887
        link_name_utf8 = link_name.encode('UTF-8')
 
1888
        expected_dirblocks = [
 
1889
                (('', '.'),
 
1890
                 [(link_name_utf8, link_name_utf8,
 
1891
                   'symlink', './' + link_name),],
 
1892
                 )]
 
1893
        result = list(osutils._walkdirs_utf8('.'))
 
1894
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1895
 
 
1896
 
 
1897
class TestReadLink(tests.TestCaseInTempDir):
 
1898
    """Exposes os.readlink() problems and the osutils solution.
 
1899
 
 
1900
    The only guarantee offered by os.readlink(), starting with 2.6, is that a
 
1901
    unicode string will be returned if a unicode string is passed.
 
1902
 
 
1903
    But prior python versions failed to properly encode the passed unicode
 
1904
    string.
 
1905
    """
 
1906
    _test_needs_features = [features.SymlinkFeature, features.UnicodeFilenameFeature]
 
1907
 
 
1908
    def setUp(self):
 
1909
        super(tests.TestCaseInTempDir, self).setUp()
 
1910
        self.link = u'l\N{Euro Sign}ink'
 
1911
        self.target = u'targe\N{Euro Sign}t'
 
1912
        os.symlink(self.target, self.link)
 
1913
 
 
1914
    def test_os_readlink_link_encoding(self):
 
1915
        self.assertEquals(self.target,  os.readlink(self.link))
 
1916
 
 
1917
    def test_os_readlink_link_decoding(self):
 
1918
        self.assertEquals(self.target.encode(osutils._fs_enc),
 
1919
                          os.readlink(self.link.encode(osutils._fs_enc)))
 
1920
 
 
1921
 
 
1922
class TestConcurrency(tests.TestCase):
 
1923
 
 
1924
    def setUp(self):
 
1925
        super(TestConcurrency, self).setUp()
 
1926
        self.overrideAttr(osutils, '_cached_local_concurrency')
 
1927
 
 
1928
    def test_local_concurrency(self):
 
1929
        concurrency = osutils.local_concurrency()
 
1930
        self.assertIsInstance(concurrency, int)
 
1931
 
 
1932
    def test_local_concurrency_environment_variable(self):
 
1933
        self.overrideEnv('BZR_CONCURRENCY', '2')
 
1934
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
 
1935
        self.overrideEnv('BZR_CONCURRENCY', '3')
 
1936
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
 
1937
        self.overrideEnv('BZR_CONCURRENCY', 'foo')
 
1938
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
 
1939
 
 
1940
    def test_option_concurrency(self):
 
1941
        self.overrideEnv('BZR_CONCURRENCY', '1')
 
1942
        self.run_bzr('rocks --concurrency 42')
 
1943
        # Command line overrides environment variable
 
1944
        self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
 
1945
        self.assertEquals(42, osutils.local_concurrency(use_cache=False))
 
1946
 
 
1947
 
 
1948
class TestFailedToLoadExtension(tests.TestCase):
 
1949
 
 
1950
    def _try_loading(self):
 
1951
        try:
 
1952
            import bzrlib._fictional_extension_py
 
1953
        except ImportError, e:
 
1954
            osutils.failed_to_load_extension(e)
 
1955
            return True
 
1956
 
 
1957
    def setUp(self):
 
1958
        super(TestFailedToLoadExtension, self).setUp()
 
1959
        self.overrideAttr(osutils, '_extension_load_failures', [])
 
1960
 
 
1961
    def test_failure_to_load(self):
 
1962
        self._try_loading()
 
1963
        self.assertLength(1, osutils._extension_load_failures)
 
1964
        self.assertEquals(osutils._extension_load_failures[0],
 
1965
            "No module named _fictional_extension_py")
 
1966
 
 
1967
    def test_report_extension_load_failures_no_warning(self):
 
1968
        self.assertTrue(self._try_loading())
 
1969
        warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
 
1970
        # it used to give a Python warning; it no longer does
 
1971
        self.assertLength(0, warnings)
 
1972
 
 
1973
    def test_report_extension_load_failures_message(self):
 
1974
        log = StringIO()
 
1975
        trace.push_log_file(log)
 
1976
        self.assertTrue(self._try_loading())
 
1977
        osutils.report_extension_load_failures()
 
1978
        self.assertContainsRe(
 
1979
            log.getvalue(),
 
1980
            r"bzr: warning: some compiled extensions could not be loaded; "
 
1981
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
 
1982
            )
 
1983
 
 
1984
 
 
1985
class TestTerminalWidth(tests.TestCase):
 
1986
 
 
1987
    def setUp(self):
 
1988
        tests.TestCase.setUp(self)
 
1989
        self._orig_terminal_size_state = osutils._terminal_size_state
 
1990
        self._orig_first_terminal_size = osutils._first_terminal_size
 
1991
        self.addCleanup(self.restore_osutils_globals)
 
1992
        osutils._terminal_size_state = 'no_data'
 
1993
        osutils._first_terminal_size = None
 
1994
 
 
1995
    def restore_osutils_globals(self):
 
1996
        osutils._terminal_size_state = self._orig_terminal_size_state
 
1997
        osutils._first_terminal_size = self._orig_first_terminal_size
 
1998
 
 
1999
    def replace_stdout(self, new):
 
2000
        self.overrideAttr(sys, 'stdout', new)
 
2001
 
 
2002
    def replace__terminal_size(self, new):
 
2003
        self.overrideAttr(osutils, '_terminal_size', new)
 
2004
 
 
2005
    def set_fake_tty(self):
 
2006
 
 
2007
        class I_am_a_tty(object):
 
2008
            def isatty(self):
 
2009
                return True
 
2010
 
 
2011
        self.replace_stdout(I_am_a_tty())
 
2012
 
 
2013
    def test_default_values(self):
 
2014
        self.assertEqual(80, osutils.default_terminal_width)
 
2015
 
 
2016
    def test_defaults_to_BZR_COLUMNS(self):
 
2017
        # BZR_COLUMNS is set by the test framework
 
2018
        self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
 
2019
        self.overrideEnv('BZR_COLUMNS', '12')
 
2020
        self.assertEqual(12, osutils.terminal_width())
 
2021
 
 
2022
    def test_BZR_COLUMNS_0_no_limit(self):
 
2023
        self.overrideEnv('BZR_COLUMNS', '0')
 
2024
        self.assertEqual(None, osutils.terminal_width())
 
2025
 
 
2026
    def test_falls_back_to_COLUMNS(self):
 
2027
        self.overrideEnv('BZR_COLUMNS', None)
 
2028
        self.assertNotEqual('42', os.environ['COLUMNS'])
 
2029
        self.set_fake_tty()
 
2030
        self.overrideEnv('COLUMNS', '42')
 
2031
        self.assertEqual(42, osutils.terminal_width())
 
2032
 
 
2033
    def test_tty_default_without_columns(self):
 
2034
        self.overrideEnv('BZR_COLUMNS', None)
 
2035
        self.overrideEnv('COLUMNS', None)
 
2036
 
 
2037
        def terminal_size(w, h):
 
2038
            return 42, 42
 
2039
 
 
2040
        self.set_fake_tty()
 
2041
        # We need to override the osutils definition as it depends on the
 
2042
        # running environment that we can't control (PQM running without a
 
2043
        # controlling terminal is one example).
 
2044
        self.replace__terminal_size(terminal_size)
 
2045
        self.assertEqual(42, osutils.terminal_width())
 
2046
 
 
2047
    def test_non_tty_default_without_columns(self):
 
2048
        self.overrideEnv('BZR_COLUMNS', None)
 
2049
        self.overrideEnv('COLUMNS', None)
 
2050
        self.replace_stdout(None)
 
2051
        self.assertEqual(None, osutils.terminal_width())
 
2052
 
 
2053
    def test_no_TIOCGWINSZ(self):
 
2054
        self.requireFeature(term_ios_feature)
 
2055
        termios = term_ios_feature.module
 
2056
        # bug 63539 is about a termios without TIOCGWINSZ attribute
 
2057
        try:
 
2058
            orig = termios.TIOCGWINSZ
 
2059
        except AttributeError:
 
2060
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
 
2061
            pass
 
2062
        else:
 
2063
            self.overrideAttr(termios, 'TIOCGWINSZ')
 
2064
            del termios.TIOCGWINSZ
 
2065
        self.overrideEnv('BZR_COLUMNS', None)
 
2066
        self.overrideEnv('COLUMNS', None)
 
2067
        # Whatever the result is, if we don't raise an exception, it's ok.
 
2068
        osutils.terminal_width()
 
2069
 
 
2070
 
 
2071
class TestCreationOps(tests.TestCaseInTempDir):
 
2072
    _test_needs_features = [features.chown_feature]
 
2073
 
 
2074
    def setUp(self):
 
2075
        tests.TestCaseInTempDir.setUp(self)
 
2076
        self.overrideAttr(os, 'chown', self._dummy_chown)
 
2077
 
 
2078
        # params set by call to _dummy_chown
 
2079
        self.path = self.uid = self.gid = None
 
2080
 
 
2081
    def _dummy_chown(self, path, uid, gid):
 
2082
        self.path, self.uid, self.gid = path, uid, gid
 
2083
 
 
2084
    def test_copy_ownership_from_path(self):
 
2085
        """copy_ownership_from_path test with specified src."""
 
2086
        ownsrc = '/'
 
2087
        f = open('test_file', 'wt')
 
2088
        osutils.copy_ownership_from_path('test_file', ownsrc)
 
2089
 
 
2090
        s = os.stat(ownsrc)
 
2091
        self.assertEquals(self.path, 'test_file')
 
2092
        self.assertEquals(self.uid, s.st_uid)
 
2093
        self.assertEquals(self.gid, s.st_gid)
 
2094
 
 
2095
    def test_copy_ownership_nonesrc(self):
 
2096
        """copy_ownership_from_path test with src=None."""
 
2097
        f = open('test_file', 'wt')
 
2098
        # should use parent dir for permissions
 
2099
        osutils.copy_ownership_from_path('test_file')
 
2100
 
 
2101
        s = os.stat('..')
 
2102
        self.assertEquals(self.path, 'test_file')
 
2103
        self.assertEquals(self.uid, s.st_uid)
 
2104
        self.assertEquals(self.gid, s.st_gid)
 
2105
 
 
2106
 
 
2107
class TestPathFromEnviron(tests.TestCase):
 
2108
 
 
2109
    def test_is_unicode(self):
 
2110
        self.overrideEnv('BZR_TEST_PATH', './anywhere at all/')
 
2111
        path = osutils.path_from_environ('BZR_TEST_PATH')
 
2112
        self.assertIsInstance(path, unicode)
 
2113
        self.assertEqual(u'./anywhere at all/', path)
 
2114
 
 
2115
    def test_posix_path_env_ascii(self):
 
2116
        self.overrideEnv('BZR_TEST_PATH', '/tmp')
 
2117
        home = osutils._posix_path_from_environ('BZR_TEST_PATH')
 
2118
        self.assertIsInstance(home, unicode)
 
2119
        self.assertEqual(u'/tmp', home)
 
2120
 
 
2121
    def test_posix_path_env_unicode(self):
 
2122
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2123
        self.overrideEnv('BZR_TEST_PATH', '/home/\xa7test')
 
2124
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2125
        self.assertEqual(u'/home/\xa7test',
 
2126
            osutils._posix_path_from_environ('BZR_TEST_PATH'))
 
2127
        osutils._fs_enc = "iso8859-5"
 
2128
        self.assertEqual(u'/home/\u0407test',
 
2129
            osutils._posix_path_from_environ('BZR_TEST_PATH'))
 
2130
        osutils._fs_enc = "utf-8"
 
2131
        self.assertRaises(errors.BadFilenameEncoding,
 
2132
            osutils._posix_path_from_environ, 'BZR_TEST_PATH')
 
2133
 
 
2134
 
 
2135
class TestGetHomeDir(tests.TestCase):
 
2136
 
 
2137
    def test_is_unicode(self):
 
2138
        home = osutils._get_home_dir()
 
2139
        self.assertIsInstance(home, unicode)
 
2140
 
 
2141
    def test_posix_homeless(self):
 
2142
        self.overrideEnv('HOME', None)
 
2143
        home = osutils._get_home_dir()
 
2144
        self.assertIsInstance(home, unicode)
 
2145
 
 
2146
    def test_posix_home_ascii(self):
 
2147
        self.overrideEnv('HOME', '/home/test')
 
2148
        home = osutils._posix_get_home_dir()
 
2149
        self.assertIsInstance(home, unicode)
 
2150
        self.assertEqual(u'/home/test', home)
 
2151
 
 
2152
    def test_posix_home_unicode(self):
 
2153
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2154
        self.overrideEnv('HOME', '/home/\xa7test')
 
2155
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2156
        self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
 
2157
        osutils._fs_enc = "iso8859-5"
 
2158
        self.assertEqual(u'/home/\u0407test', osutils._posix_get_home_dir())
 
2159
        osutils._fs_enc = "utf-8"
 
2160
        self.assertRaises(errors.BadFilenameEncoding,
 
2161
            osutils._posix_get_home_dir)
 
2162
 
 
2163
 
 
2164
class TestGetuserUnicode(tests.TestCase):
 
2165
 
 
2166
    def test_is_unicode(self):
 
2167
        user = osutils.getuser_unicode()
 
2168
        self.assertIsInstance(user, unicode)
 
2169
 
 
2170
    def envvar_to_override(self):
 
2171
        if sys.platform == "win32":
 
2172
            # Disable use of platform calls on windows so envvar is used
 
2173
            self.overrideAttr(win32utils, 'has_ctypes', False)
 
2174
            return 'USERNAME' # only variable used on windows
 
2175
        return 'LOGNAME' # first variable checked by getpass.getuser()
 
2176
 
 
2177
    def test_ascii_user(self):
 
2178
        self.overrideEnv(self.envvar_to_override(), 'jrandom')
 
2179
        self.assertEqual(u'jrandom', osutils.getuser_unicode())
 
2180
 
 
2181
    def test_unicode_user(self):
 
2182
        ue = osutils.get_user_encoding()
 
2183
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
2184
        if uni_val is None:
 
2185
            raise tests.TestSkipped(
 
2186
                'Cannot find a unicode character that works in encoding %s'
 
2187
                % (osutils.get_user_encoding(),))
 
2188
        uni_username = u'jrandom' + uni_val
 
2189
        encoded_username = uni_username.encode(ue)
 
2190
        self.overrideEnv(self.envvar_to_override(), encoded_username)
 
2191
        self.assertEqual(uni_username, osutils.getuser_unicode())
 
2192
 
 
2193
 
 
2194
class TestBackupNames(tests.TestCase):
 
2195
 
 
2196
    def setUp(self):
 
2197
        super(TestBackupNames, self).setUp()
 
2198
        self.backups = []
 
2199
 
 
2200
    def backup_exists(self, name):
 
2201
        return name in self.backups
 
2202
 
 
2203
    def available_backup_name(self, name):
 
2204
        backup_name = osutils.available_backup_name(name, self.backup_exists)
 
2205
        self.backups.append(backup_name)
 
2206
        return backup_name
 
2207
 
 
2208
    def assertBackupName(self, expected, name):
 
2209
        self.assertEqual(expected, self.available_backup_name(name))
 
2210
 
 
2211
    def test_empty(self):
 
2212
        self.assertBackupName('file.~1~', 'file')
 
2213
 
 
2214
    def test_existing(self):
 
2215
        self.available_backup_name('file')
 
2216
        self.available_backup_name('file')
 
2217
        self.assertBackupName('file.~3~', 'file')
 
2218
        # Empty slots are found, this is not a strict requirement and may be
 
2219
        # revisited if we test against all implementations.
 
2220
        self.backups.remove('file.~2~')
 
2221
        self.assertBackupName('file.~2~', 'file')
 
2222
 
 
2223
 
 
2224
class TestFindExecutableInPath(tests.TestCase):
 
2225
 
 
2226
    def test_windows(self):
 
2227
        if sys.platform != 'win32':
 
2228
            raise tests.TestSkipped('test requires win32')
 
2229
        self.assertTrue(osutils.find_executable_on_path('explorer') is not None)
 
2230
        self.assertTrue(
 
2231
            osutils.find_executable_on_path('explorer.exe') is not None)
 
2232
        self.assertTrue(
 
2233
            osutils.find_executable_on_path('EXPLORER.EXE') is not None)
 
2234
        self.assertTrue(
 
2235
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2236
        self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
 
2237
        
 
2238
    def test_windows_app_path(self):
 
2239
        if sys.platform != 'win32':
 
2240
            raise tests.TestSkipped('test requires win32')
 
2241
        # Override PATH env var so that exe can only be found on App Path
 
2242
        self.overrideEnv('PATH', '')
 
2243
        # Internt Explorer is always registered in the App Path
 
2244
        self.assertTrue(osutils.find_executable_on_path('iexplore') is not None)
 
2245
 
 
2246
    def test_other(self):
 
2247
        if sys.platform == 'win32':
 
2248
            raise tests.TestSkipped('test requires non-win32')
 
2249
        self.assertTrue(osutils.find_executable_on_path('sh') is not None)
 
2250
        self.assertTrue(
 
2251
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2252
 
 
2253
 
 
2254
class TestEnvironmentErrors(tests.TestCase):
 
2255
    """Test handling of environmental errors"""
 
2256
 
 
2257
    def test_is_oserror(self):
 
2258
        self.assertTrue(osutils.is_environment_error(
 
2259
            OSError(errno.EINVAL, "Invalid parameter")))
 
2260
 
 
2261
    def test_is_ioerror(self):
 
2262
        self.assertTrue(osutils.is_environment_error(
 
2263
            IOError(errno.EINVAL, "Invalid parameter")))
 
2264
 
 
2265
    def test_is_socket_error(self):
 
2266
        self.assertTrue(osutils.is_environment_error(
 
2267
            socket.error(errno.EINVAL, "Invalid parameter")))
 
2268
 
 
2269
    def test_is_select_error(self):
 
2270
        self.assertTrue(osutils.is_environment_error(
 
2271
            select.error(errno.EINVAL, "Invalid parameter")))
 
2272
 
 
2273
    def test_is_pywintypes_error(self):
 
2274
        self.requireFeature(features.pywintypes)
 
2275
        import pywintypes
 
2276
        self.assertTrue(osutils.is_environment_error(
 
2277
            pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))