← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Ian Clatworthy
  • Date: 2007-11-28 00:07:56 UTC
  • mto: (3054.1.1 ianc-integration)
  • mto: This revision was merged to the branch mainline in revision 3055.
  • Revision ID: ian.clatworthy@internode.on.net-20071128000756-kqm8iqmc9281roii
more cleanups and creation of tutorials directory

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
19
 
from cStringIO import StringIO
20
19
import errno
21
20
import os
22
 
import re
23
21
import socket
24
22
import stat
25
23
import sys
26
 
import time
27
24
 
 
25
import bzrlib
28
26
from bzrlib import (
29
27
    errors,
30
28
    osutils,
31
 
    tests,
32
 
    trace,
33
29
    win32utils,
34
30
    )
 
31
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
 
32
from bzrlib.osutils import (
 
33
        is_inside_any,
 
34
        is_inside_or_parent_of_any,
 
35
        pathjoin,
 
36
        )
35
37
from bzrlib.tests import (
36
 
    features,
37
 
    file_utils,
38
 
    test__walkdirs_win32,
39
 
    )
40
 
 
41
 
 
42
 
class _UTF8DirReaderFeature(tests.Feature):
43
 
 
44
 
    def _probe(self):
45
 
        try:
46
 
            from bzrlib import _readdir_pyx
47
 
            self.reader = _readdir_pyx.UTF8DirReader
48
 
            return True
49
 
        except ImportError:
50
 
            return False
51
 
 
52
 
    def feature_name(self):
53
 
        return 'bzrlib._readdir_pyx'
54
 
 
55
 
UTF8DirReaderFeature = _UTF8DirReaderFeature()
56
 
 
57
 
term_ios_feature = tests.ModuleAvailableFeature('termios')
58
 
 
59
 
 
60
 
def _already_unicode(s):
61
 
    return s
62
 
 
63
 
 
64
 
def _utf8_to_unicode(s):
65
 
    return s.decode('UTF-8')
66
 
 
67
 
 
68
 
def dir_reader_scenarios():
69
 
    # For each dir reader we define:
70
 
 
71
 
    # - native_to_unicode: a function converting the native_abspath as returned
72
 
    #   by DirReader.read_dir to its unicode representation
73
 
 
74
 
    # UnicodeDirReader is the fallback, it should be tested on all platforms.
75
 
    scenarios = [('unicode',
76
 
                  dict(_dir_reader_class=osutils.UnicodeDirReader,
77
 
                       _native_to_unicode=_already_unicode))]
78
 
    # Some DirReaders are platform specific and even there they may not be
79
 
    # available.
80
 
    if UTF8DirReaderFeature.available():
81
 
        from bzrlib import _readdir_pyx
82
 
        scenarios.append(('utf8',
83
 
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
84
 
                               _native_to_unicode=_utf8_to_unicode)))
85
 
 
86
 
    if test__walkdirs_win32.win32_readdir_feature.available():
87
 
        try:
88
 
            from bzrlib import _walkdirs_win32
89
 
            scenarios.append(
90
 
                ('win32',
91
 
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
92
 
                      _native_to_unicode=_already_unicode)))
93
 
        except ImportError:
94
 
            pass
95
 
    return scenarios
96
 
 
97
 
 
98
 
def load_tests(basic_tests, module, loader):
99
 
    suite = loader.suiteClass()
100
 
    dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
101
 
        basic_tests, tests.condition_isinstance(TestDirReader))
102
 
    tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
103
 
    suite.addTest(remaining_tests)
104
 
    return suite
105
 
 
106
 
 
107
 
class TestContainsWhitespace(tests.TestCase):
 
38
        probe_unicode_in_user_encoding,
 
39
        StringIOWrapper,
 
40
        SymlinkFeature,
 
41
        TestCase,
 
42
        TestCaseInTempDir,
 
43
        TestSkipped,
 
44
        )
 
45
 
 
46
 
 
47
class TestOSUtils(TestCaseInTempDir):
108
48
 
109
49
    def test_contains_whitespace(self):
110
50
        self.failUnless(osutils.contains_whitespace(u' '))
120
60
        self.failIf(osutils.contains_whitespace(u'hellothere'))
121
61
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
122
62
 
123
 
 
124
 
class TestRename(tests.TestCaseInTempDir):
125
 
 
126
 
    def create_file(self, filename, content):
127
 
        f = open(filename, 'wb')
128
 
        try:
129
 
            f.write(content)
130
 
        finally:
131
 
            f.close()
132
 
 
133
 
    def _fancy_rename(self, a, b):
134
 
        osutils.fancy_rename(a, b, rename_func=os.rename,
135
 
                             unlink_func=os.unlink)
136
 
 
137
63
    def test_fancy_rename(self):
138
64
        # This should work everywhere
139
 
        self.create_file('a', 'something in a\n')
140
 
        self._fancy_rename('a', 'b')
 
65
        def rename(a, b):
 
66
            osutils.fancy_rename(a, b,
 
67
                    rename_func=os.rename,
 
68
                    unlink_func=os.unlink)
 
69
 
 
70
        open('a', 'wb').write('something in a\n')
 
71
        rename('a', 'b')
141
72
        self.failIfExists('a')
142
73
        self.failUnlessExists('b')
143
74
        self.check_file_contents('b', 'something in a\n')
144
75
 
145
 
        self.create_file('a', 'new something in a\n')
146
 
        self._fancy_rename('b', 'a')
 
76
        open('a', 'wb').write('new something in a\n')
 
77
        rename('b', 'a')
147
78
 
148
79
        self.check_file_contents('a', 'something in a\n')
149
80
 
150
 
    def test_fancy_rename_fails_source_missing(self):
151
 
        # An exception should be raised, and the target should be left in place
152
 
        self.create_file('target', 'data in target\n')
153
 
        self.assertRaises((IOError, OSError), self._fancy_rename,
154
 
                          'missingsource', 'target')
155
 
        self.failUnlessExists('target')
156
 
        self.check_file_contents('target', 'data in target\n')
157
 
 
158
 
    def test_fancy_rename_fails_if_source_and_target_missing(self):
159
 
        self.assertRaises((IOError, OSError), self._fancy_rename,
160
 
                          'missingsource', 'missingtarget')
161
 
 
162
81
    def test_rename(self):
163
82
        # Rename should be semi-atomic on all platforms
164
 
        self.create_file('a', 'something in a\n')
 
83
        open('a', 'wb').write('something in a\n')
165
84
        osutils.rename('a', 'b')
166
85
        self.failIfExists('a')
167
86
        self.failUnlessExists('b')
168
87
        self.check_file_contents('b', 'something in a\n')
169
88
 
170
 
        self.create_file('a', 'new something in a\n')
 
89
        open('a', 'wb').write('new something in a\n')
171
90
        osutils.rename('b', 'a')
172
91
 
173
92
        self.check_file_contents('a', 'something in a\n')
184
103
        shape = sorted(os.listdir('.'))
185
104
        self.assertEquals(['A', 'B'], shape)
186
105
 
187
 
 
188
 
class TestRandChars(tests.TestCase):
189
 
 
190
106
    def test_01_rand_chars_empty(self):
191
107
        result = osutils.rand_chars(0)
192
108
        self.assertEqual(result, '')
197
113
        self.assertEqual(type(result), str)
198
114
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
199
115
 
200
 
 
201
 
class TestIsInside(tests.TestCase):
202
 
 
203
116
    def test_is_inside(self):
204
117
        is_inside = osutils.is_inside
205
118
        self.assertTrue(is_inside('src', 'src/foo.c'))
210
123
        self.assertTrue(is_inside('', 'foo.c'))
211
124
 
212
125
    def test_is_inside_any(self):
213
 
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
 
126
        SRC_FOO_C = pathjoin('src', 'foo.c')
214
127
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
215
128
                         (['src'], SRC_FOO_C),
216
129
                         (['src'], 'src'),
217
130
                         ]:
218
 
            self.assert_(osutils.is_inside_any(dirs, fn))
 
131
            self.assert_(is_inside_any(dirs, fn))
219
132
        for dirs, fn in [(['src'], 'srccontrol'),
220
133
                         (['src'], 'srccontrol/foo')]:
221
 
            self.assertFalse(osutils.is_inside_any(dirs, fn))
 
134
            self.assertFalse(is_inside_any(dirs, fn))
222
135
 
223
136
    def test_is_inside_or_parent_of_any(self):
224
137
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
227
140
                         (['src/bar.c', 'bla/foo.c'], 'src'),
228
141
                         (['src'], 'src'),
229
142
                         ]:
230
 
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
231
 
 
 
143
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
 
144
            
