~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Aaron Bentley
  • Date: 2006-06-21 14:30:57 UTC
  • mfrom: (1801.1.1 bzr.dev)
  • mto: This revision was merged to the branch mainline in revision 1803.
  • Revision ID: abentley@panoramicfeedback.com-20060621143057-776e4b8d707e430e
Install benchmarks. (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
 
1
# Copyright (C) 2005 by Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
19
 
from cStringIO import StringIO
20
19
import errno
21
20
import os
22
 
import re
23
21
import socket
24
22
import stat
25
23
import sys
26
 
import time
27
24
 
28
 
from bzrlib import (
29
 
    errors,
30
 
    osutils,
31
 
    tests,
32
 
    win32utils,
33
 
    )
 
25
import bzrlib
 
26
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
 
27
import bzrlib.osutils as osutils
34
28
from bzrlib.tests import (
35
 
    file_utils,
36
 
    test__walkdirs_win32,
37
 
    )
38
 
 
39
 
 
40
 
class _UTF8DirReaderFeature(tests.Feature):
41
 
 
42
 
    def _probe(self):
43
 
        try:
44
 
            from bzrlib import _readdir_pyx
45
 
            self.reader = _readdir_pyx.UTF8DirReader
46
 
            return True
47
 
        except ImportError:
48
 
            return False
49
 
 
50
 
    def feature_name(self):
51
 
        return 'bzrlib._readdir_pyx'
52
 
 
53
 
UTF8DirReaderFeature = _UTF8DirReaderFeature()
54
 
 
55
 
 
56
 
def _already_unicode(s):
57
 
    return s
58
 
 
59
 
 
60
 
def _fs_enc_to_unicode(s):
61
 
    return s.decode(osutils._fs_enc)
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.Win32ReadDirFeature.available():
87
 
        try:
88
 
            from bzrlib import _walkdirs_win32
89
 
            # TODO: check on windows, it may be that we need to use/add
90
 
            # safe_unicode instead of _fs_enc_to_unicode
91
 
            scenarios.append(
92
 
                ('win32',
93
 
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
94
 
                      _native_to_unicode=_fs_enc_to_unicode)))
95
 
        except ImportError:
96
 
            pass
97
 
    return scenarios
98
 
 
99
 
 
100
 
def load_tests(basic_tests, module, loader):
101
 
    suite = loader.suiteClass()
102
 
    dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
103
 
        basic_tests, tests.condition_isinstance(TestDirReader))
104
 
    tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
105
 
    suite.addTest(remaining_tests)
106
 
    return suite
107
 
 
108
 
 
109
 
class TestContainsWhitespace(tests.TestCase):
110
 
 
111
 
    def test_contains_whitespace(self):
112
 
        self.failUnless(osutils.contains_whitespace(u' '))
113
 
        self.failUnless(osutils.contains_whitespace(u'hello there'))
114
 
        self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
115
 
        self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
116
 
        self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
117
 
        self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
118
 
 
119
 
        # \xa0 is "Non-breaking-space" which on some python locales thinks it
120
 
        # is whitespace, but we do not.
121
 
        self.failIf(osutils.contains_whitespace(u''))
122
 
        self.failIf(osutils.contains_whitespace(u'hellothere'))
123
 
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
124
 
 
125
 
 
126
 
class TestRename(tests.TestCaseInTempDir):
 
29
        StringIOWrapper,
 
30
        TestCase, 
 
31
        TestCaseInTempDir, 
 
32
        TestSkipped,
 
33
        )
 
34
 
 
35
 
 
36
class TestOSUtils(TestCaseInTempDir):
127
37
 
128
38
    def test_fancy_rename(self):
129
39
        # This should work everywhere
158
68
 
159
69
    # TODO: test fancy_rename using a MemoryTransport
160
70
 
161
 
    def test_rename_change_case(self):
162
 
        # on Windows we should be able to change filename case by rename
163
 
        self.build_tree(['a', 'b/'])
164
 
        osutils.rename('a', 'A')
165
 
        osutils.rename('b', 'B')
166
 
        # we can't use failUnlessExists on case-insensitive filesystem
167
 
        # so try to check shape of the tree
168
 
        shape = sorted(os.listdir('.'))
169
 
        self.assertEquals(['A', 'B'], shape)
170
 
 
171
 
 
172
 
class TestRandChars(tests.TestCase):
173
 
 
174
71
    def test_01_rand_chars_empty(self):
175
72
        result = osutils.rand_chars(0)
176
73
        self.assertEqual(result, '')
182
79
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
183
80
 
184
81
 
185
 
class TestIsInside(tests.TestCase):
186
 
 
187
 
    def test_is_inside(self):
188
 
        is_inside = osutils.is_inside
189
 
        self.assertTrue(is_inside('src', 'src/foo.c'))
190
 
        self.assertFalse(is_inside('src', 'srccontrol'))
191
 
        self.assertTrue(is_inside('src', 'src/a/a/a/foo.c'))
192
 
        self.assertTrue(is_inside('foo.c', 'foo.c'))
193
 
        self.assertFalse(is_inside('foo.c', ''))
194
 
        self.assertTrue(is_inside('', 'foo.c'))
195
 
 
196
 
    def test_is_inside_any(self):
197
 
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
198
 
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
199
 
                         (['src'], SRC_FOO_C),
200
 
                         (['src'], 'src'),
201
 
                         ]:
202
 
            self.assert_(osutils.is_inside_any(dirs, fn))
203
 
        for dirs, fn in [(['src'], 'srccontrol'),
204
 
                         (['src'], 'srccontrol/foo')]:
205
 
            self.assertFalse(osutils.is_inside_any(dirs, fn))
206
 
 
207
 
    def test_is_inside_or_parent_of_any(self):
208
 
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
209
 
                         (['src'], 'src/foo.c'),
210
 
                         (['src/bar.c'], 'src'),
211
 
                         (['src/bar.c', 'bla/foo.c'], 'src'),
212
 
                         (['src'], 'src'),
213
 
                         ]:
214
 
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
215
 
 
216
 
        for dirs, fn in [(['src'], 'srccontrol'),
217
 
                         (['srccontrol/foo.c'], 'src'),
218
 
                         (['src'], 'srccontrol/foo')]:
219
 
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
220
 
 
221
 
 
222
 
class TestRmTree(tests.TestCaseInTempDir):
223
 
 
224
82
    def test_rmtree(self):
225
83
        # Check to remove tree with read-only files/dirs
226
84
        os.mkdir('dir')
239
97
        self.failIfExists('dir/file')
240
98
        self.failIfExists('dir')
241
99
 
242
 
 
243
 
class TestDeleteAny(tests.TestCaseInTempDir):
244
 
 
245
 
    def test_delete_any_readonly(self):
246
 
        # from <https://bugs.launchpad.net/bzr/+bug/218206>
247
 
        self.build_tree(['d/', 'f'])
248
 
        osutils.make_readonly('d')
249
 
        osutils.make_readonly('f')
250
 
 
251
 
        osutils.delete_any('f')
252
 
        osutils.delete_any('d')
253
 
 
254
 
 
255
 
class TestKind(tests.TestCaseInTempDir):
256
 
 
257
100
    def test_file_kind(self):
258
101
        self.build_tree(['file', 'dir/'])
259
102
        self.assertEquals('file', osutils.file_kind('file'))
261
104
        if osutils.has_symlinks():
262
105
            os.symlink('symlink', 'symlink')
263
106
            self.assertEquals('symlink', osutils.file_kind('symlink'))
264
 
 
 
107
        
265
108
        # TODO: jam 20060529 Test a block device
266
109
        try:
267
110
            os.lstat('/dev/null')
288
131
            finally:
289
132
                os.remove('socket')
290
133
 
291
 
    def test_kind_marker(self):
292
 
        self.assertEqual("", osutils.kind_marker("file"))
293
 
        self.assertEqual("/", osutils.kind_marker('directory'))
294
 
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
295
 
        self.assertEqual("@", osutils.kind_marker("symlink"))
296
 
        self.assertEqual("+", osutils.kind_marker("tree-reference"))
297
 
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
298
 
 
299
 
 
300
 
class TestUmask(tests.TestCaseInTempDir):
301
 
 
302
 
    def test_get_umask(self):
303
 
        if sys.platform == 'win32':
304
 
            # umask always returns '0', no way to set it
305
 
            self.assertEqual(0, osutils.get_umask())
306
 
            return
307
 
 
308
 
        orig_umask = osutils.get_umask()
309
 
        self.addCleanup(os.umask, orig_umask)
310
 
        os.umask(0222)
311
 
        self.assertEqual(0222, osutils.get_umask())
312
 
        os.umask(0022)
313
 
        self.assertEqual(0022, osutils.get_umask())
314
 
        os.umask(0002)
315
 
        self.assertEqual(0002, osutils.get_umask())
316
 
        os.umask(0027)
317
 
        self.assertEqual(0027, osutils.get_umask())
