Fix issue about not publishing data node updates (#2316)

* Fix linter

* update PR checklist fixing typos.

* Update .github/PULL_REQUEST_TEMPLATE.md

* rename _do_write as the name was already taken by children class

* rename _do_write as the name was already taken by children class

* fix failing tests

* Add unit tests

* Make ruff happy

* Update .github/PULL_REQUEST_TEMPLATE.md

Co-authored-by: Đỗ Trường Giang <[email protected]>

* use pytest.parametrize to factorize code.

---------

Co-authored-by: Đỗ Trường Giang <[email protected]>

.github/PULL_REQUEST_TEMPLATE.md

@@ -32,9 +32,9 @@ _Describe which projects this change will impact and that needs to be backported
 ## Checklist
 _We encourage you to keep the code coverage percentage at 80% and above._
 
-- [ ] Does this solution meet the acceptance criteria of the related issue?
-- [ ] Is the related issue checklist completed?
-- [ ] Does this PR adds unit tests for the developed code? If not, why?
-- [ ] End-to-End tests have been added or updated?
-- [ ] Was the documentation updated, or a dedicated issue for documentation created? (If applicable)
-- [ ] Is the release notes updated? (If applicable)
+- [ ] This solution meets the acceptance criteria of the related issue.
+- [ ] The related issue checklist is completed.
+- [ ] This PR adds unit tests for the developed code. If not, why?
+- [ ] End-to-End tests have been added or updated. If not, why?
+- [ ] The documentation has been updated, or a dedicated issue created. (If applicable)
+- [ ] The release notes have been updated? (If applicable)

taipy/core/_orchestrator/_dispatcher/_job_dispatcher.py

@@ -159,4 +159,7 @@ def _update_job_status(job: Job, exceptions):
                 _TaipyLogger._get_logger().error(st)
             _JobManagerFactory._build_manager()._set(job)
         else:
+            for output in job.task.output.values():
+                output.track_edit(job_id=job.id)
+                output.unlock_edit()
             job.completed()

taipy/core/_orchestrator/_dispatcher/_task_function_wrapper.py

@@ -65,7 +65,7 @@ def _write_data(self, outputs: List[DataNode], results, job_id: JobId):
                 for res, dn in zip(_results, outputs):
                     try:
                         data_node = data_manager._get(dn.id)
-                        data_node.write(res, job_id=job_id)
+                        data_node._write(res)
                     except Exception as e:
                         logger.error("Error during write", exc_info=1)
                         exceptions.append(DataNodeWritingError(f"Error writing in datanode id {dn.id}: {e}"))

taipy/core/data/data_node.py

@@ -461,8 +461,6 @@ def write(self,
             **kwargs (Any): Extra information to attach to the edit document
                 corresponding to this write.
         """
-        from ._data_manager_factory import _DataManagerFactory
-
         if (editor_id
             and self.edit_in_progress
             and self.editor_id != editor_id
@@ -471,6 +469,8 @@ def write(self,
         self._write(data)
         self.track_edit(job_id=job_id, editor_id=editor_id, comment=comment, **kwargs)
         self.unlock_edit()
+        from ._data_manager_factory import _DataManagerFactory
+
         _DataManagerFactory._build_manager()._set(self)
 
     def track_edit(self,

tests/core/_orchestrator/_dispatcher/test_dispatcher__update_job_status.py

@@ -10,19 +10,25 @@
 # specific language governing permissions and limitations under the License.
 import traceback
 
-from taipy import Job, JobId, Status, Task
+from taipy import Job, JobId, Scope, Status, Task
 from taipy.core._orchestrator._dispatcher import _JobDispatcher
 from taipy.core._orchestrator._orchestrator_factory import _OrchestratorFactory
+from taipy.core.data import InMemoryDataNode
+from taipy.core.data.data_node_id import EDIT_JOB_ID_KEY, EDIT_TIMESTAMP_KEY
 from taipy.core.job._job_manager_factory import _JobManagerFactory
 from taipy.core.task._task_manager_factory import _TaskManagerFactory
 
 
 def nothing(*args):
-    pass
+    return 42
 
 
+def _error():
+    raise RuntimeError("Something bad has happened")
+
 def test_update_job_status_no_exception():
-    task = Task("config_id", {}, nothing)
+    output = InMemoryDataNode("data_node", scope=Scope.SCENARIO)
+    task = Task("config_id",  {}, nothing, output=[output])
     _TaskManagerFactory._build_manager()._set(task)
     job = Job(JobId("id"), task, "s_id", task.id)
     _JobManagerFactory._build_manager()._set(job)
@@ -31,6 +37,14 @@ def test_update_job_status_no_exception():
 
     assert job.status == Status.COMPLETED
     assert job.stacktrace == []
+    assert len(output.edits) == 1
+    assert len(output.edits[0]) == 2
+    assert output.edits[0][EDIT_JOB_ID_KEY] == job.id
+    assert output.edits[0][EDIT_TIMESTAMP_KEY] is not None
+    assert output.last_edit_date is not None
+    assert output.editor_id is None
+    assert output.editor_expiration_date is None
+    assert not output.edit_in_progress
 
 
 def test_update_job_status_with_one_exception():

tests/core/job/test_job.py

@@ -73,9 +73,12 @@ def job(task, job_id):
 @pytest.fixture
 def replace_in_memory_write_fct():
     default_write = InMemoryDataNode.write
+    default__write = InMemoryDataNode._write
     InMemoryDataNode.write = _error
+    InMemoryDataNode._write = _error
     yield
     InMemoryDataNode.write = default_write
+    InMemoryDataNode._write = default__write
 
 
 def _foo():

tests/core/notification/test_events_published.py

@@ -12,6 +12,9 @@
 from queue import SimpleQueue
 from typing import Any, Dict, List
 
+import pytest
+
+from taipy import Orchestrator
 from taipy.common.config import Config, Frequency
 from taipy.core import taipy as tp
 from taipy.core.job.status import Status
@@ -154,8 +157,10 @@ def test_events_published_for_writing_dn():
     assert snapshot.operation_collected.get(EventOperation.UPDATE, 0) == 5
     all_evts.stop()
 
-
-def test_events_published_for_scenario_submission():
+@pytest.mark.parametrize("standalone", [False, True])
+def test_events_published_for_scenario_submission(standalone):
+    if standalone:
+        Config.configure_job_executions(mode="standalone", max_nb_of_workers=2)
     input_config = Config.configure_data_node("the_input")
     output_config = Config.configure_data_node("the_output")
     task_config = Config.configure_task("the_task", identity, input=input_config, output=output_config)
@@ -176,7 +181,11 @@ def test_events_published_for_scenario_submission():
     # 1 submission update event for jobs
     # 3 submission update events (for status: PENDING, RUNNING and COMPLETED)
     # 1 submission update event for is_completed
-    scenario.submit()
+    if standalone:
+        Orchestrator().run()
+        scenario.submit(wait=True)
+    else:
+        scenario.submit()
     snapshot = all_evts.capture()
     assert len(snapshot.collected_events) == 18
     assert snapshot.entity_type_collected.get(EventEntityType.CYCLE, 0) == 0