232
145
        for dirs, fn in [(['src'], 'srccontrol'),
233
146
                         (['srccontrol/foo.c'], 'src'),
234
147
                         (['src'], 'srccontrol/foo')]:
235
 
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
236
 
 
237
 
 
238
 
class TestRmTree(tests.TestCaseInTempDir):
 
148
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
239
149
 
240
150
    def test_rmtree(self):
241
151
        # Check to remove tree with read-only files/dirs
255
165
        self.failIfExists('dir/file')
256
166
        self.failIfExists('dir')
257
167
 
258
 
 
259
 
class TestDeleteAny(tests.TestCaseInTempDir):
260
 
 
261
 
    def test_delete_any_readonly(self):
262
 
        # from <https://bugs.launchpad.net/bzr/+bug/218206>
263
 
        self.build_tree(['d/', 'f'])
264
 
        osutils.make_readonly('d')
265
 
        osutils.make_readonly('f')
266
 
 
267
 
        osutils.delete_any('f')
268
 
        osutils.delete_any('d')
269
 
 
270
 
 
271
 
class TestKind(tests.TestCaseInTempDir):
272
 
 
273
168
    def test_file_kind(self):
274
169
        self.build_tree(['file', 'dir/'])
275
170
        self.assertEquals('file', osutils.file_kind('file'))
277
172
        if osutils.has_symlinks():
278
173
            os.symlink('symlink', 'symlink')
279
174
            self.assertEquals('symlink', osutils.file_kind('symlink'))
280
 
 
 
175
        
281
176
        # TODO: jam 20060529 Test a block device
282
177
        try:
283
178
            os.lstat('/dev/null')
305
200
                os.remove('socket')
306
201
 
307
202
    def test_kind_marker(self):
308
 
        self.assertEqual("", osutils.kind_marker("file"))
309
 
        self.assertEqual("/", osutils.kind_marker('directory'))
310
 
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
311
 
        self.assertEqual("@", osutils.kind_marker("symlink"))
312
 
        self.assertEqual("+", osutils.kind_marker("tree-reference"))
313
 
        self.assertEqual("", osutils.kind_marker("fifo"))
314
 
        self.assertEqual("", osutils.kind_marker("socket"))
315
 
        self.assertEqual("", osutils.kind_marker("unknown"))
316
 
 
317
 
 
318
 
class TestUmask(tests.TestCaseInTempDir):
 
203
        self.assertEqual(osutils.kind_marker('file'), '')
 
204
        self.assertEqual(osutils.kind_marker('directory'), '/')
 
205
        self.assertEqual(osutils.kind_marker('symlink'), '@')
 
206
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
319
207
 
320
208
    def test_get_umask(self):
321
209
        if sys.platform == 'win32':
324
212
            return
325
213
 
326
214
        orig_umask = osutils.get_umask()
327
 
        self.addCleanup(os.umask, orig_umask)
328
 
        os.umask(0222)
329
 
        self.assertEqual(0222, osutils.get_umask())
330
 
        os.umask(0022)
331
 
        self.assertEqual(0022, osutils.get_umask())
332
 
        os.umask(0002)
333
 
        self.assertEqual(0002, osutils.get_umask())
334
 
        os.umask(0027)
335
 
        self.assertEqual(0027, osutils.get_umask())
336
 
 
337
 
 
338
 
class TestDateTime(tests.TestCase):
 
215
        try:
 
216
            os.umask(0222)
 
217
            self.assertEqual(0222, osutils.get_umask())
 
218
            os.umask(0022)
 
219
            self.assertEqual(0022, osutils.get_umask())
 
220
            os.umask(0002)
 
221
            self.assertEqual(0002, osutils.get_umask())
 
222
            os.umask(0027)
 
223
            self.assertEqual(0027, osutils.get_umask())
 
224
        finally:
 
225
            os.umask(orig_umask)
339
226
 
340
227
    def assertFormatedDelta(self, expected, seconds):
341
228
        """Assert osutils.format_delta formats as expected"""
373
260
        self.assertFormatedDelta('1 second in the future', -1)
374
261
        self.assertFormatedDelta('2 seconds in the future', -2)
375
262
 
376
 
    def test_format_date(self):
377
 
        self.assertRaises(errors.UnsupportedTimezoneFormat,
378
 
            osutils.format_date, 0, timezone='foo')
379
 
        self.assertIsInstance(osutils.format_date(0), str)
380
 
        self.assertIsInstance(osutils.format_local_date(0), unicode)
381
 
        # Testing for the actual value of the local weekday without
382
 
        # duplicating the code from format_date is difficult.
383
 
        # Instead blackbox.test_locale should check for localized
384
 
        # dates once they do occur in output strings.
385
 
 
386
 
    def test_format_date_with_offset_in_original_timezone(self):
387
 
        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
388
 
            osutils.format_date_with_offset_in_original_timezone(0))
389
 
        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
390
 
            osutils.format_date_with_offset_in_original_timezone(100000))
391
 
        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
392
 
            osutils.format_date_with_offset_in_original_timezone(100000, 7200))
393
 
 
394
 
    def test_local_time_offset(self):
395
 
        """Test that local_time_offset() returns a sane value."""
396
 
        offset = osutils.local_time_offset()
397
 
        self.assertTrue(isinstance(offset, int))
398
 
        # Test that the offset is no more than a eighteen hours in
399
 
        # either direction.
400
 
        # Time zone handling is system specific, so it is difficult to
401
 
        # do more specific tests, but a value outside of this range is
402
 
        # probably wrong.
403
 
        eighteen_hours = 18 * 3600
404
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
405
 
 
406
 
    def test_local_time_offset_with_timestamp(self):
407
 
        """Test that local_time_offset() works with a timestamp."""
408
 
        offset = osutils.local_time_offset(1000000000.1234567)
409
 
        self.assertTrue(isinstance(offset, int))
410
 
        eighteen_hours = 18 * 3600
411
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
412
 
 
413
 
 
414
 
class TestLinks(tests.TestCaseInTempDir):
415
 
 
416
263
    def test_dereference_path(self):
417
 
        self.requireFeature(tests.SymlinkFeature)
 
264
        self.requireFeature(SymlinkFeature)
418
265
        cwd = osutils.realpath('.')
419
266
        os.mkdir('bar')
420
267
        bar_path = osutils.pathjoin(cwd, 'bar')
423
270
        self.assertEqual(bar_path, osutils.realpath('./bar'))
424
271
        os.symlink('bar', 'foo')
425
272
        self.assertEqual(bar_path, osutils.realpath('./foo'))
426
 
 
 
273
        
427
274
        # Does not dereference terminal symlinks
428
275
        foo_path = osutils.pathjoin(cwd, 'foo')
429
276
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
461
308
            osutils.make_readonly('dangling')
462
309
            osutils.make_writable('dangling')
463
310
 
464
 
    def test_host_os_dereferences_symlinks(self):
465
 
        osutils.host_os_dereferences_symlinks()
466
 
 
467
 
 
468
 
class TestCanonicalRelPath(tests.TestCaseInTempDir):
469
 
 
470
 
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
471
 
 
472
 
    def test_canonical_relpath_simple(self):
473
 
        f = file('MixedCaseName', 'w')
474
 
        f.close()
475
 
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
476
 
        self.failUnlessEqual('work/MixedCaseName', actual)
477
 
 
478
 
    def test_canonical_relpath_missing_tail(self):
479
 
        os.mkdir('MixedCaseParent')
480
 
        actual = osutils.canonical_relpath(self.test_base_dir,
481
 
                                           'mixedcaseparent/nochild')
482
 
        self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
483
 
 
484
 
 
485
 
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
486
 
 
487
 
    def assertRelpath(self, expected, base, path):
488
 
        actual = osutils._cicp_canonical_relpath(base, path)
489
 
        self.assertEqual(expected, actual)
490
 
 
491
 
    def test_simple(self):
492
 
        self.build_tree(['MixedCaseName'])
493
 
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
494
 
        self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
495
 
 
496
 
    def test_subdir_missing_tail(self):
497
 
        self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
498
 
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
499
 
        self.assertRelpath('MixedCaseParent/a_child', base,
500
 
                           'MixedCaseParent/a_child')
501
 
        self.assertRelpath('MixedCaseParent/a_child', base,
502
 
                           'MixedCaseParent/A_Child')
503
 
        self.assertRelpath('MixedCaseParent/not_child', base,
504
 
                           'MixedCaseParent/not_child')
505
 
 
506
 
    def test_at_root_slash(self):
507
 
        # We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
508
 
        # check...
509
 
        if osutils.MIN_ABS_PATHLENGTH > 1:
510
 
            raise tests.TestSkipped('relpath requires %d chars'
511
 
                                    % osutils.MIN_ABS_PATHLENGTH)
512
 
        self.assertRelpath('foo', '/', '/foo')
513
 
 
514
 
    def test_at_root_drive(self):
515
 
        if sys.platform != 'win32':
516
 
            raise tests.TestNotApplicable('we can only test drive-letter relative'
517
 
                                          ' paths on Windows where we have drive'
518
 
                                          ' letters.')
519
 
        # see bug #322807
520
 
        # The specific issue is that when at the root of a drive, 'abspath'
521
 
        # returns "C:/" or just "/". However, the code assumes that abspath
522
 
        # always returns something like "C:/foo" or "/foo" (no trailing slash).
523
 
        self.assertRelpath('foo', 'C:/', 'C:/foo')
524
 
        self.assertRelpath('foo', 'X:/', 'X:/foo')
525
 
        self.assertRelpath('foo', 'X:/', 'X://foo')
526
 
 
527
 
 
528
 
class TestPumpFile(tests.TestCase):
529
 
    """Test pumpfile method."""
530
 
 
531
 
    def setUp(self):
532
 
        tests.TestCase.setUp(self)
533
 
        # create a test datablock
534
 
        self.block_size = 512
535
 
        pattern = '0123456789ABCDEF'
536
 
        self.test_data = pattern * (3 * self.block_size / len(pattern))
537
 
        self.test_data_len = len(self.test_data)
538
 
 
539
 
    def test_bracket_block_size(self):
540
 
        """Read data in blocks with the requested read size bracketing the
541
 
        block size."""
542
 
        # make sure test data is larger than max read size
543
 
        self.assertTrue(self.test_data_len > self.block_size)
544
 
 
545
 
        from_file = file_utils.FakeReadFile(self.test_data)
546
 
        to_file = StringIO()
547
 
 
548
 
        # read (max / 2) bytes and verify read size wasn't affected
549
 
        num_bytes_to_read = self.block_size / 2
550
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
551
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
552
 
        self.assertEqual(from_file.get_read_count(), 1)
553
 
 
554
 
        # read (max) bytes and verify read size wasn't affected
555
 
        num_bytes_to_read = self.block_size
556
 
        from_file.reset_read_count()
557
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
558
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
559
 
        self.assertEqual(from_file.get_read_count(), 1)
560
 
 
561
 
        # read (max + 1) bytes and verify read size was limited
562
 
        num_bytes_to_read = self.block_size + 1
563
 
        from_file.reset_read_count()
564
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
565
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
566
 
        self.assertEqual(from_file.get_read_count(), 2)
567
 
 
568
 
        # finish reading the rest of the data
569
 
        num_bytes_to_read = self.test_data_len - to_file.tell()
570
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
571
 
 
572
 
        # report error if the data wasn't equal (we only report the size due
573
 
        # to the length of the data)
574
 
        response_data = to_file.getvalue()
575
 
        if response_data != self.test_data:
576
 
            message = "Data not equal.  Expected %d bytes, received %d."
577
 
            self.fail(message % (len(response_data), self.test_data_len))
578
 
 
579
 
    def test_specified_size(self):
580
 
        """Request a transfer larger than the maximum block size and verify
581
 
        that the maximum read doesn't exceed the block_size."""
582
 
        # make sure test data is larger than max read size
583
 
        self.assertTrue(self.test_data_len > self.block_size)
584
 
 
585
 
        # retrieve data in blocks
586
 
        from_file = file_utils.FakeReadFile(self.test_data)
587
 
        to_file = StringIO()
588
 
        osutils.pumpfile(from_file, to_file, self.test_data_len,
589
 
                         self.block_size)
590
 
 
591
 
        # verify read size was equal to the maximum read size
592
 
        self.assertTrue(from_file.get_max_read_size() > 0)
593
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
594
 
        self.assertEqual(from_file.get_read_count(), 3)
595
 
 
596
 
        # report error if the data wasn't equal (we only report the size due
597
 
        # to the length of the data)
598
 
        response_data = to_file.getvalue()
599
 
        if response_data != self.test_data:
600
 
            message = "Data not equal.  Expected %d bytes, received %d."
601
 
            self.fail(message % (len(response_data), self.test_data_len))
602
 
 
603
 
    def test_to_eof(self):
604
 
        """Read to end-of-file and verify that the reads are not larger than
605
 
        the maximum read size."""
606
 
        # make sure test data is larger than max read size
607
 
        self.assertTrue(self.test_data_len > self.block_size)
608
 
 
609
 
        # retrieve data to EOF
610
 
        from_file = file_utils.FakeReadFile(self.test_data)
611
 
        to_file = StringIO()
612
 
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
613
 
 
614
 
        # verify read size was equal to the maximum read size
615
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
616
 
        self.assertEqual(from_file.get_read_count(), 4)
617
 
 
618
 
        # report error if the data wasn't equal (we only report the size due
619
 
        # to the length of the data)
620
 
        response_data = to_file.getvalue()
621
 
        if response_data != self.test_data:
622
 
            message = "Data not equal.  Expected %d bytes, received %d."
623
 
            self.fail(message % (len(response_data), self.test_data_len))
624
 
 
625
 
    def test_defaults(self):
626
 
        """Verifies that the default arguments will read to EOF -- this
627
 
        test verifies that any existing usages of pumpfile will not be broken
628
 
        with this new version."""
629
 
        # retrieve data using default (old) pumpfile method
630
 
        from_file = file_utils.FakeReadFile(self.test_data)
631
 
        to_file = StringIO()
632
 
        osutils.pumpfile(from_file, to_file)
633
 
 
634
 
        # report error if the data wasn't equal (we only report the size due
635
 
        # to the length of the data)
636
 
        response_data = to_file.getvalue()
637
 
        if response_data != self.test_data:
638
 
            message = "Data not equal.  Expected %d bytes, received %d."
639
 
            self.fail(message % (len(response_data), self.test_data_len))
640
 
 
641
 
    def test_report_activity(self):
642
 
        activity = []
643
 
        def log_activity(length, direction):
644
 
            activity.append((length, direction))
645
 
        from_file = StringIO(self.test_data)
646
 
        to_file = StringIO()
647
 
        osutils.pumpfile(from_file, to_file, buff_size=500,
648
 
                         report_activity=log_activity, direction='read')
649
 
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
650
 
                          (36, 'read')], activity)
651
 
 
652
 
        from_file = StringIO(self.test_data)
653
 
        to_file = StringIO()
654
 
        del activity[:]
655
 
        osutils.pumpfile(from_file, to_file, buff_size=500,
656
 
                         report_activity=log_activity, direction='write')
657
 
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
658
 
                          (36, 'write')], activity)
659
 
 
660
 
        # And with a limited amount of data
661
 
        from_file = StringIO(self.test_data)
662
 
        to_file = StringIO()
663
 
        del activity[:]
664
 
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
665
 
                         report_activity=log_activity, direction='read')
666
 
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
667
 
 
668
 
 
669
 
 
670
 
class TestPumpStringFile(tests.TestCase):
671
 
 
672
 
    def test_empty(self):
673
 
        output = StringIO()
674
 
        osutils.pump_string_file("", output)
675
 
        self.assertEqual("", output.getvalue())
676
 
 
677
 
    def test_more_than_segment_size(self):
678
 
        output = StringIO()
679
 
        osutils.pump_string_file("123456789", output, 2)
680
 
        self.assertEqual("123456789", output.getvalue())
681
 
 
682
 
    def test_segment_size(self):
683
 
        output = StringIO()
684
 
        osutils.pump_string_file("12", output, 2)
685
 
        self.assertEqual("12", output.getvalue())
686
 
 
687
 
    def test_segment_size_multiple(self):
688
 
        output = StringIO()
689
 
        osutils.pump_string_file("1234", output, 2)
690
 
        self.assertEqual("1234", output.getvalue())
691
 
 
692
 
 
693
 
class TestRelpath(tests.TestCase):
694
 
 
695
 
    def test_simple_relpath(self):
696
 
        cwd = osutils.getcwd()
697
 
        subdir = cwd + '/subdir'
698
 
        self.assertEqual('subdir', osutils.relpath(cwd, subdir))
699
 
 
700
 
    def test_deep_relpath(self):
701
 
        cwd = osutils.getcwd()
702
 
        subdir = cwd + '/sub/subsubdir'
703
 
        self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
704
 
 
705
 
    def test_not_relative(self):
706
 
        self.assertRaises(errors.PathNotChild,
707
 
                          osutils.relpath, 'C:/path', 'H:/path')
