Add rsync_many

pyproject.toml

@@ -101,7 +101,7 @@
   name = "dycw-utilities"
   readme = "README.md"
   requires-python = ">= 3.12"
-  version = "0.174.16"
+  version = "0.174.17"
 
   [project.entry-points.pytest11]
     pytest-randomly = "utilities.pytest_plugins.pytest_randomly"
@@ -135,7 +135,7 @@
   # bump-my-version
   [tool.bumpversion]
     allow_dirty = true
-    current_version = "0.174.16"
+    current_version = "0.174.17"
 
     [[tool.bumpversion.files]]
       filename = "src/utilities/__init__.py"

src/tests/test_subprocess.py

@@ -20,6 +20,8 @@
     ChownCmdError,
     CpError,
     MvFileError,
+    RsyncCmdNoSourcesError,
+    RsyncCmdSourcesNotFoundError,
     apt_install_cmd,
     cat_cmd,
     cd_cmd,
@@ -43,6 +45,7 @@
     rm_cmd,
     rsync,
     rsync_cmd,
+    rsync_many,
     run,
     set_hostname_cmd,
     ssh,
@@ -374,9 +377,12 @@ def test_main(self) -> None:
 class TestRsync:
     @skipif_ci
     @throttle(delta=5 * MINUTE)
