Add support for export and import commands

podman/domain/volumes.py

@@ -4,6 +4,7 @@
 from typing import Any, Literal, Optional, Union
 
 import requests
+import pathlib
 
 from podman import api
 from podman.domain.manager import Manager, PodmanResource
@@ -173,3 +174,49 @@ def remove(self, name: Union[Volume, str], force: Optional[bool] = None) -> None
             name = name.name
         response = self.client.delete(f"/volumes/{name}", params={"force": force})
         response.raise_for_status()
+
+    def export_archive(self, name: Union[Volume, str]) -> bytes:
+        """Export a podman volume, returns the exported archive as bytes.
+
+        Args:
+            name: Identifier for Volume to be exported.
+
+        Raises:
+            APIError: when service reports an error
+        """
+        if isinstance(name, Volume):
+            name = name.name
+        response = self.client.get(f"/volumes/{name}/export")
+        response.raise_for_status()
+        return response._content
+
+    def import_archive(
+        self, name: Union[Volume, str], data: Optional[bytes] = None, path: Optional[str] = None
+    ):
+        """Import a podman volume from tar.
+        The podman volume archive must be provided either as bytes or as a path to the archive.
+
+        Args:
+            name: Identifier for Volume to be imported.
+            data: Uncompressed tar archive as bytes.
+            path: Path to uncompressed tar archive.
+
+        Raises:
+            APIError: when service reports an error
+        """
+        if isinstance(name, Volume):
+            name = name.name
+
+        if data is None and path is None:
+            raise RuntimeError("Either data or path must be provided !")
+        elif data is not None and path is not None:
+            raise RuntimeError("Data and path must not be set at the same time !")
+
+        if data is None:
+            file = pathlib.Path(path)
+            if not file.exists():
+                raise RuntimeError(f"Archive {path} does not exist !")
+            data = file.read_bytes()
+
+        response = self.client.post(f"/volumes/{name}/import", data=data)
+        response.raise_for_status()

podman/tests/integration/test_volumes.py

@@ -4,6 +4,7 @@
 from podman import PodmanClient
 from podman.errors import NotFound
 from podman.tests.integration import base
+from podman.tests.utils import PODMAN_VERSION
 
 
 class VolumesIntegrationTest(base.IntegrationTest):
@@ -35,6 +36,13 @@ def test_volume_crud(self):
             names = [i.name for i in report]
             self.assertIn(volume_name, names)
 
+        if PODMAN_VERSION >= (5, 6, 0):
+            with self.subTest("Export"):
+                archive_bytes = self.client.volumes.export_archive(volume_name)
+
+            with self.subTest("Import"):
+                self.client.volumes.import_archive(volume_name, archive_bytes)
+
         with self.subTest("Remove"):
             self.client.volumes.remove(volume_name, force=True)
             with self.assertRaises(NotFound):

podman/tests/unit/test_volumesmanager.py

@@ -137,6 +137,28 @@ def test_prune(self, mock):
             actual, {"VolumesDeleted": ["dbase", "source"], "SpaceReclaimed": 2048}
         )
 
+    @requests_mock.Mocker()
+    def test_import(self, mock):
+        mock.post(
+            tests.LIBPOD_URL + "/volumes/dbase/import",
+            status_code=requests.codes.no_content,
+        )
+        actual = self.client.volumes.import_archive("dbase", data=b'mocked_archive')
+        self.assertIsInstance(actual, type(None))
+
+        with self.assertRaises(RuntimeError):
+            # The archive does not exist
+            self.client.volumes.import_archive("dbase", path="/path/to/archive.tar")
+
+    @requests_mock.Mocker()
+    def test_export(self, mock):
+        mock.get(
+            tests.LIBPOD_URL + "/volumes/dbase/export",
+            content=b'exported_archive',
+        )
+        actual = self.client.volumes.export_archive("dbase")
+        self.assertIsInstance(actual, bytes)
+
 
 if __name__ == '__main__':
     unittest.main()