708
 
        self.assertRaises(errors.PathNotChild,
709
 
                          osutils.relpath, 'C:/', 'H:/path')
710
 
 
711
 
 
712
 
class TestSafeUnicode(tests.TestCase):
 
311
    def test_kind_marker(self):
 
312
        self.assertEqual("", osutils.kind_marker("file"))
 
313
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
314
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
315
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
 
316
 
 
317
 
 
318
class TestSafeUnicode(TestCase):
713
319
 
714
320
    def test_from_ascii_string(self):
715
321
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
724
330
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
725
331
 
726
332
    def test_bad_utf8_string(self):
727
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
333
        self.assertRaises(BzrBadParameterNotUnicode,
728
334
                          osutils.safe_unicode,
729
335
                          '\xbb\xbb')
730
336
 
731
337
 
732
 
class TestSafeUtf8(tests.TestCase):
 
338
class TestSafeUtf8(TestCase):
733
339
 
734
340
    def test_from_ascii_string(self):
735
341
        f = 'foobar'
745
351
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
746
352
 
747
353
    def test_bad_utf8_string(self):
748
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
354
        self.assertRaises(BzrBadParameterNotUnicode,
749
355
                          osutils.safe_utf8, '\xbb\xbb')
750
356
 
751
357
 
752
 
class TestSafeRevisionId(tests.TestCase):
 
358
class TestSafeRevisionId(TestCase):
753
359
 
754
360
    def test_from_ascii_string(self):
755
361
        # this shouldn't give a warning because it's getting an ascii string
777
383
        self.assertEqual(None, osutils.safe_revision_id(None))
778
384
 
779
385
 
780
 
class TestSafeFileId(tests.TestCase):
 
386
class TestSafeFileId(TestCase):
781
387
 
782
388
    def test_from_ascii_string(self):
783
389
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
803
409
        self.assertEqual(None, osutils.safe_file_id(None))
804
410
 
805
411
 
806
 
class TestWin32Funcs(tests.TestCase):
807
 
    """Test that _win32 versions of os utilities return appropriate paths."""
 
412
class TestWin32Funcs(TestCase):
 
413
    """Test that the _win32 versions of os utilities return appropriate paths."""
808
414
 
809
415
    def test_abspath(self):
810
416
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
817
423
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
818
424
 
819
425
    def test_pathjoin(self):
820
 
        self.assertEqual('path/to/foo',
821
 
                         osutils._win32_pathjoin('path', 'to', 'foo'))
822
 
        self.assertEqual('C:/foo',
823
 
                         osutils._win32_pathjoin('path\\to', 'C:\\foo'))
