fix: migration cli failing when theres no pipelines (#959)

* fix: migration cli failing when theres no pipelines

* feat: add core_version attribute

* Update taipy/core/_entity/_migrate/_utils.py

Co-authored-by: Đỗ Trường Giang <do.giang@avaiga.com>

* feat: update ConfigCoreVersionMismatched message with more information on how to fix the problem

---------

Co-authored-by: Đỗ Trường Giang <do.giang@avaiga.com>
Co-authored-by: trgiangdo <dtr.giang.1299@gmail.com>

taipy/core/_entity/_migrate/_utils.py

@@ -10,6 +10,7 @@
 # specific language governing permissions and limitations under the License.
 
 import json
+from importlib.metadata import version
 from typing import Dict, List, Optional, Tuple
 
 from taipy.logger._taipy_logger import _TaipyLogger
@@ -20,10 +21,22 @@ __logger = _TaipyLogger._get_logger()
 def __update_parent_ids(entity: Dict, data: Dict) -> Dict:
     # parent_ids was not present in 2.0, need to be search for in tasks
     parent_ids = entity.get("parent_ids", [])
+    id = entity["id"]
+
     if not parent_ids:
         parent_ids = __search_parent_ids(entity["id"], data)
     entity["parent_ids"] = parent_ids
 
+    if "TASK" in id:
+        parents = []
+        for parent in entity.get("parent_ids", []):
+            if "PIPELINE" in parent:
+                continue
+            parents.append(parent)
+        if entity.get("owner_id") and entity.get("owner_id") not in parents:
+            parents.append(entity.get("owner_id"))
+        entity["parent_ids"] = parents
+
     return entity
 
 
@@ -91,14 +104,15 @@ def __migrate_subscriber(fct_module, fct_name):
 
 
 def __migrate_scenario(scenario: Dict, data: Dict) -> Dict:
-    # pipelines were replaced by tasks
-    scenario["tasks"] = __fetch_tasks_from_pipelines(scenario["pipelines"], data)
+    if "pipelines" in scenario:
+        # pipelines were replaced by tasks
+        scenario["tasks"] = scenario.get("tasks") or __fetch_tasks_from_pipelines(scenario["pipelines"], data)
 
-    # pipeline attribute not removed in 3.0
-    scenario["pipelines"] = None
+        # pipeline attribute not removed in 3.0
+        scenario["pipelines"] = None
 
     # additional_data_nodes attribute added
-    scenario["additional_data_nodes"] = []
+    scenario["additional_data_nodes"] = scenario.get("additional_data_nodes", [])
 
     return scenario
 
@@ -122,12 +136,12 @@ def __migrate_task(task: Dict, data: Dict, is_entity: bool = True) -> Dict:
     if is_entity:
         # parent_id has been renamed to owner_id
         try:
-            task["owner_id"] = task["parent_id"]
+            task["owner_id"] = task.get("owner_id", task["parent_id"])
             del task["parent_id"]
         except KeyError:
             pass
 
-        # properties was not present in 2.0
+        # properties attribute was not present in 2.0
         task["properties"] = task.get("properties", {})
 
     # skippable was not present in 2.0
@@ -182,7 +196,7 @@ def __migrate_datanode(datanode: Dict) -> Dict:
 
     # parent_id has been renamed to owner_id
     try:
-        datanode["owner_id"] = datanode["parent_id"]
+        datanode["owner_id"] = datanode.get("owner_id", datanode["parent_id"])
         del datanode["parent_id"]
     except KeyError:
         pass
@@ -241,6 +255,9 @@ def __migrate_global_config(config: Dict):
     except KeyError:
         pass
 
+    if "core_version" not in config["CORE"] or config["CORE"]["core_version"] != version("taipy-core"):
+        config["CORE"]["core_version"] = version("taipy-core")
+
     return config
 
 
@@ -255,14 +272,16 @@ def __migrate_version(version: Dict) -> Dict:
     config = __migrate_global_config(config)
 
     # replace pipelines for tasks
-    pipelines_section = config["PIPELINE"]
-    for id, content in config["SCENARIO"].items():
-        tasks = []
-        for _pipeline in content["pipelines"]:
-            pipeline_id = _pipeline.split(":")[0]
-            tasks = pipelines_section[pipeline_id]["tasks"]
-        config["SCENARIO"][id]["tasks"] = tasks
-        del config["SCENARIO"][id]["pipelines"]
+    if "PIPELINE" in config:
+        pipelines_section = config["PIPELINE"]
+        for id, content in config["SCENARIO"].items():
+            tasks = []
+            for _pipeline in content["pipelines"]:
+                pipeline_id = _pipeline.split(":")[0]
+                tasks = pipelines_section[pipeline_id]["tasks"]
+            config["SCENARIO"][id]["tasks"] = tasks
+            del config["SCENARIO"][id]["pipelines"]
+        del config["PIPELINE"]
 
     for id, content in config["TASK"].items():
         config["TASK"][id] = __migrate_task_config(content, config)
@@ -270,8 +289,6 @@ def __migrate_version(version: Dict) -> Dict:
     for id, content in config["DATA_NODE"].items():
         config["DATA_NODE"][id] = __migrate_datanode_config(content)
 
-    del config["PIPELINE"]
-
     version["config"] = json.dumps(config, ensure_ascii=False, indent=0)
     return version
 

taipy/core/_version/_version_manager.py

@@ -16,6 +16,7 @@ from taipy.config import Config
 from taipy.config._config_comparator._comparator_result import _ComparatorResult
 from taipy.config.checker.issue_collector import IssueCollector
 from taipy.config.exceptions.exceptions import InconsistentEnvVariableError
+from taipy.core.exceptions.exceptions import ConfigCoreVersionMismatched
 from taipy.logger._taipy_logger import _TaipyLogger
 
 from .._manager._manager import _Manager
@@ -175,6 +176,9 @@ class _VersionManager(_Manager[_Version]):
                 return version.id
         except InconsistentEnvVariableError:  # The version exist but the Config is alternated
             return version_number
+        except ConfigCoreVersionMismatched as e:
+            cls._logger.error(e.message)
+            raise SystemExit() from e
 
         raise NonExistingVersion(version_number)
 

taipy/core/exceptions/exceptions.py

@@ -9,7 +9,6 @@
 # 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 typing import List, Optional
 
 
@@ -18,7 +17,9 @@ class ConfigCoreVersionMismatched(Exception):
 
     def __init__(self, config_core_version: str, core_version: str) -> None:
         self.message = (
-            f"Core version {config_core_version} in Config does not match with version of Taipy Core {core_version}."
+            f"The version {config_core_version} of Core's entities does not match version of Taipy Core {core_version}."
+            f" Please update the core entities to be compatible with Taipy Core {core_version}"
+            " using the `taipy migrate ...` command. For more information, please run `taipy help migrate`"
         )
 
 
@@ -186,7 +187,7 @@ class InvalidSequence(Exception):
 class NonExistingSequence(Exception):
     """Raised if a requested Sequence is not known by the Sequence Manager."""
 
-    def __init__(self, sequence_id: str, scenario_id: str=None):
+    def __init__(self, sequence_id: str, scenario_id: str = None):
         if scenario_id:
             self.message = f"Sequence: {sequence_id} does not exist in scenario {scenario_id}."
         else:

tests/core/_entity/data_sample_migrated/tasks/TASK_evaluate_57cbea3b-332e-47ac-a468-472e4885eb82.json

@@ -2,7 +2,7 @@
 "id": "TASK_evaluate_57cbea3b-332e-47ac-a468-472e4885eb82",
 "owner_id": "SCENARIO_scenario_62d29866-610e-4173-bd22-6a0555e80ff4",
 "parent_ids": [
-"PIPELINE_baseline_08c57938-83e8-4848-aca5-c98a4ca21e58"
+"SCENARIO_scenario_62d29866-610e-4173-bd22-6a0555e80ff4"
 ],
 "config_id": "evaluate",
 "input_ids": [

tests/core/_entity/data_sample_migrated/tasks/TASK_predict_cf8b3c83-5ea7-40eb-b8af-d5240852ff5b.json

@@ -2,7 +2,7 @@
 "id": "TASK_predict_cf8b3c83-5ea7-40eb-b8af-d5240852ff5b",
 "owner_id": "SCENARIO_scenario_62d29866-610e-4173-bd22-6a0555e80ff4",
 "parent_ids": [
-"PIPELINE_baseline_08c57938-83e8-4848-aca5-c98a4ca21e58"
+"SCENARIO_scenario_62d29866-610e-4173-bd22-6a0555e80ff4"
 ],
 "config_id": "predict",
 "input_ids": [

tests/core/_entity/test_migrate_cli.py

@@ -41,7 +41,8 @@ def test_migrate_fs_default(caplog):
     assert "Starting entity migration from '.taipy/' folder" in caplog.text
 
 
-def test_migrate_fs_specified_folder(caplog):
+def test_migrate_fs_specified_folder(caplog, mocker):
+    mocker.patch("taipy.core._entity._migrate._utils.version", return_value="3.1.0")
     _MigrateCLI.create_parser()
 
     # Copy data_sample to .data folder for testing
@@ -57,12 +58,14 @@ def test_migrate_fs_specified_folder(caplog):
 
     # Compare migrated .data folder with data_sample_migrated
     dircmp_result = filecmp.dircmp(data_path, "tests/core/_entity/data_sample_migrated")
+
     assert not dircmp_result.diff_files and not dircmp_result.left_only and not dircmp_result.right_only
     for subdir in dircmp_result.subdirs.values():
         assert not subdir.diff_files and not subdir.left_only and not subdir.right_only
 
 
-def test_migrate_fs_backup_and_remove(caplog):
+def test_migrate_fs_backup_and_remove(caplog, mocker):
+    mocker.patch("taipy.core._entity._migrate._utils.version", return_value="3.1.0")
     _MigrateCLI.create_parser()
 
     # Copy data_sample to .data folder for testing
@@ -95,7 +98,8 @@ def test_migrate_fs_backup_and_remove(caplog):
     assert not os.path.exists(backup_path)
 
 
-def test_migrate_fs_backup_and_restore(caplog):
+def test_migrate_fs_backup_and_restore(caplog, mocker):
+    mocker.patch("taipy.core._entity._migrate._utils.version", return_value="3.1.0")
     _MigrateCLI.create_parser()
 
     # Copy data_sample to .data folder for testing

tests/core/_entity/test_migrate_utils.py

@@ -0,0 +1,31 @@
+# Copyright 2021-2024 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.
+
+import json
+
+from taipy.core._entity._migrate._utils import _migrate
+
+
+def test_migrate_entities_from_version_2(mocker):
+    mocker.patch("taipy.core._entity._migrate._utils.version", return_value="3.1.0")
+    with open("tests/core/_entity/data_to_migrate.json") as file:
+        data = json.load(file)
+
+    with open("tests/core/_entity/expected_data.json") as file:
+        expected_data = json.load(file)
+
+    migrated_data, _ = _migrate(data)
+
+    assert expected_data == migrated_data
+
+    # Execute again on the migrated data to ensure that the migration  works on latest versions
+    migrated_data, _ = _migrate(data)
+    assert expected_data == migrated_data