1
# Copyright (C) 2006, 2007 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
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
17
"""Tests for remote bzrdir/branch/repo/etc
19
These are proxy objects which act on remote objects by sending messages
20
through a smart client. The proxies are to be created when attempting to open
21
the object given a transport that supports smartserver rpc operations.
24
from cStringIO import StringIO
26
from bzrlib import bzrdir, remote, tests
27
from bzrlib.branch import Branch
28
from bzrlib.bzrdir import BzrDir, BzrDirFormat
29
from bzrlib.remote import (
35
from bzrlib.revision import NULL_REVISION
36
from bzrlib.smart import server
37
from bzrlib.smart.client import SmartClient
38
from bzrlib.transport import remote as remote_transport
39
from bzrlib.transport.memory import MemoryTransport
42
class BasicRemoteObjectTests(tests.TestCaseWithTransport):
45
super(BasicRemoteObjectTests, self).setUp()
46
self.transport_server = server.SmartTCPServer_for_testing
47
self.transport = self.get_transport()
48
self.client = self.transport.get_smart_client()
49
# make a branch that can be opened over the smart transport
50
self.local_wt = BzrDir.create_standalone_workingtree('.')
52
def test_create_remote_bzrdir(self):
53
b = remote.RemoteBzrDir(self.transport)
54
self.assertIsInstance(b, BzrDir)
56
def test_open_remote_branch(self):
57
# open a standalone branch in the working directory
58
b = remote.RemoteBzrDir(self.transport)
59
branch = b.open_branch()
61
def test_remote_repository(self):
62
b = BzrDir.open_from_transport(self.transport)
63
repo = b.open_repository()
64
revid = u'\xc823123123'
65
self.assertFalse(repo.has_revision(revid))
66
self.local_wt.commit(message='test commit', rev_id=revid)
67
self.assertTrue(repo.has_revision(revid))
69
def test_remote_branch_revision_history(self):
70
b = BzrDir.open_from_transport(self.transport).open_branch()
71
self.assertEqual([], b.revision_history())
72
r1 = self.local_wt.commit('1st commit')
73
r2 = self.local_wt.commit('1st commit', rev_id=u'\xc8')
74
self.assertEqual([r1, r2], b.revision_history())
76
def test_find_correct_format(self):
77
"""Should open a RemoteBzrDir over a RemoteTransport"""
78
fmt = BzrDirFormat.find_format(self.transport)
79
self.assertTrue(RemoteBzrDirFormat in BzrDirFormat._control_formats)
80
self.assertIsInstance(fmt, remote.RemoteBzrDirFormat)
82
def test_open_detected_smart_format(self):
83
fmt = BzrDirFormat.find_format(self.transport)
84
d = fmt.open(self.transport)
85
self.assertIsInstance(d, BzrDir)
88
class FakeProtocol(object):
89
"""Lookalike SmartClientRequestProtocolOne allowing body reading tests."""
91
def __init__(self, body):
92
self._body_buffer = StringIO(body)
94
def read_body_bytes(self, count=-1):
95
return self._body_buffer.read(count)
98
class FakeClient(SmartClient):
99
"""Lookalike for SmartClient allowing testing."""
101
def __init__(self, responses):
102
# We don't call the super init because there is no medium.
103
"""create a FakeClient.
105
:param respones: A list of response-tuple, body-data pairs to be sent
108
self.responses = responses
111
def call(self, method, *args):
112
self._calls.append(('call', method, args))
113
return self.responses.pop(0)[0]
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])
121
class TestBranchLastRevisionInfo(tests.TestCase):
123
def test_empty_branch(self):
124
# in an empty branch we decode the response properly
125
client = FakeClient([(('ok', '0', ''), )])
126
transport = MemoryTransport()
127
transport.mkdir('quack')
128
transport = transport.clone('quack')
129
# we do not want bzrdir to make any remote calls
130
bzrdir = RemoteBzrDir(transport, _client=False)
131
branch = RemoteBranch(bzrdir, None, _client=client)
132
result = branch.last_revision_info()
135
[('call', 'Branch.last_revision_info', ('///quack/',))],
137
self.assertEqual((0, NULL_REVISION), result)
139
def test_non_empty_branch(self):
140
# in a non-empty branch we also decode the response properly
142
client = FakeClient([(('ok', '2', u'\xc8'.encode('utf8')), )])
143
transport = MemoryTransport()
144
transport.mkdir('kwaak')
145
transport = transport.clone('kwaak')
146
# we do not want bzrdir to make any remote calls
147
bzrdir = RemoteBzrDir(transport, _client=False)
148
branch = RemoteBranch(bzrdir, None, _client=client)
149
result = branch.last_revision_info()
152
[('call', 'Branch.last_revision_info', ('///kwaak/',))],
154
self.assertEqual((2, u'\xc8'), result)
157
class TestBranchControlGetBranchConf(tests.TestCase):
158
"""Test branch.control_files api munging...
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'.
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')
177
[('call2', 'Branch.get_config_file', ('///quack/',))],
179
self.assertEqual('config file body', result.read())
182
class TestRepositoryIsShared(tests.TestCase):
184
def setup_fake_client_and_repository(self, responses, transport_path):
185
"""Create the fake client and repository for testing with."""
186
client = FakeClient(responses)
187
transport = MemoryTransport()
188
transport.mkdir(transport_path)
189
transport = transport.clone(transport_path)
190
# we do not want bzrdir to make any remote calls
191
bzrdir = RemoteBzrDir(transport, _client=False)
192
repo = RemoteRepository(bzrdir, None, _client=client)
195
def test_is_shared(self):
196
# ('yes', ) for Repository.is_shared -> 'True'.
197
responses = [(('yes', ), )]
198
transport_path = 'quack'
199
repo, client = self.setup_fake_client_and_repository(
200
responses, transport_path)
201
result = repo.is_shared()
203
[('call', 'Repository.is_shared', ('///quack/',))],
205
self.assertEqual(True, result)
207
def test_is_not_shared(self):
208
# ('no', ) for Repository.is_shared -> 'False'.
209
responses = [(('no', ), )]
210
transport_path = 'qwack'
211
repo, client = self.setup_fake_client_and_repository(
212
responses, transport_path)
213
result = repo.is_shared()
215
[('call', 'Repository.is_shared', ('///qwack/',))],
217
self.assertEqual(False, result)
added
removed
bzrlib/tests/test_remote.py
