Add a unit test

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

tests/flytekit/unit/remote/requirements.txt

@@ -0,0 +1 @@
+pandas

tests/flytekit/unit/remote/test_remote.py

@@ -15,7 +15,7 @@
 from mock import ANY, MagicMock, patch
 
 import flytekit.configuration
-from flytekit import CronSchedule, LaunchPlan, WorkflowFailurePolicy, task, workflow
+from flytekit import CronSchedule, ImageSpec, LaunchPlan, WorkflowFailurePolicy, task, workflow
 from flytekit.configuration import Config, DefaultImages, Image, ImageConfig, SerializationSettings
 from flytekit.core.base_task import PythonTask
 from flytekit.core.context_manager import FlyteContextManager
@@ -387,6 +387,36 @@ def test_launch_backfill(remote):
     assert wf.workflow_metadata.on_failure == WorkflowFailurePolicy.FAIL_IMMEDIATELY
 
 
+@mock.patch("flytekit.remote.remote.FlyteRemote._version_from_hash")
+@mock.patch("flytekit.remote.remote.FlyteRemote.register_workflow")
+@mock.patch("flytekit.remote.remote.FlyteRemote.upload_file")
+@mock.patch("flytekit.remote.remote.compress_scripts")
+def test_get_image_names(compress_scripts_mock, upload_file_mock, register_workflow_mock, version_from_hash_mock):
+    md5_bytes = bytes([65, 66, 67])
+    compress_scripts_mock.return_value = "compressed"
+    upload_file_mock.return_value = md5_bytes, "localhost:30084"
+
+    image_spec = ImageSpec(requirements="requirements.txt", registry="flyteorg")
+
+    @task(container_image=image_spec)
+    def say_hello(name: str) -> str:
+        return f"hello {name}!"
+
+    @workflow
+    def sub_wf(name: str = "union"):
+        say_hello(name=name)
+
+    @workflow
+    def wf(name: str = "union"):
+        sub_wf(name=name)
+
+    flyte_remote = FlyteRemote(config=Config.auto(), default_project="p1", default_domain="d1")
+    flyte_remote.register_script(wf)
+
+    version_from_hash_mock.assert_called_once_with(md5_bytes, mock.ANY, image_spec.image_name())
+    register_workflow_mock.assert_called_once()
+
+
 @mock.patch("flytekit.remote.remote.FlyteRemote.client")
 def test_local_server(mock_client):
     ctx = FlyteContextManager.current_context()