-    def test_main(self, *, ssh_user: str, ssh_hostname: str) -> None:
-        with TemporaryFile() as src, yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp:
-            dest = temp / src.name
+    def test_file(self, *, ssh_user: str, ssh_hostname: str) -> None:
+        with (
+            TemporaryFile() as src,
+            yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp_dest,
+        ):
+            dest = temp_dest / src.name
             rsync(src, ssh_user, ssh_hostname, dest)
             ssh(
                 ssh_user,
@@ -387,14 +393,38 @@ def test_main(self, *, ssh_user: str, ssh_hostname: str) -> None:
 
     @skipif_ci
     @throttle(delta=5 * MINUTE)
-    def test_dir(self, *, ssh_user: str, ssh_hostname: str) -> None:
+    def test_dir_without_trailing_slash(
+        self, *, ssh_user: str, ssh_hostname: str
+    ) -> None:
         with (
             TemporaryDirectory() as src,
-            yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp,
+            yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp_dest,
         ):
             (src / "file.txt").touch()
-            dest = temp / src.name
+            name = src.name
+            dest = temp_dest / name
             rsync(src, ssh_user, ssh_hostname, dest)
+            ssh(
+                ssh_user,
+                ssh_hostname,
+                *BASH_LS,
+                input=strip_and_dedent(f"""
+                    if ! [ -d {dest} ]; then exit 1; fi
+                    if ! [ -d {dest}/{name} ]; then exit 1; fi
+                    if ! [ -f {dest}/{name}/file.txt ]; then exit 1; fi
+                """),
+            )
+
+    @skipif_ci
+    @throttle(delta=5 * MINUTE)
+    def test_dir_with_trailing_slash(self, *, ssh_user: str, ssh_hostname: str) -> None:
+        with (
+            TemporaryDirectory() as src,
+            yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp_dest,
+        ):
+            (src / "file.txt").touch()
+            dest = temp_dest / src.name
+            rsync(f"{src}/", ssh_user, ssh_hostname, dest)
             ssh(
                 ssh_user,
                 ssh_hostname,
@@ -407,114 +437,146 @@ def test_dir(self, *, ssh_user: str, ssh_hostname: str) -> None:
 
 
 class TestRsyncCmd:
-    def test_main(self) -> None:
-        result = rsync_cmd("src", "user", "hostname", "dest")
-        expected = [
+    def test_main(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        src.touch()
+        result = rsync_cmd(src, "user", "hostname", "dest")
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--compress",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_multiple_sources(self) -> None:
-        result = rsync_cmd(["src1", "src2"], "user", "hostname", "dest")
-        expected = [
+    def test_multiple_sources(self, *, tmp_path: Path) -> None:
+        src1, src2 = [tmp_path / f"file{i}.txt" for i in [1, 2]]
+        src1.touch()
+        src2.touch()
+        result = rsync_cmd([src1, src2], "user", "hostname", "dest")
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--compress",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src1",
-            "src2",
+            str(src1),
+            str(src2),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_archive(self) -> None:
-        result = rsync_cmd("src", "user", "hostname", "dest", archive=True)
-        expected = [
+    def test_source_with_trailing_slash(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "src"
+        src.mkdir()
+        result = rsync_cmd(f"{src}/", "user", "hostname", "dest")
+        expected: list[str] = [
+            "rsync",
+            "--checksum",
+            "--compress",
+            "--rsh",
+            "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
+            f"{src}/",
+            "user@hostname:dest",
+        ]
+        assert result == expected
+
+    def test_archive(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "src"
+        src.mkdir()
+        result = rsync_cmd(src, "user", "hostname", "dest", archive=True)
+        expected: list[str] = [
             "rsync",
             "--archive",
             "--checksum",
             "--compress",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_chown_user(self) -> None:
-        result = rsync_cmd("src", "user", "hostname", "dest", chown_user="user2")
-        expected = [
+    def test_chown_user(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        src.touch()
+        result = rsync_cmd(src, "user", "hostname", "dest", chown_user="user2")
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--chown",
             "user2",
             "--compress",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_chown_group(self) -> None:
-        result = rsync_cmd("src", "user", "hostname", "dest", chown_group="group")
-        expected = [
+    def test_chown_group(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        src.touch()
+        result = rsync_cmd(src, "user", "hostname", "dest", chown_group="group")
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--chown",
             ":group",
             "--compress",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_chown_user_and_group(self) -> None:
+    def test_chown_user_and_group(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        src.touch()
         result = rsync_cmd(
-            "src", "user", "hostname", "dest", chown_user="user2", chown_group="group"
+            src, "user", "hostname", "dest", chown_user="user2", chown_group="group"
         )
-        expected = [
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--chown",
             "user2:group",
             "--compress",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_exclude(self) -> None:
-        result = rsync_cmd("src", "user", "hostname", "dest", exclude="exclude")
-        expected = [
+    def test_exclude(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        src.touch()
+        result = rsync_cmd(src, "user", "hostname", "dest", exclude="exclude")
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--compress",
             "--exclude",
             "exclude",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_exclude_multiple(self) -> None:
+    def test_exclude_multiple(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        src.touch()
         result = rsync_cmd(
-            "src", "user", "hostname", "dest", exclude=["exclude1", "exclude2"]
+            src, "user", "hostname", "dest", exclude=["exclude1", "exclude2"]
         )
-        expected = [
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--compress",
@@ -524,38 +586,108 @@ def test_exclude_multiple(self) -> None:
             "exclude2",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_sudo(self) -> None:
-        result = rsync_cmd("src", "user", "hostname", "dest", sudo=True)
-        expected = [
+    def test_sudo(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        src.touch()
+        result = rsync_cmd(src, "user", "hostname", "dest", sudo=True)
+        expected: list[str] = [
             "rsync",
             "--checksum",
             "--compress",
             "--rsh",
             "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
             "--rsync-path",
             "sudo rsync",
-            "src",
+            str(src),
             "user@hostname:dest",
         ]
         assert result == expected
 
-    def test_parent(self) -> None:
-        result = rsync_cmd("src", "user", "hostname", "~/dest", parent=True)
-        expected = [
-            "rsync",
-            "--checksum",
-            "--compress",
-            "--rsh",
-            "ssh -o BatchMode=yes -o HostKeyAlgorithms=ssh-ed25519 -o StrictHostKeyChecking=yes -T",
-            "src",
-            "user@hostname:~",
-        ]
-        assert result == expected
+    def test_error_no_sources(self) -> None:
+        with raises(
+            RsyncCmdNoSourcesError,
+            match=r"No sources selected to send to user@hostname:dest",
+        ):
+            _ = rsync_cmd([], "user", "hostname", "dest")
+
+    def test_error_sources_not_found(self, *, tmp_path: Path) -> None:
+        src = tmp_path / "file.txt"
+        with raises(
+            RsyncCmdSourcesNotFoundError,
+            match=r"Sources selected to send to user@hostname:dest but not found: '.*/file\.txt'",
+        ):
+            _ = rsync_cmd(src, "user", "hostname", "dest")
+
+
+class TestRsyncMany:
+    @skipif_ci
+    @throttle(delta=5 * MINUTE)
+    def test_single_file(self, *, ssh_user: str, ssh_hostname: str) -> None:
+        with (
+            TemporaryDirectory() as temp_src,
+            yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp_dest,
+        ):
+            src = temp_src / "file.txt"
+            src.touch()
+            dest = temp_dest / src.name
+            rsync_many(ssh_user, ssh_hostname, (src, dest))
+            ssh(
+                ssh_user,
+                ssh_hostname,
+                *BASH_LS,
+                input=f"if ! [ -f {dest} ]; then exit 1; fi",
+            )
+
+    @skipif_ci
+    @throttle(delta=5 * MINUTE)
+    def test_multiple_files(self, *, ssh_user: str, ssh_hostname: str) -> None:
+        with (
+            TemporaryDirectory() as temp_src,
+            yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp_dest,
+        ):
+            src1, src2 = [temp_src / f"file{i}.txt" for i in [1, 2]]
+            src1.touch()
+            src2.touch()
+            dest1, dest2 = [temp_dest / src.name for src in [src1, src2]]
+            rsync_many(ssh_user, ssh_hostname, (src1, dest1), (src2, dest2))
+            ssh(
+                ssh_user,
+                ssh_hostname,
+                *BASH_LS,
+                input=strip_and_dedent(f"""
+                    if ! [ -f {dest1} ]; then exit 1; fi
+                    if ! [ -f {dest2} ]; then exit 1; fi
+                """),
+            )
+
+    @skipif_ci
+    @throttle(delta=5 * MINUTE)
+    def test_single_directory(self, *, ssh_user: str, ssh_hostname: str) -> None:
+        with (
+            TemporaryDirectory() as temp_src,
+            yield_ssh_temp_dir(ssh_user, ssh_hostname) as temp_dest,
+        ):
+            src = temp_src / "dir"
+            src.mkdir()
+            (src / "file.txt").touch()
+            dest = temp_dest / src.name
+            rsync_many(ssh_user, ssh_hostname, (src, dest))
+            ssh(
+                ssh_user,
+                ssh_hostname,
+                *BASH_LS,
+                input=(
+                    f"""
+                    if ! [ -d {dest} ]; then exit 1; fi
+                    if ! [ -f {dest}/file.txt ]; then exit 1; fi
+                """
+                ),
+            )
 
 
 class TestRun: