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

[Free Ekanayaka]

This is almost a trivial change, in preparation of some follow-up cleanup work. It's mostly about cosmetics and little details in landscape.lib.amp:

- Change the module docstring to reflect how this machinery's API will change (note
  that it hasn't been implemented yet tho). The new style is a bit more Twisted-y,
  and doesn't force the use to use unix sockets.

- MethodCallSender.send_method_call now returns directly the result of the remote
  method invokation, rather then MethodCall.response low-level dict. Tests have been
  updated accordingly.

- MethodCallSender.protocol is now public, and can be set by calling code.

This is almost a trivial change, in preparation of some follow-up cleanup work. It's mostly about cosmetics and little details in landscape.lib.amp:

- Change the module docstring to reflect how this machinery's API will change (note
  that it hasn't been implemented yet tho). The new style is a bit more Twisted-y,
  and doesn't force the use to use unix sockets.

- MethodCallSender.send_method_call now returns directly the result of the remote
  method invokation, rather then MethodCall.response low-level dict. Tests have been
  updated accordingly.

- MethodCallSender.protocol is now public, and can be set by calling code.

[Alberto Donato (ack)]

Nice! +1

review: Approve

[Geoff Teale (tealeg)]

Clean as a whistle +1

review: Approve

=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2013-04-15 11:00:01 +0000
+++ landscape/lib/amp.py	2013-04-19 11:37:27 +0000
@@ -18,29 +18,21 @@
 Process A can "publish" the greeter object by defining which methods are
 exposed remotely and opening a Unix socket for incoming connections::
 
-    class GreeterProtocol(MethodCallProtocol):
-
-        methods = ["hello"]
-
-    factory = MethodCallFactory(object=greeter)
-    factory.protocol = GreeterProtocol
+    factory = MethodCallServerFactory(greeter, ["hello"])
     reactor.listenUNIX("/some/socket/path", factory)
 
 Then a second Python process "B" can connect to that socket and build a
 "remote" greeter object, i.e. a proxy that forwards method calls to the
 real greeter object living in process A::
 
-    connector = ClientCreator(reactor, GreeterProtocol)
-    connected = connector.connectUNIX("/some/socket/path")
-
-    def got_connection(protocol):
-
-        remote_greeter = RemoteObject(protocol)
-
+    factory = MethodCallClientFactory()
+    reactor.connectUNIX("/some/socket/path", factory)
+
+    def got_remote(remote_greeter):
         deferred = remote_greeter.hello("Ted")
         deferred.addCallback(lambda result: ... # result == "hi Ted!")
 
-    connected.addCallback(got_connection)
+    factory.getRemoteObject().addCallback(got_remote)
 
 Note that when invoking a method via the remote proxy, the parameters
 are required to be serializable with bpickle, so that they can be sent
@@ -111,14 +103,6 @@
     errors = {MethodCallError: "METHOD_CALL_ERROR"}
 
 
-class MethodCallServerProtocol(AMP):
-    """XXX Placeholder"""
-
-
-class MethodCallClientProtocol(AMP):
-    """XXX Placeholder"""
-
-
 class MethodCallReceiver(CommandLocator):
     """Expose methods of a local object over AMP.
 
@@ -196,22 +180,20 @@
 
 
 class MethodCallSender(object):
-    """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}.
+    """Call methods on a remote object over L{AMP} and return the result.
+
+    @param protocol: A connected C{AMP} protocol.
+    @param clock: An object implementing the C{IReactorTime} interface.
+
+    @ivar timeout: A timeout for remote method class, see L{send_method_call}.
     """
     timeout = 60
-    chunk_size = MAX_VALUE_LENGTH
+
+    _chunk_size = MAX_VALUE_LENGTH
 
     def __init__(self, protocol, clock):
-        self._protocol = protocol
-        self._clock = clock
-        self._pending_responses = []
+        self.protocol = protocol
+        self.clock = clock
         self._sequence = 0
 
     def _create_sequence(self):
@@ -223,8 +205,8 @@
         """Send an L{AMP} command that will errback in case of a timeout.
 
         @return: A deferred resulting in the command's response (or failure) if
-            the peer responds within L{MethodClientProtocol.timeout} seconds,
-            or that errbacks with a L{MethodCallError} otherwise.
+            the peer responds within C{self.timeout} seconds, or that errbacks
+            with a L{MethodCallError} otherwise.
         """
         deferred = Deferred()
 
@@ -240,25 +222,32 @@
             # The peer didn't respond on time, raise an error.
             deferred.errback(MethodCallError("timeout"))
 
-        call = self._clock.callLater(self.timeout, handle_timeout)
+        call = self.clock.callLater(self.timeout, handle_timeout)
 
-        result = self._protocol.callRemote(command, **kwargs)
+        result = self.protocol.callRemote(command, **kwargs)
         result.addBoth(handle_response)
         return deferred
 
     def send_method_call(self, method, args=[], kwargs={}):
         """Send a L{MethodCall} command with the given arguments.
 
+        If a response from the server is not received within C{self.timeout}
+        seconds, the returned deferred will errback with a L{MethodCallError}.
+
         @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.
+
+        @return: A C{Deferred} firing with the return value of the method
+            invoked on the remote object. If the remote method itself returns
+            a deferred, we fire with the callback value of such deferred.
         """
         arguments = dumps((args, kwargs))
         sequence = self._create_sequence()
 
         # Split the given arguments in one or more chunks
-        chunks = [arguments[i:i + self.chunk_size]
-                  for i in xrange(0, len(arguments), self.chunk_size)]
+        chunks = [arguments[i:i + self._chunk_size]
+                  for i in xrange(0, len(arguments), self._chunk_size)]
 
         result = Deferred()
         if len(chunks) > 1:
@@ -266,7 +255,7 @@
             for chunk in chunks[:-1]:
 
                 def create_send_chunk(sequence, chunk):
-                    send_chunk = lambda x: self._protocol.callRemote(
+                    send_chunk = lambda x: self.protocol.callRemote(
                         MethodCallChunk, sequence=sequence, chunk=chunk)
                     return send_chunk
 
@@ -278,10 +267,19 @@
                 MethodCall, sequence=sequence, method=method, arguments=chunk)
 
         result.addCallback(send_last_chunk)
+        result.addCallback(lambda response: response["result"])
         result.callback(None)
         return result
 
 
+class MethodCallServerProtocol(AMP):
+    """XXX Placeholder"""
+
+
+class MethodCallClientProtocol(AMP):
+    """XXX Placeholder"""
+
+
 class MethodCallProtocol(MethodCallServerProtocol, MethodCallClientProtocol):
     """Can be used both for sending and receiving L{MethodCall}s."""
 
@@ -410,22 +408,21 @@
                                                    args=args,
                                                    kwargs=kwargs)
             deferred = Deferred()
-            result.addCallback(self._handle_response, deferred)
+            result.addCallback(self._handle_result, 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.
+    def _handle_result(self, result, deferred, call=None):
+        """Handles a successful C{send_method_call} result.
 
         @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)
@@ -478,7 +475,7 @@
 
         @param protocol: The newly connected protocol instance.
         """
-        self._sender._protocol = protocol
+        self._sender.protocol = protocol
         if self._retry_on_reconnect:
             self._retry()
 
@@ -495,7 +492,7 @@
         while requests:
             deferred, (method, args, kwargs, call) = requests.popitem()
             result = self._sender.send_method_call(method, args, kwargs)
-            result.addCallback(self._handle_response,
+            result.addCallback(self._handle_result,
                                deferred=deferred, call=call)
             result.addErrback(self._handle_failure, method, args, kwargs,
                               deferred=deferred, call=call)
@@ -564,6 +561,6 @@
         if self._factory:
             self._factory.stopTrying()
         if self._remote:
-            if self._remote._sender._protocol.transport:
-                self._remote._sender._protocol.transport.loseConnection()
+            if self._remote._sender.protocol.transport:
+                self._remote._sender.protocol.transport.loseConnection()
             self._remote = None
 
=== modified file 'landscape/lib/tests/test_amp.py'
--- landscape/lib/tests/test_amp.py	2013-04-15 11:00:01 +0000
+++ landscape/lib/tests/test_amp.py	2013-04-19 11:37:27 +0000
@@ -160,6 +160,19 @@
     factory = WordsFactory
 
 
+METHODS = ["empty",
+           "motd",
+           "capitalize",
+           "is_short",
+           "concatenate",
+           "lower_case",
+           "multiply_alphabetically",
+           "translate",
+           "meaning_of_life",
+           "guess",
+           "google"]
+
+
 class MethodCallTest(LandscapeTest):
 
     def setUp(self):
@@ -190,7 +203,7 @@
                                                 args=[],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": None}, self.successResultOf(deferred))
+        self.assertIs(None, self.successResultOf(deferred))
 
     def test_with_return_value(self):
         """
@@ -201,8 +214,7 @@
                                                 args=[],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": "Words are cool"},
-                         self.successResultOf(deferred))
+        self.assertEqual("Words are cool", self.successResultOf(deferred))
 
     def test_with_one_argument(self):
         """
@@ -213,7 +225,7 @@
                                                 args=["john"],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": "John"}, self.successResultOf(deferred))
+        self.assertEqual("John", self.successResultOf(deferred))
 
     def test_with_boolean_return_value(self):
         """
@@ -223,7 +235,7 @@
                                                 args=["hi"],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": True}, self.successResultOf(deferred))
+        self.assertTrue(self.successResultOf(deferred))
 
     def test_with_many_arguments(self):
         """
@@ -233,7 +245,7 @@
                                                 args=["We ", "rock"],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": "We rock"}, self.successResultOf(deferred))
+        self.assertEqual("We rock", self.successResultOf(deferred))
 
     def test_with_default_arguments(self):
         """
@@ -244,7 +256,7 @@
                                                 args=["OHH"],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": "ohh"}, self.successResultOf(deferred))
+        self.assertEqual("ohh", self.successResultOf(deferred))
 
     def test_with_overriden_default_arguments(self):
         """
@@ -256,7 +268,7 @@
                                                 args=["OHH"],
                                                 kwargs={"index": 2})
         self.connection.flush()
-        self.assertEqual({"result": "OHh"}, self.successResultOf(deferred))
+        self.assertEqual("OHh", self.successResultOf(deferred))
 
     def test_with_dictionary_arguments(self):
         """
@@ -267,8 +279,7 @@
                                                 args=[{"foo": 2, "bar": 3}],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": "barbarbarfoofoo"},
-                         self.successResultOf(deferred))
+        self.assertEqual("barbarbarfoofoo", self.successResultOf(deferred))
 
     def test_with_non_serializable_return_value(self):
         """
@@ -290,7 +301,7 @@
                                                 args=["!" * 65535],
                                                 kwargs={})
         self.connection.flush()
-        self.assertEqual({"result": False}, self.successResultOf(deferred))
+        self.assertFalse(self.successResultOf(deferred))
 
     def test_with_long_argument_multiple_calls(self):
         """
@@ -305,8 +316,8 @@
                                                  kwargs={})
 
         self.connection.flush()
-        self.assertEqual({"result": False}, self.successResultOf(deferred1))
-        self.assertEqual({"result": False}, self.successResultOf(deferred2))
+        self.assertFalse(self.successResultOf(deferred1))
+        self.assertFalse(self.successResultOf(deferred2))
 
     def test_translate(self):
         """
@@ -339,14 +350,14 @@
             return deferred
 
         sender.send_method_call = synchronous_send_method_call
-        self.words = RemoteObject(sender)
+        self.remote = RemoteObject(sender)
 
     def test_method_call_sender_with_forbidden_method(self):
         """
         A L{RemoteObject} can send L{MethodCall}s without arguments and withj
         an empty response.
         """
-        deferred = self.words.secret()
+        deferred = self.remote.secret()
         self.failureResultOf(deferred).trap(MethodCallError)
 
     def test_with_no_arguments(self):
@@ -354,7 +365,7 @@
         A L{RemoteObject} can send L{MethodCall}s without arguments and withj
         an empty response.
         """
-        deferred = self.words.empty()
+        deferred = self.remote.empty()
         self.assertIs(None, self.successResultOf(deferred))
 
     def test_with_return_value(self):
@@ -362,7 +373,7 @@
         A L{RemoteObject} can send L{MethodCall}s without arguments and get
         back the value of the commands's response.
         """
-        deferred = self.words.motd()
+        deferred = self.remote.motd()
         self.assertEqual("Words are cool", self.successResultOf(deferred))
 
     def test_with_one_argument(self):