318
 
 
319
 
 
320
 
class TestDateTime(tests.TestCase):
321
 
 
322
 
    def assertFormatedDelta(self, expected, seconds):
323
 
        """Assert osutils.format_delta formats as expected"""
324
 
        actual = osutils.format_delta(seconds)
325
 
        self.assertEqual(expected, actual)
326
 
 
327
 
    def test_format_delta(self):
328
 
        self.assertFormatedDelta('0 seconds ago', 0)
329
 
        self.assertFormatedDelta('1 second ago', 1)
330
 
        self.assertFormatedDelta('10 seconds ago', 10)
331
 
        self.assertFormatedDelta('59 seconds ago', 59)
332
 
        self.assertFormatedDelta('89 seconds ago', 89)
333
 
        self.assertFormatedDelta('1 minute, 30 seconds ago', 90)
334
 
        self.assertFormatedDelta('3 minutes, 0 seconds ago', 180)
335
 
        self.assertFormatedDelta('3 minutes, 1 second ago', 181)
336
 
        self.assertFormatedDelta('10 minutes, 15 seconds ago', 615)
337
 
        self.assertFormatedDelta('30 minutes, 59 seconds ago', 1859)
338
 
        self.assertFormatedDelta('31 minutes, 0 seconds ago', 1860)
339
 
        self.assertFormatedDelta('60 minutes, 0 seconds ago', 3600)
340
 
        self.assertFormatedDelta('89 minutes, 59 seconds ago', 5399)
341
 
        self.assertFormatedDelta('1 hour, 30 minutes ago', 5400)
342
 
        self.assertFormatedDelta('2 hours, 30 minutes ago', 9017)
343
 
        self.assertFormatedDelta('10 hours, 0 minutes ago', 36000)
344
 
        self.assertFormatedDelta('24 hours, 0 minutes ago', 86400)
345
 
        self.assertFormatedDelta('35 hours, 59 minutes ago', 129599)
346
 
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129600)
347
 
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129601)
348
 
        self.assertFormatedDelta('36 hours, 1 minute ago', 129660)
349
 
        self.assertFormatedDelta('36 hours, 1 minute ago', 129661)
350
 
        self.assertFormatedDelta('84 hours, 10 minutes ago', 303002)
351
 
 
352
 
        # We handle when time steps the wrong direction because computers
353
 
        # don't have synchronized clocks.
354
 
        self.assertFormatedDelta('84 hours, 10 minutes in the future', -303002)
355
 
        self.assertFormatedDelta('1 second in the future', -1)
356
 
        self.assertFormatedDelta('2 seconds in the future', -2)
357
 
 
358
 
    def test_format_date(self):
359
 
        self.assertRaises(errors.UnsupportedTimezoneFormat,
360
 
            osutils.format_date, 0, timezone='foo')
361
 
        self.assertIsInstance(osutils.format_date(0), str)
362
 
        self.assertIsInstance(osutils.format_local_date(0), unicode)
363
 
        # Testing for the actual value of the local weekday without
364
 
        # duplicating the code from format_date is difficult.
365
 
        # Instead blackbox.test_locale should check for localized
366
 
        # dates once they do occur in output strings.
367
 
 
368
 
    def test_local_time_offset(self):
369
 
        """Test that local_time_offset() returns a sane value."""
370
 
        offset = osutils.local_time_offset()
371
 
        self.assertTrue(isinstance(offset, int))
372
 
        # Test that the offset is no more than a eighteen hours in
373
 
        # either direction.
374
 
        # Time zone handling is system specific, so it is difficult to
375
 
        # do more specific tests, but a value outside of this range is
376
 
        # probably wrong.
377
 
        eighteen_hours = 18 * 3600
378
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
379
 
 
380
 
    def test_local_time_offset_with_timestamp(self):
381
 
        """Test that local_time_offset() works with a timestamp."""
382
 
        offset = osutils.local_time_offset(1000000000.1234567)
383
 
        self.assertTrue(isinstance(offset, int))
384
 
        eighteen_hours = 18 * 3600
385
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
386
 
 
387
 
 
388
 
class TestLinks(tests.TestCaseInTempDir):
389
 
 
390
 
    def test_dereference_path(self):
391
 
        self.requireFeature(tests.SymlinkFeature)
392
 
        cwd = osutils.realpath('.')
393
 
        os.mkdir('bar')
394
 
        bar_path = osutils.pathjoin(cwd, 'bar')
395
 
        # Using './' to avoid bug #1213894 (first path component not
396
 
        # dereferenced) in Python 2.4.1 and earlier
397
 
        self.assertEqual(bar_path, osutils.realpath('./bar'))
398
 
        os.symlink('bar', 'foo')
399
 
        self.assertEqual(bar_path, osutils.realpath('./foo'))
400
 
 
401
 
        # Does not dereference terminal symlinks
402
 
        foo_path = osutils.pathjoin(cwd, 'foo')
403
 
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
404
 
 
405
 
        # Dereferences parent symlinks
406
 
        os.mkdir('bar/baz')
407
 
        baz_path = osutils.pathjoin(bar_path, 'baz')
408
 
        self.assertEqual(baz_path, osutils.dereference_path('./foo/baz'))
409
 
 
410
 
        # Dereferences parent symlinks that are the first path element
411
 
        self.assertEqual(baz_path, osutils.dereference_path('foo/baz'))
412
 
 
413
 
        # Dereferences parent symlinks in absolute paths
414
 
        foo_baz_path = osutils.pathjoin(foo_path, 'baz')
415
 
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
416
 
 
417
 
    def test_changing_access(self):
418
 
        f = file('file', 'w')
419
 
        f.write('monkey')
420
 
        f.close()
421
 
 
422
 
        # Make a file readonly
423
 
        osutils.make_readonly('file')
424
 
        mode = os.lstat('file').st_mode
425
 
        self.assertEqual(mode, mode & 0777555)
426
 
 
427
 
        # Make a file writable
428
 
        osutils.make_writable('file')
429
 
        mode = os.lstat('file').st_mode
430
 
        self.assertEqual(mode, mode | 0200)
431
 
 
432
 
        if osutils.has_symlinks():
433
 
            # should not error when handed a symlink
434
 
            os.symlink('nonexistent', 'dangling')
435
 
            osutils.make_readonly('dangling')
436
 
            osutils.make_writable('dangling')
437
 
 
438
 
    def test_host_os_dereferences_symlinks(self):
439
 
        osutils.host_os_dereferences_symlinks()
440
 
 
441
 
 
442
 
class TestCanonicalRelPath(tests.TestCaseInTempDir):
443
 
 
444
 
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
445
 
 
446
 
    def test_canonical_relpath_simple(self):
447
 
        f = file('MixedCaseName', 'w')
448
 
        f.close()
449
 
        # Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
450
 
        real_base_dir = osutils.realpath(self.test_base_dir)
451
 
        actual = osutils.canonical_relpath(real_base_dir, 'mixedcasename')
452
 
        self.failUnlessEqual('work/MixedCaseName', actual)
453
 
 
454
 
    def test_canonical_relpath_missing_tail(self):
455
 
        os.mkdir('MixedCaseParent')
456
 
        # Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
457
 
        real_base_dir = osutils.realpath(self.test_base_dir)
458
 
        actual = osutils.canonical_relpath(real_base_dir,
459
 
                                           'mixedcaseparent/nochild')
460
 
        self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
461
 
 
462
 
 
463
 
class TestPumpFile(tests.TestCase):
464
 
    """Test pumpfile method."""
465
 
 
466
 
    def setUp(self):
467
 
        tests.TestCase.setUp(self)
468
 
        # create a test datablock
469
 
        self.block_size = 512
470
 
        pattern = '0123456789ABCDEF'
471
 
        self.test_data = pattern * (3 * self.block_size / len(pattern))
472
 
        self.test_data_len = len(self.test_data)
473
 
 
474
 
    def test_bracket_block_size(self):
475
 
        """Read data in blocks with the requested read size bracketing the
476
 
        block size."""
477
 
        # make sure test data is larger than max read size
478
 
        self.assertTrue(self.test_data_len > self.block_size)
479
 
 
480
 
        from_file = file_utils.FakeReadFile(self.test_data)
481
 
        to_file = StringIO()
482
 
 
483
 
        # read (max / 2) bytes and verify read size wasn't affected
484
 
        num_bytes_to_read = self.block_size / 2
485
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
486
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
487
 
        self.assertEqual(from_file.get_read_count(), 1)
488
 
 
489
 
        # read (max) bytes and verify read size wasn't affected
490
 
        num_bytes_to_read = self.block_size
491
 
        from_file.reset_read_count()
492
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
493
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
494
 
        self.assertEqual(from_file.get_read_count(), 1)
495
 
 
496
 
        # read (max + 1) bytes and verify read size was limited
497
 
        num_bytes_to_read = self.block_size + 1
498
 
        from_file.reset_read_count()
499
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
500
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
501
 
        self.assertEqual(from_file.get_read_count(), 2)
