← 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: John Arbash Meinel
  • Date: 2009-03-27 22:29:55 UTC
  • mto: (3735.39.2 clean)
  • mto: This revision was merged to the branch mainline in revision 4280.
  • Revision ID: john@arbash-meinel.com-20090327222955-utifmfm888zerixt
Implement apply_delta_to_source which doesn't have to malloc another string.

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, 2008, 2009 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
29
29
    errors,
30
30
    osutils,
31
31
    tests,
32
 
    trace,
33
32
    win32utils,
34
33
    )
 
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
        canonical_relpath,
 
42
        )
35
43
from bzrlib.tests import (
36
 
    features,
37
 
    file_utils,
38
 
    test__walkdirs_win32,
 
44
        Feature,
 
45
        probe_unicode_in_user_encoding,
 
46
        StringIOWrapper,
 
47
        SymlinkFeature,
 
48
        CaseInsCasePresFilenameFeature,
 
49
        TestCase,
 
50
        TestCaseInTempDir,
 
51
        TestSkipped,
 
52
        )
 
53
from bzrlib.tests.file_utils import (
 
54
    FakeReadFile,
39
55
    )
40
 
 
41
 
 
42
 
class _UTF8DirReaderFeature(tests.Feature):
 
56
from bzrlib.tests.test__walkdirs_win32 import Win32ReadDirFeature
 
57
 
 
58
 
 
59
class _UTF8DirReaderFeature(Feature):
43
60
 
44
61
    def _probe(self):
45
62
        try:
54
71
 
55
72
UTF8DirReaderFeature = _UTF8DirReaderFeature()
56
73
 
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):
 
74
 
 
75
class TestOSUtils(TestCaseInTempDir):
108
76
 
109
77
    def test_contains_whitespace(self):
110
78
        self.failUnless(osutils.contains_whitespace(u' '))
120
88
        self.failIf(osutils.contains_whitespace(u'hellothere'))
121
89
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
122
90
 
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
91
    def test_fancy_rename(self):
138
92
        # This should work everywhere
139
 
        self.create_file('a', 'something in a\n')
140
 
        self._fancy_rename('a', 'b')
 
93
        def rename(a, b):
 
94
            osutils.fancy_rename(a, b,
 
95
                    rename_func=os.rename,
 
96
                    unlink_func=os.unlink)
 
97
 
 
98
        open('a', 'wb').write('something in a\n')
 
99
        rename('a', 'b')
141
100
        self.failIfExists('a')
142
101
        self.failUnlessExists('b')
143
102
        self.check_file_contents('b', 'something in a\n')
144
103
 
145
 
        self.create_file('a', 'new something in a\n')
146
 
        self._fancy_rename('b', 'a')
 
104
        open('a', 'wb').write('new something in a\n')
 
105
        rename('b', 'a')
147
106
 
148
107
        self.check_file_contents('a', 'something in a\n')
149
108
 
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
109
    def test_rename(self):
163
110
        # Rename should be semi-atomic on all platforms
164
 
        self.create_file('a', 'something in a\n')
 
111
        open('a', 'wb').write('something in a\n')
165
112
        osutils.rename('a', 'b')
166
113
        self.failIfExists('a')
167
114
        self.failUnlessExists('b')
168
115
        self.check_file_contents('b', 'something in a\n')
169
116
 
170
 
        self.create_file('a', 'new something in a\n')
 
117
        open('a', 'wb').write('new something in a\n')
171
118
        osutils.rename('b', 'a')
172
119
 
173
120
        self.check_file_contents('a', 'something in a\n')
184
131
        shape = sorted(os.listdir('.'))
185
132
        self.assertEquals(['A', 'B'], shape)
186
133
 
187
 
 
188
 
class TestRandChars(tests.TestCase):
189
 
 
190
134
    def test_01_rand_chars_empty(self):
191
135
        result = osutils.rand_chars(0)
192
136
        self.assertEqual(result, '')
197
141
        self.assertEqual(type(result), str)
198
142
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
199
143
 
200
 
 
201
 
class TestIsInside(tests.TestCase):
202
 
 
203
144
    def test_is_inside(self):
204
145
        is_inside = osutils.is_inside
205
146
        self.assertTrue(is_inside('src', 'src/foo.c'))
210
151
        self.assertTrue(is_inside('', 'foo.c'))
211
152
 
212
153
    def test_is_inside_any(self):
213
 
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
 
154
        SRC_FOO_C = pathjoin('src', 'foo.c')
214
155
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
215
156
                         (['src'], SRC_FOO_C),
