Merge lp:~free.ekanayaka/txlongpoll/notification-source-integration-tests into lp:txlongpoll

[Free Ekanayaka]

This branch adds integration tests for the NotificationSource class.

They basically cover the same ground as the equivalent tests in the legacy txlongpoll.tests.test_integration.DeprecatedQueueManagerTest test class, but also add new failure scenarios, like network disconnects or broker restarts, showing that NotificationSource handles them gracefully.

[lp:~free.ekanayaka/txlongpoll/notification-source-integration-tests]

113. By Free Ekanayaka on 2016-06-22

Drop leftover code

[Alberto Donato (ack)]

+1, looks good

A few minor comments/nits inline

review: Approve

=== renamed file 'txlongpoll/testing/client.py' => 'txlongpoll/testing/integration.py'
--- txlongpoll/testing/client.py	2016-06-08 07:48:49 +0000
+++ txlongpoll/testing/integration.py	2016-06-22 11:06:16 +0000
@@ -14,11 +14,19 @@
     AsynchronousDeferredRunTestForBrokenTwisted,
     )
 from twisted.internet import reactor
+from twisted.internet.protocol import (
+    Protocol,
+    Factory,
+)
 from twisted.internet.defer import (
     Deferred,
     DeferredQueue,
     inlineCallbacks,
     )
+from twisted.internet.endpoints import (
+    TCP4ClientEndpoint,
+)
+from twisted.application.service import Service
 from txamqp.client import Closed
 from txlongpoll.client import AMQFactory
 
@@ -61,13 +69,16 @@
         """
 
 
-class AMQTest(ResourcedTestCase, TestCase):
+class IntegrationTest(ResourcedTestCase, TestCase):
 
     run_tests_with = AsynchronousDeferredRunTestForBrokenTwisted.make_factory(
-        timeout=5)
+        timeout=10)
 
     resources = [('rabbit', FixtureResource(RabbitServerWithoutReset()))]
 
+
+class AMQTest(IntegrationTest):
+
     VHOST = "/"
     USER = "guest"
     PASSWORD = "guest"
@@ -172,3 +183,87 @@
         """
         self.exchanges.add(exchange)
         return self.real_exchange_declare(exchange=exchange, **kwargs)
+
+
+class ProxyService(Service):
+    """A TCP proxy that can be instructed to drop packets on the floor."""
+
+    def __init__(self, host, port):
+        """
+        @param host: The backend host to proxy.
+        @param port: The port on the backend host to proxy.
+        """
+        self._host = host
+        self._port = port
+        self._listener = None
+        self._factory = None
+
+    def startService(self):
+        super(ProxyService, self).startService()
+        self._factory = Factory()
+        self._factory.protocol = _FrontendProtocol
+        self._factory.blocked = False
+        self._factory.connections = 0
+        self._factory.backend = TCP4ClientEndpoint(
+            reactor, self._host, self._port)
+
+        self._listener = reactor.listenTCP(0, self._factory)
+
+    def stopService(self):
+        super(ProxyService, self).startService()
+        self._listener.stopListening()
+
+    def block(self):
+        """Drop all packets on the floor."""
+        self._factory.blocked = True
+
+    def unblock(self):
+        """Let packets flow again."""
+        self._factory.blocked = False
+
+    @property
+    def port(self):
+        """Get the frontend port of the proxy."""
+        return self._listener.getHost().port
+
+    @property
+    def connections(self):
+        """Get the number of frontend connections created so far."""
+        return self._factory.connections
+
+
+class _FrontendProtocol(Protocol):
+
+    def connectionMade(self):
+        self.factory.connections += 1
+        self.buffer = ""  # Pending writes
+        self.backend = None  # Backend protocol
+        factory = Factory()
+        factory.protocol = _BackendProtocol
+        factory.frontend = self
+
+        deferred = self.factory.backend.connect(factory)
+        deferred.addCallback(self._backendConnected)
+
+    def connectionLost(self, reason):
+        self.backend.transport.loseConnection()
+
+    def dataReceived(self, data):
+        if self.factory.blocked:
+            return
+        if self.backend:
+            self.backend.transport.write(data)
+        else:
+            self.buffer += data
+
+    def _backendConnected(self, backend):
+        self.backend = backend
+        self.dataReceived(self.buffer)
+
+
+class _BackendProtocol(Protocol):
+
+    def dataReceived(self, data):
+        if self.factory.frontend.factory.blocked:
+            return
+        self.factory.frontend.transport.write(data)