502
 
 
503
 
        # finish reading the rest of the data
504
 
        num_bytes_to_read = self.test_data_len - to_file.tell()
505
 
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
506
 
 
507
 
        # report error if the data wasn't equal (we only report the size due
508
 
        # to the length of the data)
509
 
        response_data = to_file.getvalue()
510
 
        if response_data != self.test_data:
511
 
            message = "Data not equal.  Expected %d bytes, received %d."
512
 
            self.fail(message % (len(response_data), self.test_data_len))
513
 
 
514
 
    def test_specified_size(self):
515
 
        """Request a transfer larger than the maximum block size and verify
516
 
        that the maximum read doesn't exceed the block_size."""
517
 
        # make sure test data is larger than max read size
518
 
        self.assertTrue(self.test_data_len > self.block_size)
519
 
 
520
 
        # retrieve data in blocks
521
 
        from_file = file_utils.FakeReadFile(self.test_data)
522
 
        to_file = StringIO()
523
 
        osutils.pumpfile(from_file, to_file, self.test_data_len,
524
 
                         self.block_size)
525
 
 
526
 
        # verify read size was equal to the maximum read size
527
 
        self.assertTrue(from_file.get_max_read_size() > 0)
528
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
529
 
        self.assertEqual(from_file.get_read_count(), 3)
530
 
 
531
 
        # report error if the data wasn't equal (we only report the size due
532
 
        # to the length of the data)
533
 
        response_data = to_file.getvalue()
534
 
        if response_data != self.test_data:
535
 
            message = "Data not equal.  Expected %d bytes, received %d."
536
 
            self.fail(message % (len(response_data), self.test_data_len))
537
 
 
538
 
    def test_to_eof(self):
539
 
        """Read to end-of-file and verify that the reads are not larger than
540
 
        the maximum read size."""
541
 
        # make sure test data is larger than max read size
542
 
        self.assertTrue(self.test_data_len > self.block_size)
543
 
 
544
 
        # retrieve data to EOF
545
 
        from_file = file_utils.FakeReadFile(self.test_data)
546
 
        to_file = StringIO()
547
 
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
548
 
 
549
 
        # verify read size was equal to the maximum read size
550
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
551
 
        self.assertEqual(from_file.get_read_count(), 4)
552
 
 
553
 
        # report error if the data wasn't equal (we only report the size due
554
 
        # to the length of the data)
555
 
        response_data = to_file.getvalue()
556
 
        if response_data != self.test_data:
557
 
            message = "Data not equal.  Expected %d bytes, received %d."
558
 
            self.fail(message % (len(response_data), self.test_data_len))
559
 
 
560
 
    def test_defaults(self):
561
 
        """Verifies that the default arguments will read to EOF -- this
562
 
        test verifies that any existing usages of pumpfile will not be broken
563
 
        with this new version."""
564
 
        # retrieve data using default (old) pumpfile method
565
 
        from_file = file_utils.FakeReadFile(self.test_data)
566
 
        to_file = StringIO()
567
 
        osutils.pumpfile(from_file, to_file)
568
 
 
569
 
        # report error if the data wasn't equal (we only report the size due
570
 
        # to the length of the data)
571
 
        response_data = to_file.getvalue()
572
 
        if response_data != self.test_data:
573
 
            message = "Data not equal.  Expected %d bytes, received %d."
574
 
            self.fail(message % (len(response_data), self.test_data_len))
575
 
 
576
 
    def test_report_activity(self):
577
 
        activity = []
578
 
        def log_activity(length, direction):
579
 
            activity.append((length, direction))
580
 
        from_file = StringIO(self.test_data)
581
 
        to_file = StringIO()
582
 
        osutils.pumpfile(from_file, to_file, buff_size=500,
583
 
                         report_activity=log_activity, direction='read')
584
 
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
585
 
                          (36, 'read')], activity)
586
 
 
587
 
        from_file = StringIO(self.test_data)
588
 
        to_file = StringIO()
589
 
        del activity[:]
590
 
        osutils.pumpfile(from_file, to_file, buff_size=500,
591
 
                         report_activity=log_activity, direction='write')
592
 
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
593
 
                          (36, 'write')], activity)
594
 
 
595
 
        # And with a limited amount of data
596
 
        from_file = StringIO(self.test_data)
597
 
        to_file = StringIO()
598
 
        del activity[:]
599
 
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
600
 
                         report_activity=log_activity, direction='read')
601
 
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
602
 
 
603
 
 
604
 
 
605
 
class TestPumpStringFile(tests.TestCase):
606
 
 
607
 
    def test_empty(self):
608
 
        output = StringIO()
609
 
        osutils.pump_string_file("", output)
610
 
        self.assertEqual("", output.getvalue())
611
 
 
612
 
    def test_more_than_segment_size(self):
613
 
        output = StringIO()
614
 
        osutils.pump_string_file("123456789", output, 2)
615
 
        self.assertEqual("123456789", output.getvalue())
616
 
 
617
 
    def test_segment_size(self):
618
 
        output = StringIO()
619
 
        osutils.pump_string_file("12", output, 2)
620
 
        self.assertEqual("12", output.getvalue())
621
 
 
622
 
    def test_segment_size_multiple(self):
623
 
        output = StringIO()
624
 
        osutils.pump_string_file("1234", output, 2)
625
 
        self.assertEqual("1234", output.getvalue())
626
 
 
627
 
 
628
 
class TestRelpath(tests.TestCase):
629
 
 
630
 
    def test_simple_relpath(self):
631
 
        cwd = osutils.getcwd()
632
 
        subdir = cwd + '/subdir'
633
 
        self.assertEqual('subdir', osutils.relpath(cwd, subdir))
634
 
 
635
 
    def test_deep_relpath(self):
636
 
        cwd = osutils.getcwd()
637
 
        subdir = cwd + '/sub/subsubdir'
638
 
        self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
639
 
 
640
 
    def test_not_relative(self):
641
 
        self.assertRaises(errors.PathNotChild,
642
 
                          osutils.relpath, 'C:/path', 'H:/path')
643
 
        self.assertRaises(errors.PathNotChild,
644
 
                          osutils.relpath, 'C:/', 'H:/path')
645
 
 
646
 
 
647
 
class TestSafeUnicode(tests.TestCase):
 
134
 
 
135
class TestSafeUnicode(TestCase):
648
136
 
649
137
    def test_from_ascii_string(self):
650
138
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
659
147
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
660
148
 
661
149
    def test_bad_utf8_string(self):
662
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
150
        self.assertRaises(BzrBadParameterNotUnicode,
663
151
                          osutils.safe_unicode,
664
152
                          '\xbb\xbb')
665
153
 
666
154
 
667
 
class TestSafeUtf8(tests.TestCase):
668
 
 
669
 
    def test_from_ascii_string(self):
670
 
        f = 'foobar'
671
 
        self.assertEqual('foobar', osutils.safe_utf8(f))
672
 
 
673
 
    def test_from_unicode_string_ascii_contents(self):
674
 
        self.assertEqual('bargam', osutils.safe_utf8(u'bargam'))
675
 
 
676
 
    def test_from_unicode_string_unicode_contents(self):