216
157
                         (['src'], 'src'),
217
158
                         ]:
218
 
            self.assert_(osutils.is_inside_any(dirs, fn))
 
159
            self.assert_(is_inside_any(dirs, fn))
219
160
        for dirs, fn in [(['src'], 'srccontrol'),
220
161
                         (['src'], 'srccontrol/foo')]:
221
 
            self.assertFalse(osutils.is_inside_any(dirs, fn))
 
162
            self.assertFalse(is_inside_any(dirs, fn))
222
163
 
223
164
    def test_is_inside_or_parent_of_any(self):
224
165
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
227
168
                         (['src/bar.c', 'bla/foo.c'], 'src'),
228
169
                         (['src'], 'src'),
229
170
                         ]:
230
 
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
171
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
231
172
 
232
173
        for dirs, fn in [(['src'], 'srccontrol'),
233
174
                         (['srccontrol/foo.c'], 'src'),
234
175
                         (['src'], 'srccontrol/foo')]:
235
 
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
236
 
 
237
 
 
238
 
class TestRmTree(tests.TestCaseInTempDir):
 
176
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
239
177
 
240
178
    def test_rmtree(self):
241
179
        # Check to remove tree with read-only files/dirs
255
193
        self.failIfExists('dir/file')
256
194
        self.failIfExists('dir')
257
195
 
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
196
    def test_file_kind(self):
274
197
        self.build_tree(['file', 'dir/'])
275
198
        self.assertEquals('file', osutils.file_kind('file'))
305
228
                os.remove('socket')
306
229
 
307
230
    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):
 
231
        self.assertEqual(osutils.kind_marker('file'), '')
 
232
        self.assertEqual(osutils.kind_marker('directory'), '/')
 
233
        self.assertEqual(osutils.kind_marker('symlink'), '@')
 
234
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
319
235
 
320
236
    def test_get_umask(self):
321
237
        if sys.platform == 'win32':
324
240
            return
325
241
 
326
242
        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):
 
243
        try:
 
244
            os.umask(0222)
 
245
            self.assertEqual(0222, osutils.get_umask())
 
246
            os.umask(0022)
 
247
            self.assertEqual(0022, osutils.get_umask())
 
248
            os.umask(0002)
 
249
            self.assertEqual(0002, osutils.get_umask())
 
250
            os.umask(0027)
 
251
            self.assertEqual(0027, osutils.get_umask())
 
252
        finally:
 
253
            os.umask(orig_umask)
339
254
 
340
255
    def assertFormatedDelta(self, expected, seconds):
341
256
        """Assert osutils.format_delta formats as expected"""
383
298
        # Instead blackbox.test_locale should check for localized
384
299
        # dates once they do occur in output strings.
385
300
 
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
301
    def test_dereference_path(self):
417
 
        self.requireFeature(tests.SymlinkFeature)
 
302
        self.requireFeature(SymlinkFeature)
418
303
        cwd = osutils.realpath('.')
419
304
        os.mkdir('bar')
420
305
        bar_path = osutils.pathjoin(cwd, 'bar')
461
346
            osutils.make_readonly('dangling')
462
347
            osutils.make_writable('dangling')
463
348
 
 
349
    def test_kind_marker(self):
 
350
        self.assertEqual("", osutils.kind_marker("file"))
 
351
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
352
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
353
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
 
354
 
464
355
    def test_host_os_dereferences_symlinks(self):
465
356
        osutils.host_os_dereferences_symlinks()
466
357
 
467
358
 
468
 
class TestCanonicalRelPath(tests.TestCaseInTempDir):
 
359
class TestCanonicalRelPath(TestCaseInTempDir):
469
360
 
470
 
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
 
361
    _test_needs_features = [CaseInsCasePresFilenameFeature]
471
362
 
472
363
    def test_canonical_relpath_simple(self):
473
364
        f = file('MixedCaseName', 'w')
474
365
        f.close()
475
 
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
476
 
        self.failUnlessEqual('work/MixedCaseName', actual)
 
366
        self.failUnlessEqual(
 
367
            canonical_relpath(self.test_base_dir, 'mixedcasename'),
 
368
            'work/MixedCaseName')
477
369
 
478
370
    def test_canonical_relpath_missing_tail(self):
479
371
        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):
 
372
        self.failUnlessEqual(
 
373
            canonical_relpath(self.test_base_dir, 'mixedcaseparent/nochild'),
 
374
            'work/MixedCaseParent/nochild')
 
