Merge lp:~free.ekanayaka/landscape-client/amp-cleanup-8 into lp:~landscape/landscape-client/trunk

[Free Ekanayaka]

Some more cleanup around the AMP-based machinery:

- landscape.amp.ComponentConnector: the disconnect method now uses Twisted's IConnector.disconnect(), without messing with the lower-level ITransport.loseConnection(). The objects implementing IConnector are the ones returned by reactor.connectUNIX/connectTCP/connectSSH and afaiu they're what is supposed to be used for terminating a connection.

- rename RemoteComponentRegistry to ComponentRegistry, to match ComponentPublisher and ComponentConnector

- landscape.configuration: the changes here are due because we know use IConnector.disconnect() vs ITransport.loseConnection(). Unfortunately the code in this module is factored and tested in a far less than ideal way, eventually we should take time to clean it up.

- landscape.lib.amp.RemoteObject: update the constructor docstring and reduce a bit of logic duplication by adding a new _send_method_call helper method, which is now used by __getattr__ and _retry (this makes _retry flush the fake asynchronous connection used in tests, in the same way __getattr__ does).

- landscape.lib.amp.MethodCallServerFactory: rename the "reactor" init parameter to "clock", since only the IReactorTime interface is needed (in this case instances are normally called "clock" and can be mocked with a twisted.internet.task.Clock instance in tests)

- landscape.lib.tests.test_amp: Add a FakeConnector class implementing IConnector, so we can use it unit-tests. Also, I've converted a few deferred-returning (asynchronous) tests to synchronous equivalent.

- landscape.tests.test_amp: Concerted asynchronous tests to synchronous equivalent.

Some more cleanup around the AMP-based machinery:

- landscape.amp.ComponentConnector: the disconnect method now uses Twisted's IConnector.disconnect(), without messing with the lower-level ITransport.loseConnection(). The objects implementing IConnector are the ones returned by reactor.connectUNIX/connectTCP/connectSSH and afaiu they're what is supposed to be used for terminating a connection.

- rename RemoteComponentRegistry to ComponentRegistry, to match ComponentPublisher and ComponentConnector

- landscape.configuration: the changes here are due because we know use IConnector.disconnect() vs ITransport.loseConnection(). Unfortunately the code in this module is factored and tested in a far less than ideal way, eventually we should take time to clean it up.

- landscape.lib.amp.RemoteObject: update the constructor docstring and reduce a bit of logic duplication by adding a new _send_method_call helper method, which is now used by __getattr__ and _retry (this makes _retry flush the fake asynchronous connection used in tests, in the same way __getattr__ does).

- landscape.lib.amp.MethodCallServerFactory: rename the "reactor" init parameter to "clock", since only the IReactorTime interface is needed (in this case instances are normally called "clock" and can be mocked with a twisted.internet.task.Clock instance in tests)

- landscape.lib.tests.test_amp: Add a FakeConnector class implementing IConnector, so we can use it unit-tests. Also, I've converted a few deferred-returning (asynchronous) tests to synchronous equivalent.

- landscape.tests.test_amp: Concerted asynchronous tests to synchronous equivalent.

[Alberto Donato (ack)]

Great! +1

#1:
+ If C{max_retries} is passed to L{RemoteObjectConnector.connet},
+ then it will give up trying to connect after that amout of times.

typos, "connet" and "amout".

review: Approve

[Geoff Teale (tealeg)]

+1 Only trivial comments:

[1] "invokation" should be "invocation"
226 + # invokation letting tests simulate a synchronous transport.

[2] "till" -> "until"
439 + # queue, till the call gets a chance to be retried upon reconnection

review: Approve

[lp:~free.ekanayaka/landscape-client/amp-cleanup-8]

673. By Free Ekanayaka on 2013-05-06

Fix typos

[Free Ekanayaka (free.ekanayaka)]

#1, [1] and [2] all fixed.

=== modified file 'landscape/amp.py'
--- landscape/amp.py	2013-04-30 14:01:55 +0000
+++ landscape/amp.py	2013-05-06 18:17:25 +0000
@@ -95,6 +95,9 @@
         factory.initialDelay = factory.delay = 0.05
         factory.retryOnReconnect = self._retry_on_reconnect
         factory.remote = self.remote
