1
# Copyright (C) 2005, 2006 Canonical Ltd
1
# Copyright (C) 2007 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
20
18
from bzrlib.builtins import cmd_init
21
from bzrlib.hooks import Hooks
22
from bzrlib.tests.test_ftp_transport import TestCaseWithFTPServer
23
from bzrlib.transport import (
27
from bzrlib.transport.ftp import FtpTransport
30
class TransportHooks(Hooks):
31
"""Dict-mapping hook name to a list of callables for transport hooks"""
35
# invoked when the transport is about to create or reuse
36
# an ftp connection. The api signature is (transport, ftp_instance)
40
class InstrumentedTransport(FtpTransport):
41
"""Instrumented transport class to test use by init command"""
43
hooks = TransportHooks()
46
"""See FtpTransport._get_FTP.
48
This is where we can detect if the connection is reused
49
or if a new one is created. This a bit ugly, but it's the
50
easiest until transport classes are refactored.
52
instance = super(InstrumentedTransport, self)._get_FTP()
53
for hook in self.hooks['get_FTP']:
58
class TestInit(TestCaseWithFTPServer):
19
from bzrlib.tests.transport_util import TestCaseWithConnectionHookedTransport
22
class TestInit(TestCaseWithConnectionHookedTransport):
61
25
super(TestInit, self).setUp()
62
InstrumentedTransport.hooks.install_hook('get_FTP',
63
self.get_connection_hook)
64
# Make our instrumented transport the default ftp transport
65
register_transport('ftp://', InstrumentedTransport)
68
InstrumentedTransport.hooks = TransportHooks()
69
unregister_transport('ftp://', InstrumentedTransport)
71
self.addCleanup(cleanup)
75
def get_connection_hook(self, transport, connection):
76
if connection is not None and connection not in self.connections:
77
self.connections.append(connection)
79
28
def test_init(self):