← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

Finish making Tree.ids2paths support the file_ids_across_trees api.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
 
1
# Copyright (C) 2005, 2006 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
19
 
from cStringIO import StringIO
20
19
import errno
21
20
import os
22
 
import re
23
21
import socket
24
22
import stat
25
23
import sys
26
 
import time
27
24
 
 
25
import bzrlib
28
26
from bzrlib import (
29
27
    errors,
30
28
    osutils,
31
 
    tests,
32
29
    win32utils,
33
30
    )
34
31
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
 
        )
43
32
from bzrlib.tests import (
44
 
        Feature,
45
 
        probe_unicode_in_user_encoding,
46
33
        StringIOWrapper,
47
 
        SymlinkFeature,
48
 
        CaseInsCasePresFilenameFeature,
49
 
        TestCase,
50
 
        TestCaseInTempDir,
 
34
        TestCase, 
 
35
        TestCaseInTempDir, 
51
36
        TestSkipped,
52
37
        )
53
 
from bzrlib.tests.file_utils import (
54
 
    FakeReadFile,
55
 
    )
56
 
from bzrlib.tests.test__walkdirs_win32 import Win32ReadDirFeature
57
 
 
58
 
 
59
 
class _UTF8DirReaderFeature(Feature):
60
 
 
61
 
    def _probe(self):
62
 
        try:
63
 
            from bzrlib import _readdir_pyx
64
 
            self.reader = _readdir_pyx.UTF8DirReader
65
 
            return True
66
 
        except ImportError:
67
 
            return False
68
 
 
69
 
    def feature_name(self):
70
 
        return 'bzrlib._readdir_pyx'
71
 
 
72
 
UTF8DirReaderFeature = _UTF8DirReaderFeature()
73
38
 
74
39
 
75
40
class TestOSUtils(TestCaseInTempDir):
121
86
 
122
87
    # TODO: test fancy_rename using a MemoryTransport
123
88
 
124
 
    def test_rename_change_case(self):
125
 
        # on Windows we should be able to change filename case by rename
126
 
        self.build_tree(['a', 'b/'])
127
 
        osutils.rename('a', 'A')
128
 
        osutils.rename('b', 'B')
129
 
        # we can't use failUnlessExists on case-insensitive filesystem
130
 
        # so try to check shape of the tree
131
 
        shape = sorted(os.listdir('.'))
132
 
        self.assertEquals(['A', 'B'], shape)
133
 
 
134
89
    def test_01_rand_chars_empty(self):
135
90
        result = osutils.rand_chars(0)
136
91
        self.assertEqual(result, '')
150
105
        self.assertFalse(is_inside('foo.c', ''))
151
106
        self.assertTrue(is_inside('', 'foo.c'))
152
107
 
153
 
    def test_is_inside_any(self):
154
 
        SRC_FOO_C = pathjoin('src', 'foo.c')
155
 
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
156
 
                         (['src'], SRC_FOO_C),
157
 
                         (['src'], 'src'),
158
 
                         ]:
159
 
            self.assert_(is_inside_any(dirs, fn))
160
 
        for dirs, fn in [(['src'], 'srccontrol'),
161
 
                         (['src'], 'srccontrol/foo')]:
162
 
            self.assertFalse(is_inside_any(dirs, fn))
163
 
 
164
 
    def test_is_inside_or_parent_of_any(self):
165
 
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
166
 
                         (['src'], 'src/foo.c'),
167
 
                         (['src/bar.c'], 'src'),
168
 
                         (['src/bar.c', 'bla/foo.c'], 'src'),
169
 
                         (['src'], 'src'),
170
 
                         ]:
171
 
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
172
 
 
173
 
        for dirs, fn in [(['src'], 'srccontrol'),
174
 
                         (['srccontrol/foo.c'], 'src'),
175
 
                         (['src'], 'srccontrol/foo')]:
176
 
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
177
 
 
178
108
    def test_rmtree(self):
179
109
        # Check to remove tree with read-only files/dirs
180
110
        os.mkdir('dir')
200
130
        if osutils.has_symlinks():
201
131
            os.symlink('symlink', 'symlink')
202
132
            self.assertEquals('symlink', osutils.file_kind('symlink'))
203
 
 
 
133
        
204
134
        # TODO: jam 20060529 Test a block device
205
135
        try:
206
136
            os.lstat('/dev/null')
227
157
            finally:
228
158
                os.remove('socket')
229
159
 
230
 
    def test_kind_marker(self):
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'), '+')
235
 
 
236
160
    def test_get_umask(self):
237
161
        if sys.platform == 'win32':
238
162
            # umask always returns '0', no way to set it
288
212
        self.assertFormatedDelta('1 second in the future', -1)
289
213
        self.assertFormatedDelta('2 seconds in the future', -2)
290
214
 
291
 
    def test_format_date(self):
292
 
        self.assertRaises(errors.UnsupportedTimezoneFormat,
293
 
            osutils.format_date, 0, timezone='foo')
294
 
        self.assertIsInstance(osutils.format_date(0), str)
295
 
        self.assertIsInstance(osutils.format_local_date(0), unicode)