824
 
        self.assertEqual('C:/foo',
825
 
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
826
 
        self.assertEqual('path/to/foo',
827
 
                         osutils._win32_pathjoin('path/to/', 'foo'))
828
 
        self.assertEqual('/foo',
829
 
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
830
 
        self.assertEqual('/foo',
831
 
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
426
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
 
427
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
428
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
 
429
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
 
430
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
431
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
832
432
 
833
433
    def test_normpath(self):
834
 
        self.assertEqual('path/to/foo',
835
 
                         osutils._win32_normpath(r'path\\from\..\to\.\foo'))
836
 
        self.assertEqual('path/to/foo',
837
 
                         osutils._win32_normpath('path//from/../to/./foo'))
 
434
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
435
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
838
436
 
839
437
    def test_getcwd(self):
840
438
        cwd = osutils._win32_getcwd()
869
467
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
870
468
 
871
469
 
872
 
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
 
470
class TestWin32FuncsDirs(TestCaseInTempDir):
873
471
    """Test win32 functions that create files."""
 
472
    
 
473
    def test_getcwd(self):
 
474
        if win32utils.winver == 'Windows 98':
 
475
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
 
476
        # Make sure getcwd can handle unicode filenames
 
477
        try:
 
478
            os.mkdir(u'mu-\xb5')
 
479
        except UnicodeError:
 
480
            raise TestSkipped("Unable to create Unicode filename")
874
481
 
875
 
    def test_getcwd(self):
876
 
        self.requireFeature(tests.UnicodeFilenameFeature)
877
 
        os.mkdir(u'mu-\xb5')
878
482
        os.chdir(u'mu-\xb5')
879
483
        # TODO: jam 20060427 This will probably fail on Mac OSX because
880
484
        #       it will change the normalization of B\xe5gfors
885
489
    def test_minimum_path_selection(self):
886
490
        self.assertEqual(set(),
887
491
            osutils.minimum_path_selection([]))
888
 
        self.assertEqual(set(['a']),
889
 
            osutils.minimum_path_selection(['a']))
890
492
        self.assertEqual(set(['a', 'b']),
891
493
            osutils.minimum_path_selection(['a', 'b']))
892
494
        self.assertEqual(set(['a/', 'b']),
893
495
            osutils.minimum_path_selection(['a/', 'b']))
894
496
        self.assertEqual(set(['a/', 'b']),
895
497
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
896
 
        self.assertEqual(set(['a-b', 'a', 'a0b']),
897
 
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
898
498
 
899
499
    def test_mkdtemp(self):
900
500
        tmpdir = osutils._win32_mkdtemp(dir='.')
956
556
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
957
557
 
958
558
 
959
 
class TestParentDirectories(tests.TestCaseInTempDir):
960
 
    """Test osutils.parent_directories()"""
961
 
 
962
 
    def test_parent_directories(self):
963
 
        self.assertEqual([], osutils.parent_directories('a'))
964
 
        self.assertEqual(['a'], osutils.parent_directories('a/b'))
965
 
        self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
966
 
 
967
 
 
968
 
class TestMacFuncsDirs(tests.TestCaseInTempDir):
 
559
class TestMacFuncsDirs(TestCaseInTempDir):
969
560
    """Test mac special functions that require directories."""
970
561
 
971
562
    def test_getcwd(self):
972
 
        self.requireFeature(tests.UnicodeFilenameFeature)
973
 
        os.mkdir(u'B\xe5gfors')
 
563
        # On Mac, this will actually create Ba\u030agfors
 
564
        # but chdir will still work, because it accepts both paths
 
565
        try:
 
566
            os.mkdir(u'B\xe5gfors')
 
567
        except UnicodeError:
 
568
            raise TestSkipped("Unable to create Unicode filename")
 
569
 
974
570
        os.chdir(u'B\xe5gfors')
975
571
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
976
572
 
977
573
    def test_getcwd_nonnorm(self):
978
 
        self.requireFeature(tests.UnicodeFilenameFeature)
979
574
        # Test that _mac_getcwd() will normalize this path
980
 
        os.mkdir(u'Ba\u030agfors')
 
575
        try:
 
576
            os.mkdir(u'Ba\u030agfors')
 
577
        except UnicodeError:
 
578
            raise TestSkipped("Unable to create Unicode filename")
 
579
 
981
580
        os.chdir(u'Ba\u030agfors')
982
581
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
983
582
 
984
583
 
985
 
class TestChunksToLines(tests.TestCase):
986
 
 
987
 
    def test_smoketest(self):
988
 
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
989
 
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
990
 
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
991
 
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
992
 
 
993
 
    def test_osutils_binding(self):
994
 
        from bzrlib.tests import test__chunks_to_lines
995
 
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
996
 
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
997
 
        else:
998
 
            from bzrlib._chunks_to_lines_py import chunks_to_lines
999
 
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1000
 
 
1001
 
 
1002
 
class TestSplitLines(tests.TestCase):
 
584
class TestSplitLines(TestCase):
1003
585
 
1004
586
    def test_split_unicode(self):
1005
587
        self.assertEqual([u'foo\n', u'bar\xae'],
1012
594
                         osutils.split_lines('foo\rbar\n'))
1013
595
 
1014
596
 
1015
 
class TestWalkDirs(tests.TestCaseInTempDir):
1016
 
 
1017
 
    def assertExpectedBlocks(self, expected, result):
1018
 
        self.assertEqual(expected,
1019
 
                         [(dirinfo, [line[0:3] for line in block])
1020
 
                          for dirinfo, block in result])
 
597
class TestWalkDirs(TestCaseInTempDir):
1021
598
 
1022
599
    def test_walkdirs(self):
1023
600
        tree = [
1056
633
            result.append((dirdetail, dirblock))
1057
634
 
1058
635
        self.assertTrue(found_bzrdir)
1059
 
        self.assertExpectedBlocks(expected_dirblocks, result)
 
636
        self.assertEqual(expected_dirblocks,
 
637
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1060
638
        # you can search a subdir only, with a supplied prefix.
1061
639
        result = []
1062
640
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1063
641
            result.append(dirblock)
1064
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
1065
 
 
1066
 
    def test_walkdirs_os_error(self):
1067
 
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
1068
 
        # Pyrex readdir didn't raise useful messages if it had an error
1069
 
        # reading the directory
1070
 
        if sys.platform == 'win32':
1071
 
            raise tests.TestNotApplicable(
1072
 
                "readdir IOError not tested on win32")
1073
 
        os.mkdir("test-unreadable")
1074
 
        os.chmod("test-unreadable", 0000)
1075
 
        # must chmod it back so that it can be removed
1076
 
        self.addCleanup(os.chmod, "test-unreadable", 0700)
1077
 
        # The error is not raised until the generator is actually evaluated.
1078
 
        # (It would be ok if it happened earlier but at the moment it
1079
 
        # doesn't.)
1080
 
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1081
 
        self.assertEquals('./test-unreadable', e.filename)
1082
 
        self.assertEquals(errno.EACCES, e.errno)
1083
 
        # Ensure the message contains the file name
1084
 
        self.assertContainsRe(str(e), "\./test-unreadable")
 
642
        self.assertEqual(expected_dirblocks[1:],
 
643
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1085
644
 
1086
645
    def test__walkdirs_utf8(self):
1087
646
        tree = [
1120
679
            result.append((dirdetail, dirblock))
1121
680
 
1122
681
        self.assertTrue(found_bzrdir)
1123
 
        self.assertExpectedBlocks(expected_dirblocks, result)
1124
 
 
 
682
        self.assertEqual(expected_dirblocks,
 
683
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1125
684
        # you can search a subdir only, with a supplied prefix.
1126
685
        result = []
1127
686
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1128
687
            result.append(dirblock)
1129
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
688
        self.assertEqual(expected_dirblocks[1:],
 
689
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1130
690
 
1131
691
    def _filter_out_stat(self, result):
1132
692
        """Filter out the stat value from the walkdirs result"""
1137
697
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1138
698
            dirblock[:] = new_dirblock
1139
699
 
1140
 
    def _save_platform_info(self):
1141
 
        self.overrideAttr(win32utils, 'winver')
1142
 
        self.overrideAttr(osutils, '_fs_enc')
1143
 
        self.overrideAttr(osutils, '_selected_dir_reader')
1144
 
 
1145
 
    def assertDirReaderIs(self, expected):
1146
 
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1147
 
        # Force it to redetect
1148
 
        osutils._selected_dir_reader = None
1149
 
        # Nothing to list, but should still trigger the selection logic
1150
 
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1151
 
        self.assertIsInstance(osutils._selected_dir_reader, expected)
1152
 
 
1153
 
    def test_force_walkdirs_utf8_fs_utf8(self):
1154
 
        self.requireFeature(UTF8DirReaderFeature)
1155
 
        self._save_platform_info()
1156
 
        win32utils.winver = None # Avoid the win32 detection code
1157
 
        osutils._fs_enc = 'UTF-8'
1158
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1159
 
 
1160
 
    def test_force_walkdirs_utf8_fs_ascii(self):
1161
 
        self.requireFeature(UTF8DirReaderFeature)
1162
 
        self._save_platform_info()
1163
 
        win32utils.winver = None # Avoid the win32 detection code
1164
 
        osutils._fs_enc = 'US-ASCII'
1165
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1166
 
 
1167
 
    def test_force_walkdirs_utf8_fs_ANSI(self):
1168
 
        self.requireFeature(UTF8DirReaderFeature)
1169
 
        self._save_platform_info()
1170
 
        win32utils.winver = None # Avoid the win32 detection code
1171
 
        osutils._fs_enc = 'ANSI_X3.4-1968'
1172
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1173
 
 
1174
 
    def test_force_walkdirs_utf8_fs_latin1(self):
1175
 
        self._save_platform_info()
1176
 
        win32utils.winver = None # Avoid the win32 detection code
1177
 
        osutils._fs_enc = 'latin1'
1178
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1179
 
 
1180
 
    def test_force_walkdirs_utf8_nt(self):
1181
 
        # Disabled because the thunk of the whole walkdirs api is disabled.
1182
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1183
 
        self._save_platform_info()
1184
 
        win32utils.winver = 'Windows NT'
1185
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1186
 
        self.assertDirReaderIs(Win32ReadDir)
1187
 
 
1188
 
    def test_force_walkdirs_utf8_98(self):
1189
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1190
 
        self._save_platform_info()
1191
 
        win32utils.winver = 'Windows 98'
1192
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1193
 
 
1194
700
    def test_unicode_walkdirs(self):
1195
701
        """Walkdirs should always return unicode paths."""
1196
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1197
702
        name0 = u'0file-\xb6'
1198
703
        name1 = u'1dir-\u062c\u0648'
1199
704
        name2 = u'2file-\u0633'
1204
709
            name1 + '/' + name1 + '/',
1205
710
            name2,
1206
711
            ]
1207
 
        self.build_tree(tree)
 
712
        try:
 
713
            self.build_tree(tree)
 
714
        except UnicodeError:
 
715
            raise TestSkipped('Could not represent Unicode chars'
 
716
                              ' in current encoding.')
1208
717
        expected_dirblocks = [
1209
718
                ((u'', u'.'),
1210
719
                 [(name0, name0, 'file', './' + name0),
1236
745
 
1237
746
        The abspath portion might be in unicode or utf-8
1238
747
        """
1239
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1240
748
        name0 = u'0file-\xb6'
1241
749
        name1 = u'1dir-\u062c\u0648'
1242
750
        name2 = u'2file-\u0633'
1247
755
            name1 + '/' + name1 + '/',
1248
756
            name2,
1249
757
            ]
1250
 
        self.build_tree(tree)
 
758
        try:
 
759
            self.build_tree(tree)
 
760
        except UnicodeError:
 
761
            raise TestSkipped('Could not represent Unicode chars'
 
762
                              ' in current encoding.')
1251
763
        name0 = name0.encode('utf8')
1252
764
        name1 = name1.encode('utf8')
1253
765
        name2 = name2.encode('utf8')
1292
804
            result.append((dirdetail, new_dirblock))
1293
805
        self.assertEqual(expected_dirblocks, result)
1294
806
 
1295
 
    def test__walkdirs_utf8_with_unicode_fs(self):
1296
 
        """UnicodeDirReader should be a safe fallback everywhere
 
807
    def test_unicode__walkdirs_unicode_to_utf8(self):
 
808
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
1297
809
 
1298
810
        The abspath portion should be in unicode
1299
811
        """
1300
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1301
 
        # Use the unicode reader. TODO: split into driver-and-driven unit
1302
 
        # tests.
1303
 
        self._save_platform_info()
1304
 
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
1305
 
        name0u = u'0file-\xb6'
1306
 
        name1u = u'1dir-\u062c\u0648'
1307
 
        name2u = u'2file-\u0633'
1308
 
        tree = [
1309
 
            name0u,
1310
 
            name1u + '/',
1311
 
            name1u + '/' + name0u,
1312
 
            name1u + '/' + name1u + '/',
1313
 
            name2u,
1314
 
            ]
1315
 
        self.build_tree(tree)
1316
 
        name0 = name0u.encode('utf8')
1317
 
        name1 = name1u.encode('utf8')
1318
 
        name2 = name2u.encode('utf8')
1319
 
 
1320
 
        # All of the abspaths should be in unicode, all of the relative paths
1321
 
        # should be in utf8
1322
 
        expected_dirblocks = [
1323
 
                (('', '.'),
1324
 
                 [(name0, name0, 'file', './' + name0u),
1325
 
                  (name1, name1, 'directory', './' + name1u),
1326
 
                  (name2, name2, 'file', './' + name2u),
1327
 
                 ]
1328
 
                ),
1329
 
                ((name1, './' + name1u),
1330
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1331
 
                                                        + '/' + name0u),
1332
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1333
 
                                                            + '/' + name1u),
1334
 
                 ]
1335
 
                ),
1336
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1337
 
                 [
1338
 
                 ]
1339
 
                ),
1340
 
            ]
1341
 
        result = list(osutils._walkdirs_utf8('.'))
1342
 
        self._filter_out_stat(result)
1343
 
        self.assertEqual(expected_dirblocks, result)
1344
 
 
1345
 
    def test__walkdirs_utf8_win32readdir(self):
1346
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1347
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1348
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1349
 
        self._save_platform_info()
1350
 
        osutils._selected_dir_reader = Win32ReadDir()
1351
 
        name0u = u'0file-\xb6'
1352
 
        name1u = u'1dir-\u062c\u0648'
1353
 
        name2u = u'2file-\u0633'
1354
 
        tree = [
1355
 
            name0u,
1356
 
            name1u + '/',
1357
 
            name1u + '/' + name0u,
1358
 
            name1u + '/' + name1u + '/',
1359
 
            name2u,
1360
 
            ]
1361
 
        self.build_tree(tree)
1362
 
        name0 = name0u.encode('utf8')
1363
 
        name1 = name1u.encode('utf8')
1364
 
        name2 = name2u.encode('utf8')
1365
 
 
1366
 
        # All of the abspaths should be in unicode, all of the relative paths
1367
 
        # should be in utf8
1368
 
        expected_dirblocks = [
1369
 
                (('', '.'),
1370
 
                 [(name0, name0, 'file', './' + name0u),
1371
 
                  (name1, name1, 'directory', './' + name1u),
1372
 
                  (name2, name2, 'file', './' + name2u),
1373
 
                 ]
1374
 
                ),
1375
 
                ((name1, './' + name1u),
1376
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1377
 
                                                        + '/' + name0u),
1378
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1379
 
                                                            + '/' + name1u),
1380
 
                 ]
1381
 
                ),
1382
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1383
 
                 [
1384
 
                 ]
1385
 
                ),
1386
 
            ]
1387
 
        result = list(osutils._walkdirs_utf8(u'.'))
1388
 
        self._filter_out_stat(result)
1389
 
        self.assertEqual(expected_dirblocks, result)
1390
 
 
1391
 
    def assertStatIsCorrect(self, path, win32stat):
1392
 
        os_stat = os.stat(path)
1393
 
        self.assertEqual(os_stat.st_size, win32stat.st_size)
1394
 
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1395
 
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1396
 
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1397
 
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1398
 
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1399
 
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1400
 
 
1401
 
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1402
 
        """make sure our Stat values are valid"""
1403
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1404
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1405
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1406
 
        name0u = u'0file-\xb6'
1407
 
        name0 = name0u.encode('utf8')
1408
 
        self.build_tree([name0u])
1409
 
        # I hate to sleep() here, but I'm trying to make the ctime different
1410
 
        # from the mtime
1411
 
        time.sleep(2)
1412
 
        f = open(name0u, 'ab')
 
812
        name0u = u'0file-\xb6'
 
813
        name1u = u'1dir-\u062c\u0648'
 
814
        name2u = u'2file-\u0633'
 
815
        tree = [
 
816
            name0u,
 
817
            name1u + '/',
 
818
            name1u + '/' + name0u,
 
819
            name1u + '/' + name1u + '/',
 
820
            name2u,
 
821
            ]
1413
822
        try:
1414
 
            f.write('just a small update')
1415
 
        finally:
1416
 
            f.close()
1417
 
 
1418
 
        result = Win32ReadDir().read_dir('', u'.')
1419
 
        entry = result[0]
1420
 
        self.assertEqual((name0, name0, 'file'), entry[:3])
1421
 
        self.assertEqual(u'./' + name0u, entry[4])
1422
 
        self.assertStatIsCorrect(entry[4], entry[3])
1423
 
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1424
 
 
1425
 
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1426
 
        """make sure our Stat values are valid"""
1427
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1428
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1429
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1430
 
        name0u = u'0dir-\u062c\u0648'
 
823
            self.build_tree(tree)
 
824
        except UnicodeError:
 
825
            raise TestSkipped('Could not represent Unicode chars'
 
826
                              ' in current encoding.')
1431
827
        name0 = name0u.encode('utf8')
1432
 
        self.build_tree([name0u + '/'])
 
828
        name1 = name1u.encode('utf8')
 
829
        name2 = name2u.encode('utf8')
1433
830
 
1434
 
        result = Win32ReadDir().read_dir('', u'.')
1435
 
        entry = result[0]
1436
 
        self.assertEqual((name0, name0, 'directory'), entry[:3])
1437
 
        self.assertEqual(u'./' + name0u, entry[4])
1438
 
        self.assertStatIsCorrect(entry[4], entry[3])
 
831
        # All of the abspaths should be in unicode, all of the relative paths
 
832
        # should be in utf8
 
833
        expected_dirblocks = [
 
834
                (('', '.'),
 
835
                 [(name0, name0, 'file', './' + name0u),
 
836
                  (name1, name1, 'directory', './' + name1u),
 
837
                  (name2, name2, 'file', './' + name2u),
 
838
                 ]
 
839
                ),
 
840
                ((name1, './' + name1u),
 
841
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
842
                                                        + '/' + name0u),
 
843
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
844
                                                            + '/' + name1u),
 
845
                 ]
 
846
                ),
 
847
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
848
                 [
 
849
                 ]
 
850
                ),
 
851
            ]
 
852
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
 
853
        self._filter_out_stat(result)
 
854
        self.assertEqual(expected_dirblocks, result)
1439
855
 
1440
856
    def assertPathCompare(self, path_less, path_greater):
1441
857
        """check that path_less and path_greater compare correctly."""
1515
931
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1516
932
 
1517
933
 
1518
 
class TestCopyTree(tests.TestCaseInTempDir):
1519
 
 
 
934
class TestCopyTree(TestCaseInTempDir):
 
935
    
1520
936
    def test_copy_basic_tree(self):
1521
937
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1522
938
        osutils.copy_tree('source', 'target')
1531
947
        self.assertEqual(['c'], os.listdir('target/b'))
1532
948
 
1533
949
    def test_copy_tree_symlinks(self):
1534
 
        self.requireFeature(tests.SymlinkFeature)
 
950
        self.requireFeature(SymlinkFeature)
1535
951
        self.build_tree(['source/'])
1536
952
        os.symlink('a/generic/path', 'source/lnk')
1537
953
        osutils.copy_tree('source', 'target')
1567
983
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1568
984
 
1569
985
 
1570
 
class TestSetUnsetEnv(tests.TestCase):
 
986
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
 
987
# [bialix] 2006/12/26
 
988
 
 
989
 
 
990
class TestSetUnsetEnv(TestCase):
1571
991
    """Test updating the environment"""
1572
992
 
1573
993
    def setUp(self):
1579
999
        def cleanup():
1580
1000
            if 'BZR_TEST_ENV_VAR' in os.environ:
1581
1001
                del os.environ['BZR_TEST_ENV_VAR']
 
1002
 
1582
1003
        self.addCleanup(cleanup)
1583
1004
 
1584
1005
    def test_set(self):
1596
1017
 
1597
1018
    def test_unicode(self):
1598
1019
        """Environment can only contain plain strings
1599
 
 
 
1020
        
1600
1021
        So Unicode strings must be encoded.
1601
1022
        """
1602
 
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
1023
        uni_val, env_val = probe_unicode_in_user_encoding()
1603
1024
        if uni_val is None:
1604
 
            raise tests.TestSkipped(
1605
 
                'Cannot find a unicode character that works in encoding %s'
1606
 
                % (osutils.get_user_encoding(),))
 
1025
            raise TestSkipped('Cannot find a unicode character that works in'
 
1026
                              ' encoding %s' % (bzrlib.user_encoding,))
1607
1027
 
1608
1028
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1609
1029
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1617
1037
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1618
1038
 
1619
1039
 
1620
 
class TestSizeShaFile(tests.TestCaseInTempDir):
1621
 
 
1622
 
    def test_sha_empty(self):
1623
 
        self.build_tree_contents([('foo', '')])
1624
 
        expected_sha = osutils.sha_string('')
1625
 
        f = open('foo')
1626
 
        self.addCleanup(f.close)
1627
 
        size, sha = osutils.size_sha_file(f)
1628
 
        self.assertEqual(0, size)
1629
 
        self.assertEqual(expected_sha, sha)
1630
 
 
1631
 
    def test_sha_mixed_endings(self):
1632
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1633
 
        self.build_tree_contents([('foo', text)])
1634
 
        expected_sha = osutils.sha_string(text)
1635
 
        f = open('foo', 'rb')
1636
 
        self.addCleanup(f.close)
1637
 
        size, sha = osutils.size_sha_file(f)
1638
 
        self.assertEqual(38, size)
1639
 
        self.assertEqual(expected_sha, sha)
1640
 
 
1641
 
 
1642
 
class TestShaFileByName(tests.TestCaseInTempDir):
1643
 
 
1644
 
    def test_sha_empty(self):
1645
 
        self.build_tree_contents([('foo', '')])
1646
 
        expected_sha = osutils.sha_string('')
1647
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1648
 
 
1649
 
    def test_sha_mixed_endings(self):
1650
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1651
 
        self.build_tree_contents([('foo', text)])
1652
 
        expected_sha = osutils.sha_string(text)
1653
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1654
 
 
1655
 
 
1656
 
class TestResourceLoading(tests.TestCaseInTempDir):
1657
 
 
1658
 
    def test_resource_string(self):
1659
 
        # test resource in bzrlib
1660
 
        text = osutils.resource_string('bzrlib', 'debug.py')
1661
 
        self.assertContainsRe(text, "debug_flags = set()")
1662
 
        # test resource under bzrlib
1663
 
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1664
 
        self.assertContainsRe(text, "class TextUIFactory")
1665
 
        # test unsupported package
1666
 
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1667
 
            'yyy.xx')
1668
 
        # test unknown resource
1669
 
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1670
 
 
1671
 
 
1672
 
class TestReCompile(tests.TestCase):
1673
 
 
1674
 
    def test_re_compile_checked(self):
1675
 
        r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1676
 
        self.assertTrue(r.match('aaaa'))
1677
 
        self.assertTrue(r.match('aAaA'))
1678
 
 
1679
 
    def test_re_compile_checked_error(self):
1680
 
        # like https://bugs.launchpad.net/bzr/+bug/251352
1681
 
        err = self.assertRaises(
1682
 
            errors.BzrCommandError,
1683
 
            osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1684
 
        self.assertEqual(
1685
 
            "Invalid regular expression in test case: '*': "
1686
 
            "nothing to repeat",
1687
 
            str(err))
1688
 
 
1689
 
 
1690
 
class TestDirReader(tests.TestCaseInTempDir):
1691
 
 
1692
 
    # Set by load_tests
1693
 
    _dir_reader_class = None
1694
 
    _native_to_unicode = None
1695
 
 
1696
 
    def setUp(self):
1697
 
        tests.TestCaseInTempDir.setUp(self)
1698
 
        self.overrideAttr(osutils,
1699
 
                          '_selected_dir_reader', self._dir_reader_class())
1700
 
 
1701
 
    def _get_ascii_tree(self):
1702
 
        tree = [
1703
 
            '0file',
1704
 
            '1dir/',
1705
 
            '1dir/0file',
1706
 
            '1dir/1dir/',
1707
 
            '2file'
1708
 
            ]
1709
 
        expected_dirblocks = [
1710
 
                (('', '.'),
1711
 
                 [('0file', '0file', 'file'),
1712
 
                  ('1dir', '1dir', 'directory'),
1713
 
                  ('2file', '2file', 'file'),
1714
 
                 ]
1715
 
                ),
1716
 
                (('1dir', './1dir'),
1717
 
                 [('1dir/0file', '0file', 'file'),
1718
 
                  ('1dir/1dir', '1dir', 'directory'),
1719
 
                 ]
1720
 
                ),
1721
 
                (('1dir/1dir', './1dir/1dir'),
1722
 
                 [
1723
 
                 ]
1724
 
                ),
1725
 
            ]
1726
 
        return tree, expected_dirblocks
1727
 
 
1728
 
    def test_walk_cur_dir(self):
1729
 
        tree, expected_dirblocks = self._get_ascii_tree()
1730
 
        self.build_tree(tree)
1731
 
        result = list(osutils._walkdirs_utf8('.'))
1732
 
        # Filter out stat and abspath
1733
 
        self.assertEqual(expected_dirblocks,
1734
 
                         [(dirinfo, [line[0:3] for line in block])
1735
 
                          for dirinfo, block in result])
1736
 
 
1737
 
    def test_walk_sub_dir(self):
1738
 
        tree, expected_dirblocks = self._get_ascii_tree()
1739
 
        self.build_tree(tree)
1740
 
        # you can search a subdir only, with a supplied prefix.
1741
 
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1742
 
        # Filter out stat and abspath
1743
 
        self.assertEqual(expected_dirblocks[1:],
1744
 
                         [(dirinfo, [line[0:3] for line in block])
1745
 
                          for dirinfo, block in result])
1746
 
 
1747
 
    def _get_unicode_tree(self):
1748
 
        name0u = u'0file-\xb6'
1749
 
        name1u = u'1dir-\u062c\u0648'
1750
 
        name2u = u'2file-\u0633'
1751
 
        tree = [
1752
 
            name0u,
1753
 
            name1u + '/',
1754
 
            name1u + '/' + name0u,
1755
 
            name1u + '/' + name1u + '/',
1756
 
            name2u,
1757
 
            ]
1758
 
        name0 = name0u.encode('UTF-8')
1759
 
        name1 = name1u.encode('UTF-8')
1760
 
        name2 = name2u.encode('UTF-8')
1761
 
        expected_dirblocks = [
1762
 
                (('', '.'),
1763
 
                 [(name0, name0, 'file', './' + name0u),
1764
 
                  (name1, name1, 'directory', './' + name1u),
1765
 
                  (name2, name2, 'file', './' + name2u),
1766
 
                 ]
1767
 
                ),
1768
 
                ((name1, './' + name1u),
1769
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1770
 
                                                        + '/' + name0u),
1771
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1772
 
                                                            + '/' + name1u),
1773
 
                 ]
1774
 
                ),
1775
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1776
 
                 [
1777
 
                 ]
1778
 
                ),
1779
 
            ]
1780
 
        return tree, expected_dirblocks
1781
 
 
1782
 
    def _filter_out(self, raw_dirblocks):
1783
 
        """Filter out a walkdirs_utf8 result.
1784
 
 
1785
 
        stat field is removed, all native paths are converted to unicode
1786
 
        """
1787
 
        filtered_dirblocks = []
1788
 
        for dirinfo, block in raw_dirblocks:
1789
 
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1790
 
            details = []
1791
 
            for line in block:
1792
 
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1793
 
            filtered_dirblocks.append((dirinfo, details))
1794
 
        return filtered_dirblocks
1795
 
 
1796
 
    def test_walk_unicode_tree(self):
1797
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1798
 
        tree, expected_dirblocks = self._get_unicode_tree()
1799
 
        self.build_tree(tree)
1800
 
        result = list(osutils._walkdirs_utf8('.'))
1801
 
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1802
 
 
1803
 
    def test_symlink(self):
1804
 
        self.requireFeature(tests.SymlinkFeature)
1805
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1806
 
        target = u'target\N{Euro Sign}'
1807
 
        link_name = u'l\N{Euro Sign}nk'
1808
 
        os.symlink(target, link_name)
1809
 
        target_utf8 = target.encode('UTF-8')
1810
 
        link_name_utf8 = link_name.encode('UTF-8')
1811
 
        expected_dirblocks = [
1812
 
                (('', '.'),
1813
 
                 [(link_name_utf8, link_name_utf8,
1814
 
                   'symlink', './' + link_name),],
1815
 
                 )]
1816
 
        result = list(osutils._walkdirs_utf8('.'))
1817
 
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1818
 
 
1819
 
 
1820
 
class TestReadLink(tests.TestCaseInTempDir):
1821
 
    """Exposes os.readlink() problems and the osutils solution.
1822
 
 
1823
 
    The only guarantee offered by os.readlink(), starting with 2.6, is that a
1824
 
    unicode string will be returned if a unicode string is passed.
1825
 
 
1826
 
    But prior python versions failed to properly encode the passed unicode
1827
 
    string.
1828
 
    """
1829
 
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1830
 
 
1831
 
    def setUp(self):
1832
 
        super(tests.TestCaseInTempDir, self).setUp()
1833
 
        self.link = u'l\N{Euro Sign}ink'
1834
 
        self.target = u'targe\N{Euro Sign}t'
1835
 
        os.symlink(self.target, self.link)
1836
 
 
1837
 
    def test_os_readlink_link_encoding(self):
1838
 
        if sys.version_info < (2, 6):
1839
 
            self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1840
 
        else:
1841
 
            self.assertEquals(self.target,  os.readlink(self.link))
1842
 
 
1843
 
    def test_os_readlink_link_decoding(self):
1844
 
        self.assertEquals(self.target.encode(osutils._fs_enc),
1845
 
                          os.readlink(self.link.encode(osutils._fs_enc)))
1846
 
 
1847
 
 
1848
 
class TestConcurrency(tests.TestCase):
1849
 
 
1850
 
    def setUp(self):
1851
 
        super(TestConcurrency, self).setUp()
1852
 
        self.overrideAttr(osutils, '_cached_local_concurrency')
1853
 
 
1854
 
    def test_local_concurrency(self):
1855
 
        concurrency = osutils.local_concurrency()
1856
 
        self.assertIsInstance(concurrency, int)
1857
 
 
1858
 
    def test_local_concurrency_environment_variable(self):
1859
 
        os.environ['BZR_CONCURRENCY'] = '2'
1860
 
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1861
 
        os.environ['BZR_CONCURRENCY'] = '3'
1862
 
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1863
 
        os.environ['BZR_CONCURRENCY'] = 'foo'
1864
 
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1865
 
 
1866
 
    def test_option_concurrency(self):
1867
 
        os.environ['BZR_CONCURRENCY'] = '1'
1868
 
        self.run_bzr('rocks --concurrency 42')
1869
 
        # Command line overrides envrionment variable
1870
 
        self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1871
 
        self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1872
 
 
1873
 
 
1874
 
class TestFailedToLoadExtension(tests.TestCase):
1875
 
 
1876
 
    def _try_loading(self):
1877
 
        try:
1878
 
            import bzrlib._fictional_extension_py
1879
 
        except ImportError, e:
1880
 
            osutils.failed_to_load_extension(e)
1881
 
            return True
1882
 
 
1883
 
    def setUp(self):
1884
 
        super(TestFailedToLoadExtension, self).setUp()
1885
 
        self.overrideAttr(osutils, '_extension_load_failures', [])
1886
 
 
1887
 
    def test_failure_to_load(self):
1888
 
        self._try_loading()
1889
 
        self.assertLength(1, osutils._extension_load_failures)
1890
 
        self.assertEquals(osutils._extension_load_failures[0],
1891
 
            "No module named _fictional_extension_py")
1892
 
 
1893
 
    def test_report_extension_load_failures_no_warning(self):
1894
 
        self.assertTrue(self._try_loading())
1895
 
        warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
1896
 
        # it used to give a Python warning; it no longer does
1897
 
        self.assertLength(0, warnings)
1898
 
 
1899
 
    def test_report_extension_load_failures_message(self):
1900
 
        log = StringIO()
1901
 
        trace.push_log_file(log)
1902
 
        self.assertTrue(self._try_loading())
1903
 
        osutils.report_extension_load_failures()
1904
 
        self.assertContainsRe(
1905
 
            log.getvalue(),
1906
 
            r"bzr: warning: some compiled extensions could not be loaded; "
1907
 
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1908
 
            )
1909
 
 
1910
 
 
1911
 
class TestTerminalWidth(tests.TestCase):
1912
 
 
1913
 
    def replace_stdout(self, new):
1914
 
        self.overrideAttr(sys, 'stdout', new)
1915
 
 
1916
 
    def replace__terminal_size(self, new):
1917
 
        self.overrideAttr(osutils, '_terminal_size', new)
1918
 
 
1919
 
    def set_fake_tty(self):
1920
 
 
1921
 
        class I_am_a_tty(object):
1922
 
            def isatty(self):
1923
 
                return True
1924
 
 
1925
 
        self.replace_stdout(I_am_a_tty())
1926
 
 
1927
 
    def test_default_values(self):
1928
 
        self.assertEqual(80, osutils.default_terminal_width)
1929
 
 
1930
 
    def test_defaults_to_BZR_COLUMNS(self):
1931
 
        # BZR_COLUMNS is set by the test framework
1932
 
        self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
1933
 
        os.environ['BZR_COLUMNS'] = '12'
1934
 
        self.assertEqual(12, osutils.terminal_width())
1935
 
 
1936
 
    def test_falls_back_to_COLUMNS(self):
1937
 
        del os.environ['BZR_COLUMNS']
1938
 
        self.assertNotEqual('42', os.environ['COLUMNS'])
1939
 
        self.set_fake_tty()
1940
 
        os.environ['COLUMNS'] = '42'
1941
 
        self.assertEqual(42, osutils.terminal_width())
1942
 
 
1943
 
    def test_tty_default_without_columns(self):
1944
 
        del os.environ['BZR_COLUMNS']
1945
 
        del os.environ['COLUMNS']
1946
 
 
1947
 
        def terminal_size(w, h):
1948
 
            return 42, 42
1949
 
 
1950
 
        self.set_fake_tty()
1951
 
        # We need to override the osutils definition as it depends on the
1952
 
        # running environment that we can't control (PQM running without a
1953
 
        # controlling terminal is one example).
1954
 
        self.replace__terminal_size(terminal_size)
1955
 
        self.assertEqual(42, osutils.terminal_width())
1956
 
 
1957
 
    def test_non_tty_default_without_columns(self):
1958
 
        del os.environ['BZR_COLUMNS']
1959
 
        del os.environ['COLUMNS']
1960
 
        self.replace_stdout(None)
1961
 
        self.assertEqual(None, osutils.terminal_width())
1962
 
 
1963
 
    def test_no_TIOCGWINSZ(self):
1964
 
        self.requireFeature(term_ios_feature)
1965
 
        termios = term_ios_feature.module
1966
 
        # bug 63539 is about a termios without TIOCGWINSZ attribute
1967
 
        try:
1968
 
            orig = termios.TIOCGWINSZ
1969
 
        except AttributeError:
1970
 
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
1971
 
            pass
1972
 
        else:
1973
 
            self.overrideAttr(termios, 'TIOCGWINSZ')
1974
 
            del termios.TIOCGWINSZ
1975
 
        del os.environ['BZR_COLUMNS']
1976
 
        del os.environ['COLUMNS']
1977
 
        # Whatever the result is, if we don't raise an exception, it's ok.
1978
 
        osutils.terminal_width()
1979
 
 
1980
 
class TestCreationOps(tests.TestCaseInTempDir):
1981
 
    _test_needs_features = [features.chown_feature]
1982
 
 
1983
 
    def setUp(self):
1984
 
        tests.TestCaseInTempDir.setUp(self)
1985
 
        self.overrideAttr(os, 'chown', self._dummy_chown)
1986
 
 
1987
 
        # params set by call to _dummy_chown
1988
 
        self.path = self.uid = self.gid = None
1989
 
 
1990
 
    def _dummy_chown(self, path, uid, gid):
1991
 
        self.path, self.uid, self.gid = path, uid, gid
1992
 
 
1993
 
    def test_mkdir_with_ownership_chown(self):
1994
 
        """Ensure that osutils.mkdir_with_ownership chowns correctly with ownership_src.
1995
 
        """
1996
 
        ownsrc = '/'
1997
 
        osutils.mkdir_with_ownership('foo', ownsrc)
1998
 
 
1999
 
        s = os.stat(ownsrc)
2000
 
        self.assertEquals(self.path, 'foo')
2001
 
        self.assertEquals(self.uid, s.st_uid)
2002
 
        self.assertEquals(self.gid, s.st_gid)
2003
 
 
2004
 
    def test_open_with_ownership_chown(self):
2005
 
        """Ensure that osutils.open_with_ownership chowns correctly with ownership_src.
2006
 
        """
2007
 
        ownsrc = '/'
2008
 
        f = osutils.open_with_ownership('foo', 'w', ownership_src=ownsrc)
2009
 
 
2010
 
        # do a test write and close
2011
 
        f.write('hello')
2012
 
        f.close()
2013
 
 
2014
 
        s = os.stat(ownsrc)
2015
 
        self.assertEquals(self.path, 'foo')
2016
 
        self.assertEquals(self.uid, s.st_uid)
2017
 
        self.assertEquals(self.gid, s.st_gid)
2018
 
 
 
1040
class TestLocalTimeOffset(TestCase):
 
1041
 
 
1042
    def test_local_time_offset(self):
 
1043
        """Test that local_time_offset() returns a sane value."""
 
1044
        offset = osutils.local_time_offset()
 
1045
        self.assertTrue(isinstance(offset, int))
 
1046
        # Test that the offset is no more than a eighteen hours in
 
1047
        # either direction.
 
1048
        # Time zone handling is system specific, so it is difficult to
 
1049
        # do more specific tests, but a value outside of this range is
 
1050
        # probably wrong.
 
1051
        eighteen_hours = 18 * 3600
 
1052
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1053
 
 
1054
    def test_local_time_offset_with_timestamp(self):
 
1055
        """Test that local_time_offset() works with a timestamp."""
 
1056
        offset = osutils.local_time_offset(1000000000.1234567)
 
1057
        self.assertTrue(isinstance(offset, int))
 
1058
        eighteen_hours = 18 * 3600
 
1059
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1060
 
 
1061
 
 
1062
class TestShaFileByName(TestCaseInTempDir):
 
1063
 
 
1064
    def test_sha_empty(self):
 
1065
        self.build_tree_contents([('foo', '')])
 
1066
        expected_sha = osutils.sha_string('')
 
1067
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1068
 
 
1069
    def test_sha_mixed_endings(self):
 
1070
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1071
        self.build_tree_contents([('foo', text)])
 
1072
        expected_sha = osutils.sha_string(text)
 
1073
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))