Merge ~cgrabowski/maas:reverse_unsubscribe_logic into maas:master

[Christian Grabowski]

add back a set of unsubscribed pks to know when to ignore an update

update units to reflect allow-list behavior

reverse unsubscribe endpoint logic to receive an allow-list

[MAAS Lander (maas-lander)]

UNIT TESTS
-b reverse_unsubscribe_logic lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/34/consoleText
COMMIT: 93893ed47c2552b86d8032e92909657784f22bd4

review: Needs Fixing

c36b1b1... by Christian Grabowski on 2022-07-12

remove no longer relevant test

[MAAS Lander (maas-lander)]

UNIT TESTS
-b reverse_unsubscribe_logic lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/36/consoleText
COMMIT: 27d6b8cf2b0a3787969123b5972af1019883986c

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b reverse_unsubscribe_logic lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: c36b1b185951f71429cbc34ccc1be904f99f6d2c

review: Approve

02a30d6... by Christian Grabowski on 2022-07-13

override machine's listen function to only listen for machines in loaded_pks

7cd89f4... by Christian Grabowski on 2022-07-13

safely access active_pk

[MAAS Lander (maas-lander)]

UNIT TESTS
-b reverse_unsubscribe_logic lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: a704ef6445d66fdf0cd6a7511a637a7cd0c17f8c

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b reverse_unsubscribe_logic lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 7cd89f4e08129cb5519836200e7d325a6f6cc7cf

review: Approve

[Alexsander de Souza (alexsander-souza)]

+1

review: Approve

diff --git a/src/maasserver/websockets/base.py b/src/maasserver/websockets/base.py
index 3063777..8787174 100644
--- a/src/maasserver/websockets/base.py
+++ b/src/maasserver/websockets/base.py
@@ -103,7 +103,6 @@ class HandlerOptions:
     pk = "id"
     bulk_pk = "ids"
     pk_type = int
-    unsubscribed_pks = set()
     fields = None
     exclude = None
     list_fields = None
@@ -340,7 +339,6 @@ class Handler(metaclass=HandlerMetaclass):
                 permission = self._meta.view_permission
             if not self.user.has_perm(permission, obj):
                 raise HandlerPermissionError()
-        self._meta.unsubscribed_pks.discard(pk)
         return obj
 
     def get_queryset(self, for_list=False):
@@ -465,7 +463,6 @@ class Handler(metaclass=HandlerMetaclass):
         """Cache all loaded object pks."""
         getpk = attrgetter(self._meta.pk)
         self.cache["loaded_pks"].update(getpk(obj) for obj in objs)
-        [self._meta.unsubscribed_pks.discard(getpk(obj)) for obj in objs]
 
     def _filter(self, qs, action, params):
         """Return a filtered queryset
@@ -812,11 +809,7 @@ class Handler(metaclass=HandlerMetaclass):
         :param action: Action that caused this event.
         :param pk: Id of the object.
         """
-        return (
-            self.get_object({self._meta.pk: pk})
-            if pk not in self._meta.unsubscribed_pks
-            else None
-        )
+        return self.get_object({self._meta.pk: pk})
 
     def refetch(self, obj):
         """Refetch an object using the handler queryset.
@@ -826,22 +819,19 @@ class Handler(metaclass=HandlerMetaclass):
         """
         return self.get_object({self._meta.pk: getattr(obj, self._meta.pk)})
 
-    def _unsubscribe(self, pk):
-        if pk == self.cache.get("active_pk"):
-            del self.cache["active_pk"]
-
-        if pk in self.cache.get("loaded_pks", []):
-            self.cache["loaded_pks"].remove(pk)
-
-        self._meta.unsubscribed_pks.add(pk)
-
-        return pk
-
     def unsubscribe(self, params):
         if self._meta.pk in params:
-            return self._unsubscribe(params[self._meta.pk])
+            pk = params[self._meta.pk]
+            if pk == self.cache.get("active_pk"):
+                del self.cache["active_pk"]
+            self.cache["loaded_pks"] = self.cache["loaded_pks"] - set(pk)
+            return [pk]
         elif self._meta.bulk_pk in params:
