Merge pull request #989 from sul-dlss/t986-gather-files-task

Updates ftp transmission task to use the gather_files dict.

libsys_airflow/plugins/data_exports/transmission_tasks.py

@@ -77,7 +77,7 @@ def transmit_data_http_task(gather_files, **kwargs) -> dict:
 
 
 @task(multiple_outputs=True)
-def transmit_data_ftp_task(conn_id, local_files) -> dict:
+def transmit_data_ftp_task(conn_id, gather_files) -> dict:
     """
     Transmit the data via ftp
     Returns lists of files successfully transmitted and failures
@@ -87,7 +87,7 @@ def transmit_data_ftp_task(conn_id, local_files) -> dict:
     remote_path = connection.extra_dejson["remote_path"]
     success = []
     failures = []
-    for f in local_files:
+    for f in gather_files["file_list"]:
         remote_file_path = f"{remote_path}/{Path(f).name}"
         try:
             logger.info(f"Start transmission of file {f}")

tests/data_exports/test_transmission_tasks.py

@@ -11,6 +11,7 @@
 from libsys_airflow.plugins.data_exports.transmission_tasks import (
     gather_files_task,
     transmit_data_http_task,
+    transmit_data_ftp_task,
     archive_transmitted_data_task,
 )
 
@@ -60,6 +61,19 @@ def mock_httpx_connection():
     )
 
 
+@pytest.fixture
+def mock_ftphook_connection():
+    return Connection(
+        conn_id="ftp-example.com",
+        conn_type="ftp",
+        host="ftp://www.example.com",
+        login="username",
+        password="pass",
+        extra={"remote_path": "/remote/path/dir"},
+        schema="ftp",
+    )
+
+
 @pytest.fixture
 def mock_httpx_success():
     return httpx.Client(
@@ -88,6 +102,25 @@ def test_gather_full_dump_files(mocker):
     assert marc_files["s3"]
 
 
+def test_transmit_data_ftp_task(
+    mocker, mock_ftphook_connection, mock_marc_files, caplog
+):
+    ftp_hook = mocker.patch(
+        "airflow.providers.ftp.hooks.ftp.FTPHook.store_file", return_value=True
+    )
+    mocker.patch(
+        "libsys_airflow.plugins.data_exports.transmission_tasks.Connection.get_connection_from_secrets",
+        return_value=mock_ftphook_connection,
+    )
+
+    transmit_data = transmit_data_ftp_task.function("ftp-example.com", mock_marc_files)
+    assert len(transmit_data["success"]) == 3
+    assert "Start transmission of file" in caplog.text
+    assert ftp_hook.store_file.called_with(
+        "/remote/path/dir/2024022914.mrc", "2024022914.mrc"
+    )
+
+
 def test_transmit_data_task(
     mocker, mock_httpx_connection, mock_httpx_success, mock_marc_files, caplog
 ):