Merge ~adam-collard/maas:flaky-lease-socket-service-fix into maas:master

[Adam Collard]

fix(lease_service): flaky tests

Introduce configure_lease_service_for_one_shot helper coroutine

Refactor test_helper_notify and test_lease_socket_service to use the
helper which guarantees that we don't lose a race with the LoopingTask
at the heart of the LeaseSocketService

Fixes LP:2026360 and LP:2019216

[Christian Grabowski (cgrabowski)]

+1

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b flaky-lease-socket-service-fix lp:~adam-collard/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 5ab1a6674dde7bda111167ca95f879087da0a057

review: Approve

diff --git a/src/provisioningserver/dhcp/tests/test_helper_notify.py b/src/provisioningserver/dhcp/tests/test_helper_notify.py
index d5ef091..b00f4f4 100644
--- a/src/provisioningserver/dhcp/tests/test_helper_notify.py
+++ b/src/provisioningserver/dhcp/tests/test_helper_notify.py
@@ -4,22 +4,21 @@
 """Tests for maas-dhcp-support notify command."""
 
 
+from functools import partial
 import os
 import random
 from unittest.mock import sentinel
 
-from testtools.content import text_content
 from twisted.internet import defer, reactor
 
 from maastesting import dev_root, get_testing_timeout
 from maastesting.factory import factory
 from maastesting.testcase import MAASTestCase, MAASTwistedRunTest
 from provisioningserver.rackdservices import lease_socket_service
