/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/per_transport.py

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

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
1
 
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Tests for Transport implementations.
18
18
 
48
48
from bzrlib.smart import medium
49
49
from bzrlib.tests import (
50
50
    TestCaseInTempDir,
51
 
    TestScenarioApplier,
52
51
    TestSkipped,
53
52
    TestNotApplicable,
 
53
    multiply_tests,
54
54
    )
 
55
from bzrlib.tests import test_server
55
56
from bzrlib.tests.test_transport import TestTransportImplementation
56
57
from bzrlib.transport import (
57
58
    ConnectedTransport,
61
62
from bzrlib.transport.memory import MemoryTransport
62
63
 
63
64
 
64
 
class TransportTestProviderAdapter(TestScenarioApplier):
65
 
    """A tool to generate a suite testing all transports for a single test.
66
 
 
67
 
    This is done by copying the test once for each transport and injecting
68
 
    the transport_class and transport_server classes into each copy. Each copy
69
 
    is also given a new id() to make it easy to identify.
70
 
    """
71
 
 
72
 
    def __init__(self):
73
 
        self.scenarios = self._test_permutations()
74
 
 
75
 
    def get_transport_test_permutations(self, module):
76
 
        """Get the permutations module wants to have tested."""
77
 
        if getattr(module, 'get_test_permutations', None) is None:
78
 
            raise AssertionError(
79
 
                "transport module %s doesn't provide get_test_permutations()"
80
 
                % module.__name__)
81
 
            return []
82
 
        return module.get_test_permutations()
83
 
 
84
 
    def _test_permutations(self):
85
 
        """Return a list of the klass, server_factory pairs to test."""
86
 
        result = []
87
 
        for module in _get_transport_modules():
88
 
            try:
89
 
                permutations = self.get_transport_test_permutations(
90
 
                    reduce(getattr, (module).split('.')[1:], __import__(module)))
91
 
                for (klass, server_factory) in permutations:
92
 
                    scenario = (server_factory.__name__,
93
 
                        {"transport_class":klass,
94
 
                         "transport_server":server_factory})
95
 
                    result.append(scenario)
96
 
            except errors.DependencyNotPresent, e:
97
 
                # Continue even if a dependency prevents us 
98
 
                # from adding this test
99
 
                pass
100
 
        return result
 
65
def get_transport_test_permutations(module):
 
66
    """Get the permutations module wants to have tested."""
 
67
    if getattr(module, 'get_test_permutations', None) is None:
 
68
        raise AssertionError(
 
69
            "transport module %s doesn't provide get_test_permutations()"
 
70
            % module.__name__)
 
71
        return []
 
72
    return module.get_test_permutations()
 
73
 
 
74
 
 
75
def transport_test_permutations():
 
76
    """Return a list of the klass, server_factory pairs to test."""
 
77
    result = []
 
78
    for module in _get_transport_modules():
 
79
        try:
 
80
            permutations = get_transport_test_permutations(
 
81
                reduce(getattr, (module).split('.')[1:], __import__(module)))
 
82
            for (klass, server_factory) in permutations:
 
83
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
 
84
                    {"transport_class":klass,
 
85
                     "transport_server":server_factory})
 
86
                result.append(scenario)
 
87
        except errors.DependencyNotPresent, e:
 
88
            # Continue even if a dependency prevents us
 
89
            # from adding this test
 
90
            pass
 
91
    return result
101
92
 
102
93
 
103
94
def load_tests(standard_tests, module, loader):
104
95
    """Multiply tests for tranport implementations."""
105
96
    result = loader.suiteClass()
106
 
    adapter = TransportTestProviderAdapter()
107
 
    for test in tests.iter_suite_tests(standard_tests):
108
 
        result.addTests(adapter.adapt(test))
109
 
    return result
 
97
    scenarios = transport_test_permutations()
 
98
    return multiply_tests(standard_tests, scenarios, result)
110
99
 
111
100
 
112
101
class TransportTests(TestTransportImplementation):
167
156
        self.assertEqual(True, t.has('a'))
168
157
        self.assertEqual(False, t.has('c'))