296
 
        # Testing for the actual value of the local weekday without
297
 
        # duplicating the code from format_date is difficult.
298
 
        # Instead blackbox.test_locale should check for localized
299
 
        # dates once they do occur in output strings.
300
 
 
301
215
    def test_dereference_path(self):
302
 
        self.requireFeature(SymlinkFeature)
 
216
        if not osutils.has_symlinks():
 
217
            raise TestSkipped('Symlinks are not supported on this platform')
303
218
        cwd = osutils.realpath('.')
304
219
        os.mkdir('bar')
305
220
        bar_path = osutils.pathjoin(cwd, 'bar')
308
223
        self.assertEqual(bar_path, osutils.realpath('./bar'))
309
224
        os.symlink('bar', 'foo')
310
225
        self.assertEqual(bar_path, osutils.realpath('./foo'))
311
 
 
 
226
        
312
227
        # Does not dereference terminal symlinks
313
228
        foo_path = osutils.pathjoin(cwd, 'foo')
314
229
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
325
240
        foo_baz_path = osutils.pathjoin(foo_path, 'baz')
326
241
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
327
242
 
328
 
    def test_changing_access(self):
329
 
        f = file('file', 'w')
330
 
        f.write('monkey')
331
 
        f.close()
332
 
 
333
 
        # Make a file readonly
334
 
        osutils.make_readonly('file')
335
 
        mode = os.lstat('file').st_mode
336
 
        self.assertEqual(mode, mode & 0777555)
337
 
 
338
 
        # Make a file writable
339
 
        osutils.make_writable('file')
340
 
        mode = os.lstat('file').st_mode
341
 
        self.assertEqual(mode, mode | 0200)
342
 
 
343
 
        if osutils.has_symlinks():
344
 
            # should not error when handed a symlink
345
 
            os.symlink('nonexistent', 'dangling')
346
 
            osutils.make_readonly('dangling')
347
 
            osutils.make_writable('dangling')
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
 
 
355
 
    def test_host_os_dereferences_symlinks(self):
356
 
        osutils.host_os_dereferences_symlinks()
357
 
 
358
 
 
359
 
class TestCanonicalRelPath(TestCaseInTempDir):
360
 
 
361
 
    _test_needs_features = [CaseInsCasePresFilenameFeature]
362
 
 
363
 
    def test_canonical_relpath_simple(self):
364
 
        f = file('MixedCaseName', 'w')
365
 
        f.close()
366
 
        self.failUnlessEqual(
367
 
            canonical_relpath(self.test_base_dir, 'mixedcasename'),
368
 
            'work/MixedCaseName')
369
 
 
370
 
    def test_canonical_relpath_missing_tail(self):
371
 
        os.mkdir('MixedCaseParent')
372
 
        self.failUnlessEqual(
373
 
            canonical_relpath(self.test_base_dir, 'mixedcaseparent/nochild'),
374
 
            'work/MixedCaseParent/nochild')
375
 
 
376
 
 
377
 
class TestPumpFile(TestCase):
378
 
    """Test pumpfile method."""
379
 
    def setUp(self):
380
 
        TestCase.setUp(self)
381
 
        # create a test datablock
382
 
        self.block_size = 512
383
 
        pattern = '0123456789ABCDEF'
384
 
        self.test_data = pattern * (3 * self.block_size / len(pattern))
385
 
        self.test_data_len = len(self.test_data)
386
 
 
387
 
    def test_bracket_block_size(self):
388
 
        """Read data in blocks with the requested read size bracketing the
389
 
        block size."""
390
 
        # make sure test data is larger than max read size
391
 
        self.assertTrue(self.test_data_len > self.block_size)
392
 
 
393
 
        from_file = FakeReadFile(self.test_data)
394
 
        to_file = StringIO()
395
 
 
396
 
        # read (max / 2) bytes and verify read size wasn't affected
397
 
        num_bytes_to_read = self.block_size / 2
398
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
399
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
400
 
        self.assertEqual(from_file.get_read_count(), 1)
401
 
 
402
 
        # read (max) bytes and verify read size wasn't affected
403
 
        num_bytes_to_read = self.block_size
404
 
        from_file.reset_read_count()
405
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
406
 
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
407
 
        self.assertEqual(from_file.get_read_count(), 1)
408
 
 
409
 
        # read (max + 1) bytes and verify read size was limited
410
 
        num_bytes_to_read = self.block_size + 1
411
 
        from_file.reset_read_count()
412
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
413
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
414
 
        self.assertEqual(from_file.get_read_count(), 2)
415
 
 
416
 
        # finish reading the rest of the data
417
 
        num_bytes_to_read = self.test_data_len - to_file.tell()
418
 
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
419
 
 
420
 
        # report error if the data wasn't equal (we only report the size due
421
 
        # to the length of the data)
422
 
        response_data = to_file.getvalue()
423
 
        if response_data != self.test_data:
424
 
            message = "Data not equal.  Expected %d bytes, received %d."
425
 
            self.fail(message % (len(response_data), self.test_data_len))
426
 
 
427
 
    def test_specified_size(self):
