Merge lp:~free.ekanayaka/landscape-client/service-protocol into lp:~landscape/landscape-client/amp-trunk

[Free Ekanayaka]

[lp:~free.ekanayaka/landscape-client/service-protocol]

223. By Free Ekanayaka on 2010-03-04

Remove unused max_retries

[Kevin McDermott (bigkevmcd)]

Looks nice and small, thanks for keeping the branches small :-)

review: Approve

[Free Ekanayaka (free.ekanayaka)]

Thanks Kevin!

@reviewers: For some reason the diff showed below is gigantic, but the actual one is only 282 lines (maybe LP is having troubles with criss-cross merges), however merging this branch to amp-trunk locally with bzr should be just fine.

[lp:~free.ekanayaka/landscape-client/service-protocol]

224. By Free Ekanayaka on 2010-03-05

Simplified MethodCall classes by introducing a factory that can be used both for server-side and client-side protocols

[Kapil Thangavelu (hazmat)]

looks good to me, nice work +1

review: Approve

[lp:~free.ekanayaka/landscape-client/service-protocol]

225. By Free Ekanayaka on 2010-03-08

Drop the Landscape suffix for sake of brevity

=== modified file 'landscape/__init__.py'
--- landscape/__init__.py	2010-03-03 17:06:50 +0000
+++ landscape/__init__.py	2010-03-05 12:44:28 +0000
@@ -1,5 +1,10 @@
+<<<<<<< TREE
 DEBIAN_REVISION = ""
 UPSTREAM_VERSION = "1.4.4"
+=======
+DEBIAN_REVISION = "-0ubuntu0.9.10.0"
+UPSTREAM_VERSION = "1.4.0"
+>>>>>>> MERGE-SOURCE
 VERSION = "%s%s" % (UPSTREAM_VERSION, DEBIAN_REVISION)
 
 # The "server-api" field of outgoing messages will be set to this value, and

=== added file 'landscape/amp.py'
--- landscape/amp.py	1970-01-01 00:00:00 +0000
+++ landscape/amp.py	2010-03-05 12:44:28 +0000
@@ -0,0 +1,75 @@
+import os
+import logging
+
+from landscape.lib.amp import (
+    MethodCallProtocol, MethodCallFactory, RemoteObjectCreator)
+
+
+class LandscapeComponentProtocol(MethodCallProtocol):
+    """Communication protocol between the various Landscape components.
+
+    It can be used both as server-side protocol for exposing the methods of a
+    certain Landscape component, or as client-side protocol for connecting to
+    another Landscape component we want to call the methods of.
+    """
+    methods = ["ping", "exit"]
+    timeout = 60
+
+
+class LandscapeComponentFactory(MethodCallFactory):
+
+    protocol = LandscapeComponentProtocol
+
+
+class RemoteLandscapeComponentCreator(RemoteObjectCreator):
+    """Utility superclass for creating connections with a Landscape component.
+
+    @cvar component: The class of the component to connect to, it is expected
+        to define a C{name} class attribute, which will be used to find out
+        the socket to use. It must be defined by sub-classes.
+    """
+
+    factory = LandscapeComponentFactory
+
+    def __init__(self, reactor, config, *args, **kwargs):
+        """
+        @param reactor: A L{TwistedReactor} object.
+        @param config: A L{LandscapeConfiguration}.
+        @param args: Positional arguments for protocol factory constructor.
+        @param kwargs: Keyword arguments for protocol factory constructor.
+
+        @see: L{MethodCallClientFactory}.
+        """
+        self._twisted_reactor = reactor
+        socket = os.path.join(config.data_path, self.component.name + ".sock")
+        super(RemoteLandscapeComponentCreator, self).__init__(
+            self._twisted_reactor._reactor, socket, *args, **kwargs)
+
+    def connect(self, max_retries=None):
+        """Connect to the remote Landscape component.
+
+        If the connection is lost after having been established, and then
+        it is established again by the reconnect mechanism, an event will
+        be fired.
+
+        @param max_retries: If given, the connector will keep trying to connect
+            up to that number of times, if the first connection attempt fails.
+        """
+
+        def fire_reconnect(remote):
+            self._twisted_reactor.fire("%s-reconnect" %
+                                       self.component.name)
+
+        def connected(remote):
+            self._factory.add_notifier(fire_reconnect)
+            return remote
+
+        def log_error(failure):
+            logging.error("Error while connecting to %s", self.component.name)
+            return failure
+
+        result = super(RemoteLandscapeComponentCreator, self).connect(
+            max_retries=max_retries)
+        result.addErrback(log_error)
+        result.addCallback(connected)
+        return result

=== modified file 'landscape/broker/amp.py'
--- landscape/broker/amp.py	2010-02-11 14:29:50 +0000
+++ landscape/broker/amp.py	2010-03-05 12:44:28 +0000
@@ -1,8 +1,21 @@
-class RemoteClient(object):
-    """A connected client utilizing features provided by a L{BrokerServer}."""
-
-    def __init__(self, name):
-        """
-        @param name: Name of the broker client.
-        """
-        self.name = name
+<<<<<<< TREE
+class RemoteClient(object):
+    """A connected client utilizing features provided by a L{BrokerServer}."""
+
+    def __init__(self, name):
+        """
+        @param name: Name of the broker client.
+        """
+        self.name = name
+=======
+class RemoteClient(object):
+    """A connected client utilizing features provided by a L{BrokerServer}."""
+
+    def __init__(self, name):
+        """
+        @param name: Name of the broker client.
+        @param protocol: A L{BrokerServerProtocol} connection with the broker
+            server.
+        """
+        self.name = name
+>>>>>>> MERGE-SOURCE

=== modified file 'landscape/broker/tests/helpers.py'
--- landscape/broker/tests/helpers.py	2010-02-11 14:29:50 +0000
+++ landscape/broker/tests/helpers.py	2010-03-05 12:44:28 +0000
@@ -99,6 +99,7 @@
             test_case.config, test_case.identity, test_case.reactor,
             test_case.exchanger, test_case.pinger, test_case.mstore,
             fetch_async=fetch_func)
+<<<<<<< TREE
 
 
 class BrokerServerHelper(RegistrationHelper):
@@ -113,3 +114,19 @@
         test_case.broker = BrokerServer(test_case.config, test_case.reactor,
                                         test_case.exchanger, test_case.handler,
                                         test_case.mstore)
+=======
+
+
+class BrokerServerHelper(RegistrationHelper):
+    """
+    This helper adds a broker server to the L{RegistrationHelper}.  The
+    following attributes will be set in your test case:
+      - server: A L{BrokerServer}.
+    """
+
+    def set_up(self, test_case):
+        super(BrokerServerHelper, self).set_up(test_case)
+        test_case.broker = BrokerServer(test_case.config, test_case.reactor,
+                                        test_case.exchanger, test_case.handler,
+                                        test_case.mstore)
+>>>>>>> MERGE-SOURCE

=== modified file 'landscape/broker/tests/test_server.py'
--- landscape/broker/tests/test_server.py	2010-03-03 16:17:11 +0000
+++ landscape/broker/tests/test_server.py	2010-03-05 12:44:28 +0000
@@ -1,3 +1,4 @@
+<<<<<<< TREE
 from twisted.internet.defer import succeed, fail
 
 from landscape.broker.amp import RemoteClient
