/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Martin von Gagern
  • Date: 2010-04-20 08:47:38 UTC
  • mfrom: (5167 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5195.
  • Revision ID: martin.vgagern@gmx.net-20100420084738-ygymnqmdllzrhpfn
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
1
 
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2005-2010 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
19
from cStringIO import StringIO
20
20
import errno
21
21
import os
 
22
import re
22
23
import socket
23
24
import stat
24
25
import sys
25
26
import time
26
27
 
27
 
import bzrlib
28
28
from bzrlib import (
29
29
    errors,
30
30
    osutils,
31
31
    tests,
 
32
    trace,
32
33
    win32utils,
33
34
    )
34
 
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
35
 
from bzrlib.osutils import (
36
 
        is_inside_any,
37
 
        is_inside_or_parent_of_any,
38
 
        pathjoin,
39
 
        pumpfile,
40
 
        pump_string_file,
41
 
        )
42
35
from bzrlib.tests import (
43
 
        adapt_tests,
44
 
        Feature,
45
 
        probe_unicode_in_user_encoding,
46
 
        split_suite_by_re,
47
 
        StringIOWrapper,
48
 
        SymlinkFeature,
49
 
        TestCase,
50
 
        TestCaseInTempDir,
51
 
        TestScenarioApplier,
52
 
        TestSkipped,
53
 
        )
54
 
from bzrlib.tests.file_utils import (
55
 
    FakeReadFile,
 
36
    features,
 
37
    file_utils,
 
38
    test__walkdirs_win32,
56
39
    )
57
 
from bzrlib.tests.test__walkdirs_win32 import Win32ReadDirFeature
58
 
 
59
 
 
60
 
class _UTF8DirReaderFeature(Feature):
 
40
 
 
41
 
 
42
class _UTF8DirReaderFeature(tests.Feature):
61
43
 
62
44
    def _probe(self):
63
45
        try:
72
54
 
73
55
UTF8DirReaderFeature = _UTF8DirReaderFeature()
74
56
 
75
 
 
76
 
class TestOSUtils(TestCaseInTempDir):
 
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):
77
108
 
78
109
    def test_contains_whitespace(self):
79
110
        self.failUnless(osutils.contains_whitespace(u' '))
89
120
        self.failIf(osutils.contains_whitespace(u'hellothere'))
90
121
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
91
122
 
 
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
 
92
137
    def test_fancy_rename(self):
93
138
        # This should work everywhere
94
 
        def rename(a, b):
95
 
            osutils.fancy_rename(a, b,
96
 
                    rename_func=os.rename,
97
 
                    unlink_func=os.unlink)
98
 
 
99
 
        open('a', 'wb').write('something in a\n')
100
 
        rename('a', 'b')
 
139
        self.create_file('a', 'something in a\n')
 
140
        self._fancy_rename('a', 'b')
101
141
        self.failIfExists('a')
102
142
        self.failUnlessExists('b')
103
143
        self.check_file_contents('b', 'something in a\n')
104
144
 
105
 
        open('a', 'wb').write('new something in a\n')
106
 
        rename('b', 'a')
 
145
        self.create_file('a', 'new something in a\n')
 
146
        self._fancy_rename('b', 'a')
107
147
 
108
148
        self.check_file_contents('a', 'something in a\n')
109
149
 
 
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
 
110
162
    def test_rename(self):
111
163
        # Rename should be semi-atomic on all platforms
112
 
        open('a', 'wb').write('something in a\n')
 
164
        self.create_file('a', 'something in a\n')
113
165
        osutils.rename('a', 'b')
114
166
        self.failIfExists('a')
115
167
        self.failUnlessExists('b')
116
168
        self.check_file_contents('b', 'something in a\n')
117
169
 
118
 
        open('a', 'wb').write('new something in a\n')
 
170
        self.create_file('a', 'new something in a\n')
119
171
        osutils.rename('b', 'a')
120
172
 
121
173
        self.check_file_contents('a', 'something in a\n')
132
184
        shape = sorted(os.listdir('.'))
133
185
        self.assertEquals(['A', 'B'], shape)
134
186
 
 
187
 
 
188
class TestRandChars(tests.TestCase):
 
189
 
135
190
    def test_01_rand_chars_empty(self):
136
191
        result = osutils.rand_chars(0)
137
192
        self.assertEqual(result, '')