428
 
        """Request a transfer larger than the maximum block size and verify
429
 
        that the maximum read doesn't exceed the block_size."""
430
 
        # make sure test data is larger than max read size
431
 
        self.assertTrue(self.test_data_len > self.block_size)
432
 
 
433
 
        # retrieve data in blocks
434
 
        from_file = FakeReadFile(self.test_data)
435
 
        to_file = StringIO()
436
 
        pumpfile(from_file, to_file, self.test_data_len, self.block_size)
437
 
 
438
 
        # verify read size was equal to the maximum read size
439
 
        self.assertTrue(from_file.get_max_read_size() > 0)
440
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
441
 
        self.assertEqual(from_file.get_read_count(), 3)
442
 
 
443
 
        # report error if the data wasn't equal (we only report the size due
444
 
        # to the length of the data)
445
 
        response_data = to_file.getvalue()
446
 
        if response_data != self.test_data:
447
 
            message = "Data not equal.  Expected %d bytes, received %d."
448
 
            self.fail(message % (len(response_data), self.test_data_len))
449
 
 
450
 
    def test_to_eof(self):
451
 
        """Read to end-of-file and verify that the reads are not larger than
452
 
        the maximum read size."""
453
 
        # make sure test data is larger than max read size
454
 
        self.assertTrue(self.test_data_len > self.block_size)
455
 
 
456
 
        # retrieve data to EOF
457
 
        from_file = FakeReadFile(self.test_data)
458
 
        to_file = StringIO()
459
 
        pumpfile(from_file, to_file, -1, self.block_size)
460
 
 
461
 
        # verify read size was equal to the maximum read size
462
 
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
463
 
        self.assertEqual(from_file.get_read_count(), 4)
464
 
 
465
 
        # report error if the data wasn't equal (we only report the size due
466
 
        # to the length of the data)
467
 
        response_data = to_file.getvalue()
468
 
        if response_data != self.test_data:
469
 
            message = "Data not equal.  Expected %d bytes, received %d."
470
 
            self.fail(message % (len(response_data), self.test_data_len))
471
 
 
472
 
    def test_defaults(self):
473
 
        """Verifies that the default arguments will read to EOF -- this
474
 
        test verifies that any existing usages of pumpfile will not be broken
475
 
        with this new version."""
476
 
        # retrieve data using default (old) pumpfile method
477
 
        from_file = FakeReadFile(self.test_data)
478
 
        to_file = StringIO()
479
 
        pumpfile(from_file, to_file)
480
 
 
481
 
        # report error if the data wasn't equal (we only report the size due
482
 
        # to the length of the data)
483
 
        response_data = to_file.getvalue()
484
 
        if response_data != self.test_data:
485
 
            message = "Data not equal.  Expected %d bytes, received %d."
486
 
            self.fail(message % (len(response_data), self.test_data_len))
487
 
 
488
 
    def test_report_activity(self):
489
 
        activity = []
490
 
        def log_activity(length, direction):
491
 
            activity.append((length, direction))
492
 
        from_file = StringIO(self.test_data)
493
 
        to_file = StringIO()
494
 
        pumpfile(from_file, to_file, buff_size=500,
495
 
                 report_activity=log_activity, direction='read')
496
 
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
497
 
                          (36, 'read')], activity)
498
 
 
499
 
        from_file = StringIO(self.test_data)
500
 
        to_file = StringIO()
501
 
        del activity[:]
502
 
        pumpfile(from_file, to_file, buff_size=500,
503
 
                 report_activity=log_activity, direction='write')
504
 
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
505
 
                          (36, 'write')], activity)
506
 
 
507
 
        # And with a limited amount of data
508
 
        from_file = StringIO(self.test_data)
509
 
        to_file = StringIO()
510
 
        del activity[:]
511
 
        pumpfile(from_file, to_file, buff_size=500, read_length=1028,
512
 
                 report_activity=log_activity, direction='read')
513
 
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
514
 
 
515
 
 
516
 
 
517
 
class TestPumpStringFile(TestCase):
518
 
 
519
 
    def test_empty(self):
520
 
        output = StringIO()
521
 
        pump_string_file("", output)
522
 
        self.assertEqual("", output.getvalue())
523
 
 
524
 
    def test_more_than_segment_size(self):
525
 
        output = StringIO()
526
 
        pump_string_file("123456789", output, 2)
527
 
        self.assertEqual("123456789", output.getvalue())
528
 
 
529
 
    def test_segment_size(self):
530
 
        output = StringIO()
531
 
        pump_string_file("12", output, 2)
532
 
        self.assertEqual("12", output.getvalue())
533
 
 
534
 
    def test_segment_size_multiple(self):
535
 
        output = StringIO()
536
 
        pump_string_file("1234", output, 2)
537
 
        self.assertEqual("1234", output.getvalue())
538
 
 
539
243
 
540
244
class TestSafeUnicode(TestCase):
541
245
 
580
284
class TestSafeRevisionId(TestCase):
581
285
 
582
286
    def test_from_ascii_string(self):
583
 
        # this shouldn't give a warning because it's getting an ascii string
584
 
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
 
287
        f = 'foobar'
 
