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

[Free Ekanayaka]

Another episode of the AMP-related cleanup:

- Rename landscape.amp.RemoteComponentConnector to simply ComponentConnector (the fact that it's remote is implied by the fact that you connect to it, and the new name complements the existing ComponentPublisher nicely)

- Drop the landscape.amp.ComponentProtocolClientFactory sub-class and use MethodCallClientFactory directly, tweaking the needed instance attributes with explicit assignements (this is a sort of composition-over-inheritance change)

- Drop landscape.reactor.UnixReactorMixin. The landscape.reactor.FakeReactor class now really fakes the Unix socket transport, by re-using the machinery in landscape.lib.tests.test_amp, instead of delegating the behavior to the real twisted reactor (hence making it not fake at all)

- Update tests that were using FakeReactor.listen_unix/connect_unix (directly or via helpers) to be synchronous. This means essentially using FakeReactor.advance to simulate time progresses and trigger scheduled calls.

- Make landscape.lib.amp.RemoteObject aware of fake transports, so it can flush it transparently. This is a hacky workaround to the fact that twisted's AMP code doesn't like synchronous transports, and without it we'd need to explicitly flush the fake transport buffers explicitly all the times in a test we communicate with a remote component connected via a fake transport.

- Drop landscape.service.LandscapeService.socket instance attribute, as LandscapeService-derived classes now all use the new landscape.amp.ComponentPublisher convenience introduced in former branches, that takes care of that.

Another episode of the AMP-related cleanup:

- Rename landscape.amp.RemoteComponentConnector to simply ComponentConnector (the fact that it's remote is implied by the fact that you connect to it, and the new name complements the existing ComponentPublisher nicely)

- Drop the landscape.amp.ComponentProtocolClientFactory sub-class and use MethodCallClientFactory directly, tweaking the needed instance attributes with explicit assignements (this is a sort of composition-over-inheritance change)

- Drop landscape.reactor.UnixReactorMixin. The landscape.reactor.FakeReactor class now really fakes the Unix socket transport, by re-using the machinery in landscape.lib.tests.test_amp, instead of delegating the behavior to the real twisted reactor (hence making it not fake at all)

- Update tests that were using FakeReactor.listen_unix/connect_unix (directly or via helpers) to be synchronous. This means essentially using FakeReactor.advance to simulate time progresses and trigger scheduled calls.

- Make landscape.lib.amp.RemoteObject aware of fake transports, so it can flush it transparently. This is a hacky workaround to the fact that twisted's AMP code doesn't like synchronous transports, and without it we'd need to explicitly flush the fake transport buffers explicitly all the times in a test we communicate with a remote component connected via a fake transport.

- Drop landscape.service.LandscapeService.socket instance attribute, as LandscapeService-derived classes now all use the new landscape.amp.ComponentPublisher convenience introduced in former branches, that takes care of that.

[Geoff Teale (tealeg)]

+1

The duplication of comments in test_configuration is a bit odd, but I guess you're trying to ensure that point isn't missed.

review: Approve

[Christopher Armstrong (radix)]

Have you filed a bug about Twisted's AMP not supporting synchronous transports? I'm not really sure what that means, that's why I ask, so I can go check out the explanation on that bug.

[1] The code around the call to self._factory.connection.flush() is unclear to me. It *seems* that this code is only meant to be run in unit tests, not in normal execution, but the conditional (if the factory has a non-None connection) does not scream "testing-only" to me. Or is there a "flush" method on the real connection object that I'm unfamiliar with?

[2] It doesn't seem right that _socket_paths is a class attribute instead of an instance attribute

Everything else looks fine!

review: Approve

[Chris Glass (tribaal)]

Looks good! Thanks for doing this cleanup.

[1]
landscape/monitor/tests/test_service.py:1: 'ANY' imported but unused
landscape/service.py:1: 'os' imported but unused

review: Approve

[Free Ekanayaka (free.ekanayaka)]

Hi Chris, thanks for the careful review. I've filed:

http://twistedmatrix.com/trac/ticket/6502

describing the issue.

[1]

You're right, that code is only meant for unit tests. I've renamed MethodCallClientFactory.connection to fake_connection and added a comment:

    # XXX support exposing fake asynchronous connections created by tests, so
    # they can be flushed transparently and emulate a synchronous behavior. See
    # also http://twistedmatrix.com/trac/ticket/6502, once that's fixed this
    # hack can be removed.
    fake_connection = None

The FakeReactor.connect_unix is the place where it gets set for unit-tests:

    connection = FakeConnection(factory.buildProtocol(path),
                                server.buildProtocol(path))
    factory.fake_connection = connection

[2]

Yeah, that's indeed a hack. The problem is that FakeReactor instances for clients and servers need to be different. I added a comment:

    # XXX probably this shouldn't be a class attribute, but we need client-side
    # FakeReactor instaces to be aware of listening sockets created by
    # server-side FakeReactor instances.
    _socket_paths = {}

Maybe the way to solve it would be to add API for declaring a reactor "aware" of another reactor (so tests would need to call it explicitly). Suggestions are welcome.

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

669. By Free Ekanayaka on 2013-05-02

Address radix's comments

670. By Free Ekanayaka on 2013-05-02

Drop unused imports

[Free Ekanayaka (free.ekanayaka)]

Hey Chris (Glass), unused imports fixed.

=== modified file 'landscape/amp.py'
--- landscape/amp.py	2013-04-29 15:46:13 +0000
+++ landscape/amp.py	2013-05-02 08:30:35 +0000
@@ -17,11 +17,6 @@
     MethodCallClientFactory, MethodCallServerFactory, RemoteObject)
 
 
-class ComponentProtocolClientFactory(MethodCallClientFactory):
-
-    initialDelay = 0.05
-
-
 class ComponentPublisher(object):
     """Publish a Landscape client component using a UNIX socket.
 
@@ -35,6 +30,7 @@
     """
 
     methods = ("ping", "exit")
+    factory = MethodCallServerFactory
 
     def __init__(self, component, reactor, config):
         self._reactor = reactor
@@ -53,7 +49,7 @@
         return self._port.stopListening()
 
 
-class RemoteComponentConnector(object):
+class ComponentConnector(object):
     """Utility superclass for creating connections with a Landscape component.
 
     @cvar component: The class of the component to connect to, it is expected
@@ -71,8 +67,8 @@
 
     @see: L{MethodCallClientFactory}.
     """
+    factory = MethodCallClientFactory
     component = None  # Must be defined by sub-classes
-    factory = ComponentProtocolClientFactory
     remote = RemoteObject
 
     def __init__(self, reactor, config, retry_on_reconnect=False):
@@ -95,8 +91,8 @@
             result in a faster reconnection attempts pace.
         @param quiet: A boolean indicating whether to log errors.
         """
-        reactor = self._reactor._reactor
-        factory = self.factory(reactor)
+        factory = self.factory(self._reactor._reactor)
+        factory.initialDelay = factory.delay = 0.05
         factory.retryOnReconnect = self._retry_on_reconnect
         factory.remote = self.remote
 
@@ -115,8 +111,8 @@
         if factor:
             factory.factor = factor
         socket_path = _get_socket_path(self.component, self._config)
-        self._connector = reactor.connectUNIX(socket_path, factory)
         deferred = factory.getRemoteObject()
+        self._connector = self._reactor.connect_unix(socket_path, factory)
 
         if not quiet:
             deferred.addErrback(log_error)
@@ -155,7 +151,7 @@
     def register(cls, connector_class):
         """Register a connector for a Landscape component.
 
-        @param connector_class: A sub-class of L{RemoteComponentConnector}
+        @param connector_class: A sub-class of L{ComponentConnector}
             that can be used to connect to a certain component.
         """
         cls._by_name[connector_class.component.name] = connector_class
 
=== modified file 'landscape/broker/amp.py'
--- landscape/broker/amp.py	2013-04-29 11:44:32 +0000
+++ landscape/broker/amp.py	2013-05-02 08:30:35 +0000
@@ -2,7 +2,7 @@
 
 from landscape.lib.amp import RemoteObject, MethodCallArgument
 from landscape.amp import (
-    RemoteComponentConnector, RemoteComponentsRegistry, ComponentPublisher)
+    ComponentConnector, RemoteComponentsRegistry, ComponentPublisher)
 from landscape.broker.server import BrokerServer
 from landscape.broker.client import BrokerClient
 from landscape.monitor.monitor import Monitor
@@ -95,14 +95,14 @@
         "message")
 
 
-class RemoteBrokerConnector(RemoteComponentConnector):
+class RemoteBrokerConnector(ComponentConnector):
     """Helper to create connections with the L{BrokerServer}."""
 
     remote = RemoteBroker
     component = BrokerServer
 
 
-class RemoteClientConnector(RemoteComponentConnector):
+class RemoteClientConnector(ComponentConnector):
     """Helper to create connections with the L{BrokerServer}."""
 
     component = BrokerClient
 
=== modified file 'landscape/broker/tests/test_amp.py'
--- landscape/broker/tests/test_amp.py	2013-04-15 09:03:43 +0000
+++ landscape/broker/tests/test_amp.py	2013-05-02 08:30:35 +0000
@@ -168,9 +168,11 @@
         L{RemoteBroker.listen_events} returns a deferred which fires when
         the first of the given events occurs in the broker reactor.
         """
-        result = self.remote.listen_events(["event1", "event2"])
-        self.reactor._reactor.callLater(0.05, self.reactor.fire, "event2")
-        return self.assertSuccess(result, "event2")
+        deferred = self.remote.listen_events(["event1", "event2"])
+        self.reactor.call_later(0.05, self.reactor.fire, "event2")
+        self.reactor.advance(0.05)
+        self.remote._factory.fake_connection.flush()
+        self.assertEqual("event2", self.successResultOf(deferred))
 
     def test_call_on_events(self):
         """
@@ -182,10 +184,12 @@
         callback2 = self.mocker.mock()
         self.expect(callback2()).result(123)
         self.mocker.replay()
-        result = self.remote.call_on_event({"event1": callback1,
-                                            "event2": callback2})
-        self.reactor._reactor.callLater(0.05, self.reactor.fire, "event2")
-        return self.assertSuccess(result, 123)
+        deferred = self.remote.call_on_event({"event1": callback1,
+                                              "event2": callback2})
+        self.reactor.call_later(0.05, self.reactor.fire, "event2")
+        self.reactor.advance(0.05)
+        self.remote._factory.fake_connection.flush()
+        self.assertEqual(123, self.successResultOf(deferred))
 
     def test_fire_event(self):
         """
@@ -202,9 +206,8 @@
         """
         Trying to call an non-exposed broker method results in a failure.
         """
-        result = self.remote._sender.send_method_call(
-            method="get_clients", args=[], kwargs={})
-        return self.assertFailure(result, MethodCallError)
+        deferred = self.remote.get_clients()
+        self.failureResultOf(deferred).trap(MethodCallError)
 
 
 class RemoteClientTest(LandscapeTest):
@@ -263,6 +266,5 @@
         """
         Trying to call an non-exposed client method results in a failure.
         """
-        result = self.remote_client._sender.send_method_call(
-            method="get_plugins", args=[], kwargs={})
-        return self.assertFailure(result, MethodCallError)
+        deferred = self.remote_client.get_plugins()
+        self.failureResultOf(deferred).trap(MethodCallError)
 
=== modified file 'landscape/broker/tests/test_service.py'
--- landscape/broker/tests/test_service.py	2013-04-29 11:37:27 +0000
+++ landscape/broker/tests/test_service.py	2013-05-02 08:30:35 +0000
@@ -14,7 +14,11 @@
 
     def setUp(self):
         super(BrokerServiceTest, self).setUp()
-        self.service = BrokerService(self.config)
+
+        class FakeBrokerService(BrokerService):
+            reactor_factory = FakeReactor
+
+        self.service = FakeBrokerService(self.config)
 
     def test_persist(self):
         """
 
=== modified file 'landscape/configuration.py'
--- landscape/configuration.py	2013-03-27 22:58:20 +0000
+++ landscape/configuration.py	2013-05-02 08:30:35 +0000
@@ -618,7 +618,8 @@
 
         0.05 * (1 - 1.62 ** 14) / (1 - 1.62) = 69 seconds
    """
-    reactor = TwistedReactor()
+    if reactor is None:
+        reactor = TwistedReactor()
     exit_with_error = []
 
     def stop(error=None):
 
=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2013-04-30 16:21:01 +0000
+++ landscape/lib/amp.py	2013-05-02 08:30:35 +0000
@@ -349,6 +349,14 @@
             result.addCallback(self._handle_result, deferred)
             result.addErrback(self._handle_failure, method, args, kwargs,
                               deferred)
+
+            if self._factory.fake_connection is not None:
+                # Transparently flush the connection after a send_method_call
+                # invokation letting tests simulate a synchronous transport.
+                # This is needed because the Twisted's AMP implementation
+                # assume that the transport is asynchronous.
+                self._factory.fake_connection.flush()
+
             return deferred
 
         return send_method_call
@@ -487,6 +495,12 @@
     retryOnReconnect = False
     retryTimeout = None
 
+    # XXX support exposing fake asynchronous connections created by tests, so
+    # they can be flushed transparently and emulate a synchronous behavior. See
+    # also http://twistedmatrix.com/trac/ticket/6502, once that's fixed this
+    # hack can be removed.
+    fake_connection = None
+
     def __init__(self, reactor=None):
         """
         @param object: The object exposed by the L{MethodCallProtocol}s
 
=== modified file 'landscape/lib/tests/test_amp.py'
--- landscape/lib/tests/test_amp.py	2013-04-30 16:21:01 +0000
+++ landscape/lib/tests/test_amp.py	2013-05-02 08:30:35 +0000
@@ -1,28 +1,28 @@
 from twisted.internet import reactor
+from twisted.internet.error import ConnectError, ConnectionDone
+from twisted.internet.task import Clock
 from twisted.internet.defer import Deferred, DeferredList
-from twisted.internet.error import ConnectionDone, ConnectError
-from twisted.internet.task import Clock
-from twisted.protocols.amp import AMP
 from twisted.python.failure import Failure
 
 from landscape.lib.amp import (
     MethodCallError, MethodCallServerProtocol, MethodCallClientProtocol,
     MethodCallServerFactory, MethodCallClientFactory, RemoteObject,
-    RemoteObjectConnector, MethodCallReceiver, MethodCallSender)
+    RemoteObjectConnector, MethodCallSender)
 from landscape.tests.helpers import LandscapeTest
 
 
 class FakeTransport(object):
     """Accumulate written data into a list."""
 
-    def __init__(self):
+    def __init__(self, connection):
         self.stream = []
+        self.connection = connection
 
     def write(self, data):
         self.stream.append(data)
 
     def loseConnection(self):
-        raise NotImplemented()
+        self.connection.disconnect()
 
     def getPeer(self):
         pass
@@ -37,8 +37,16 @@
     def __init__(self, client, server):
         self.client = client
         self.server = server
-        self.server.makeConnection(FakeTransport())
-        self.client.makeConnection(FakeTransport())
+        self.connect()
+
+    def connect(self):
+        self.server.makeConnection(FakeTransport(self))
+        self.client.makeConnection(FakeTransport(self))
+
+    def disconnect(self):
+        reason = Failure(ConnectionDone())
+        connector = self
+        self.client.factory.clientConnectionLost(connector, reason)
 
     def flush(self):
         """
@@ -338,17 +346,7 @@
         client.factory = WordsFactory(self.clock)
         self.remote = RemoteObject(client.factory)
         self.connection = FakeConnection(client, server)
-
-        send_method_call = self.remote._sender.send_method_call
-
-        def synchronous_send_method_call(method, args=[], kwargs={}):
-            # Transparently flush the connection after a send_method_call
-            # invocation
-            deferred = send_method_call(method, args=args, kwargs=kwargs)
-            self.connection.flush()
-            return deferred
-
-        self.remote._sender.send_method_call = synchronous_send_method_call
+        client.factory.fake_connection = self.connection
 
     def test_method_call_sender_with_forbidden_method(self):
         """
 
=== modified file 'landscape/manager/tests/test_usermanager.py'
--- landscape/manager/tests/test_usermanager.py	2013-03-20 08:58:22 +0000
+++ landscape/manager/tests/test_usermanager.py	2013-05-02 08:30:35 +0000
@@ -422,7 +422,7 @@
         def handle_callback(ignored):
             messages = self.broker_service.message_store.get_pending_messages()
             # Ignore the message created by plugin.run.
-            messages = sorted(messages[1:3],
+            messages = sorted([messages[1], messages[3]],
                               key=lambda message: message["operation-id"])
             self.assertMessages(messages,
                                 [{"type": "operation-result",
 
=== modified file 'landscape/manager/usermanager.py'
--- landscape/manager/usermanager.py	2013-04-22 09:32:25 +0000
+++ landscape/manager/usermanager.py	2013-05-02 08:30:35 +0000
@@ -1,9 +1,7 @@
-import os
 import logging
 
-from landscape.lib.amp import RemoteObject
 from landscape.lib.encoding import encode_dict_if_needed
-from landscape.amp import ComponentPublisher, RemoteComponentConnector
+from landscape.amp import ComponentPublisher, ComponentConnector
 
 from landscape.user.management import UserManagement
 from landscape.manager.plugin import ManagerPlugin
@@ -158,6 +156,6 @@
     methods = ["get_locked_usernames"]
 
 
-class RemoteUserManagerConnector(RemoteComponentConnector):
+class RemoteUserManagerConnector(ComponentConnector):
 
     component = UserManager
 
=== modified file 'landscape/monitor/tests/test_service.py'
--- landscape/monitor/tests/test_service.py	2013-04-29 11:37:27 +0000
+++ landscape/monitor/tests/test_service.py	2013-05-02 08:30:35 +0000
@@ -1,4 +1,3 @@
-from landscape.tests.mocker import ANY
 from landscape.tests.helpers import LandscapeTest, FakeBrokerServiceHelper
 from landscape.reactor import FakeReactor
 from landscape.monitor.config import MonitorConfiguration, ALL_PLUGINS
@@ -46,13 +45,6 @@
         starts the plugins and register the monitor as broker client.  It also
         start listening on its own socket for incoming connections.
         """
-        # FIXME: don't actually run the real register method, because at the
-        # moment the UserMonitor plugin still depends on DBus. We can probably
-        # drop this mocking once the AMP migration is completed.
-        for plugin in self.service.plugins:
-            plugin.register = self.mocker.mock()
-            plugin.register(ANY)
-        self.mocker.replay()
 
         def stop_service(ignored):
             [connector] = self.broker_service.broker.get_connectors()
 
=== modified file 'landscape/monitor/usermonitor.py'
--- landscape/monitor/usermonitor.py	2013-04-22 09:32:25 +0000
+++ landscape/monitor/usermonitor.py	2013-05-02 08:30:35 +0000
@@ -1,7 +1,7 @@
 from twisted.internet.defer import maybeDeferred
 
 from landscape.lib.log import log_failure
-from landscape.amp import ComponentPublisher, RemoteComponentConnector
+from landscape.amp import ComponentPublisher, ComponentConnector
 
 from landscape.monitor.plugin import MonitorPlugin
 from landscape.user.changes import UserChanges
@@ -111,6 +111,6 @@
     methods = ["detect_changes"]
 
 
-class RemoteUserMonitorConnector(RemoteComponentConnector):
+class RemoteUserMonitorConnector(ComponentConnector):
 
     component = UserMonitor
 
=== modified file 'landscape/package/taskhandler.py'
--- landscape/package/taskhandler.py	2013-02-28 10:09:09 +0000
+++ landscape/package/taskhandler.py	2013-05-02 08:30:35 +0000
@@ -276,10 +276,7 @@
 
     def finish():
         connector.disconnect()
-        # For some obscure reason our TwistedReactor.stop method calls
-        # reactor.crash() instead of reactor.stop(), which doesn't work
-        # here. Maybe TwistedReactor.stop should simply use reactor.stop().
-        reactor.call_later(0, reactor._reactor.stop)
+        reactor.call_later(0, reactor.stop)
 
     def got_error(failure):
         log_failure(failure)
 
=== modified file 'landscape/package/tests/test_changer.py'
--- landscape/package/tests/test_changer.py	2013-03-18 03:34:11 +0000
+++ landscape/package/tests/test_changer.py	2013-05-02 08:30:35 +0000
@@ -1479,7 +1479,7 @@
             self.broker_service.reactor.advance(100)
             self.twisted_reactor.advance(10)
             payloads = self.broker_service.exchanger._transport.payloads
-            self.assertEqual(1, len(payloads))
+            self.assertEqual(0, len(payloads))
 
         [arguments] = self.process_factory.spawns
         protocol = arguments[0]
 
=== modified file 'landscape/package/tests/test_taskhandler.py'
--- landscape/package/tests/test_taskhandler.py	2012-05-10 00:34:13 +0000
+++ landscape/package/tests/test_taskhandler.py	2013-05-02 08:30:35 +0000
@@ -1,10 +1,9 @@
 import os
 
-from twisted.internet import reactor
 from twisted.internet.defer import Deferred, fail
 
 from landscape.lib.lock import lock_path
-from landscape.reactor import TwistedReactor
+from landscape.reactor import TwistedReactor, FakeReactor
 from landscape.broker.amp import RemoteBrokerConnector
 from landscape.package.taskhandler import (
     PackageTaskHandlerConfiguration, PackageTaskHandler, run_task_handler,
@@ -366,7 +365,8 @@
         reactor_mock.run()
         self.mocker.call(lambda: call_when_running[0]())
         connector_mock.disconnect()
-        reactor_mock.call_later(0, reactor.stop)
+        reactor_mock.call_later(0, ANY)
+        self.mocker.result(None)
 
         # Okay, the whole playground is set.
         self.mocker.replay()
@@ -433,12 +433,10 @@
 
     def test_errors_in_tasks_are_printed_and_exit_program(self):
         # Ignore a bunch of crap that we don't care about
-        reactor_mock = self.mocker.patch(TwistedReactor)
         init_logging_mock = self.mocker.replace("landscape.deployment"
                                                 ".init_logging",
                                                 passthrough=False)
         init_logging_mock(ARGS)
-        reactor_mock.run()
 
         class MyException(Exception):
             pass
@@ -450,8 +448,6 @@
         handler_mock = handler_factory_mock(ARGS)
         self.expect(handler_mock.run()).result(fail(MyException("Hey error")))
 
-        reactor_mock.call_later(0, reactor.stop)
-
         self.mocker.replay()
 
         # Ok now for some real stuff
@@ -460,7 +456,8 @@
             self.assertIn("MyException", self.logfile.getvalue())
 
         result = run_task_handler(handler_factory_mock,
-                                  ["-c", self.config_filename])
+                                  ["-c", self.config_filename],
+                                  reactor=FakeReactor())
         return result.addCallback(assert_log)
 
 
@@ -473,7 +470,7 @@
         The L{LazyRemoteBroker} class doesn't initialize the actual remote
         broker until one of its attributes gets actually accessed.
         """
-        reactor = TwistedReactor()
+        reactor = FakeReactor()
         connector = RemoteBrokerConnector(reactor, self.broker_service.config)
         self.broker = LazyRemoteBroker(connector)
         self.assertIs(self.broker._remote, None)
 
=== modified file 'landscape/reactor.py'
--- landscape/reactor.py	2013-04-12 11:49:59 +0000
+++ landscape/reactor.py	2013-05-02 08:30:35 +0000
@@ -6,6 +6,8 @@
 import logging
 import bisect
 
+from twisted.python.failure import Failure
+from twisted.internet.error import ConnectError
 from twisted.internet.threads import deferToThread
 
 from landscape.log import format_object
@@ -110,33 +112,13 @@
             raise InvalidID("EventID instance expected, received %r" % id)
 
 
-class UnixReactorMixin(object):
-    """Support listening on Unix domain sockets.
-
-    Note that this mixin uses the *real* Twisted reactor to open a *real*
-    socket.
-
-    Since it is used by *both* L{TwistedReactor} and L{FakeReactor}, this
-    means that the latter is not really fake in this sense and that unit
-    tests involving calls to this method won't be synchronous anymore.
-
-    For example, many tests under L{landscape.broker.tests} use C{listen_unix}
-    to setup a "real" remote broker and exercise the RPC/AMP functionality. See
-    in particular L{landscape.broker.tests.helpers.RemoteBrokerHelper}.
-    """
-
-    def listen_unix(self, socket, factory):
-        """Start listening on a Unix socket."""
-        return self._reactor.listenUNIX(socket, factory, wantPID=True)
-
-
 class ReactorID(object):
 
     def __init__(self, timeout):
         self._timeout = timeout
 
 
-class TwistedReactor(EventHandlingReactorMixin, UnixReactorMixin):
+class TwistedReactor(EventHandlingReactorMixin):
     """Wrap and add functionalities to the Twisted C{reactor}.
 
     This essentially a facade around the C{twisted.internet.reactor} and
@@ -239,6 +221,14 @@
         deferred.addCallback(on_success)
         deferred.addErrback(on_failure)
 
+    def listen_unix(self, socket, factory):
+        """Start listening on a Unix socket."""
+        return self._reactor.listenUNIX(socket, factory, wantPID=True)
+
+    def connect_unix(self, socket, factory):
+        """Connect to a Unix socket."""
+        return self._reactor.connectUNIX(socket, factory)
+
     def run(self):
         """Start the reactor, a C{"run"} event will be fired."""
 
@@ -266,7 +256,7 @@
         self._data = data
 
 
-class FakeReactor(EventHandlingReactorMixin, UnixReactorMixin):
+class FakeReactor(EventHandlingReactorMixin):
     """A fake reactor with the same API of L{TwistedReactor}.
 
     This reactor emulates the asychronous interface of L{TwistedReactor}, but
@@ -279,6 +269,10 @@
     around Unix sockets), and implement a fake version C{listen_unix}, but this
     hasn't been done yet.
     """
+    # XXX probably this shouldn't be a class attribute, but we need client-side
+    # FakeReactor instaces to be aware of listening sockets created by
+    # server-side FakeReactor instances.
+    _socket_paths = {}
 
     def __init__(self):
         super(FakeReactor, self).__init__()
@@ -287,8 +281,8 @@
         self.hosts = {}
         self._threaded_callbacks = []
 
-        # We need a reference to the Twisted reactor as well to
-        # let Landscape services listen to Unix sockets
+        # XXX we need a reference to the Twisted reactor as well because
+        # some tests use it
         from twisted.internet import reactor
         self._reactor = reactor
 
@@ -326,7 +320,8 @@
             super(FakeReactor, self).cancel_call(id)
 
     def call_when_running(self, f):
-        raise NotImplemented("The FakeReactor doesn't implement this.")
+        # Just schedule a call that will be kicked by the run() method.
+        self.call_later(0, f)
 
     def call_in_main(self, f, *args, **kwargs):
         """Schedule a function for execution in the main thread."""
@@ -347,6 +342,28 @@
         self._in_thread(callback, errback, f, args, kwargs)
         self._run_threaded_callbacks()
 
+    def listen_unix(self, socket_path, factory):
+
+        class FakePort(object):
+
+            def stopListening(oself):
+                self._socket_paths.pop(socket_path)
+
+        self._socket_paths[socket_path] = factory
+        return FakePort()
+
+    def connect_unix(self, path, factory):
+        server = self._socket_paths.get(path)
+        from landscape.lib.tests.test_amp import FakeConnection
+        if server:
+            connection = FakeConnection(factory.buildProtocol(path),
+                                        server.buildProtocol(path))
+            factory.fake_connection = connection
+        else:
+            connector = object()  # Fake connector
+            failure = Failure(ConnectError("No such file or directory"))
+            factory.clientConnectionFailed(connector, failure)
+
     def run(self):
         """Continuously advance this reactor until reactor.stop() is called."""
         self.fire("run")
 
=== modified file 'landscape/service.py'
--- landscape/service.py	2013-04-29 11:44:32 +0000
+++ landscape/service.py	2013-05-02 08:30:35 +0000
@@ -1,4 +1,3 @@
-import os
 import logging
 import signal
 
@@ -42,15 +41,11 @@
             signal.signal(
                 signal.SIGUSR1,
                 lambda signal, frame: reactor.callFromThread(rotate_logs))
-        self.socket = os.path.join(self.config.sockets_path,
-                                   self.service_name + ".sock")
 
     def startService(self):
         Service.startService(self)
         logging.info("%s started with config %s" % (
             self.service_name.capitalize(), self.config.get_config_filename()))
-        if hasattr(self, "factory"):
-            self.port = self.reactor.listen_unix(self.socket, self.factory)
 
     def stopService(self):
         # We don't need to call port.stopListening(), because the reactor
 
=== modified file 'landscape/tests/test_amp.py'
--- landscape/tests/test_amp.py	2013-04-22 09:32:25 +0000
+++ landscape/tests/test_amp.py	2013-05-02 08:30:35 +0000
@@ -4,9 +4,7 @@
 from landscape.tests.helpers import LandscapeTest
 from landscape.reactor import FakeReactor
 from landscape.deployment import Configuration
-from landscape.amp import (
-    ComponentProtocolClientFactory, RemoteComponentConnector,
-    ComponentPublisher)
+from landscape.amp import ComponentPublisher, ComponentConnector
 
 
 class TestComponent(object):
@@ -14,22 +12,21 @@
     name = "test"
 
 
-class TestComponentProtocolFactory(ComponentProtocolClientFactory):
-
-    maxRetries = 0
-    initialDelay = 0.01
-
-
-class RemoteTestComponentConnector(RemoteComponentConnector):
-
-    factory = TestComponentProtocolFactory
+class TestComponentConnector(ComponentConnector):
+
     component = TestComponent
 
 
-class RemoteComponentTest(LandscapeTest):
+class FakeAMP(object):
+
+    def __init__(self, locator):
+        self._locator = locator
+
+
+class ComponentPublisherTest(LandscapeTest):
 
     def setUp(self):
-        super(RemoteComponentTest, self).setUp()
+        super(ComponentPublisherTest, self).setUp()
         reactor = FakeReactor()
         config = Configuration()
         config.data_path = self.makeDir()
@@ -38,7 +35,7 @@
         self.publisher = ComponentPublisher(self.component, reactor, config)
         self.publisher.start()
 
-        self.connector = RemoteTestComponentConnector(reactor, config)
+        self.connector = TestComponentConnector(reactor, config)
         connected = self.connector.connect()
         connected.addCallback(lambda remote: setattr(self, "remote", remote))
         return connected
@@ -46,7 +43,7 @@
     def tearDown(self):
         self.connector.disconnect()
         self.publisher.stop()
-        super(RemoteComponentTest, self).tearDown()
+        super(ComponentPublisherTest, self).tearDown()
 
     def test_ping(self):
         """
@@ -71,16 +68,15 @@
         return self.assertSuccess(result)
 
 
-class RemoteComponentConnectorTest(LandscapeTest):
+class ComponentConnectorTest(LandscapeTest):
 
     def setUp(self):
-        super(RemoteComponentConnectorTest, self).setUp()
+        super(ComponentConnectorTest, self).setUp()
         self.reactor = FakeReactor()
         self.config = Configuration()
         self.config.data_path = self.makeDir()
         self.makeDir(path=self.config.sockets_path)
-        self.connector = RemoteTestComponentConnector(self.reactor,
-                                                      self.config)
+        self.connector = TestComponentConnector(self.reactor, self.config)
 
     def test_connect_logs_errors(self):
         """
 
=== modified file 'landscape/tests/test_configuration.py'
--- landscape/tests/test_configuration.py	2013-04-18 09:32:04 +0000
+++ landscape/tests/test_configuration.py	2013-05-02 08:30:35 +0000
@@ -5,9 +5,10 @@
 from cStringIO import StringIO
 
 from twisted.internet.defer import succeed, fail
+from twisted.internet.task import Clock
 
 from landscape.lib.amp import MethodCallSender
-from landscape.reactor import TwistedReactor
+from landscape.reactor import TwistedReactor, FakeReactor
 from landscape.lib.fetch import HTTPCodeError, PyCurlError
 from landscape.configuration import (
     print_text, LandscapeSetupScript, LandscapeSetupConfiguration,
@@ -1634,7 +1635,7 @@
         registration_mock = self.mocker.replace(service.registration)
         config_mock = self.mocker.replace(service.config)
         print_text_mock = self.mocker.replace(print_text)
-        reactor_mock = self.mocker.patch(TwistedReactor)
+        reactor_mock = self.mocker.patch(FakeReactor)
 
         # This must necessarily happen in the following order.
         self.mocker.order()
@@ -1646,8 +1647,6 @@
         time_mock.sleep(ANY)
         self.mocker.count(1)
 
-        reactor_mock.run()
-
         # After a nice dance the configuration is reloaded.
         config_mock.reload()
 
@@ -1665,6 +1664,10 @@
 
         reactor_mock.stop()
 
+        # This is actually called after everything else since all deferreds
+        # are synchronous and callbacks will be executed immediately.
+        reactor_mock.run()
+
         # Nothing else is printed!
         print_text_mock(ANY)
         self.mocker.count(0)
@@ -1672,7 +1675,8 @@
         self.mocker.replay()
 
         # DO IT!
-        return register(self.config, print_text, sys.exit)
+        return register(self.config, print_text, sys.exit,
+                        reactor=FakeReactor())
 
     def test_register_failure(self):
         """
@@ -1685,7 +1689,7 @@
         registration_mock = self.mocker.replace(service.registration)
         config_mock = self.mocker.replace(service.config)
         print_text_mock = self.mocker.replace(print_text)
-        reactor_mock = self.mocker.patch(TwistedReactor)
+        reactor_mock = self.mocker.patch(FakeReactor)
 
         # This must necessarily happen in the following order.
         self.mocker.order()
@@ -1697,8 +1701,6 @@
         time_mock.sleep(ANY)
         self.mocker.count(1)
 
-        reactor_mock.run()
-
         # After a nice dance the configuration is reloaded.
         config_mock.reload()
 
@@ -1717,6 +1719,10 @@
 
         reactor_mock.stop()
 
+        # This is actually called after everything else since all deferreds
+        # are synchronous and callbacks will be executed immediately.
+        reactor_mock.run()
+
         # Nothing else is printed!
         print_text_mock(ANY)
         self.mocker.count(0)
@@ -1724,7 +1730,9 @@
         self.mocker.replay()
 
         # DO IT!
-        return register(self.config, print_text, sys.exit)
+        exit = []
+        register(self.config, print_text, exit.append, reactor=FakeReactor())
+        self.assertEqual([2], exit)
 
     def test_register_exchange_failure(self):
         """
@@ -1736,7 +1744,7 @@
         registration_mock = self.mocker.replace(service.registration)
         config_mock = self.mocker.replace(service.config)
         print_text_mock = self.mocker.replace(print_text)
-        reactor_mock = self.mocker.patch(TwistedReactor)
+        reactor_mock = self.mocker.patch(FakeReactor)
 
         # This must necessarily happen in the following order.
         self.mocker.order()
@@ -1748,8 +1756,6 @@
         time_mock.sleep(ANY)
         self.mocker.count(1)
 
-        reactor_mock.run()
-
         # After a nice dance the configuration is reloaded.
         config_mock.reload()
 
@@ -1767,6 +1773,10 @@
 
         reactor_mock.stop()
 
+        # This is actually called after everything else since all deferreds
+        # are synchronous and callbacks will be executed immediately.
+        reactor_mock.run()
+
         # Nothing else is printed!
         print_text_mock(ANY)
         self.mocker.count(0)
@@ -1774,7 +1784,9 @@
         self.mocker.replay()
 
         # DO IT!
-        return register(self.config, print_text, sys.exit)
+        exit = []
+        register(self.config, print_text, exit.append, reactor=FakeReactor())
+        self.assertEqual([2], exit)
 
     def test_register_timeout_failure(self):
         service = self.broker_service
@@ -1782,7 +1794,7 @@
         registration_mock = self.mocker.replace(service.registration)
         config_mock = self.mocker.replace(service.config)
         print_text_mock = self.mocker.replace(print_text)
-        reactor_mock = self.mocker.patch(TwistedReactor)
+        reactor_mock = self.mocker.patch(FakeReactor)
         remote_mock = self.mocker.patch(RemoteBroker)
 
         protocol_mock = self.mocker.patch(MethodCallSender)
@@ -1800,17 +1812,19 @@
         time_mock.sleep(ANY)
         self.mocker.count(1)
 
-        reactor_mock.run()
+        # After a nice dance the configuration is reloaded.
+        config_mock.reload()
 
         remote_mock.call_on_event(ANY)
         self.mocker.result(succeed(None))
 
-        # After a nice dance the configuration is reloaded.
-        config_mock.reload()
-
         registration_mock.register()
         self.mocker.passthrough()
 
+        # This is actually called after everything else since all deferreds
+        # are synchronous and callbacks will be executed immediately.
+        reactor_mock.run()
+
         # Nothing else is printed!
         print_text_mock(ANY)
         self.mocker.count(0)
@@ -1818,7 +1832,12 @@
         self.mocker.replay()
 
         # DO IT!
-        return register(self.config, print_text, sys.exit)
+        fake_reactor = FakeReactor()
+        fake_reactor._reactor = Clock()
+        deferred = register(self.config, print_text, sys.exit,
+                            reactor=fake_reactor)
+        fake_reactor._reactor.advance(100)
+        return deferred
 
     def test_register_bus_connection_failure(self):
         """
@@ -1993,14 +2012,13 @@
         remote_broker = self.mocker.mock()
 
         print_text_mock = self.mocker.replace(print_text)
-        reactor_mock = self.mocker.patch(TwistedReactor)
+        reactor_mock = self.mocker.patch(FakeReactor)
 
         # This is unordered. It's just way too much of a pain.
         print_text_mock("Please wait... ", "")
         time_mock = self.mocker.replace("time")
         time_mock.sleep(ANY)
         self.mocker.count(1)
-        reactor_mock.run()
 
         # SNORE
         connector = connector_factory(ANY, configuration)
@@ -2025,10 +2043,12 @@
         # WHOAH DUDE. This waits for callLater(0, reactor.stop).
         connector.disconnect()
         reactor_mock.stop()
+        reactor_mock.run()
 
         self.mocker.replay()
 
-        return register(configuration, print_text, sys.exit, max_retries=0)
+        return register(configuration, print_text, sys.exit, max_retries=0,
+                        reactor=FakeReactor())
 
 
 class SSLCertificateDataTest(LandscapeConfigurationTest):
 
=== modified file 'landscape/tests/test_service.py'
--- landscape/tests/test_service.py	2013-04-22 09:32:25 +0000
+++ landscape/tests/test_service.py	2013-05-02 08:30:35 +0000
@@ -7,7 +7,6 @@
 from landscape.reactor import FakeReactor
 from landscape.deployment import Configuration
 from landscape.service import LandscapeService
-from landscape.amp import RemoteComponentConnector
 from landscape.tests.helpers import LandscapeTest
 
 
@@ -15,10 +14,6 @@
     name = "monitor"
 
 
-class RemoteTestComponentCreator(RemoteComponentConnector):
-    component = TestComponent
-
-
 class TestService(LandscapeService):
     service_name = TestComponent.name
 
 
=== modified file 'landscape/tests/test_watchdog.py'
--- landscape/tests/test_watchdog.py	2013-04-29 11:44:32 +0000
+++ landscape/tests/test_watchdog.py	2013-05-02 08:30:35 +0000
@@ -22,7 +22,7 @@
 from landscape.lib.dns import discover_server
 from landscape.configuration import (
     fetch_base64_ssl_public_certificate, print_text)
-from landscape.amp import ComponentProtocolClientFactory, RemoteComponentConnector
+from landscape.amp import ComponentConnector
 from landscape.broker.amp import RemoteBrokerConnector
 from landscape.deployment import Configuration
 from landscape.reactor import TwistedReactor
@@ -472,15 +472,9 @@
     name = "broker"
 
 
-class StubBrokerProtocolFactory(ComponentProtocolClientFactory):
-
-    initialDelay = 0.1
-
-
-class RemoteStubBrokerConnector(RemoteComponentConnector):
+class RemoteStubBrokerConnector(ComponentConnector):
 
     component = StubBroker
-    factory = StubBrokerProtocolFactory
 
 
 class DaemonTestBase(LandscapeTest):
@@ -945,6 +939,7 @@
         return RemoteBrokerConnector
 
     def test_is_running(self):
+        self.daemon._connector._reactor = self.broker_service.reactor
         result = self.daemon.is_running()
         result.addCallback(self.assertTrue)
         return result
 
=== modified file 'landscape/watchdog.py'
--- landscape/watchdog.py	2013-04-22 09:32:25 +0000
+++ landscape/watchdog.py	2013-05-02 08:30:35 +0000
@@ -68,7 +68,7 @@
     @cvar factor: The factor by which the delay between subsequent connection
         attempts will increase.
 
-    @param connector: The L{RemoteComponentConnector} of the daemon.
+    @param connector: The L{ComponentConnector} of the daemon.
     @param reactor: The reactor used to spawn the process and schedule timed
         calls.
     @param verbose: Optionally, report more information when running this