@@ -370,27 +381,27 @@
         A L{RemoteObject} can send L{MethodCall}s with one argument and get
         the response value.
         """
-        deferred = self.words.capitalize("john")
+        deferred = self.remote.capitalize("john")
         self.assertEqual("John", self.successResultOf(deferred))
 
     def test_with_one_keyword_argument(self):
         """
         A L{RemoteObject} can send L{MethodCall}s with a named argument.
         """
-        deferred = self.words.capitalize(word="john")
+        deferred = self.remote.capitalize(word="john")
         self.assertEqual("John", self.successResultOf(deferred))
 
     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)
+        return self.assertSuccess(self.remote.is_short("hi"), True)
 
     def test_with_many_arguments(self):
         """
         A L{RemoteObject} can send L{MethodCall}s with more than one argument.
         """
-        deferred = self.words.concatenate("You ", "rock")
+        deferred = self.remote.concatenate("You ", "rock")
         self.assertEqual("You rock", self.successResultOf(deferred))
 
     def test_with_many_keyword_arguments(self):
@@ -398,7 +409,7 @@
         A L{RemoteObject} can send L{MethodCall}s with several
         named arguments.
         """
-        deferred = self.words.concatenate(word2="rock", word1="You ")
+        deferred = self.remote.concatenate(word2="rock", word1="You ")
         self.assertEqual("You rock", self.successResultOf(deferred))
 
     def test_with_default_arguments(self):
@@ -406,7 +417,7 @@
         A L{RemoteObject} can send a L{MethodCall} having an argument with
         a default value.
         """
-        deferred = self.words.lower_case("OHH")
+        deferred = self.remote.lower_case("OHH")
         self.assertEqual("ohh", self.successResultOf(deferred))
 
     def test_with_overriden_default_arguments(self):
@@ -414,7 +425,7 @@
         A L{RemoteObject} can send L{MethodCall}s overriding the default
         value of an argument.
         """
-        deferred = self.words.lower_case("OHH", 2)
+        deferred = self.remote.lower_case("OHH", 2)
         self.assertEqual("OHh", self.successResultOf(deferred))
 
     def test_with_dictionary_arguments(self):
@@ -422,7 +433,7 @@
         A L{RemoteObject} can send a L{MethodCall}s for methods requiring
         a dictionary arguments.
         """
-        deferred = self.words.multiply_alphabetically({"foo": 2, "bar": 3})
+        deferred = self.remote.multiply_alphabetically({"foo": 2, "bar": 3})
         self.assertEqual("barbarbarfoofoo", self.successResultOf(deferred))
 
     def test_with_generic_args_and_kwargs(self):
@@ -430,7 +441,7 @@
         A L{RemoteObject} behaves well with L{MethodCall}s for methods
         having generic C{*args} and C{**kwargs} arguments.
         """
-        deferred = self.words.guess("word", "cool", value=4)
+        deferred = self.remote.guess("word", "cool", value=4)
         self.assertEqual("Guessed!", self.successResultOf(deferred))
 
     def test_with_successful_deferred(self):
@@ -439,7 +450,7 @@
         transparently.
         """
         result = []
-        deferred = self.words.google("Landscape")
+        deferred = self.remote.google("Landscape")
         deferred.addCallback(result.append)
 
         # At this point the receiver is waiting for method to complete, so
@@ -458,7 +469,7 @@
         L{MethodCallError} is raised.
         """
         result = []
-        deferred = self.words.google("Weird stuff")
+        deferred = self.remote.google("Weird stuff")
         deferred.addErrback(result.append)
 
         # At this point the receiver is waiting for method to complete, so
@@ -476,14 +487,14 @@
         """
         The target object method can return an already fired L{Deferred}.
         """
-        deferred = self.words.google("Easy query")
+        deferred = self.remote.google("Easy query")
         self.assertEqual("Done!", self.successResultOf(deferred))
 
     def test_with_already_errback_deferred(self):
         """
         If the target object method can return an already failed L{Deferred}.
         """
-        deferred = self.words.google("Censored")
+        deferred = self.remote.google("Censored")
         self.failureResultOf(deferred).trap(MethodCallError)
 
     def test_with_deferred_timeout(self):
@@ -492,7 +503,7 @@
         the given timeout, the method call fails.
         """
         result = []
-        deferred = self.words.google("Long query")
+        deferred = self.remote.google("Long query")
         deferred.addErrback(result.append)
 
         self.clock.advance(60.0)
@@ -506,7 +517,7 @@
         already timeout, that response is ignored.
         """
         result = []
