Prevent register_launch_plan from re-registering already registered workflow

flytekit/remote/remote.py

@@ -870,7 +870,12 @@ async def _serialize_and_register(
                 loop.run_in_executor(
                     None,
                     functools.partial(
-                        self.raw_register, cp_entity, serialization_settings, version, og_entity=task_entity
+                        self.raw_register,
+                        cp_entity,
+                        serialization_settings,
+                        version,
+                        create_default_launchplan=create_default_launchplan,
+                        og_entity=task_entity,
                     ),
                 )
             )
@@ -1271,16 +1276,36 @@ def register_launch_plan(
                 source_root=project_root,
                 project=project or self.default_project,
                 domain=domain or self.default_domain,
+                version=version,
             )
 
-        ident = run_sync(
-            self._serialize_and_register,
-            entity,
-            serialization_settings,
-            version,
-            options,
-            False,
+        workflow_id = self._resolve_identifier(
+            t=ResourceType.WORKFLOW, name=entity.workflow.name, version=version, ss=serialization_settings
         )
+        try:
+            remote_wf = self.client.get_workflow(workflow_id)
+        except FlyteEntityNotExistException:
+            remote_wf = None
+
+        if remote_wf is None:
+            ident = run_sync(
+                self._serialize_and_register,
+                entity,
+                serialization_settings,
+                version,
+                options,
+                False,
+            )
+        else:
+            launch_plan_model = get_serializable(
+                OrderedDict(), settings=serialization_settings, entity=entity, options=options
+            )
+            ident = self.raw_register(
+                launch_plan_model, serialization_settings, version, create_default_launchplan=False
+            )
+            if ident is None:
+                raise ValueError("Failed to register launch plan")
+
         flp = self.fetch_launch_plan(ident.project, ident.domain, ident.name, ident.version)
         flp.python_interface = entity.python_interface
         return flp

tests/flytekit/unit/remote/test_remote.py

@@ -852,3 +852,43 @@ def workflow1():
     registered_workflow = rr.register_workflow(workflow1, ss)
     assert isinstance(registered_workflow, FlyteWorkflow)
     assert registered_workflow.id == Identifier(ResourceType.WORKFLOW, "flytesnacks", "development", "tests.flytekit.unit.remote.test_remote.workflow1", "dummy_version")
+
+
+@mock.patch("flytekit.remote.remote.get_serializable")
+@mock.patch("flytekit.remote.remote.FlyteRemote.fetch_launch_plan")
+@mock.patch("flytekit.remote.remote.FlyteRemote.raw_register")
+@mock.patch("flytekit.remote.remote.FlyteRemote._serialize_and_register")
+@mock.patch("flytekit.remote.remote.FlyteRemote.client")
+def test_register_launch_plan(mock_client, mock_serialize_and_register, mock_raw_register,mock_fetch_launch_plan, mock_get_serializable):
+    serialization_settings = SerializationSettings(
+        image_config=ImageConfig.auto_default_image(),
+        version="dummy_version",
+    )
+
+    rr = FlyteRemote(
+        Config.for_sandbox(),
+        default_project="flytesnacks",
+        default_domain="development",
+    )
+
+    @task
+    def say_hello() -> str:
+        return "Hello, World!"
+
+    @workflow
+    def hello_world_wf() -> str:
+        res = say_hello()
+        return res
+
+    lp = LaunchPlan.get_or_create(workflow=hello_world_wf, name="additional_lp_for_hello_world", default_inputs={})
+
+    mock_get_serializable.return_value = MagicMock()
+    mock_client.get_workflow.return_value = MagicMock()
+
+    mock_remote_lp = MagicMock()
+    mock_fetch_launch_plan.return_value = mock_remote_lp
+
+    remote_lp = rr.register_launch_plan(lp, version="dummy_version", project="flytesnacks", domain="development", serialization_settings=serialization_settings)
+    assert remote_lp is mock_remote_lp
+    assert not mock_serialize_and_register.called
+    assert mock_raw_register.called