~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/commands/test_init.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2007-07-22 18:09:04 UTC
  • mfrom: (2485.8.63 bzr.connection.sharing)
  • Revision ID: pqm@pqm.ubuntu.com-20070722180904-wy7y7oyi32wbghgf
Transport connection sharing

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006 Canonical Ltd
 
1
# Copyright (C) 2007 Canonical Ltd
2
2
#
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
16
16
 
17
17
 
18
 
import os
19
 
 
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 (
24
 
    register_transport,
25
 
    unregister_transport,
26
 
    )
27
 
from bzrlib.transport.ftp import FtpTransport
28
 
 
29
 
 
30
 
class TransportHooks(Hooks):
31
 
    """Dict-mapping hook name to a list of callables for transport hooks"""
32
 
 
33
 
    def __init__(self):
34
 
        Hooks.__init__(self)
35
 
        # invoked when the transport is about to create or reuse
36
 
        # an ftp connection. The api signature is (transport, ftp_instance)
37
 
        self['get_FTP'] = []
38
 
 
39
 
 
40
 
class InstrumentedTransport(FtpTransport):
41
 
    """Instrumented transport class to test use by init command"""
42
 
 
43
 
    hooks = TransportHooks()
44
 
 
45
 
    def _get_FTP(self):
46
 
        """See FtpTransport._get_FTP.
47
 
 
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.
51
 
        """
52
 
        instance = super(InstrumentedTransport, self)._get_FTP()
53
 
        for hook in self.hooks['get_FTP']:
54
 
            hook(self, instance)
55
 
        return instance
56
 
 
57
 
 
58
 
class TestInit(TestCaseWithFTPServer):
 
19
from bzrlib.tests.transport_util import TestCaseWithConnectionHookedTransport
 
20
 
 
21
 
 
22
class TestInit(TestCaseWithConnectionHookedTransport):
59
23
 
60
24
    def setUp(self):
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)
66
 
 
67
 
        def cleanup():
68
 
            InstrumentedTransport.hooks = TransportHooks()
69
 
            unregister_transport('ftp://', InstrumentedTransport)
70
 
 
71
 
        self.addCleanup(cleanup)
72
 
        self.connections = []
73
 
 
74
 
 
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)
 
26
        self.install_hooks()
78
27
 
79
28
    def test_init(self):
80
29
        cmd = cmd_init()