677
 
        self.assertEqual('bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
678
 
 
679
 
    def test_from_utf8_string(self):
680
 
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
681
 
 
682
 
    def test_bad_utf8_string(self):
683
 
        self.assertRaises(errors.BzrBadParameterNotUnicode,
684
 
                          osutils.safe_utf8, '\xbb\xbb')
685
 
 
686
 
 
687
 
class TestSafeRevisionId(tests.TestCase):
688
 
 
689
 
    def test_from_ascii_string(self):
690
 
        # this shouldn't give a warning because it's getting an ascii string
691
 
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
692
 
 
693
 
    def test_from_unicode_string_ascii_contents(self):
694
 
        self.assertEqual('bargam',
695
 
                         osutils.safe_revision_id(u'bargam', warn=False))
696
 
 
697
 
    def test_from_unicode_deprecated(self):
698
 
        self.assertEqual('bargam',
699
 
            self.callDeprecated([osutils._revision_id_warning],
700
 
                                osutils.safe_revision_id, u'bargam'))
701
 
 
702
 
    def test_from_unicode_string_unicode_contents(self):
703
 
        self.assertEqual('bargam\xc2\xae',
704
 
                         osutils.safe_revision_id(u'bargam\xae', warn=False))
705
 
 
706
 
    def test_from_utf8_string(self):
707
 
        self.assertEqual('foo\xc2\xae',
708
 
                         osutils.safe_revision_id('foo\xc2\xae'))
709
 
 
710
 
    def test_none(self):
711
 
        """Currently, None is a valid revision_id"""
712
 
        self.assertEqual(None, osutils.safe_revision_id(None))
713
 
 
714
 
 
715
 
class TestSafeFileId(tests.TestCase):
716
 
 
717
 
    def test_from_ascii_string(self):
718
 
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
719
 
 
720
 
    def test_from_unicode_string_ascii_contents(self):
721
 
        self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
722
 
 
723
 
    def test_from_unicode_deprecated(self):
724
 
        self.assertEqual('bargam',
725
 
            self.callDeprecated([osutils._file_id_warning],
726
 
                                osutils.safe_file_id, u'bargam'))
727
 
 
728
 
    def test_from_unicode_string_unicode_contents(self):
729
 
        self.assertEqual('bargam\xc2\xae',
730
 
                         osutils.safe_file_id(u'bargam\xae', warn=False))
731
 
 
732
 
    def test_from_utf8_string(self):
733
 
        self.assertEqual('foo\xc2\xae',
734
 
                         osutils.safe_file_id('foo\xc2\xae'))
735
 
 
736
 
    def test_none(self):
737
 
        """Currently, None is a valid revision_id"""
738
 
        self.assertEqual(None, osutils.safe_file_id(None))
739
 
 
740
 
 
741
 
class TestWin32Funcs(tests.TestCase):
742
 
    """Test that _win32 versions of os utilities return appropriate paths."""
 
155
class TestWin32Funcs(TestCase):
 
156
    """Test that the _win32 versions of os utilities return appropriate paths."""
743
157
 
744
158
    def test_abspath(self):
745
159
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
746
160
        self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
747
 
        self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
748
 
        self.assertEqual('//HOST/path', osutils._win32_abspath('//HOST/path'))
749
161
 
750
162
    def test_realpath(self):
751
163
        self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
752
164
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
753
165
 
754
166
    def test_pathjoin(self):
755
 
        self.assertEqual('path/to/foo',
756
 
                         osutils._win32_pathjoin('path', 'to', 'foo'))
757
 
        self.assertEqual('C:/foo',
758
 
                         osutils._win32_pathjoin('path\\to', 'C:\\foo'))
759
 
        self.assertEqual('C:/foo',
760
 
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
761
 
        self.assertEqual('path/to/foo',
762
 
                         osutils._win32_pathjoin('path/to/', 'foo'))
763
 
        self.assertEqual('/foo',
764
 
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
765
 
        self.assertEqual('/foo',
766
 
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
167
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
 
168
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
169
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
 
170
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
 
171
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
172
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
767
173
 
768
174
    def test_normpath(self):
769
 
        self.assertEqual('path/to/foo',
770
 
                         osutils._win32_normpath(r'path\\from\..\to\.\foo'))
771
 
        self.assertEqual('path/to/foo',
772
 
                         osutils._win32_normpath('path//from/../to/./foo'))
 
175
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
176
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
773
177
 
774
178
    def test_getcwd(self):
775
 
        cwd = osutils._win32_getcwd()
776
 
        os_cwd = os.getcwdu()
777
 
        self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
778
 
        # win32 is inconsistent whether it returns lower or upper case
779
 
        # and even if it was consistent the user might type the other
780
 
        # so we force it to uppercase
781
 
        # running python.exe under cmd.exe return capital C:\\
782
 
        # running win32 python inside a cygwin shell returns lowercase
783
 
        self.assertEqual(os_cwd[0].upper(), cwd[0])
784
 
 
785
 
    def test_fixdrive(self):
786
 
        self.assertEqual('H:/foo', osutils._win32_fixdrive('h:/foo'))
787
 
        self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
788
 
        self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
789
 
 
790
 
    def test_win98_abspath(self):
791
 
        # absolute path
792
 
        self.assertEqual('C:/foo', osutils._win98_abspath('C:\\foo'))
793
 
        self.assertEqual('C:/foo', osutils._win98_abspath('C:/foo'))
794
 
        # UNC path
795
 
        self.assertEqual('//HOST/path', osutils._win98_abspath(r'\\HOST\path'))
796
 
        self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
797
 
        # relative path
798
 
        cwd = osutils.getcwd().rstrip('/')
799
 
        drive = osutils._nt_splitdrive(cwd)[0]
800
 
        self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
801
 
        self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
802
 
        # unicode path
803
 
        u = u'\u1234'
804
 
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
805
 
 
806
 
 
807
 
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
 
179
        self.assertEqual(os.getcwdu().replace('\\', '/'), osutils._win32_getcwd())
 
180
 
 
181
 
 
182
class TestWin32FuncsDirs(TestCaseInTempDir):
808
183
    """Test win32 functions that create files."""
 
184
    
 
185
    def test_getcwd(self):
 
186
        # Make sure getcwd can handle unicode filenames
 
187
        try:
 
188
            os.mkdir(u'B\xe5gfors')
 
189
        except UnicodeError:
 
190
            raise TestSkipped("Unable to create Unicode filename")
809
191
 
810
 
    def test_getcwd(self):
811
 
        self.requireFeature(tests.UnicodeFilenameFeature)
812
 
        os.mkdir(u'mu-\xb5')
813
 
        os.chdir(u'mu-\xb5')
 
192
        os.chdir(u'B\xe5gfors')
814
193
        # TODO: jam 20060427 This will probably fail on Mac OSX because
815
194
        #       it will change the normalization of B\xe5gfors
816
195
        #       Consider using a different unicode character, or make
817
196
        #       osutils.getcwd() renormalize the path.
818
 
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
819
 
 
820
 
    def test_minimum_path_selection(self):
821
 
        self.assertEqual(set(),
822
 
            osutils.minimum_path_selection([]))
823
 
        self.assertEqual(set(['a']),
824
 
            osutils.minimum_path_selection(['a']))
825
 
        self.assertEqual(set(['a', 'b']),
826
 
            osutils.minimum_path_selection(['a', 'b']))
827
 
        self.assertEqual(set(['a/', 'b']),
828
 
            osutils.minimum_path_selection(['a/', 'b']))
829
 
        self.assertEqual(set(['a/', 'b']),
830
 
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
831
 
        self.assertEqual(set(['a-b', 'a', 'a0b']),
832
 
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
 
197
        self.assertTrue(osutils._win32_getcwd().endswith(u'/B\xe5gfors'))
833
198
 
834
199
    def test_mkdtemp(self):
835
200
        tmpdir = osutils._win32_mkdtemp(dir='.')
848
213
        self.failIfExists('b')
849
214
        self.assertFileEqual('baz\n', 'a')
850
215
 
851
 
    def test_rename_missing_file(self):
852
 
        a = open('a', 'wb')
853
 
        a.write('foo\n')
854
 
        a.close()
855
 
 
856
 
        try:
857
 
            osutils._win32_rename('b', 'a')
858
 
        except (IOError, OSError), e:
859
 
            self.assertEqual(errno.ENOENT, e.errno)
860
 
        self.assertFileEqual('foo\n', 'a')
861
 
 
862
 
    def test_rename_missing_dir(self):
863
 
        os.mkdir('a')
864
 
        try:
865
 
            osutils._win32_rename('b', 'a')
866
 
        except (IOError, OSError), e:
867
 
            self.assertEqual(errno.ENOENT, e.errno)
868
 
 
869
 
    def test_rename_current_dir(self):
870
 
        os.mkdir('a')
871
 
        os.chdir('a')
872
 
        # You can't rename the working directory
873
 
        # doing rename non-existant . usually
874
 
        # just raises ENOENT, since non-existant
875
 
        # doesn't exist.
876
 
        try:
877
 
            osutils._win32_rename('b', '.')
878
 
        except (IOError, OSError), e:
879
 
            self.assertEqual(errno.ENOENT, e.errno)
880
 
 
881
 
    def test_splitpath(self):
882
 
        def check(expected, path):
883
 
            self.assertEqual(expected, osutils.splitpath(path))
884
 
 
885
 
        check(['a'], 'a')
886
 
        check(['a', 'b'], 'a/b')
887
 
        check(['a', 'b'], 'a/./b')
888
 
        check(['a', '.b'], 'a/.b')
889
 
        check(['a', '.b'], 'a\\.b')
890
 
 
891
 
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
892
 
 
893
 
 
894
 
class TestParentDirectories(tests.TestCaseInTempDir):
895
 
    """Test osutils.parent_directories()"""
896
 
 
897
 
    def test_parent_directories(self):
898
 
        self.assertEqual([], osutils.parent_directories('a'))
899
 
        self.assertEqual(['a'], osutils.parent_directories('a/b'))
900
 
        self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
901
 
 
902
 
 
903
 
class TestMacFuncsDirs(tests.TestCaseInTempDir):
904
 
    """Test mac special functions that require directories."""
905
 
 
906
 
    def test_getcwd(self):
907
 
        self.requireFeature(tests.UnicodeFilenameFeature)
908
 
        os.mkdir(u'B\xe5gfors')
909
 
        os.chdir(u'B\xe5gfors')
910
 
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
911
 
 
912
 
    def test_getcwd_nonnorm(self):
913
 
        self.requireFeature(tests.UnicodeFilenameFeature)
914
 
        # Test that _mac_getcwd() will normalize this path
915
 
        os.mkdir(u'Ba\u030agfors')
916
 
        os.chdir(u'Ba\u030agfors')
917
 
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
918
 
 
919
 
 
920
 
class TestChunksToLines(tests.TestCase):
921
 
 
922
 
    def test_smoketest(self):
923
 
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
924
 
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
925
 
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
926
 
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
927
 
 
928
 
    def test_osutils_binding(self):
929
 
        from bzrlib.tests import test__chunks_to_lines
930
 
        if test__chunks_to_lines.CompiledChunksToLinesFeature.available():
931
 
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
932
 
        else:
933
 
            from bzrlib._chunks_to_lines_py import chunks_to_lines
934
 
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
935
 
 
936
 
 
937
 
class TestSplitLines(tests.TestCase):
 
216
 
 
217
class TestSplitLines(TestCase):
938
218
 
939
219
    def test_split_unicode(self):
940
220
        self.assertEqual([u'foo\n', u'bar\xae'],
947
227
                         osutils.split_lines('foo\rbar\n'))
948
228
 
949
229
 
950
 
class TestWalkDirs(tests.TestCaseInTempDir):
951
 
 
952
 
    def assertExpectedBlocks(self, expected, result):
953
 
        self.assertEqual(expected,
954
 
                         [(dirinfo, [line[0:3] for line in block])
955
 
                          for dirinfo, block in result])
 
230
class TestWalkDirs(TestCaseInTempDir):
956
231
 
957
232
    def test_walkdirs(self):
958
233
        tree = [
965
240
            ]
966
241
        self.build_tree(tree)
967
242
        expected_dirblocks = [
968
 
                (('', '.'),
969
 
                 [('0file', '0file', 'file'),
970
 
                  ('1dir', '1dir', 'directory'),
971
 
                  ('2file', '2file', 'file'),
972
 
                 ]
973
 
                ),
974
 
                (('1dir', './1dir'),
975
 
                 [('1dir/0file', '0file', 'file'),
976
 
                  ('1dir/1dir', '1dir', 'directory'),
977
 
                 ]
978
 
                ),
979
 
                (('1dir/1dir', './1dir/1dir'),
980
 
                 [
981
 
                 ]
982
 
                ),
983
 
            ]
984
 
        result = []
985
 
        found_bzrdir = False
986
 
        for dirdetail, dirblock in osutils.walkdirs('.'):
987
 
            if len(dirblock) and dirblock[0][1] == '.bzr':
988
 
                # this tests the filtering of selected paths
989
 
                found_bzrdir = True
990
 
                del dirblock[0]
991
 
            result.append((dirdetail, dirblock))
992
 
 
993
 
        self.assertTrue(found_bzrdir)
994
 
        self.assertExpectedBlocks(expected_dirblocks, result)
995
 
        # you can search a subdir only, with a supplied prefix.
996
 
        result = []
997
 
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
998
 
            result.append(dirblock)
999
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
1000
 
 
1001
 
    def test_walkdirs_os_error(self):
1002
 
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
1003
 
        # Pyrex readdir didn't raise useful messages if it had an error
1004
 
        # reading the directory
1005
 
        if sys.platform == 'win32':
1006
 
            raise tests.TestNotApplicable(
1007
 
                "readdir IOError not tested on win32")
1008
 
        os.mkdir("test-unreadable")
1009
 
        os.chmod("test-unreadable", 0000)
1010
 
        # must chmod it back so that it can be removed
1011
 
        self.addCleanup(os.chmod, "test-unreadable", 0700)
1012
 
        # The error is not raised until the generator is actually evaluated.
1013
 
        # (It would be ok if it happened earlier but at the moment it
1014
 
        # doesn't.)
1015
 
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1016
 
        self.assertEquals('./test-unreadable', e.filename)
1017
 
        self.assertEquals(errno.EACCES, e.errno)
1018
 
        # Ensure the message contains the file name
1019
 
        self.assertContainsRe(str(e), "\./test-unreadable")
1020
 
 
1021
 
    def test__walkdirs_utf8(self):
1022
 
        tree = [
1023
 
            '.bzr',
1024
 
            '0file',
1025
 
            '1dir/',
1026
 
            '1dir/0file',
1027
 
            '1dir/1dir/',
1028
 
            '2file'
1029
 
            ]
1030
 
        self.build_tree(tree)
1031
 
        expected_dirblocks = [
1032
 
                (('', '.'),
1033
 
                 [('0file', '0file', 'file'),
1034
 
                  ('1dir', '1dir', 'directory'),
1035
 
                  ('2file', '2file', 'file'),
1036
 
                 ]
1037
 
                ),
1038
 
                (('1dir', './1dir'),
1039
 
                 [('1dir/0file', '0file', 'file'),
1040
 
                  ('1dir/1dir', '1dir', 'directory'),
1041
 
                 ]
1042
 
                ),
1043
 
                (('1dir/1dir', './1dir/1dir'),
1044
 
                 [
1045
 
                 ]
1046
 
                ),
1047
 
            ]
1048
 
        result = []
1049
 
        found_bzrdir = False
1050
 
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1051
 
            if len(dirblock) and dirblock[0][1] == '.bzr':
1052
 
                # this tests the filtering of selected paths
1053
 
                found_bzrdir = True
1054
 
                del dirblock[0]
1055
 
            result.append((dirdetail, dirblock))
1056
 
 
1057
 
        self.assertTrue(found_bzrdir)
1058
 
        self.assertExpectedBlocks(expected_dirblocks, result)
1059
 
 
1060
 
        # you can search a subdir only, with a supplied prefix.
1061
 
        result = []
1062
 
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
1063
 
            result.append(dirblock)
1064
 
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
1065
 
 
1066
 
    def _filter_out_stat(self, result):
1067
 
        """Filter out the stat value from the walkdirs result"""
1068
 
        for dirdetail, dirblock in result:
1069
 
            new_dirblock = []
1070
 
            for info in dirblock:
1071
 
                # Ignore info[3] which is the stat
1072
 
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1073
 
            dirblock[:] = new_dirblock
1074
 
 
1075
 
    def _save_platform_info(self):
1076
 
        cur_winver = win32utils.winver
1077
 
        cur_fs_enc = osutils._fs_enc
1078
 
        cur_dir_reader = osutils._selected_dir_reader
1079
 
        def restore():
1080
 
            win32utils.winver = cur_winver
1081
 
            osutils._fs_enc = cur_fs_enc
1082
 
            osutils._selected_dir_reader = cur_dir_reader
1083
 
        self.addCleanup(restore)
1084
 
 
1085
 
    def assertDirReaderIs(self, expected):
1086
 
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1087
 
        # Force it to redetect
1088
 
        osutils._selected_dir_reader = None
1089
 
        # Nothing to list, but should still trigger the selection logic
1090
 
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1091
 
        self.assertIsInstance(osutils._selected_dir_reader, expected)
1092
 
 
1093
 
    def test_force_walkdirs_utf8_fs_utf8(self):
1094
 
        self.requireFeature(UTF8DirReaderFeature)
1095
 
        self._save_platform_info()
1096
 
        win32utils.winver = None # Avoid the win32 detection code
1097
 
        osutils._fs_enc = 'UTF-8'
1098
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1099
 
 
1100
 
    def test_force_walkdirs_utf8_fs_ascii(self):
1101
 
        self.requireFeature(UTF8DirReaderFeature)
1102
 
        self._save_platform_info()
1103
 
        win32utils.winver = None # Avoid the win32 detection code
1104
 
        osutils._fs_enc = 'US-ASCII'
1105
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1106
 
 
1107
 
    def test_force_walkdirs_utf8_fs_ANSI(self):
1108
 
        self.requireFeature(UTF8DirReaderFeature)
1109
 
        self._save_platform_info()
1110
 
        win32utils.winver = None # Avoid the win32 detection code
1111
 
        osutils._fs_enc = 'ANSI_X3.4-1968'
1112
 
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1113
 
 
1114
 
    def test_force_walkdirs_utf8_fs_latin1(self):
1115
 
        self._save_platform_info()
1116
 
        win32utils.winver = None # Avoid the win32 detection code
1117
 
        osutils._fs_enc = 'latin1'
1118
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1119
 
 
1120
 
    def test_force_walkdirs_utf8_nt(self):
1121
 
        # Disabled because the thunk of the whole walkdirs api is disabled.
1122
 
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1123
 
        self._save_platform_info()
1124
 
        win32utils.winver = 'Windows NT'
1125
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1126
 
        self.assertDirReaderIs(Win32ReadDir)
1127
 
 
1128
 
    def test_force_walkdirs_utf8_98(self):
1129
 
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1130
 
        self._save_platform_info()
1131
 
        win32utils.winver = 'Windows 98'
1132
 
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1133
 
 
1134
 
    def test_unicode_walkdirs(self):
1135
 
        """Walkdirs should always return unicode paths."""
1136
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1137
 
        name0 = u'0file-\xb6'
1138
 
        name1 = u'1dir-\u062c\u0648'
1139
 
        name2 = u'2file-\u0633'
1140
 
        tree = [
1141
 
            name0,
1142
 
            name1 + '/',
1143
 
            name1 + '/' + name0,
1144
 
            name1 + '/' + name1 + '/',
1145
 
            name2,
1146
 
            ]
1147
 
        self.build_tree(tree)
1148
 
        expected_dirblocks = [
1149
 
                ((u'', u'.'),
1150
 
                 [(name0, name0, 'file', './' + name0),
1151
 
                  (name1, name1, 'directory', './' + name1),
1152
 
                  (name2, name2, 'file', './' + name2),
1153
 
                 ]
1154
 
                ),
1155
 
                ((name1, './' + name1),
1156
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1
1157
 
                                                        + '/' + name0),
1158
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1
1159
 
                                                            + '/' + name1),
1160
 
                 ]
1161
 
                ),
1162
 
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
1163
 
                 [
1164
 
                 ]
1165
 
                ),
1166
 
            ]
1167
 
        result = list(osutils.walkdirs('.'))
1168
 
        self._filter_out_stat(result)
1169
 
        self.assertEqual(expected_dirblocks, result)
1170
 
        result = list(osutils.walkdirs(u'./'+name1, name1))
1171
 
        self._filter_out_stat(result)
1172
 
        self.assertEqual(expected_dirblocks[1:], result)
1173
 
 
1174
 
    def test_unicode__walkdirs_utf8(self):
1175
 
        """Walkdirs_utf8 should always return utf8 paths.
1176
 
 
1177
 
        The abspath portion might be in unicode or utf-8
1178
 
        """
1179
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1180
 
        name0 = u'0file-\xb6'
1181
 
        name1 = u'1dir-\u062c\u0648'
1182
 
        name2 = u'2file-\u0633'
1183
 
        tree = [
1184
 
            name0,
1185
 
            name1 + '/',
1186
 
            name1 + '/' + name0,
1187
 
            name1 + '/' + name1 + '/',
1188
 
            name2,
1189
 
            ]
1190
 
        self.build_tree(tree)
1191
 
        name0 = name0.encode('utf8')
1192
 
        name1 = name1.encode('utf8')
1193
 
        name2 = name2.encode('utf8')
1194
 
 
1195
 
        expected_dirblocks = [
1196
 
                (('', '.'),
1197
 
                 [(name0, name0, 'file', './' + name0),
1198
 
                  (name1, name1, 'directory', './' + name1),
1199
 
                  (name2, name2, 'file', './' + name2),
1200
 
                 ]
1201
 
                ),
1202
 
                ((name1, './' + name1),
1203
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1
1204
 
                                                        + '/' + name0),
1205
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1
1206
 
                                                            + '/' + name1),
1207
 
                 ]
1208
 
                ),