288
        self.assertEqual('foobar', osutils.safe_revision_id(f))
 
289
        self.assertIs(osutils.safe_utf8(f), f)
585
290
 
586
291
    def test_from_unicode_string_ascii_contents(self):
587
 
        self.assertEqual('bargam',
588
 
                         osutils.safe_revision_id(u'bargam', warn=False))
589
 
 
590
 
    def test_from_unicode_deprecated(self):
591
 
        self.assertEqual('bargam',
592
 
            self.callDeprecated([osutils._revision_id_warning],
593
 
                                osutils.safe_revision_id, u'bargam'))
 
292
        self.assertEqual('bargam', osutils.safe_revision_id(u'bargam'))
594
293
 
595
294
    def test_from_unicode_string_unicode_contents(self):
596
295
        self.assertEqual('bargam\xc2\xae',
597
 
                         osutils.safe_revision_id(u'bargam\xae', warn=False))
 
296
                         osutils.safe_revision_id(u'bargam\xae'))
598
297
 
599
298
    def test_from_utf8_string(self):
600
299
        self.assertEqual('foo\xc2\xae',
601
300
                         osutils.safe_revision_id('foo\xc2\xae'))
602
301
 
 
302
    def test_bad_utf8_string(self):
 
303
        # This check may eventually go away
 
304
        self.assertRaises(BzrBadParameterNotUnicode,
 
305
                          osutils.safe_utf8, '\xbb\xbb')
 
306
 
603
307
    def test_none(self):
604
308
        """Currently, None is a valid revision_id"""
605
309
        self.assertEqual(None, osutils.safe_revision_id(None))
606
310
 
607
311
 
608
 
class TestSafeFileId(TestCase):
609
 
 
610
 
    def test_from_ascii_string(self):
611
 
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
612
 
 
613
 
    def test_from_unicode_string_ascii_contents(self):
614
 
        self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
615
 
 
616
 
    def test_from_unicode_deprecated(self):
617
 
        self.assertEqual('bargam',
618
 
            self.callDeprecated([osutils._file_id_warning],
619
 
                                osutils.safe_file_id, u'bargam'))
620
 
 
621
 
    def test_from_unicode_string_unicode_contents(self):
622
 
        self.assertEqual('bargam\xc2\xae',
623
 
                         osutils.safe_file_id(u'bargam\xae', warn=False))
624
 
 
625
 
    def test_from_utf8_string(self):
626
 
        self.assertEqual('foo\xc2\xae',
627
 
                         osutils.safe_file_id('foo\xc2\xae'))
628
 
 
629
 
    def test_none(self):
630
 
        """Currently, None is a valid revision_id"""
631
 
        self.assertEqual(None, osutils.safe_file_id(None))
632
 
 
633
 
 
634
312
class TestWin32Funcs(TestCase):
635
313
    """Test that the _win32 versions of os utilities return appropriate paths."""
636
314
 
691
369
 
692
370
class TestWin32FuncsDirs(TestCaseInTempDir):
693
371
    """Test win32 functions that create files."""
694
 
 
 
372
    
695
373
    def test_getcwd(self):
696
374
        if win32utils.winver == 'Windows 98':
697
375
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
708
386
        #       osutils.getcwd() renormalize the path.
709
387
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
710
388
 
711
 
    def test_minimum_path_selection(self):
712
 
        self.assertEqual(set(),
713
 
            osutils.minimum_path_selection([]))
714
 
        self.assertEqual(set(['a', 'b']),
715
 
            osutils.minimum_path_selection(['a', 'b']))
716
 
        self.assertEqual(set(['a/', 'b']),
717
 
            osutils.minimum_path_selection(['a/', 'b']))