+        factory.maxRetries = max_retries
+        if factor:
+            factory.factor = factor
 
         def fire_reconnect(ignored):
             self._reactor.fire("%s-reconnect" % self.component.name)
@@ -107,9 +110,6 @@
             logging.error("Error while connecting to %s", self.component.name)
             return failure
 
-        factory.maxRetries = max_retries
-        if factor:
-            factory.factor = factor
         socket_path = _get_socket_path(self.component, self._config)
         deferred = factory.getRemoteObject()
         self._connector = self._reactor.connect_unix(socket_path, factory)
@@ -124,14 +124,11 @@
         if self._connector is not None:
             factory = self._connector.factory
             factory.stopTrying()
-            # XXX we should be using self._connector.disconnect() here
-            remote = factory._remote
-            if remote:
-                if remote._sender.protocol.transport:
-                    remote._sender.protocol.transport.loseConnection()
-
-
-class RemoteComponentsRegistry(object):
+            self._connector.disconnect()
+            self._connector = None
+
+
+class ComponentsRegistry(object):
     """
     A global registry for looking up Landscape component connectors by name.
     """

=== modified file 'landscape/broker/amp.py'
--- landscape/broker/amp.py	2013-04-30 14:01:55 +0000
+++ landscape/broker/amp.py	2013-05-06 18:17:25 +0000
@@ -2,7 +2,7 @@
 
 from landscape.lib.amp import RemoteObject, MethodCallArgument
 from landscape.amp import (
-    ComponentConnector, RemoteComponentsRegistry, ComponentPublisher)
+    ComponentConnector, ComponentsRegistry, ComponentPublisher)
 from landscape.broker.server import BrokerServer
 from landscape.broker.client import BrokerClient
 from landscape.monitor.monitor import Monitor
@@ -119,7 +119,7 @@
 
     component = Manager
 
-RemoteComponentsRegistry.register(RemoteBrokerConnector)
-RemoteComponentsRegistry.register(RemoteClientConnector)
-RemoteComponentsRegistry.register(RemoteMonitorConnector)
-RemoteComponentsRegistry.register(RemoteManagerConnector)
+ComponentsRegistry.register(RemoteBrokerConnector)
+ComponentsRegistry.register(RemoteClientConnector)
+ComponentsRegistry.register(RemoteMonitorConnector)
+ComponentsRegistry.register(RemoteManagerConnector)

=== modified file 'landscape/broker/server.py'
--- landscape/broker/server.py	2013-04-29 11:37:27 +0000
+++ landscape/broker/server.py	2013-05-06 18:17:25 +0000
@@ -3,7 +3,7 @@
 from twisted.internet.defer import Deferred
 
 from landscape.lib.twisted_util import gather_results
-from landscape.amp import RemoteComponentsRegistry
+from landscape.amp import ComponentsRegistry
 from landscape.manager.manager import FAILED
 
 
@@ -37,7 +37,7 @@
     @param message_store: The broker's L{MessageStore}.
     """
     name = "broker"
-    connectors_registry = RemoteComponentsRegistry
+    connectors_registry = ComponentsRegistry
 
     def __init__(self, config, reactor, exchange, registration,
                  message_store, pinger):

=== modified file 'landscape/configuration.py'
--- landscape/configuration.py	2013-04-23 11:18:33 +0000
+++ landscape/configuration.py	2013-05-06 18:17:25 +0000
@@ -492,7 +492,7 @@
 
 
 def fetch_base64_ssl_public_certificate(hostname, on_info=print_text,
-    on_error=print_text):
+                                        on_error=print_text):
     """
     Fetch base64 encoded SSL CA certificate from the discovered landscape
     server and return that decoded info.
@@ -622,20 +622,21 @@
         reactor = TwistedReactor()
     exit_with_error = []
 
-    def stop(error=None):
-        if not config.ok_no_register and error is not None:
-            exit_with_error.append(error)
+    def stop(errors):
+        if not config.ok_no_register:
+            for error in errors:
+                if error is not None:
+                    exit_with_error.append(error)
         connector.disconnect()
         reactor.stop()
 
     def failure():
         on_message("Invalid account name or "
                    "registration key.", error=True)
-        stop(2)
+        return 2
 
     def success():
         on_message("System successfully registered.")
-        stop()
 
     def exchange_failure():
         on_message("We were unable to contact the server. "
@@ -643,7 +644,7 @@
                    "The landscape client will continue to try and contact "
                    "the server periodically.",
                    error=True)