-            return [self._unsubscribe(pk) for pk in params[self._meta.bulk_pk]]
+            pks = set(params[self._meta.bulk_pk])
+            if self.cache.get("active_pk") in pks:
+                del self.cache["active_pk"]
+            self.cache["loaded_pks"] = self.cache["loaded_pks"] - pks
+            return pks
         else:
             raise HandlerValidationError(
                 f"'{self._meta.pk}' or '{self._meta.bulk_pk}' must be provided in params for unsubscribe"
diff --git a/src/maasserver/websockets/handlers/machine.py b/src/maasserver/websockets/handlers/machine.py
index 93cd2eb..955c350 100644
--- a/src/maasserver/websockets/handlers/machine.py
+++ b/src/maasserver/websockets/handlers/machine.py
@@ -521,6 +521,18 @@ class MachineHandler(NodeHandler):
                 % (op, " or ".join(status_names))
             )
 
+    def listen(self, channel, action, pk):
+        """Called when the handler listens for events on channels with
+        `Meta.listen_channels`.
+
+        :param channel: Channel event occured on.
+        :param action: Action that caused this event.
+        :param pk: Id of the object.
+        """
+        # if loaded / not unsubscrived, allow listen events
+        if pk in self.cache["loaded_pks"] or pk == self.cache.get("active_pk"):
+            return self.get_object({self._meta.pk: pk})
+
     def update_filesystem(self, params):
         node = self._get_node_or_permission_error(
             params, permission=NodePermission.edit
diff --git a/src/maasserver/websockets/handlers/tests/test_machine.py b/src/maasserver/websockets/handlers/tests/test_machine.py
index 352d4ea..c4d3c6f 100644
--- a/src/maasserver/websockets/handlers/tests/test_machine.py
+++ b/src/maasserver/websockets/handlers/tests/test_machine.py
@@ -6271,3 +6271,74 @@ class TestMachineHandlerFilter(MAASServerTestCase):
                 _assert_value_in(machine.bmc.power_type, "not_pod_type")
                 _assert_value_in(machine.bmc.id, "pod")
                 _assert_value_in(machine.bmc.id, "not_pod")
+
+    def test_unsubscribe_prevents_further_updates_for_pk(self):
+        admin = factory.make_admin()
+        handler = MachineHandler(admin, {}, None)
+        node = factory.make_Node()
+        handler.list({})
+        listen_result = handler.on_listen("machine", "update", node.system_id)
+        self.assertIsNotNone(listen_result)
+        handler.unsubscribe({"system_ids": [node.system_id]})
+        self.assertIsNone(
+            handler.on_listen("machine", "update", node.system_id)
+        )
+        list_result = handler.list({})
+        self.assertEqual(len(list_result), 1)
+
+    def test_unsubscribe_raises_validation_error_with_no_pk(self):
+        admin = factory.make_admin()
+        handler = MachineHandler(admin, {}, None)
+        self.assertRaises(HandlerValidationError, handler.unsubscribe, {})
+
+    def test_read_an_unsubscribed_object_subscribes(self):
+        admin = factory.make_admin()
+        handler = MachineHandler(admin, {}, None)
+        node1 = factory.make_Node()
+        node2 = factory.make_Node()
+        handler.list({})
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node1.system_id)
+        )
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node2.system_id)
+        )
+        handler.unsubscribe({"system_ids": [node2.system_id]})
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node1.system_id)
+        )
+        self.assertIsNone(
+            handler.on_listen("machine", "update", node2.system_id)
+        )
+        self.assertIsNotNone(handler.get({"system_id": node2.system_id}))
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node2.system_id)
+        )
+
+    def test_list_an_unsubscribed_object_subscribes(self):
+        admin = factory.make_admin()
+        handler = MachineHandler(admin, {}, None)
+        node1 = factory.make_Node()
+        node2 = factory.make_Node()
+        handler.list({})
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node1.system_id)
+        )
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node2.system_id)
+        )
+        handler.unsubscribe({"system_ids": [node1.system_id]})
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node2.system_id)
+        )
+        self.assertIsNone(
+            handler.on_listen("machine", "update", node1.system_id)
+        )
+        list_result = handler.list({})
+        self.assertEqual(len(list_result), 2)
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node1.system_id)
+        )
+        self.assertIsNotNone(
+            handler.on_listen("machine", "update", node2.system_id)
+        )
diff --git a/src/maasserver/websockets/tests/test_base.py b/src/maasserver/websockets/tests/test_base.py
index 77f3fd3..e7632b1 100644
--- a/src/maasserver/websockets/tests/test_base.py
+++ b/src/maasserver/websockets/tests/test_base.py
@@ -986,56 +986,6 @@ class TestHandler(MAASServerTestCase, FakeNodesHandlerMixin):
         for idx in range(3):
             self.assertEqual(f"host-{2-idx}", result[idx]["hostname"])
 