718
 
        self.assertEqual(set(['a/', 'b']),
719
 
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
720
 
 
721
389
    def test_mkdtemp(self):
722
390
        tmpdir = osutils._win32_mkdtemp(dir='.')
723
391
        self.assertFalse('\\' in tmpdir)
803
471
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
804
472
 
805
473
 
806
 
class TestChunksToLines(TestCase):
807
 
 
808
 
    def test_smoketest(self):
809
 
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
810
 
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
811
 
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
812
 
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
813
 
 
814
 
    def test_osutils_binding(self):
815
 
        from bzrlib.tests import test__chunks_to_lines
816
 
        if test__chunks_to_lines.CompiledChunksToLinesFeature.available():
817
 
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
818
 
        else:
819
 
            from bzrlib._chunks_to_lines_py import chunks_to_lines
820
 
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
821
 
 
822
 
 
823
474
class TestSplitLines(TestCase):
824
475
 
825
476
    def test_split_unicode(self):
881
532
        self.assertEqual(expected_dirblocks[1:],
882
533
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
883
534
 
884
 
    def test_walkdirs_os_error(self):
885
 
        # <https://bugs.edge.launchpad.net/bzr/+bug/338653>
886
 
        # Pyrex readdir didn't raise useful messages if it had an error
887
 
        # reading the directory
888
 
        if sys.platform == 'win32':
889
 
            raise tests.TestNotApplicable(
890
 
                "readdir IOError not tested on win32")
891
 
        os.mkdir("test-unreadable")
892
 
        os.chmod("test-unreadable", 0000)
893
 
        # must chmod it back so that it can be removed
894
 
        self.addCleanup(lambda: os.chmod("test-unreadable", 0700))
895
 
        # The error is not raised until the generator is actually evaluated.
896
 
        # (It would be ok if it happened earlier but at the moment it
897
 
        # doesn't.)
898
 
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
899
 
        self.assertEquals('./test-unreadable', e.filename)
900
 
        self.assertEquals(errno.EACCES, e.errno)
901
 
        # Ensure the message contains the file name
902
 
        self.assertContainsRe(str(e), "\./test-unreadable")
903
 
 
904
 
    def test__walkdirs_utf8(self):
905
 
        tree = [
906
 
            '.bzr',
907
 
            '0file',
908
 
            '1dir/',
909
 
            '1dir/0file',
910
 
            '1dir/1dir/',
911
 
            '2file'
912
 
            ]
913
 
        self.build_tree(tree)
914
 
        expected_dirblocks = [
915
 
                (('', '.'),
916
 
                 [('0file', '0file', 'file'),
917
 
                  ('1dir', '1dir', 'directory'),
918
 
                  ('2file', '2file', 'file'),
919
 
                 ]
920
 
                ),
921
 
                (('1dir', './1dir'),
922
 
                 [('1dir/0file', '0file', 'file'),
923
 
                  ('1dir/1dir', '1dir', 'directory'),
924
 
                 ]
925
 
                ),
926
 
                (('1dir/1dir', './1dir/1dir'),
927
 
                 [
928
 
                 ]
929
 
                ),
930
 
            ]
931
 
        result = []
932
 
        found_bzrdir = False
933
 
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
934
 
            if len(dirblock) and dirblock[0][1] == '.bzr':
935
 
                # this tests the filtering of selected paths
936
 
                found_bzrdir = True
937
 
                del dirblock[0]
938
 
            result.append((dirdetail, dirblock))
939
 
 
940
 
        self.assertTrue(found_bzrdir)
941
 
        self.assertEqual(expected_dirblocks,
942
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
943
 
        # you can search a subdir only, with a supplied prefix.
944
 
        result = []
945
 
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
946
 
            result.append(dirblock)
947
 
        self.assertEqual(expected_dirblocks[1:],
948
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
949
 
 
950
 
    def _filter_out_stat(self, result):
951
 
        """Filter out the stat value from the walkdirs result"""
952
 
        for dirdetail, dirblock in result:
953
 
            new_dirblock = []
954
 
            for info in dirblock:
955
 
                # Ignore info[3] which is the stat
956
 
                new_dirblock.append((info[0], info[1], info[2], info[4]))
957
 
            dirblock[:] = new_dirblock
958
 
 
959
 
    def _save_platform_info(self):
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)
968
 
 
969
 
    def assertReadFSDirIs(self, expected):
970
 
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
971
 
        # Force it to redetect
972
 
        osutils._selected_dir_reader = None
973
 
        # Nothing to list, but should still trigger the selection logic
974
 
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
975
 
        self.assertIsInstance(osutils._selected_dir_reader, expected)
976
 
 
977
 
    def test_force_walkdirs_utf8_fs_utf8(self):
978
 
        self.requireFeature(UTF8DirReaderFeature)
979
 
        self._save_platform_info()
980
 
        win32utils.winver = None # Avoid the win32 detection code
981
 
        osutils._fs_enc = 'UTF-8'
982
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
983
 
 
984
 
    def test_force_walkdirs_utf8_fs_ascii(self):
985
 
        self.requireFeature(UTF8DirReaderFeature)
986
 
        self._save_platform_info()
987
 
        win32utils.winver = None # Avoid the win32 detection code
988
 
        osutils._fs_enc = 'US-ASCII'
989
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
990
 
 
991
 
    def test_force_walkdirs_utf8_fs_ANSI(self):
992
 
        self.requireFeature(UTF8DirReaderFeature)
993
 
        self._save_platform_info()
994
 
        win32utils.winver = None # Avoid the win32 detection code
995
 
        osutils._fs_enc = 'ANSI_X3.4-1968'
996
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
997
 
 
998
 
    def test_force_walkdirs_utf8_fs_latin1(self):
999
 
        self._save_platform_info()
1000
 
        win32utils.winver = None # Avoid the win32 detection code
1001
 
        osutils._fs_enc = 'latin1'
1002
 
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
1003
 
 
1004
 
    def test_force_walkdirs_utf8_nt(self):
1005
 
        # Disabled because the thunk of the whole walkdirs api is disabled.
1006
 
        self.requireFeature(Win32ReadDirFeature)
1007
 
        self._save_platform_info()
1008
 
        win32utils.winver = 'Windows NT'
1009
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1010
 
        self.assertReadFSDirIs(Win32ReadDir)
1011
 
 
1012
 
    def test_force_walkdirs_utf8_98(self):
1013
 
        self.requireFeature(Win32ReadDirFeature)
1014
 
        self._save_platform_info()
1015
 
        win32utils.winver = 'Windows 98'
1016
 
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
1017
 
 
1018
 
    def test_unicode_walkdirs(self):
1019
 
        """Walkdirs should always return unicode paths."""
1020
 
        name0 = u'0file-\xb6'
1021
 
        name1 = u'1dir-\u062c\u0648'
1022
 
        name2 = u'2file-\u0633'
1023
 
        tree = [
1024
 
            name0,
1025
 
            name1 + '/',
1026
 
            name1 + '/' + name0,
1027
 
            name1 + '/' + name1 + '/',
1028
 
            name2,
1029
 
            ]
1030
 
        try:
1031
 
            self.build_tree(tree)
1032
 
        except UnicodeError:
1033
 
            raise TestSkipped('Could not represent Unicode chars'
1034
 
                              ' in current encoding.')
1035
 
        expected_dirblocks = [
1036
 
                ((u'', u'.'),
1037
 
                 [(name0, name0, 'file', './' + name0),
1038
 
                  (name1, name1, 'directory', './' + name1),
1039
 
                  (name2, name2, 'file', './' + name2),
1040
 
                 ]
1041
 
                ),
1042
 
                ((name1, './' + name1),
1043
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1
1044
 
                                                        + '/' + name0),
1045
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1
1046
 
                                                            + '/' + name1),
1047
 
                 ]
1048
 
                ),
1049
 
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
1050
 
                 [
1051
 
                 ]
1052
 
                ),