375
 
 
376
 
 
377
class TestPumpFile(TestCase):
529
378
    """Test pumpfile method."""
530
 
 
531
379
    def setUp(self):
532
 
        tests.TestCase.setUp(self)
 
380
        TestCase.setUp(self)
533
381
        # create a test datablock
534
382
        self.block_size = 512
535
383
        pattern = '0123456789ABCDEF'
542
390
        # make sure test data is larger than max read size
543
391
        self.assertTrue(self.test_data_len > self.block_size)
544
392
 
545
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
393
        from_file = FakeReadFile(self.test_data)
546
394
        to_file = StringIO()
547
395
 
548
396
        # read (max / 2) bytes and verify read size wasn't affected
549
397
        num_bytes_to_read = self.block_size / 2
550
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
398
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
551
399
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
552
400
        self.assertEqual(from_file.get_read_count(), 1)
553
401
 
554
402
        # read (max) bytes and verify read size wasn't affected
555
403
        num_bytes_to_read = self.block_size
556
404
        from_file.reset_read_count()
557
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
405
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
558
406
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
559
407
        self.assertEqual(from_file.get_read_count(), 1)
560
408
 
561
409
        # read (max + 1) bytes and verify read size was limited
562
410
        num_bytes_to_read = self.block_size + 1
563
411
        from_file.reset_read_count()
564
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
412
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
565
413
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
566
414
        self.assertEqual(from_file.get_read_count(), 2)
567
415
 
568
416
        # finish reading the rest of the data
569
417
        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)
 
418
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
571
419
 
572
420
        # report error if the data wasn't equal (we only report the size due
573
421
        # to the length of the data)
583
431
        self.assertTrue(self.test_data_len > self.block_size)
584
432
 
585
433
        # retrieve data in blocks
586
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
434
        from_file = FakeReadFile(self.test_data)
587
435
        to_file = StringIO()
588
 
        osutils.pumpfile(from_file, to_file, self.test_data_len,
589
 
                         self.block_size)
 
436
        pumpfile(from_file, to_file, self.test_data_len, self.block_size)
590
437
 
591
438
        # verify read size was equal to the maximum read size
592
439
        self.assertTrue(from_file.get_max_read_size() > 0)
607
454
        self.assertTrue(self.test_data_len > self.block_size)
608
455
 
609
456
        # retrieve data to EOF
610
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
457
        from_file = FakeReadFile(self.test_data)
611
458
        to_file = StringIO()
612
 
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
 
459
        pumpfile(from_file, to_file, -1, self.block_size)
613
460
 
614
461
        # verify read size was equal to the maximum read size
615
462
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
627
474
        test verifies that any existing usages of pumpfile will not be broken
