[2922] Test implicit unsubscription notifications

Test that notifications for unsubscription are sent even in case a
client disconnects without explicit unsubscription.

src/bin/msgq/msgq.py.in

@@ -197,6 +197,28 @@ class MsgQ:
         # side.
         self.__lock = threading.Lock()
 
+    def members_notify(self, event, params):
+        """
+        Thin wrapper around ccs's notify. Send a notification about change
+        of some list that can be requested by the members command.
+
+        The event is either one of:
+        - connected (client connected to MsgQ)
+        - disconected (client disconnected from MsgQ)
+        - subscribed (client subscribed to a group)
+        - unsubscribed (client unsubscribed from a group)
+
+        The params is dict containing:
+        - client: The lname of the client in question.
+        - group (only the 3rd and 4th): The group the client subscribed
+          or unsubscribed from.
+
+        It is expected to happen after the event (so client subscribing for these
+        notifications gets a notification about itself, but not in the case
+        of unsubscribing).
+        """
+        # Empty for now.
+
     def cfgmgr_ready(self, ready=True):
         """Notify that the config manager is either subscribed, or
            that the msgq is shutting down and it won't connect, but
@@ -339,6 +361,8 @@ class MsgQ:
         else:
             self.add_kqueue_socket(newsocket)
 
+        self.members_notify('connected', {'client': lname})
+
     def kill_socket(self, fd, sock):
         """Fully close down the socket."""
         # Unregister events on the socket.  Note that we don't have to do
@@ -356,6 +380,7 @@ class MsgQ:
         if fd in self.sendbuffs:
             del self.sendbuffs[fd]
         logger.debug(TRACE_BASIC, MSGQ_SOCK_CLOSE, fd)
+        self.members_notify('disconnected', {'client': lname})
 
     def __getbytes(self, fd, sock, length, continued):
         """Get exactly the requested bytes, or raise an exception if
@@ -647,6 +672,12 @@ class MsgQ:
         if group == None or instance == None:
             return  # ignore invalid packets entirely
         self.subs.subscribe(group, instance, sock)
+        lname = self.fd_to_lname[sock.fileno()]
+        self.members_notify('subscribed',
+                            {
+                                'client': lname,
+                                'group': group
+                            })
 
     def process_command_unsubscribe(self, sock, routing, data):
         group = routing[CC_HEADER_GROUP]
@@ -654,6 +685,12 @@ class MsgQ:
         if group == None or instance == None:
             return  # ignore invalid packets entirely
         self.subs.unsubscribe(group, instance, sock)
+        lname = self.fd_to_lname[sock.fileno()]
+        self.members_notify('unsubscribed',
+                            {
+                                'client': lname,
+                                'group': group
+                            })
 
     def run(self):
         """Process messages.  Forever.  Mostly."""

src/bin/msgq/tests/msgq_test.py

@@ -240,10 +240,9 @@ class MsgQTest(unittest.TestCase):
         # Omitting the parameters completely in such case is OK
         check_both(self.__msgq.command_handler('members', None))
 
-    def test_notifies(self):
+    def notifications_setup(self):
         """
-        Test the message queue sends notifications about connecting,
-        disconnecting and subscription changes.
+        Common setup of some notifications tests. Mock several things.
         """
         # Mock the method to send notifications (we don't really want
         # to send them now, just see they'd be sent).
@@ -255,7 +254,7 @@ class MsgQTest(unittest.TestCase):
         class FakePoller:
             def register(self, socket, mode):
                 pass
-            def unregister(self, fd, sock):
+            def unregister(self, sock):
                 pass
         self.__msgq.members_notify = send_notification
         self.__msgq.poller = FakePoller()
@@ -264,7 +263,17 @@ class MsgQTest(unittest.TestCase):
         class Sock:
             def __init__(self, fileno):
                 self.fileno = lambda: fileno
+            def close(self):
+                pass
         sock = Sock(1)
+        return notifications, sock
+
+    def test_notifies(self):
+        """
+        Test the message queue sends notifications about connecting,
+        disconnecting and subscription changes.
+        """
+        notifications, sock = self.notifications_setup()
 
         # We should notify about new cliend when we register it
         self.__msgq.register_socket(sock)
@@ -292,6 +301,28 @@ class MsgQTest(unittest.TestCase):
         self.__msgq.kill_socket(sock.fileno(), sock)
         self.assertEqual([('disconnected', {'client': lname})], notifications)
 
+    def test_notifies_implicit_kill(self):
+        """
+        Test that the unsubscription notifications are sent before the socket
+        is dropped, even in case it does not unsubscribe explicitly.
+        """
+        notifications, sock = self.notifications_setup()
+
+        # Register and subscribe. Notifications for these are in above test.
+        self.__msgq.register_socket(sock)
+        lname = list(self.__msgq.lnames.keys())[0] # Steal the lname
+        self.__msgq.process_command_subscribe(sock, {'group': 'G',
+                                                     'instance': '*'},
+                                              None)
+        notifications.clear()
+
+        self.__msgq.kill_socket(sock.fileno(), sock)
+        # Now, the notification for unsubscribe should be first, second for
+        # the disconnection.
+        self.assertEqual([('unsubscribed', {'client': lname, 'group': 'G'}),
+                          ('disconnected', {'client': lname})
+                         ], notifications)
+
     def test_undeliverable_errors(self):
         """
         Send several packets through the MsgQ and check it generates