1053
 
            ]
1054
 
        result = list(osutils.walkdirs('.'))
1055
 
        self._filter_out_stat(result)
1056
 
        self.assertEqual(expected_dirblocks, result)
1057
 
        result = list(osutils.walkdirs(u'./'+name1, name1))
1058
 
        self._filter_out_stat(result)
1059
 
        self.assertEqual(expected_dirblocks[1:], result)
1060
 
 
1061
 
    def test_unicode__walkdirs_utf8(self):
1062
 
        """Walkdirs_utf8 should always return utf8 paths.
1063
 
 
1064
 
        The abspath portion might be in unicode or utf-8
1065
 
        """
1066
 
        name0 = u'0file-\xb6'
1067
 
        name1 = u'1dir-\u062c\u0648'
1068
 
        name2 = u'2file-\u0633'
1069
 
        tree = [
1070
 
            name0,
1071
 
            name1 + '/',
1072
 
            name1 + '/' + name0,
1073
 
            name1 + '/' + name1 + '/',
1074
 
            name2,
1075
 
            ]
1076
 
        try:
1077
 
            self.build_tree(tree)
1078
 
        except UnicodeError:
1079
 
            raise TestSkipped('Could not represent Unicode chars'
1080
 
                              ' in current encoding.')
1081
 
        name0 = name0.encode('utf8')
1082
 
        name1 = name1.encode('utf8')
1083
 
        name2 = name2.encode('utf8')
1084
 
 
1085
 
        expected_dirblocks = [
1086
 
                (('', '.'),
1087
 
                 [(name0, name0, 'file', './' + name0),
1088
 
                  (name1, name1, 'directory', './' + name1),
1089
 
                  (name2, name2, 'file', './' + name2),
1090
 
                 ]
1091
 
                ),
1092
 
                ((name1, './' + name1),
1093
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1
1094
 
                                                        + '/' + name0),
1095
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1
1096
 
                                                            + '/' + name1),
1097
 
                 ]
1098
 
                ),
1099
 
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
1100
 
                 [
1101
 
                 ]
1102
 
                ),
1103
 
            ]
1104
 
        result = []
1105
 
        # For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1106
 
        # all abspaths are Unicode, and encode them back into utf8.
1107
 
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1108
 
            self.assertIsInstance(dirdetail[0], str)
1109
 
            if isinstance(dirdetail[1], unicode):
1110
 
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1111
 
                dirblock = [list(info) for info in dirblock]
1112
 
                for info in dirblock:
1113
 
                    self.assertIsInstance(info[4], unicode)
1114
 
                    info[4] = info[4].encode('utf8')
1115
 
            new_dirblock = []
1116
 
            for info in dirblock:
1117
 
                self.assertIsInstance(info[0], str)
1118
 
                self.assertIsInstance(info[1], str)
1119
 
                self.assertIsInstance(info[4], str)
1120
 
                # Remove the stat information
1121
 
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1122
 
            result.append((dirdetail, new_dirblock))
1123
 
        self.assertEqual(expected_dirblocks, result)
1124
 
 
1125
 
    def test__walkdirs_utf8_with_unicode_fs(self):
1126
 
        """UnicodeDirReader should be a safe fallback everywhere
1127
 
 
1128
 
        The abspath portion should be in unicode
1129
 
        """
1130
 
        # Use the unicode reader. TODO: split into driver-and-driven unit
1131
 
        # tests.
1132
 
        self._save_platform_info()
1133
 
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
1134
 
        name0u = u'0file-\xb6'
1135
 
        name1u = u'1dir-\u062c\u0648'
1136
 
        name2u = u'2file-\u0633'
1137
 
        tree = [
1138
 
            name0u,
1139
 
            name1u + '/',
1140
 
            name1u + '/' + name0u,
1141
 
            name1u + '/' + name1u + '/',
1142
 
            name2u,
1143
 
            ]