=== modified file 'txlongpoll/tests/test_client.py'
--- txlongpoll/tests/test_client.py	2011-09-30 09:56:12 +0000
+++ txlongpoll/tests/test_client.py	2016-06-22 11:06:16 +0000
@@ -16,7 +16,7 @@
 from txamqp.queue import Closed
 from txamqp.spec import Spec
 from txlongpoll.client import AMQFactory
-from txlongpoll.testing.client import AMQTest
+from txlongpoll.testing.integration import AMQTest
 
 
 class AMQFactoryTest(TestCase):

=== modified file 'txlongpoll/tests/test_integration.py'
--- txlongpoll/tests/test_integration.py	2016-05-31 11:28:08 +0000
+++ txlongpoll/tests/test_integration.py	2016-06-22 11:06:16 +0000
@@ -3,6 +3,8 @@
 
 """Integration tests running a real RabbitMQ broker."""
 
+from rabbitfixture.server import RabbitServerResources
+
 from twisted.internet import reactor
 from twisted.internet.defer import (
     inlineCallbacks,
@@ -12,26 +14,317 @@
     Clock,
     deferLater,
     )
+from twisted.application.internet import (
+    backoffPolicy,
+    ClientService,
+)
 
 from txamqp.content import Content
 from txamqp.protocol import (
     AMQChannel,
     AMQClient,
 )
+from txamqp.factory import AMQFactory
+from txamqp.endpoint import AMQEndpoint
 
 from testtools.deferredruntest import assert_fails_with
 
 from txlongpoll.notification import (
+    NotificationConnector,
+    NotificationSource,
     NotFound,
     Timeout,
 )
 from txlongpoll.frontend import DeprecatedQueueManager
