process events under authorization. (#2540)

* process events under authorization.
* Add unit tests
* fix linter

taipy/gui_core/_context.py

@@ -116,17 +116,15 @@ class _GuiCoreContext(CoreEventConsumerBase):
 
     def process_event(self, event: Event):
         self.__lazy_start()
-        if event.entity_type is EventEntityType.SCENARIO:
-            with self.gui._get_authorization(system=True): # type: ignore
+        with self.gui._get_authorization(system=True): # type: ignore
+            if event.entity_type is EventEntityType.SCENARIO:
                 self.scenario_refresh(
                     event.entity_id
                     if event.operation is EventOperation.DELETION or is_readable(t.cast(ScenarioId, event.entity_id))
                     else None
                 )
-        elif event.entity_type is EventEntityType.SEQUENCE and event.entity_id:
-            sequence = None
-            try:
-                with self.gui._get_authorization(system=True): # type: ignore
+            elif event.entity_type is EventEntityType.SEQUENCE and event.entity_id:
+                try:
                     sequence = (
                         core_get(event.entity_id)
                         if event.operation is not EventOperation.DELETION
@@ -135,22 +133,22 @@ class _GuiCoreContext(CoreEventConsumerBase):
                     )
                     if sequence and hasattr(sequence, "parent_ids") and sequence.parent_ids:  # type: ignore
                         self.broadcast_core_changed({"scenario": list(sequence.parent_ids)})  # type: ignore
-            except Exception as e:
-                _warn(f"Access to sequence {event.entity_id} failed", e)
-        elif event.entity_type is EventEntityType.JOB:
-            with self.lock:
-                self.jobs_list = None
-            # no broadcast because the submission status will do the job
-            if event.operation is EventOperation.DELETION:
-                self.broadcast_core_changed({"jobs": True})
-        elif event.entity_type is EventEntityType.SUBMISSION:
-            self.submission_status_callback(event.entity_id, event)
-        elif event.entity_type is EventEntityType.DATA_NODE:
-            with self.lock:
-                self.data_nodes_by_owner = None
-            self.broadcast_core_changed(
-                {"datanode": event.entity_id if event.operation != EventOperation.DELETION else True}
-            )
+                except Exception as e:
+                    _warn(f"Access to sequence '{event.entity_id}' failed", e)
+            elif event.entity_type is EventEntityType.JOB:
+                with self.lock:
+                    self.jobs_list = None
+                # no broadcast because the submission status will do the job
+                if event.operation is EventOperation.DELETION:
+                    self.broadcast_core_changed({"jobs": True})
+            elif event.entity_type is EventEntityType.SUBMISSION:
+                self.submission_status_callback(event.entity_id, event)
+            elif event.entity_type is EventEntityType.DATA_NODE:
+                with self.lock:
+                    self.data_nodes_by_owner = None
+                self.broadcast_core_changed(
+                    {"datanode": event.entity_id if event.operation != EventOperation.DELETION else True}
+                )
 
     def broadcast_core_changed(self, payload: t.Dict[str, t.Any], client_id: t.Optional[str] = None):
         self.gui._broadcast(_GuiCoreContext._CORE_CHANGED_NAME, payload, client_id) # type: ignore

tests/gui_core/test_context_process_event_datanode.py

@@ -0,0 +1,56 @@
+# Copyright 2021-2025 Avaiga Private Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+from collections import defaultdict
+from unittest.mock import patch
+
+import pytest
+
+from taipy import DataNode, Gui, Scope
+from taipy.core.notification import Event, EventEntityType, EventOperation
+from taipy.gui_core._context import _GuiCoreContext
+
+
+class TestGuiCoreContextProcessDatanodeEvent:
+
+    @pytest.mark.parametrize("operation", [EventOperation.CREATION, EventOperation.UPDATE])
+    def test_datanode_event(self, operation):
+        event = Event(entity_type=EventEntityType.DATA_NODE,
+                      operation=operation,
+                      entity_id="whatever")
+        gui_core_context = _GuiCoreContext(Gui())
+        gui_core_context.data_nodes_by_owner = defaultdict(list)
+        gui_core_context.data_nodes_by_owner["owner_id"] = [DataNode(config_id="cfg_id", scope=Scope.SCENARIO)]
+        assert len(gui_core_context.data_nodes_by_owner) == 1
+
+        with (patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+              patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth):
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            assert gui_core_context.data_nodes_by_owner is None
+            mock_broadcast.assert_called_once_with("core_changed", {"datanode": "whatever"}, None)
+
+    def test_datanode_deletion_event(self):
+        event = Event(entity_type=EventEntityType.DATA_NODE,
+                      operation=EventOperation.DELETION,
+                      entity_id="whatever")
+        gui_core_context = _GuiCoreContext(Gui())
+        gui_core_context.data_nodes_by_owner = defaultdict(list)
+        gui_core_context.data_nodes_by_owner["owner_id"] = [DataNode(config_id="cfg_id", scope=Scope.SCENARIO)]
+        assert len(gui_core_context.data_nodes_by_owner) == 1
+
+        with (patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+              patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth):
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            assert gui_core_context.data_nodes_by_owner is None
+            mock_broadcast.assert_called_once_with("core_changed", {"datanode": True}, None)

tests/gui_core/test_context_process_event_job.py

@@ -0,0 +1,49 @@
+# Copyright 2021-2025 Avaiga Private Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+from unittest.mock import patch
+
+import pytest
+
+from taipy import Gui
+from taipy.core.notification import Event, EventEntityType, EventOperation
+from taipy.gui_core._context import _GuiCoreContext
+
+
+class TestGuiCoreContextProcessJobEvent:
+
+    @pytest.mark.parametrize("operation", [EventOperation.CREATION, EventOperation.UPDATE])
+    def test_job_event(self, operation):
+        event = Event(entity_type=EventEntityType.JOB,
+                      operation=operation,
+                      entity_id="whatever")
+        gui_core_context = _GuiCoreContext(Gui())
+        gui_core_context.jobs_list = ["job_id"]
+        assert gui_core_context.jobs_list == ["job_id"]
+        with patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth:
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            assert gui_core_context.jobs_list is None
+
+
+    def test_job_deletion(self):
+        event = Event(entity_type=EventEntityType.JOB,
+                      operation=EventOperation.DELETION,
+                      entity_id="job_id")
+        with (patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+              patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth):
+            gui_core_context = _GuiCoreContext(Gui())
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            mock_broadcast.assert_called_once_with("core_changed", {"jobs": True}, None)
+            assert gui_core_context.jobs_list is None
+

tests/gui_core/test_context_process_event_scenario.py

@@ -0,0 +1,65 @@
+# Copyright 2021-2025 Avaiga Private Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+from unittest.mock import patch
+
+import pytest
+
+from taipy import Gui
+from taipy.core.notification import Event, EventEntityType, EventOperation
+from taipy.gui_core._context import _GuiCoreContext
+
+
+class TestGuiCoreContextProcessScenarioEvent:
+
+    @pytest.mark.parametrize("operation, scenario_is_valid", [(EventOperation.CREATION, True),
+                                                              (EventOperation.CREATION, False),
+                                                              (EventOperation.UPDATE, True),
+                                                              (EventOperation.UPDATE, False),
+                                                              (EventOperation.SUBMISSION, True),
+                                                              (EventOperation.SUBMISSION, False),
+                                                              ])
+    def test_scenario_event(self, operation, scenario_is_valid):
+        scenario_id = "scenario_id"
+        event = Event(entity_type=EventEntityType.SCENARIO,
+                      operation=operation,
+                      entity_id=scenario_id)
+        with (
+            patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+            patch("taipy.gui_core._context.is_readable") as mock_is_readable,
+            patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth
+        ):
+            mock_is_readable.return_value = scenario_is_valid
+            gui_core_context = _GuiCoreContext(Gui())
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            if scenario_is_valid:
+                mock_broadcast.assert_called_once_with("core_changed", {"scenario": scenario_id}, None)
+            else:
+                mock_broadcast.assert_called_once_with("core_changed", {"scenario": True}, None)
+
+    @pytest.mark.parametrize("scenario_is_valid", [True, False])
+    def test_scenario_deletion(self, scenario_is_valid):
+        event = Event(entity_type=EventEntityType.SCENARIO,
+                      operation=EventOperation.DELETION,
+                      entity_id="scenario_id")
+        with (
+            patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+            patch("taipy.gui_core._context.is_readable") as mock_is_readable,
+            patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth
+        ):
+            mock_is_readable.return_value = scenario_is_valid
+            gui_core_context = _GuiCoreContext(Gui())
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            mock_broadcast.assert_called_once_with("core_changed", {"scenario": "scenario_id"}, None)
+

tests/gui_core/test_context_process_event_sequence.py

@@ -0,0 +1,90 @@
+# Copyright 2021-2025 Avaiga Private Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+from unittest.mock import patch
+
+import pytest
+
+from taipy import Gui, Sequence, SequenceId
+from taipy.core.notification import Event, EventEntityType, EventOperation
+from taipy.gui_core._context import _GuiCoreContext
+
+
+class TestGuiCoreContextProcessSequenceEvent:
+
+    @pytest.mark.parametrize("operation, sequence_is_valid", [(EventOperation.CREATION, True),
+                                                              (EventOperation.CREATION, False),
+                                                              (EventOperation.UPDATE, True),
+                                                              (EventOperation.UPDATE, False),
+                                                              (EventOperation.SUBMISSION, True),
+                                                              (EventOperation.SUBMISSION, False),
+                                                              ])
+    def test_sequence_event(self, operation, sequence_is_valid):
+        seq_id = "sequence_id"
+        seq_parent_ids = {"a_scenario_id", "s_id"}
+        sequence = Sequence({}, [], SequenceId(seq_id), parent_ids=seq_parent_ids)
+        event = Event(entity_type=EventEntityType.SEQUENCE,
+                      operation=operation,
+                      entity_id=seq_id)
+        with (
+            patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+            patch("taipy.gui_core._context.is_readable") as mock_is_readable,
+            patch("taipy.gui_core._context.core_get") as mock_core_get,
+            patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth
+        ):
+            mock_core_get.return_value = sequence
+            mock_is_readable.return_value = sequence_is_valid
+            gui_core_context = _GuiCoreContext(Gui())
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            if sequence_is_valid:
+                mock_broadcast.assert_called_once_with("core_changed", {"scenario": list(seq_parent_ids)}, None)
+            else:
+                mock_broadcast.assert_not_called()
+
+    @pytest.mark.parametrize("sequence_is_valid", [True, False])
+    def test_sequence_deletion(self, sequence_is_valid):
+        event = Event(entity_type=EventEntityType.SEQUENCE,
+                      operation=EventOperation.DELETION,
+                      entity_id="sequence_id")
+        with (
+            patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+            patch("taipy.gui_core._context.is_readable") as mock_is_readable,
+            patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth
+        ):
+            mock_is_readable.return_value = sequence_is_valid
+            gui_core_context = _GuiCoreContext(Gui())
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            mock_broadcast.assert_not_called()
+
+    @pytest.mark.parametrize("operation", [EventOperation.CREATION, EventOperation.UPDATE, EventOperation.SUBMISSION])
+    def test_sequence_without_parent_ids(self, operation):
+        seq_id = "sequence_id"
+        seq_parent_ids = set()
+        sequence = Sequence({}, [], SequenceId(seq_id), parent_ids=seq_parent_ids)
+        event = Event(entity_type=EventEntityType.SEQUENCE,
+                      operation=operation,
+                      entity_id=seq_id)
+        with (
+            patch("taipy.gui.gui.Gui._broadcast") as mock_broadcast,
+            patch("taipy.gui_core._context.is_readable") as mock_is_readable,
+            patch("taipy.gui_core._context.core_get") as mock_core_get,
+            patch("taipy.gui.gui.Gui._get_authorization") as mock_get_auth
+        ):
+            mock_core_get.return_value = sequence
+            mock_is_readable.return_value = True
+            gui_core_context = _GuiCoreContext(Gui())
+            gui_core_context.process_event(event=event)
+
+            mock_get_auth.assert_called_once_with(system=True)
+            mock_broadcast.assert_not_called()