Fix bug in FlyteDirectory.listdir on local files

flytekit/types/directory/types.py

@@ -367,10 +367,11 @@ def listdir(cls, directory: FlyteDirectory) -> typing.List[typing.Union[FlyteDir
         file_access = FlyteContextManager.current_context().file_access
         if not file_access.is_remote(final_path):
             for p in os.listdir(final_path):
-                if os.path.isfile(os.path.join(final_path, p)):
-                    paths.append(FlyteFile(p))
+                joined_path = os.path.join(final_path, p) 
+                if os.path.isfile(joined_path):
+                    paths.append(FlyteFile(joined_path))
                 else:
-                    paths.append(FlyteDirectory(p))
+                    paths.append(FlyteDirectory(joined_path))
             return paths
 
         def create_downloader(_remote_path: str, _local_path: str, is_multipart: bool):

tests/flytekit/unit/types/directory/test_listdir.py

@@ -0,0 +1,31 @@
+import tempfile
+from pathlib import Path
+
+from flytekit import FlyteDirectory, FlyteFile, map_task, task, workflow
+
+def test_listdir():
+    @task
+    def setup() -> FlyteDirectory:
+        tmpdir = Path(tempfile.mkdtemp())
+        (tmpdir / "file.txt").write_text("Hello, World!")
+        return FlyteDirectory(tmpdir)
+
+
+    @task
+    def read_file(file: FlyteFile) -> str:
+        with open(file, "r") as f:
+            return f.read()
+
+
+    @task
+    def list_dir(dir: FlyteDirectory) -> list[FlyteFile]:
+        return FlyteDirectory.listdir(dir)
+
+
+    @workflow
+    def wf() -> list[str]:
+        tmpdir = setup()
+        files = list_dir(dir=tmpdir)
+        return map_task(read_file)(file=files)
+
+    assert wf() == ["Hello, World!"]