628
475
        with this new version."""
629
476
        # retrieve data using default (old) pumpfile method
630
 
        from_file = file_utils.FakeReadFile(self.test_data)
 
477
        from_file = FakeReadFile(self.test_data)
631
478
        to_file = StringIO()
632
 
        osutils.pumpfile(from_file, to_file)
 
479
        pumpfile(from_file, to_file)
633
480
 
634
481
        # report error if the data wasn't equal (we only report the size due
635
482
        # to the length of the data)
644
491
            activity.append((length, direction))
645
492
        from_file = StringIO(self.test_data)
646
493
        to_file = StringIO()
647
 
        osutils.pumpfile(from_file, to_file, buff_size=500,
648
 
                         report_activity=log_activity, direction='read')
 
494
        pumpfile(from_file, to_file, buff_size=500,
 
495
                 report_activity=log_activity, direction='read')
649
496
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
650
497
                          (36, 'read')], activity)
651
498
 
652
499
        from_file = StringIO(self.test_data)
653
500
        to_file = StringIO()
654
501
        del activity[:]
655
 
        osutils.pumpfile(from_file, to_file, buff_size=500,
656
 
                         report_activity=log_activity, direction='write')
 
502
        pumpfile(from_file, to_file, buff_size=500,
 
503
                 report_activity=log_activity, direction='write')
657
504
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
658
505
                          (36, 'write')], activity)
659
506
 
661
508
        from_file = StringIO(self.test_data)
662
509
        to_file = StringIO()
663
510
        del activity[:]
664
 
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
665
 
                         report_activity=log_activity, direction='read')
 
511
        pumpfile(from_file, to_file, buff_size=500, read_length=1028,
 
512
                 report_activity=log_activity, direction='read')
666
513
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
667
514
 
668
515
 
669
516
 
670
 
class TestPumpStringFile(tests.TestCase):
 
517
class TestPumpStringFile(TestCase):
671
518
 
672
519
    def test_empty(self):
673
520
        output = StringIO()
674
 
        osutils.pump_string_file("", output)
 
521
        pump_string_file("", output)
675
522
        self.assertEqual("", output.getvalue())
676
523
 
677
524
    def test_more_than_segment_size(self):
678
525
        output = StringIO()
679
 
        osutils.pump_string_file("123456789", output, 2)
 
526
        pump_string_file("123456789", output, 2)
680
527
        self.assertEqual("123456789", output.getvalue())
681
528
 
682
529
    def test_segment_size(self):
683
530
        output = StringIO()
684
 
        osutils.pump_string_file("12", output, 2)
 
531
        pump_string_file("12", output, 2)
685
532
        self.assertEqual("12", output.getvalue())
686
533
 
687
534
    def test_segment_size_multiple(self):
688
535
        output = StringIO()
689
 
        osutils.pump_string_file("1234", output, 2)
 
536
        pump_string_file("1234", output, 2)
690
537
        self.assertEqual("1234", output.getvalue())
691
538
 
692
539
 
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):
 
540
class TestSafeUnicode(TestCase):
713
541
 
714
542
    def test_from_ascii_string(self):
715
543
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
724
552
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
725
553
 
726
554
    def test_bad_utf8_string(self):
727
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
555
        self.assertRaises(BzrBadParameterNotUnicode,
728
556
                          osutils.safe_unicode,
729
557
                          '\xbb\xbb')
730
558
 
731
559
 
732
 
class TestSafeUtf8(tests.TestCase):
 
560
class TestSafeUtf8(TestCase):
733
561
 
734
562
    def test_from_ascii_string(self):
735
563
        f = 'foobar'
745
573
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
746
574
 
747
575
    def test_bad_utf8_string(self):
748
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
576
        self.assertRaises(BzrBadParameterNotUnicode,
749
577
                          osutils.safe_utf8, '\xbb\xbb')
750
578
 
751
579
 
752
 
class TestSafeRevisionId(tests.TestCase):
 
580
class TestSafeRevisionId(TestCase):
753
581
 
754
582
    def test_from_ascii_string(self):
755
583
        # this shouldn't give a warning because it's getting an ascii string
777
605
        self.assertEqual(None, osutils.safe_revision_id(None))
778
606
 
779
607
 
780
 
class TestSafeFileId(tests.TestCase):
 
608
class TestSafeFileId(TestCase):
781
609
 
782
610
    def test_from_ascii_string(self):
783
611
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
803
631
        self.assertEqual(None, osutils.safe_file_id(None))
804
632
 
805
633
 
806
 
class TestWin32Funcs(tests.TestCase):
807
 
    """Test that _win32 versions of os utilities return appropriate paths."""
 
634
class TestWin32Funcs(TestCase):
 
635
    """Test that the _win32 versions of os utilities return appropriate paths."""
808
636
 
809
637
    def test_abspath(self):
810
638
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
817
645
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
818
646
 
819
647
    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'))
 
648
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
 
649
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
650
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
 
651
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
 
652
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
653
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
832
654
 
833
655
    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'))
 
656
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
657
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
838
658
 
839
659
    def test_getcwd(self):
840
660
        cwd = osutils._win32_getcwd()
869
689
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
870
690
 
871
691
 
872
 
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
 
692
class TestWin32FuncsDirs(TestCaseInTempDir):
873
693
    """Test win32 functions that create files."""
874
694
 
875
695
    def test_getcwd(self):
876
 
        self.requireFeature(tests.UnicodeFilenameFeature)
877
 
        os.mkdir(u'mu-\xb5')
 
696
        if win32utils.winver == 'Windows 98':
 
697
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
 
698
        # Make sure getcwd can handle unicode filenames
 
699
        try:
 
700
            os.mkdir(u'mu-\xb5')
 
701
        except UnicodeError:
 
702
            raise TestSkipped("Unable to create Unicode filename")
 
703
 
878
704
        os.chdir(u'mu-\xb5')
879
705
        # TODO: jam 20060427 This will probably fail on Mac OSX because
880
706
        #       it will change the normalization of B\xe5gfors
885
711
    def test_minimum_path_selection(self):
886
712
        self.assertEqual(set(),
887
713
            osutils.minimum_path_selection([]))
888
 
        self.assertEqual(set(['a']),
889
 
            osutils.minimum_path_selection(['a']))
890
714
        self.assertEqual(set(['a', 'b']),
891
715
            osutils.minimum_path_selection(['a', 'b']))
892
716
        self.assertEqual(set(['a/', 'b']),
893
717
            osutils.minimum_path_selection(['a/', 'b']))
894
718
        self.assertEqual(set(['a/', 'b']),
895
719
            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
720
 
899
721
    def test_mkdtemp(self):
900
722
        tmpdir = osutils._win32_mkdtemp(dir='.')
956
778
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
957
779
 
958
780
 
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):
 
781
class TestMacFuncsDirs(TestCaseInTempDir):
969
782
    """Test mac special functions that require directories."""
970
783
 
971
784
    def test_getcwd(self):
972
 
        self.requireFeature(tests.UnicodeFilenameFeature)
973
 
        os.mkdir(u'B\xe5gfors')
 
785
        # On Mac, this will actually create Ba\u030agfors
 
786
        # but chdir will still work, because it accepts both paths
 
787
        try:
 
788
            os.mkdir(u'B\xe5gfors')
 
789
        except UnicodeError:
 
790
            raise TestSkipped("Unable to create Unicode filename")
 
791
 
974
792
        os.chdir(u'B\xe5gfors')
975
793
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
976
794
 
977
795
    def test_getcwd_nonnorm(self):
978
 
        self.requireFeature(tests.UnicodeFilenameFeature)
979
796
        # Test that _mac_getcwd() will normalize this path
980
 
        os.mkdir(u'Ba\u030agfors')
 
797
        try:
 
798
            os.mkdir(u'Ba\u030agfors')
 
799
        except UnicodeError:
 
800
            raise TestSkipped("Unable to create Unicode filename")
 
801
 
981
802
        os.chdir(u'Ba\u030agfors')
982
803
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
983
804
 
984
805
 
985
 
class TestChunksToLines(tests.TestCase):
 
806
class TestChunksToLines(TestCase):
986
807
 
987
808
    def test_smoketest(self):
988
809
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
992
813
 
993
814
    def test_osutils_binding(self):
994
815
        from bzrlib.tests import test__chunks_to_lines
995
 
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
 
816
        if test__chunks_to_lines.CompiledChunksToLinesFeature.available():
996
817
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
997
818
        else:
998
819
            from bzrlib._chunks_to_lines_py import chunks_to_lines
999
820
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1000
821
 
1001
822
 
1002
 
class TestSplitLines(tests.TestCase):
 
823
class TestSplitLines(TestCase):
1003
824
 
1004
825
    def test_split_unicode(self):
1005
826
        self.assertEqual([u'foo\n', u'bar\xae'],
1012
833
                         osutils.split_lines('foo\rbar\n'))
1013
834
 
1014
835
 
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])
 
836
class TestWalkDirs(TestCaseInTempDir):
1021
837
 
1022
838
    def test_walkdirs(self):
1023
839
        tree = [
1056
872
            result.append((dirdetail, dirblock))
1057
873
 
1058
874
        self.assertTrue(found_bzrdir)
1059
 
        self.assertExpectedBlocks(expected_dirblocks, result)
 
875
        self.assertEqual(expected_dirblocks,
 
876
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1060
877
        # you can search a subdir only, with a supplied prefix.
1061
878
        result = []
1062
879
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1063
880
            result.append(dirblock)
1064
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
881
        self.assertEqual(expected_dirblocks[1:],
 
882
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1065
883
 
1066
884
    def test_walkdirs_os_error(self):
1067
 
        # <https://bugs.launchpad.net/bzr/+bug/338653>
 
885
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
1068
886
        # Pyrex readdir didn't raise useful messages if it had an error
1069
887
        # reading the directory
1070
888
        if sys.platform == 'win32':
1073
891
        os.mkdir("test-unreadable")
1074
892
        os.chmod("test-unreadable", 0000)
1075
893
        # must chmod it back so that it can be removed
1076
 
        self.addCleanup(os.chmod, "test-unreadable", 0700)
 
894
        self.addCleanup(lambda: os.chmod("test-unreadable", 0700))
1077
895
        # The error is not raised until the generator is actually evaluated.
1078
896
        # (It would be ok if it happened earlier but at the moment it
1079
897
        # doesn't.)
1120
938
            result.append((dirdetail, dirblock))
1121
939
 
1122
940
        self.assertTrue(found_bzrdir)
1123
 
        self.assertExpectedBlocks(expected_dirblocks, result)
1124
 
 
 
941
        self.assertEqual(expected_dirblocks,
 
942
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1125
943
        # you can search a subdir only, with a supplied prefix.
1126
944
        result = []
1127
945
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1128
946
            result.append(dirblock)
1129
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
947
        self.assertEqual(expected_dirblocks[1:],
 
948
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
1130
949
 
1131
950
    def _filter_out_stat(self, result):
1132
951
        """Filter out the stat value from the walkdirs result"""
1138
957
            dirblock[:] = new_dirblock
1139
958
 
1140
959
    def _save_platform_info(self):
1141
 
        self.overrideAttr(win32utils, 'winver')
1142
 
        self.overrideAttr(osutils, '_fs_enc')
1143
 
        self.overrideAttr(osutils, '_selected_dir_reader')
 
960
        cur_winver = win32utils.winver
 
961
        cur_fs_enc = osutils._fs_enc
 
962
        cur_dir_reader = osutils._selected_dir_reader
 
963
        def restore():
 
964
            win32utils.winver = cur_winver
 
965
            osutils._fs_enc = cur_fs_enc
 
966
            osutils._selected_dir_reader = cur_dir_reader
 
967
        self.addCleanup(restore)
1144
968
 
1145
 
    def assertDirReaderIs(self, expected):
 
969
    def assertReadFSDirIs(self, expected):
1146
970
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1147
971
        # Force it to redetect
1148
972
        osutils._selected_dir_reader = None
1155
979
        self._save_platform_info()
1156
980
        win32utils.winver = None # Avoid the win32 detection code
1157
981
        osutils._fs_enc = 'UTF-8'
1158
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
982
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
1159
983
 
1160
984
    def test_force_walkdirs_utf8_fs_ascii(self):
1161
985
        self.requireFeature(UTF8DirReaderFeature)
1162
986
        self._save_platform_info()
1163
987
        win32utils.winver = None # Avoid the win32 detection code
1164
988
        osutils._fs_enc = 'US-ASCII'
1165
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
989
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
1166
990
 
1167
991
    def test_force_walkdirs_utf8_fs_ANSI(self):
1168
992
        self.requireFeature(UTF8DirReaderFeature)
1169
993
        self._save_platform_info()
1170
994
        win32utils.winver = None # Avoid the win32 detection code
1171
995
        osutils._fs_enc = 'ANSI_X3.4-1968'
1172
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
996
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
1173
997
 
1174
998
    def test_force_walkdirs_utf8_fs_latin1(self):
1175
999
        self._save_platform_info()
1176
1000
        win32utils.winver = None # Avoid the win32 detection code
1177
1001
        osutils._fs_enc = 'latin1'
1178
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1002
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
1179
1003
 
1180
1004
    def test_force_walkdirs_utf8_nt(self):
1181
1005
        # Disabled because the thunk of the whole walkdirs api is disabled.
1182
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1006
        self.requireFeature(Win32ReadDirFeature)
1183
1007
        self._save_platform_info()
1184
1008
        win32utils.winver = 'Windows NT'
1185
1009
        from bzrlib._walkdirs_win32 import Win32ReadDir
1186
 
        self.assertDirReaderIs(Win32ReadDir)
 
1010
        self.assertReadFSDirIs(Win32ReadDir)
1187
1011
 
1188
1012
    def test_force_walkdirs_utf8_98(self):
1189
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1013
        self.requireFeature(Win32ReadDirFeature)
1190
1014
        self._save_platform_info()
1191
1015
        win32utils.winver = 'Windows 98'
1192
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1016
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
1193
1017
 
1194
1018
    def test_unicode_walkdirs(self):
1195
1019
        """Walkdirs should always return unicode paths."""
1196
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1197
1020
        name0 = u'0file-\xb6'
1198
1021
        name1 = u'1dir-\u062c\u0648'
1199
1022
        name2 = u'2file-\u0633'
1204
1027
            name1 + '/' + name1 + '/',
1205
1028
            name2,
1206
1029
            ]
1207
 
        self.build_tree(tree)
 
1030
        try:
 
1031
            self.build_tree(tree)
 
1032
        except UnicodeError:
 
1033
            raise TestSkipped('Could not represent Unicode chars'
 
1034
                              ' in current encoding.')
1208
1035
        expected_dirblocks = [
1209
1036
                ((u'', u'.'),
1210
1037
                 [(name0, name0, 'file', './' + name0),
1236
1063
 
1237
1064
        The abspath portion might be in unicode or utf-8
1238
1065
        """