-        stop(2)
+        return 2
 
     def handle_registration_errors(failure):
         # We'll get invalid credentials through the signal.
@@ -653,7 +654,7 @@
     def catch_all(failure):
         on_message(failure.getTraceback(), error=True)
         on_message("Unknown error occurred.", error=True)
-        stop(2)
+        return 2
 
     on_message("Please wait... ", "")
 
@@ -670,14 +671,15 @@
         # We consume errors here to ignore errors after the first one.
         # catch_all will be called for the very first deferred that fails.
         results = gather_results(deferreds, consume_errors=True)
-        return results.addErrback(catch_all)
+        results.addErrback(catch_all)
+        results.addCallback(stop)
 
     def got_error(failure):
         on_message("There was an error communicating with the Landscape"
                    " client.", error=True)
         on_message("This machine will be registered with the provided "
                    "details when the client runs.", error=True)
-        stop(2)
+        stop([2])
 
     connector = RemoteBrokerConnector(reactor, config)
     result = connector.connect(max_retries=max_retries, quiet=True)

=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2013-05-02 08:24:39 +0000
+++ landscape/lib/amp.py	2013-05-06 18:17:25 +0000
@@ -318,14 +318,9 @@
 
     def __init__(self, factory):
         """
-        @param protocol: A reference to a connected L{AMP} protocol instance,
-            which will be used to send L{MethodCall} commands.
-        @param retry_on_reconnect: If C{True}, this L{RemoteObject} will retry
-            to perform again requests that failed due to a lost connection, as
-            soon as a new connection is available.
-        @param timeout: A timeout for failed requests, if the L{RemoteObject}
-            can't perform them again successfully within this number of
-            seconds, they will errback with a L{MethodCallError}.
+        @param factory: The L{MethodCallClientFactory} used for connecting to
+            the other peer. Look there if you need to tweak the behavior of
+            this L{RemoteObject}.
         """
         self._sender = None
         self._pending_requests = {}
@@ -340,27 +335,29 @@
         keyword arguments it was called with, and returning a L{Deferred}
         resulting in the L{MethodCall}'s response value.
         """
-
         def send_method_call(*args, **kwargs):
-            result = self._sender.send_method_call(method=method,
-                                                   args=args,
-                                                   kwargs=kwargs)
             deferred = Deferred()
-            result.addCallback(self._handle_result, deferred)
-            result.addErrback(self._handle_failure, method, args, kwargs,
-                              deferred)
-
-            if self._factory.fake_connection is not None:
-                # Transparently flush the connection after a send_method_call
-                # invokation letting tests simulate a synchronous transport.
-                # This is needed because the Twisted's AMP implementation
-                # assume that the transport is asynchronous.
-                self._factory.fake_connection.flush()
-
+            self._send_method_call(method, args, kwargs, deferred)
             return deferred
 
         return send_method_call
 
+    def _send_method_call(self, method, args, kwargs, deferred, call=None):
+        """Send a L{MethodCall} command, adding callbacks to handle retries."""
+        result = self._sender.send_method_call(method=method,
+                                               args=args,
+                                               kwargs=kwargs)
+        result.addCallback(self._handle_result, deferred, call=call)
+        result.addErrback(self._handle_failure, method, args, kwargs,
+                          deferred, call=call)
+
+        if self._factory.fake_connection is not None:
+            # Transparently flush the connection after a send_method_call
+            # invocation letting tests simulate a synchronous transport.
+            # This is needed because the Twisted's AMP implementation
+            # assume that the transport is asynchronous.
+            self._factory.fake_connection.flush()
+
     def _handle_result(self, result, deferred, call=None):
         """Handles a successful C{send_method_call} result.
 
@@ -437,11 +434,7 @@
 
         while requests:
             deferred, (method, args, kwargs, call) = requests.popitem()
-            result = self._sender.send_method_call(method, args, kwargs)
-            result.addCallback(self._handle_result,
-                               deferred=deferred, call=call)
-            result.addErrback(self._handle_failure, method, args, kwargs,
-                              deferred=deferred, call=call)
+            self._send_method_call(method, args, kwargs, deferred, call=call)
 
 
 class MethodCallServerFactory(ServerFactory):
