/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

merge bzr.dev r3564

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
16
16
 
17
17
"""Tests for the osutils wrapper."""
18
18
 
 
19
from cStringIO import StringIO
19
20
import errno
20
21
import os
21
22
import socket
22
23
import stat
23
24
import sys
 
25
import time
24
26
 
25
27
import bzrlib
26
28
from bzrlib import (
27
29
    errors,
28
30
    osutils,
 
31
    tests,
29
32
    win32utils,
30
33
    )
31
34
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
33
36
        is_inside_any,
34
37
        is_inside_or_parent_of_any,
35
38
        pathjoin,
 
39
        pumpfile,
36
40
        )
37
41
from bzrlib.tests import (
38
42
        probe_unicode_in_user_encoding,
42
46
        TestCaseInTempDir,
43
47
        TestSkipped,
44
48
        )
 
49
from bzrlib.tests.file_utils import (
 
50
    FakeReadFile,
 
51
    )
 
52
from bzrlib.tests.test__walkdirs_win32 import WalkdirsWin32Feature
45
53
 
46
54
 
47
55
class TestOSUtils(TestCaseInTempDir):
318
326
        self.assertEqual("@", osutils.kind_marker("symlink"))
319
327
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
320
328
 
 
329
    def test_host_os_dereferences_symlinks(self):
 
330
        osutils.host_os_dereferences_symlinks()
 
331
 
 
332
 
 
333
class TestPumpFile(TestCase):
 
334
    """Test pumpfile method."""
 
335
    def setUp(self):
 
336
        # create a test datablock
 
337
        self.block_size = 512
 
338
        pattern = '0123456789ABCDEF'
 
339
        self.test_data = pattern * (3 * self.block_size / len(pattern))
 
340
        self.test_data_len = len(self.test_data)
 
341
 
 
342
    def test_bracket_block_size(self):
 
343
        """Read data in blocks with the requested read size bracketing the
 
344
        block size."""
 
345
        # make sure test data is larger than max read size
 
346
        self.assertTrue(self.test_data_len > self.block_size)
 
347
 
 
348
        from_file = FakeReadFile(self.test_data)
 
349
        to_file = StringIO()
 
350
 
 
351
        # read (max / 2) bytes and verify read size wasn't affected
 
352
        num_bytes_to_read = self.block_size / 2
 
353
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
354
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
355
        self.assertEqual(from_file.get_read_count(), 1)
 
356
 
 
357
        # read (max) bytes and verify read size wasn't affected
 
358
        num_bytes_to_read = self.block_size
 
359
        from_file.reset_read_count()
 
360
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
361
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
362
        self.assertEqual(from_file.get_read_count(), 1)
 
363
 
 
364
        # read (max + 1) bytes and verify read size was limited
 
365
        num_bytes_to_read = self.block_size + 1
 
366
        from_file.reset_read_count()
 
367
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
368
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
369
        self.assertEqual(from_file.get_read_count(), 2)
 
370
 
 
371
        # finish reading the rest of the data
 
372
        num_bytes_to_read = self.test_data_len - to_file.tell()
 
373
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
374
 
 
375
        # report error if the data wasn't equal (we only report the size due
 
376
        # to the length of the data)
 
377
        response_data = to_file.getvalue()
 
378
        if response_data != self.test_data:
 
379
            message = "Data not equal.  Expected %d bytes, received %d."
 
380
            self.fail(message % (len(response_data), self.test_data_len))
 
381
 
 
382
    def test_specified_size(self):
 
383
        """Request a transfer larger than the maximum block size and verify
 
384
        that the maximum read doesn't exceed the block_size."""
 
385
        # make sure test data is larger than max read size
 
386
        self.assertTrue(self.test_data_len > self.block_size)
 
387
 
 
388
        # retrieve data in blocks
 
389
        from_file = FakeReadFile(self.test_data)
 
390
        to_file = StringIO()
 
391
        pumpfile(from_file, to_file, self.test_data_len, self.block_size)
 
392
 
 
393
        # verify read size was equal to the maximum read size
 
394
        self.assertTrue(from_file.get_max_read_size() > 0)
 
395
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
396
        self.assertEqual(from_file.get_read_count(), 3)
 
397
 
 
398
        # report error if the data wasn't equal (we only report the size due
 
399
        # to the length of the data)
 
400
        response_data = to_file.getvalue()
 
401
        if response_data != self.test_data:
 
402
            message = "Data not equal.  Expected %d bytes, received %d."
 
403
            self.fail(message % (len(response_data), self.test_data_len))
 
404
 
 
405
    def test_to_eof(self):
 
