28
29
from bzrlib.osutils import abspath
29
30
from bzrlib.repofmt.knitrepo import RepositoryFormatKnit1
30
31
from bzrlib.tests.blackbox import ExternalBase
32
from bzrlib.transport import register_transport, unregister_transport
33
from bzrlib.transport.memory import MemoryServer, MemoryTransport
31
34
from bzrlib.uncommit import uncommit
32
35
from bzrlib.urlutils import local_path_from_url
33
36
from bzrlib.workingtree import WorkingTree
244
247
self.run_bzr_error(['At ../dir you have a valid .bzr control'],
246
249
working_dir='tree')
252
class RedirectingMemoryTransport(MemoryTransport):
254
def mkdir(self, path, mode=None):
255
path = self.abspath(path)[len(self._scheme):]
256
if path == '/source':
257
raise errors.RedirectRequested(
258
path, self._scheme + '/target', is_permanent=True)
259
elif path == '/infinite-loop':
260
raise errors.RedirectRequested(
261
path, self._scheme + '/infinite-loop', is_permanent=True)
263
return super(RedirectingMemoryTransport, self).mkdir(
267
class RedirectingMemoryServer(MemoryServer):
270
self._dirs = {'/': None}
273
self._scheme = 'redirecting-memory+%s:///' % id(self)
274
register_transport(self._scheme, self._memory_factory)
276
def _memory_factory(self, url):
277
result = RedirectingMemoryTransport(url)
278
result._dirs = self._dirs
279
result._files = self._files
280
result._locks = self._locks
284
unregister_transport(self._scheme, self._memory_factory)
287
class TestPushRedirect(ExternalBase):
290
ExternalBase.setUp(self)
291
self.memory_server = RedirectingMemoryServer()
292
self.memory_server.setUp()
293
self.addCleanup(self.memory_server.tearDown)
295
# Make the branch and tree that we'll be pushing.
296
t = self.make_branch_and_tree('tree')
297
self.build_tree(['tree/file'])
301
def test_push_redirects_on_mkdir(self):
302
"""If the push requires a mkdir, push respects redirect requests.
304
This is added primarily to handle lp:/ URI support, so that users can
305
push to new branches by specifying lp:/ URIs.
308
destination_url = self.memory_server.get_url() + 'source'
309
self.run_bzr('push %s' % destination_url)
312
local_revision = Branch.open('tree').last_revision()
313
remote_revision = Branch.open(
314
self.memory_server.get_url() + 'target').last_revision()
315
self.assertEqual(remote_revision, local_revision)
317
def test_push_gracefully_handles_too_many_redirects(self):
318
"""Push fails gracefully if the mkdir generates a large number of
322
destination_url = self.memory_server.get_url() + 'infinite-loop'
323
out, err = self.run_bzr_error(
324
['Too many redirections trying to make %s\\.\n'
325
% re.escape(destination_url)],
326
'push %s' % destination_url, retcode=3)
328
self.assertEqual('', out)