From 382d30a74d9c0086aea4c26671fd0080087643fb Mon Sep 17 00:00:00 2001 From: Riushda Date: Sun, 28 Sep 2025 12:37:17 +0200 Subject: [PATCH] Add volume import and export functions Signed-off-by: Riushda fix export and import functions Signed-off-by: Riushda [docs] Update testing section Add common issues section Add docs on how to skip tests based on podman's version and os Signed-off-by: Nicola Sella Signed-off-by: Riushda format Signed-off-by: Riushda Add import and export test Signed-off-by: Riushda Do not test if podman version is older than 5.6.0 Signed-off-by: Riushda Fix docstring Signed-off-by: Riushda Fix export archive command Signed-off-by: Riushda Add path to import function Signed-off-by: Riushda Add default values Signed-off-by: Riushda Fix typing Signed-off-by: Riushda Add mock tests Signed-off-by: Riushda --- podman/domain/volumes.py | 47 ++++++++++++++++++++++++ podman/tests/integration/test_volumes.py | 8 ++++ podman/tests/unit/test_volumesmanager.py | 22 +++++++++++ 3 files changed, 77 insertions(+) diff --git a/podman/domain/volumes.py b/podman/domain/volumes.py index e2daed02..51bd10fa 100644 --- a/podman/domain/volumes.py +++ b/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() diff --git a/podman/tests/integration/test_volumes.py b/podman/tests/integration/test_volumes.py index 34dc19d2..e88d94c7 100644 --- a/podman/tests/integration/test_volumes.py +++ b/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): diff --git a/podman/tests/unit/test_volumesmanager.py b/podman/tests/unit/test_volumesmanager.py index 6650929b..1880287c 100644 --- a/podman/tests/unit/test_volumesmanager.py +++ b/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()