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

  • Committer: John Arbash Meinel
  • Date: 2009-06-17 19:08:25 UTC
  • mto: This revision was merged to the branch mainline in revision 4460.
  • Revision ID: john@arbash-meinel.com-20090617190825-ktfk82li57rf2im6
It seems that fetch() no longer returns the number of revisions fetched.
It still does for *some* InterRepository fetch paths, but the generic one does not.
It is also not easy to get it to, since the Source and Sink are the ones
that would know how many keys were transmitted, and they are potentially 'remote'
objects.

This was also only tested to occur as a by-product in a random 'test_commit' test.
I assume if we really wanted the assurance, we would have a per_repo or interrepo
test for it.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_read_bundle.py
26
26
from bzrlib.symbol_versioning import deprecated_in
27
27
from bzrlib import tests
28
28
from bzrlib.tests.test_transport import TestTransportImplementation
29
 
from bzrlib.tests.per_transport import transport_test_permutations
 
29
from bzrlib.tests.test_transport_implementations import transport_test_permutations
30
30
import bzrlib.transport
31
31
from bzrlib.transport.memory import MemoryTransport
32
32
import bzrlib.urlutils
84
84
class TestReadBundleFromURL(TestTransportImplementation):
85
85
    """Test that read_bundle works properly across multiple transports"""
86
86
 
87
 
    def setUp(self):
88
 
        super(TestReadBundleFromURL, self).setUp()
89
 
        self.bundle_name = 'test_bundle'
90
 
        # read_mergeable_from_url will invoke get_transport which may *not*
91
 
        # respect self._transport (i.e. returns a transport that is different
92
 
        # from the one we want to test, so we must inject a correct transport
93
 
        # into possible_transports first).
94
 
        self.possible_transports = [self.get_transport(self.bundle_name)]
95
 
        self._captureVar('BZR_NO_SMART_VFS', None)
96
 
        wt = self.create_test_bundle()
97
 
 
98
 
    def read_mergeable_from_url(self, url):
99
 
        return bzrlib.bundle.read_mergeable_from_url(
100
 
            url, possible_transports=self.possible_transports)
101
 
 
102
87
    def get_url(self, relpath=''):
103
88
        return bzrlib.urlutils.join(self._server.get_url(), relpath)
104
89
 
105
90
    def create_test_bundle(self):
106
91
        out, wt = create_bundle_file(self)
107
92
        if self.get_transport().is_readonly():
108
 
            self.build_tree_contents([(self.bundle_name, out.getvalue())])
 
93
            f = open('test_bundle', 'wb')
 
94
            try:
 
95
                f.write(out.getvalue())
 
96
            finally:
 
97
                f.close()
109
98
        else:
110
 
            self.get_transport().put_file(self.bundle_name, out)
111
 
            self.log('Put to: %s', self.get_url(self.bundle_name))
 
99
            self.get_transport().put_file('test_bundle', out)
 
100
            self.log('Put to: %s', self.get_url('test_bundle'))
112
101
        return wt
113
102
 
114
103
    def test_read_mergeable_from_url(self):
115
 
        info = self.read_mergeable_from_url(
116
 
            unicode(self.get_url(self.bundle_name)))
 
104
        self._captureVar('BZR_NO_SMART_VFS', None)
 
105
        wt = self.create_test_bundle()
 
106
        if wt is None:
 
107
            return
 
108
        # read_mergeable_from_url will invoke get_transport which may *not*
 
109
        # respect self._transport (i.e. returns a transport that is different
 
110
        # from the one we want to test, so we must inject a correct transport
 
111
        # into possible_transports first.
 
112
        t = self.get_transport('test_bundle')
 
113
        possible_transports = [t]
 
114
        info = bzrlib.bundle.read_mergeable_from_url(
 
115
                    unicode(self.get_url('test_bundle')),
 
116
                    possible_transports=possible_transports)
117
117
        revision = info.real_revisions[-1]
118
118
        self.assertEqual('commit-1', revision.revision_id)
119
119
 
120
120
    def test_read_fail(self):
121
121
        # Trying to read from a directory, or non-bundle file
122
122
        # should fail with NotABundle
123
 
        self.assertRaises(errors.NotABundle,
124
 
                          self.read_mergeable_from_url, self.get_url('tree'))
125
 
        self.assertRaises(errors.NotABundle,
126
 
                          self.read_mergeable_from_url, self.get_url('tree/a'))
 
123
        self._captureVar('BZR_NO_SMART_VFS', None)
 
124
        wt = self.create_test_bundle()
 
125
        if wt is None:
 
126
            return
 
127
 
 
128
        self.assertRaises(errors.NotABundle,
 
129
            bzrlib.bundle.read_mergeable_from_url,
 
130
            self.get_url('tree'))
 
131
        self.assertRaises(errors.NotABundle,
 
132
            bzrlib.bundle.read_mergeable_from_url,
 
133
            self.get_url('tree/a'))
127
134
 
128
135
    def test_read_mergeable_respects_possible_transports(self):
129
 
        if not isinstance(self.get_transport(self.bundle_name),
130
 
                          bzrlib.transport.ConnectedTransport):
 
136
        t = self.get_transport('test_bundle')
 
137
        if not isinstance(t, bzrlib.transport.ConnectedTransport):
131
138
            # There is no point testing transport reuse for not connected
132
139
            # transports (the test will fail even).
133
 
            raise tests.TestSkipped(
134
 
                'Need a ConnectedTransport to test transport reuse')
135
 
        url = unicode(self.get_url(self.bundle_name))
136
 
        info = self.read_mergeable_from_url(url)
137
 
        self.assertEqual(1, len(self.possible_transports))
 
140
            return
 
141
        self._captureVar('BZR_NO_SMART_VFS', None)
 
142
        wt = self.create_test_bundle()
 
143
        if wt is None:
 
144
            return
 
145
        # read_mergeable_from_url will invoke get_transport which may *not*
 
146
        # respect self._transport (i.e. returns a transport that is different
 
147
        # from the one we want to test, so we must inject a correct transport
 
148
        # into possible_transports first.
 
149
        possible_transports = [t]
 
150
        url = unicode(self.get_url('test_bundle'))
 
151
        info = bzrlib.bundle.read_mergeable_from_url(url,
 
152
            possible_transports=possible_transports)
 
153
        self.assertEqual(1, len(possible_transports))