142
197
        self.assertEqual(type(result), str)
143
198
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
144
199
 
 
200
 
 
201
class TestIsInside(tests.TestCase):
 
202
 
145
203
    def test_is_inside(self):
146
204
        is_inside = osutils.is_inside
147
205
        self.assertTrue(is_inside('src', 'src/foo.c'))
152
210
        self.assertTrue(is_inside('', 'foo.c'))
153
211
 
154
212
    def test_is_inside_any(self):
155
 
        SRC_FOO_C = pathjoin('src', 'foo.c')
 
213
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
156
214
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
157
215
                         (['src'], SRC_FOO_C),
158
216
                         (['src'], 'src'),
159
217
                         ]:
160
 
            self.assert_(is_inside_any(dirs, fn))
 
218
            self.assert_(osutils.is_inside_any(dirs, fn))
161
219
        for dirs, fn in [(['src'], 'srccontrol'),
162
220
                         (['src'], 'srccontrol/foo')]:
163
 
            self.assertFalse(is_inside_any(dirs, fn))
 
221
            self.assertFalse(osutils.is_inside_any(dirs, fn))
164
222
 
165
223
    def test_is_inside_or_parent_of_any(self):
166
224
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
169
227
                         (['src/bar.c', 'bla/foo.c'], 'src'),
170
228
                         (['src'], 'src'),
171
229
                         ]:
172
 
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
173
 
            
 
230
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
231
 
174
232
        for dirs, fn in [(['src'], 'srccontrol'),
175
233
                         (['srccontrol/foo.c'], 'src'),
176
234
                         (['src'], 'srccontrol/foo')]:
177
 
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
 
235
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
 
236
 
 
237
 
 
238
class TestRmTree(tests.TestCaseInTempDir):
178
239
 
179
240
    def test_rmtree(self):
180
241
        # Check to remove tree with read-only files/dirs
194
255
        self.failIfExists('dir/file')
195
256
        self.failIfExists('dir')
196
257
 
 
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
 
197
273
    def test_file_kind(self):
198
274
        self.build_tree(['file', 'dir/'])
199
275
        self.assertEquals('file', osutils.file_kind('file'))
201
277
        if osutils.has_symlinks():
202
278
            os.symlink('symlink', 'symlink')
203
279
            self.assertEquals('symlink', osutils.file_kind('symlink'))
204
 
        
 
280
 
205
281
        # TODO: jam 20060529 Test a block device
206
282
        try:
207
283
            os.lstat('/dev/null')
229
305
                os.remove('socket')
230
306
 
231
307
    def test_kind_marker(self):
232
 
        self.assertEqual(osutils.kind_marker('file'), '')
233
 
        self.assertEqual(osutils.kind_marker('directory'), '/')
234
 
        self.assertEqual(osutils.kind_marker('symlink'), '@')
235
 
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
 
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):
236
319
 
237
320
    def test_get_umask(self):
238
321
        if sys.platform == 'win32':
241
324
            return
242
325
 
243
326
        orig_umask = osutils.get_umask()
244
 
        try:
245
 
            os.umask(0222)
246
 
            self.assertEqual(0222, osutils.get_umask())
247
 
            os.umask(0022)
248
 
            self.assertEqual(0022, osutils.get_umask())
249
 
            os.umask(0002)
250
 
            self.assertEqual(0002, osutils.get_umask())
251
 
            os.umask(0027)
252
 
            self.assertEqual(0027, osutils.get_umask())
253
 
        finally:
254
 
            os.umask(orig_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):
255
339
 
256
340
    def assertFormatedDelta(self, expected, seconds):
257
341
        """Assert osutils.format_delta formats as expected"""
292
376
    def test_format_date(self):
293
377
        self.assertRaises(errors.UnsupportedTimezoneFormat,
294
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):
295
415
 
296
416
    def test_dereference_path(self):
297
 
        self.requireFeature(SymlinkFeature)
 
417
        self.requireFeature(tests.SymlinkFeature)
298
418
        cwd = osutils.realpath('.')
299
419
        os.mkdir('bar')
300
420
        bar_path = osutils.pathjoin(cwd, 'bar')
303
423
        self.assertEqual(bar_path, osutils.realpath('./bar'))
304
424
        os.symlink('bar', 'foo')
305
425
        self.assertEqual(bar_path, osutils.realpath('./foo'))