169
158
        self.assertEqual(True, t.has(urlutils.escape('%')))
170
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
171
 
                [True, True, False, False, True, False, True, False])
 
159
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd',
 
160
                                           'e', 'f', 'g', 'h'])),
 
161
                         [True, True, False, False,
 
162
                          True, False, True, False])
172
163
        self.assertEqual(True, t.has_any(['a', 'b', 'c']))
173
 
        self.assertEqual(False, t.has_any(['c', 'd', 'f', urlutils.escape('%%')]))
174
 
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
175
 
                [True, True, False, False, True, False, True, False])
 
164
        self.assertEqual(False, t.has_any(['c', 'd', 'f',
 
165
                                           urlutils.escape('%%')]))
 
166
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd',
 
167
                                                'e', 'f', 'g', 'h']))),
 
168
                         [True, True, False, False,
 
169
                          True, False, True, False])
176
170
        self.assertEqual(False, t.has_any(['c', 'c', 'c']))
177
171
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
178
172
 
179
173
    def test_has_root_works(self):
180
 
        from bzrlib.smart import server
181
 
        if self.transport_server is server.SmartTCPServer_for_testing:
 
174
        if self.transport_server is test_server.SmartTCPServer_for_testing:
182
175
            raise TestNotApplicable(
183
176
                "SmartTCPServer_for_testing intentionally does not allow "
184
177
                "access to /.")
210
203
        for content, f in itertools.izip(contents, content_f):
211
204
            self.assertEqual(content, f.read())
212
205
 
 
206
    def test_get_unknown_file(self):
 
207
        t = self.get_transport()
 
208
        files = ['a', 'b']
 
209
        contents = ['contents of a\n',
 
210
                    'contents of b\n',
 
211
                    ]
 
212
        self.build_tree(files, transport=t, line_endings='binary')
213
213
        self.assertRaises(NoSuchFile, t.get, 'c')
214
214
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
215
215
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
249
249
        for content, fname in zip(contents, files):
250
250
            self.assertEqual(content, t.get_bytes(fname))
251
251
 
 
252
    def test_get_bytes_unknown_file(self):
 
253
        t = self.get_transport()
 
254
 
252
255
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
253
256
 
254
257
    def test_get_with_open_write_stream_sees_all_content(self):
325
328
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
326
329
                               create_parent_dir=True)
327
330
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
328
 
        
 
331
 
329
332
        # But we still get NoSuchFile if we can't make the parent dir
330
333
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
331
334
                                       'contents\n',
353
356
        umask = osutils.get_umask()
354
357
        t.put_bytes('nomode', 'test text\n', mode=None)
355
358
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
356
 
        
 
359
 
357
360
    def test_put_bytes_non_atomic_permissions(self):
358
361
        t = self.get_transport()
359
362
 
387
390
        t.put_bytes_non_atomic('dir777/mode664', 'test text\n', mode=0664,
388
391
                               dir_mode=0777, create_parent_dir=True)
389
392
        self.assertTransportMode(t, 'dir777', 0777)
390
 
        
 
393
 
391
394
    def test_put_file(self):
392
395
        t = self.get_transport()
393
396
 
441
444
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
442
445
                              create_parent_dir=True)
443
446
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
444
 
        
 
447
 
445
448
        # But we still get NoSuchFile if we can't make the parent dir
446
449
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
447
450
                                       StringIO('contents\n'),
469
472
        umask = osutils.get_umask()
470
473
        t.put_file('nomode', StringIO('test text\n'), mode=None)
471
474
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
472
 
        
 
475
 
473
476
    def test_put_file_non_atomic_permissions(self):
474
477
        t = self.get_transport()
475
478
 
492
495
        umask = osutils.get_umask()
493
496
        t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
494
497
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
495
 
        
 
498
 
496
499
        # We should also be able to set the mode for a parent directory
497
500
        # when it is created
498
501
        sio = StringIO()
538
541
        t = self.get_transport()
539
542
 
540
543
        if t.is_readonly():
541
 
            # cannot mkdir on readonly transports. We're not testing for 
 
