expose duplicate and can duplicate function in taipy

taipy/core/scenario/_scenario_manager.py

@@ -523,7 +523,9 @@ def _get_by_config_id(cls, config_id: str, version_number: Optional[str] = None)
         return cls._repository._load_all(filters)
 
     @classmethod
-    def _duplicate(cls, scenario: Scenario, creation_date: Optional[datetime] = None) -> Scenario:
+    def _duplicate(
+        cls, scenario: Scenario, creation_date: Optional[datetime] = None, name: Optional[str] = None
+    ) -> Scenario:
         """
         Clone a scenario.
 
@@ -567,6 +569,10 @@ def _duplicate(cls, scenario: Scenario, creation_date: Optional[datetime] = None
                 dn._parent_ids.update([cloned_scenario.id])
                 _data_manager._set(dn)
 
+        if name:
+            if hasattr(cloned_scenario._properties, "_entity_owner"):
+                del cloned_scenario._properties._entity_owner
+            cloned_scenario._properties["name"] = name
         cloned_scenario._cycle = cycle
         cloned_scenario._creation_date = creation_date
         cloned_scenario._primary_scenario = len(cls._get_all_by_cycle(cycle)) == 0 if cycle else False
@@ -576,13 +582,13 @@ def _duplicate(cls, scenario: Scenario, creation_date: Optional[datetime] = None
         return cloned_scenario
 
     @classmethod
-    def _can_duplicate(cls, scenario: Scenario) -> ReasonCollection:
+    def _can_duplicate(cls, scenario: Optional[Scenario]) -> ReasonCollection:
         reason_collector = ReasonCollection()
 
         if isinstance(scenario, Scenario):
             scenario_id = scenario.id
         else:
-            scenario_id = scenario
+            scenario_id = str(scenario)
 
         if not cls._repository._exists(scenario_id):
             reason_collector._add_reason(scenario_id, EntityDoesNotExist(scenario_id))

taipy/core/taipy.py

@@ -1070,3 +1070,35 @@ def get_entities_by_config_id(
     if entities := _DataManagerFactory._build_manager()._get_by_config_id(config_id):
         return entities
     return entities
+
+
+def can_duplicate(entity: Optional[Scenario] = None) -> ReasonCollection:
+    """Indicate if a scenario can be duplicated.
+
+    Returns:
+        True if the given scenario can be created. False otherwise.
+    """
+    return _ScenarioManagerFactory._build_manager()._can_duplicate(entity)
+
+
+def duplicate_scenario(
+    scenario: Scenario, creation_date: Optional[datetime] = None, name: Optional[str] = None
+) -> Scenario:
+    """Duplicate an existing scenario and return a new scenario.
+
+    This function duplicates the provided scenario, optionally setting a new creation date and name.
+
+    If the scenario belongs to a cycle, the cycle (corresponding to the creation_date and the configuration
+    frequency attribute) is created if it does not exist yet.
+
+    Arguments:
+        scenario (Scenario): The scenario to duplicate.
+        creation_date (Optional[datetime.datetime]): The creation date of the new scenario.
+            If None, the current date and time is used.
+        name (Optional[str]): The displayable name of the new scenario.
+
+    Returns:
+        Scenario: The newly duplicated scenario.
+    """
+
+    return _ScenarioManagerFactory._build_manager()._duplicate(scenario, creation_date, name)

tests/core/scenario/test_scenario_manager.py

@@ -1567,9 +1567,10 @@ def test_clone_scenario():
     assert len(_DataManager._get_all()) == 3
     assert len(_TaskManager._get_all()) == 1
 
-    new_scenario = _ScenarioManager._duplicate(scenario)
+    new_scenario = _ScenarioManager._duplicate(scenario, name="New Scenario")
 
     assert scenario.id != new_scenario.id
+    assert new_scenario.name == "New Scenario"
     assert len(_ScenarioManager._get_all()) == 2
     assert len(_DataManager._get_all()) == 6
     assert len(_TaskManager._get_all()) == 2

tests/core/test_taipy.py

@@ -869,3 +869,29 @@ def test_get_entities_by_config_id_in_multiple_versions_environment(self):
         assert len(tp.get_scenarios()) == 5
         assert len(tp.get_entities_by_config_id(scenario_config_1.id)) == 3
         assert len(tp.get_entities_by_config_id(scenario_config_2.id)) == 2
+
+    def test_can_duplicate(self):
+        dn_config = Config.configure_in_memory_data_node("dn", 10)
+        task_config = Config.configure_task("task", print, [dn_config])
+        scenario_config = Config.configure_scenario("sc", {task_config}, [], Frequency.DAILY)
+
+        scenario = tp.create_scenario(scenario_config)
+        assert tp.can_duplicate(scenario)
+        assert not tp.can_duplicate("1")
+
+    def test_duplicate_scenario(self):
+        dn_config = Config.configure_in_memory_data_node("dn", 10)
+        task_config = Config.configure_task("task", print, [dn_config])
+        scenario_config = Config.configure_scenario("sc", {task_config}, [], Frequency.DAILY)
+
+        scenario = tp.create_scenario(scenario_config)
+
+        with mock.patch("taipy.core.scenario._scenario_manager._ScenarioManager._duplicate") as mck:
+            tp.duplicate_scenario(scenario)
+            mck.assert_called_once_with(scenario, None, None)
+        with mock.patch("taipy.core.scenario._scenario_manager._ScenarioManager._duplicate") as mck:
+            tp.duplicate_scenario(scenario, datetime.datetime(2022, 2, 5))
+            mck.assert_called_once_with(scenario, datetime.datetime(2022, 2, 5), None)
+        with mock.patch("taipy.core.scenario._scenario_manager._ScenarioManager._duplicate") as mck:
+            tp.duplicate_scenario(scenario, datetime.datetime(2022, 2, 5), "displayable_name")
+            mck.assert_called_once_with(scenario, datetime.datetime(2022, 2, 5), "displayable_name")