306
 
        
 
426
 
307
427
        # Does not dereference terminal symlinks
308
428
        foo_path = osutils.pathjoin(cwd, 'foo')
309
429
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
341
461
            osutils.make_readonly('dangling')
342
462
            osutils.make_writable('dangling')
343
463
 
344
 
    def test_kind_marker(self):
345
 
        self.assertEqual("", osutils.kind_marker("file"))
346
 
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
347
 
        self.assertEqual("@", osutils.kind_marker("symlink"))
348
 
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
349
 
 
350
464
    def test_host_os_dereferences_symlinks(self):
351
465
        osutils.host_os_dereferences_symlinks()
352
466
 
353
467
 
354
 
class TestPumpFile(TestCase):
 
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):
355
529
    """Test pumpfile method."""
 
530
 
356
531
    def setUp(self):
 
532
        tests.TestCase.setUp(self)
357
533
        # create a test datablock
358
534
        self.block_size = 512
359
535
        pattern = '0123456789ABCDEF'
366
542
        # make sure test data is larger than max read size
367
543
        self.assertTrue(self.test_data_len > self.block_size)
368
544
 
369
 
        from_file = FakeReadFile(self.test_data)
 
545
        from_file = file_utils.FakeReadFile(self.test_data)
370
546
        to_file = StringIO()
371
547
 
372
548
        # read (max / 2) bytes and verify read size wasn't affected
373
549
        num_bytes_to_read = self.block_size / 2
374
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
550
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
375
551
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
376
552
        self.assertEqual(from_file.get_read_count(), 1)
377
553
 
378
554
        # read (max) bytes and verify read size wasn't affected
379
555
        num_bytes_to_read = self.block_size
380
556
        from_file.reset_read_count()
381
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
557
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
382
558
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
383
559
        self.assertEqual(from_file.get_read_count(), 1)
384
560
 
385
561
        # read (max + 1) bytes and verify read size was limited
386
562
        num_bytes_to_read = self.block_size + 1
387
563
        from_file.reset_read_count()
388
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
564
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
389
565
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
390
566
        self.assertEqual(from_file.get_read_count(), 2)
391
567
 
392
568
        # finish reading the rest of the data
393
569
        num_bytes_to_read = self.test_data_len - to_file.tell()
394
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
570
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
395
571
 
396
572
        # report error if the data wasn't equal (we only report the size due
397
573
        # to the length of the data)
407
583
        self.assertTrue(self.test_data_len > self.block_size)
408
584
 
409
585
        # retrieve data in blocks
410
 
        from_file = FakeReadFile(self.test_data)
 
586
        from_file = file_utils.FakeReadFile(self.test_data)
411
587
        to_file = StringIO()
412
 
        pumpfile(from_file, to_file, self.test_data_len, self.block_size)
 
588
        osutils.pumpfile(from_file, to_file, self.test_data_len,
 
589
                         self.block_size)
413
590
 
414
591
        # verify read size was equal to the maximum read size
415
592
        self.assertTrue(from_file.get_max_read_size() > 0)
430
607
        self.assertTrue(self.test_data_len > self.block_size)
431
608
 
432
609
        # retrieve data to EOF
433
 
        from_file = FakeReadFile(self.test_data)
 
610
        from_file = file_utils.FakeReadFile(self.test_data)
434
611
        to_file = StringIO()
435
 
        pumpfile(from_file, to_file, -1, self.block_size)
 
612
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
436
613
 
437
614
        # verify read size was equal to the maximum read size
438
615
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
450
627
        test verifies that any existing usages of pumpfile will not be broken
