~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

Ensure that all transports close their underlying connection.

* bzrlib/transport/sftp.py:
(SFTPTransport.disconnect): Close the sftp connection.

* bzrlib/transport/remote.py:
(RemoteTransport.disconnect): The medium can be None.

* bzrlib/transport/http/_urllib.py:
(HttpTransport_urllib.disconnect): Close the http connection.

* bzrlib/transport/http/_pycurl.py:
(PyCurlTransport.disconnect): Close the Curl object.

* bzrlib/transport/http/__init__.py:
(SmartClientHTTPMedium.disconnect): Disconnect the backing http
transport.

* bzrlib/transport/gio_transport.py:
(GioTransport.disconnect): Hmm, nothing obvious here.

* bzrlib/transport/ftp/__init__.py:
(FtpTransport.disconnect): Close the ftplib connection.

* bzrlib/transport/__init__.py:
(Transport.disconnect): A do-nothing implementation by default.
(ConnectedTransport.disconnect): A new required method.

* bzrlib/tests/test_smart_transport.py:
(TestSmartTCPServer.test_get_error_unexpected)
(SmartTCPTests.start_server): Revert to SmartServer waiting for a
better way to capture the connections.

* bzrlib/tests/per_transport.py:
(TransportTests.test_connection_sharing): 'None' is better than
'object()' for a dummy connection.

* bzrlib/tests/__init__.py:
(TestCase._check_leaked_threads): Don't fail at first leak or all
subsequent tests will fail too. A more elaborate scheme would be
to use threading.enumerate() but that's a big hammer with little
benefits as long as there exist "valid" leaks (like the smart
server).
(TestCaseWithMemoryTransport.setUp): Make sure we call
transport.disconnect() for all created transports.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1265
1265
        self.assertIs(t._get_connection(), c._get_connection())
1266
1266
 
1267
1267
        # Temporary failure, we need to create a new dummy connection
1268
 
        new_connection = object()
 
1268
        new_connection = None
1269
1269
        t._set_connection(new_connection)
1270
1270
        # Check that both transports use the same connection
1271
1271
        self.assertIs(new_connection, t._get_connection())