← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_read_bundle.py

  • Committer: John Arbash Meinel
  • Date: 2009-07-29 21:35:05 UTC
  • mfrom: (4576 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4577.
  • Revision ID: john@arbash-meinel.com-20090729213505-tkqsvy1zfpocu75w
Merge bzr.dev 4576 in prep for NEWS

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.test_transport_implementations import transport_test_permutations
 
29
from bzrlib.tests.per_transport 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
 
87
102
    def get_url(self, relpath=''):
88
103
        return bzrlib.urlutils.join(self._server.get_url(), relpath)
89
104
 
90
105
    def create_test_bundle(self):
91
106
        out, wt = create_bundle_file(self)
92
107
        if self.get_transport().is_readonly():
93
 
            f = open('test_bundle', 'wb')
94
 
            try:
95
 
                f.write(out.getvalue())
96
 
            finally:
97
 
                f.close()
 
108
            self.build_tree_contents([(self.bundle_name, out.getvalue())])
98
109
        else:
99
 
            self.get_transport().put_file('test_bundle', out)
100
 
            self.log('Put to: %s', self.get_url('test_bundle'))
 
110
            self.get_transport().put_file(self.bundle_name, out)
 
111
            self.log('Put to: %s', self.get_url(self.bundle_name))
101
112
        return wt
102
113
 
103
114
    def test_read_mergeable_from_url(self):
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)
 
115
        info = self.read_mergeable_from_url(
 
116
            unicode(self.get_url(self.bundle_name)))
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._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'))
 
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'))
134
127
 
135
128
    def test_read_mergeable_respects_possible_transports(self):
136
 
        t = self.get_transport('test_bundle')
137
 
        if not isinstance(t, bzrlib.transport.ConnectedTransport):
 
129
        if not isinstance(self.get_transport(self.bundle_name),
 
130
                          bzrlib.transport.ConnectedTransport):
138
131
            # There is no point testing transport reuse for not connected
139
132
            # transports (the test will fail even).
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))
 
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))