← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_remote.py

Get BranchConfig working somewhat on RemoteBranches (Robert Collins, Vincent Ladeuil).

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_remote.py
21
21
the object given a transport that supports smartserver rpc operations. 
22
22
"""
23
23
 
 
24
from cStringIO import StringIO
 
25
 
24
26
from bzrlib import bzrdir, remote, tests
25
27
from bzrlib.branch import Branch
26
28
from bzrlib.bzrdir import BzrDir, BzrDirFormat
83
85
        self.assertIsInstance(d, BzrDir)
84
86
 
85
87
 
 
88
class FakeProtocol(object):
 
89
    """Lookalike SmartClientRequestProtocolOne allowing body reading tests."""
 
90
 
 
91
    def __init__(self, body):
 
92
        self._body_buffer = StringIO(body)
 
93
 
 
94
    def read_body_bytes(self, count=-1):
 
95
        return self._body_buffer.read(count)
 
96
 
 
97
 
86
98
class FakeClient(SmartClient):
87
99
    """Lookalike for SmartClient allowing testing."""
88
100
    
89
101
    def __init__(self, responses):
90
102
        # We don't call the super init because there is no medium.
 
103
        """create a FakeClient.
 
104
 
 
105
        :param respones: A list of response-tuple, body-data pairs to be sent
 
106
            back to callers.
 
107
        """
91
108
        self.responses = responses
92
109
        self._calls = []
93
110
 
94
111
    def call(self, method, *args):
95
112
        self._calls.append(('call', method, args))
96
 
        return self.responses.pop(0)
 
113
        return self.responses.pop(0)[0]
 
114
 
 
115
    def call2(self, method, *args):
 
116
        self._calls.append(('call2', method, args))
 
117
        result = self.responses.pop(0)
 
118
        return result[0], FakeProtocol(result[1])
97
119
 
98
120
 
99
121
class TestBranchLastRevisionInfo(tests.TestCase):
100
122
 
101
123
    def test_empty_branch(self):
102
124
        # in an empty branch we decode the response properly
103
 
        client = FakeClient([('ok', '0', '')])
 
125
        client = FakeClient([(('ok', '0', ''), )])
104
126
        transport = MemoryTransport()
105
127
        transport.mkdir('quack')
106
128
        transport = transport.clone('quack')
117
139
    def test_non_empty_branch(self):
118
140
        # in a non-empty branch we also decode the response properly
119
141
 
120
 
        client = FakeClient([('ok', '2', u'\xc8'.encode('utf8'))])
 
142
        client = FakeClient([(('ok', '2', u'\xc8'.encode('utf8')), )])
121
143
        transport = MemoryTransport()
122
144
        transport.mkdir('kwaak')
123
145
        transport = transport.clone('kwaak')
132
154
        self.assertEqual((2, u'\xc8'), result)
133
155
 
134
156
 
 
157
class TestBranchControlGetBranchConf(tests.TestCase):
 
158
    """Test branch.control_files api munging...
 
159
 
 
160
    we special case RemoteBranch.control_files.get('branch.conf') to 
 
161
    call a specific API so that RemoteBranch's can intercept configuration
 
162
    file reading, allowing them to signal to the client about things like
 
163
    'email is configured for commits'.
 
164
    """
 
165
 
 
166
    def test_get_branch_conf(self):
 
167
        # in an empty branch we decode the response properly
 
168
        client = FakeClient([(('ok', ), 'config file body')])
 
169
        transport = MemoryTransport()
 
170
        transport.mkdir('quack')
 
171
        transport = transport.clone('quack')
 
172
        # we do not want bzrdir to make any remote calls
 
173
        bzrdir = RemoteBzrDir(transport, _client=False)
 
174
        branch = RemoteBranch(bzrdir, None, _client=client)
 
175
        result = branch.control_files.get('branch.conf')
 
176
        self.assertEqual(
 
177
            [('call2', 'Branch.get_config_file', ('///quack/',))],
 
178
            client._calls)
 
179
        self.assertEqual('config file body', result.read())
 
180
 
 
181
 
135
182
class TestRepositoryIsShared(tests.TestCase):
136
183
 
137
184
    def setup_fake_client_and_repository(self, responses, transport_path):
147
194
 
148
195
    def test_is_shared(self):
149
196
        # ('yes', ) for Repository.is_shared -> 'True'.
150
 
        responses = [('yes', )]
 
197
        responses = [(('yes', ), )]
151
198
        transport_path = 'quack'
152
199
        repo, client = self.setup_fake_client_and_repository(
153
200
            responses, transport_path)
159
206
 
160
207
    def test_is_not_shared(self):
161
208
        # ('no', ) for Repository.is_shared -> 'False'.
162
 
        responses = [('no', )]
 
209
        responses = [(('no', ), )]
163
210
        transport_path = 'qwack'
164
211
        repo, client = self.setup_fake_client_and_repository(
165
212
            responses, transport_path)