@@ -501,15 +494,14 @@
     # hack can be removed.
     fake_connection = None
 
-    def __init__(self, reactor=None):
+    def __init__(self, clock):
         """
         @param object: The object exposed by the L{MethodCallProtocol}s
             instances created by this factory.
         @param reactor: The reactor used by the created protocols
             to schedule notifications and timeouts.
         """
-        self.reactor = reactor
-        self.clock = self.reactor
+        self.clock = clock
         self.delay = self.initialDelay
         self._connects = []
         self._requests = []
@@ -600,7 +592,7 @@
             delay between subsequent retries should increase. Smaller values
             result in a faster reconnection attempts pace.
         """
-        factory = self.factory(reactor=self._reactor)
+        factory = self.factory(self._reactor)
         factory.maxRetries = max_retries
         if factor:
             factory.factor = factor

=== modified file 'landscape/lib/tests/test_amp.py'
--- landscape/lib/tests/test_amp.py	2013-05-02 08:24:39 +0000
+++ landscape/lib/tests/test_amp.py	2013-05-06 18:17:25 +0000
@@ -22,7 +22,7 @@
         self.stream.append(data)
 
     def loseConnection(self):
-        self.connection.disconnect()
+        raise NotImplemented()
 
     def getPeer(self):
         pass
@@ -37,15 +37,14 @@
     def __init__(self, client, server):
         self.client = client
         self.server = server
-        self.connect()
 
-    def connect(self):
+    def make(self):
         self.server.makeConnection(FakeTransport(self))
         self.client.makeConnection(FakeTransport(self))
 
-    def disconnect(self):
-        reason = Failure(ConnectionDone())
-        connector = self
+    def lose(self, connector, reason):
+        self.server.connectionLost(reason)
+        self.client.connectionLost(reason)
         self.client.factory.clientConnectionLost(connector, reason)
 
     def flush(self):
@@ -53,14 +52,41 @@
         Notify the server of any data written by the client and viceversa.
         """
         while True:
-            if self.client.transport.stream:
+            if self.client.transport and self.client.transport.stream:
                 self.server.dataReceived(self.client.transport.stream.pop(0))
-            elif self.server.transport.stream:
+            elif self.server.transport and self.server.transport.stream:
                 self.client.dataReceived(self.server.transport.stream.pop(0))
             else:
                 break
 
 
+class FakeConnector(object):
+    """Make L{FakeConnection}s using the given server and client factories."""
+
+    def __init__(self, client, server):
+        self.client = client
+        self.server = server
+        self.connection = None
+
+    @property
+    def factory(self):
+        return self.client
+
+    def connect(self):
+        self.connection = FakeConnection(self.client.buildProtocol(None),
+                                         self.server.buildProtocol(None))
+
+        # XXX Let the client factory be aware of this fake connection, so
+        # it can flush it when needed. This is to workaround AMP not
+        # supporting synchronous transports
+        self.client.fake_connection = self.connection
+
+        self.connection.make()
+
+    def disconnect(self):
+        self.connection.lose(self, Failure(ConnectionDone()))
+
+
 class WordsException(Exception):
     """Test exception."""
 
@@ -185,6 +211,7 @@
         server = MethodCallServerProtocol(Words(), METHODS)
         client = MethodCallClientProtocol()
         self.connection = FakeConnection(client, server)
+        self.connection.make()
         self.clock = Clock()
         self.sender = MethodCallSender(client, self.clock)
 
@@ -341,12 +368,11 @@
     def setUp(self):
         super(RemoteObjectTest, self).setUp()
         self.clock = Clock()