451
628
        with this new version."""
452
629
        # retrieve data using default (old) pumpfile method
453
 
        from_file = FakeReadFile(self.test_data)
 
630
        from_file = file_utils.FakeReadFile(self.test_data)
454
631
        to_file = StringIO()
455
 
        pumpfile(from_file, to_file)
 
632
        osutils.pumpfile(from_file, to_file)
456
633
 
457
634
        # report error if the data wasn't equal (we only report the size due
458
635
        # to the length of the data)
461
638
            message = "Data not equal.  Expected %d bytes, received %d."
462
639
            self.fail(message % (len(response_data), self.test_data_len))
463
640
 
464
 
 
465
 
class TestPumpStringFile(TestCase):
 
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):
466
671
 
467
672
    def test_empty(self):
468
673
        output = StringIO()
469
 
        pump_string_file("", output)
 
674
        osutils.pump_string_file("", output)
470
675
        self.assertEqual("", output.getvalue())
471
676
 
472
677
    def test_more_than_segment_size(self):
473
678
        output = StringIO()
474
 
        pump_string_file("123456789", output, 2)
 
679
        osutils.pump_string_file("123456789", output, 2)
475
680
        self.assertEqual("123456789", output.getvalue())
476
681
 
477
682
    def test_segment_size(self):
478
683
        output = StringIO()
479
 
        pump_string_file("12", output, 2)
 
684
        osutils.pump_string_file("12", output, 2)
480
685
        self.assertEqual("12", output.getvalue())
481
686
 
482
687
    def test_segment_size_multiple(self):
483
688
        output = StringIO()
484
 
        pump_string_file("1234", output, 2)
 
689
        osutils.pump_string_file("1234", output, 2)
485
690
        self.assertEqual("1234", output.getvalue())
486
691
 
487
692
 
488
 
class TestSafeUnicode(TestCase):
 
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):
489
713
 
490
714
    def test_from_ascii_string(self):
491
715
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
500
724
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
501
725
 
502
726
    def test_bad_utf8_string(self):
503
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
727
        self.assertRaises(errors.BzrBadParameterNotUnicode,
504
728
                          osutils.safe_unicode,
505
729
                          '\xbb\xbb')
506
730
 
507
731
 
508
 
class TestSafeUtf8(TestCase):
 
732
class TestSafeUtf8(tests.TestCase):
509
733
 
510
734
    def test_from_ascii_string(self):
511
735
        f = 'foobar'
521
745
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
522
746
 
523
747
    def test_bad_utf8_string(self):
524
 
        self.assertRaises(BzrBadParameterNotUnicode,
 
748
        self.assertRaises(errors.BzrBadParameterNotUnicode,
525
749
                          osutils.safe_utf8, '\xbb\xbb')
526
750
 
527
751
 
528
 
class TestSafeRevisionId(TestCase):
 
752
class TestSafeRevisionId(tests.TestCase):
529
753
 
530
754
    def test_from_ascii_string(self):
531
755
        # this shouldn't give a warning because it's getting an ascii string
553
777
        self.assertEqual(None, osutils.safe_revision_id(None))
554
778
 
555
779
 
556
 
class TestSafeFileId(TestCase):
 
780
class TestSafeFileId(tests.TestCase):
557
781
 
558
782
    def test_from_ascii_string(self):
559
783
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
579
803
        self.assertEqual(None, osutils.safe_file_id(None))
580
804
 
581
805
 
582
 
class TestWin32Funcs(TestCase):
583
 
    """Test that the _win32 versions of os utilities return appropriate paths."""
 
806
class TestWin32Funcs(tests.TestCase):
 
807
    """Test that _win32 versions of os utilities return appropriate paths."""
584
808
 
585
809
    def test_abspath(self):
586
810
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
593
817
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
594
818
 
595
819
    def test_pathjoin(self):
596
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
597
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
598
 
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
599
 
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
600
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
601
 
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
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'))
602
832
 
603
833
    def test_normpath(self):
604
 
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
605
 
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
 
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'))
606
838
 
607
839
    def test_getcwd(self):
608
840
        cwd = osutils._win32_getcwd()
637
869
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
638
870
 
639
871
 
640
 
class TestWin32FuncsDirs(TestCaseInTempDir):
 
872
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
641
873
    """Test win32 functions that create files."""
642
 
    
 
874
 
643
875
    def test_getcwd(self):
644
 
        if win32utils.winver == 'Windows 98':
645
 
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
646
 
        # Make sure getcwd can handle unicode filenames
647
 
        try:
648
 
            os.mkdir(u'mu-\xb5')
649
 
        except UnicodeError:
650
 
            raise TestSkipped("Unable to create Unicode filename")
651
 
 
 
876
        self.requireFeature(tests.UnicodeFilenameFeature)
 
877
        os.mkdir(u'mu-\xb5')
652
878
        os.chdir(u'mu-\xb5')
653
879
        # TODO: jam 20060427 This will probably fail on Mac OSX because
654
880
        #       it will change the normalization of B\xe5gfors
659
885
    def test_minimum_path_selection(self):
660
886
        self.assertEqual(set(),
661
887
            osutils.minimum_path_selection([]))
 
888
        self.assertEqual(set(['a']),
 
889
            osutils.minimum_path_selection(['a']))
662
890
        self.assertEqual(set(['a', 'b']),
663
891
            osutils.minimum_path_selection(['a', 'b']))
664
892
        self.assertEqual(set(['a/', 'b']),
665
893
            osutils.minimum_path_selection(['a/', 'b']))
666
894
        self.assertEqual(set(['a/', 'b']),
667
895
            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']))
668
898
 
669
899
    def test_mkdtemp(self):
670
900
        tmpdir = osutils._win32_mkdtemp(dir='.')
726
956
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
727
957
 
728
958
 
729
 
class TestMacFuncsDirs(TestCaseInTempDir):
 
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):
730
969
    """Test mac special functions that require directories."""
731
970
 
732
971
    def test_getcwd(self):
733
 
        # On Mac, this will actually create Ba\u030agfors
734
 
        # but chdir will still work, because it accepts both paths
735
 
        try:
736
 
            os.mkdir(u'B\xe5gfors')
737
 
        except UnicodeError:
738
 
            raise TestSkipped("Unable to create Unicode filename")
739
 
 
 
972
        self.requireFeature(tests.UnicodeFilenameFeature)
 
973
        os.mkdir(u'B\xe5gfors')
740
974
        os.chdir(u'B\xe5gfors')
741
975
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
742
976
 
743
977
    def test_getcwd_nonnorm(self):
 
978
        self.requireFeature(tests.UnicodeFilenameFeature)
744
979
        # Test that _mac_getcwd() will normalize this path
745
 
        try:
746
 
            os.mkdir(u'Ba\u030agfors')
747
 
        except UnicodeError:
748
 
            raise TestSkipped("Unable to create Unicode filename")
749
 
 
 
980
        os.mkdir(u'Ba\u030agfors')
750
981
        os.chdir(u'Ba\u030agfors')
751
982
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
752
983
 
753
984
 
754
 
class TestSplitLines(TestCase):
 
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):
755
1003
 
756
1004
    def test_split_unicode(self):
757
1005
        self.assertEqual([u'foo\n', u'bar\xae'],
764
1012
                         osutils.split_lines('foo\rbar\n'))
765
1013
 
766
1014
 
767
 
class TestWalkDirs(TestCaseInTempDir):
 
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])
768
1021
 
769
1022
    def test_walkdirs(self):
770
1023
        tree = [
803
1056
            result.append((dirdetail, dirblock))
804
1057
 
805
1058
        self.assertTrue(found_bzrdir)
806
 
        self.assertEqual(expected_dirblocks,
807
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1059
        self.assertExpectedBlocks(expected_dirblocks, result)
808
1060
        # you can search a subdir only, with a supplied prefix.
809
1061
        result = []
810
1062
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
811
1063
            result.append(dirblock)
812
 
        self.assertEqual(expected_dirblocks[1:],
813
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
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")
814
1085
 
815
1086
    def test__walkdirs_utf8(self):
816
1087
        tree = [
849
1120
            result.append((dirdetail, dirblock))
850
1121
 
851
1122
        self.assertTrue(found_bzrdir)
852
 
        self.assertEqual(expected_dirblocks,
853
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1123
        self.assertExpectedBlocks(expected_dirblocks, result)
 
1124
 
854
1125
        # you can search a subdir only, with a supplied prefix.
855
1126
        result = []
856
1127
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
857
1128
            result.append(dirblock)
858
 
        self.assertEqual(expected_dirblocks[1:],
859
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
1129
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
860
1130
 
861
1131
    def _filter_out_stat(self, result):
862
1132
        """Filter out the stat value from the walkdirs result"""
868
1138
            dirblock[:] = new_dirblock
869
1139
 
870
1140
    def _save_platform_info(self):
871
 
        cur_winver = win32utils.winver
872
 
        cur_fs_enc = osutils._fs_enc
873
 
        cur_dir_reader = osutils._selected_dir_reader
874
 
        def restore():
875
 
            win32utils.winver = cur_winver
876
 
            osutils._fs_enc = cur_fs_enc
877
 
            osutils._selected_dir_reader = cur_dir_reader
878
 
        self.addCleanup(restore)
 
1141
        self.overrideAttr(win32utils, 'winver')
 
1142
        self.overrideAttr(osutils, '_fs_enc')
 
1143
        self.overrideAttr(osutils, '_selected_dir_reader')
879
1144
 
880
 
    def assertReadFSDirIs(self, expected):
 
1145
    def assertDirReaderIs(self, expected):
881
1146
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
882
1147
        # Force it to redetect
883
1148
        osutils._selected_dir_reader = None
890
1155
        self._save_platform_info()
891
1156
        win32utils.winver = None # Avoid the win32 detection code
892
1157
        osutils._fs_enc = 'UTF-8'
893
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
 
1158
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
894
1159
 
895
1160
    def test_force_walkdirs_utf8_fs_ascii(self):
896
1161
        self.requireFeature(UTF8DirReaderFeature)
897
1162
        self._save_platform_info()
898
1163
        win32utils.winver = None # Avoid the win32 detection code
899
1164
        osutils._fs_enc = 'US-ASCII'
900
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
 
1165
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
901
1166
 
902
1167
    def test_force_walkdirs_utf8_fs_ANSI(self):
903
1168
        self.requireFeature(UTF8DirReaderFeature)
904
1169
        self._save_platform_info()
905
1170
        win32utils.winver = None # Avoid the win32 detection code
906
1171
        osutils._fs_enc = 'ANSI_X3.4-1968'
907
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
 
1172
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
908
1173
 
909
1174
    def test_force_walkdirs_utf8_fs_latin1(self):
910
1175
        self._save_platform_info()
911
1176
        win32utils.winver = None # Avoid the win32 detection code
912
1177
        osutils._fs_enc = 'latin1'
913
 
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
 
1178
        self.assertDirReaderIs(osutils.UnicodeDirReader)
914
1179
 
915
1180
    def test_force_walkdirs_utf8_nt(self):
916
1181
        # Disabled because the thunk of the whole walkdirs api is disabled.
917
 
        self.requireFeature(Win32ReadDirFeature)
 
1182
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
918
1183
        self._save_platform_info()
919
1184
        win32utils.winver = 'Windows NT'
920
1185
        from bzrlib._walkdirs_win32 import Win32ReadDir
921
 
        self.assertReadFSDirIs(Win32ReadDir)
 
1186
        self.assertDirReaderIs(Win32ReadDir)
922
1187
 
923
1188
    def test_force_walkdirs_utf8_98(self):
924
 
        self.requireFeature(Win32ReadDirFeature)
 
1189
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
925
1190
        self._save_platform_info()
926
1191
        win32utils.winver = 'Windows 98'
927
 
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
 
1192
        self.assertDirReaderIs(osutils.UnicodeDirReader)
928
1193
 
929
1194
    def test_unicode_walkdirs(self):
930
1195
        """Walkdirs should always return unicode paths."""
 
1196
        self.requireFeature(tests.UnicodeFilenameFeature)
931
1197
        name0 = u'0file-\xb6'
932
1198
        name1 = u'1dir-\u062c\u0648'
933
1199
        name2 = u'2file-\u0633'
938
1204
            name1 + '/' + name1 + '/',
939
1205
            name2,
940
1206
            ]
941
 
        try:
942
 
            self.build_tree(tree)
943
 
        except UnicodeError:
944
 
            raise TestSkipped('Could not represent Unicode chars'
945
 
                              ' in current encoding.')
 
1207
        self.build_tree(tree)
946
1208
        expected_dirblocks = [
947
1209
                ((u'', u'.'),
948
1210
                 [(name0, name0, 'file', './' + name0),
974
1236
 
975
1237
        The abspath portion might be in unicode or utf-8
976
1238
        """
 