406
        """Read to end-of-file and verify that the reads are not larger than
 
407
        the maximum read size."""
 
408
        # make sure test data is larger than max read size
 
409
        self.assertTrue(self.test_data_len > self.block_size)
 
410
 
 
411
        # retrieve data to EOF
 
412
        from_file = FakeReadFile(self.test_data)
 
413
        to_file = StringIO()
 
414
        pumpfile(from_file, to_file, -1, self.block_size)
 
415
 
 
416
        # verify read size was equal to the maximum read size
 
417
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
418
        self.assertEqual(from_file.get_read_count(), 4)
 
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_defaults(self):
 
428
        """Verifies that the default arguments will read to EOF -- this
 
429
        test verifies that any existing usages of pumpfile will not be broken
 
430
        with this new version."""
 
431
        # retrieve data using default (old) pumpfile method
 
432
        from_file = FakeReadFile(self.test_data)
 
433
        to_file = StringIO()
 
434
        pumpfile(from_file, to_file)
 
435
 
 
436
        # report error if the data wasn't equal (we only report the size due
 
437
        # to the length of the data)
 
438
        response_data = to_file.getvalue()
 
439
        if response_data != self.test_data:
 
440
            message = "Data not equal.  Expected %d bytes, received %d."
 
441
            self.fail(message % (len(response_data), self.test_data_len))
321
442
 
322
443
class TestSafeUnicode(TestCase):
323
444
 
701
822
                new_dirblock.append((info[0], info[1], info[2], info[4]))
702
823
            dirblock[:] = new_dirblock
703
824
 
 
825
    def test__walkdirs_utf8_selection(self):
 
826
        # Just trigger the function once, to make sure it has selected a real
 
827
        # implementation.
 
828
        list(osutils._walkdirs_utf8('.'))
 
829
        if WalkdirsWin32Feature.available():
 
830
            # If the compiled form is available, make sure it is used
 
831
            from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
832
            self.assertIs(_walkdirs_utf8_win32_find_file,
 
833
                          osutils._real_walkdirs_utf8)
 
834
        elif sys.platform == 'win32':
 
835
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
 
836
                          osutils._real_walkdirs_utf8)
 
837
        elif osutils._fs_enc.upper() in ('UTF-8', 'ASCII', 'ANSI_X3.4-1968'): # ascii
 
838
            self.assertIs(osutils._walkdirs_fs_utf8,
 
839
                          osutils._real_walkdirs_utf8)
 
840
        else:
 
841
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
 
842
                          osutils._real_walkdirs_utf8)
 
843
 
 
844
    def _save_platform_info(self):
 
845
        cur_winver = win32utils.winver
 
846
        cur_fs_enc = osutils._fs_enc
 
847
        cur_real_walkdirs_utf8 = osutils._real_walkdirs_utf8
 
848
        def restore():
 
849
            win32utils.winver = cur_winver
 
850
            osutils._fs_enc = cur_fs_enc
 
851
            osutils._real_walkdirs_utf8 = cur_real_walkdirs_utf8
 
852
        self.addCleanup(restore)
 
853
 
 
854
    def assertWalkdirsUtf8Is(self, expected):
 
855
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
 
856
        # Force it to redetect
 
857
        osutils._real_walkdirs_utf8 = None
 
858
        # Nothing to list, but should still trigger the selection logic
 
859
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
 
860
        self.assertIs(expected, osutils._real_walkdirs_utf8)
 
861
 
 
862
    def test_force_walkdirs_utf8_fs_utf8(self):
 
863
        self._save_platform_info()
 
864
        win32utils.winver = None # Avoid the win32 detection code
 
865
        osutils._fs_enc = 'UTF-8'
 
866
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
867
 
 
868
    def test_force_walkdirs_utf8_fs_ascii(self):
 
869
        self._save_platform_info()
 
870
        win32utils.winver = None # Avoid the win32 detection code
 
871
        osutils._fs_enc = 'US-ASCII'
 
872
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
873
 
 
874
    def test_force_walkdirs_utf8_fs_ANSI(self):
 
875
        self._save_platform_info()
 
876
        win32utils.winver = None # Avoid the win32 detection code
 
877
        osutils._fs_enc = 'ANSI_X3.4-1968'
 
878
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
879
 
 
880
    def test_force_walkdirs_utf8_fs_latin1(self):
 
881
        self._save_platform_info()
 
882
        win32utils.winver = None # Avoid the win32 detection code
 
883
        osutils._fs_enc = 'latin1'
 
884
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
 
885
 
 
886
    def test_force_walkdirs_utf8_nt(self):
 
887
        self.requireFeature(WalkdirsWin32Feature)
 
888
        self._save_platform_info()
 