@@ -212,3 +213,219 @@
         self.reactor.call_on("pre-exit", pre_exit)
         self.reactor.call_on("post-exit", post_exit)
         return self.assertSuccess(self.broker.exit())
+=======
+from twisted.internet.defer import succeed, fail
+
+from landscape.broker.amp import RemoteClient
+from landscape.tests.helpers import LandscapeTest, DEFAULT_ACCEPTED_TYPES
+from landscape.broker.tests.helpers import BrokerServerHelper
+
+
+class BrokerServerTest(LandscapeTest):
+
+    helpers = [BrokerServerHelper]
+
+    def test_ping(self):
+        """
+        The L{BrokerServer.ping} simply returns C{True}.
+        """
+        self.assertTrue(self.broker.ping())
+
+    def test_send_message(self):
+        """
+        The L{BrokerServer.send_message} method forwards a message to the
+        broker's exchanger.
+        """
+        message = {"type": "test"}
+        self.mstore.set_accepted_types(["test"])
+        self.broker.send_message(message)
+        self.assertMessages(self.mstore.get_pending_messages(), [message])
+        self.assertFalse(self.exchanger.is_urgent())
+
+    def test_send_message_with_urgent(self):
+        """
+        The L{BrokerServer.send_message} can optionally specify the urgency
+        of the message.
+        """
+        message = {"type": "test"}
+        self.mstore.set_accepted_types(["test"])
+        self.broker.send_message(message, True)
+        self.assertMessages(self.mstore.get_pending_messages(), [message])
+        self.assertTrue(self.exchanger.is_urgent())
+
+    def test_is_pending(self):
+        """
+        The L{BrokerServer.is_pending} method indicates if a message with
+        the given id is pending waiting for delivery in the message store.
+        """
+        self.assertFalse(self.broker.is_message_pending(123))
+        message = {"type": "test"}
+        self.mstore.set_accepted_types(["test"])
+        message_id = self.broker.send_message(message)
+        self.assertTrue(self.broker.is_message_pending(message_id))
+
+    def test_register_client(self):
+        """
+        The L{BrokerServer.register_client} method can be used to register
+        client components that need to communicate with the server. After
+        the registration they can be fetched with L{BrokerServer.get_clients}.
+        """
+        self.assertEquals(self.broker.get_clients(), [])
+        self.broker.register_client("test")
+        [client] = self.broker.get_clients()
+        self.assertTrue(isinstance(client, RemoteClient))
+        self.assertEquals(client.name, "test")
+
+    def test_stop_clients(self):
+        """
+        The L{BrokerServer.stop_clients} method calls the C{exit} method
+        of each registered client, and returns a deferred resulting in C{None}
+        if all C{exit} calls were successful.
+        """
+        self.broker.register_client("foo")
+        self.broker.register_client("bar")
+        for client in self.broker.get_clients():
+            client.exit = self.mocker.mock()
+            self.expect(client.exit()).result(succeed(None))
+        self.mocker.replay()
+        return self.assertSuccess(self.broker.stop_clients())
+
+    def test_stop_clients_with_failure(self):
+        """
+        The L{BrokerServer.stop_clients} method calls the C{exit} method
+        of each registered client, and returns a deferred resulting in C{None}
+        if all C{exit} calls were successful.
+        """
+        self.broker.register_client("foo")
+        self.broker.register_client("bar")
+        [client1, client2] = self.broker.get_clients()
+        client1.exit = self.mocker.mock()
+        client2.exit = self.mocker.mock()
+        self.expect(client1.exit()).result(succeed(None))
+        self.expect(client2.exit()).result(fail(Exception()))
+        self.mocker.replay()
+        return self.assertFailure(self.broker.stop_clients(), Exception)
+
+    def test_reload_configuration(self):
+        """
+        The L{BrokerServer.reload_configuration} method forces the config
+        file associated with the broker server to be reloaded.
+        """
+        open(self.config_filename, "a").write("computer_title = New Title")
+        result = self.broker.reload_configuration()
+        result.addCallback(lambda x: self.assertEquals(
+            self.config.computer_title, "New Title"))
+        return result
+
+    def test_reload_configuration_stops_clients(self):
+        """
+        The L{BrokerServer.reload_configuration} method forces the config
+        file associated with the broker server to be reloaded.
+        """
+        self.broker.register_client("foo")
+        self.broker.register_client("bar")
+        for client in self.broker.get_clients():
+            client.exit = self.mocker.mock()
+            self.expect(client.exit()).result(succeed(None))
+        self.mocker.replay()
+        return self.assertSuccess(self.broker.reload_configuration())
+
+    def test_register(self):
+        """
+        The L{BrokerServer.register} method attempts to register with the
+        Ladscape server and waits for a C{set-id} message from it.
+        """
+        registered = self.broker.register()
+        # This should callback the deferred.
+        self.exchanger.handle_message({"type": "set-id", "id": "abc",
+                                       "insecure-id": "def"})
+        return self.assertSuccess(registered)
+
+    def test_get_accepted_types_empty(self):
+        """
+        The L{BrokerServer.get_accepted_message_types} returns an empty list
+        if no message types are accepted by the Landscape server.
+        """
+        self.mstore.set_accepted_types([])
+        self.assertEquals(self.broker.get_accepted_message_types(), [])
+
+    def test_get_accepted_message_types(self):
+        """
+        The L{BrokerServer.get_accepted_message_types} returns the list of
+        message types accepted by the Landscape server.
+        """
+        self.mstore.set_accepted_types(["foo", "bar"])
+        self.assertEquals(sorted(self.broker.get_accepted_message_types()),
+                          ["bar", "foo"])
+
+    def test_get_server_uuid_with_unset_uuid(self):
+        """
+        The L{BrokerServer.get_server_uuid} method returns C{None} if the uuid
+        of the Landscape server we're pointing at is unknown.
+        """
+        self.assertEquals(self.broker.get_server_uuid(), None)
+
+    def test_get_server_uuid(self):
+        """
+        The L{BrokerServer.get_server_uuid} method returns the uuid of the
+        Landscape server we're pointing at.
+        """
+        self.mstore.set_server_uuid("the-uuid")
+        self.assertEquals(self.broker.get_server_uuid(), "the-uuid")
+
+    def test_register_client_accepted_message_type(self):
+        """
+        The L{BrokerServer.register_client_accepted_message_type} method can
+        register new message types accepted by this Landscape client.
+        """
+        self.broker.register_client_accepted_message_type("type1")
+        self.broker.register_client_accepted_message_type("type2")
+        self.assertEquals(self.exchanger.get_client_accepted_message_types(),
+                          sorted(["type1", "type2"] + DEFAULT_ACCEPTED_TYPES))
+
+    def test_exit(self):
+        """
+        The L{BrokerServer.exit} method stops all registered clients.
+        """
+        self.broker.register_client("foo")
+        self.broker.register_client("bar")
+        for client in self.broker.get_clients():
+            client.exit = self.mocker.mock()
+            self.expect(client.exit()).result(succeed(None))
+        self.mocker.replay()
+        return self.assertSuccess(self.broker.exit())
+
+    def test_exit_exits_when_other_daemons_blow_up(self):
+        """
+        If a broker client blow up in its exit() methods, exit should ignore
+        the error and exit anyway.
+        """
+        self.broker.register_client("foo")
+        [client] = self.broker.get_clients()
+        client.exit = self.mocker.mock()
+        post_exit = self.mocker.mock()
+        self.expect(client.exit()).result(fail(ZeroDivisionError()))
+        post_exit()
+        self.mocker.replay()
+        self.reactor.call_on("post-exit", post_exit)
+        return self.assertSuccess(self.broker.exit())
+
+    def test_exit_fires_reactor_events(self):
+        """
+        The L{BrokerServer.exit} method fires a C{pre-exit} event before the
+        clients are stopped and a C{post-exit} event after.
+        """
+        self.broker.register_client("foo")
+        [client] = self.broker.get_clients()
+        self.mocker.order()
+        pre_exit = self.mocker.mock()
+        client.exit = self.mocker.mock()
+        post_exit = self.mocker.mock()
+        pre_exit()
+        self.expect(client.exit()).result(fail(ZeroDivisionError()))
+        post_exit()
+        self.mocker.replay()
+        self.reactor.call_on("pre-exit", pre_exit)
+        self.reactor.call_on("post-exit", post_exit)
+        return self.assertSuccess(self.broker.exit())
+>>>>>>> MERGE-SOURCE