1239
        self.requireFeature(tests.UnicodeFilenameFeature)
977
1240
        name0 = u'0file-\xb6'
978
1241
        name1 = u'1dir-\u062c\u0648'
979
1242
        name2 = u'2file-\u0633'
984
1247
            name1 + '/' + name1 + '/',
985
1248
            name2,
986
1249
            ]
987
 
        try:
988
 
            self.build_tree(tree)
989
 
        except UnicodeError:
990
 
            raise TestSkipped('Could not represent Unicode chars'
991
 
                              ' in current encoding.')
 
1250
        self.build_tree(tree)
992
1251
        name0 = name0.encode('utf8')
993
1252
        name1 = name1.encode('utf8')
994
1253
        name2 = name2.encode('utf8')
1038
1297
 
1039
1298
        The abspath portion should be in unicode
1040
1299
        """
 
1300
        self.requireFeature(tests.UnicodeFilenameFeature)
1041
1301
        # Use the unicode reader. TODO: split into driver-and-driven unit
1042
1302
        # tests.
1043
1303
        self._save_platform_info()
1052
1312
            name1u + '/' + name1u + '/',
1053
1313
            name2u,
1054
1314
            ]
1055
 
        try:
1056
 
            self.build_tree(tree)
1057
 
        except UnicodeError:
1058
 
            raise TestSkipped('Could not represent Unicode chars'
1059
 
                              ' in current encoding.')
 
1315
        self.build_tree(tree)
1060
1316
        name0 = name0u.encode('utf8')
1061
1317
        name1 = name1u.encode('utf8')
1062
1318
        name2 = name2u.encode('utf8')
1087
1343
        self.assertEqual(expected_dirblocks, result)
1088
1344
 
1089
1345
    def test__walkdirs_utf8_win32readdir(self):
1090
 
        self.requireFeature(Win32ReadDirFeature)
 
1346
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1091
1347
        self.requireFeature(tests.UnicodeFilenameFeature)
1092
1348
        from bzrlib._walkdirs_win32 import Win32ReadDir
1093
1349
        self._save_platform_info()
1144
1400
 
1145
1401
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1146
1402
        """make sure our Stat values are valid"""
1147
 
        self.requireFeature(Win32ReadDirFeature)
 
1403
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1148
1404
        self.requireFeature(tests.UnicodeFilenameFeature)
1149
1405
        from bzrlib._walkdirs_win32 import Win32ReadDir
1150
1406
        name0u = u'0file-\xb6'
1168
1424
 
1169
1425
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1170
1426
        """make sure our Stat values are valid"""
1171
 
        self.requireFeature(Win32ReadDirFeature)
 
1427
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1172
1428
        self.requireFeature(tests.UnicodeFilenameFeature)
1173
1429
        from bzrlib._walkdirs_win32 import Win32ReadDir
1174
1430
        name0u = u'0dir-\u062c\u0648'
1259
1515
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1260
1516
 
1261
1517
 
1262
 
class TestCopyTree(TestCaseInTempDir):
1263
 
    
 
1518
class TestCopyTree(tests.TestCaseInTempDir):
 
1519
 
1264
1520
    def test_copy_basic_tree(self):
1265
1521
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1266
1522
        osutils.copy_tree('source', 'target')
1275
1531
        self.assertEqual(['c'], os.listdir('target/b'))
1276
1532
 
1277
1533
    def test_copy_tree_symlinks(self):
1278
 
        self.requireFeature(SymlinkFeature)
 
1534
        self.requireFeature(tests.SymlinkFeature)
1279
1535
        self.build_tree(['source/'])
1280
1536
        os.symlink('a/generic/path', 'source/lnk')
1281
1537
        osutils.copy_tree('source', 'target')
1311
1567
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1312
1568
 
1313
1569
 
1314
 
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
1315
 
# [bialix] 2006/12/26
1316
 
 
1317
 
 
1318
 
class TestSetUnsetEnv(TestCase):
 
1570
class TestSetUnsetEnv(tests.TestCase):
1319
1571
    """Test updating the environment"""
1320
1572
 
1321
1573
    def setUp(self):
1327
1579
        def cleanup():
1328
1580
            if 'BZR_TEST_ENV_VAR' in os.environ:
1329
1581
                del os.environ['BZR_TEST_ENV_VAR']
1330
 
 
1331
1582
        self.addCleanup(cleanup)
1332
1583
 
1333
1584
    def test_set(self):
1345
1596
 
1346
1597
    def test_unicode(self):
1347
1598
        """Environment can only contain plain strings