-    def test_unsubscribe_prevents_further_updates_for_pk(self):
-        handler = self.make_nodes_handler()
-        node = factory.make_Node()
-        handler._meta.queryset = Node.objects.all()
-        handler._meta.listen_channels = ["node"]
-        handler._meta.bulk_pk = "system_ids"
-        handler._meta.pk = "system_id"
-        listen_result = handler.listen("node", "update", node.system_id)
-        self.assertIsNotNone(listen_result)
-        handler.unsubscribe({"system_ids": [node.system_id]})
-        self.assertIsNone(handler.listen("node", "update", node.system_id))
-        list_result = handler.list({})
-        self.assertEqual(len(list_result), 1)
-
-    def test_unsubscribe_raises_validation_error_with_no_pk(self):
-        handler = self.make_nodes_handler()
-        handler._meta.queryset = Node.objects.all()
-        handler._meta.listen_channels = ["node"]
-        handler._meta.bulk_pk = "system_ids"
-        handler._meta.pk = "system_id"
-        self.assertRaises(HandlerValidationError, handler.unsubscribe, {})
-
-    def test_read_an_unsubscribed_object_subscribes(self):
-        handler = self.make_nodes_handler()
-        node = factory.make_Node()
-        handler._meta.queryset = Node.objects.all()
-        handler._meta.listen_channels = ["node"]
-        handler._meta.bulk_pk = "system_ids"
-        handler._meta.pk = "system_id"
-        self.assertIsNotNone(handler.listen("node", "update", node.system_id))
-        handler.unsubscribe({"system_ids": [node.system_id]})
-        self.assertIsNone(handler.listen("node", "update", node.system_id))
-        self.assertIsNotNone(handler.get({"system_id": node.system_id}))
-        self.assertIsNotNone(handler.listen("node", "update", node.system_id))
-
-    def test_list_an_unsubscribed_object_subscribes(self):
-        handler = self.make_nodes_handler()
-        node = factory.make_Node()
-        handler._meta.queryset = Node.objects.all()
-        handler._meta.listen_channels = ["node"]
-        handler._meta.bulk_pk = "system_ids"
-        handler._meta.pk = "system_id"
-        listen_result = handler.listen("node", "update", node.system_id)
-        self.assertIsNotNone(listen_result)
-        handler.unsubscribe({"system_ids": [node.system_id]})
-        self.assertIsNone(handler.listen("node", "update", node.system_id))
-        list_result = handler.list({})
-        self.assertEqual(len(list_result), 1)
-        self.assertIsNotNone(handler.listen("node", "update", node.system_id))
-
 
 class TestHandlerGrouping(MAASServerTestCase, FakeNodesHandlerMixin):
     def test_group_simple(self):
diff --git a/src/maasserver/websockets/tests/test_protocol.py b/src/maasserver/websockets/tests/test_protocol.py
index b5cbdc7..5181adb 100644
--- a/src/maasserver/websockets/tests/test_protocol.py
+++ b/src/maasserver/websockets/tests/test_protocol.py
@@ -20,10 +20,7 @@ from apiclient.utils import ascii_url
 from maasserver.eventloop import services
 from maasserver.testing.factory import factory as maas_factory
 from maasserver.testing.listener import FakePostgresListenerService
-from maasserver.testing.testcase import (
-    MAASServerTestCase,
-    MAASTransactionServerTestCase,
-)
+from maasserver.testing.testcase import MAASTransactionServerTestCase
 from maasserver.utils.orm import transactional
 from maasserver.utils.threads import deferToDatabase
 from maasserver.websockets import protocol as protocol_module
@@ -1003,20 +1000,3 @@ class TestWebSocketFactoryTransactional(
                 controller.system_id,
             ),
         )
-
-
-class TestWebSocketFactoryServer(MAASServerTestCase, MakeProtocolFactoryMixin):
-    def test_processNotify_unsubscribed_object(self):
-        factory = self.make_factory()
-        machine = maas_factory.make_Machine()
-        handler = MachineHandler(maas_factory.make_User(), {}, None)
-        factory.handlers["machine"] = handler
-        result = factory.processNotify(
-            handler, "machine", "update", machine.system_id
-        )
-        self.assertIsNotNone(result)
-        handler.unsubscribe({"system_ids": [machine.system_id]})
-        result = factory.processNotify(
-            handler, "machine", "update", machine.system_id
-        )
-        self.assertIsNone(result)