Feature/add image import

podman/domain/images_manager.py

@@ -166,6 +166,82 @@ def _generator(body: dict) -> Generator[Image, None, None]:
         # Pass the response body to the generator
         return _generator(response.json())
 
+    def import_image(
+        self,
+        data: Optional[bytes] = None,
+        file_path: Optional[os.PathLike] = None,
+        url: Optional[str] = None,
+        **kwargs,
+    ) -> "Image":
+        """Import a tarball as an image (equivalent of 'podman import').
+
+        Args:
+            file_path: Path to the tarball to import.
+            data: tarball raw data (bytes)
+            url: Url to the tarball to import.
+
+        Keyword Args:
+            reference: Optional reference for the new image (e.g. 'myimage:latest').
+            message: Optional commit message.
+            changes: Optional list of Dockerfile-style instructions
+                     (e.g. ['CMD /bin/bash', 'ENV FOO=bar']).
+
+        Returns:
+            An Image object for the newly imported image.
+
+        Raises:
+            APIError: when service returns an error.
+        """
+        # Check that exactly one of the data or file_path is provided
+        if data is None and file_path is None and url is None:
+            raise PodmanError("The 'data' or 'file_path' or 'url' parameter should be set.")
+
+        if (data and file_path) or (url and data) or (url and file_path):
+            raise PodmanError(
+                "Only one parameter should be set from 'data', 'file_path' and 'url' parameters."
+            )
+
+        # Check if url given it is supported
+        if url:
+            uri = urllib.parse.urlparse(url)
+            if uri.scheme not in api.APIClient.supported_schemes:
+                raise ValueError(
+                    f"The scheme '{uri.scheme}' must be one of {api.APIClient.supported_schemes}"
+                )
+
+        # Set the parameters
+        params = {}
+        if reference := kwargs.get("reference"):
+            params["reference"] = reference
+        if message := kwargs.get("message"):
+            params["message"] = message
+        if changes := kwargs.get("changes"):
+            params["changes"] = changes  # requests sends repeated keys as a list
+        if url:
+            params["url"] = url
+
+        # Get either from file or from raw data
+        post_data_context = Path(file_path).open("rb") if file_path else io.BytesIO(data)
+
+        # Post it
+        image_id = None
+        with post_data_context as post_data:
+            response = self.client.post(
+                "/images/import",
+                params=params,
+                data=post_data,
+                headers={"Content-Type": "application/x-tar"},
+            )
+            response.raise_for_status()
+
+            body = response.json()
+            image_id = body.get("Id")
+
+        if image_id is None:
+            raise APIError(response.url, response=response, explanation="No image id was returned")
+
+        return self.get(image_id)
+
     def prune(
         self,
         all: Optional[bool] = False,  # pylint: disable=redefined-builtin

podman/tests/integration/test_images.py

@@ -17,9 +17,11 @@
 import io
 import os
 import json
+import http.server
 import platform
 import tarfile
 import tempfile
+import threading
 import types
 import unittest
 import random
@@ -292,3 +294,80 @@ def test_scp(self):
             e.exception.explanation,
             r"failed to connect: dial tcp: lookup fake\.ip\.addr.+no such host",
         )