1239
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1240
1066
        name0 = u'0file-\xb6'
1241
1067
        name1 = u'1dir-\u062c\u0648'
1242
1068
        name2 = u'2file-\u0633'
1247
1073
            name1 + '/' + name1 + '/',
1248
1074
            name2,
1249
1075
            ]
1250
 
        self.build_tree(tree)
 
1076
        try:
 
1077
            self.build_tree(tree)
 
1078
        except UnicodeError:
 
1079
            raise TestSkipped('Could not represent Unicode chars'
 
1080
                              ' in current encoding.')
1251
1081
        name0 = name0.encode('utf8')
1252
1082
        name1 = name1.encode('utf8')
1253
1083
        name2 = name2.encode('utf8')
1297
1127
 
1298
1128
        The abspath portion should be in unicode
1299
1129
        """
1300
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1301
1130
        # Use the unicode reader. TODO: split into driver-and-driven unit
1302
1131
        # tests.
1303
1132
        self._save_platform_info()
1312
1141
            name1u + '/' + name1u + '/',
1313
1142
            name2u,
1314
1143
            ]
1315
 
        self.build_tree(tree)
 
1144
        try:
 
1145
            self.build_tree(tree)
 
1146
        except UnicodeError:
 
1147
            raise TestSkipped('Could not represent Unicode chars'
 
1148
                              ' in current encoding.')
1316
1149
        name0 = name0u.encode('utf8')
1317
1150
        name1 = name1u.encode('utf8')
1318
1151
        name2 = name2u.encode('utf8')
1343
1176
        self.assertEqual(expected_dirblocks, result)
1344
1177
 
1345
1178
    def test__walkdirs_utf8_win32readdir(self):
1346
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1179
        self.requireFeature(Win32ReadDirFeature)
1347
1180
        self.requireFeature(tests.UnicodeFilenameFeature)
1348
1181
        from bzrlib._walkdirs_win32 import Win32ReadDir
1349
1182
        self._save_platform_info()
1400
1233
 
1401
1234
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1402
1235
        """make sure our Stat values are valid"""
1403
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1236
        self.requireFeature(Win32ReadDirFeature)
1404
1237
        self.requireFeature(tests.UnicodeFilenameFeature)
1405
1238
        from bzrlib._walkdirs_win32 import Win32ReadDir
1406
1239
        name0u = u'0file-\xb6'
1424
1257
 
1425
1258
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1426
1259
        """make sure our Stat values are valid"""
1427
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1260
        self.requireFeature(Win32ReadDirFeature)
1428
1261
        self.requireFeature(tests.UnicodeFilenameFeature)
1429
1262
        from bzrlib._walkdirs_win32 import Win32ReadDir
1430
1263
        name0u = u'0dir-\u062c\u0648'
1515
1348
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1516
1349
 
1517
1350
 
1518
 
class TestCopyTree(tests.TestCaseInTempDir):
 
1351
class TestCopyTree(TestCaseInTempDir):
1519
1352
 
1520
1353
    def test_copy_basic_tree(self):
1521
1354
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1531
1364
        self.assertEqual(['c'], os.listdir('target/b'))
1532
1365
 
1533
1366
    def test_copy_tree_symlinks(self):
1534
 
        self.requireFeature(tests.SymlinkFeature)
 
1367
        self.requireFeature(SymlinkFeature)
1535
1368
        self.build_tree(['source/'])
1536
1369
        os.symlink('a/generic/path', 'source/lnk')
1537
1370
        osutils.copy_tree('source', 'target')
1567
1400
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1568
1401
 
1569
1402
 
1570
 
class TestSetUnsetEnv(tests.TestCase):
 
1403
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
 
1404
# [bialix] 2006/12/26
 
1405
 
 
1406
 
 
1407
class TestSetUnsetEnv(TestCase):
1571
1408
    """Test updating the environment"""
1572
1409
 
1573
1410
    def setUp(self):
1579
1416
        def cleanup():
1580
1417
            if 'BZR_TEST_ENV_VAR' in os.environ:
1581
1418
                del os.environ['BZR_TEST_ENV_VAR']
 
1419
 
1582
1420
        self.addCleanup(cleanup)
1583
1421
 
1584
1422
    def test_set(self):
1599
1437
 
1600
1438
        So Unicode strings must be encoded.
1601
1439
        """
