~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Colin Watson
  • Date: 2015-07-02 10:37:05 UTC
  • mto: This revision was merged to the branch mainline in revision 6605.
  • Revision ID: cjwatson@canonical.com-20150702103705-zdfslk3wb70sz37n
Avoid associating dirty patch headers with the previous file in the patch.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006 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 tempfile
 
27
import time
24
28
 
25
 
import bzrlib
26
29
from bzrlib import (
27
30
    errors,
 
31
    lazy_regex,
28
32
    osutils,
 
33
    symbol_versioning,
 
34
    tests,
 
35
    trace,
29
36
    win32utils,
30
37
    )
31
 
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
32
38
from bzrlib.tests import (
33
 
        StringIOWrapper,
34
 
        TestCase, 
35
 
        TestCaseInTempDir, 
36
 
        TestSkipped,
37
 
        )
38
 
 
39
 
 
40
 
class TestOSUtils(TestCaseInTempDir):
 
39
    features,
 
40
    file_utils,
 
41
    test__walkdirs_win32,
 
42
    )
 
43
from bzrlib.tests.scenarios import load_tests_apply_scenarios
 
44
 
 
45
 
 
46
class _UTF8DirReaderFeature(features.Feature):
 
47
 
 
48
    def _probe(self):
 
49
        try:
 
50
            from bzrlib import _readdir_pyx
 
51
            self.reader = _readdir_pyx.UTF8DirReader
 
52
            return True
 
53
        except ImportError:
 
54
            return False
 
55
 
 
56
    def feature_name(self):
 
57
        return 'bzrlib._readdir_pyx'
 
58
 
 
59
UTF8DirReaderFeature = features.ModuleAvailableFeature('bzrlib._readdir_pyx')
 
60
 
 
61
term_ios_feature = features.ModuleAvailableFeature('termios')
 
62
 
 
63
 
 
64
def _already_unicode(s):
 
65
    return s
 
66
 
 
67
 
 
68
def _utf8_to_unicode(s):
 
69
    return s.decode('UTF-8')
 
70
 
 
71
 
 
72
def dir_reader_scenarios():
 
73
    # For each dir reader we define:
 
74
 
 
75
    # - native_to_unicode: a function converting the native_abspath as returned
 
76
    #   by DirReader.read_dir to its unicode representation
 
77
 
 
78
    # UnicodeDirReader is the fallback, it should be tested on all platforms.
 
79
    scenarios = [('unicode',
 
80
                  dict(_dir_reader_class=osutils.UnicodeDirReader,
 
81
                       _native_to_unicode=_already_unicode))]
 
82
    # Some DirReaders are platform specific and even there they may not be
 
83
    # available.
 
84
    if UTF8DirReaderFeature.available():
 
85
        from bzrlib import _readdir_pyx
 
86
        scenarios.append(('utf8',
 
87
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
 
88
                               _native_to_unicode=_utf8_to_unicode)))
 
89
 
 
90
    if test__walkdirs_win32.win32_readdir_feature.available():
 
91
        try:
 
92
            from bzrlib import _walkdirs_win32
 
93
            scenarios.append(
 
94
                ('win32',
 
95
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
 
96
                      _native_to_unicode=_already_unicode)))
 
97
        except ImportError:
 
98
            pass
 
99
    return scenarios
 
100
 
 
101
 
 
102
load_tests = load_tests_apply_scenarios
 
103
 
 
104
 
 
105
class TestContainsWhitespace(tests.TestCase):
41
106
 
42
107
    def test_contains_whitespace(self):
43
 
        self.failUnless(osutils.contains_whitespace(u' '))
44
 
        self.failUnless(osutils.contains_whitespace(u'hello there'))
45
 
        self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
46
 
        self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
47
 
        self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
48
 
        self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
 
108
        self.assertTrue(osutils.contains_whitespace(u' '))
 
109
        self.assertTrue(osutils.contains_whitespace(u'hello there'))
 
110
        self.assertTrue(osutils.contains_whitespace(u'hellothere\n'))
 
111
        self.assertTrue(osutils.contains_whitespace(u'hello\nthere'))
 
112
        self.assertTrue(osutils.contains_whitespace(u'hello\rthere'))
 
113
        self.assertTrue(osutils.contains_whitespace(u'hello\tthere'))
49
114
 
50
115
        # \xa0 is "Non-breaking-space" which on some python locales thinks it
51
116
        # is whitespace, but we do not.
52
 
        self.failIf(osutils.contains_whitespace(u''))
53
 
        self.failIf(osutils.contains_whitespace(u'hellothere'))
54
 
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
 
117
        self.assertFalse(osutils.contains_whitespace(u''))
 
118
        self.assertFalse(osutils.contains_whitespace(u'hellothere'))
 
119
        self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
 
120
 
 
121
 
 
122
class TestRename(tests.TestCaseInTempDir):
 
123
 
 
124
    def create_file(self, filename, content):
 
125
        f = open(filename, 'wb')
 
126
        try:
 
127
            f.write(content)
 
128
        finally:
 
129
            f.close()
 
130
 
 
131
    def _fancy_rename(self, a, b):
 
132
        osutils.fancy_rename(a, b, rename_func=os.rename,
 
133
                             unlink_func=os.unlink)
55
134
 
56
135
    def test_fancy_rename(self):
57
136
        # This should work everywhere
58
 
        def rename(a, b):
59
 
            osutils.fancy_rename(a, b,
60
 
                    rename_func=os.rename,
61
 
                    unlink_func=os.unlink)
62
 
 
63
 
        open('a', 'wb').write('something in a\n')
64
 
        rename('a', 'b')
65
 
        self.failIfExists('a')
66
 
        self.failUnlessExists('b')
 
137
        self.create_file('a', 'something in a\n')
 
138
        self._fancy_rename('a', 'b')
 
139
        self.assertPathDoesNotExist('a')
 
140
        self.assertPathExists('b')
67
141
        self.check_file_contents('b', 'something in a\n')
68
142
 
69
 
        open('a', 'wb').write('new something in a\n')
70
 
        rename('b', 'a')
 
143
        self.create_file('a', 'new something in a\n')
 
144
        self._fancy_rename('b', 'a')
71
145
 
72
146
        self.check_file_contents('a', 'something in a\n')
73
147
 
 
148
    def test_fancy_rename_fails_source_missing(self):
 
149
        # An exception should be raised, and the target should be left in place
 
150
        self.create_file('target', 'data in target\n')
 
151
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
152
                          'missingsource', 'target')
 
153
        self.assertPathExists('target')
 
154
        self.check_file_contents('target', 'data in target\n')
 
155
 
 
156
    def test_fancy_rename_fails_if_source_and_target_missing(self):
 
157
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
158
                          'missingsource', 'missingtarget')
 
159
 
74
160
    def test_rename(self):
75
161
        # Rename should be semi-atomic on all platforms
76
 
        open('a', 'wb').write('something in a\n')
 
162
        self.create_file('a', 'something in a\n')
77
163
        osutils.rename('a', 'b')
78
 
        self.failIfExists('a')
79
 
        self.failUnlessExists('b')
 
164
        self.assertPathDoesNotExist('a')
 
165
        self.assertPathExists('b')
80
166
        self.check_file_contents('b', 'something in a\n')
81
167
 
82
 
        open('a', 'wb').write('new something in a\n')
 
168
        self.create_file('a', 'new something in a\n')
83
169
        osutils.rename('b', 'a')
84
170
 
85
171
        self.check_file_contents('a', 'something in a\n')
86
172
 
87
173
    # TODO: test fancy_rename using a MemoryTransport
88
174
 
 
175
    def test_rename_change_case(self):
 
176
        # on Windows we should be able to change filename case by rename
 
177
        self.build_tree(['a', 'b/'])
 
178
        osutils.rename('a', 'A')
 
179
        osutils.rename('b', 'B')
 
180
        # we can't use failUnlessExists on case-insensitive filesystem
 
181
        # so try to check shape of the tree
 
182
        shape = sorted(os.listdir('.'))
 
183
        self.assertEquals(['A', 'B'], shape)
 
184
 
 
185
    def test_rename_exception(self):
 
186
        try:
 
187
            osutils.rename('nonexistent_path', 'different_nonexistent_path')
 
188
        except OSError, e:
 
189
            self.assertEqual(e.old_filename, 'nonexistent_path')
 
190
            self.assertEqual(e.new_filename, 'different_nonexistent_path')
 
191
            self.assertTrue('nonexistent_path' in e.strerror)
 
192
            self.assertTrue('different_nonexistent_path' in e.strerror)
 
193
 
 
194
 
 
195
class TestRandChars(tests.TestCase):
 
196
 
89
197
    def test_01_rand_chars_empty(self):
90
198
        result = osutils.rand_chars(0)
91
199
        self.assertEqual(result, '')
96
204
        self.assertEqual(type(result), str)
97
205
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
98
206
 
 
207
 
 
208
class TestIsInside(tests.TestCase):
 
209
 
99
210
    def test_is_inside(self):
100
211
        is_inside = osutils.is_inside
101
212
        self.assertTrue(is_inside('src', 'src/foo.c'))
105
216
        self.assertFalse(is_inside('foo.c', ''))
106
217
        self.assertTrue(is_inside('', 'foo.c'))
107
218
 
 
219
    def test_is_inside_any(self):
 
220
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
 
221
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
 
222
                         (['src'], SRC_FOO_C),
 
223
                         (['src'], 'src'),
 
224
                         ]:
 
225
            self.assert_(osutils.is_inside_any(dirs, fn))
 
226
        for dirs, fn in [(['src'], 'srccontrol'),
 
227
                         (['src'], 'srccontrol/foo')]:
 
228
            self.assertFalse(osutils.is_inside_any(dirs, fn))
 
229
 
 
230
    def test_is_inside_or_parent_of_any(self):
 
231
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
 
232
                         (['src'], 'src/foo.c'),
 
233
                         (['src/bar.c'], 'src'),
 
234
                         (['src/bar.c', 'bla/foo.c'], 'src'),
 
235
                         (['src'], 'src'),
 
236
                         ]:
 
237
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
238
 
 
239
        for dirs, fn in [(['src'], 'srccontrol'),
 
240
                         (['srccontrol/foo.c'], 'src'),
 
241
                         (['src'], 'srccontrol/foo')]:
 
242
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
 
243
 
 
244
 
 
245
class TestLstat(tests.TestCaseInTempDir):
 
246
 
 
247
    def test_lstat_matches_fstat(self):
 
248
        # On Windows, lstat and fstat don't always agree, primarily in the
 
249
        # 'st_ino' and 'st_dev' fields. So we force them to be '0' in our
 
250
        # custom implementation.
 
251
        if sys.platform == 'win32':
 
252
            # We only have special lstat/fstat if we have the extension.
 
253
            # Without it, we may end up re-reading content when we don't have
 
254
            # to, but otherwise it doesn't effect correctness.
 
255
            self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
256
        f = open('test-file.txt', 'wb')
 
257
        self.addCleanup(f.close)
 
258
        f.write('some content\n')
 
259
        f.flush()
 
260
        self.assertEqualStat(osutils.fstat(f.fileno()),
 
261
                             osutils.lstat('test-file.txt'))
 