1348
 
        
 
1599
 
1349
1600
        So Unicode strings must be encoded.
1350
1601
        """
1351
 
        uni_val, env_val = probe_unicode_in_user_encoding()
 
1602
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
1352
1603
        if uni_val is None:
1353
 
            raise TestSkipped('Cannot find a unicode character that works in'
1354
 
                              ' encoding %s' % (bzrlib.user_encoding,))
 
1604
            raise tests.TestSkipped(
 
1605
                'Cannot find a unicode character that works in encoding %s'
 
1606
                % (osutils.get_user_encoding(),))
1355
1607
 
1356
1608
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1357
1609
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1365
1617
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1366
1618
 
1367
1619
 
1368
 
class TestLocalTimeOffset(TestCase):
1369
 
 
1370
 
    def test_local_time_offset(self):
1371
 
        """Test that local_time_offset() returns a sane value."""
1372
 
        offset = osutils.local_time_offset()
1373
 
        self.assertTrue(isinstance(offset, int))
1374
 
        # Test that the offset is no more than a eighteen hours in
1375
 
        # either direction.
1376
 
        # Time zone handling is system specific, so it is difficult to
1377
 
        # do more specific tests, but a value outside of this range is
1378
 
        # probably wrong.
1379
 
        eighteen_hours = 18 * 3600
1380
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1381
 
 
1382
 
    def test_local_time_offset_with_timestamp(self):
1383
 
        """Test that local_time_offset() works with a timestamp."""
1384
 
        offset = osutils.local_time_offset(1000000000.1234567)
1385
 
        self.assertTrue(isinstance(offset, int))
1386
 
        eighteen_hours = 18 * 3600
1387
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1388
 
 
1389
 
 
1390
 
class TestShaFileByName(TestCaseInTempDir):
1391
 
 
1392
 
    def test_sha_empty(self):
1393
 
        self.build_tree_contents([('foo', '')])
1394
 
        expected_sha = osutils.sha_string('')
1395
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1396
 
 
1397
 
    def test_sha_mixed_endings(self):
1398
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1399
 
        self.build_tree_contents([('foo', text)])
1400
 
        expected_sha = osutils.sha_string(text)
1401
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1402
 
 
1403
 
 
1404
 
_debug_text = \
1405
 
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1406
 
#
1407
 
# This program is free software; you can redistribute it and/or modify
1408
 
# it under the terms of the GNU General Public License as published by
1409
 
# the Free Software Foundation; either version 2 of the License, or
1410
 
# (at your option) any later version.
1411
 
#
1412
 
# This program is distributed in the hope that it will be useful,
1413
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1414
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
1415
 
# GNU General Public License for more details.
1416
 
#
1417
 
# You should have received a copy of the GNU General Public License
1418
 
# along with this program; if not, write to the Free Software
1419
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
1420
 
 
1421
 
 
1422
 
# NOTE: If update these, please also update the help for global-options in
1423
 
#       bzrlib/help_topics/__init__.py
1424
 
 
1425
 
debug_flags = set()
1426
 
"""Set of flags that enable different debug behaviour.
1427
 
 
1428
 