1602
 
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
1440
        uni_val, env_val = probe_unicode_in_user_encoding()
1603
1441
        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(),))
 
1442
            raise TestSkipped('Cannot find a unicode character that works in'
 
1443
                              ' encoding %s' % (osutils.get_user_encoding(),))
1607
1444
 
1608
1445
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1609
1446
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1617
1454
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1618
1455
 
1619
1456
 
1620
 
class TestSizeShaFile(tests.TestCaseInTempDir):
 
1457
class TestLocalTimeOffset(TestCase):
 
1458
 
 
1459
    def test_local_time_offset(self):
 
1460
        """Test that local_time_offset() returns a sane value."""
 
1461
        offset = osutils.local_time_offset()
 
1462
        self.assertTrue(isinstance(offset, int))
 
1463
        # Test that the offset is no more than a eighteen hours in
 
1464
        # either direction.
 
1465
        # Time zone handling is system specific, so it is difficult to
 
1466
        # do more specific tests, but a value outside of this range is
 
1467
        # probably wrong.
 
1468
        eighteen_hours = 18 * 3600
 
1469
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1470
 
 
1471
    def test_local_time_offset_with_timestamp(self):
 
1472
        """Test that local_time_offset() works with a timestamp."""
 
1473
        offset = osutils.local_time_offset(1000000000.1234567)
 