1209
 
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
1210
 
                 [
1211
 
                 ]
1212
 
                ),
1213
 
            ]
1214
 
        result = []
1215
 
        # For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1216
 
        # all abspaths are Unicode, and encode them back into utf8.
1217
 
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1218
 
            self.assertIsInstance(dirdetail[0], str)
1219
 
            if isinstance(dirdetail[1], unicode):
1220
 
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1221
 
                dirblock = [list(info) for info in dirblock]
1222
 
                for info in dirblock:
1223
 
                    self.assertIsInstance(info[4], unicode)
1224
 
                    info[4] = info[4].encode('utf8')
1225
 
            new_dirblock = []
1226
 
            for info in dirblock:
1227
 
                self.assertIsInstance(info[0], str)
1228
 
                self.assertIsInstance(info[1], str)
1229
 
                self.assertIsInstance(info[4], str)
1230
 
                # Remove the stat information
1231
 
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1232
 
            result.append((dirdetail, new_dirblock))
1233
 
        self.assertEqual(expected_dirblocks, result)
1234
 
 
1235
 
    def test__walkdirs_utf8_with_unicode_fs(self):
1236
 
        """UnicodeDirReader should be a safe fallback everywhere
1237
 
 
1238
 
        The abspath portion should be in unicode
1239
 
        """