1144
 
        try:
1145
 
            self.build_tree(tree)
1146
 
        except UnicodeError:
1147
 
            raise TestSkipped('Could not represent Unicode chars'
1148
 
                              ' in current encoding.')
1149
 
        name0 = name0u.encode('utf8')
1150
 
        name1 = name1u.encode('utf8')
1151
 
        name2 = name2u.encode('utf8')
1152
 
 
1153
 
        # All of the abspaths should be in unicode, all of the relative paths
1154
 
        # should be in utf8
1155
 
        expected_dirblocks = [
1156
 
                (('', '.'),
1157
 
                 [(name0, name0, 'file', './' + name0u),
1158
 
                  (name1, name1, 'directory', './' + name1u),
1159
 
                  (name2, name2, 'file', './' + name2u),
1160
 
                 ]
1161
 
                ),
1162
 
                ((name1, './' + name1u),
1163
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1164
 
                                                        + '/' + name0u),
1165
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1166
 
                                                            + '/' + name1u),
1167
 
                 ]
1168
 
                ),
1169
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1170
 
                 [
1171
 
                 ]
1172
 
                ),
1173
 
            ]
1174
 
        result = list(osutils._walkdirs_utf8('.'))
1175
 
        self._filter_out_stat(result)
1176
 
        self.assertEqual(expected_dirblocks, result)
1177
 
 
1178
 
    def test__walkdirs_utf8_win32readdir(self):
1179
 
        self.requireFeature(Win32ReadDirFeature)
1180
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1181
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1182
 
        self._save_platform_info()
1183
 
        osutils._selected_dir_reader = Win32ReadDir()
1184
 
        name0u = u'0file-\xb6'
1185
 
        name1u = u'1dir-\u062c\u0648'
1186
 
        name2u = u'2file-\u0633'
1187
 
        tree = [
1188
 
            name0u,
1189
 
            name1u + '/',
1190
 
            name1u + '/' + name0u,
1191
 
            name1u + '/' + name1u + '/',
1192
 
            name2u,
1193
 
            ]
1194
 
        self.build_tree(tree)
1195
 
        name0 = name0u.encode('utf8')
1196
 
        name1 = name1u.encode('utf8')
1197
 
        name2 = name2u.encode('utf8')
1198
 
 
1199
 
        # All of the abspaths should be in unicode, all of the relative paths
1200
 
        # should be in utf8
1201
 
        expected_dirblocks = [
1202
 
                (('', '.'),
1203
 
                 [(name0, name0, 'file', './' + name0u),
1204
 
                  (name1, name1, 'directory', './' + name1u),
1205
 
                  (name2, name2, 'file', './' + name2u),
1206
 
                 ]
1207
 
                ),
1208
 
                ((name1, './' + name1u),
1209
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1210
 
                                                        + '/' + name0u),
1211
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1212
 
                                                            + '/' + name1u),
1213
 
                 ]
1214
 
                ),
1215
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1216
 
                 [
1217
 
                 ]
1218
 
                ),
1219
 
            ]
1220
 
        result = list(osutils._walkdirs_utf8(u'.'))
1221
 
        self._filter_out_stat(result)
1222
 
        self.assertEqual(expected_dirblocks, result)
1223
 
 
1224
 
    def assertStatIsCorrect(self, path, win32stat):
1225
 
        os_stat = os.stat(path)
1226
 
        self.assertEqual(os_stat.st_size, win32stat.st_size)
1227
 
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1228
 
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1229
 
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1230
 
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1231
 
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1232
 
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1233
 
 
1234
 
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1235
 
        """make sure our Stat values are valid"""
1236
 
        self.requireFeature(Win32ReadDirFeature)
1237
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1238
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1239
 
        name0u = u'0file-\xb6'
1240
 
        name0 = name0u.encode('utf8')
1241
 
        self.build_tree([name0u])
1242
 
        # I hate to sleep() here, but I'm trying to make the ctime different
1243
 
        # from the mtime
1244
 
        time.sleep(2)
1245
 
        f = open(name0u, 'ab')
1246
 
        try:
1247
 
            f.write('just a small update')
1248
 
        finally:
1249
 
            f.close()
1250
 
 
1251
 
        result = Win32ReadDir().read_dir('', u'.')
1252
 
        entry = result[0]
1253
 
        self.assertEqual((name0, name0, 'file'), entry[:3])
1254
 
        self.assertEqual(u'./' + name0u, entry[4])
1255
 
        self.assertStatIsCorrect(entry[4], entry[3])
1256
 
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1257
 
 
1258
 
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1259
 
        """make sure our Stat values are valid"""
1260
 
        self.requireFeature(Win32ReadDirFeature)
1261
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1262
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1263
 
        name0u = u'0dir-\u062c\u0648'
1264
 
        name0 = name0u.encode('utf8')
1265
 
        self.build_tree([name0u + '/'])
1266
 
 
1267
 
        result = Win32ReadDir().read_dir('', u'.')
1268
 
        entry = result[0]
1269
 
        self.assertEqual((name0, name0, 'directory'), entry[:3])
