/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Jelmer Vernooij
  • Date: 2012-01-03 22:13:09 UTC
  • mfrom: (6421 +trunk)
  • mto: (6437.3.23 2.5)
  • mto: This revision was merged to the branch mainline in revision 6451.
  • Revision ID: jelmer@samba.org-20120103221309-i37nc7q09erm17xx
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
2734
2734
 
2735
2735
    def setUp(self):
2736
2736
        super(TestCaseWithMemoryTransport, self).setUp()
2737
 
        # Ensure that ConnectedTransport doesn't leak sockets
2738
 
        def get_transport_from_url_with_cleanup(*args, **kwargs):
2739
 
            t = orig_get_transport_from_url(*args, **kwargs)
2740
 
            if isinstance(t, _mod_transport.ConnectedTransport):
2741
 
                self.addCleanup(t.disconnect)
2742
 
            return t
2743
 
 
2744
 
        orig_get_transport_from_url = self.overrideAttr(
2745
 
            _mod_transport, 'get_transport_from_url',
2746
 
            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
 
2747
2751
        self._make_test_root()
2748
2752
        self.addCleanup(os.chdir, os.getcwdu())
2749
2753
        self.makeAndChdirToTestDir()
2755
2759
    def setup_smart_server_with_call_log(self):
2756
2760
        """Sets up a smart server as the transport server with a call log."""
2757
2761
        self.transport_server = test_server.SmartTCPServer_for_testing
 
2762
        self.hpss_connections = []
2758
2763
        self.hpss_calls = []
2759
2764
        import traceback
2760
2765
        # Skip the current stack down to the caller of
2763
2768
        def capture_hpss_call(params):
2764
2769
            self.hpss_calls.append(
2765
2770
                CapturedCall(params, prefix_length))
 
2771
        def capture_connect(transport):
 
2772
            self.hpss_connections.append(transport)
2766
2773
        client._SmartClient.hooks.install_named_hook(
2767
2774
            'call', capture_hpss_call, None)
 
2775
        _mod_transport.Transport.hooks.install_named_hook(
 
2776
            'post_connect', capture_connect, None)
2768
2777
 
2769
2778
    def reset_smart_call_log(self):
2770
2779
        self.hpss_calls = []
 
2780
        self.hpss_connections = []
2771
2781
 
2772
2782
 
2773
2783
class TestCaseInTempDir(TestCaseWithMemoryTransport):