diff --git a/podman/domain/images_manager.py b/podman/domain/images_manager.py index 23da8667..2d5f2348 100644 --- a/podman/domain/images_manager.py +++ b/podman/domain/images_manager.py @@ -8,6 +8,7 @@ import urllib.parse from typing import Any, Literal, Optional, Union from collections.abc import Iterator, Mapping, Generator +from contextlib import nullcontext from pathlib import Path import requests @@ -166,6 +167,87 @@ 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 sum(x is not None for x in (data, file_path, url)) != 1: + raise PodmanError( + "Exactly 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 = None + if file_path is not None: + post_data_context = Path(file_path).open("rb") + elif data is not None: + post_data_context = io.BytesIO(data) + elif url is not None: + post_data_context = nullcontext() + + # 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 diff --git a/podman/tests/integration/test_images.py b/podman/tests/integration/test_images.py index 24f448c9..f80f0722 100644 --- a/podman/tests/integration/test_images.py +++ b/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,94 @@ 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") + + # Clean up + container.remove(force=True) + image.remove(force=True) + + 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") + + # Clean up + container.remove(force=True) + image.remove(force=True) + + 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 + def handler(*a): + return http.server.SimpleHTTPRequestHandler(*a, directory=tmpdir) + + with http.server.HTTPServer(("", 0), handler) as httpd: + threading.Thread(target=httpd.serve_forever, daemon=True).start() + + # Import it + image = self.client.images.import_image( + url=f"http://localhost:{httpd.server_port}/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") + + # Clean up + container.remove(force=True) + image.remove(force=True) diff --git a/podman/tests/unit/test_imagesmanager.py b/podman/tests/unit/test_imagesmanager.py index 00faa7ac..fd3908db 100644 --- a/podman/tests/unit/test_imagesmanager.py +++ b/podman/tests/unit/test_imagesmanager.py @@ -1,4 +1,6 @@ +import io import types +import urllib.parse import unittest from unittest.mock import patch @@ -44,6 +46,11 @@ "Containers": 0, } +IMPORTED_IMAGE = { + "Id": "b07571c5220ab38f303c854bd150aede9e5d1d1501a34b54f84916e9f5f8d000", + "Digest": "sha256:122ae1d154d021971cd7a739b5d6f029a80b627763b649adeb34c6f13f3c451b", +} + class ImagesManagerTestCase(unittest.TestCase): """Test ImagesManager area of concern. @@ -430,6 +437,88 @@ 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": IMPORTED_IMAGE["Digest"]}, + ) + mock.get( + tests.LIBPOD_URL + + "/images/" + + urllib.parse.quote_plus(IMPORTED_IMAGE["Digest"]) + + "/json", + json=IMPORTED_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, IMPORTED_IMAGE["Id"]) + + mock.post( + tests.LIBPOD_URL + "/images/import", + json={"Id": IMPORTED_IMAGE["Digest"]}, + ) + mock.get( + tests.LIBPOD_URL + + "/images/" + + urllib.parse.quote_plus(IMPORTED_IMAGE["Digest"]) + + "/json", + json=IMPORTED_IMAGE, + ) + + image = self.client.images.import_image(b'This is a weird tarball...') + self.assertIsInstance(image, Image) + + self.assertEqual(image.id, IMPORTED_IMAGE["Id"]) + + mock.post( + tests.LIBPOD_URL + "/images/import", + json={"Id": IMPORTED_IMAGE["Digest"]}, + ) + mock.get( + tests.LIBPOD_URL + + "/images/" + + urllib.parse.quote_plus(IMPORTED_IMAGE["Digest"]) + + "/json", + json=IMPORTED_IMAGE, + ) + + image = self.client.images.import_image(url="http://example.com") + self.assertIsInstance(image, Image) + + self.assertEqual(image.id, IMPORTED_IMAGE["Id"]) + @requests_mock.Mocker() def test_search(self, mock): mock.get(