-from txlongpoll.testing.client import (
+from txlongpoll.testing.integration import (
+    IntegrationTest,
+    ProxyService,
     AMQTest,
     QueueWrapper,
 )
 
 
+class NotificationSourceIntegrationTest(IntegrationTest):
+
+    @inlineCallbacks
+    def setUp(self):
+        super(NotificationSourceIntegrationTest, self).setUp()
+        self.endpoint = AMQEndpoint(
+            reactor, self.rabbit.config.hostname, self.rabbit.config.port,
+            username="guest", password="guest", heartbeat=1)
+        self.policy = backoffPolicy(initialDelay=0)
+        self.service = ClientService(
+            self.endpoint, AMQFactory(), retryPolicy=self.policy)
+        self.connector = NotificationConnector(self.service)
+        self.source = NotificationSource(self.connector)
+
+        self.client = yield self.endpoint.connect(AMQFactory())
+        self.channel = yield self.client.channel(1)
+        yield self.channel.channel_open()
+        yield self.channel.queue_declare(queue="uuid")
+
+        self.service.startService()
+
+    @inlineCallbacks
+    def tearDown(self):
+        self.service.stopService()
+        super(NotificationSourceIntegrationTest, self).tearDown()
+        # Wrap resetting queues and client in a try/except, since the broker
+        # may have been stopped (e.g. when this is the last test being run).
+        try:
+            yield self.channel.queue_delete(queue="uuid")
+        except:
+            pass
+        finally:
+            yield self.client.close()
+
+    @inlineCallbacks
+    def test_get_after_publish(self):
+        """
+        Calling get() after a message has been published in the associated
+        queue, returns a Notification for that message.
+        """
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+        notification = yield self.source.get("uuid", 0)
+        self.assertEqual("hello", notification.payload)
+
+    @inlineCallbacks
+    def test_get_before_publish(self):
+        """
+        Calling get() before a message has been published in the associated
+        queue, will wait until publication.
+        """
+        deferred = self.source.get("uuid", 0)
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+        notification = yield deferred
+        self.assertEqual("hello", notification.payload)
+
+    @inlineCallbacks
+    def test_get_with_error(self):
+        """
+        If an error occurs in during get(), the client is closed so
+        we can query messages again.
+        """
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+        try:
+            yield self.source.get("uuid-unknown", 0)
+        except NotFound:
+            pass
+        else:
+            self.fail("NotFound wasn't raised")
+        notification = yield self.source.get("uuid", 0)
+        self.assertEqual("hello", notification.payload)
+
+    @inlineCallbacks
+    def test_get_concurrent_with_error(self):
+        """
+        If an error occurs in a call to get(), other calls don't
+        fail, and are retried on reconnection instead.
+        """
+        client1 = yield self.service.whenConnected()
+        deferred = self.source.get("uuid", 0)
+
+        try:
+            yield self.source.get("uuid-unknown", 0)
+        except NotFound:
+            pass
+        else:
+            self.fail("NotFound wasn't raised")
+
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+
+        notification = yield deferred
+        self.assertEquals("hello", notification.payload)
+        client2 = yield self.service.whenConnected()
+        # The ClientService has reconnected, yielding a new client.
+        self.assertIsNot(client1, client2)
+
+    @inlineCallbacks
+    def test_get_timeout(self):
+        """
+        Calls to get() timeout after a certain amount of time if no message
+        arrived on the queue.
+        """
+        self.source.timeout = 1
+        try:
+            yield self.source.get("uuid", 0)
+        except Timeout:
+            pass
+        else:
+            self.fail("Timeout not raised")
+        client = yield self.service.whenConnected()
+        channel = yield client.channel(1)
+        # The channel is still opened
+        self.assertFalse(channel.closed)
+        # The consumer has been deleted
+        self.assertNotIn("uuid.0", client.queues)
+
+    @inlineCallbacks
+    def test_get_with_broker_shutdown_during_consume(self):
+        """
+        If rabbitmq gets shutdown during the basic-consume call, we wait
+        for the reconection and retry transparently.
+        """
+        # This will make the connector setup the channel before we call
+        # get(), so by the time we call it in the next line all
+        # connector-related deferreds will fire synchronously and the
+        # code will block on basic-consume.
+        yield self.connector()
+
+        d = self.source.get("uuid", 0)
+
+        # Restart rabbitmq
+        yield self.client.close()
+        yield self.client.disconnected.wait()
+        self.rabbit.cleanUp()
+        self.rabbit.config = RabbitServerResources(
+            port=self.rabbit.config.port)  # Ensure that we use the same port
+        self.rabbit.setUp()
+
+        # Get a new channel and re-declare the queue, since the restart has
+        # destroyed it.
+        self.client = yield self.endpoint.connect(AMQFactory())
+        self.channel = yield self.client.channel(1)
+        yield self.channel.channel_open()
+        yield self.channel.queue_declare(queue="uuid")
+
+        # Publish a message in the queue
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+
+        notification = yield d
+        self.assertEqual("hello", notification.payload)
+
+    @inlineCallbacks
+    def test_get_with_broker_die_during_consume(self):
+        """
+        If rabbitmq dies during the basic-consume call, we wait for the
+        reconection and retry transparently.
+        """
+        # This will make the connector setup the channel before we call
+        # get(), so by the time we call it in the next line all
+        # connector-related deferreds will fire synchronously and the
+        # code will block on basic-consume.
+        yield self.connector()
+
+        d = self.source.get("uuid", 0)
+
+        # Kill rabbitmq and start it again
+        yield self.client.close()
+        yield self.client.disconnected.wait()
+        self.rabbit.runner.kill()
+        self.rabbit.cleanUp()
+        self.rabbit.config = RabbitServerResources(
+            port=self.rabbit.config.port)  # Ensure that we use the same port
+        self.rabbit.setUp()
+
+        # Get a new channel and re-declare the queue, since the crash has
+        # destroyed it.
+        self.client = yield self.endpoint.connect(AMQFactory())
+        self.channel = yield self.client.channel(1)
+        yield self.channel.channel_open()
+        yield self.channel.queue_declare(queue="uuid")
+
+        # Publish a message in the queue
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+
+        notification = yield d
+        self.assertEqual("hello", notification.payload)
+
+    @inlineCallbacks
+    def test_wb_get_with_broker_shutdown_during_message_wait(self):
+        """
+        If rabbitmq gets shutdown while we wait for messages, we transparently
+        wait for the reconnection and try again.
+        """
+        # This will make the connector setup the channel before we call
+        # get(), so by the time we call it in the next line all
+        # connector-related deferreds will fire synchronously and the
+        # code will block on basic-consume.
+        yield self.connector()
+
+        d = self.source.get("uuid", 0)
+
+        # Acquiring the channel lock makes sure that basic-consume has
+        # succeeded and we started waiting for the message.
+        yield self.source._channel_lock.acquire()
+        self.source._channel_lock.release()
+
+        # Restart rabbitmq
+        yield self.client.close()
+        yield self.client.disconnected.wait()
+        self.rabbit.cleanUp()
+        self.rabbit.config = RabbitServerResources(
+            port=self.rabbit.config.port)  # Ensure that we use the same port
+        self.rabbit.setUp()
+
+        # Get a new channel and re-declare the queue, since the restart has
+        # destroyed it.
+        self.client = yield self.endpoint.connect(AMQFactory())
+        self.channel = yield self.client.channel(1)
+        yield self.channel.channel_open()
+        yield self.channel.queue_declare(queue="uuid")
+
+        # Publish a message in the queue
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+
+        notification = yield d
+        self.assertEqual("hello", notification.payload)
+
+    @inlineCallbacks
+    def test_wb_heartbeat(self):
+        """
+        If heartbeat checks fail due to network issues, we keep re-trying
+        until the network recovers.
+        """
+        self.service.stopService()
+
+        # Put a TCP proxy between NotificationSource and RabbitMQ, to simulate
+        # packets getting dropped on the floor.
+        proxy = ProxyService(
+            self.rabbit.config.hostname, self.rabbit.config.port)
+        proxy.startService()
+        self.addCleanup(proxy.stopService)
+        self.endpoint._port = proxy.port
+        self.service = ClientService(
+            self.endpoint, AMQFactory(), retryPolicy=self.policy)
+        self.connector._service = self.service
+        self.service.startService()
+
+        # This will make the connector setup the channel before we call
+        # get(), so by the time we call it in the next line all
+        # connector-related deferreds will fire synchronously and the
+        # code will block on basic-consume.
+        channel = yield self.connector()
+
+        deferred = self.source.get("uuid", 0)
+
+        # Start dropping packets on the floor
+        proxy.block()
+
+        # Publish a notification, which won't be delivered just yet.
+        yield self.channel.basic_publish(
+            routing_key="uuid", content=Content("hello"))
+
+        # Wait for the first connection to terminate, because heartbeat
+        # checks will fail.
+        yield channel.client.disconnected.wait()
+
+        # Now let packets flow again.
+        proxy.unblock()
+
+        # The situation got recovered.
+        notification = yield deferred
+        self.assertEqual("hello", notification.payload)
+        self.assertEqual(2, proxy.connections)
+
+
 class DeprecatedQueueManagerTest(AMQTest):
 
     prefix = None