-from provisioningserver.rackdservices.lease_socket_service import (
-    LeaseSocketService,
+from provisioningserver.rackdservices.testing import (
+    configure_lease_service_for_one_shot,
 )
 from provisioningserver.utils.shell import call_and_check
-from provisioningserver.utils.twisted import DeferredValue
 
 TIMEOUT = get_testing_timeout()
 
@@ -37,15 +36,13 @@ class TestDHCPNotify(MAASTestCase):
 
     def catch_packet_on_socket(self):
         socket_path = self.patch_socket_path()
-        service = LeaseSocketService(sentinel.service, reactor)
-        dv = DeferredValue()
-
-        def mock_processNotification(*args, **kwargs):
-            dv.set(args)
-
-        self.patch(service, "processNotification", mock_processNotification)
-
-        return socket_path, service, dv
+        service = lease_socket_service.LeaseSocketService(
+            sentinel.service, reactor
+        )
+        helper = configure_lease_service_for_one_shot(self, service)
+        # Get to the point of the done callback
+        next(helper)
+        return socket_path, service, helper
 
     @defer.inlineCallbacks
     def test_sends_notification_over_socket_for_processing(self):
@@ -56,36 +53,37 @@ class TestDHCPNotify(MAASTestCase):
         lease_time = random.randint(30, 1000)
         hostname = factory.make_name("host")
 
-        socket_path, service, done = self.catch_packet_on_socket()
-        service.startService()
-        self.addCleanup(service.stopService)
-        cmd = [
-            f"{dev_root}/package-files/usr/sbin/maas-dhcp-helper",
-            "notify",
-            "--action",
-            action,
-            "--mac",
-            mac,
-            "--ip-family",
-            ip_family,
-            "--ip",
-            ip,
-            "--lease-time",
-            str(lease_time),
-            "--hostname",
-            hostname,
-            "--socket",
-            socket_path,
-        ]
-        self.addDetail("cmd", text_content(repr(cmd)))
-        call_and_check(cmd)
-        yield done.get(timeout=TIMEOUT)
-
-        self.addDetail(
-            "notifications", text_content(repr(service.notifications))
+        socket_path, service, helper = self.catch_packet_on_socket()
+
+        send_notification = partial(
+            call_and_check,
+            [
+                f"{dev_root}/package-files/usr/sbin/maas-dhcp-helper",
+                "notify",
+                "--action",
+                action,
+                "--mac",
+                mac,
+                "--ip-family",
+                ip_family,
+                "--ip",
+                ip,
+                "--lease-time",
+                str(lease_time),
+                "--hostname",
+                hostname,
+                "--socket",
+                socket_path,
+            ],
         )
-        self.assertEqual(1, len(done.value[0]["updates"]))
-        update = done.value[0]["updates"][0]
+
+        # Wait for the service to be done
+        yield next(helper)
+
+        helper.send(send_notification)
+        notifications = yield from helper
+        self.assertEqual(len(notifications), 1)
+        update = notifications[0]
         self.assertEqual(action, update["action"])
         self.assertEqual(mac, update["mac"])
         self.assertEqual(ip_family, update["ip_family"])
diff --git a/src/provisioningserver/rackdservices/testing.py b/src/provisioningserver/rackdservices/testing.py
index 1a5881f..52b4bd7 100644
--- a/src/provisioningserver/rackdservices/testing.py
+++ b/src/provisioningserver/rackdservices/testing.py
@@ -3,12 +3,14 @@
 
 """Testing resources for `provisioningserver.rackdservices`."""
 
+from unittest.mock import MagicMock
 
 from maastesting.factory import factory
 from maastesting.twisted import always_succeed_with
 from provisioningserver import services
 from provisioningserver.rpc import region
 from provisioningserver.rpc.testing import MockLiveClusterToRegionRPCFixture
+from provisioningserver.utils.twisted import pause, retries
 
 
 def prepareRegionForGetControllerType(test, is_region=False, is_rack=True):
@@ -33,3 +35,36 @@ def prepareRegionForGetControllerType(test, is_region=False, is_rack=True):
         return services.getServiceNamed("rpc"), protocol
 
     return connecting.addCallback(connected)
+
+
+def configure_lease_service_for_one_shot(testcase, service):
+    """Helper for getting notifications out of a LeaseSocketService.
+
+    I'm a coroutine that works alongside an @inlineCallbacks test.
+
+    ```
+    coroutine = configure_lease_service_for_one_shot(self, service)
+
+    next(coroutine) # First one initialises
+    yield next(coroutine) # This one is for putting the service.done in the callback chain
+    notifications = yield from coroutine
+    assert notifications is not None
+    ```
+    """
+    service.startService()
+    testcase.addCleanup(service.stopService)
+    # Stop the looping call to check that the notification gets added
+    # to notifications.
+    process_done = service.done
+    service.processor.stop()
+    yield process_done
+    service.processor = MagicMock()
+    create_notification = yield
+    create_notification()
+    # Loop until the notifications has a notification.
+    for elapsed, remaining, wait in retries(5, 0.1, service.reactor):
+        if len(service.notifications) > 0:
+            return service.notifications
+        else:
+            yield pause(wait, service.reactor)
+    assert False, "No notifications found"
diff --git a/src/provisioningserver/rackdservices/tests/test_lease_socket_service.py b/src/provisioningserver/rackdservices/tests/test_lease_socket_service.py
index 4433884..309e551 100644
--- a/src/provisioningserver/rackdservices/tests/test_lease_socket_service.py
+++ b/src/provisioningserver/rackdservices/tests/test_lease_socket_service.py
@@ -4,6 +4,7 @@
 """Tests for src/provisioningserver/rackdservices/lease_socket_service.py"""
 
 
+from functools import partial
 import json
 import os
 import socket
@@ -14,20 +15,20 @@ from testtools.matchers import Not, PathExists
 from twisted.application.service import Service
 from twisted.internet import defer, reactor
 from twisted.internet.protocol import DatagramProtocol
-from twisted.internet.threads import deferToThread
 
 from maastesting import get_testing_timeout
 from maastesting.factory import factory
-from maastesting.matchers import MockCalledOnceWith
 from maastesting.testcase import MAASTestCase, MAASTwistedRunTest
 from provisioningserver.rackdservices import lease_socket_service
 from provisioningserver.rackdservices.lease_socket_service import (
     LeaseSocketService,
 )
+from provisioningserver.rackdservices.testing import (
+    configure_lease_service_for_one_shot,
+)
 from provisioningserver.rpc import clusterservice, getRegionClient
 from provisioningserver.rpc.region import UpdateLeases
 from provisioningserver.rpc.testing import MockLiveClusterToRegionRPCFixture
-from provisioningserver.utils.twisted import DeferredValue, pause, retries
 
 
 class TestLeaseSocketService(MAASTestCase):
@@ -99,82 +100,50 @@ class TestLeaseSocketService(MAASTestCase):
     def test_notification_gets_added_to_notifications(self):
         socket_path = self.patch_socket_path()
         service = LeaseSocketService(sentinel.service, reactor)
-        service.startService()
-        self.addCleanup(service.stopService)
 
-        # Stop the looping call to check that the notification gets added
-        # to notifications.
-        process_done = service.done
-        service.processor.stop()
-        yield process_done
-        service.processor = MagicMock()
-
-        # Create test payload to send.
         packet = self.create_lease_notification()
+        helper = configure_lease_service_for_one_shot(self, service)
+        next(helper)
+        yield next(helper)
+        helper.send(partial(self.send_notification, socket_path, packet))
 
-        # Send notification to the socket should appear in notifications.
-        yield deferToThread(self.send_notification, socket_path, packet)
-
-        # Loop until the notifications has a notification.
-        for elapsed, remaining, wait in retries(5, 0.1, reactor):
-            if len(service.notifications) > 0:
-                break
-            else:
-                yield pause(wait, reactor)
+        notifications = yield from helper
 
-        # Should have one notitication.
-        self.assertEqual([packet], list(service.notifications))
+        # Should have one notification.
+        self.assertItemsEqual([packet], notifications)
 
     @defer.inlineCallbacks
     def test_processNotification_gets_called_with_notification(self):
         socket_path = self.patch_socket_path()
         service = LeaseSocketService(sentinel.service, reactor)
-        dv = DeferredValue()
 
-        # Mock processNotifcation to catch the call.
-        def mock_processNotification(*args, **kwargs):
-            dv.set(args)
-
-        self.patch(service, "processNotification", mock_processNotification)
-
-        # Start the service and stop it at the end of the test.
-        service.startService()
-        self.addCleanup(service.stopService)
+        helper = configure_lease_service_for_one_shot(self, service)
+        next(helper)
+        yield next(helper)
 
         # Create test payload to send.
         packet1 = self.create_lease_notification()
         packet2 = self.create_lease_notification()
-        payload = {
-            "cluster_uuid": None,
-            "updates": [packet1, packet2],
-        }
 
         # Send notification to the socket and wait for notification.
-        yield deferToThread(self.send_notification, socket_path, packet1)
-        yield deferToThread(self.send_notification, socket_path, packet2)
-        yield dv.get(timeout=10)
+        def _send_notifications():
+            self.send_notification(socket_path, packet1)
+            self.send_notification(socket_path, packet2)
+
+        helper.send(_send_notifications)
+
+        notifications = yield from helper
 
-        # Payload should be the argument passed to processNotifcation
-        self.assertEqual((payload,), dv.value)
+        self.assertItemsEqual([packet1, packet2], notifications)
 
     @defer.inlineCallbacks
     def test_processNotification_dont_allow_same_address(self):
         socket_path = self.patch_socket_path()
         service = LeaseSocketService(sentinel.service, reactor)
-        dvs = [DeferredValue(), DeferredValue()]
 
-        # Mock processNotifcation to catch the call.
-        def mock_processNotification(*args, **kwargs):
-            for dv in dvs:
-                if not dv.isSet:
-                    dv.set(args)
-                    break
-
-        self.patch(service, "processNotification", mock_processNotification)
-
-        # Start the service and stop it at the end of the test.
-        service.startService()
-        self.addCleanup(service.stopService)
+        helper = configure_lease_service_for_one_shot(self, service)
+        next(helper)
+        yield next(helper)
 
         # Create test payload to send.
         ip = factory.make_ipv4_address()
@@ -182,31 +151,17 @@ class TestLeaseSocketService(MAASTestCase):
         packet2 = self.create_lease_notification(ip=ip)
 
         # Send notifications to the socket and wait for notifications.
-        yield deferToThread(self.send_notification, socket_path, packet1)
-        yield deferToThread(self.send_notification, socket_path, packet2)
-        yield dvs[0].get(timeout=10)
-        yield dvs[1].get(timeout=10)
+        def _send_notifications():
+            self.send_notification(socket_path, packet1)
+            self.send_notification(socket_path, packet2)
+
+        helper.send(_send_notifications)
+
+        notifications = yield from helper
 
         # Packet should be the argument passed to processNotification in
         # order.
-        self.assertEqual(
-            (
-                {
-                    "cluster_uuid": None,
-                    "updates": [packet1],
-                },
-            ),
-            dvs[0].value,
-        )
-        self.assertEqual(
-            (
-                {
-                    "cluster_uuid": None,
-                    "updates": [packet2],
-                },
-            ),
-            dvs[1].value,
-        )
+        self.assertEqual([packet1, packet2], list(notifications))
 
     @defer.inlineCallbacks
     def test_processNotification_send_to_region(self):
@@ -225,11 +180,6 @@ class TestLeaseSocketService(MAASTestCase):
             "updates": [packet],
         }
         yield service.processNotification(payload, clock=reactor)
-        self.assertThat(
-            protocol.UpdateLeases,
-            MockCalledOnceWith(
-                protocol,
-                cluster_uuid=client.localIdent,
-                updates=[packet],
-            ),
+        protocol.UpdateLeases.assert_called_once_with(
+            protocol, cluster_uuid=client.localIdent, updates=[packet]
         )