1240
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1241
 
        # Use the unicode reader. TODO: split into driver-and-driven unit
1242
 
        # tests.
1243
 
        self._save_platform_info()
1244
 
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
1245
 
        name0u = u'0file-\xb6'
1246
 
        name1u = u'1dir-\u062c\u0648'
1247
 
        name2u = u'2file-\u0633'
1248
 
        tree = [
1249
 
            name0u,
1250
 
            name1u + '/',
1251
 
            name1u + '/' + name0u,
1252
 
            name1u + '/' + name1u + '/',
1253
 
            name2u,
1254
 
            ]
1255
 
        self.build_tree(tree)
1256
 
        name0 = name0u.encode('utf8')
1257
 
        name1 = name1u.encode('utf8')
1258
 
        name2 = name2u.encode('utf8')
1259
 
 
1260
 
        # All of the abspaths should be in unicode, all of the relative paths
1261
 
        # should be in utf8
1262
 
        expected_dirblocks = [
1263
 
                (('', '.'),
1264
 
                 [(name0, name0, 'file', './' + name0u),
1265
 
                  (name1, name1, 'directory', './' + name1u),
1266
 
                  (name2, name2, 'file', './' + name2u),
1267
 
                 ]
1268
 
                ),
1269
 
                ((name1, './' + name1u),
1270
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1271
 
                                                        + '/' + name0u),
1272
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1273
 
                                                            + '/' + name1u),
1274
 
                 ]
1275
 
                ),
1276
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1277
 
                 [
1278
 
                 ]
1279
 
                ),
1280
 
            ]
1281
 
        result = list(osutils._walkdirs_utf8('.'))
1282
 
        self._filter_out_stat(result)
1283
 
        self.assertEqual(expected_dirblocks, result)
1284
 
 
1285
 
    def test__walkdirs_utf8_win32readdir(self):
1286
 
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1287
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1288
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1289
 
        self._save_platform_info()
1290
 
        osutils._selected_dir_reader = Win32ReadDir()
1291
 
        name0u = u'0file-\xb6'
1292
 
        name1u = u'1dir-\u062c\u0648'
1293
 
        name2u = u'2file-\u0633'
1294
 
        tree = [
1295
 
            name0u,
1296
 
            name1u + '/',
1297
 
            name1u + '/' + name0u,
1298
 
            name1u + '/' + name1u + '/',
1299
 
            name2u,
1300
 
            ]
1301
 
        self.build_tree(tree)
1302
 
        name0 = name0u.encode('utf8')
1303
 
        name1 = name1u.encode('utf8')
1304
 
        name2 = name2u.encode('utf8')
1305
 
 
1306
 
        # All of the abspaths should be in unicode, all of the relative paths
1307
 
        # should be in utf8
1308
 
        expected_dirblocks = [
1309
 
                (('', '.'),
1310
 
                 [(name0, name0, 'file', './' + name0u),
1311
 
                  (name1, name1, 'directory', './' + name1u),
1312
 
                  (name2, name2, 'file', './' + name2u),
1313
 
                 ]
1314
 
                ),
1315
 
                ((name1, './' + name1u),
1316
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1317
 
                                                        + '/' + name0u),
1318
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1319
 
                                                            + '/' + name1u),
1320
 
                 ]
1321
 
                ),
1322
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1323
 
                 [
1324
 
                 ]
1325
 
                ),
1326
 
            ]
1327
 
        result = list(osutils._walkdirs_utf8(u'.'))
1328
 
        self._filter_out_stat(result)
1329
 
        self.assertEqual(expected_dirblocks, result)
1330
 
 
1331
 
    def assertStatIsCorrect(self, path, win32stat):
1332
 
        os_stat = os.stat(path)
1333
 
        self.assertEqual(os_stat.st_size, win32stat.st_size)
1334
 
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1335
 
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1336
 
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1337
 
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1338
 
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1339
 
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1340
 
 
1341
 
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1342
 
        """make sure our Stat values are valid"""
1343
 
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1344
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1345
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1346
 
        name0u = u'0file-\xb6'
1347
 
        name0 = name0u.encode('utf8')
1348
 
        self.build_tree([name0u])
1349
 
        # I hate to sleep() here, but I'm trying to make the ctime different
1350
 
        # from the mtime
1351
 
        time.sleep(2)
1352
 
        f = open(name0u, 'ab')
1353
 
        try:
1354
 
            f.write('just a small update')
1355
 
        finally:
1356
 
            f.close()
1357
 
 
1358
 
        result = Win32ReadDir().read_dir('', u'.')
1359
 
        entry = result[0]
1360
 
        self.assertEqual((name0, name0, 'file'), entry[:3])
1361
 
        self.assertEqual(u'./' + name0u, entry[4])
1362
 
        self.assertStatIsCorrect(entry[4], entry[3])
1363
 
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1364
 
 
1365
 
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1366
 
        """make sure our Stat values are valid"""