-        server = MethodCallServerProtocol(Words(self.clock), METHODS)
-        client = MethodCallClientProtocol()
-        client.factory = WordsFactory(self.clock)
-        self.remote = RemoteObject(client.factory)
-        self.connection = FakeConnection(client, server)
-        client.factory.fake_connection = self.connection
+        self.factory = WordsFactory(self.clock)
+        server_factory = MethodCallServerFactory(Words(self.clock), METHODS)
+        self.connector = FakeConnector(self.factory, server_factory)
+        self.connector.connect()
+        self.remote = self.successResultOf(self.factory.getRemoteObject())
 
     def test_method_call_sender_with_forbidden_method(self):
         """
@@ -455,7 +481,7 @@
 
         # Simulate time advancing and the receiver responding
         self.clock.advance(0.5)
-        self.connection.flush()
+        self.connector.connection.flush()
 
         self.assertEqual(["Cool!"], result)
 
@@ -474,7 +500,7 @@
 
         # Simulate time advancing and the receiver responding
         self.clock.advance(0.5)
-        self.connection.flush()
+        self.connector.connection.flush()
 
         [failure] = result
         failure.trap(MethodCallError)
@@ -521,13 +547,62 @@
         [failure] = result
         failure.trap(MethodCallError)
 
+    def test_method_call_error(self):
+        """
+        If a L{MethodCall} fails due to a L{MethodCallError},
+        the L{RemoteObject} won't try to perform it again, even if the
+        C{retryOnReconnect} error is set, as a L{MethodCallError} is a
+        permanent failure that is not likely to ever succeed.
+        """
+        self.factory.retryOnReconnect = True
+        deferred = self.remote.secret()
+        self.failureResultOf(deferred).trap(MethodCallError)
+
+    def test_retry(self):
+        """
+        If the connection is lost and C{retryOnReconnect} is C{True} on the
+        factory, the L{RemoteObject} will transparently retry to perform
+        the L{MethodCall} requests that failed due to the broken connections.
+        """
+        self.connector.disconnect()
+        deferred = self.remote.capitalize("john")
+
+        # The deferred has not fired yet, because it's been put in the pending
+        # queue, until the call gets a chance to be retried upon reconnection
+        self.assertFalse(deferred.called)
+
+        # Time passes and the factory successfully reconnects
+        self.clock.advance(1)
+
+        # We finally get the result
+        self.assertEqual("John", self.successResultOf(deferred))
+
+    def test_retry_with_method_call_error(self):
+        """
+        If a retried L{MethodCall} request fails due to a L{MethodCallError},
+        the L{RemoteObject} will properly propagate the error to the original
+        caller.
+        """
+        self.connector.disconnect()
+        deferred = self.remote.secret()
+
+        # The deferred has not fired yet, because it's been put in the pending
+        # queue, until the call gets a chance to be retried upon reconnection
+        self.assertFalse(deferred.called)
+
+        # Time passes and the factory successfully reconnects
+        self.clock.advance(1)
+
+        failure = self.failureResultOf(deferred)
+        self.assertEqual("Forbidden method 'secret'", str(failure.value))
+
 
 class MethodCallClientFactoryTest(LandscapeTest):
 
     def setUp(self):
         super(MethodCallClientFactoryTest, self).setUp()
         self.clock = Clock()
-        self.factory = MethodCallClientFactory(reactor=self.clock)
+        self.factory = MethodCallClientFactory(self.clock)
 
     def test_max_delay(self):
         """
@@ -598,6 +673,21 @@
         self.clock.advance(5)
         self.assertTrue(connector.called)
 
+    def test_reconnect(self):
+        """
+        If the connection is lost, the L{RemoteObject} created by the creator
+        will transparently handle the reconnection.
+        """
+        server_factory = MethodCallServerFactory(Words(self.clock), METHODS)
+        connector = FakeConnector(self.factory, server_factory)
+        connector.connect()
+        remote = self.successResultOf(self.factory.getRemoteObject())
+
+        connector.disconnect()
+        self.clock.advance(5)
+        deferred = remote.empty()
+        self.assertIsNone(self.successResultOf(deferred))
+
 
 class RemoteObjectConnectorTest(LandscapeTest):
 