262
 
 
263
 
 
264
class TestRmTree(tests.TestCaseInTempDir):
 
265
 
108
266
    def test_rmtree(self):
109
267
        # Check to remove tree with read-only files/dirs
110
268
        os.mkdir('dir')
120
278
 
121
279
        osutils.rmtree('dir')
122
280
 
123
 
        self.failIfExists('dir/file')
124
 
        self.failIfExists('dir')
 
281
        self.assertPathDoesNotExist('dir/file')
 
282
        self.assertPathDoesNotExist('dir')
 
283
 
 
284
 
 
285
class TestDeleteAny(tests.TestCaseInTempDir):
 
286
 
 
287
    def test_delete_any_readonly(self):
 
288
        # from <https://bugs.launchpad.net/bzr/+bug/218206>
 
289
        self.build_tree(['d/', 'f'])
 
290
        osutils.make_readonly('d')
 
291
        osutils.make_readonly('f')
 
292
 
 
293
        osutils.delete_any('f')
 
294
        osutils.delete_any('d')
 
295
 
 
296
 
 
297
class TestKind(tests.TestCaseInTempDir):
125
298
 
126
299
    def test_file_kind(self):
127
300
        self.build_tree(['file', 'dir/'])
130
303
        if osutils.has_symlinks():
131
304
            os.symlink('symlink', 'symlink')
132
305
            self.assertEquals('symlink', osutils.file_kind('symlink'))
133
 
        
 
306
 
134
307
        # TODO: jam 20060529 Test a block device
135
308
        try:
136
309
            os.lstat('/dev/null')
157
330
            finally:
158
331
                os.remove('socket')
159
332
 
 
333
    def test_kind_marker(self):
 
334
        self.assertEqual("", osutils.kind_marker("file"))
 
335
        self.assertEqual("/", osutils.kind_marker('directory'))
 
336
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
337
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
338
        self.assertEqual("+", osutils.kind_marker("tree-reference"))
 
339
        self.assertEqual("", osutils.kind_marker("fifo"))
 
340
        self.assertEqual("", osutils.kind_marker("socket"))
 
341
        self.assertEqual("", osutils.kind_marker("unknown"))
 
342
 
 
343
 
 
344
class TestUmask(tests.TestCaseInTempDir):
 
345
 
160
346
    def test_get_umask(self):
161
347
        if sys.platform == 'win32':
162
348
            # umask always returns '0', no way to set it
164
350
            return
165
351
 
166
352
        orig_umask = osutils.get_umask()
167
 
        try:
168
 
            os.umask(0222)
169
 
            self.assertEqual(0222, osutils.get_umask())
170
 
            os.umask(0022)
171
 
            self.assertEqual(0022, osutils.get_umask())
172
 
            os.umask(0002)
173
 
            self.assertEqual(0002, osutils.get_umask())
174
 
            os.umask(0027)
175
 
            self.assertEqual(0027, osutils.get_umask())
176
 
        finally:
177
 
            os.umask(orig_umask)
 
353
        self.addCleanup(os.umask, orig_umask)
 
354
        os.umask(0222)
 
355
        self.assertEqual(0222, osutils.get_umask())
 
356
        os.umask(0022)
 
357
        self.assertEqual(0022, osutils.get_umask())
 
358
        os.umask(0002)
 
359
        self.assertEqual(0002, osutils.get_umask())
 
360
        os.umask(0027)
 
361
        self.assertEqual(0027, osutils.get_umask())
 
362
 
 
363
 
 
364
class TestDateTime(tests.TestCase):
178
365
 
179
366
    def assertFormatedDelta(self, expected, seconds):
180
367
        """Assert osutils.format_delta formats as expected"""
212
399
        self.assertFormatedDelta('1 second in the future', -1)
213
400
        self.assertFormatedDelta('2 seconds in the future', -2)
214
401
 
 
402
    def test_format_date(self):
 
403
        self.assertRaises(errors.UnsupportedTimezoneFormat,
 
404
            osutils.format_date, 0, timezone='foo')
 
405
        self.assertIsInstance(osutils.format_date(0), str)
 
406
        self.assertIsInstance(osutils.format_local_date(0), unicode)
 
407
        # Testing for the actual value of the local weekday without
 
408
        # duplicating the code from format_date is difficult.
 
409
        # Instead blackbox.test_locale should check for localized
 
410
        # dates once they do occur in output strings.
 
411
 
 
412
    def test_format_date_with_offset_in_original_timezone(self):
 
413
        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
 
414
            osutils.format_date_with_offset_in_original_timezone(0))
 
415
        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
 
416
            osutils.format_date_with_offset_in_original_timezone(100000))
 
417
        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
 
418
            osutils.format_date_with_offset_in_original_timezone(100000, 7200))
 
419
 
 
420
    def test_local_time_offset(self):
 
421
        """Test that local_time_offset() returns a sane value."""
 
422
        offset = osutils.local_time_offset()
 
423
        self.assertTrue(isinstance(offset, int))
 
424
        # Test that the offset is no more than a eighteen hours in
 
425
        # either direction.
 
426
        # Time zone handling is system specific, so it is difficult to
 
427
        # do more specific tests, but a value outside of this range is
 
428
        # probably wrong.
 
429
        eighteen_hours = 18 * 3600
 
430
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
431
 
 
432
    def test_local_time_offset_with_timestamp(self):
 
433
        """Test that local_time_offset() works with a timestamp."""
 
434
        offset = osutils.local_time_offset(1000000000.1234567)
 
435
        self.assertTrue(isinstance(offset, int))
 
436
        eighteen_hours = 18 * 3600
 
437
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
438
 
 
439
 
 
440
class TestFdatasync(tests.TestCaseInTempDir):
 
441
 
 
442
    def do_fdatasync(self):
 
443
        f = tempfile.NamedTemporaryFile()
 
444
        osutils.fdatasync(f.fileno())
 
445
        f.close()
 
446
 
 
447
    @staticmethod
 
448
    def raise_eopnotsupp(*args, **kwargs):
 
449
        raise IOError(errno.EOPNOTSUPP, os.strerror(errno.EOPNOTSUPP))
 
450
 
 
451
    @staticmethod
 
452
    def raise_enotsup(*args, **kwargs):
 
453
        raise IOError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
 
454
 
 
455
    def test_fdatasync_handles_system_function(self):
 
456
        self.overrideAttr(os, "fdatasync")
 
457
        self.do_fdatasync()
 
458
 
 
459
    def test_fdatasync_handles_no_fdatasync_no_fsync(self):
 
460
        self.overrideAttr(os, "fdatasync")
 
461
        self.overrideAttr(os, "fsync")
 
462
        self.do_fdatasync()
 
463
 
 
464
    def test_fdatasync_handles_no_EOPNOTSUPP(self):
 
465
        self.overrideAttr(errno, "EOPNOTSUPP")
 
466
        self.do_fdatasync()
 
467
 
 
468
    def test_fdatasync_catches_ENOTSUP(self):
 
469
        enotsup = getattr(errno, "ENOTSUP", None)
 
470
        if enotsup is None:
 
471
            raise tests.TestNotApplicable("No ENOTSUP on this platform")
 
472
        self.overrideAttr(os, "fdatasync", self.raise_enotsup)
 
473
        self.do_fdatasync()
 
474
 
 
475
    def test_fdatasync_catches_EOPNOTSUPP(self):
 
476
        enotsup = getattr(errno, "EOPNOTSUPP", None)
 
477
        if enotsup is None:
 
478
            raise tests.TestNotApplicable("No EOPNOTSUPP on this platform")
 
479
        self.overrideAttr(os, "fdatasync", self.raise_eopnotsupp)
 
480
        self.do_fdatasync()
 
481
 
 
482
 
 
483
class TestLinks(tests.TestCaseInTempDir):
 
484
 
215
485
    def test_dereference_path(self):
216
 
        if not osutils.has_symlinks():
217
 
            raise TestSkipped('Symlinks are not supported on this platform')
 
486
        self.requireFeature(features.SymlinkFeature)
218
487
        cwd = osutils.realpath('.')
219
488
        os.mkdir('bar')
220
489
        bar_path = osutils.pathjoin(cwd, 'bar')
223
492
        self.assertEqual(bar_path, osutils.realpath('./bar'))
224
493
        os.symlink('bar', 'foo')
225
494
        self.assertEqual(bar_path, osutils.realpath('./foo'))
226
 
        
 
495
 
227
496
        # Does not dereference terminal symlinks
228
497
        foo_path = osutils.pathjoin(cwd, 'foo')
229
498
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
240
509
        foo_baz_path = osutils.pathjoin(foo_path, 'baz')
241
510
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
242
511
 
243
 
 
244
 
class TestSafeUnicode(TestCase):
 
512
    def test_changing_access(self):
 
513
        f = file('file', 'w')
 
514
        f.write('monkey')
 
515
        f.close()
 
516
 
 
517
        # Make a file readonly
 
518
        osutils.make_readonly('file')
 
519
        mode = os.lstat('file').st_mode
 
520
        self.assertEqual(mode, mode & 0777555)
 
521
 
 
522
        # Make a file writable
 
523
        osutils.make_writable('file')
 
524
        mode = os.lstat('file').st_mode
 
525
        self.assertEqual(mode, mode | 0200)
 
526
 
 
527
        if osutils.has_symlinks():
 
528
            # should not error when handed a symlink
 
529
            os.symlink('nonexistent', 'dangling')
 
530
            osutils.make_readonly('dangling')
 
531
            osutils.make_writable('dangling')
 
532
 
 
533
    def test_host_os_dereferences_symlinks(self):
 
534
        osutils.host_os_dereferences_symlinks()
 
535
 
 
536
 
 
537
class TestCanonicalRelPath(tests.TestCaseInTempDir):
 
538
 
 
539
    _test_needs_features = [features.CaseInsCasePresFilenameFeature]
 
540
 
 
541
    def test_canonical_relpath_simple(self):
 
542
        f = file('MixedCaseName', 'w')
 
543
        f.close()
 
544
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
 
545
        self.assertEqual('work/MixedCaseName', actual)
 
546
 
 
547
    def test_canonical_relpath_missing_tail(self):
 
548
        os.mkdir('MixedCaseParent')
 
549
        actual = osutils.canonical_relpath(self.test_base_dir,
 
550
                                           'mixedcaseparent/nochild')
 
551
        self.assertEqual('work/MixedCaseParent/nochild', actual)
 
552
 
 
553
 
 
554
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
 
555
 
 
556
    def assertRelpath(self, expected, base, path):
 
557
        actual = osutils._cicp_canonical_relpath(base, path)
 
558
        self.assertEqual(expected, actual)
 
559
 
 
560
    def test_simple(self):
 
561
        self.build_tree(['MixedCaseName'])
 
562
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
563
        self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
 
564
 
 
565
    def test_subdir_missing_tail(self):
 
566
        self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
 
567
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
568
        self.assertRelpath('MixedCaseParent/a_child', base,
 
569
                           'MixedCaseParent/a_child')
 
570
        self.assertRelpath('MixedCaseParent/a_child', base,
 
571
                           'MixedCaseParent/A_Child')
 
572
        self.assertRelpath('MixedCaseParent/not_child', base,
 
573
                           'MixedCaseParent/not_child')
 
