Merge ~cgrabowski/maas:add_machine.unsubscribe_websocket_endpoint into maas:master

[Christian Grabowski]

add 'unsubscribe' to base WebSocket Handler

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas/job/branch-tester/12932/console
COMMIT: 54344b905972fd3db412178fdd80e8aa32c30248

review: Needs Fixing

[Adam Collard (adam-collard)]

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas/job/branch-tester/12941/console
COMMIT: aaf7e5934669af2479123696954dde257cab7b72

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas/job/branch-tester/12945/console
COMMIT: 3501c22ca277043ddd91d8df3ab81eeea26fe2c1

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 5f990f6d6bea87212185d405cbc44118970435ab

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: d35818db4fc8ee0c600b7e20904884a0f946ddf5

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: f2b9749326b1f54f3de2de4ac24b5ff8452afc8e

review: Approve

[Alexsander de Souza (alexsander-souza)]

The object should continue to be visible after unsubscribe()

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas/job/branch-tester/13024/console
COMMIT: 117c14626846b52d721c3902effa224051a28da1

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas/job/branch-tester/13026/console
COMMIT: 44b0b138ceb23d65d37a75a68bc759b93d3f3ef6

review: Needs Fixing

[Alexsander de Souza (alexsander-souza)]

+1

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: c5c877056548bbf19c5f7396eeca5aba759afb3a

review: Approve

[MAAS Lander (maas-lander)]

LANDING
-b add_machine.unsubscribe_websocket_endpoint lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci.internal:8080/job/maas/job/branch-tester/13030/consoleText

diff --git a/src/maasserver/websockets/base.py b/src/maasserver/websockets/base.py
index 9d4ed77..f93ae0b 100644
--- a/src/maasserver/websockets/base.py
+++ b/src/maasserver/websockets/base.py
@@ -94,13 +94,16 @@ class HandlerOptions:
         "update",
         "delete",
         "set_active",
+        "unsubscribe",
     ]
     handler_name = None
     object_class = None
     queryset = None
     list_queryset = None
     pk = "id"
+    bulk_pk = "ids"
     pk_type = int
+    unsubscribed_pks = set()
     fields = None
     exclude = None
     list_fields = None
@@ -337,6 +340,7 @@ 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):
@@ -461,6 +465,7 @@ 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
@@ -740,7 +745,11 @@ 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})
+        return (
+            self.get_object({self._meta.pk: pk})
+            if pk not in self._meta.unsubscribed_pks
+            else None
+        )
 
     def refetch(self, obj):
         """Refetch an object using the handler queryset.
@@ -750,6 +759,27 @@ 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])
+        elif self._meta.bulk_pk in params:
+            return [self._unsubscribe(pk) for pk in params[self._meta.bulk_pk]]
+        else:
+            raise HandlerValidationError(
+                f"'{self._meta.pk}' or '{self._meta.bulk_pk}' must be provided in params for unsubscribe"
+            )
+
 
 class AdminOnlyMixin(Handler):
     class Meta:
diff --git a/src/maasserver/websockets/handlers/machine.py b/src/maasserver/websockets/handlers/machine.py
index 8c27240..78d9efd 100644
--- a/src/maasserver/websockets/handlers/machine.py
+++ b/src/maasserver/websockets/handlers/machine.py
@@ -161,6 +161,7 @@ class MachineHandler(NodeHandler):
             "update",
             "action",
             "set_active",
+            "unsubscribe",
             "check_power",
             "create_physical",
             "create_vlan",
diff --git a/src/maasserver/websockets/handlers/node.py b/src/maasserver/websockets/handlers/node.py
index 7fd590b..bc3ced5 100644
--- a/src/maasserver/websockets/handlers/node.py
+++ b/src/maasserver/websockets/handlers/node.py
@@ -128,6 +128,7 @@ class NodeHandler(TimestampedModelHandler):
     class Meta:
         abstract = True
         pk = "system_id"
+        bulk_pk = "system_ids"
         pk_type = str
 
     def __init__(self, user, cache, request):
diff --git a/src/maasserver/websockets/tests/test_base.py b/src/maasserver/websockets/tests/test_base.py
index 9d5ea16..f13a6fc 100644
--- a/src/maasserver/websockets/tests/test_base.py
+++ b/src/maasserver/websockets/tests/test_base.py
@@ -59,6 +59,7 @@ class TestHandlerMeta(MAASTestCase):
                         "update",
                         "delete",
                         "set_active",
+                        "unsubscribe",
                     ]
                 ),
                 handler_name=Equals(""),
@@ -981,6 +982,56 @@ 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 TestHandlerTransaction(
     MAASTransactionServerTestCase, FakeNodesHandlerMixin
diff --git a/src/maasserver/websockets/tests/test_protocol.py b/src/maasserver/websockets/tests/test_protocol.py
index 5181adb..b5cbdc7 100644
--- a/src/maasserver/websockets/tests/test_protocol.py
+++ b/src/maasserver/websockets/tests/test_protocol.py
@@ -20,7 +20,10 @@ 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 MAASTransactionServerTestCase
+from maasserver.testing.testcase import (
+    MAASServerTestCase,
+    MAASTransactionServerTestCase,
+)
 from maasserver.utils.orm import transactional
 from maasserver.utils.threads import deferToDatabase
 from maasserver.websockets import protocol as protocol_module
@@ -1000,3 +1003,20 @@ 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)