+
+    def test_import_from_file(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Create test folder with test file
+            base_dir = os.path.join(tmpdir, "test")
+            os.mkdir(base_dir)
+            open(os.path.join(base_dir, "foobar"), "w").close()
+
+            # Pack the testfile with the test folder in a tar
+            tar_path = os.path.join(tmpdir, "test.tar.gz")
+            with tarfile.open(tar_path, "w:gz") as tar:
+                tar.add(base_dir, arcname="test")
+
+            # Import it
+            image = self.client.images.import_image(file_path=tar_path, message="test")
+            self.assertIsInstance(image, Image)
+            self.assertEqual(image.attrs.get("Comment"), "test")
+            container = self.client.containers.create(image, command=["."])
+
+            # Check the imported image
+            actual = container.get_archive("./test/foobar")
+            self.assertEqual(len(actual), 2)
+            self.assertEqual(actual[1]["linkTarget"], "/test/foobar")
+
+    def test_import_from_data(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Create test folder with test file
+            base_dir = os.path.join(tmpdir, "test")
+            os.mkdir(base_dir)
+            open(os.path.join(base_dir, "foobar"), "w").close()
+
+            # Pack the testfile with the test folder in a tar buffer
+            tar_buffer = io.BytesIO()
+            with tarfile.open(fileobj=tar_buffer, mode="w:gz") as tar:
+                tar.add(base_dir, arcname="test")
+            tar_buffer.seek(0)
+
+            # Import it
+            image = self.client.images.import_image(data=tar_buffer.read(), changes=["ENV FOO=bar"])
+            self.assertIsInstance(image, Image)
+            self.assertEqual(image.attrs.get("Config", {}).get("Env"), ["FOO=bar"])
+            container = self.client.containers.create(image, command=["."])
+
+            # Check the imported image
+            actual = container.get_archive("./test/foobar")
+            self.assertEqual(len(actual), 2)
+            self.assertEqual(actual[1]["linkTarget"], "/test/foobar")
+
+    def test_import_from_url(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Create test folder with test file
+            base_dir = os.path.join(tmpdir, "test")
+            os.mkdir(base_dir)
+            open(os.path.join(base_dir, "foobar"), "w").close()
+
+            # Pack the testfile with the test folder in a tar
+            tar_path = os.path.join(tmpdir, "test.tar.gz")
+            with tarfile.open(tar_path, "w:gz") as tar:
+                tar.add(base_dir, arcname="test")
+
+            # Serve it on a http server
+            handler = lambda *a: http.server.SimpleHTTPRequestHandler(*a, directory=tmpdir)
+            with http.server.HTTPServer(("", 8000), handler) as httpd:
+                threading.Thread(target=httpd.serve_forever, daemon=True).start()
+
+                # Import it
+                image = self.client.images.import_image(
+                    url="http://localhost:8000/test.tar.gz", message="test"
+                )
+                self.assertIsInstance(image, Image)
+                self.assertEqual(image.attrs.get("Comment"), "test")
+                container = self.client.containers.create(image, command=["."])
+
+                # Check the imported image
+                actual = container.get_archive("./test/foobar")
+                self.assertEqual(len(actual), 2)
+                self.assertEqual(actual[1]["linkTarget"], "/test/foobar")

podman/tests/unit/test_imagesmanager.py

@@ -1,3 +1,4 @@
+import io
 import types
 import unittest
 from unittest.mock import patch
@@ -430,6 +431,86 @@ def test_load(self, mock):
             report[0].id, "sha256:326dd9d7add24646a325e8eaa82125294027db2332e49c5828d96312c5d773ab"
         )
 
+    @requests_mock.Mocker()
+    def test_import(self, mock):
+        # Check for forbidden parameter usage
+        with self.assertRaises(PodmanError):
+            self.client.images.import_image()
+
+        with self.assertRaises(PodmanError):
+            self.client.images.import_image(b'data', "file_path")
+
+        with self.assertRaises(PodmanError):
+            self.client.images.import_image(b'data', "file_path", "url")
+
+        with self.assertRaises(PodmanError):
+            self.client.images.import_image(data=b'data', file_path="file_path")
+
+        with self.assertRaises(PodmanError):
+            self.client.images.import_image(url="url", file_path="file_path")
+
+        with self.assertRaises(PodmanError):
+            self.client.images.import_image(url="url", data=b'data')
+
+        with self.assertRaises(PodmanError):
+            self.client.images.import_image(data=b'data', file_path="file_path", url="url")
+
+        # Check if url is valid
+        with self.assertRaises(ValueError):
+            self.client.images.import_image(url="not-an-url")
+
+        # Patch Path.read_bytes to mock the file reading behavior
+        with patch("pathlib.Path.open", return_value=io.BytesIO(b"mock tarball data")):
+            mock.post(
+                tests.LIBPOD_URL + "/images/import",
+                json={"Id": "quay.io/fedora:latest"},
+            )
+            mock.get(
+                tests.LIBPOD_URL + "/images/quay.io%2ffedora%3Alatest/json",
+                json=FIRST_IMAGE,
+            )
+
+            # 3a. Test the case where only 'file_path' is provided
+            image = self.client.images.import_image(file_path="mock_file.tar")
+            self.assertIsInstance(image, Image)
+
+            self.assertEqual(
+                image.id,
+                "sha256:326dd9d7add24646a325e8eaa82125294027db2332e49c5828d96312c5d773ab",
+            )
+
+        mock.post(
+            tests.LIBPOD_URL + "/images/import",
+            json={"Id": "quay.io/fedora:latest"},
+        )
+        mock.get(
+            tests.LIBPOD_URL + "/images/quay.io%2ffedora%3Alatest/json",
+            json=FIRST_IMAGE,
+        )
+
+        image = self.client.images.import_image(b'This is a weird tarball...')
+        self.assertIsInstance(image, Image)
+
+        self.assertEqual(
+            image.id, "sha256:326dd9d7add24646a325e8eaa82125294027db2332e49c5828d96312c5d773ab"
+        )
+
+        mock.post(
+            tests.LIBPOD_URL + "/images/import",
+            json={"Id": "quay.io/fedora:latest"},
+        )
+        mock.get(
+            tests.LIBPOD_URL + "/images/quay.io%2ffedora%3Alatest/json",
+            json=FIRST_IMAGE,
+        )
+
+        image = self.client.images.import_image(url="http://example.com")
+        self.assertIsInstance(image, Image)
+
+        self.assertEqual(
+            image.id, "sha256:326dd9d7add24646a325e8eaa82125294027db2332e49c5828d96312c5d773ab"
+        )
+
     @requests_mock.Mocker()
     def test_search(self, mock):
         mock.get(