process events under authorization - Issue #662 - Backport #2540

taipy/gui_core/_context.py

@@ -104,17 +104,15 @@ def __init__(self, gui: Gui) -> None:
         self.start()
 
     def process_event(self, event: Event):
-        if event.entity_type == EventEntityType.SCENARIO:
-            with self.gui._get_autorization(system=True):
+        with self.gui._get_autorization(system=True):
+            if event.entity_type == EventEntityType.SCENARIO:
                 self.scenario_refresh(
                     event.entity_id
                     if event.operation != EventOperation.DELETION and is_readable(t.cast(ScenarioId, event.entity_id))
                     else None
                 )
-        elif event.entity_type == EventEntityType.SEQUENCE and event.entity_id:
-            sequence = None
-            try:
-                with self.gui._get_autorization(system=True):
+            elif event.entity_type == EventEntityType.SEQUENCE and event.entity_id:
+                try:
                     sequence = (
                         core_get(event.entity_id)
                         if event.operation != EventOperation.DELETION
@@ -126,20 +124,20 @@ def process_event(self, event: Event):
                             _GuiCoreContext._CORE_CHANGED_NAME,
                             {"scenario": [x for x in sequence.parent_ids]},  # type: ignore
                         )
-            except Exception as e:
-                _warn(f"Access to sequence {event.entity_id} failed", e)
-        elif event.entity_type == EventEntityType.JOB:
-            with self.lock:
-                self.jobs_list = None
-        elif event.entity_type == EventEntityType.SUBMISSION:
-            self.submission_status_callback(event.entity_id)
-        elif event.entity_type == EventEntityType.DATA_NODE:
-            with self.lock:
-                self.data_nodes_by_owner = None
-            self.gui._broadcast(
-                _GuiCoreContext._CORE_CHANGED_NAME,
-                {"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 == EventEntityType.JOB:
+                with self.lock:
+                    self.jobs_list = None
+            elif event.entity_type == EventEntityType.SUBMISSION:
+                self.submission_status_callback(event.entity_id)
+            elif event.entity_type == EventEntityType.DATA_NODE:
+                with self.lock:
+                    self.data_nodes_by_owner = None
+                self.gui._broadcast(
+                    _GuiCoreContext._CORE_CHANGED_NAME,
+                    {"datanode": event.entity_id if event.operation != EventOperation.DELETION else True},
+                )
 
     def scenario_refresh(self, scenario_id: t.Optional[str]):
         with self.lock:

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:
+            with patch("taipy.gui.gui.Gui._get_autorization") 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"})
+
+    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:
+            with patch("taipy.gui.gui.Gui._get_autorization") 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})

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_autorization") 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:
+            with patch("taipy.gui.gui.Gui._get_autorization") 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_not_called()
+                assert gui_core_context.jobs_list is None
+

tests/gui_core/test_context_process_event_scenario.py

@@ -0,0 +1,61 @@
+# 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:
+            with patch("taipy.gui_core._context.is_readable") as mock_is_readable:
+                with patch("taipy.gui.gui.Gui._get_autorization") 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})
+                    else:
+                        mock_broadcast.assert_called_once_with("core_changed", {"scenario": True})
+
+    @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:
+            with patch("taipy.gui_core._context.is_readable") as mock_is_readable:
+                with patch("taipy.gui.gui.Gui._get_autorization") 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": True})
+

tests/gui_core/test_context_process_event_sequence.py

@@ -0,0 +1,85 @@
+# 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:
+            with patch("taipy.gui_core._context.is_readable") as mock_is_readable:
+                with patch("taipy.gui_core._context.core_get") as mock_core_get:
+                    with patch("taipy.gui.gui.Gui._get_autorization") 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)})
+                        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:
+            with patch("taipy.gui_core._context.is_readable") as mock_is_readable:
+                with patch("taipy.gui.gui.Gui._get_autorization") 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:
+            with patch("taipy.gui_core._context.is_readable") as mock_is_readable:
+                with patch("taipy.gui_core._context.core_get") as mock_core_get:
+                    with patch("taipy.gui.gui.Gui._get_autorization") 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()