Fix S3 and HDFSt tests

Author: dolfinus

syncmaster/worker/handlers/file/base.py

@@ -54,14 +54,13 @@ class FileHandler(Handler):
     def _rename_files(self, tmp_path: str) -> None:
         files = self.file_connection.list_dir(tmp_path)
 
-        for index, file_name in enumerate(files):
+        for index, old_path in enumerate(files):
             extension = self._get_file_extension()
-            new_name = self._get_file_name(str(index), extension)
-            old_path = os.path.join(tmp_path, file_name)
+            new_name = self._get_file_name(index, extension)
             new_path = os.path.join(tmp_path, new_name)
             self.file_connection.rename_file(old_path, new_path)
 
-    def _get_file_name(self, index: str, extension: str) -> str:
+    def _get_file_name(self, index: int, extension: str) -> str:
         return self.transfer_dto.file_name_template.format(
             index=index,
             extension=extension,

tests/test_integration/test_run_transfer/test_hdfs.py

@@ -248,7 +248,7 @@ async def test_run_transfer_postgres_to_hdfs_with_full_strategy(
 
     await run_transfer_and_verify(client, group_owner, postgres_to_hdfs.id)
 
-    files = [os.fspath(file) for file in hdfs_file_connection.list_dir(target_path) if file.is_file()]
+    files = [file for file in hdfs_file_connection.list_dir(target_path) if file.is_file()]
     verify_file_name_template(files, expected_extension)
 
     spark.catalog.clearCache()
@@ -304,7 +304,7 @@ async def test_run_transfer_postgres_to_hdfs_with_incremental_strategy(
     fill_with_data(first_transfer_df)
     await run_transfer_and_verify(client, group_owner, postgres_to_hdfs.id)
 
-    files = [os.fspath(file) for file in hdfs_file_connection.list_dir(target_path) if file.is_file()]
+    files = [file for file in hdfs_file_connection.list_dir(target_path) if file.is_file()]
     verify_file_name_template(files, expected_extension)
 
     spark.catalog.clearCache()
@@ -323,7 +323,7 @@ async def test_run_transfer_postgres_to_hdfs_with_incremental_strategy(
     fill_with_data(second_transfer_df)
     await run_transfer_and_verify(client, group_owner, postgres_to_hdfs.id)
 
-    files = [os.fspath(file) for file in hdfs_file_connection.list_dir(target_path) if file.is_file()]
+    files = [file for file in hdfs_file_connection.list_dir(target_path) if file.is_file()]
     verify_file_name_template(files, expected_extension)
 
     spark.catalog.clearCache()

tests/test_integration/test_run_transfer/test_s3.py

@@ -283,7 +283,7 @@ async def test_run_transfer_postgres_to_s3_with_full_strategy(
 
     await run_transfer_and_verify(client, group_owner, postgres_to_s3.id, target_auth="s3")
 
-    files = [os.fspath(file) for file in s3_file_connection.list_dir(target_path)]
+    files = [file for file in s3_file_connection.list_dir(target_path)]
     verify_file_name_template(files, expected_extension)
 
     reader = FileDFReader(
@@ -339,7 +339,7 @@ async def test_run_transfer_postgres_to_s3_with_incremental_strategy(
     fill_with_data(first_transfer_df)
     await run_transfer_and_verify(client, group_owner, postgres_to_s3.id, target_auth="s3")
 
-    files = [os.fspath(file) for file in s3_file_connection.list_dir(target_path)]
+    files = [file for file in s3_file_connection.list_dir(target_path)]
     verify_file_name_template(files, expected_extension)
 
     reader = FileDFReader(
@@ -357,7 +357,7 @@ async def test_run_transfer_postgres_to_s3_with_incremental_strategy(
     fill_with_data(second_transfer_df)
     await run_transfer_and_verify(client, group_owner, postgres_to_s3.id, target_auth="s3")
 
-    files = [os.fspath(file) for file in s3_file_connection.list_dir(target_path)]
+    files = [file for file in s3_file_connection.list_dir(target_path)]
     verify_file_name_template(files, expected_extension)
 
     df_with_increment = reader.run()

tests/utils.py

@@ -12,6 +12,7 @@
 from httpx import AsyncClient
 from onetl.connection import FileConnection
 from onetl.file import FileDownloader, FileUploader
+from onetl.impl import RemoteFile
 from pyspark.sql import DataFrame
 from pyspark.sql.functions import (
     col,
@@ -240,9 +241,9 @@ def add_increment_to_files_and_upload(file_connection: FileConnection, remote_pa
     uploader.run()
 
 
-def verify_file_name_template(files: list, expected_extension: str) -> None:
-    for file_name in files:
-        run_created_at, index_and_extension = file_name.split("-")
+def verify_file_name_template(files: list[RemoteFile], expected_extension: str) -> None:
+    for file in files:
+        run_created_at, index_and_extension = file.name.split("-")
         assert len(run_created_at.split("_")) == 6, f"Got wrong {run_created_at=}"
         assert index_and_extension.split(".", 1)[1] == expected_extension