← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_init.py

  • Committer: v.ladeuil+lp at free
  • Date: 2007-05-15 17:40:32 UTC
  • mto: (2485.8.44 bzr.connection.sharing)
  • mto: This revision was merged to the branch mainline in revision 2646.
  • Revision ID: v.ladeuil+lp@free.fr-20070515174032-qzdkangpv29l9e7g
Add a test that check that init connect only once. It fails.

* __init__.py:
(test_suite): Register the new test class.

* test_init.py: 
(InstrumentedTransport): A transport that can track connections.
(TransportHooks): Transport specific hooks.
(TestInit): Iniit command behavior tests.

* ftp.py:
(FtpTransport.__init__): Mark place that need fixing regarding
transport connection sharing

* builtins.py:
(cmd_init.run): Mark places that need fixing regarding transport
connection sharing.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_init.py
 
1
# Copyright (C) 2005, 2006 Canonical Ltd
 
2
#
 
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.
 
7
#
 
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.
 
12
#
 
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
 
16
 
 
17
 
 
18
import os
 
19
 
 
20
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 InstrumentedTransport(FtpTransport):
 
31
    """Instrumented transport class to test use by init command"""
 
32
 
 
33
    def _get_FTP(self):
 
34
        """See FtpTransport._get_FTP.
 
35
 
 
36
        This is where we can detect if the connection is reused
 
37
        or if a new one is created. This a bit ugly, but it's the
 
38
        easiest until transport classes are refactored.
 
39
        """
 
40
        instance = super(InstrumentedTransport, self)._get_FTP()
 
41
        self.hooks.run_hook('get_FTP', self, instance)
 
42
        return instance
 
43
 
 
44
 
 
45
class TransportHooks(Hooks):
 
46
    """Dict-mapping hook name to a list of callables for transport hooks"""
 
47
 
 
48
    def __init__(self):
 
49
        Hooks.__init__(self)
 
50
        # invoked when the transport is about to create or reuse
 
51
        # an ftp connection. The api signature is (transport, ftp_instance)
 
52
        self['get_FTP'] = []
 
53
 
 
54
    # FIXME: Why don't we have Hooks.run_hooks ?
 
55
    def run_hook(self, hook_name, *args, **kwargs):
 
56
        try:
 
57
            hooks = self[hook_name]
 
58
        except KeyError:
 
59
            raise errors.UnknownHook(self.__class__.__name__, hook_name)
 
60
        for hook in hooks:
 
61
            hook(*args, **kwargs)
 
62
 
 
63
    # FIXME: Why don't we have Hooks.uninstall_hook ?
 
64
    def uninstall_hook(self, hook_name, a_callable):
 
65
        try:
 
66
            self[hook_name].remove(a_callable)
 
67
        except KeyError:
 
68
            raise errors.UnknownHook(self.__class__.__name__, hook_name)
 
69
        # FIXME: catch ValueError and raise errors.UnknownHookValue ?
 
70
 
 
71
 
 
72
# install the default hooks into the Branch class.
 
73
InstrumentedTransport.hooks = TransportHooks()
 
74
 
 
75
 
 
76
class TestInit(TestCaseWithFTPServer):
 
77
 
 
78
    def setUp(self):
 
79
        super(TestInit, self).setUp()
 
80
        InstrumentedTransport.hooks.install_hook('get_FTP',
 
81
                                                 self.get_connection_hook)
 
82
        # Make our instrumented transport the default ftp transport
 
83
        register_transport('ftp://', InstrumentedTransport)
 
84
 
 
85
        def cleanup():
 
86
            InstrumentedTransport.hooks.uninstall_hook('get_FTP',
 
87
                                                       self.get_connection_hook)
 
88
            unregister_transport('ftp://', InstrumentedTransport)
 
89
 
 
90
        self.addCleanup(cleanup)
 
91
        self.connections = []
 
92
 
 
93
 
 
94
    def get_connection_hook(self, transport, connection):
 
95
        if connection is not None and connection not in self.connections:
 
96
            self.connections.append(connection)
 
97
 
 
98
    def test_init(self):
 
99
        cmd = cmd_init()
 
100
        cmd.run(self.get_url())
 
101
        self.assertEquals(1, len(self.connections))
 
102
 
 
103
    def test_branch(self):
 
104