1474
        self.assertTrue(isinstance(offset, int))
 
1475
        eighteen_hours = 18 * 3600
 
1476
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1477
 
 
1478
 
 
1479
class TestSizeShaFile(TestCaseInTempDir):
1621
1480
 
1622
1481
    def test_sha_empty(self):
1623
1482
        self.build_tree_contents([('foo', '')])
1632
1491
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1633
1492
        self.build_tree_contents([('foo', text)])
1634
1493
        expected_sha = osutils.sha_string(text)
1635
 
        f = open('foo', 'rb')
 
1494
        f = open('foo')
1636
1495
        self.addCleanup(f.close)
1637
1496
        size, sha = osutils.size_sha_file(f)
1638
1497
        self.assertEqual(38, size)
1639
1498
        self.assertEqual(expected_sha, sha)
1640
1499
 
1641
1500
 
1642
 
class TestShaFileByName(tests.TestCaseInTempDir):
 
1501
class TestShaFileByName(TestCaseInTempDir):
1643
1502
 
1644
1503
    def test_sha_empty(self):
1645
1504
        self.build_tree_contents([('foo', '')])
1653
1512
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1654
1513
 
1655
1514
 
1656
 
class TestResourceLoading(tests.TestCaseInTempDir):
 
1515
class TestResourceLoading(TestCaseInTempDir):
1657
1516
 