1367
 
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1368
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1369
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1370
 
        name0u = u'0dir-\u062c\u0648'
1371
 
        name0 = name0u.encode('utf8')
1372
 
        self.build_tree([name0u + '/'])
1373
 
 
1374
 
        result = Win32ReadDir().read_dir('', u'.')
1375
 
        entry = result[0]
1376
 
        self.assertEqual((name0, name0, 'directory'), entry[:3])
1377
 
        self.assertEqual(u'./' + name0u, entry[4])
1378
 
        self.assertStatIsCorrect(entry[4], entry[3])
 
243
                [
 
244
                    ('0file', '0file', 'file'),
 
245
                    ('1dir', '1dir', 'directory'),
 
246
                    ('2file', '2file', 'file'),
 
247
                ],
 
248
                [
 
249
                    ('1dir/0file', '0file', 'file'),
 
250
                    ('1dir/1dir', '1dir', 'directory'),
 
251
                ],
 
252
                [
 
253
                ],
 
254
            ]
 
255
        result = []
 
256
        found_bzrdir = False
 
257
        for dirblock in osutils.walkdirs('.'):
 
258
            if len(dirblock) and dirblock[0][1] == '.bzr':
 
259
                # this tests the filtering of selected paths
 
260
                found_bzrdir = True
 
261
                del dirblock[0]
 
262
            result.append(dirblock)
 
263
 
 
264
        self.assertTrue(found_bzrdir)
 
265
        self.assertEqual(expected_dirblocks,
 
266
            [[line[0:3] for line in block] for block in result])
 
267
        # you can search a subdir only, with a supplied prefix.
 
268
        result = []
 
269
        for dirblock in osutils.walkdirs('1dir', '1dir'):
 
270
            result.append(dirblock)
 
271
        self.assertEqual(expected_dirblocks[1:],
 
272
            [[line[0:3] for line in block] for block in result])
1379
273
 
1380
274
    def assertPathCompare(self, path_less, path_greater):
1381
275
        """check that path_less and path_greater compare correctly."""
1455
349
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1456
350
 
1457
351
 
1458
 
class TestCopyTree(tests.TestCaseInTempDir):
1459
 
 
1460
 
    def test_copy_basic_tree(self):
1461
 
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1462
 
        osutils.copy_tree('source', 'target')
1463
 
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
1464
 
        self.assertEqual(['c'], os.listdir('target/b'))
1465
 
 
1466
 
    def test_copy_tree_target_exists(self):
1467
 
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c',
1468
 
                         'target/'])
1469
 
        osutils.copy_tree('source', 'target')
1470
 
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
1471
 
        self.assertEqual(['c'], os.listdir('target/b'))
1472
 
 
1473
 
    def test_copy_tree_symlinks(self):
1474
 
        self.requireFeature(tests.SymlinkFeature)
1475
 
        self.build_tree(['source/'])
1476
 
        os.symlink('a/generic/path', 'source/lnk')
1477
 
        osutils.copy_tree('source', 'target')
1478
 
        self.assertEqual(['lnk'], os.listdir('target'))
1479
 
        self.assertEqual('a/generic/path', os.readlink('target/lnk'))
1480
 
 
1481
 
    def test_copy_tree_handlers(self):
1482
 
        processed_files = []
1483
 
        processed_links = []
1484
 
        def file_handler(from_path, to_path):
1485
 
            processed_files.append(('f', from_path, to_path))
1486
 
        def dir_handler(from_path, to_path):
1487
 
            processed_files.append(('d', from_path, to_path))
1488
 
        def link_handler(from_path, to_path):
1489
 
            processed_links.append((from_path, to_path))
1490
 
        handlers = {'file':file_handler,
1491
 
                    'directory':dir_handler,
1492
 
                    'symlink':link_handler,
1493
 
                   }
1494
 
 
1495
 
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1496
 
        if osutils.has_symlinks():
1497
 
            os.symlink('a/generic/path', 'source/lnk')
1498
 
        osutils.copy_tree('source', 'target', handlers=handlers)
1499
 
 
1500
 
        self.assertEqual([('d', 'source', 'target'),
1501
 
                          ('f', 'source/a', 'target/a'),
1502
 
                          ('d', 'source/b', 'target/b'),
1503
 
                          ('f', 'source/b/c', 'target/b/c'),
1504
 
                         ], processed_files)
1505
 
        self.failIfExists('target')
1506
 
        if osutils.has_symlinks():
1507
 
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1508
 
 
1509
 
 
1510
 
class TestSetUnsetEnv(tests.TestCase):
1511
 
    """Test updating the environment"""
1512
 
 
1513
 
    def setUp(self):
1514
 
        super(TestSetUnsetEnv, self).setUp()
1515
 
 
1516
 
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
1517
 
                         'Environment was not cleaned up properly.'
1518
 
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
1519
 
        def cleanup():
1520
 
            if 'BZR_TEST_ENV_VAR' in os.environ:
1521
 
                del os.environ['BZR_TEST_ENV_VAR']
1522
 
 
1523
 
        self.addCleanup(cleanup)
1524
 
 
1525
 
    def test_set(self):
1526
 
        """Test that we can set an env variable"""
1527
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1528
 
        self.assertEqual(None, old)
1529
 
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
1530
 
 
1531
 
    def test_double_set(self):
1532
 
        """Test that we get the old value out"""
1533
 
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1534
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
1535
 
        self.assertEqual('foo', old)
1536
 
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
1537
 
 
1538
 
    def test_unicode(self):
1539
 
        """Environment can only contain plain strings
1540
 
 
1541
 
        So Unicode strings must be encoded.
1542
 
        """
1543
 
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
1544
 
        if uni_val is None:
1545
 
            raise tests.TestSkipped(
1546
 
                'Cannot find a unicode character that works in encoding %s'
1547
 
                % (osutils.get_user_encoding(),))
1548
 
 
1549
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1550
 
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1551
 
 
1552
 
    def test_unset(self):
1553
 
        """Test that passing None will remove the env var"""
1554
 
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1555
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1556
 
        self.assertEqual('foo', old)
1557
 
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1558
 
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1559
 
 
1560
 
 
1561
 
class TestSizeShaFile(tests.TestCaseInTempDir):
1562
 
 
1563
 
    def test_sha_empty(self):
1564
 
        self.build_tree_contents([('foo', '')])
1565
 
        expected_sha = osutils.sha_string('')
1566
 
        f = open('foo')
1567
 
        self.addCleanup(f.close)
1568
 
        size, sha = osutils.size_sha_file(f)
1569
 
        self.assertEqual(0, size)
1570
 
        self.assertEqual(expected_sha, sha)
1571
 
 
1572
 
    def test_sha_mixed_endings(self):
1573
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1574
 
        self.build_tree_contents([('foo', text)])
1575
 
        expected_sha = osutils.sha_string(text)
1576
 
        f = open('foo')
1577
 
        self.addCleanup(f.close)
1578
 
        size, sha = osutils.size_sha_file(f)
1579
 
        self.assertEqual(38, size)
1580
 
        self.assertEqual(expected_sha, sha)
1581
 
 
1582
 
 
1583
 
class TestShaFileByName(tests.TestCaseInTempDir):
1584
 
 
1585
 
    def test_sha_empty(self):
1586
 
        self.build_tree_contents([('foo', '')])
1587
 
        expected_sha = osutils.sha_string('')
1588
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1589
 
 
1590
 
    def test_sha_mixed_endings(self):
1591
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1592
 
        self.build_tree_contents([('foo', text)])
1593
 
        expected_sha = osutils.sha_string(text)
1594
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1595
 
 
1596
 
 
1597
 
class TestResourceLoading(tests.TestCaseInTempDir):
1598
 
 
1599
 
    def test_resource_string(self):
1600
 
        # test resource in bzrlib
1601
 
        text = osutils.resource_string('bzrlib', 'debug.py')
1602
 
        self.assertContainsRe(text, "debug_flags = set()")
1603
 
        # test resource under bzrlib
1604
 
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1605
 
        self.assertContainsRe(text, "class TextUIFactory")
1606
 
        # test unsupported package
1607
 
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1608
 
            'yyy.xx')
1609
 
        # test unknown resource
1610
 
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1611
 
 
1612
 
 
1613
 
class TestReCompile(tests.TestCase):
1614
 
 
1615
 
    def test_re_compile_checked(self):
1616
 
        r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1617
 
        self.assertTrue(r.match('aaaa'))
1618
 
        self.assertTrue(r.match('aAaA'))
1619
 
 
1620
 
    def test_re_compile_checked_error(self):
1621
 
        # like https://bugs.launchpad.net/bzr/+bug/251352
