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):
61
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)
19
from bzrlib.tests.TransportUtil import TestCaseWithConnectionHookedTransport
21
class TestInit(TestCaseWithConnectionHookedTransport):
79
23
def test_init(self):