=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2010-02-08 08:10:48 +0000
+++ landscape/lib/amp.py	2010-03-05 12:44:28 +0000
@@ -1,3 +1,4 @@
+<<<<<<< TREE
 """Expose the methods of a remote object over AMP."""
 
 from uuid import uuid4
@@ -456,3 +457,463 @@
         """Disconnect the L{RemoteObject} that we have created."""
         self._factory.stopTrying()
         self._remote._protocol.transport.loseConnection()
+=======
+"""Expose the methods of a remote object over AMP."""
+
+from uuid import uuid4
+from twisted.internet.defer import Deferred, maybeDeferred
+from twisted.internet.protocol import ReconnectingClientFactory
+from twisted.protocols.amp import Argument, String, Command, AMP
+from twisted.python.failure import Failure
+
+from landscape.lib.bpickle import loads, dumps, dumps_table
+
+
+class MethodCallArgument(Argument):
+    """A bpickle-compatible argument."""
+
+    def toString(self, inObject):
+        """Serialize an argument."""
+        return dumps(inObject)
+
+    def fromString(self, inString):
+        """Unserialize an argument."""
+        return loads(inString)
+
+    @classmethod
+    def check(cls, inObject):
+        """Check if an argument is serializable."""
+        return type(inObject) in dumps_table
+
+
+class MethodCallError(Exception):
+    """Raised when a L{MethodCall} command fails."""
+
+
+class MethodCall(Command):
+    """Call a method on the object exposed by a L{MethodCallProtocol}."""
+
+    arguments = [("method", String()),
+                 ("args", MethodCallArgument()),
+                 ("kwargs", MethodCallArgument())]
+
+    response = [("result", MethodCallArgument()),
+                ("deferred", String(optional=True))]
+
+    errors = {MethodCallError: "METHOD_CALL_ERROR"}
+
+
+class DeferredResponse(Command):
+    """Fire a L{Deferred} associated with an outstanding method call result."""
+
+    arguments = [("uuid", String()),
+                 ("result", MethodCallArgument(optional=True)),
+                 ("failure", String(optional=True))]
+    requiresAnswer = False
+
+
+class MethodCallServerProtocol(AMP):
+    """Expose methods of a local object over AMP.
+
+    The object to be exposed is expected to be the C{object} attribute of our
+    protocol factory.
+
+    @cvar methods: The list of exposed object's methods that can be called with
+        the protocol. It must be defined by sub-classes.
+    """
+
+    methods = []
+
+    @MethodCall.responder
+    def receive_method_call(self, method, args, kwargs):
+        """Call an object's method with the given arguments.
+
+        If a connected client sends a L{MethodCall} for method C{foo_bar}, then
+        the actual method C{foo_bar} of the object associated with the protocol
+        will be called with the given C{args} and C{kwargs} and its return
+        value delivered back to the client as response to the command.
+
+        @param method: The name of the object's method to call.
+        @param args: The arguments to pass to the method.
+        @param kwargs: The keywords arguments to pass to the method.
+        """
+        if not method in self.methods:
+            raise MethodCallError("Forbidden method '%s'" % method)
+
+        method_func = getattr(self.factory.object, method)
+        result = maybeDeferred(method_func, *args, **kwargs)
+
+        # If the Deferred was already fired, we can return its result
+        if result.called:
+            if isinstance(result.result, Failure):
+                failure = str(result.result.value)
+                result.addErrback(lambda error: None) # Stop propagating
+                raise MethodCallError(failure)
+            return {"result": self._check_result(result.result)}
+
+        uuid = str(uuid4())
+        result.addBoth(self.send_deferred_response, uuid)
+        return {"result": None, "deferred": uuid}
+
+    def _check_result(self, result):
+        """Check that the C{result} we're about to return is serializable.
+
+        @return: The C{result} itself if valid.
+        @raises: L{MethodCallError} if C{result} is not serializable.
+        """
+        if not MethodCallArgument.check(result):
+            raise MethodCallError("Non-serializable result")
+        return result
+
+    def send_deferred_response(self, result, uuid):
+        """Send a L{DeferredResponse} for the deferred with given C{uuid}.
+
+        This is called when the result of a L{Deferred} returned by an
+        object's method becomes available. A L{DeferredResponse} notifying
+        such result (either success or failure) is sent to the peer.
+        """
+        kwargs = {"uuid": uuid}
+        if isinstance(result, Failure):
+            kwargs["failure"] = str(result.value)
+        else:
+            kwargs["result"] = self._check_result(result)
+        self.callRemote(DeferredResponse, **kwargs)
+
+
+class MethodCallClientProtocol(AMP):
+    """Calls methods of a remote object over L{AMP}.
+
+    @note: If the remote method returns a deferred, the associated local
+        deferred returned by L{send_method_call} will result in the same
+        callback value of the remote deferred.
+    @cvar timeout: A timeout for remote methods returning L{Deferred}s, if a
+        response for the deferred is not received within this amount of
+        seconds, the remote method call will errback with a L{MethodCallError}.
+    """
+    timeout = 60
+
+    def __init__(self):
+        AMP.__init__(self)
+        self._pending_responses = {}
+
+    @DeferredResponse.responder
+    def receive_deferred_response(self, uuid, result, failure):
+        """Receive the deferred L{MethodCall} response.
+
+        @param uuid: The id of the L{MethodCall} we're getting the result of.
+        @param result: The result of the associated deferred if successful.
+        @param failure: The failure message of the deferred if it failed.
+        """
+        self.fire_pending_response(uuid, result, failure)
+        return {}
+
+    def fire_pending_response(self, uuid, result, failure):
+        """Fire the L{Deferred} associated with a pending response.
+
+        @param uuid: The id of the L{MethodCall} we're getting the result of.
+        @param result: The result of the associated deferred if successful.
+        @param failure: The failure message of the deferred if it failed.
+        """
+        try:
+            deferred, call = self._pending_responses.pop(uuid)
+        except KeyError:
+            # Late response for a request that has timeout, just ignore it
+            return
+        if not call.called:
+            call.cancel()
+        if failure is None:
+            deferred.callback({"result": result})
+        else:
+            deferred.errback(MethodCallError(failure))
+
+    def handle_response(self, response):
+        """Handle a L{MethodCall} response.
+
+        If the response is tagged as deferred, it will be queued as pending,
+        and a L{Deferred} is returned, which will be fired as soon as the
+        final response becomes available, or the timeout is reached.
+        """
+        if response["deferred"]:
+            uuid = response["deferred"]
+            deferred = Deferred()
+            call = self.factory.reactor.callLater(self.timeout,
+                                                  self.fire_pending_response,
+                                                  uuid, None, "timeout")
+            self._pending_responses[uuid] = (deferred, call)
+            return deferred
+
+        return response
+
+    def send_method_call(self, method, args=[], kwargs={}):
+        """Send a L{MethodCall} command with the given arguments.
+
+        @param method: The name of the remote method to invoke.
+        @param args: The positional arguments to pass to the remote method.
+        @param kwargs: The keyword arguments to pass to the remote method.
+        """
+        result = self.callRemote(MethodCall,
+                                 method=method, args=args, kwargs=kwargs)
+        # The result can be C{None} only if the requested command is a
+        # DeferredResponse, which has requiresAnswer set to False
+        if result is not None:
+            return result.addCallback(self.handle_response)
+
+
+class MethodCallProtocol(MethodCallServerProtocol, MethodCallClientProtocol):
+    """Can be used both for sending and receiving L{MethodCall}s."""
+
+    def __init__(self):
+        MethodCallServerProtocol.__init__(self)
+        MethodCallClientProtocol.__init__(self)
+
+
+class MethodCallFactory(ReconnectingClientFactory):
+    """
+    Factory for L{MethodCallProtocol}s exposing an object or connecting to
+    to L{MethodCall} servers.
+
+    When used to connect, if the connection fails or is lost the factory
+    will keep retrying to establish it.
+
+    @cvar protocol: The factory used to build protocol instances.
+    @cvar factor: The time factor by which the delay between two subsequent
+        connection retries will increase.
+    """
+
+    protocol = MethodCallProtocol
+    factor = 1.6180339887498948
+
+    def __init__(self, object=None, reactor=None):
+        """
+        @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.object = object
+        self.reactor = reactor
+        self.clock = self.reactor
+        self._notifiers = []
+
+    def add_notifier(self, callback, errback=None):
+        """Call the given function on connection, reconnection or give up.
+
+        @param notifier: A function that will be called when the factory builds
+            a new connected protocol or gives up connecting.  It will be passed
+            the new protocol instance as argument, or the connectionf failure.
+        """
+        self._notifiers.append((callback, errback))
+
+    def remove_notifier(self, callback, errback=None):
+        """Remove a notifier."""
+        self._notifiers.remove((callback, errback))
+
+    def notify_success(self, *args, **kwargs):
+        """Notify all registered notifier callbacks."""
+        for callback, _ in self._notifiers:
+            self.reactor.callLater(0, callback, *args, **kwargs)
+
+    def notify_failure(self, failure):
+        """Notify all registered notifier errbacks."""
+        for _, errback in self._notifiers:
+            if errback is not None:
+                self.reactor.callLater(0, errback, failure)
+
+    def clientConnectionFailed(self, connector, reason):
+        ReconnectingClientFactory.clientConnectionFailed(self, connector,
+                                                         reason)
+        if self.maxRetries is not None and (self.retries > self.maxRetries):
+            self.notify_failure(reason) # Give up
+
+    def buildProtocol(self, addr):
+        self.resetDelay()
+        protocol = self.protocol()
+        protocol.factory = self
+        self.notify_success(protocol)
+        return protocol
+
+
+class RemoteObject(object):
+    """An object able to transparently call methods on a remote object.
+
+    Any method call on a L{RemoteObject} instance will return a L{Deferred}
+    resulting in the return value of the same method call performed on
+    the remote object exposed by the peer.
+    """
+
+    def __init__(self, protocol, retry_on_reconnect=False, timeout=None):
+        """
+        @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}.
+        """
+        self._protocol = protocol
+        self._factory = protocol.factory
+        self._reactor = protocol.factory.reactor
+        self._retry_on_reconnect = retry_on_reconnect
+        self._timeout = timeout
+        self._pending_requests = {}
+        self._factory.add_notifier(self._handle_reconnect)
+
+    def __getattr__(self, method):
+        """Return a function sending a L{MethodCall} for the given C{method}.
+
+        When the created function is called, it sends the an appropriate
+        L{MethodCall} to the remote peer passing it the arguments and
+        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._protocol.send_method_call(method=method,
+                                                     args=args,
+                                                     kwargs=kwargs)
+            deferred = Deferred()
+            result.addCallback(self._handle_response, deferred)
+            result.addErrback(self._handle_failure, method, args, kwargs,
+                              deferred)
+            return deferred
+
+        return send_method_call
+
+    def _handle_response(self, response, deferred, call=None):
+        """Handles a successful L{MethodCall} response.
+
+        @param response: The L{MethodCall} response.
+        @param deferred: The deferred that was returned to the caller.
+        @param call: If not C{None}, the scheduled timeout call associated with
+            the given deferred.
+        """
+        result = response["result"]
+        if call is not None:
+            call.cancel() # This is a successful retry, cancel the timeout.
+        deferred.callback(result)
+
+    def _handle_failure(self, failure, method, args, kwargs, deferred,
+                        call=None):
+        """Called when a L{MethodCall} command fails.
+
+        If a failure is due to a connection error and if C{retry_on_reconnect}
+        is C{True}, we will try to perform the requested L{MethodCall} again
+        as soon as a new connection becomes available, giving up after the
+        specified C{timeout}, if any.
+
+        @param failure: The L{Failure} raised by the requested L{MethodCall}.
+        @param name: The method name associated with the failed L{MethodCall}.
+        @param args: The positional arguments of the failed L{MethodCall}.
+        @param kwargs: The keyword arguments of the failed L{MethodCall}.
+        @param deferred: The deferred that was returned to the caller.
+        @param call: If not C{None}, the scheduled timeout call associated with
+            the given deferred.
+        """
+        is_method_call_error = failure.type is MethodCallError
+        dont_retry = self._retry_on_reconnect == False
+
+        if is_method_call_error or dont_retry:
+            # This means either that the connection is working, and a
+            # MethodCall protocol error occured, or that we gave up
+            # trying and raised a timeout. In any case just propagate
+            # the error.
+            if deferred in self._pending_requests:
+                self._pending_requests.pop(deferred)
+            if call:
+                call.cancel()
+            deferred.errback(failure)
+            return
+
+        if self._timeout and call is None:
+            # This is the first failure for this request, let's schedule a
+            # timeout call.
+            timeout = Failure(MethodCallError("timeout"))
+            call = self._reactor.callLater(self._timeout,
+                                           self._handle_failure,
+                                           timeout, method, args,
+                                           kwargs, deferred=deferred)
+
+        self._pending_requests[deferred] = (method, args, kwargs, call)
+
+    def _handle_reconnect(self, protocol):
+        """Handles a reconnection.
+
+        @param protocol: The newly connected protocol instance.
+        """
+        self._protocol = protocol
+        if self._retry_on_reconnect:
+            self._retry()
+
+    def _retry(self):
+        """Try to perform again requests that failed."""
+
+        # We need to copy the requests list before iterating over it, because
+        # if we are actually still disconnected, callRemote will return a
+        # failed deferred and the _handle_failure errback will be executed
+        # synchronously during the loop, modifing the requests list itself.
+        requests = self._pending_requests.copy()
+        self._pending_requests.clear()
+
+        while requests:
+            deferred, (method, args, kwargs, call) = requests.popitem()
+            result = self._protocol.send_method_call(method, args, kwargs)
+            result.addCallback(self._handle_response,
+                               deferred=deferred, call=call)
+            result.addErrback(self._handle_failure, method, args, kwargs,
+                              deferred=deferred, call=call)
+
+
+class RemoteObjectCreator(object):
+    """Connect to remote objects exposed by a L{MethodCallProtocol}."""
+
+    factory = MethodCallFactory
+    remote = RemoteObject
+
+    def __init__(self, reactor, socket_path, *args, **kwargs):
+        """
+        @param reactor: A reactor able to connect to Unix sockets.
+        @param socket: The path to the socket we want to connect to.
+        @param args: Arguments to be passed to the created L{RemoteObject}.
+        @param kwargs: Keyword arguments for the created L{RemoteObject}.
+        """
+        self._socket_path = socket_path
+        self._reactor = reactor
+        self._args = args
+        self._kwargs = kwargs
+
+    def connect(self, max_retries=None):
+        """Connect to a remote object exposed by a L{MethodCallProtocol}.
+
+        This method will connect to the socket provided in the constructor
+        and return a L{Deferred} resulting in a connected L{RemoteObject}.
+
+        @param max_retries: If not C{None} give up try to connect after this
+            amount of times.
+        """
+        self._connected = Deferred()
+        self._factory = self.factory(reactor=self._reactor)
+        self._factory.maxRetries = max_retries
+        self._factory.add_notifier(self._success, self._failure)
+        self._reactor.connectUNIX(self._socket_path, self._factory)
+        return self._connected
+
+    def _success(self, result):
+        """Called when the first connection has been established"""
+
+        # We did our job, remove our own notifier and let the remote object
+        # handle reconnections.
+        self._factory.remove_notifier(self._success, self._failure)
+        self._remote = self.remote(result, *self._args, **self._kwargs)
+        self._connected.callback(self._remote)
+
+    def _failure(self, failure):
+        """Called when the first connection has failed"""
+        self._connected.errback(failure)
+
+    def disconnect(self):
+        """Disconnect the L{RemoteObject} that we have created."""
+        self._factory.stopTrying()
+        self._remote._protocol.transport.loseConnection()
+>>>>>>> MERGE-SOURCE

