← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Jelmer Vernooij
  • Date: 2012-01-05 16:03:11 UTC
  • mfrom: (6432 +trunk)
  • mto: This revision was merged to the branch mainline in revision 6433.
  • Revision ID: jelmer@samba.org-20120105160311-12o5p67kin1v3zps
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
16
16
 
17
17
"""Testing framework extensions"""
18
18
 
 
19
from __future__ import absolute_import
 
20
 
19
21
# NOTE: Some classes in here use camelCaseNaming() rather than
20
22
# underscore_naming().  That's for consistency with unittest; it's not the
21
23
# general style of bzrlib.  Please continue that consistency when adding e.g.
2732
2734
 
2733
2735
    def setUp(self):
2734
2736
        super(TestCaseWithMemoryTransport, self).setUp()
2735
 
        # Ensure that ConnectedTransport doesn't leak sockets
2736
 
        def get_transport_from_url_with_cleanup(*args, **kwargs):
2737
 
            t = orig_get_transport_from_url(*args, **kwargs)
2738
 
            if isinstance(t, _mod_transport.ConnectedTransport):
2739
 
                self.addCleanup(t.disconnect)
2740
 
            return t
2741
 
 
2742
 
        orig_get_transport_from_url = self.overrideAttr(
2743
 
            _mod_transport, 'get_transport_from_url',
2744
 
            get_transport_from_url_with_cleanup)
 
2737
 
 
2738
        def _add_disconnect_cleanup(transport):
 
2739
            """Schedule disconnection of given transport at test cleanup
 
2740
 
 
2741
            This needs to happen for all connected transports or leaks occur.
 
2742
 
 
2743
            Note reconnections may mean we call disconnect multiple times per
 
2744
            transport which is suboptimal but seems harmless.
 
2745
            """
 
2746
            self.addCleanup(transport.disconnect)
 
2747
 
 
2748
        _mod_transport.Transport.hooks.install_named_hook('post_connect',
 
2749
            _add_disconnect_cleanup, None)
 
2750
 
2745
2751
        self._make_test_root()
2746
2752
        self.addCleanup(os.chdir, os.getcwdu())
2747
2753
        self.makeAndChdirToTestDir()
2753
2759
    def setup_smart_server_with_call_log(self):
2754
2760
        """Sets up a smart server as the transport server with a call log."""
2755
2761
        self.transport_server = test_server.SmartTCPServer_for_testing
 
2762
        self.hpss_connections = []
2756
2763
        self.hpss_calls = []
2757
2764
        import traceback
2758
2765
        # Skip the current stack down to the caller of
2761
2768
        def capture_hpss_call(params):
2762
2769
            self.hpss_calls.append(
2763
2770
                CapturedCall(params, prefix_length))
 
2771
        def capture_connect(transport):
 
2772
            self.hpss_connections.append(transport)
2764
2773
        client._SmartClient.hooks.install_named_hook(
2765
2774
            'call', capture_hpss_call, None)
 
2775
        _mod_transport.Transport.hooks.install_named_hook(
 
2776
            'post_connect', capture_connect, None)
2766
2777
 
2767
2778
    def reset_smart_call_log(self):
2768
2779
        self.hpss_calls = []
 
2780
        self.hpss_connections = []
2769
2781
 
2770
2782
 
2771
2783
class TestCaseInTempDir(TestCaseWithMemoryTransport):