Fix tests

Signed-off-by: Kevin Su <[email protected]>

flytekit/remote/remote.py

@@ -17,6 +17,7 @@
 from flytekit.clients.friendly import SynchronousFlyteClient
 from flytekit.common import utils as common_utils
 from flytekit.common.exceptions.user import FlyteEntityAlreadyExistsException, FlyteEntityNotExistException
+from flytekit.configuration import internal
 from flytekit.configuration import platform as platform_config
 from flytekit.configuration import sdk as sdk_config
 from flytekit.configuration import set_flyte_config_file
@@ -522,6 +523,7 @@ def _serialize(
                 domain or self.default_domain,
                 version or self.version,
                 self.image_config,
+                env={internal.IMAGE.env_var: self.image_config.default_image.full},
             ),
             entity=entity,
         )
@@ -621,8 +623,6 @@ def _register_entity_if_not_exists(self, entity: WorkflowBase, resolved_identifi
                     raise NotImplementedError(f"We don't support registering this kind of entity: {node.flyte_entity}")
             except FlyteEntityAlreadyExistsException:
                 logging.info(f"{entity.name} already exists")
-            except Exception as e:
-                logging.info(f"Failed to register Flyte entity {entity.name} with error v{e}")
 
     ####################
     # Execute Entities #

tests/flytekit/integration/remote/test_remote.py

@@ -285,15 +285,23 @@ def test_execute_joblib_workflow(flyteclient, flyte_workflows_register, flyte_re
 
 
 def test_execute_with_default_launch_plan(flyteclient, flyte_workflows_register, flyte_remote_env):
-    from mock_flyte_repo.workflows.basic.list_float_wf import my_wf
+    from mock_flyte_repo.workflows.basic.subworkflows import parent_wf
 
     # make sure the task name is the same as the name used during registration
-    my_wf._name = my_wf.name.replace("mock_flyte_repo.", "")
+    parent_wf._name = parent_wf.name.replace("mock_flyte_repo.", "")
+
     remote = FlyteRemote.from_config(PROJECT, "development")
+    execution = remote.execute(parent_wf, {"a": 101}, version=f"v{VERSION}", wait=True)
+    # check node execution inputs and outputs
+    assert execution.node_executions["n0"].inputs == {"a": 101}
+    assert execution.node_executions["n0"].outputs == {"t1_int_output": 103, "c": "world"}
+    assert execution.node_executions["n1"].inputs == {"a": 103}
+    assert execution.node_executions["n1"].outputs == {"o0": "world", "o1": "world"}
 
-    xs: typing.List[float] = [42.24, 999.1, 0.0001]
-    execution = remote.execute(my_wf, inputs={"xs": xs}, version=f"v{VERSION}", wait=True)
-    assert execution.outputs["o0"] == "[42.24, 999.1, 0.0001]"
+    # check subworkflow task execution inputs and outputs
+    subworkflow_node_executions = execution.node_executions["n1"].subworkflow_node_executions
+    subworkflow_node_executions["n1-0-n0"].inputs == {"a": 103}
+    subworkflow_node_executions["n1-0-n1"].outputs == {"t1_int_output": 107, "c": "world"}
 
 
 def test_fetch_not_exist_launch_plan(flyteclient):