-        deferred = self.words.google("Slowish query")
+        deferred = self.remote.google("Slowish query")
         deferred.addErrback(result.append)
 
         self.clock.advance(120.0)
@@ -520,7 +531,7 @@
     def setUp(self):
         super(MethodCallFactoryTest, self).setUp()
         self.clock = Clock()
-        self.factory = WordsFactory(reactor=self.clock)
+        self.factory = MethodCallFactory(reactor=self.clock)
 
     def test_max_delay(self):
         """
@@ -669,7 +680,7 @@
         If the connection is lost, the L{RemoteObject} created by the creator
         will transparently handle the reconnection.
         """
-        self.words._sender._protocol.transport.loseConnection()
+        self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
 
         def restart_listening():
@@ -698,7 +709,7 @@
         will transparently retry to perform the L{MethodCall} requests that
         failed due to the broken connection.
         """
-        self.words._sender._protocol.transport.loseConnection()
+        self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
 
         def restart_listening():
@@ -713,7 +724,7 @@
         the L{RemoteObject} will properly propagate the error to the original
         caller.
         """
-        self.words._sender._protocol.transport.loseConnection()
+        self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
 
         def restart_listening():
@@ -733,12 +744,12 @@
         connection is ready. If for whatever reason the connection drops
         again very quickly, the C{_retry} method will behave as expected.
         """
-        self.words._sender._protocol.transport.loseConnection()
+        self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
 
         def handle_reconnect(protocol):
             # In this precise moment we have a newly connected protocol
-            self.words._sender._protocol = protocol
+            self.words._sender.protocol = protocol
 
             # Pretend that the connection is lost again very quickly
             protocol.transport.loseConnection()
@@ -773,7 +784,7 @@
         will be all eventually completed when the connection gets established
         again.
         """
-        self.words._sender._protocol.transport.loseConnection()
+        self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
 
         def restart_listening():
@@ -804,7 +815,7 @@
         retry to perform requests which failed because the connection was
         lost, however requests made after a reconnection will still succeed.
         """
-        self.words._sender._protocol.transport.loseConnection()
+        self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
 
         def restart_listening():
@@ -829,7 +840,7 @@
         L{MethodCall}s after that amount of seconds, without retrying them when
         the connection established again.
         """
-        self.words._sender._protocol.transport.loseConnection()
+        self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
 
         def restart_listening():
 
=== modified file 'landscape/tests/test_amp.py'
--- landscape/tests/test_amp.py	2013-04-15 09:03:43 +0000
+++ landscape/tests/test_amp.py	2013-04-19 11:37:27 +0000
@@ -118,7 +118,7 @@
             ports.append(self.reactor.listen_unix(socket, factory))
 
         def connected(remote):
-            remote._sender._protocol.transport.loseConnection()
+            remote._sender.protocol.transport.loseConnection()
             ports[0].stopListening()
             self.reactor._reactor.callLater(0.01, listen_again)
 
 
=== modified file 'landscape/tests/test_configuration.py'
--- landscape/tests/test_configuration.py	2013-03-27 22:58:20 +0000
+++ landscape/tests/test_configuration.py	2013-04-19 11:37:27 +0000
@@ -6,6 +6,7 @@
 
 from twisted.internet.defer import succeed, fail
 
+from landscape.lib.amp import MethodCallSender
 from landscape.reactor import TwistedReactor
 from landscape.lib.fetch import HTTPCodeError, PyCurlError
 from landscape.configuration import (
@@ -19,7 +20,7 @@
 from landscape.tests.helpers import (
     LandscapeTest, BrokerServiceHelper, EnvironSaverHelper)
 from landscape.tests.mocker import ARGS, ANY, MATCH, CONTAINS, expect
-from landscape.broker.amp import RemoteBroker, BrokerClientProtocol
+from landscape.broker.amp import RemoteBroker
 
 
 class LandscapeConfigurationTest(LandscapeTest):
@@ -1784,7 +1785,7 @@
         reactor_mock = self.mocker.patch(TwistedReactor)
         remote_mock = self.mocker.patch(RemoteBroker)
 
-        protocol_mock = self.mocker.patch(BrokerClientProtocol)
+        protocol_mock = self.mocker.patch(MethodCallSender)
         protocol_mock.timeout
         self.mocker.result(0.1)
         self.mocker.count(0, None)