574
 
 
575
    def test_at_root_slash(self):
 
576
        # We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
 
577
        # check...
 
578
        if osutils.MIN_ABS_PATHLENGTH > 1:
 
579
            raise tests.TestSkipped('relpath requires %d chars'
 
580
                                    % osutils.MIN_ABS_PATHLENGTH)
 
581
        self.assertRelpath('foo', '/', '/foo')
 
582
 
 
583
    def test_at_root_drive(self):
 
584
        if sys.platform != 'win32':
 
585
            raise tests.TestNotApplicable('we can only test drive-letter relative'
 
586
                                          ' paths on Windows where we have drive'
 
587
                                          ' letters.')
 
588
        # see bug #322807
 
589
        # The specific issue is that when at the root of a drive, 'abspath'
 
590
        # returns "C:/" or just "/". However, the code assumes that abspath
 
591
        # always returns something like "C:/foo" or "/foo" (no trailing slash).
 
592
        self.assertRelpath('foo', 'C:/', 'C:/foo')
 
593
        self.assertRelpath('foo', 'X:/', 'X:/foo')
 
594
        self.assertRelpath('foo', 'X:/', 'X://foo')
 
595
 
 
596
 
 
597
class TestPumpFile(tests.TestCase):
 
598
    """Test pumpfile method."""
 
599
 
 
600
    def setUp(self):
 
601
        super(TestPumpFile, self).setUp()
 
602
        # create a test datablock
 
603
        self.block_size = 512
 
604
        pattern = '0123456789ABCDEF'
 
605
        self.test_data = pattern * (3 * self.block_size / len(pattern))
 
606
        self.test_data_len = len(self.test_data)
 
607
 
 
608
    def test_bracket_block_size(self):
 
609
        """Read data in blocks with the requested read size bracketing the
 
610
        block size."""
 
611
        # make sure test data is larger than max read size
 
612
        self.assertTrue(self.test_data_len > self.block_size)
 
613
 
 
614
        from_file = file_utils.FakeReadFile(self.test_data)
 
615
        to_file = StringIO()
 
616
 
 
617
        # read (max / 2) bytes and verify read size wasn't affected
 
618
        num_bytes_to_read = self.block_size / 2
 
619
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
620
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
621
        self.assertEqual(from_file.get_read_count(), 1)
 
622
 
 
623
        # read (max) bytes and verify read size wasn't affected
 
624
        num_bytes_to_read = self.block_size
 
625
        from_file.reset_read_count()
 
626
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
627
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
628
        self.assertEqual(from_file.get_read_count(), 1)
 
629
 
 
630
        # read (max + 1) bytes and verify read size was limited
 
631
        num_bytes_to_read = self.block_size + 1
 
632
        from_file.reset_read_count()
 
633
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
634
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
635
        self.assertEqual(from_file.get_read_count(), 2)
 
636
 
 
637
        # finish reading the rest of the data
 
638
        num_bytes_to_read = self.test_data_len - to_file.tell()
 
639
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
640
 
 
641
        # report error if the data wasn't equal (we only report the size due
 
642
        # to the length of the data)
 
643
        response_data = to_file.getvalue()
 
644
        if response_data != self.test_data:
 
645
            message = "Data not equal.  Expected %d bytes, received %d."
 
646
            self.fail(message % (len(response_data), self.test_data_len))
 
647
 
 
648
    def test_specified_size(self):
 
649
        """Request a transfer larger than the maximum block size and verify
 
650
        that the maximum read doesn't exceed the block_size."""
 
651
        # make sure test data is larger than max read size
 
652
        self.assertTrue(self.test_data_len > self.block_size)
 
653
 
 
654
        # retrieve data in blocks
 
655
        from_file = file_utils.FakeReadFile(self.test_data)
 
656
        to_file = StringIO()
 
657
        osutils.pumpfile(from_file, to_file, self.test_data_len,
 
658
                         self.block_size)
 
659
 
 
660
        # verify read size was equal to the maximum read size
 
661
        self.assertTrue(from_file.get_max_read_size() > 0)
 
662
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
663
        self.assertEqual(from_file.get_read_count(), 3)
 
664
 
 
665
        # report error if the data wasn't equal (we only report the size due
 
666
        # to the length of the data)
 
667
        response_data = to_file.getvalue()
 
668
        if response_data != self.test_data:
 
669
            message = "Data not equal.  Expected %d bytes, received %d."
 
670
            self.fail(message % (len(response_data), self.test_data_len))
 
671
 
 
672
    def test_to_eof(self):
 
673
        """Read to end-of-file and verify that the reads are not larger than
 
674
        the maximum read size."""
 
675
        # make sure test data is larger than max read size
 
676
        self.assertTrue(self.test_data_len > self.block_size)
 
677
 
 
678
        # retrieve data to EOF
 
679
        from_file = file_utils.FakeReadFile(self.test_data)
 
680
        to_file = StringIO()
 
681
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
 
682
 
 
683
        # verify read size was equal to the maximum read size
 
684
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
685
        self.assertEqual(from_file.get_read_count(), 4)
 
686
 
 
687
        # report error if the data wasn't equal (we only report the size due
 
688
        # to the length of the data)
 
689
        response_data = to_file.getvalue()
 
690
        if response_data != self.test_data:
 
691
            message = "Data not equal.  Expected %d bytes, received %d."
 
692
            self.fail(message % (len(response_data), self.test_data_len))
 
693
 
 
694
    def test_defaults(self):
 
695
        """Verifies that the default arguments will read to EOF -- this
 
696
        test verifies that any existing usages of pumpfile will not be broken
 
697
        with this new version."""
 
698
        # retrieve data using default (old) pumpfile method
 
699
        from_file = file_utils.FakeReadFile(self.test_data)
 
700
        to_file = StringIO()
 
701
        osutils.pumpfile(from_file, to_file)
 
702
 
 
703
        # report error if the data wasn't equal (we only report the size due
 
704
        # to the length of the data)
 
705
        response_data = to_file.getvalue()
 
706
        if response_data != self.test_data:
 
707
            message = "Data not equal.  Expected %d bytes, received %d."
 
708
            self.fail(message % (len(response_data), self.test_data_len))
 
709
 
 
710
    def test_report_activity(self):
 
711
        activity = []
 
712
        def log_activity(length, direction):
 
713
            activity.append((length, direction))
 
714
        from_file = StringIO(self.test_data)
 
715
        to_file = StringIO()
 
716
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
717
                         report_activity=log_activity, direction='read')
 
718
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
 
719
                          (36, 'read')], activity)
 
720
 
 
721
        from_file = StringIO(self.test_data)
 
722
        to_file = StringIO()
 
723
        del activity[:]
 
724
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
725
                         report_activity=log_activity, direction='write')
 
726
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
 
727
                          (36, 'write')], activity)
 
728
 
 
729
        # And with a limited amount of data
 
730
        from_file = StringIO(self.test_data)
 
731
        to_file = StringIO()
 
732
        del activity[:]
 
733
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
 
734
                         report_activity=log_activity, direction='read')
 
735
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
 
736
 
 
737
 
 
738
 
 
739
class TestPumpStringFile(tests.TestCase):
 
740
 
 
741
    def test_empty(self):
 
742
        output = StringIO()
 
743
        osutils.pump_string_file("", output)
 
744
        self.assertEqual("", output.getvalue())
 
745
 
 
746
    def test_more_than_segment_size(self):
 
747
        output = StringIO()
 
748
        osutils.pump_string_file("123456789", output, 2)
 
749
        self.assertEqual("123456789", output.getvalue())
 
750
 
 
751
    def test_segment_size(self):
 
752
        output = StringIO()
 
753
        osutils.pump_string_file("12", output, 2)
 
754
        self.assertEqual("12", output.getvalue())
 
755
 
 
756
    def test_segment_size_multiple(self):
 
757
        output = StringIO()
 
758
        osutils.pump_string_file("1234", output, 2)
 
759
        self.assertEqual("1234", output.getvalue())
 
760
 
 
761
 
 
762
class TestRelpath(tests.TestCase):
 
763
 
 
764
    def test_simple_relpath(self):
 
765
        cwd = osutils.getcwd()
 
766
        subdir = cwd + '/subdir'
 
767
        self.assertEqual('subdir', osutils.relpath(cwd, subdir))
 
768
 
 
769
    def test_deep_relpath(self):
 
770
        cwd = osutils.getcwd()
 
771
        subdir = cwd + '/sub/subsubdir'
 
772
        self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
 
773
 
 
774
    def test_not_relative(self):
 
775
        self.assertRaises(errors.PathNotChild,
 
776
                          osutils.relpath, 'C:/path', 'H:/path')
 
777
        self.assertRaises(errors.PathNotChild,
 
778
                          osutils.relpath, 'C:/', 'H:/path')
 
779
 
 
780
 
 
781
class TestSafeUnicode(tests.TestCase):
245
782
 
246
783
    def test_from_ascii_string(self):
247
784
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
256
793
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
257
794
 
258
795
    def test_bad_utf8_string(self):
259
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
796
        self.assertRaises(errors.BzrBadParameterNotUnicode,
260
797
                          osutils.safe_unicode,
261
798
                          '\xbb\xbb')
262
799
 
263
800
 
264
 
class TestSafeUtf8(TestCase):
 
801
class TestSafeUtf8(tests.TestCase):
265
802
 
266
803
    def test_from_ascii_string(self):
267
804
        f = 'foobar'
277
814
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
278
815
 
279
816
    def test_bad_utf8_string(self):
280
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
817
        self.assertRaises(errors.BzrBadParameterNotUnicode,
281
818
                          osutils.safe_utf8, '\xbb\xbb')
282
819
 
283
820
 
284
 
class TestSafeRevisionId(TestCase):
 
821
class TestSafeRevisionId(tests.TestCase):
285
822
 
286
823
    def test_from_ascii_string(self):
 
824
        # this shouldn't give a warning because it's getting an ascii string
287
825
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
288
826
 
289
827
    def test_from_unicode_string_ascii_contents(self):
308
846
        self.assertEqual(None, osutils.safe_revision_id(None))
309
847
 
310
848
 
311
 
class TestSafeFileId(TestCase):
 
849
class TestSafeFileId(tests.TestCase):
312
850
 
313
851
    def test_from_ascii_string(self):
314
852
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
334
872
        self.assertEqual(None, osutils.safe_file_id(None))
335
873
 
336
874
 
337
 
class TestWin32Funcs(TestCase):
338
 
    """Test that the _win32 versions of os utilities return appropriate paths."""
 
875
class TestSendAll(tests.TestCase):
 
876
 
 
877
    def test_send_with_disconnected_socket(self):
 
878
        class DisconnectedSocket(object):
 
879
            def __init__(self, err):
 
880
                self.err = err
 
881
            def send(self, content):
 
882
                raise self.err
 
883
            def close(self):
 
884
                pass
 
885
        # All of these should be treated as ConnectionReset
 
886
        errs = []
 
887
        for err_cls in (IOError, socket.error):
 
888
            for errnum in osutils._end_of_stream_errors:
 
889
                errs.append(err_cls(errnum))
 
890
        for err in errs:
 
891
            sock = DisconnectedSocket(err)
 