544
            # cannot mkdir on readonly transports. We're not testing for
542
545
            # cache coherency because cache behaviour is not currently
543
546
            # defined for the transport interface.
544
547
            self.assertRaises(TransportNotPossible, t.mkdir, '.')
565
568
 
566
569
        # we were testing that a local mkdir followed by a transport
567
570
        # mkdir failed thusly, but given that we * in one process * do not
568
 
        # concurrently fiddle with disk dirs and then use transport to do 
 
571
        # concurrently fiddle with disk dirs and then use transport to do
569
572
        # things, the win here seems marginal compared to the constraint on
570
573
        # the interface. RBC 20051227
571
574
        t.mkdir('dir_g')
681
684
            for f in files:
682
685
                self.assertTransportMode(temp_transport, f, mode)
683
686
 
 
687
    def test_create_prefix(self):
 
688
        t = self.get_transport()
 
689
        sub = t.clone('foo').clone('bar')
 
690
        try:
 
691
            sub.create_prefix()
 
692
        except TransportNotPossible:
 
693
            self.assertTrue(t.is_readonly())
 
694
        else:
 
695
            self.assertTrue(t.has('foo/bar'))
 
696
 
684
697
    def test_append_file(self):
685
698
        t = self.get_transport()
686
699
 
790
803
                t.append_file, 'f', StringIO('f'), mode=None)
791
804
            return
792
805
        t.append_file('f', StringIO('f'), mode=None)
793
 
        
 
806
 
794
807
    def test_append_bytes_mode(self):
795
808
        # check append_bytes accepts a mode
796
809
        t = self.get_transport()
799
812
                t.append_bytes, 'f', 'f', mode=None)
800
813
            return
801
814
        t.append_bytes('f', 'f', mode=None)
802
 
        
 
815
 
803
816
    def test_delete(self):
804
817
        # TODO: Test Transport.delete
805
818
        t = self.get_transport()
866
879
 
867
880
    def test_rmdir_not_empty(self):
868
881
        """Deleting a non-empty directory raises an exception
869
 
        
 
882
 
870
883
        sftp (and possibly others) don't give us a specific "directory not
871
884
        empty" exception -- we can just see that the operation failed.
872
885
        """
879
892
 
880
893
    def test_rmdir_empty_but_similar_prefix(self):
881
894
        """rmdir does not get confused by sibling paths.
882
 
        
 
895
 
883
896
        A naive implementation of MemoryTransport would refuse to rmdir
884
897
        ".bzr/branch" if there is a ".bzr/branch-format" directory, because it
885
898
        uses "path.startswith(dir)" on all file paths to determine if directory
1021
1034
 
1022
1035
    def test_connection_error(self):
1023
1036
        """ConnectionError is raised when connection is impossible.
1024
 
        
 
1037
 
1025
1038
        The error should be raised from the first operation on the transport.