889
        win32utils.winver = 'Windows NT'
 
890
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
891
        self.assertWalkdirsUtf8Is(_walkdirs_utf8_win32_find_file)
 
892
 
 
893
    def test_force_walkdirs_utf8_nt(self):
 
894
        self.requireFeature(WalkdirsWin32Feature)
 
895
        self._save_platform_info()
 
896
        win32utils.winver = 'Windows 98'
 
897
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
 
898
 
704
899
    def test_unicode_walkdirs(self):
705
900
        """Walkdirs should always return unicode paths."""
706
901
        name0 = u'0file-\xb6'
857
1052
        self._filter_out_stat(result)
858
1053
        self.assertEqual(expected_dirblocks, result)
859
1054
 
 
1055
    def test__walkdirs_utf_win32_find_file(self):
 
1056
        self.requireFeature(WalkdirsWin32Feature)
 
1057
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1058
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1059
        name0u = u'0file-\xb6'
 
1060
        name1u = u'1dir-\u062c\u0648'
 
1061
        name2u = u'2file-\u0633'
 
1062
        tree = [
 
1063
            name0u,
 
1064
            name1u + '/',
 
1065
            name1u + '/' + name0u,
 
1066
            name1u + '/' + name1u + '/',
 
1067
            name2u,
 
1068
            ]
 
1069
        self.build_tree(tree)
 
1070
        name0 = name0u.encode('utf8')
 
1071
        name1 = name1u.encode('utf8')
 
1072
        name2 = name2u.encode('utf8')
 
1073
 
 
1074
        # All of the abspaths should be in unicode, all of the relative paths
 
1075
        # should be in utf8
 
1076
        expected_dirblocks = [
 
1077
                (('', '.'),
 
1078
                 [(name0, name0, 'file', './' + name0u),
 
1079
                  (name1, name1, 'directory', './' + name1u),
 
1080
                  (name2, name2, 'file', './' + name2u),
 
1081
                 ]
 
1082
                ),
 
1083
                ((name1, './' + name1u),
 
1084
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1085
                                                        + '/' + name0u),
 
1086
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1087
                                                            + '/' + name1u),
 
1088
                 ]
 
1089
                ),
 
1090
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1091
                 [
 
1092
                 ]
 
1093
                ),
 
1094
            ]
 
1095
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1096
        self._filter_out_stat(result)
 
1097
        self.assertEqual(expected_dirblocks, result)
 
1098
 
 
1099
    def assertStatIsCorrect(self, path, win32stat):
 
1100
        os_stat = os.stat(path)
 
1101
        self.assertEqual(os_stat.st_size, win32stat.st_size)
 
1102
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
 
1103
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
 
1104
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
 
1105
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
 
1106
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
 
1107
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
 
1108
 
 
1109
    def test__walkdirs_utf_win32_find_file_stat_file(self):
 
1110
        """make sure our Stat values are valid"""
 
1111
        self.requireFeature(WalkdirsWin32Feature)
 
1112
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1113
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1114
        name0u = u'0file-\xb6'
 
1115
        name0 = name0u.encode('utf8')
 
1116
        self.build_tree([name0u])
 
1117
        # I hate to sleep() here, but I'm trying to make the ctime different
 
1118
        # from the mtime
 
1119
        time.sleep(2)
 
1120
        f = open(name0u, 'ab')
 
1121
        try:
 
1122
            f.write('just a small update')
 
1123
        finally:
 
1124
            f.close()
 
1125
 
 
1126
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1127
        entry = result[0][1][0]
 
1128
        self.assertEqual((name0, name0, 'file'), entry[:3])
 
1129
        self.assertEqual(u'./' + name0u, entry[4])
 
1130
        self.assertStatIsCorrect(entry[4], entry[3])
 
1131
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
 
1132
 
 
1133
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
 
1134
        """make sure our Stat values are valid"""
 
1135
        self.requireFeature(WalkdirsWin32Feature)
 
1136
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1137
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1138
        name0u = u'0dir-\u062c\u0648'
 
1139
        name0 = name0u.encode('utf8')
 
1140
        self.build_tree([name0u + '/'])
 
1141
 
 
1142
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1143
        entry = result[0][1][0]
 
1144
        self.assertEqual((name0, name0, 'directory'), entry[:3])
 
1145
        self.assertEqual(u'./' + name0u, entry[4])
 
1146
        self.assertStatIsCorrect(entry[4], entry[3])
 
1147
 
860
1148
    def assertPathCompare(self, path_less, path_greater):
861
1149
        """check that path_less and path_greater compare correctly."""
862
1150
        self.assertEqual(0, osutils.compare_paths_prefix_order(