892
            self.assertRaises(errors.ConnectionReset,
 
893
                osutils.send_all, sock, 'some more content')
 
894
 
 
895
    def test_send_with_no_progress(self):
 
896
        # See https://bugs.launchpad.net/bzr/+bug/1047309
 
897
        # It seems that paramiko can get into a state where it doesn't error,
 
898
        # but it returns 0 bytes sent for requests over and over again.
 
899
        class NoSendingSocket(object):
 
900
            def __init__(self):
 
901
                self.call_count = 0
 
902
            def send(self, bytes):
 
903
                self.call_count += 1
 
904
                if self.call_count > 100:
 
905
                    # Prevent the test suite from hanging
 
906
                    raise RuntimeError('too many calls')
 
907
                return 0
 
908
        sock = NoSendingSocket()
 
909
        self.assertRaises(errors.ConnectionReset,
 
910
                          osutils.send_all, sock, 'content')
 
911
        self.assertEqual(1, sock.call_count)
 
912
 
 
913
 
 
914
class TestPosixFuncs(tests.TestCase):
 
915
    """Test that the posix version of normpath returns an appropriate path
 
916
       when used with 2 leading slashes."""
 
917
 
 
918
    def test_normpath(self):
 
919
        self.assertEqual('/etc/shadow', osutils._posix_normpath('/etc/shadow'))
 
920
        self.assertEqual('/etc/shadow', osutils._posix_normpath('//etc/shadow'))
 
921
        self.assertEqual('/etc/shadow', osutils._posix_normpath('///etc/shadow'))
 
922
 
 
923
 
 
924
class TestWin32Funcs(tests.TestCase):
 
925
    """Test that _win32 versions of os utilities return appropriate paths."""
339
926
 
340
927
    def test_abspath(self):
341
928
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
348
935
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
349
936
 
350
937
    def test_pathjoin(self):
351
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
352
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
353
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
354
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
355
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
356
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
938
        self.assertEqual('path/to/foo',
 
939
                         osutils._win32_pathjoin('path', 'to', 'foo'))
 
