cleaning input path to account for ending '/' as well as 'incoming' (#82)

* cleaning input path to account for ending '/' as well as 'incoming'

* Using pathlib for ensuring destination is /incoming directory

Author: davezuckerman

mokelumne/dags/copy_files.py

@@ -14,6 +14,7 @@
 from mokelumne.util.storage import run_dir
 from mokelumne.util.file_transfer import (
     build_file_manifest,
+    clean_destination_path,
     copy_files_from_manifest,
     load_json,
     save_json,
@@ -64,20 +65,11 @@ def validate_source() -> str:
 
     @task
     def prepare_destination() -> str:
-        """
-        Prepare the destination directory.
-        TODO: one of the requirements that we got from Lynne is that the destination 
-        should be an incoming subdirectory of any directory on PA or DA, 
-        e.g. /srv/pa/aerial/ucb/incoming. i don't have a good answer for how to 
-        approach this, but i'm curious about how you think we should confirm that 1) 
-        we're actually writing to the appropriate incoming subdirectory, and if the 
-        incoming directory exists and has files in it, we should fail the job.
-        """
+        """Prepare the destination directory."""
 
         ctx = get_current_context()
         destination = ctx["params"]["destination"]
-
-        destination_path = Path(destination)
+        destination_path = clean_destination_path(Path(destination))
 
         if not destination_path.exists():
             logger.info("Creating destination directory: %s", destination_path)
@@ -89,7 +81,7 @@ def prepare_destination() -> str:
         if next(os.scandir(destination_path), None) is not None:
             raise AirflowFailException(f"Destination directory contains files: {destination_path}")
 
-        return destination
+        return str(destination_path)
 
     @task
     def build_manifest(source: str) -> str:

mokelumne/util/file_transfer.py

@@ -64,6 +64,12 @@ def load_json(path: Path) -> Manifest:
     with open(path, "r", encoding="utf-8") as json_file:
         return json.load(json_file)
 
+def clean_destination_path(destination_path: Path) -> Path:
+    """Normalize a destination path to end with /incoming."""
+    if destination_path.name.casefold() == "incoming":
+        destination_path = destination_path.parent
+
+    return Path(destination_path / "incoming")
 
 def verify_file_manifest(destination_path: Path, manifest_path: Path) -> list[ManifestEntry]:
     """Verify all copied files exist at the destination."""

test/unit/test_file_transfer.py

@@ -95,6 +95,32 @@ def test_build_manifest_excludes_hidden_and_excluded_files(self, tmp_path: Path)
         assert paths == {"visible.tif"}
         assert len(result["files"]) == 1
 
+    @pytest.mark.parametrize(
+        ("input_path", "expected_path"),
+        [
+            (
+                Path("/srv/pa/aerial/ucb"),
+                Path("/srv/pa/aerial/ucb/incoming"),
+            ),
+            (
+                Path("/srv/pa/aerial/ucb/"),
+                Path("/srv/pa/aerial/ucb/incoming"),
+            ),
+            (
+                Path("/srv/pa/aerial/ucb/incoming"),
+                Path("/srv/pa/aerial/ucb/incoming"),
+            ),
+            (
+                Path("/srv/pa/aerial/ucb/incoming/"),
+                Path("/srv/pa/aerial/ucb/incoming"),
+            ),
+        ],
+    )
+    def test_clean_destination_path(self, input_path: Path, expected_path: Path):
+        """Ensure clean_destination_path normalizes the incoming directory suffix."""
+
+        assert file_transfer.clean_destination_path(input_path) == expected_path
+
     def test_verify_manifest_success(self, tmp_path: Path):
         """Ensure verify_manifest succeeds for valid files."""