These are set with eg ``-Dlock`` on the bzr command line.
1429
 
 
1430
 
Options include:
1431
 
 
1432
 
 * auth - show authentication sections used
1433
 
 * error - show stack traces for all top level exceptions
1434
 
 * evil - capture call sites that do expensive or badly-scaling operations.
1435
 
 * fetch - trace history copying between repositories
1436
 
 * graph - trace graph traversal information
1437
 
 * hashcache - log every time a working file is read to determine its hash
1438
 
 * hooks - trace hook execution
1439
 
 * hpss - trace smart protocol requests and responses
1440
 
 * http - trace http connections, requests and responses
1441
 
 * index - trace major index operations
1442
 
 * knit - trace knit operations
1443
 
 * lock - trace when lockdir locks are taken or released
1444
 
 * merge - emit information for debugging merges
1445
 
 * pack - emit information about pack operations
1446
 
 
1447
 
"""
1448
 
'''
1449
 
 
1450
 
 
1451
 
class TestResourceLoading(TestCaseInTempDir):
 
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):
1452
1657
 
1453
1658
    def test_resource_string(self):
1454
1659
        # test resource in bzrlib
1455
1660
        text = osutils.resource_string('bzrlib', 'debug.py')
1456
 
        self.assertEquals(_debug_text, text)
 
1661
        self.assertContainsRe(text, "debug_flags = set()")
1457
1662
        # test resource under bzrlib
1458
1663
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1459
1664
        self.assertContainsRe(text, "class TextUIFactory")
1462
1667
            'yyy.xx')
1463
1668
        # test unknown resource
1464
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