940
        self.assertEqual('C:/foo',
 
941
                         osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
942
        self.assertEqual('C:/foo',
 
943
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
 
944
        self.assertEqual('path/to/foo',
 
945
                         osutils._win32_pathjoin('path/to/', 'foo'))
 
946
 
 
947
    def test_pathjoin_late_bugfix(self):
 
948
        if sys.version_info < (2, 7, 6):
 
949
            expected = '/foo'
 
950
        else:
 
951
            expected = 'C:/foo'
 
952
        self.assertEqual(expected,
 
953
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
954
        self.assertEqual(expected,
 
955
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
357
956
 
358
957
    def test_normpath(self):
359
 
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
360
 
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
 
958
        self.assertEqual('path/to/foo',
 
959
                         osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
960
        self.assertEqual('path/to/foo',
 
961
                         osutils._win32_normpath('path//from/../to/./foo'))
361
962
 
362
963
    def test_getcwd(self):
363
964
        cwd = osutils._win32_getcwd()
384
985
        self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
385
986
        # relative path
386
987
        cwd = osutils.getcwd().rstrip('/')
387
 
        drive = osutils._nt_splitdrive(cwd)[0]
 
988
        drive = osutils.ntpath.splitdrive(cwd)[0]
388
989
        self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
389
990
        self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
390
991
        # unicode path
392
993
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
393
994
 
394
995
 
395
 
class TestWin32FuncsDirs(TestCaseInTempDir):
 
996
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
396
997
    """Test win32 functions that create files."""
397
 
    
 
998
 
398
999
    def test_getcwd(self):
399
 
        if win32utils.winver == 'Windows 98':
400
 
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
401
 
        # Make sure getcwd can handle unicode filenames
402
 
        try:
403
 
            os.mkdir(u'mu-\xb5')
404
 
        except UnicodeError:
405
 
            raise TestSkipped("Unable to create Unicode filename")
406
 
 
 
1000
        self.requireFeature(features.UnicodeFilenameFeature)
 
1001
        os.mkdir(u'mu-\xb5')
407
1002
        os.chdir(u'mu-\xb5')
408
1003
        # TODO: jam 20060427 This will probably fail on Mac OSX because
409
1004
        #       it will change the normalization of B\xe5gfors
411
1006
        #       osutils.getcwd() renormalize the path.
412
1007
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
413
1008
 
 
1009
    def test_minimum_path_selection(self):
 
1010
        self.assertEqual(set(),
 
1011
            osutils.minimum_path_selection([]))
 
1012
        self.assertEqual(set(['a']),
 
1013
            osutils.minimum_path_selection(['a']))
 
1014
        self.assertEqual(set(['a', 'b']),
 
1015
            osutils.minimum_path_selection(['a', 'b']))
 
1016
        self.assertEqual(set(['a/', 'b']),
 
1017
            osutils.minimum_path_selection(['a/', 'b']))
 
1018
        self.assertEqual(set(['a/', 'b']),
 
1019
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
 
1020
        self.assertEqual(set(['a-b', 'a', 'a0b']),
 
1021
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
 
1022
 
414
1023
    def test_mkdtemp(self):
415
1024
        tmpdir = osutils._win32_mkdtemp(dir='.')
416
1025
        self.assertFalse('\\' in tmpdir)
424
1033
        b.close()
425
1034
 
426
1035
        osutils._win32_rename('b', 'a')
427
 
        self.failUnlessExists('a')
428
 
        self.failIfExists('b')
 
1036
        self.assertPathExists('a')
 
1037
        self.assertPathDoesNotExist('b')
429
1038
        self.assertFileEqual('baz\n', 'a')
430
1039
 
431
1040
    def test_rename_missing_file(self):
471
1080
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
472
1081
 
473
1082
 
474
 
class TestMacFuncsDirs(TestCaseInTempDir):
 
1083
class TestParentDirectories(tests.TestCaseInTempDir):
 
1084
    """Test osutils.parent_directories()"""
 
1085
 
 
1086
    def test_parent_directories(self):
 
1087
        self.assertEqual([], osutils.parent_directories('a'))
 
1088
        self.assertEqual(['a'], osutils.parent_directories('a/b'))
 
1089
        self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
 
1090
 
 
1091
 
 
1092
class TestMacFuncsDirs(tests.TestCaseInTempDir):
475
1093
    """Test mac special functions that require directories."""
476
1094
 
477
1095
    def test_getcwd(self):
478
 
        # On Mac, this will actually create Ba\u030agfors
479
 
        # but chdir will still work, because it accepts both paths
480
 
        try:
481
 
            os.mkdir(u'B\xe5gfors')
482
 
        except UnicodeError:
483
 
            raise TestSkipped("Unable to create Unicode filename")
484
 
 
 
1096
        self.requireFeature(features.UnicodeFilenameFeature)
 
1097
        os.mkdir(u'B\xe5gfors')
485
1098
        os.chdir(u'B\xe5gfors')
486
1099
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
487
1100
 
488
1101
    def test_getcwd_nonnorm(self):
 
1102
        self.requireFeature(features.UnicodeFilenameFeature)
489
1103
        # Test that _mac_getcwd() will normalize this path
490
 
        try:
491
 
            os.mkdir(u'Ba\u030agfors')
492
 
        except UnicodeError:
493
 
            raise TestSkipped("Unable to create Unicode filename")
494
 
 
 
1104
        os.mkdir(u'Ba\u030agfors')
495
1105
        os.chdir(u'Ba\u030agfors')
496
1106
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
497
1107
 
498
1108
 
499
 
class TestSplitLines(TestCase):
 
1109
class TestChunksToLines(tests.TestCase):
 
1110
 
 
1111
    def test_smoketest(self):
 
1112
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
1113
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
 
1114
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
1115
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
 
1116
 
 
1117
    def test_osutils_binding(self):
 
1118
        from bzrlib.tests import test__chunks_to_lines
 
1119
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
 
1120
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
 
1121
        else:
 
1122
            from bzrlib._chunks_to_lines_py import chunks_to_lines
 
1123
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
 
1124
 
 
1125
 
 
1126
class TestSplitLines(tests.TestCase):
500
1127
 
501
1128
    def test_split_unicode(self):
502
1129
        self.assertEqual([u'foo\n', u'bar\xae'],
509
1136
                         osutils.split_lines('foo\rbar\n'))
510
1137
 
511
1138
 
512
 
class TestWalkDirs(TestCaseInTempDir):
 
1139
class TestWalkDirs(tests.TestCaseInTempDir):
 
1140
 
 
1141
    def assertExpectedBlocks(self, expected, result):
 
1142
        self.assertEqual(expected,
 
1143
                         [(dirinfo, [line[0:3] for line in block])
 
1144
                          for dirinfo, block in result])
513
1145
 
514
1146
    def test_walkdirs(self):
515
1147
        tree = [
548
1180
            result.append((dirdetail, dirblock))
549
1181
 
550
1182
        self.assertTrue(found_bzrdir)
551
 
        self.assertEqual(expected_dirblocks,
552
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1183
        self.assertExpectedBlocks(expected_dirblocks, result)
553
1184
        # you can search a subdir only, with a supplied prefix.
554
1185
        result = []
555
1186
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
556
1187
            result.append(dirblock)
557
 
        self.assertEqual(expected_dirblocks[1:],
558
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1188
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
1189
 
 
1190
    def test_walkdirs_os_error(self):
 
1191
        # <https://bugs.launchpad.net/bzr/+bug/338653>
 
1192
        # Pyrex readdir didn't raise useful messages if it had an error
 
1193
        # reading the directory
 
1194
        if sys.platform == 'win32':
 
1195
            raise tests.TestNotApplicable(
 
1196
                "readdir IOError not tested on win32")
 
1197
        self.requireFeature(features.not_running_as_root)
 
1198
        os.mkdir("test-unreadable")
 
1199
        os.chmod("test-unreadable", 0000)
 
1200
        # must chmod it back so that it can be removed
 
1201
        self.addCleanup(os.chmod, "test-unreadable", 0700)
 
1202
        # The error is not raised until the generator is actually evaluated.
 
1203
        # (It would be ok if it happened earlier but at the moment it
 
1204
        # doesn't.)
 
1205
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
 
1206
        self.assertEquals('./test-unreadable', e.filename)
 
1207
        self.assertEquals(errno.EACCES, e.errno)
 
1208
        # Ensure the message contains the file name
 
1209
        self.assertContainsRe(str(e), "\./test-unreadable")
 
1210
 
 
1211
 
 
1212
    def test_walkdirs_encoding_error(self):
 
1213
        # <https://bugs.launchpad.net/bzr/+bug/488519>
 
1214
        # walkdirs didn't raise a useful message when the filenames
 
1215
        # are not using the filesystem's encoding
 
1216
 
 
1217
        # require a bytestring based filesystem
 
1218
        self.requireFeature(features.ByteStringNamedFilesystem)
 
1219
 
 
1220
        tree = [
 
1221
            '.bzr',
 
1222
            '0file',
 
1223
            '1dir/',
 
1224
            '1dir/0file',
 
1225
            '1dir/1dir/',
 
1226
            '1file'
 
1227
            ]
 
1228
 
 
1229
        self.build_tree(tree)
 
1230
 
 
1231
        # rename the 1file to a latin-1 filename
 
1232
        os.rename("./1file", "\xe8file")
 
1233
        if "\xe8file" not in os.listdir("."):
 
1234
            self.skip("Lack filesystem that preserves arbitrary bytes")
 
1235
 
 
1236
        self._save_platform_info()
 
1237
        win32utils.winver = None # Avoid the win32 detection code
 
1238
        osutils._fs_enc = 'UTF-8'
 
1239
 
 
1240
        # this should raise on error
 
1241
        def attempt():
 
1242
            for dirdetail, dirblock in osutils.walkdirs('.'):
 
1243
                pass
 
1244
 
 
1245
        self.assertRaises(errors.BadFilenameEncoding, attempt)
559
1246
 
560
1247
    def test__walkdirs_utf8(self):
561
1248
        tree = [
594
1281
            result.append((dirdetail, dirblock))
595
1282
 
596
1283
        self.assertTrue(found_bzrdir)
597
 
        self.assertEqual(expected_dirblocks,
598
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1284
        self.assertExpectedBlocks(expected_dirblocks, result)
 
1285
 
599
1286
        # you can search a subdir only, with a supplied prefix.
600
1287
        result = []
601
1288
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
602
1289
            result.append(dirblock)
603
 
        self.assertEqual(expected_dirblocks[1:],
604
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1290
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
605
1291
 
606
1292
    def _filter_out_stat(self, result):
607
1293
        """Filter out the stat value from the walkdirs result"""
612
1298
                new_dirblock.append((info[0], info[1], info[2], info[4]))
613
1299
            dirblock[:] = new_dirblock
614
1300
 
 
1301
    def _save_platform_info(self):
 
1302
        self.overrideAttr(win32utils, 'winver')
 
1303
        self.overrideAttr(osutils, '_fs_enc')
 
1304
        self.overrideAttr(osutils, '_selected_dir_reader')
 
1305
 
 
1306
    def assertDirReaderIs(self, expected):
 
1307
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
 
1308
        # Force it to redetect
 
1309
        osutils._selected_dir_reader = None
 
1310
        # Nothing to list, but should still trigger the selection logic
 
1311
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
 
1312
        self.assertIsInstance(osutils._selected_dir_reader, expected)
 
1313
 
 
1314
    def test_force_walkdirs_utf8_fs_utf8(self):
 
1315
        self.requireFeature(UTF8DirReaderFeature)
 
1316
        self._save_platform_info()
 
1317
        win32utils.winver = None # Avoid the win32 detection code
 
1318
        osutils._fs_enc = 'utf-8'
 
1319
        self.assertDirReaderIs(
 
1320
            UTF8DirReaderFeature.module.UTF8DirReader)
 
1321
 
 
1322
    def test_force_walkdirs_utf8_fs_ascii(self):
 
1323
        self.requireFeature(UTF8DirReaderFeature)
 
1324
        self._save_platform_info()
 
1325
        win32utils.winver = None # Avoid the win32 detection code
 
1326
        osutils._fs_enc = 'ascii'
 
1327
        self.assertDirReaderIs(
 
1328
            UTF8DirReaderFeature.module.UTF8DirReader)
 
1329
 
 
1330
    def test_force_walkdirs_utf8_fs_latin1(self):
 
1331
        self._save_platform_info()
 
1332
        win32utils.winver = None # Avoid the win32 detection code
 
1333
        osutils._fs_enc = 'iso-8859-1'
 
1334
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1335
 
 
1336
    def test_force_walkdirs_utf8_nt(self):
 
1337
        # Disabled because the thunk of the whole walkdirs api is disabled.
 
1338
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1339
        self._save_platform_info()
 
1340
        win32utils.winver = 'Windows NT'
 
1341
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1342
        self.assertDirReaderIs(Win32ReadDir)
 
1343
 
 
1344
    def test_force_walkdirs_utf8_98(self):
 
1345
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1346
        self._save_platform_info()
 
1347
        win32utils.winver = 'Windows 98'
 
1348
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1349
 
615
1350
    def test_unicode_walkdirs(self):
616
1351
        """Walkdirs should always return unicode paths."""
 
1352
        self.requireFeature(features.UnicodeFilenameFeature)
617
1353
        name0 = u'0file-\xb6'
618
1354
        name1 = u'1dir-\u062c\u0648'
619
1355
        name2 = u'2file-\u0633'
624
1360
            name1 + '/' + name1 + '/',
625
1361
            name2,
626
1362
            ]
627
 
        try:
628
 
            self.build_tree(tree)
629
 
        except UnicodeError:
630
 
            raise TestSkipped('Could not represent Unicode chars'
631
 
                              ' in current encoding.')
 
1363
        self.build_tree(tree)
632
1364
        expected_dirblocks = [
633
1365
                ((u'', u'.'),
634
1366
                 [(name0, name0, 'file', './' + name0),
660
1392
 
661
1393
        The abspath portion might be in unicode or utf-8
662
1394
        """
 
1395
        self.requireFeature(features.UnicodeFilenameFeature)
663
1396
        name0 = u'0file-\xb6'
664
1397
        name1 = u'1dir-\u062c\u0648'
665
1398
        name2 = u'2file-\u0633'
670
1403
            name1 + '/' + name1 + '/',
671
1404
            name2,
672
1405
            ]
673
 
        try:
674
 
            self.build_tree(tree)
675
 
        except UnicodeError:
676
 
            raise TestSkipped('Could not represent Unicode chars'
677
 
                              ' in current encoding.')
 
1406
        self.build_tree(tree)
678
1407
        name0 = name0.encode('utf8')
679
1408
        name1 = name1.encode('utf8')
680
1409
        name2 = name2.encode('utf8')
704
1433
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
705
1434
            self.assertIsInstance(dirdetail[0], str)
706
1435
            if isinstance(dirdetail[1], unicode):
707
 
                dirdetail[1] = dirdetail[1].encode('utf8')
 
1436
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
 
1437
                dirblock = [list(info) for info in dirblock]
708
1438
                for info in dirblock:
709
1439
                    self.assertIsInstance(info[4], unicode)
710
1440
                    info[4] = info[4].encode('utf8')
718
1448
            result.append((dirdetail, new_dirblock))
719
1449
        self.assertEqual(expected_dirblocks, result)
720
1450
 
721
 
    def test_unicode__walkdirs_unicode_to_utf8(self):
722
 
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
 
1451
    def test__walkdirs_utf8_with_unicode_fs(self):
 
1452
        """UnicodeDirReader should be a safe fallback everywhere
723
1453
 
724
1454
        The abspath portion should be in unicode
725
1455
        """
726
 
        name0u = u'0file-\xb6'
727
 
        name1u = u'1dir-\u062c\u0648'
728
 
        name2u = u'2file-\u0633'
729
 
        tree = [
730
 
            name0u,
731
 
            name1u + '/',
732
 
            name1u + '/' + name0u,
733
 
            name1u + '/' + name1u + '/',
734
 
            name2u,
735
 
            ]
 
1456
        self.requireFeature(features.UnicodeFilenameFeature)
 
1457
        # Use the unicode reader. TODO: split into driver-and-driven unit
 
1458
        # tests.
 
1459
        self._save_platform_info()
 
1460
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
 
1461
        name0u = u'0file-\xb6'
 
1462
        name1u = u'1dir-\u062c\u0648'
 
1463
        name2u = u'2file-\u0633'
 
1464
        tree = [
 
1465
            name0u,
 
1466
            name1u + '/',
 
1467
            name1u + '/' + name0u,
 
1468
            name1u + '/' + name1u + '/',
 
1469
            name2u,
 
1470
            ]
 
1471
        self.build_tree(tree)
 
1472
        name0 = name0u.encode('utf8')
 
1473
        name1 = name1u.encode('utf8')
 
1474
        name2 = name2u.encode('utf8')
 
1475
 
 
1476
        # All of the abspaths should be in unicode, all of the relative paths
 
1477
        # should be in utf8
 
1478
        expected_dirblocks = [
 
1479
                (('', '.'),
 
1480
                 [(name0, name0, 'file', './' + name0u),
 
1481
                  (name1, name1, 'directory', './' + name1u),
 
1482
                  (name2, name2, 'file', './' + name2u),
 
1483
                 ]
 
1484
                ),
 
1485
                ((name1, './' + name1u),
 
1486
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1487
                                                        + '/' + name0u),
 
1488
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1489
                                                            + '/' + name1u),
 
1490
                 ]
 
1491
                ),
 
1492
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1493
                 [
 
1494
                 ]
 
1495
                ),
 
1496
            ]
 
1497
        result = list(osutils._walkdirs_utf8('.'))
 
1498
        self._filter_out_stat(result)
 
1499
        self.assertEqual(expected_dirblocks, result)
 
1500
 
 
1501
    def test__walkdirs_utf8_win32readdir(self):
 
1502
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1503
        self.requireFeature(features.UnicodeFilenameFeature)
 
1504
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1505
        self._save_platform_info()
 
1506
        osutils._selected_dir_reader = Win32ReadDir()
 
1507
        name0u = u'0file-\xb6'
 
1508
        name1u = u'1dir-\u062c\u0648'
 
1509
        name2u = u'2file-\u0633'
 
1510
        tree = [
 
1511
            name0u,
 
1512
            name1u + '/',
 
1513
            name1u + '/' + name0u,
 
1514
            name1u + '/' + name1u + '/',
 
1515
            name2u,
 
1516
            ]
 
1517
        self.build_tree(tree)
 
1518
        name0 = name0u.encode('utf8')
 
1519
        name1 = name1u.encode('utf8')
 
1520
        name2 = name2u.encode('utf8')
 
1521
 
 
1522
        # All of the abspaths should be in unicode, all of the relative paths
 
1523
        # should be in utf8
 
1524
        expected_dirblocks = [
 
1525
                (('', '.'),
 
1526
                 [(name0, name0, 'file', './' + name0u),
 
1527
                  (name1, name1, 'directory', './' + name1u),
 
1528
                  (name2, name2, 'file', './' + name2u),
 
1529
                 ]
 
1530
                ),
 
1531
                ((name1, './' + name1u),
 
1532
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1533
                                                        + '/' + name0u),
 
1534
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1535
                                                            + '/' + name1u),
 
1536
                 ]
 
1537
                ),
 
1538
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1539
                 [
 
1540
                 ]
 
1541
                ),
 
1542
            ]
 
1543
        result = list(osutils._walkdirs_utf8(u'.'))
 
1544
        self._filter_out_stat(result)
 
1545
        self.assertEqual(expected_dirblocks, result)
 
1546
 
 
1547
    def assertStatIsCorrect(self, path, win32stat):
 
1548
        os_stat = os.stat(path)
 
1549
        self.assertEqual(os_stat.st_size, win32stat.st_size)
 
1550
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
 
1551
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
 
1552
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
 
1553
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
 
1554
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
 
1555
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
 
1556
 
 
1557
    def test__walkdirs_utf_win32_find_file_stat_file(self):
 
1558
        """make sure our Stat values are valid"""
 
1559
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1560
        self.requireFeature(features.UnicodeFilenameFeature)
 
1561
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1562
        name0u = u'0file-\xb6'
 
1563
        name0 = name0u.encode('utf8')
 
1564
        self.build_tree([name0u])
 
1565
        # I hate to sleep() here, but I'm trying to make the ctime different
 
1566
        # from the mtime
 
1567
        time.sleep(2)
 
1568
        f = open(name0u, 'ab')
736
1569
        try:
737
 
            self.build_tree(tree)
738
 
        except UnicodeError:
739
 
            raise TestSkipped('Could not represent Unicode chars'
740
 
                              ' in current encoding.')
 
1570
            f.write('just a small update')
 
1571
        finally:
 
1572
            f.close()
 
1573
 
 
1574
        result = Win32ReadDir().read_dir('', u'.')
 
1575
        entry = result[0]
 
1576
        self.assertEqual((name0, name0, 'file'), entry[:3])
 
1577
        self.assertEqual(u'./' + name0u, entry[4])
 
1578
        self.assertStatIsCorrect(entry[4], entry[3])
 
1579
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
 
1580
 
 
1581
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
 
1582
        """make sure our Stat values are valid"""
 
1583
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1584
        self.requireFeature(features.UnicodeFilenameFeature)
 
1585
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1586
        name0u = u'0dir-\u062c\u0648'
741
1587
        name0 = name0u.encode('utf8')
742
 
        name1 = name1u.encode('utf8')
743
 
        name2 = name2u.encode('utf8')
 
1588
        self.build_tree([name0u + '/'])
744
1589
 
745
 
        # All of the abspaths should be in unicode, all of the relative paths
746
 
        # should be in utf8
747
 
        expected_dirblocks = [
748
 
                (('', '.'),
749
 
                 [(name0, name0, 'file', './' + name0u),
750
 
                  (name1, name1, 'directory', './' + name1u),
751
 
                  (name2, name2, 'file', './' + name2u),
752
 
                 ]
753
 
                ),
754
 
                ((name1, './' + name1u),
755
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
756
 
                                                        + '/' + name0u),
757
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
758
 
                                                            + '/' + name1u),
759
 
                 ]
760
 
                ),
761
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
762
 
                 [
763
 
                 ]
764
 
                ),
765
 
            ]
766
 
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
767
 
        self._filter_out_stat(result)
768
 
        self.assertEqual(expected_dirblocks, result)
 
1590
        result = Win32ReadDir().read_dir('', u'.')
 
1591
        entry = result[0]
 
1592
        self.assertEqual((name0, name0, 'directory'), entry[:3])
 
1593
        self.assertEqual(u'./' + name0u, entry[4])
 
1594
        self.assertStatIsCorrect(entry[4], entry[3])
769
1595
 
770
1596
    def assertPathCompare(self, path_less, path_greater):
771
1597
        """check that path_less and path_greater compare correctly."""
845
1671
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
846
1672
 
847
1673
 
848
 
class TestCopyTree(TestCaseInTempDir):
849
 
    
 
1674
class TestCopyTree(tests.TestCaseInTempDir):
 
1675
 
850
1676
    def test_copy_basic_tree(self):
851
1677
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
852
1678
        osutils.copy_tree('source', 'target')
861
1687
        self.assertEqual(['c'], os.listdir('target/b'))
862
1688
 
863
1689
    def test_copy_tree_symlinks(self):
864
 
        if not osutils.has_symlinks():
865
 
            return
 
1690
        self.requireFeature(features.SymlinkFeature)
866
1691
        self.build_tree(['source/'])
867
1692
        os.symlink('a/generic/path', 'source/lnk')
868
1693
        osutils.copy_tree('source', 'target')
893
1718
                          ('d', 'source/b', 'target/b'),
894
1719
                          ('f', 'source/b/c', 'target/b/c'),
895
1720
                         ], processed_files)
896
 
        self.failIfExists('target')
 
1721
        self.assertPathDoesNotExist('target')
897
1722
        if osutils.has_symlinks():
898
1723
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
899
1724
 
900
1725
 
901
 
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
902
 
# [bialix] 2006/12/26
903
 
 
904
 
 
905
 
class TestSetUnsetEnv(TestCase):
 
1726
class TestSetUnsetEnv(tests.TestCase):
906
1727
    """Test updating the environment"""
907
1728
 
908
1729
    def setUp(self):
914
1735
        def cleanup():
915
1736
            if 'BZR_TEST_ENV_VAR' in os.environ:
916
1737
                del os.environ['BZR_TEST_ENV_VAR']
917
 
 
918
1738
        self.addCleanup(cleanup)
919
1739
 
920
1740
    def test_set(self):
932
1752
 
933
1753
    def test_unicode(self):
934
1754
        """Environment can only contain plain strings
935
 
        
 
1755
 
936
1756
        So Unicode strings must be encoded.
937
1757
        """
938
 
        # Try a few different characters, to see if we can get
939
 
        # one that will be valid in the user_encoding
940
 
        possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
941
 
        for uni_val in possible_vals:
942
 
            try:
943
 
                env_val = uni_val.encode(bzrlib.user_encoding)
944
 
            except UnicodeEncodeError:
945
 
                # Try a different character
946
 
                pass
947
 
            else:
948
 
                break
949
 
        else:
950
 
            raise TestSkipped('Cannot find a unicode character that works in'
951
 
                              ' encoding %s' % (bzrlib.user_encoding,))
 
1758
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
1759
        if uni_val is None:
 
1760
            raise tests.TestSkipped(
 
1761
                'Cannot find a unicode character that works in encoding %s'
 
1762
                % (osutils.get_user_encoding(),))
952
1763
 
953
1764
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
954
1765
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
959
1770
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
960
1771
        self.assertEqual('foo', old)
961
1772
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
962
 
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
963
 
 
964
 
 
965
 
class TestLocalTimeOffset(TestCase):
966
 
 
967
 
    def test_local_time_offset(self):
968
 
        """Test that local_time_offset() returns a sane value."""
969
 
        offset = osutils.local_time_offset()
970
 
        self.assertTrue(isinstance(offset, int))
971
 
        # Test that the offset is no more than a eighteen hours in
972
 
        # either direction.
973
 
        # Time zone handling is system specific, so it is difficult to
974
 
        # do more specific tests, but a value outside of this range is
975
 
        # probably wrong.
976
 
        eighteen_hours = 18 * 3600
977
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
978
 
 
979
 
    def test_local_time_offset_with_timestamp(self):
980
 
        """Test that local_time_offset() works with a timestamp."""
981
 
        offset = osutils.local_time_offset(1000000000.1234567)
982
 
        self.assertTrue(isinstance(offset, int))
983
 
        eighteen_hours = 18 * 3600
984
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1773
        self.assertFalse('BZR_TEST_ENV_VAR' in os.environ)
 
1774
 
 
1775
 
 
1776
class TestSizeShaFile(tests.TestCaseInTempDir):
 
1777
 
 
1778
    def test_sha_empty(self):
 
1779
        self.build_tree_contents([('foo', '')])
 
1780
        expected_sha = osutils.sha_string('')
 
1781
        f = open('foo')
 
1782
        self.addCleanup(f.close)
 
1783
        size, sha = osutils.size_sha_file(f)
 
1784
        self.assertEqual(0, size)
 
1785
        self.assertEqual(expected_sha, sha)
 
1786
 
 
1787
    def test_sha_mixed_endings(self):
 
1788
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1789
        self.build_tree_contents([('foo', text)])
 
1790
        expected_sha = osutils.sha_string(text)
 
1791
        f = open('foo', 'rb')
 
1792
        self.addCleanup(f.close)
 
1793
        size, sha = osutils.size_sha_file(f)
 
1794
        self.assertEqual(38, size)
 
1795
        self.assertEqual(expected_sha, sha)
 
1796
 
 
1797
 
 
1798
class TestShaFileByName(tests.TestCaseInTempDir):
 
1799
 
 
1800
    def test_sha_empty(self):
 
1801
        self.build_tree_contents([('foo', '')])
 
1802
        expected_sha = osutils.sha_string('')
 
1803
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1804
 
 
1805
    def test_sha_mixed_endings(self):
 
1806
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1807
        self.build_tree_contents([('foo', text)])
 
1808
        expected_sha = osutils.sha_string(text)
 
1809
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1810
 
 
1811
 
 
1812
class TestResourceLoading(tests.TestCaseInTempDir):
 
1813
 
 
1814
    def test_resource_string(self):
 
1815
        # test resource in bzrlib
 
1816
        text = osutils.resource_string('bzrlib', 'debug.py')
 
1817
        self.assertContainsRe(text, "debug_flags = set()")
 
1818
        # test resource under bzrlib
 
1819
        text = osutils.resource_string('bzrlib.ui', 'text.py')
 
1820
        self.assertContainsRe(text, "class TextUIFactory")
 
1821
        # test unsupported package
 
1822
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
 
1823
            'yyy.xx')
 
1824
        # test unknown resource
 
1825
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
 
1826
 
 
1827
 
 
1828
class TestReCompile(tests.TestCase):
 
1829
 
 
1830
    def _deprecated_re_compile_checked(self, *args, **kwargs):
 
1831
        return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
 
1832
            osutils.re_compile_checked, *args, **kwargs)
 
1833
 
 
1834
    def test_re_compile_checked(self):
 
1835
        r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
 
1836
        self.assertTrue(r.match('aaaa'))
 
1837
        self.assertTrue(r.match('aAaA'))
 
1838
 
 
1839
    def test_re_compile_checked_error(self):
 
1840
        # like https://bugs.launchpad.net/bzr/+bug/251352
 
1841
 
 
1842
        # Due to possible test isolation error, re.compile is not lazy at
 
1843
        # this point. We re-install lazy compile.
 
1844
        lazy_regex.install_lazy_compile()
 
1845
        err = self.assertRaises(
 
1846
            errors.BzrCommandError,
 
1847
            self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
 
1848
        self.assertEqual(
 
1849
            'Invalid regular expression in test case: '
 
1850
            '"*" nothing to repeat',
 
1851
            str(err))
 
1852
 
 
1853
 
 
1854
class TestDirReader(tests.TestCaseInTempDir):
 
1855
 
 
1856
    scenarios = dir_reader_scenarios()
 
1857
 
 
1858
    # Set by load_tests
 
1859
    _dir_reader_class = None
 
1860
    _native_to_unicode = None
 
1861
 
 
1862
    def setUp(self):
 
1863
        super(TestDirReader, self).setUp()
 
1864
        self.overrideAttr(osutils,
 
1865
                          '_selected_dir_reader', self._dir_reader_class())
 
1866
 
 
1867
    def _get_ascii_tree(self):
 
1868
        tree = [
 
1869
            '0file',
 
1870
            '1dir/',
 
1871
            '1dir/0file',
 
1872
            '1dir/1dir/',
 
1873
            '2file'
 
1874
            ]
 
1875
        expected_dirblocks = [
 
1876
                (('', '.'),
 
1877
                 [('0file', '0file', 'file'),
 
1878
                  ('1dir', '1dir', 'directory'),
 
1879
                  ('2file', '2file', 'file'),
 
1880
                 ]
 
1881
                ),
 
1882
                (('1dir', './1dir'),
 
1883
                 [('1dir/0file', '0file', 'file'),
 
1884
                  ('1dir/1dir', '1dir', 'directory'),
 
1885
                 ]
 
1886
                ),
 
1887
                (('1dir/1dir', './1dir/1dir'),
 
1888
                 [
 
1889
                 ]
 
1890
                ),
 
1891
            ]
 
1892
        return tree, expected_dirblocks
 
1893
 
 
1894
    def test_walk_cur_dir(self):
 
1895
        tree, expected_dirblocks = self._get_ascii_tree()
 
1896
        self.build_tree(tree)
 
1897
        result = list(osutils._walkdirs_utf8('.'))
 
1898
        # Filter out stat and abspath
 
1899
        self.assertEqual(expected_dirblocks,
 
1900
                         [(dirinfo, [line[0:3] for line in block])
 
1901
                          for dirinfo, block in result])
 
1902
 
 
1903
    def test_walk_sub_dir(self):
 
1904
        tree, expected_dirblocks = self._get_ascii_tree()
 
1905
        self.build_tree(tree)
 
1906
        # you can search a subdir only, with a supplied prefix.
 
1907
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
 
1908
        # Filter out stat and abspath
 
1909
        self.assertEqual(expected_dirblocks[1:],
 
1910
                         [(dirinfo, [line[0:3] for line in block])
 
1911
                          for dirinfo, block in result])
 
1912
 
 
1913
    def _get_unicode_tree(self):
 
1914
        name0u = u'0file-\xb6'
 
1915
        name1u = u'1dir-\u062c\u0648'
 
1916
        name2u = u'2file-\u0633'
 
1917
        tree = [
 
1918
            name0u,
 
1919
            name1u + '/',
 
1920
            name1u + '/' + name0u,
 
1921
            name1u + '/' + name1u + '/',
 
1922
            name2u,
 
1923
            ]
 
1924
        name0 = name0u.encode('UTF-8')
 
1925
        name1 = name1u.encode('UTF-8')
 
1926
        name2 = name2u.encode('UTF-8')
 
1927
        expected_dirblocks = [
 
1928
                (('', '.'),
 
1929
                 [(name0, name0, 'file', './' + name0u),
 
1930
                  (name1, name1, 'directory', './' + name1u),
 
1931
                  (name2, name2, 'file', './' + name2u),
 
1932
                 ]
 
1933
                ),
 
1934
                ((name1, './' + name1u),
 
1935
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1936
                                                        + '/' + name0u),
 
1937
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1938
                                                            + '/' + name1u),
 
1939
                 ]
 
1940
                ),
 
1941
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1942
                 [
 
1943
                 ]
 
1944
                ),
 
1945
            ]
 
1946
        return tree, expected_dirblocks
 
1947
 
 
1948
    def _filter_out(self, raw_dirblocks):
 
1949
        """Filter out a walkdirs_utf8 result.
 
1950
 
 
1951
        stat field is removed, all native paths are converted to unicode
 
1952
        """
 
1953
        filtered_dirblocks = []
 
1954
        for dirinfo, block in raw_dirblocks:
 
1955
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
 
1956
            details = []
 
1957
            for line in block:
 
1958
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
 
1959
            filtered_dirblocks.append((dirinfo, details))
 
1960
        return filtered_dirblocks
 
1961
 
 
1962
    def test_walk_unicode_tree(self):
 
1963
        self.requireFeature(features.UnicodeFilenameFeature)
 
1964
        tree, expected_dirblocks = self._get_unicode_tree()
 
1965
        self.build_tree(tree)
 
1966
        result = list(osutils._walkdirs_utf8('.'))
 
1967
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1968
 
 
1969
    def test_symlink(self):
 
1970
        self.requireFeature(features.SymlinkFeature)
 
1971
        self.requireFeature(features.UnicodeFilenameFeature)
 
1972
        target = u'target\N{Euro Sign}'
 
1973
        link_name = u'l\N{Euro Sign}nk'
 
1974
        os.symlink(target, link_name)
 
1975
        target_utf8 = target.encode('UTF-8')
 
1976
        link_name_utf8 = link_name.encode('UTF-8')
 
1977
        expected_dirblocks = [
 
1978
                (('', '.'),
 
1979
                 [(link_name_utf8, link_name_utf8,
 
1980
                   'symlink', './' + link_name),],
 
1981
                 )]
 
1982
        result = list(osutils._walkdirs_utf8('.'))
 
1983
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1984
 
 
1985
 
 
1986
class TestReadLink(tests.TestCaseInTempDir):
 
1987
    """Exposes os.readlink() problems and the osutils solution.
 
1988
 
 
1989
    The only guarantee offered by os.readlink(), starting with 2.6, is that a
 
1990
    unicode string will be returned if a unicode string is passed.
 
1991
 
 
1992
    But prior python versions failed to properly encode the passed unicode
 
1993
    string.
 
1994
    """
 
1995
    _test_needs_features = [features.SymlinkFeature, features.UnicodeFilenameFeature]
 
1996
 
 
1997
    def setUp(self):
 
1998
        super(tests.TestCaseInTempDir, self).setUp()
 
1999
        self.link = u'l\N{Euro Sign}ink'
 
2000
        self.target = u'targe\N{Euro Sign}t'
 
2001
        os.symlink(self.target, self.link)
 
2002
 
 
2003
    def test_os_readlink_link_encoding(self):
 
2004
        self.assertEquals(self.target,  os.readlink(self.link))
 
2005
 
 
2006
    def test_os_readlink_link_decoding(self):
 
2007
        self.assertEquals(self.target.encode(osutils._fs_enc),
 
2008
                          os.readlink(self.link.encode(osutils._fs_enc)))
 
2009
 
 
2010
 
 
2011
class TestConcurrency(tests.TestCase):
 
2012
 
 
2013
    def setUp(self):
 
2014
        super(TestConcurrency, self).setUp()
 
2015
        self.overrideAttr(osutils, '_cached_local_concurrency')
 
2016
 
 
2017
    def test_local_concurrency(self):
 
2018
        concurrency = osutils.local_concurrency()
 
2019
        self.assertIsInstance(concurrency, int)
 
2020
 
 
2021
    def test_local_concurrency_environment_variable(self):
 
2022
        self.overrideEnv('BZR_CONCURRENCY', '2')
 
2023
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
 
2024
        self.overrideEnv('BZR_CONCURRENCY', '3')
 
2025
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
 
2026
        self.overrideEnv('BZR_CONCURRENCY', 'foo')
 
2027
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
 
2028
 
 
2029
    def test_option_concurrency(self):
 
2030
        self.overrideEnv('BZR_CONCURRENCY', '1')
 
2031
        self.run_bzr('rocks --concurrency 42')
 
2032
        # Command line overrides environment variable
 
2033
        self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
 
2034
        self.assertEquals(42, osutils.local_concurrency(use_cache=False))
 
2035
 
 
2036
 
 
2037
class TestFailedToLoadExtension(tests.TestCase):
 
2038
 
 
2039
    def _try_loading(self):
 
2040
        try:
 
2041
            import bzrlib._fictional_extension_py
 
2042
        except ImportError, e:
 
2043
            osutils.failed_to_load_extension(e)
 
2044
            return True
 
2045
 
 
2046
    def setUp(self):
 
2047
        super(TestFailedToLoadExtension, self).setUp()
 
2048
        self.overrideAttr(osutils, '_extension_load_failures', [])
 
2049
 
 
2050
    def test_failure_to_load(self):
 
2051
        self._try_loading()
 
2052
        self.assertLength(1, osutils._extension_load_failures)
 
2053
        self.assertEquals(osutils._extension_load_failures[0],
 
2054
            "No module named _fictional_extension_py")
 
2055
 
 
2056
    def test_report_extension_load_failures_no_warning(self):
 
2057
        self.assertTrue(self._try_loading())
 
2058
        warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
 
2059
        # it used to give a Python warning; it no longer does
 
2060
        self.assertLength(0, warnings)
 
2061
 
 
2062
    def test_report_extension_load_failures_message(self):
 
2063
        log = StringIO()
 
2064
        trace.push_log_file(log)
 
2065
        self.assertTrue(self._try_loading())
 
2066
        osutils.report_extension_load_failures()
 
2067
        self.assertContainsRe(
 
2068
            log.getvalue(),
 
2069
            r"bzr: warning: some compiled extensions could not be loaded; "
 
2070
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
 
2071
            )
 
2072
 
 
2073
 
 
2074
class TestTerminalWidth(tests.TestCase):
 
2075
 
 
2076
    def setUp(self):
 
2077
        super(TestTerminalWidth, self).setUp()
 
2078
        self._orig_terminal_size_state = osutils._terminal_size_state
 
2079
        self._orig_first_terminal_size = osutils._first_terminal_size
 
2080
        self.addCleanup(self.restore_osutils_globals)
 
2081
        osutils._terminal_size_state = 'no_data'
 
2082
        osutils._first_terminal_size = None
 
2083
 
 
2084
    def restore_osutils_globals(self):
 
2085
        osutils._terminal_size_state = self._orig_terminal_size_state
 
2086
        osutils._first_terminal_size = self._orig_first_terminal_size
 
2087
 
 
2088
    def replace_stdout(self, new):
 
2089
        self.overrideAttr(sys, 'stdout', new)
 
2090
 
 
2091
    def replace__terminal_size(self, new):
 
2092
        self.overrideAttr(osutils, '_terminal_size', new)
 
2093
 
 
2094
    def set_fake_tty(self):
 
2095
 
 
2096
        class I_am_a_tty(object):
 
2097
            def isatty(self):
 
2098
                return True
 
2099
 
 
2100
        self.replace_stdout(I_am_a_tty())
 
2101
 
 
2102
    def test_default_values(self):
 
2103
        self.assertEqual(80, osutils.default_terminal_width)
 
2104
 
 
2105
    def test_defaults_to_BZR_COLUMNS(self):
 
2106
        # BZR_COLUMNS is set by the test framework
 
2107
        self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
 
2108
        self.overrideEnv('BZR_COLUMNS', '12')
 
2109
        self.assertEqual(12, osutils.terminal_width())
 
2110
 
 
2111
    def test_BZR_COLUMNS_0_no_limit(self):
 
2112
        self.overrideEnv('BZR_COLUMNS', '0')
 
2113
        self.assertEqual(None, osutils.terminal_width())
 
2114
 
 
2115
    def test_falls_back_to_COLUMNS(self):
 
2116
        self.overrideEnv('BZR_COLUMNS', None)
 
2117
        self.assertNotEqual('42', os.environ['COLUMNS'])
 
2118
        self.set_fake_tty()
 
2119
        self.overrideEnv('COLUMNS', '42')
 
2120
        self.assertEqual(42, osutils.terminal_width())
 
2121
 
 
2122
    def test_tty_default_without_columns(self):
 
2123
        self.overrideEnv('BZR_COLUMNS', None)
 
2124
        self.overrideEnv('COLUMNS', None)
 
2125
 
 
2126
        def terminal_size(w, h):
 
2127
            return 42, 42
 
2128
 
 
2129
        self.set_fake_tty()
 
2130
        # We need to override the osutils definition as it depends on the
 
2131
        # running environment that we can't control (PQM running without a
 
2132
        # controlling terminal is one example).
 
2133
        self.replace__terminal_size(terminal_size)
 
2134
        self.assertEqual(42, osutils.terminal_width())
 
2135
 
 
2136
    def test_non_tty_default_without_columns(self):
 
2137
        self.overrideEnv('BZR_COLUMNS', None)
 
2138
        self.overrideEnv('COLUMNS', None)
 
2139
        self.replace_stdout(None)
 
2140
        self.assertEqual(None, osutils.terminal_width())
 
2141
 
 
2142
    def test_no_TIOCGWINSZ(self):
 
2143
        self.requireFeature(term_ios_feature)
 
2144
        termios = term_ios_feature.module
 
2145
        # bug 63539 is about a termios without TIOCGWINSZ attribute
 
2146
        try:
 
2147
            orig = termios.TIOCGWINSZ
 
2148
        except AttributeError:
 
2149
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
 
2150
            pass
 
2151
        else:
 
2152
            self.overrideAttr(termios, 'TIOCGWINSZ')
 
2153
            del termios.TIOCGWINSZ
 
2154
        self.overrideEnv('BZR_COLUMNS', None)
 
2155
        self.overrideEnv('COLUMNS', None)
 
2156
        # Whatever the result is, if we don't raise an exception, it's ok.
 
2157
        osutils.terminal_width()
 
2158
 
 
2159
 
 
2160
class TestCreationOps(tests.TestCaseInTempDir):
 
2161
    _test_needs_features = [features.chown_feature]
 
2162
 
 
2163
    def setUp(self):
 
2164
        super(TestCreationOps, self).setUp()
 
2165
        self.overrideAttr(os, 'chown', self._dummy_chown)
 
2166
 
 
2167
        # params set by call to _dummy_chown
 
2168
        self.path = self.uid = self.gid = None
 
2169
 
 
2170
    def _dummy_chown(self, path, uid, gid):
 
2171
        self.path, self.uid, self.gid = path, uid, gid
 
2172
 
 
2173
    def test_copy_ownership_from_path(self):
 
2174
        """copy_ownership_from_path test with specified src."""
 
2175
        ownsrc = '/'
 
2176
        f = open('test_file', 'wt')
 
2177
        osutils.copy_ownership_from_path('test_file', ownsrc)
 
2178
 
 
2179
        s = os.stat(ownsrc)
 
2180
        self.assertEquals(self.path, 'test_file')
 
2181
        self.assertEquals(self.uid, s.st_uid)
 
2182
        self.assertEquals(self.gid, s.st_gid)
 
2183
 
 
2184
    def test_copy_ownership_nonesrc(self):
 
2185
        """copy_ownership_from_path test with src=None."""
 
2186
        f = open('test_file', 'wt')
 
2187
        # should use parent dir for permissions
 
2188
        osutils.copy_ownership_from_path('test_file')
 
2189
 
 
2190
        s = os.stat('..')
 
2191
        self.assertEquals(self.path, 'test_file')
 
2192
        self.assertEquals(self.uid, s.st_uid)
 
2193
        self.assertEquals(self.gid, s.st_gid)
 
2194
 
 
2195
 
 
2196
class TestPathFromEnviron(tests.TestCase):
 
2197
 
 
2198
    def test_is_unicode(self):
 
2199
        self.overrideEnv('BZR_TEST_PATH', './anywhere at all/')
 
2200
        path = osutils.path_from_environ('BZR_TEST_PATH')
 
2201
        self.assertIsInstance(path, unicode)
 
2202
        self.assertEqual(u'./anywhere at all/', path)
 
2203
 
 
2204
    def test_posix_path_env_ascii(self):
 
2205
        self.overrideEnv('BZR_TEST_PATH', '/tmp')
 
2206
        home = osutils._posix_path_from_environ('BZR_TEST_PATH')
 
2207
        self.assertIsInstance(home, unicode)
 
2208
        self.assertEqual(u'/tmp', home)
 
2209
 
 
2210
    def test_posix_path_env_unicode(self):
 
2211
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2212
        self.overrideEnv('BZR_TEST_PATH', '/home/\xa7test')
 
2213
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2214
        self.assertEqual(u'/home/\xa7test',
 
2215
            osutils._posix_path_from_environ('BZR_TEST_PATH'))
 
2216
        osutils._fs_enc = "iso8859-5"
 
2217
        self.assertEqual(u'/home/\u0407test',
 
2218
            osutils._posix_path_from_environ('BZR_TEST_PATH'))
 
2219
        osutils._fs_enc = "utf-8"
 
2220
        self.assertRaises(errors.BadFilenameEncoding,
 
2221
            osutils._posix_path_from_environ, 'BZR_TEST_PATH')
 
2222
 
 
2223
 
 
2224
class TestGetHomeDir(tests.TestCase):
 
2225
 
 
2226
    def test_is_unicode(self):
 
2227
        home = osutils._get_home_dir()
 
2228
        self.assertIsInstance(home, unicode)
 
2229
 
 
2230
    def test_posix_homeless(self):
 
2231
        self.overrideEnv('HOME', None)
 
2232
        home = osutils._get_home_dir()
 
2233
        self.assertIsInstance(home, unicode)
 
2234
 
 
2235
    def test_posix_home_ascii(self):
 
2236
        self.overrideEnv('HOME', '/home/test')
 
2237
        home = osutils._posix_get_home_dir()
 
2238
        self.assertIsInstance(home, unicode)
 
2239
        self.assertEqual(u'/home/test', home)
 
2240
 
 
2241
    def test_posix_home_unicode(self):
 
2242
        self.requireFeature(features.ByteStringNamedFilesystem)
 
2243
        self.overrideEnv('HOME', '/home/\xa7test')
 
2244
        self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
 
2245
        self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
 
2246
        osutils._fs_enc = "iso8859-5"
 
2247
        self.assertEqual(u'/home/\u0407test', osutils._posix_get_home_dir())
 
2248
        osutils._fs_enc = "utf-8"
 
2249
        self.assertRaises(errors.BadFilenameEncoding,
 
2250
            osutils._posix_get_home_dir)
 
2251
 
 
2252
 
 
2253
class TestGetuserUnicode(tests.TestCase):
 
2254
 
 
2255
    def test_is_unicode(self):
 
2256
        user = osutils.getuser_unicode()
 
2257
        self.assertIsInstance(user, unicode)
 
2258
 
 
2259
    def envvar_to_override(self):
 
2260
        if sys.platform == "win32":
 
2261
            # Disable use of platform calls on windows so envvar is used
 
2262
            self.overrideAttr(win32utils, 'has_ctypes', False)
 
2263
            return 'USERNAME' # only variable used on windows
 
2264
        return 'LOGNAME' # first variable checked by getpass.getuser()
 
2265
 
 
2266
    def test_ascii_user(self):
 
2267
        self.overrideEnv(self.envvar_to_override(), 'jrandom')
 
2268
        self.assertEqual(u'jrandom', osutils.getuser_unicode())
 
2269
 
 
2270
    def test_unicode_user(self):
 
2271
        ue = osutils.get_user_encoding()
 
2272
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
2273
        if uni_val is None:
 
2274
            raise tests.TestSkipped(
 
2275
                'Cannot find a unicode character that works in encoding %s'
 
2276
                % (osutils.get_user_encoding(),))
 
2277
        uni_username = u'jrandom' + uni_val
 
2278
        encoded_username = uni_username.encode(ue)
 
2279
        self.overrideEnv(self.envvar_to_override(), encoded_username)
 
2280
        self.assertEqual(uni_username, osutils.getuser_unicode())
 
2281
 
 
2282
 
 
2283
class TestBackupNames(tests.TestCase):
 
2284
 
 
2285
    def setUp(self):
 
2286
        super(TestBackupNames, self).setUp()
 
2287
        self.backups = []
 
2288
 
 
2289
    def backup_exists(self, name):
 
2290
        return name in self.backups
 
2291
 
 
2292
    def available_backup_name(self, name):
 
2293
        backup_name = osutils.available_backup_name(name, self.backup_exists)
 
2294
        self.backups.append(backup_name)
 
2295
        return backup_name
 
2296
 
 
2297
    def assertBackupName(self, expected, name):
 
2298
        self.assertEqual(expected, self.available_backup_name(name))
 
2299
 
 
2300
    def test_empty(self):
 
2301
        self.assertBackupName('file.~1~', 'file')
 
2302
 
 
2303
    def test_existing(self):
 
2304
        self.available_backup_name('file')
 
2305
        self.available_backup_name('file')
 
2306
        self.assertBackupName('file.~3~', 'file')
 
2307
        # Empty slots are found, this is not a strict requirement and may be
 
2308
        # revisited if we test against all implementations.
 
2309
        self.backups.remove('file.~2~')
 
2310
        self.assertBackupName('file.~2~', 'file')
 
2311
 
 
2312
 
 
2313
class TestFindExecutableInPath(tests.TestCase):
 
2314
 
 
2315
    def test_windows(self):
 
2316
        if sys.platform != 'win32':
 
2317
            raise tests.TestSkipped('test requires win32')
 
2318
        self.assertTrue(osutils.find_executable_on_path('explorer') is not None)
 
2319
        self.assertTrue(
 
2320
            osutils.find_executable_on_path('explorer.exe') is not None)
 
2321
        self.assertTrue(
 
2322
            osutils.find_executable_on_path('EXPLORER.EXE') is not None)
 
2323
        self.assertTrue(
 
2324
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2325
        self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
 
2326
        
 
2327
    def test_windows_app_path(self):
 
2328
        if sys.platform != 'win32':
 
2329
            raise tests.TestSkipped('test requires win32')
 
2330
        # Override PATH env var so that exe can only be found on App Path
 
2331
        self.overrideEnv('PATH', '')
 
2332
        # Internt Explorer is always registered in the App Path
 
2333
        self.assertTrue(osutils.find_executable_on_path('iexplore') is not None)
 
2334
 
 
2335
    def test_other(self):
 
2336
        if sys.platform == 'win32':
 
2337
            raise tests.TestSkipped('test requires non-win32')
 
2338
        self.assertTrue(osutils.find_executable_on_path('sh') is not None)
 
2339
        self.assertTrue(
 
2340
            osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
 
2341
 
 
2342
 
 
2343
class TestEnvironmentErrors(tests.TestCase):
 
2344
    """Test handling of environmental errors"""
 
2345
 
 
2346
    def test_is_oserror(self):
 
2347
        self.assertTrue(osutils.is_environment_error(
 
2348
            OSError(errno.EINVAL, "Invalid parameter")))
 
2349
 
 
2350
    def test_is_ioerror(self):
 
2351
        self.assertTrue(osutils.is_environment_error(
 
2352
            IOError(errno.EINVAL, "Invalid parameter")))
 
2353
 
 
2354
    def test_is_socket_error(self):
 
2355
        self.assertTrue(osutils.is_environment_error(
 
2356
            socket.error(errno.EINVAL, "Invalid parameter")))
 
2357
 
 
2358
    def test_is_select_error(self):
 
2359
        self.assertTrue(osutils.is_environment_error(
 
2360
            select.error(errno.EINVAL, "Invalid parameter")))
 
2361
 
 
2362
    def test_is_pywintypes_error(self):
 
2363
        self.requireFeature(features.pywintypes)
 
2364
        import pywintypes
 
2365
        self.assertTrue(osutils.is_environment_error(
 
2366
            pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))