@@ -627,115 +717,6 @@
         """
         return self.assertSuccess(self.words.empty())
 
-    def test_connect_with_max_retries(self):
-        """
-        If C{max_retries} is passed to L{RemoteObjectConnector.connet},
-        then it will give up trying to connect after that amout of times.
-        """
-        self.connector.disconnect()
-        self.port.stopListening()
-
-        def reconnect(ignored):
-            self.port = reactor.listenUNIX(self.socket, self.server_factory)
-            return self.connector.connect()
-
-        result = self.connector.connect(max_retries=0)
-        self.assertFailure(result, ConnectError)
-        return result.addCallback(reconnect)
-
-    def test_connect_with_factor(self):
-        """
-        If C{factor} is passed to L{RemoteObjectConnector.connect} method,
-        then the associated protocol factory will be set to that value.
-        """
-        self.connector.disconnect()
-
-        def assert_factor(ignored):
-            self.assertEqual(self.connector._connector.factory.factor, 1.0)
-
-        result = self.connector.connect(factor=1.0)
-        return result.addCallback(assert_factor)
-
-    def test_disconnect(self):
-        """
-        It is possible to call L{RemoteObjectConnector.disconnect} multiple
-        times, even if the connection has been already closed.
-        """
-        self.connector.disconnect()
-        self.connector.disconnect()
-
-    def test_disconnect_without_connect(self):
-        """
-        It is possible to call L{RemoteObjectConnector.disconnect} even
-        if the connection was never established. In that case the method
-        is effectively a no-op.
-        """
-        connector = RemoteWordsConnector(None, None)
-        connector.disconnect()
-
-    def test_reconnect(self):
-        """
-        If the connection is lost, the L{RemoteObject} created by the creator
-        will transparently handle the reconnection.
-        """
-        self.words._sender.protocol.transport.loseConnection()
-        self.port.stopListening()
-
-        def restart_listening():
-            self.port = reactor.listenUNIX(self.socket, self.server_factory)
-            reactor.callLater(0.3, assert_remote)
-
-        def assert_remote():
-            result = self.words.empty()
-            result.addCallback(lambda x: reconnected.callback(None))
-            return result
-
-        reactor.callLater(0.01, restart_listening)
-        reconnected = Deferred()
-        return reconnected
-
-    def test_method_call_error(self):
-        """
-        If a L{MethodCall} fails due to a L{MethodCallError}, the
-        L{RemoteObject} won't try to perform it again.
-        """
-        return self.assertFailure(self.words.secret(), MethodCallError)
-
-    def test_retry(self):
-        """
-        If the connection is lost, the L{RemoteObject} created by the creator
-        will transparently retry to perform the L{MethodCall} requests that
-        failed due to the broken connection.
-        """
-        self.words._sender.protocol.transport.loseConnection()
-        self.port.stopListening()
-
-        def restart_listening():
-            self.port = reactor.listenUNIX(self.socket, self.server_factory)
-
-        reactor.callLater(0.1, restart_listening)
-        return self.words.empty()
-
-    def test_retry_with_method_call_error(self):
-        """
-        If a retried L{MethodCall} request fails due to a L{MethodCallError},
-        the L{RemoteObject} will properly propagate the error to the original
-        caller.
-        """
-        self.words._sender.protocol.transport.loseConnection()
-        self.port.stopListening()
-
-        def restart_listening():
-            self.port = reactor.listenUNIX(self.socket, self.server_factory)
-
-        def assert_failure(error):
-            self.assertEqual(str(error), "Forbidden method 'secret'")
-
-        reactor.callLater(0.5, restart_listening)
-        result = self.words.secret()
-        self.assertFailure(result, MethodCallError)
-        return result.addCallback(assert_failure)
-
     def test_wb_retry_with_while_still_disconnected(self):
         """
         The L{RemoteObject._retry} method gets called as soon as a new

=== modified file 'landscape/reactor.py'
--- landscape/reactor.py	2013-05-02 08:24:39 +0000
+++ landscape/reactor.py	2013-05-06 18:17:25 +0000
@@ -354,15 +354,15 @@
 
     def connect_unix(self, path, factory):
         server = self._socket_paths.get(path)
-        from landscape.lib.tests.test_amp import FakeConnection
+        from landscape.lib.tests.test_amp import FakeConnector
         if server:
-            connection = FakeConnection(factory.buildProtocol(path),
-                                        server.buildProtocol(path))
-            factory.fake_connection = connection
+            connector = FakeConnector(factory, server)
+            connector.connect()
         else:
             connector = object()  # Fake connector
             failure = Failure(ConnectError("No such file or directory"))
             factory.clientConnectionFailed(connector, failure)
+        return connector
 
     def run(self):
         """Continuously advance this reactor until reactor.stop() is called."""

=== modified file 'landscape/tests/helpers.py'
--- landscape/tests/helpers.py	2013-04-30 14:02:14 +0000
+++ landscape/tests/helpers.py	2013-05-06 18:17:25 +0000
@@ -148,6 +148,19 @@
         else:
             return result[0]
 
