← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport_implementations.py

  • Committer: Matt Nordhoff
  • Date: 2009-04-04 02:50:01 UTC
  • mfrom: (4253 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4256.
  • Revision ID: mnordhoff@mattnordhoff.com-20090404025001-z1403k0tatmc8l91
Merge bzr.dev, fixing conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport_implementations.py
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
55
from bzrlib.tests.test_transport import TestTransportImplementation
56
56
from bzrlib.transport import (
61
61
from bzrlib.transport.memory import MemoryTransport
62
62
 
63
63
 
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
 
64
def get_transport_test_permutations(module):
 
65
    """Get the permutations module wants to have tested."""
 
66
    if getattr(module, 'get_test_permutations', None) is None:
 
67
        raise AssertionError(
 
68
            "transport module %s doesn't provide get_test_permutations()"
 
69
            % module.__name__)
 
70
        return []
 
71
    return module.get_test_permutations()
 
72
 
 
73
 
 
74
def transport_test_permutations():
 
75
    """Return a list of the klass, server_factory pairs to test."""
 
76
    result = []
 
77
    for module in _get_transport_modules():
 
78
        try:
 
79
            permutations = get_transport_test_permutations(
 
80
                reduce(getattr, (module).split('.')[1:], __import__(module)))
 
81
            for (klass, server_factory) in permutations:
 
82
                scenario = (server_factory.__name__,
 
83
                    {"transport_class":klass,
 
84
                     "transport_server":server_factory})
 
85
                result.append(scenario)
 
86
        except errors.DependencyNotPresent, e:
 
87
            # Continue even if a dependency prevents us
 
88
            # from adding this test
 
89
            pass
 
90
    return result
101
91
 
102
92
 
103
93
def load_tests(standard_tests, module, loader):
104
94
    """Multiply tests for tranport implementations."""
105
95
    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
 
96
    scenarios = transport_test_permutations()
 
97
    return multiply_tests(standard_tests, scenarios, result)
110
98
 
111
99
 
112
100
class TransportTests(TestTransportImplementation):
167
155
        self.assertEqual(True, t.has('a'))
168
156
        self.assertEqual(False, t.has('c'))
169
157
        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])
 
158
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd',
 
159
                                           'e', 'f', 'g', 'h'])),
 
160
                         [True, True, False, False,
 
161
                          True, False, True, False])
172
162
        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])
 
163
        self.assertEqual(False, t.has_any(['c', 'd', 'f',
 
164
                                           urlutils.escape('%%')]))
 
165
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd',
 
166
                                                'e', 'f', 'g', 'h']))),
 
167
                         [True, True, False, False,
 
168
                          True, False, True, False])
176
169
        self.assertEqual(False, t.has_any(['c', 'c', 'c']))
177
170
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
178
171
 
325
318
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
326
319
                               create_parent_dir=True)
327
320
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
328
 
        
 
321
 
329
322
        # But we still get NoSuchFile if we can't make the parent dir
330
323
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
331
324
                                       'contents\n',
353
346
        umask = osutils.get_umask()
354
347
        t.put_bytes('nomode', 'test text\n', mode=None)
355
348
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
356
 
        
 
349
 
357
350
    def test_put_bytes_non_atomic_permissions(self):
358
351
        t = self.get_transport()
359
352
 
387
380
        t.put_bytes_non_atomic('dir777/mode664', 'test text\n', mode=0664,
388
381
                               dir_mode=0777, create_parent_dir=True)
389
382
        self.assertTransportMode(t, 'dir777', 0777)
390
 
        
 
383
 
391
384
    def test_put_file(self):
392
385
        t = self.get_transport()
393
386
 
441
434
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
442
435
                              create_parent_dir=True)
443
436
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
444
 
        
 
437
 
445
438
        # But we still get NoSuchFile if we can't make the parent dir
446
439
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
447
440
                                       StringIO('contents\n'),
469
462
        umask = osutils.get_umask()
470
463
        t.put_file('nomode', StringIO('test text\n'), mode=None)
471
464
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
472
 
        
 
465
 
473
466
    def test_put_file_non_atomic_permissions(self):
474
467
        t = self.get_transport()
475
468
 
492
485
        umask = osutils.get_umask()
493
486
        t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
494
487
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
495
 
        
 
488
 
496
489
        # We should also be able to set the mode for a parent directory
497
490
        # when it is created
498
491
        sio = StringIO()
538
531
        t = self.get_transport()
539
532
 
540
533
        if t.is_readonly():
541
 
            # cannot mkdir on readonly transports. We're not testing for 
 
534
            # cannot mkdir on readonly transports. We're not testing for
542
535
            # cache coherency because cache behaviour is not currently
543
536
            # defined for the transport interface.
544
537
            self.assertRaises(TransportNotPossible, t.mkdir, '.')
565
558
 
566
559
        # we were testing that a local mkdir followed by a transport
567
560
        # mkdir failed thusly, but given that we * in one process * do not
568
 
        # concurrently fiddle with disk dirs and then use transport to do 
 
561
        # concurrently fiddle with disk dirs and then use transport to do
569
562
        # things, the win here seems marginal compared to the constraint on
570
563
        # the interface. RBC 20051227
571
564
        t.mkdir('dir_g')
790
783
                t.append_file, 'f', StringIO('f'), mode=None)
791
784
            return
792
785
        t.append_file('f', StringIO('f'), mode=None)
793
 
        
 
786
 
794
787
    def test_append_bytes_mode(self):
795
788
        # check append_bytes accepts a mode
796
789
        t = self.get_transport()
799
792
                t.append_bytes, 'f', 'f', mode=None)
800
793
            return
801
794
        t.append_bytes('f', 'f', mode=None)
802
 
        
 
795
 
803
796
    def test_delete(self):
804
797
        # TODO: Test Transport.delete
805
798
        t = self.get_transport()
866
859
 
867
860
    def test_rmdir_not_empty(self):
868
861
        """Deleting a non-empty directory raises an exception
869
 
        
 
862
 
870
863
        sftp (and possibly others) don't give us a specific "directory not
871
864
        empty" exception -- we can just see that the operation failed.
872
865
        """
879
872
 
880
873
    def test_rmdir_empty_but_similar_prefix(self):
881
874
        """rmdir does not get confused by sibling paths.
882
 
        
 
875
 
883
876
        A naive implementation of MemoryTransport would refuse to rmdir
884
877
        ".bzr/branch" if there is a ".bzr/branch-format" directory, because it
885
878
        uses "path.startswith(dir)" on all file paths to determine if directory
1021
1014
 
1022
1015
    def test_connection_error(self):
1023
1016
        """ConnectionError is raised when connection is impossible.
1024
 
        
 
1017
 
1025
1018
        The error should be raised from the first operation on the transport.
1026
1019
        """
1027
1020
        try:
1115
1108
 
1116
1109
        self.assertListRaises(PathError, t.list_dir, 'q')
1117
1110
        self.assertListRaises(PathError, t.list_dir, 'c/f')
 
1111
        # 'a' is a file, list_dir should raise an error
1118
1112
        self.assertListRaises(PathError, t.list_dir, 'a')
1119
1113
 
1120
1114
    def test_list_dir_result_is_url_escaped(self):
1506
1500
        transport.put_bytes('foo', 'bar')
1507
1501
        transport3 = self.get_transport()
1508
1502
        self.check_transport_contents('bar', transport3, 'foo')
1509
 
        # its base should be usable.
1510
 
        transport4 = get_transport(transport.base)
1511
 
        self.check_transport_contents('bar', transport4, 'foo')
1512
1503
 
1513
1504
        # now opening at a relative url should give use a sane result:
1514
1505
        transport.mkdir('newdir')
1515
 
        transport5 = get_transport(transport.base + "newdir")
 
1506
        transport5 = self.get_transport('newdir')
1516
1507
        transport6 = transport5.clone('..')
1517
1508
        self.check_transport_contents('bar', transport6, 'foo')
1518
1509