1658
1517
    def test_resource_string(self):
1659
1518
        # test resource in bzrlib
1669
1528
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1670
1529
 
1671
1530
 
1672
 
class TestReCompile(tests.TestCase):
 
1531
class TestReCompile(TestCase):
1673
1532
 
1674
1533
    def test_re_compile_checked(self):
1675
1534
        r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1685
1544
            "Invalid regular expression in test case: '*': "
1686
1545
            "nothing to repeat",
1687
1546
            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_copy_ownership_from_path(self):
1994
 
        """copy_ownership_from_path test with specified src."""
1995
 
        ownsrc = '/'
1996
 
        f = open('test_file', 'wt')
1997
 
        osutils.copy_ownership_from_path('test_file', ownsrc)
1998
 
 
1999
 
        s = os.stat(ownsrc)
2000
 
        self.assertEquals(self.path, 'test_file')
2001
 
        self.assertEquals(self.uid, s.st_uid)
2002
 
        self.assertEquals(self.gid, s.st_gid)
2003
 
 
2004
 
    def test_copy_ownership_nonesrc(self):
2005
 
        """copy_ownership_from_path test with src=None."""
2006
 
        f = open('test_file', 'wt')
2007
 
        # should use parent dir for permissions
2008
 
        osutils.copy_ownership_from_path('test_file')
2009
 
 
2010
 
        s = os.stat('..')
2011
 
        self.assertEquals(self.path, 'test_file')
2012
 
        self.assertEquals(self.uid, s.st_uid)
2013
 
        self.assertEquals(self.gid, s.st_gid)
2014
 
 
2015
 
class TestGetuserUnicode(tests.TestCase):
2016
 
 
2017
 
    def test_ascii_user(self):
2018
 
        osutils.set_or_unset_env('LOGNAME', 'jrandom')
2019
 
        self.assertEqual(u'jrandom', osutils.getuser_unicode())
2020
 
 
2021
 
    def test_unicode_user(self):
2022
 
        ue = osutils.get_user_encoding()
2023
 
        osutils.set_or_unset_env('LOGNAME', u'jrandom\xb6'.encode(ue))
2024
 
        self.assertEqual(u'jrandom\xb6', osutils.getuser_unicode())