+    def assertNoResult(self, deferred):
+        """See C{twisted.trial._synctest._Assertions.assertNoResult}.
+
+        This is a copy of the original method, which is available only
+        since Twisted 12.3.0 (from 2012-12-20).
+        """
+        result = []
+        deferred.addBoth(result.append)
+        if result:
+            self.fail(
+                "No result expected on %r, found %r instead" % (
+                    deferred, result[0]))
+
     def assertDeferredSucceeded(self, deferred):
         self.assertTrue(isinstance(deferred, Deferred))
         called = []

=== modified file 'landscape/tests/test_amp.py'
--- landscape/tests/test_amp.py	2013-04-23 11:18:33 +0000
+++ landscape/tests/test_amp.py	2013-05-06 18:17:25 +0000
@@ -1,5 +1,6 @@
 from twisted.internet.defer import Deferred
 from twisted.internet.error import ConnectError
+from twisted.internet.task import Clock
 
 from landscape.tests.helpers import LandscapeTest
 from landscape.reactor import FakeReactor
@@ -73,11 +74,25 @@
     def setUp(self):
         super(ComponentConnectorTest, self).setUp()
         self.reactor = FakeReactor()
+        # XXX this should be dropped once the FakeReactor doesn't use the
+        # real reactor anymore under the hood.
+        self.reactor._reactor = Clock()
         self.config = Configuration()
         self.config.data_path = self.makeDir()
         self.makeDir(path=self.config.sockets_path)
         self.connector = TestComponentConnector(self.reactor, self.config)
 
+    def test_connect_with_max_retries(self):
+        """
+        If C{max_retries} is passed to L{RemoteObjectConnector.connect},
+        then it will give up trying to connect after that amount of times.
+        """
+        self.log_helper.ignore_errors("Error while connecting to test")
+        deferred = self.connector.connect(max_retries=2)
+        self.assertNoResult(deferred)
+        return
+        self.failureResultOf(deferred).trap(ConnectError)
+
     def test_connect_logs_errors(self):
         """
         Connection errors are logged.
@@ -104,25 +119,47 @@
         An event is fired whenever the connection is established again after
         it has been lost.
         """
-        component = TestComponent()
-        publisher = ComponentPublisher(component, self.reactor, self.config)
-        publisher.start()
-
-        def listen_again():
-            publisher.start()
-
-        def connected(remote):
-            remote._sender.protocol.transport.loseConnection()
-            publisher.stop()
-            self.reactor._reactor.callLater(0.01, listen_again)
-
-        def reconnected():
-            self.connector.disconnect()
-            publisher.stop()
-            deferred.callback(None)
-
-        deferred = Deferred()
-        self.reactor.call_on("test-reconnect", reconnected)
-        result = self.connector.connect()
-        result.addCallback(connected)
-        return deferred
+        reconnects = []
+        self.reactor.call_on("test-reconnect", lambda: reconnects.append(True))
+
+        component = TestComponent()
+        publisher = ComponentPublisher(component, self.reactor, self.config)
+        publisher.start()
+        deferred = self.connector.connect()
+        self.successResultOf(deferred)
+        self.connector._connector.disconnect()  # Simulate a disconnection
+        self.assertEqual([], reconnects)
+        self.reactor._reactor.advance(10)
+        self.assertEqual([True], reconnects)
+
+    def test_connect_with_factor(self):
+        """
+        If C{factor} is passed to the L{ComponentConnector.connect} method,
+        then the associated protocol factory will be set to that value.
+        """
+        component = TestComponent()
+        publisher = ComponentPublisher(component, self.reactor, self.config)
+        publisher.start()
+        deferred = self.connector.connect(factor=1.0)
+        remote = self.successResultOf(deferred)
+        self.assertEqual(1.0, remote._factory.factor)
+
+    def test_disconnect(self):
+        """
+        It is possible to call L{ComponentConnector.disconnect} multiple times,
+        even if the connection has been already closed.
+        """
+        component = TestComponent()
+        publisher = ComponentPublisher(component, self.reactor, self.config)
+        publisher.start()
+        self.connector.connect()
+        self.connector.disconnect()
+        self.connector.disconnect()
+
+    def test_disconnect_without_connect(self):
+        """
+        It is possible to call L{ComponentConnector.disconnect} even if the
+        connection was never established. In that case the method is
+        effectively a no-op.
+        """
+        self.connector.disconnect()