1270
 
        self.assertEqual(u'./' + name0u, entry[4])
1271
 
        self.assertStatIsCorrect(entry[4], entry[3])
1272
 
 
1273
535
    def assertPathCompare(self, path_less, path_greater):
1274
536
        """check that path_less and path_greater compare correctly."""
1275
537
        self.assertEqual(0, osutils.compare_paths_prefix_order(
1349
611
 
1350
612
 
1351
613
class TestCopyTree(TestCaseInTempDir):
1352
 
 
 
614
    
1353
615
    def test_copy_basic_tree(self):
1354
616
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1355
617
        osutils.copy_tree('source', 'target')
1364
626
        self.assertEqual(['c'], os.listdir('target/b'))
1365
627
 
1366
628
    def test_copy_tree_symlinks(self):
1367
 
        self.requireFeature(SymlinkFeature)
 
629
        if not osutils.has_symlinks():
 
630
            return
1368
631
        self.build_tree(['source/'])
1369
632
        os.symlink('a/generic/path', 'source/lnk')
1370
633
        osutils.copy_tree('source', 'target')
1434
697
 
1435
698
    def test_unicode(self):
1436
699
        """Environment can only contain plain strings
1437
 
 
 
700
        
1438
701
        So Unicode strings must be encoded.
1439
702
        """
1440
 
        uni_val, env_val = probe_unicode_in_user_encoding()
1441
 
        if uni_val is None:
 
703
        # Try a few different characters, to see if we can get
 
704
        # one that will be valid in the user_encoding
 
705
        possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
 
706
        for uni_val in possible_vals:
 
707
            try:
 
708
                env_val = uni_val.encode(bzrlib.user_encoding)
 
709
            except UnicodeEncodeError:
 
710
                # Try a different character
 
711
                pass
 
712
            else:
 
713
                break
 
714
        else:
1442
715
            raise TestSkipped('Cannot find a unicode character that works in'
1443
 
                              ' encoding %s' % (osutils.get_user_encoding(),))
 
716
                              ' encoding %s' % (bzrlib.user_encoding,))
1444
717
 
1445
718
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1446
719
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1474
747
        self.assertTrue(isinstance(offset, int))
1475
748
        eighteen_hours = 18 * 3600
1476
749
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1477
 
 
1478
 
 
1479
 
class TestSizeShaFile(TestCaseInTempDir):
1480
 
 
1481
 
    def test_sha_empty(self):
1482
 
        self.build_tree_contents([('foo', '')])
1483
 
        expected_sha = osutils.sha_string('')
1484
 
        f = open('foo')
1485
 
        self.addCleanup(f.close)
1486
 
        size, sha = osutils.size_sha_file(f)
1487
 
        self.assertEqual(0, size)
1488
 
        self.assertEqual(expected_sha, sha)
1489
 
 
1490
 
    def test_sha_mixed_endings(self):
1491
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1492
 
        self.build_tree_contents([('foo', text)])
1493
 
        expected_sha = osutils.sha_string(text)
1494
 
        f = open('foo')
1495
 
        self.addCleanup(f.close)
1496
 
        size, sha = osutils.size_sha_file(f)
1497
 
        self.assertEqual(38, size)
1498
 
        self.assertEqual(expected_sha, sha)
1499
 
 
1500
 
 
1501
 
class TestShaFileByName(TestCaseInTempDir):
1502
 
 
1503
 
    def test_sha_empty(self):
1504
 
        self.build_tree_contents([('foo', '')])
1505
 
        expected_sha = osutils.sha_string('')
1506
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1507
 
 
1508
 
    def test_sha_mixed_endings(self):
1509
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1510
 
        self.build_tree_contents([('foo', text)])
1511
 
        expected_sha = osutils.sha_string(text)
1512
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1513
 
 
1514
 
 
1515
 
class TestResourceLoading(TestCaseInTempDir):
1516
 
 
1517
 
    def test_resource_string(self):
1518
 
        # test resource in bzrlib
1519
 
        text = osutils.resource_string('bzrlib', 'debug.py')
1520
 
        self.assertContainsRe(text, "debug_flags = set()")
1521
 
        # test resource under bzrlib
1522
 
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1523
 
        self.assertContainsRe(text, "class TextUIFactory")
1524
 
        # test unsupported package
1525
 
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1526
 
            'yyy.xx')
1527
 
        # test unknown resource
1528
 
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1529
 
 
1530
 
 
1531
 
class TestReCompile(TestCase):
1532
 
 
1533
 
    def test_re_compile_checked(self):
1534
 
        r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1535
 
        self.assertTrue(r.match('aaaa'))
1536
 
        self.assertTrue(r.match('aAaA'))
1537
 
 
1538
 
    def test_re_compile_checked_error(self):
1539
 
        # like https://bugs.launchpad.net/bzr/+bug/251352
1540
 
        err = self.assertRaises(
1541
 
            errors.BzrCommandError,
1542
 
            osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1543
 
        self.assertEqual(
1544
 
            "Invalid regular expression in test case: '*': "
1545
 
            "nothing to repeat",
1546
 
            str(err))