1026
1039
        """
1027
1040
        try:
1070
1083
        subdir.stat('./file')
1071
1084
        subdir.stat('.')
1072
1085
 
 
1086
    def test_hardlink(self):
 
1087
        from stat import ST_NLINK
 
1088
 
 
1089
        t = self.get_transport()
 
1090
 
 
1091
        source_name = "original_target"
 
1092
        link_name = "target_link"
 
1093
 
 
1094
        self.build_tree([source_name], transport=t)
 
1095
 
 
1096
        try:
 
1097
            t.hardlink(source_name, link_name)
 
1098
 
 
1099
            self.failUnless(t.has(source_name))
 
1100
            self.failUnless(t.has(link_name))
 
1101
 
 
1102
            st = t.stat(link_name)
 
1103
            self.failUnlessEqual(st[ST_NLINK], 2)
 
1104
        except TransportNotPossible:
 
1105
            raise TestSkipped("Transport %s does not support hardlinks." %
 
1106
                              self._server.__class__)
 
1107
 
 
1108
    def test_symlink(self):
 
1109
        from stat import S_ISLNK
 
1110
 
 
1111
        t = self.get_transport()
 
1112
 
 
1113
        source_name = "original_target"
 
1114
        link_name = "target_link"
 
1115
 
 
1116
        self.build_tree([source_name], transport=t)
 
1117
 
 
1118
        try:
 
1119
            t.symlink(source_name, link_name)
 
1120
 
 
1121
            self.failUnless(t.has(source_name))
 
1122
            self.failUnless(t.has(link_name))
 
1123
 
 
1124
            st = t.stat(link_name)
 
1125
            self.failUnless(S_ISLNK(st.st_mode))
 
1126
        except TransportNotPossible:
 
1127
            raise TestSkipped("Transport %s does not support symlinks." %
 
1128
                              self._server.__class__)
 
1129
        except IOError:
 
1130
            raise tests.KnownFailure("Paramiko fails to create symlinks during tests")
 
1131
 
1073
1132
    def test_list_dir(self):
1074
1133
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1075
1134
        t = self.get_transport()
1115
1174
 
1116
1175
        self.assertListRaises(PathError, t.list_dir, 'q')
1117
1176
        self.assertListRaises(PathError, t.list_dir, 'c/f')
 
1177
        # 'a' is a file, list_dir should raise an error
1118
1178
        self.assertListRaises(PathError, t.list_dir, 'a')
1119
1179
 
1120
1180
    def test_list_dir_result_is_url_escaped(self):
1437
1497
                         'to/dir/b%2525z',
1438
1498
                         'to/bar',]))
1439
1499
 
 
1500
    def test_copy_tree_to_transport(self):
 
1501
        transport = self.get_transport()
 
1502
        if not transport.listable():
 
1503
            self.assertRaises(TransportNotPossible,
 
1504
                              transport.iter_files_recursive)
 
1505
            return
 
1506
        if transport.is_readonly():
 
1507
            return
 
1508
        self.build_tree(['from/',
 
1509
                         'from/dir/',
 
1510
                         'from/dir/foo',
 
1511
                         'from/dir/bar',
 
1512
                         'from/dir/b%25z', # make sure quoting is correct
 
1513
                         'from/bar'],
 
1514
                        transport=transport)
 
1515
        from_transport = transport.clone('from')
 
1516
        to_transport = transport.clone('to')
 
1517
        to_transport.ensure_base()
 
1518
        from_transport.copy_tree_to_transport(to_transport)
 
1519
        paths = set(transport.iter_files_recursive())
 
1520
        self.assertEqual(paths,
 
1521
                    set(['from/dir/foo',
 
1522
                         'from/dir/bar',
 
1523
                         'from/dir/b%2525z',
 
1524
                         'from/bar',
 
1525
                         'to/dir/foo',
 
1526
                         'to/dir/bar',
 
1527
                         'to/dir/b%2525z',
 
1528
                         'to/bar',]))
 
1529
 
1440
1530
    def test_unicode_paths(self):
1441
1531
        """Test that we can read/write files with Unicode names."""
1442
1532
        t = self.get_transport()
1453
1543
                 u'\u65e5', # Kanji person
1454
1544
                ]
1455
1545
 
 
1546
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
 
1547
        if no_unicode_support:
 
1548
            raise tests.KnownFailure("test server cannot handle unicode paths")
 
1549
 
1456
1550
        try:
1457
1551
            self.build_tree(files, transport=t, line_endings='binary')
1458
1552
        except UnicodeError:
1476
1570
        transport.put_bytes('foo', 'bar')
1477
1571
        transport3 = self.get_transport()
1478
1572
        self.check_transport_contents('bar', transport3, 'foo')
1479
 
        # its base should be usable.
1480
 
        transport4 = get_transport(transport.base)
1481
 
        self.check_transport_contents('bar', transport4, 'foo')
1482
1573
 
1483
1574
        # now opening at a relative url should give use a sane result:
1484
1575
        transport.mkdir('newdir')
1485
 
        transport5 = get_transport(transport.base + "newdir")
 
1576
        transport5 = self.get_transport('newdir')
1486
1577
        transport6 = transport5.clone('..')
1487
1578
        self.check_transport_contents('bar', transport6, 'foo')
1488
1579