=== modified file 'landscape/lib/tests/test_amp.py'
--- landscape/lib/tests/test_amp.py	2010-02-12 09:50:07 +0000
+++ landscape/lib/tests/test_amp.py	2010-03-05 12:44:28 +0000
@@ -1,3 +1,4 @@
+<<<<<<< TREE
 from twisted.internet import reactor
 from twisted.internet.defer import Deferred, DeferredList
 from twisted.internet.protocol import ClientCreator
@@ -720,3 +721,718 @@
         self.assertFailure(result, MethodCallError)
         reconnected = Deferred()
         return result.addCallback(assert_failure)
+=======
+from twisted.internet import reactor
+from twisted.internet.defer import Deferred, DeferredList
+from twisted.internet.protocol import ClientCreator
+from twisted.internet.error import ConnectionDone, ConnectError
+from twisted.internet.task import Clock
+
+from landscape.lib.amp import (
+    MethodCallError, MethodCallProtocol, MethodCallFactory, RemoteObject,
+    RemoteObjectCreator)
+from landscape.tests.helpers import LandscapeTest
+from landscape.tests.mocker import KWARGS
+
+
+class WordsException(Exception):
+    """Test exception."""
+
+
+class Words(object):
+    """Test class to be used as target object of a L{MethodCallProtocol}."""
+
+    def secret(self):
+        raise RuntimeError("I'm not supposed to be called!")
+
+    def empty(self):
+        pass
+
+    def motd(self):
+        return "Words are cool"
+
+    def capitalize(self, word):
+        return word.capitalize()
+
+    def is_short(self, word):
+        return len(word) < 4
+
+    def concatenate(self, word1, word2):
+        return word1 + word2
+
+    def lower_case(self, word, index=None):
+        if index is None:
+            return word.lower()
+        else:
+            return word[:index] + word[index:].lower()
+
+    def multiply_alphabetically(self, word_times):
+        result = ""
+        for word, times in sorted(word_times.iteritems()):
+            result += word * times
+        return result
+
+    def meaning_of_life(self):
+
+        class Complex(object):
+            pass
+        return Complex()
+
+    def _check(self, word, seed, value=3):
+        if seed == "cool" and value == 4:
+            return "Guessed!"
+
+    def guess(self, word, *args, **kwargs):
+        return self._check(word, *args, **kwargs)
+
+    def translate(self, word):
+        raise WordsException("Unknown word")
+
+    def google(self, word):
+        deferred = Deferred()
+        if word == "Landscape":
+            reactor.callLater(0.01, lambda: deferred.callback("Cool!"))
+        elif word == "Easy query":
+            deferred.callback("Done!")
+        elif word == "Weird stuff":
+            reactor.callLater(0.01, lambda: deferred.errback(Exception("bad")))
+        elif word == "Censored":
+            deferred.errback(Exception("very bad"))
+        elif word == "Long query":
+            # Do nothing, the deferred won't be fired at all
+            pass
+        elif word == "Slowish query":
+            # Fire the result after a while.
+            reactor.callLater(0.05, lambda: deferred.callback("Done!"))
+        return deferred
+
+
+class WordsProtocol(MethodCallProtocol):
+
+    methods = ["empty",
+               "motd",
+               "capitalize",
+               "is_short",
+               "concatenate",
+               "lower_case",
+               "multiply_alphabetically",
+               "translate",
+               "meaning_of_life",
+               "guess",
+               "google"]
+
+    timeout = 0.1
+
+
+class WordsFactory(MethodCallFactory):
+
+    protocol = WordsProtocol
+    factor = 0.19
+
+
+class RemoteWordsCreator(RemoteObjectCreator):
+
+    factory = WordsFactory
+
+
+class MethodCallProtocolTest(LandscapeTest):
+
+    def setUp(self):
+        super(MethodCallProtocolTest, self).setUp()
+        socket = self.mktemp()
+        factory = WordsFactory(object=Words())
+        self.port = reactor.listenUNIX(socket, factory)
+
+        def set_protocol(protocol):
+            self.protocol = protocol
+
+        connector = ClientCreator(reactor, WordsProtocol)
+        connected = connector.connectUNIX(socket)
+        return connected.addCallback(set_protocol)
+
+    def tearDown(self):
+        self.protocol.transport.loseConnection()
+        self.port.stopListening()
+        super(MethodCallProtocolTest, self).tearDown()
+
+    def test_with_forbidden_method(self):
+        """
+        If a method is not included in L{MethodCallProtocol.methods} it
+        can't be called.
+        """
+        result = self.protocol.send_method_call(method="secret",
+                                                args=[],
+                                                kwargs={})
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_no_arguments(self):
+        """
+        A connected client can issue a L{MethodCall} without arguments and
+        with an empty response.
+        """
+        result = self.protocol.send_method_call(method="empty",
+                                                args=[],
+                                                kwargs={})
+        return self.assertSuccess(result, {"result": None,
+                                           "deferred": None})
+
+    def test_with_return_value(self):
+        """
+        A connected client can issue a L{MethodCall} targeted to an
+        object method with a return value.
+        """
+        result = self.protocol.send_method_call(method="motd",
+                                                args=[],
+                                                kwargs={})
+        return self.assertSuccess(result, {"result": "Words are cool",
+                                           "deferred": None})
+
+    def test_with_one_argument(self):
+        """
+        A connected AMP client can issue a L{MethodCall} with one argument and
+        a response value.
+        """
+        result = self.protocol.send_method_call(method="capitalize",
+                                                args=["john"],
+                                                kwargs={})
+        return self.assertSuccess(result, {"result": "John",
+                                           "deferred": None})
+
+    def test_with_boolean_return_value(self):
+        """
+        The return value of a L{MethodCall} argument can be a boolean.
+        """
+        result = self.protocol.send_method_call(method="is_short",
+                                                args=["hi"],
+                                                kwargs={})
+        return self.assertSuccess(result, {"result": True,
+                                           "deferred": None})
+
+    def test_with_many_arguments(self):
+        """
+        A connected client can issue a L{MethodCall} with many arguments.
+        """
+        result = self.protocol.send_method_call(method="concatenate",
+                                                args=["You ", "rock"],
+                                                kwargs={})
+        return self.assertSuccess(result, {"result": "You rock",
+                                           "deferred": None})
+
+    def test_with_default_arguments(self):
+        """
+        A connected client can issue a L{MethodCall} for methods having
+        default arguments.
+        """
+        result = self.protocol.send_method_call(method="lower_case",
+                                                args=["OHH"],
+                                                kwargs={})
+        return self.assertSuccess(result, {"result": "ohh",
+                                           "deferred": None})
+
+    def test_with_overriden_default_arguments(self):
+        """
+        A connected client can issue a L{MethodCall} with keyword arguments
+        having default values in the target object.  If a value is specified by
+        the caller it will be used in place of the default value
+        """
+        result = self.protocol.send_method_call(method="lower_case",
+                                                args=["OHH"],
+                                                kwargs={"index": 2})
+        return self.assertSuccess(result, {"result": "OHh",
+                                           "deferred": None})
+
+    def test_with_dictionary_arguments(self):
+        """
+        Method arguments passed to a L{MethodCall} can be dictionaries.
+        """
+        result = self.protocol.send_method_call(method="multiply_"
+                                                       "alphabetically",
+                                                args=[{"foo": 2, "bar": 3}],
+                                                kwargs={})
+        return self.assertSuccess(result, {"result": "barbarbarfoofoo",
+                                           "deferred": None})
+
+    def test_with_non_serializable_return_value(self):
+        """
+        If the target object method returns an object that can't be serialized,
+        the L{MethodCall} result is C{None}.
+        """
+        result = self.protocol.send_method_call(method="meaning_of_life",
+                                                args=[],
+                                                kwargs={})
+        return self.assertFailure(result, MethodCallError)
+
+    def test_translate(self):
+        """
+        If the target object method raises an exception, the remote call fails
+        with a L{MethodCallError}.
+        """
+        result = self.protocol.send_method_call(method="translate",
+                                                args=["hi"],
+                                                kwargs={})
+        return self.assertFailure(result, MethodCallError)
+
+
+class RemoteObjectTest(LandscapeTest):
+
+    def setUp(self):
+        super(RemoteObjectTest, self).setUp()
+        socket = self.mktemp()
+        server_factory = WordsFactory(object=Words())
+        self.port = reactor.listenUNIX(socket, server_factory)
+
+        def set_remote(protocol):
+            self.protocol = protocol
+            self.words = RemoteObject(protocol)
+            client_factory.stopTrying()
+
+        connected = Deferred()
+        connected.addCallback(set_remote)
+        client_factory = WordsFactory(reactor=reactor)
+        client_factory.add_notifier(connected.callback)
+        reactor.connectUNIX(socket, client_factory)
+        return connected
+
+    def tearDown(self):
+        self.protocol.transport.loseConnection()
+        self.port.stopListening()
+        super(RemoteObjectTest, self).tearDown()
+
+    def test_method_call_sender_with_forbidden_method(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s without arguments and withj
+        an empty response.
+        """
+        result = self.words.secret()
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_no_arguments(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s without arguments and withj
+        an empty response.
+        """
+        return self.assertSuccess(self.words.empty())
+
+    def test_with_return_value(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s without arguments and get
+        back the value of the commands's response.
+        """
+        result = self.words.motd()
+        return self.assertSuccess(result, "Words are cool")
+
+    def test_with_one_argument(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s with one argument and get
+        the response value.
+        """
+        result = self.words.capitalize("john")
+        return self.assertSuccess(result, "John")
+
+    def test_with_one_keyword_argument(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s with a named argument.
+        """
+        result = self.words.capitalize(word="john")
+        return self.assertSuccess(result, "John")
+
+    def test_with_boolean_return_value(self):
+        """
+        The return value of a L{MethodCall} argument can be a boolean.
+        """
+        return self.assertSuccess(self.words.is_short("hi"), True)
+
+    def test_with_many_arguments(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s with more than one argument.
+        """
+        result = self.words.concatenate("You ", "rock")
+        return self.assertSuccess(result, "You rock")
+
+    def test_with_many_keyword_arguments(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s with several
+        named arguments.
+        """
+        result = self.words.concatenate(word2="rock", word1="You ")
+        return self.assertSuccess(result, "You rock")
+
+    def test_with_default_arguments(self):
+        """
+        A L{RemoteObject} can send a L{MethodCall} having an argument with
+        a default value.
+        """
+        result = self.words.lower_case("OHH")
+        return self.assertSuccess(result, "ohh")
+
+    def test_with_overriden_default_arguments(self):
+        """
+        A L{RemoteObject} can send L{MethodCall}s overriding the default
+        value of an argument.
+        """
+        result = self.words.lower_case("OHH", 2)
+        return self.assertSuccess(result, "OHh")
+
+    def test_with_dictionary_arguments(self):
+        """
+        A L{RemoteObject} can send a L{MethodCall}s for methods requiring
+        a dictionary arguments.
+        """
+        result = self.words.multiply_alphabetically({"foo": 2, "bar": 3})
+        return self.assertSuccess(result, "barbarbarfoofoo")
+
+    def test_with_generic_args_and_kwargs(self):
+        """
+        A L{RemoteObject} behaves well with L{MethodCall}s for methods
+        having generic C{*args} and C{**kwargs} arguments.
+        """
+        result = self.words.guess("word", "cool", value=4)
+        return self.assertSuccess(result, "Guessed!")
+
+    def test_with_success_full_deferred(self):
+        """
+        If the target object method returns a L{Deferred}, it is handled
+        transparently.
+        """
+        result = self.words.google("Landscape")
+        return self.assertSuccess(result, "Cool!")
+
+    def test_with_failing_deferred(self):
+        """
+        If the target object method returns a failing L{Deferred}, a
+        L{MethodCallError} is raised.
+        """
+        result = self.words.google("Weird stuff")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_already_callback_deferred(self):
+        """
+        The target object method can return an already fired L{Deferred}.
+        """
+        result = self.words.google("Easy query")
+        return self.assertSuccess(result, "Done!")
+
+    def test_with_already_errback_deferred(self):
+        """
+        If the target object method can return an already failed L{Deferred}.
+        """
+        result = self.words.google("Censored")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_deferred_timeout(self):
+        """
+        If the peer protocol doesn't send a response for a deferred within
+        the given timeout, the method call fails.
+        """
+        result = self.words.google("Long query")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_late_response(self):
+        """
+        If the peer protocol sends a late response for a request that has
+        already timeout, that response is ignored.
+        """
+        self.protocol.timeout = 0.01
+        result = self.words.google("Slowish query")
+        self.assertFailure(result, MethodCallError)
+
+        def assert_late_response_is_handled(ignored):
+            deferred = Deferred()
+            # We wait a bit to be sure that the late response gets delivered
+            reactor.callLater(0.1, lambda: deferred.callback(None))
+            return deferred
+
+        return result.addCallback(assert_late_response_is_handled)
+
+
+class MethodCallFactoryTest(LandscapeTest):
+
+    def setUp(self):
+        super(MethodCallFactoryTest, self).setUp()
+        self.clock = Clock()
+        self.factory = WordsFactory(reactor=self.clock)
+
+    def test_add_notifier(self):
+        """
+        The L{MethodCallClientFactory.add_notifier} method can be used to
+        add a callback function to be called when a connection is made and
+        a new protocol instance built.
+        """
+        protocol = self.factory.protocol()
+        self.factory.protocol = lambda: protocol
+        callback = self.mocker.mock()
+        callback(protocol)
+        self.mocker.replay()
+        self.factory.add_notifier(callback)
+        self.factory.buildProtocol(None)
+        self.clock.advance(0)
+
+    def test_remove_notifier(self):
+        """
+        The L{MethodCallClientFactory.remove_notifier} method can be used to
+        remove a previously added notifier callback.
+        """
+        callback = lambda protocol: 1 / 0
+        self.factory.add_notifier(callback)
+        self.factory.remove_notifier(callback)
+        self.factory.buildProtocol(None)
+        self.clock.advance(0)
+
+    def test_client_connection_failed(self):
+        """
+        The L{MethodCallClientFactory} keeps trying to connect if maxRetries
+        is not reached.
+        """
+        # This is sub-optimal but the ReconnectingFactory in Hardy's Twisted
+        # doesn't support task.Clock
+        self.factory.retry = self.mocker.mock()
+        self.factory.retry(KWARGS)
+        self.mocker.replay()
+        self.assertEquals(self.factory.retries, 0)
+        self.factory.clientConnectionFailed(None, None)
+
+    def test_client_connection_failed_with_max_retries_reached(self):
+        """
+        The L{MethodCallClientFactory} stops trying to connect if maxRetries
+        is reached.
+        """
+        callback = lambda protocol: 1 / 0
+        errback = self.mocker.mock()
+        errback("failure")
+        self.mocker.replay()
+
+        self.factory.add_notifier(callback, errback)
+        self.factory.maxRetries = 1
+        self.factory.retries = self.factory.maxRetries
+        self.factory.clientConnectionFailed(object(), "failure")
+        self.clock.advance(0)
+
+
+class RemoteObjectCreatorTest(LandscapeTest):
+
+    def setUp(self):
+        super(RemoteObjectCreatorTest, self).setUp()
+        self.socket = self.mktemp()
+        self.server_factory = WordsFactory(object=Words())
+        self.port = reactor.listenUNIX(self.socket, self.server_factory)
+        self.connector = RemoteWordsCreator(reactor, self.socket,
+                                            retry_on_reconnect=True,
+                                            timeout=0.7)
+
+        def set_remote(words):
+            self.words = words
+
+        connected = self.connector.connect()
+        return connected.addCallback(set_remote)
+
+    def tearDown(self):
+        self.connector.disconnect()
+        self.port.stopListening()
+        super(RemoteObjectCreatorTest, self).tearDown()
+
+    def test_connect(self):
+        """
+        The L{RemoteObject} resulting form the deferred returned by
+        L{RemoteObjectCreator.connect} is properly connected to the
+        remote peer.
+        """
+        return self.assertSuccess(self.words.empty())
+
+    def test_connect_with_max_retries(self):
+        """
+        If C{max_retries} is passed to the L{RemoteObjectCreator} method,
+        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_reconnect(self):
+        """
+        If the connection is lost, the L{RemoteObject} created by the creator
+        will transparently handle the reconnection.
+        """
+        self.words._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._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._protocol.transport.loseConnection()
+        self.port.stopListening()
+
+        def restart_listening():
+            self.port = reactor.listenUNIX(self.socket, self.server_factory)
+
+        def assert_failure(error):
+            self.assertEquals(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
+        connection is ready. If for whatever reason the connection drops
+        again very quickly, the C{_retry} method will behave as expected.
+        """
+        self.words._protocol.transport.loseConnection()
+        self.port.stopListening()
+
+        def handle_reconnect(protocol):
+            # In this precise moment we have a newly connected protocol
+            self.words._protocol = protocol
+
+            # Pretend that the connection is lost again very quickly
+            protocol.transport.loseConnection()
+            self.port.stopListening()
+
+            # Force RemoteObject._retry to run using a disconnected protocol
+            reactor.callLater(0, self.words._retry)
+
+            # Restore the real handler and start listening again very soon
+            self.connector._factory.remove_notifier(handle_reconnect)
+            self.connector._factory.add_notifier(self.words._handle_reconnect)
+            reactor.callLater(0.2, restart_listening)
+
+        def restart_listening():
+            self.port = reactor.listenUNIX(self.socket, self.server_factory)
+
+        def assert_failure(error):
+            self.assertEquals(str(error), "Forbidden method 'secret'")
+
+        # Use our own reconnect handler
+        self.connector._factory.remove_notifier(self.words._handle_reconnect)
+        self.connector._factory.add_notifier(handle_reconnect)
+
+        reactor.callLater(0.2, restart_listening)
+        result = self.words.secret()
+        self.assertFailure(result, MethodCallError)
+        return result.addCallback(assert_failure)
+
+    def test_retry_with_many_method_calls(self):
+        """
+        If several L{MethodCall} requests were issued while disconnected, they
+        will be all eventually completed when the connection gets established
+        again.
+        """
+        self.words._protocol.transport.loseConnection()
+        self.port.stopListening()
+
+        def restart_listening():
+            self.port = reactor.listenUNIX(self.socket, self.server_factory)
+
+        def assert_guess(response):
+            self.assertEquals(response, "Guessed!")
+
+        def assert_secret(failure):
+            self.assertEquals(str(failure.value), "Forbidden method 'secret'")
+
+        def assert_motd(response):
+            self.assertEquals(response, "Words are cool")
+
+        reactor.callLater(0.1, restart_listening)
+
+        results = [self.words.guess("word", "cool", value=4),
+                   self.words.secret(),
+                   self.words.motd()]
+        results[0].addCallback(assert_guess)
+        results[1].addErrback(assert_secret)
+        results[2].addCallback(assert_motd)
+        return DeferredList(results)
+
+    def test_retry_without_retry_on_reconnect(self):
+        """
+        If C{retry_on_reconnect} is C{False}, the L{RemoteObject} object won't
+        retry to perform requests which failed because the connection was
+        lost, however requests made after a reconnection will still succeed.
+        """
+        self.words._protocol.transport.loseConnection()
+        self.port.stopListening()
+
+        def restart_listening():
+            self.port = reactor.listenUNIX(self.socket, self.server_factory)
+            reactor.callLater(0.3, assert_reconnected)
+
+        def assert_reconnected():
+            result = self.words.empty()
+            result.addCallback(lambda x: reconnected.callback(None))
+            return result
+
+        reactor.callLater(0.1, restart_listening)
+        self.words._retry_on_reconnect = False
+        result = self.words.empty()
+        self.assertFailure(result, ConnectionDone)
+        reconnected = Deferred()
+        return result.addCallback(lambda x: reconnected)
+
+    def test_retry_with_timeout(self):
+        """
+        If a C{timeout} is set, the L{RemoteObject} object will errback failed
+        L{MethodCall}s after that amount of seconds, without retrying them when
+        the connection established again.
+        """
+        self.words._protocol.transport.loseConnection()
+        self.port.stopListening()
+
+        def restart_listening():
+            self.port = reactor.listenUNIX(self.socket, self.server_factory)
+            reactor.callLater(0.1, reconnected.callback, None)
+
+        def assert_failure(error):
+            self.assertEquals(str(error), "timeout")
+            return reconnected
+
+        reactor.callLater(0.9, restart_listening)
+        result = self.words.empty()
+        self.assertFailure(result, MethodCallError)
+        reconnected = Deferred()
+        return result.addCallback(assert_failure)
+>>>>>>> MERGE-SOURCE

=== modified file 'landscape/package/reporter.py'
=== modified file 'landscape/package/tests/test_reporter.py'
=== modified file 'landscape/reactor.py'
--- landscape/reactor.py	2009-11-02 13:49:26 +0000
+++ landscape/reactor.py	2010-03-05 12:44:28 +0000
@@ -158,6 +158,13 @@
         self.cancel_call(self._run_threaded_callbacks_id)
 
 
+class UnixReactorMixin(object):
+
+    def listen_unix(self, *args, **kwargs):
+        """Start listen on a Unix socket."""
+        return self._reactor.listenUNIX(*args, **kwargs)
+
+
 class ReactorID(object):
 
     def __init__(self, timeout):
@@ -215,7 +222,7 @@
 
 
 class FakeReactor(EventHandlingReactorMixin,
-                  ThreadedCallsReactorMixin):
+                  ThreadedCallsReactorMixin, UnixReactorMixin):
     """
     @ivar udp_transports: dict of {port: (protocol, transport)}
     @ivar hosts: Dict of {hostname: ip}. Users should populate this
@@ -228,6 +235,11 @@
         self.udp_transports = {}
         self.hosts = {}
 
+        # We need a reference to the Twisted reactor as well to
+        # let Landscape services listen to Unix sockets
+        from twisted.internet import reactor
+        self._reactor = reactor
+
     def time(self):
         return float(self._current_time)
 
@@ -331,9 +343,8 @@
             return succeed(hostname)
 
 
-
 class TwistedReactor(EventHandlingReactorMixin,
-                     ThreadedCallsReactorMixin):
+                     ThreadedCallsReactorMixin, UnixReactorMixin):
     """Wrap and add functionalities to the Twisted C{reactor}."""
 
     def __init__(self):

=== added file 'landscape/tests/test_amp.py'
--- landscape/tests/test_amp.py	1970-01-01 00:00:00 +0000
+++ landscape/tests/test_amp.py	2010-03-05 12:44:28 +0000
@@ -0,0 +1,126 @@
+import os
+
+from twisted.internet.defer import Deferred
+from twisted.internet.error import ConnectError
+
+from landscape.tests.helpers import LandscapeTest
+from landscape.reactor import FakeReactor
+from landscape.deployment import Configuration
+from landscape.amp import (
+    LandscapeComponentFactory, RemoteLandscapeComponentCreator)
+
+
+class TestComponent(object):
+
+    name = "test"
+
+
+class TestComponentFactory(LandscapeComponentFactory):
+
+    maxRetries = 0
+    initialDelay = 0.01
+
+
+class RemoteTestComponentCreator(RemoteLandscapeComponentCreator):
+
+    factory = TestComponentFactory
+    component = TestComponent
+
+
+class RemoteLandscapeComponentTest(LandscapeTest):
+
+    def setUp(self):
+        super(RemoteLandscapeComponentTest, self).setUp()
+        reactor = FakeReactor()
+        config = Configuration()
+        config.data_path = self.makeDir()
+        socket = os.path.join(config.data_path, "test.sock")
+        self.component = TestComponent()
+        factory = LandscapeComponentFactory(object=self.component)
+        self.port = reactor.listen_unix(socket, factory)
+
+
+        self.connector = RemoteTestComponentCreator(reactor, config)
+        connected = self.connector.connect()
+        connected.addCallback(lambda remote: setattr(self, "remote", remote))
+        return connected
+
+    def tearDown(self):
+        self.connector.disconnect()
+        self.port.stopListening()
+        super(RemoteLandscapeComponentTest, self).tearDown()
+
+    def test_ping(self):
+        """
+        The L{LandscapeComponentProtocol} exposes the C{ping} method of a
+        remote Landscape component.
+        """
+        self.component.ping = self.mocker.mock()
+        self.expect(self.component.ping()).result(True)
+        self.mocker.replay()
+        result = self.remote.ping()
+        return self.assertSuccess(result, True)
+
+    def test_exit(self):
+        """
+        The L{LandscapeComponentProtocol} exposes the C{exit} method of a
+        remote Landscape component.
+        """
+        self.component.exit = self.mocker.mock()
+        self.component.exit()
+        self.mocker.replay()
+        result = self.remote.exit()
+        return self.assertSuccess(result)
+
+
+class RemoteLandscapeComponentCreatorTest(LandscapeTest):
+
+    def setUp(self):
+        super(RemoteLandscapeComponentCreatorTest, self).setUp()
+        self.reactor = FakeReactor()
+        self.config = Configuration()
+        self.config.data_path = self.makeDir()
+        self.connector = RemoteTestComponentCreator(self.reactor, self.config)
+
+    def test_connect_logs_errors(self):
+        """
+        Connection errors are logged.
+        """
+        self.log_helper.ignore_errors("Error while connecting to test")
+
+        def assert_log(ignored):
+            self.assertIn("Error while connecting to test",
+                          self.logfile.getvalue())
+
+        result = self.connector.connect(max_retries=0)
+        self.assertFailure(result, ConnectError)
+        return result.addCallback(assert_log)
+
+    def test_reconnect_fires_event(self):
+        """
+        An event is fired whenever the connection is established again after
+        it has been lost.
+        """
+        socket = os.path.join(self.config.data_path, "test.sock")
+        factory = LandscapeComponentFactory()
+        ports = []
+        ports.append(self.reactor.listen_unix(socket, factory))
+
+        def listen_again():
+            ports.append(self.reactor.listen_unix(socket, factory))
+
+        def connected(remote):
+            remote._protocol.transport.loseConnection()
+            ports[0].stopListening()
+            self.reactor._reactor.callLater(0.01, listen_again)
+
+        def reconnected():
+            self.connector.disconnect()
+            ports[1].stopListening()
+            deferred.callback(None)
+
+        deferred = Deferred()
+        self.reactor.call_on("test-reconnect", reconnected)
+        result = self.connector.connect()
+        result.addCallback(connected)
+        return deferred