1622
 
        err = self.assertRaises(
1623
 
            errors.BzrCommandError,
1624
 
            osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1625
 
        self.assertEqual(
1626
 
            "Invalid regular expression in test case: '*': "
1627
 
            "nothing to repeat",
1628
 
            str(err))
1629
 
 
1630
 
 
1631
 
class TestDirReader(tests.TestCaseInTempDir):
1632
 
 
1633
 
    # Set by load_tests
1634
 
    _dir_reader_class = None
1635
 
    _native_to_unicode = None
1636
 
 
1637
 
    def setUp(self):
1638
 
        tests.TestCaseInTempDir.setUp(self)
1639
 
 
1640
 
        # Save platform specific info and reset it
1641
 
        cur_dir_reader = osutils._selected_dir_reader
1642
 
 
1643
 
        def restore():
1644
 
            osutils._selected_dir_reader = cur_dir_reader
1645
 
        self.addCleanup(restore)
1646
 
 
1647
 
        osutils._selected_dir_reader = self._dir_reader_class()
1648
 
 
1649
 
    def _get_ascii_tree(self):
1650
 
        tree = [
1651
 
            '0file',
1652
 
            '1dir/',
1653
 
            '1dir/0file',
1654
 
            '1dir/1dir/',
1655
 
            '2file'
1656
 
            ]
1657
 
        expected_dirblocks = [
1658
 
                (('', '.'),
1659
 
                 [('0file', '0file', 'file'),
1660
 
                  ('1dir', '1dir', 'directory'),
1661
 
                  ('2file', '2file', 'file'),
1662
 
                 ]
1663
 
                ),
1664
 
                (('1dir', './1dir'),
1665
 
                 [('1dir/0file', '0file', 'file'),
1666
 
                  ('1dir/1dir', '1dir', 'directory'),
1667
 
                 ]
1668
 
                ),
1669
 
                (('1dir/1dir', './1dir/1dir'),
1670
 
                 [
1671
 
                 ]
1672
 
                ),
1673
 
            ]
1674
 
        return tree, expected_dirblocks
1675
 
 
1676
 
    def test_walk_cur_dir(self):
1677
 
        tree, expected_dirblocks = self._get_ascii_tree()
1678
 
        self.build_tree(tree)
1679
 
        result = list(osutils._walkdirs_utf8('.'))
1680
 
        # Filter out stat and abspath
1681
 
        self.assertEqual(expected_dirblocks,
1682
 
                         [(dirinfo, [line[0:3] for line in block])
1683
 
                          for dirinfo, block in result])
1684
 
 
1685
 
    def test_walk_sub_dir(self):
1686
 
        tree, expected_dirblocks = self._get_ascii_tree()
1687
 
        self.build_tree(tree)
1688
 
        # you can search a subdir only, with a supplied prefix.
1689
 
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1690
 
        # Filter out stat and abspath
1691
 
        self.assertEqual(expected_dirblocks[1:],
1692
 
                         [(dirinfo, [line[0:3] for line in block])
1693
 
                          for dirinfo, block in result])
1694
 
 
1695
 
    def _get_unicode_tree(self):
1696
 
        name0u = u'0file-\xb6'
1697
 
        name1u = u'1dir-\u062c\u0648'
1698
 
        name2u = u'2file-\u0633'
1699
 
        tree = [
1700
 
            name0u,
1701
 
            name1u + '/',
1702
 
            name1u + '/' + name0u,
1703
 
            name1u + '/' + name1u + '/',
1704
 
            name2u,
1705
 
            ]
1706
 
        name0 = name0u.encode('UTF-8')
1707
 
        name1 = name1u.encode('UTF-8')
1708
 
        name2 = name2u.encode('UTF-8')
1709
 
        expected_dirblocks = [
1710
 
                (('', '.'),
1711
 
                 [(name0, name0, 'file', './' + name0u),
1712
 
                  (name1, name1, 'directory', './' + name1u),
1713
 
                  (name2, name2, 'file', './' + name2u),
1714
 
                 ]
1715
 
                ),
1716
 
                ((name1, './' + name1u),
1717
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1718
 
                                                        + '/' + name0u),
1719
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1720
 
                                                            + '/' + name1u),
1721
 
                 ]
1722
 
                ),
1723
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1724
 
                 [
1725
 
                 ]
1726
 
                ),
1727
 
            ]
1728
 
        return tree, expected_dirblocks
1729
 
 
1730
 
    def _filter_out(self, raw_dirblocks):
1731
 
        """Filter out a walkdirs_utf8 result.
1732
 
 
1733
 
        stat field is removed, all native paths are converted to unicode
1734
 
        """
1735
 
        filtered_dirblocks = []
1736
 
        for dirinfo, block in raw_dirblocks:
1737
 
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1738
 
            details = []
1739
 
            for line in block:
1740
 
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1741
 
            filtered_dirblocks.append((dirinfo, details))
1742
 
        return filtered_dirblocks
1743
 
 
1744
 
    def test_walk_unicode_tree(self):
1745
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1746
 
        tree, expected_dirblocks = self._get_unicode_tree()
1747
 
        self.build_tree(tree)
1748
 
        result = list(osutils._walkdirs_utf8('.'))
1749
 
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1750
 
 
1751
 
    def test_symlink(self):
1752
 
        self.requireFeature(tests.SymlinkFeature)
1753
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1754
 
        target = u'target\N{Euro Sign}'
1755
 
        link_name = u'l\N{Euro Sign}nk'
1756
 
        os.symlink(target, link_name)
1757
 
        target_utf8 = target.encode('UTF-8')
1758
 
        link_name_utf8 = link_name.encode('UTF-8')
1759
 
        expected_dirblocks = [
1760
 
                (('', '.'),
1761
 
                 [(link_name_utf8, link_name_utf8,
1762
 
                   'symlink', './' + link_name),],
1763
 
                 )]
1764
 
        result = list(osutils._walkdirs_utf8('.'))
1765
 
        self.assertEqual(expected_dirblocks, self._filter_out(result))
1766
 
 
1767
 
 
1768
 
class TestReadLink(tests.TestCaseInTempDir):
1769
 
    """Exposes os.readlink() problems and the osutils solution.
1770
 
 
1771
 
    The only guarantee offered by os.readlink(), starting with 2.6, is that a
1772
 
    unicode string will be returned if a unicode string is passed.
1773
 
 
1774
 
    But prior python versions failed to properly encode the passed unicode
1775
 
    string.
1776
 
    """
1777
 
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1778
 
 
1779
 
    def setUp(self):
1780
 
        super(tests.TestCaseInTempDir, self).setUp()
1781
 
        self.link = u'l\N{Euro Sign}ink'
1782
 
        self.target = u'targe\N{Euro Sign}t'
1783
 
        os.symlink(self.target, self.link)
1784
 
 
1785
 
    def test_os_readlink_link_encoding(self):
1786
 
        if sys.version_info < (2, 6):
1787
 
            self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1788
 
        else:
1789
 
            self.assertEquals(self.target,  os.readlink(self.link))
1790
 
 
1791
 
    def test_os_readlink_link_decoding(self):
1792
 
        self.assertEquals(self.target.encode(osutils._fs_enc),
1793
 
                          os.readlink(self.link.encode(osutils._fs_enc)))
1794
 
 
1795
 
 
1796
 
class TestConcurrency(tests.TestCase):
1797
 
 
1798
 
    def test_local_concurrency(self):
1799
 
        concurrency = osutils.local_concurrency()
1800
 
        self.assertIsInstance(concurrency, int)
 
352
class TestTerminalEncoding(TestCase):
 
353
    """Test the auto-detection of proper terminal encoding."""
 
354
 
 
355
    def setUp(self):
 
356
        self._stdout = sys.stdout
 
357
        self._stderr = sys.stderr
 
358
        self._stdin = sys.stdin
 
359
        self._user_encoding = bzrlib.user_encoding
 
360
 
 
361
        self.addCleanup(self._reset)
 
362
 
 
363
        sys.stdout = StringIOWrapper()
 
364
        sys.stdout.encoding = 'stdout_encoding'
 
365
        sys.stderr = StringIOWrapper()
 
366
        sys.stderr.encoding = 'stderr_encoding'
 
367
        sys.stdin = StringIOWrapper()
 
368
        sys.stdin.encoding = 'stdin_encoding'
 
369
        bzrlib.user_encoding = 'user_encoding'
 
370
 
 
371
    def _reset(self):
 
372
        sys.stdout = self._stdout
 
373
        sys.stderr = self._stderr
 
374
        sys.stdin = self._stdin
 
375
        bzrlib.user_encoding = self._user_encoding
 
376
 
 
377
    def test_get_terminal_encoding(self):
 
378
        # first preference is stdout encoding
 
379
        self.assertEqual('stdout_encoding', osutils.get_terminal_encoding())
 
380
 
 
381
        sys.stdout.encoding = None
 
382
        # if sys.stdout is None, fall back to sys.stdin
 
383
        self.assertEqual('stdin_encoding', osutils.get_terminal_encoding())
 
384
 
 
385
        sys.stdin.encoding = None
 
386
        # and in the worst case, use bzrlib.user_encoding
 
387
        self.assertEqual('user_encoding', osutils.get_terminal_encoding())
 
388