From 7003774dadf1f1369863ef79fe52e9bd3ae5ade6 Mon Sep 17 00:00:00 2001 From: Nicola Sella Date: Tue, 17 Feb 2026 17:53:33 +0100 Subject: [PATCH 1/2] Implement client.quadlets.install Install quadlet passing them to the install function as as a tuples (name, content) or a file path or a list of both. Fixes: https://issues.redhat.com/browse/RUN-4367 Signed-off-by: Nicola Sella --- podman/api/client.py | 7 + podman/domain/quadlets.py | 113 ++++++- podman/tests/unit/test_quadletsmanager.py | 367 ++++++++++++++++++++++ 3 files changed, 486 insertions(+), 1 deletion(-) diff --git a/podman/api/client.py b/podman/api/client.py index c670c3db..d341e2c9 100644 --- a/podman/api/client.py +++ b/podman/api/client.py @@ -317,6 +317,7 @@ def post( *, params: _Params = None, data: _Data = None, + files: Optional[Any] = None, headers: Optional[Mapping[str, str]] = None, timeout: _Timeout = None, stream: Optional[bool] = False, @@ -327,6 +328,8 @@ def post( Args: path: Relative path to RESTful resource. data: HTTP body for operation + files: Dictionary or list of tuples for multipart file upload. + Follows the ``requests`` library ``files`` parameter format. params: Optional parameters to include with URL. headers: Optional headers to include in request. timeout: Number of seconds to wait on request, or (connect timeout, read timeout) tuple @@ -344,6 +347,7 @@ def post( path=path, params=params, data=data, + files=files, headers=headers, timeout=timeout, stream=stream, @@ -394,6 +398,7 @@ def _request( path: Union[str, bytes], *, data: _Data = None, + files: Optional[Any] = None, params: _Params = None, headers: Optional[Mapping[str, str]] = None, timeout: _Timeout = None, @@ -406,6 +411,7 @@ def _request( method: HTTP method to use for request path: Relative path to RESTful resource. params: Optional parameters to include with URL. + files: Dictionary or list of tuples for multipart file upload. headers: Optional headers to include in request. timeout: Number of seconds to wait on request, or (connect timeout, read timeout) tuple @@ -445,6 +451,7 @@ def _request( uri.geturl(), params=params, data=data, + files=files, headers=(headers or {}), stream=stream, verify=kwargs.get("verify", None), diff --git a/podman/domain/quadlets.py b/podman/domain/quadlets.py index 0f38ca40..b8c3cde5 100644 --- a/podman/domain/quadlets.py +++ b/podman/domain/quadlets.py @@ -2,7 +2,9 @@ import builtins import logging -from typing import Optional, Union +import os +import pathlib +from typing import Any, Optional, Union import requests @@ -223,3 +225,112 @@ def delete( response.raise_for_status() return response.json()["Removed"] + + def install( + self, + files: Union[ + "QuadletFileItem", + builtins.list["QuadletFileItem"], + ], + *, + replace: bool = False, + reload_systemd: bool = True, + ) -> dict[str, Any]: + """Install a Quadlet file and additional asset files. + + The function will make a single request. Each request must contain exactly + one quadlet file (identified by its extension: .container, .volume, + .network, ... etc.) and may optionally include asset files such as + Containerfiles, kube YAML, or other configuration files. + + Quadlets and asset files can be provided as a tuple (filename, content) + or a string/path that represents a file path. Both can be combined in a + list to be included as part of the same request. + + The path to a single ``.tar`` file can also be provided; it will be posted + directly with ``Content-Type: application/x-tar``. In all other cases the + items are uploaded as ``multipart/form-data``. + + Args: + files (Union[QuadletFileItem, list[QuadletFileItem]]): File(s) to install. + QuadletFileItem is a Union[tuple[str, str], str, os.PathLike]. + replace (bool): Replace existing files if they already exist. + Defaults to False. + reload_systemd (bool): Reload systemd after installing quadlets. + Defaults to True. + + Returns: + A dict with keys: + - ``InstalledQuadlets``: mapping of source path to installed + path for successfully installed files. + - ``QuadletErrors``: mapping of source path to error message + for failed installations (empty on success). + + Raises: + APIError: when the service reports an error (e.g. no quadlet files + found, multiple quadlet files, file already exists without + replace, or other server errors). + FileNotFoundError: when a provided file path does not exist on + disk. + """ + if not isinstance(files, builtins.list): + files = [files] + + params = { + "replace": replace, + "reload-systemd": reload_systemd, + } + + first = files[0] + if len(files) == 1 and isinstance(first, (str, os.PathLike)) and self._is_tar_path(first): + tar_path = pathlib.Path(first) + if not tar_path.is_file(): + raise FileNotFoundError(f"No such file: '{tar_path}'") + response = self.client.post( + "/quadlets", + params=params, + data=tar_path.read_bytes(), + headers={"Content-Type": "application/x-tar"}, + ) + else: + multipart = self._prepare_install_body(files) + response = self.client.post( + "/quadlets", + params=params, + files=multipart, + ) + + response.raise_for_status() + return response.json() + + @staticmethod + def _is_tar_path(item: "QuadletFileItem") -> bool: + """Return True if *item* looks like a path to a ``.tar`` archive.""" + return isinstance(item, (str, os.PathLike)) and str(item).endswith(".tar") + + def _prepare_install_body( + self, + items: builtins.list["QuadletFileItem"], + ) -> dict[str, tuple[str, bytes]]: + """Build a ``files`` dict for :pymethod:`requests.Session.request`. + + Returns a dictionary of ``{field_name: (filename, file_bytes)}`` + suitable for passing as the ``files`` keyword argument to + :pymethod:`requests.Session.request`, which encodes them as + ``multipart/form-data``. + """ + result: dict[str, tuple[str, bytes]] = {} + for item in items: + if isinstance(item, tuple): + filename, content = item + result[filename] = (filename, content.encode("utf-8")) + elif isinstance(item, (str, os.PathLike)): + fp = pathlib.Path(item) + if not fp.is_file(): + raise FileNotFoundError(f"No such file: '{fp}'") + result[fp.name] = (fp.name, fp.read_bytes()) + return result + + +# Type alias – importable for type annotations in calling code. +QuadletFileItem = Union[tuple[str, str], str, os.PathLike] diff --git a/podman/tests/unit/test_quadletsmanager.py b/podman/tests/unit/test_quadletsmanager.py index 9ab10099..16ec24ad 100644 --- a/podman/tests/unit/test_quadletsmanager.py +++ b/podman/tests/unit/test_quadletsmanager.py @@ -1,5 +1,12 @@ """Unit tests for QuadletsManager.""" +import os +import pathlib +import tarfile +import tempfile +from email.message import Message +from email.parser import BytesParser + import unittest from unittest.mock import patch @@ -9,6 +16,34 @@ from podman.domain.quadlets import Quadlet, QuadletsManager from podman.errors import APIError, NotFound, PodmanError +INSTALL_REPORT = { + "InstalledQuadlets": { + "test.container": "/home/user/.config/containers/systemd/test.container", + }, + "QuadletErrors": {}, +} + + +def _parse_multipart_files(request) -> dict[str, bytes]: + """Parse a multipart/form-data request into ``{filename: content_bytes}``.""" + content_type = request.headers["Content-Type"] + body = request.body + if not isinstance(body, bytes): + body = body.read() if hasattr(body, "read") else body.encode() + raw = b"Content-Type: " + content_type.encode() + b"\r\n\r\n" + body + msg = BytesParser().parsebytes(raw) + result: dict[str, bytes] = {} + if msg.is_multipart(): + for part in msg.get_payload(): + if not isinstance(part, Message): + continue + filename = part.get_filename() + payload = part.get_payload(decode=True) + if filename and isinstance(payload, bytes): + result[filename] = payload + return result + + FIRST_QUADLET = { "Name": "myapp.container", "UnitName": "myapp.service", @@ -329,6 +364,333 @@ def test_delete_running_quadlet_error(self, mock): f"Expected error about running quadlet or force, got: {error_message}", ) + # -- install: single items ------------------------------------------------- + + @requests_mock.Mocker() + def test_install_single_str_path(self, mock): + """Test installing a single quadlet file by path (str).""" + adapter = mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_path = os.path.join(tmpdir, "test.container") + with open(quadlet_path, "w") as f: + f.write("[Container]\nImage=alpine\n") + + result = self.client.quadlets.install(quadlet_path) + + self.assertEqual(result, INSTALL_REPORT) + self.assertIn("test.container", result["InstalledQuadlets"]) + self.assertEqual(result["QuadletErrors"], {}) + self.assertTrue(adapter.called_once) + + @requests_mock.Mocker() + def test_install_single_pathlib_path(self, mock): + """Test installing a single quadlet file by path (pathlib.Path).""" + adapter = mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_path = pathlib.Path(tmpdir) / "test.container" + quadlet_path.write_text("[Container]\nImage=alpine\n") + + result = self.client.quadlets.install(quadlet_path) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(sorted(uploaded), ["test.container"]) + self.assertEqual(uploaded["test.container"], b"[Container]\nImage=alpine\n") + self.assertEqual(result, INSTALL_REPORT) + self.assertTrue(adapter.called_once) + + @requests_mock.Mocker() + def test_install_single_tuple(self, mock): + """Test installing a single quadlet file by tuple (filename, content).""" + adapter = mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + content = "[Container]\nImage=alpine\n" + result = self.client.quadlets.install(("test.container", content)) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(sorted(uploaded), ["test.container"]) + self.assertEqual(uploaded["test.container"], content.encode()) + self.assertEqual(result, INSTALL_REPORT) + self.assertTrue(adapter.called_once) + + @requests_mock.Mocker() + def test_install_list_of_tuples(self, mock): + """Test installing a list of (filename, content) tuples.""" + adapter = mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + items = [ + ("myapp.container", "[Container]\nImage=alpine\n"), + ("Containerfile", "FROM alpine\nCMD echo hello\n"), + ] + self.client.quadlets.install(items) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertIn("myapp.container", uploaded) + self.assertIn("Containerfile", uploaded) + self.assertEqual(uploaded["myapp.container"], b"[Container]\nImage=alpine\n") + self.assertTrue(adapter.called_once) + + # -- install: tar passthrough ------------------------------------------ + + @requests_mock.Mocker() + def test_install_single_tar_by_string_path(self, mock): + """Test that a single .tar file by string path is sent directly.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_file = os.path.join(tmpdir, "test.container") + with open(quadlet_file, "w") as f: + f.write("[Container]\nImage=alpine\n") + + tar_path = os.path.join(tmpdir, "quadlet.tar") + with tarfile.open(tar_path, "w") as tar: + tar.add(quadlet_file, arcname="test.container") + + original_bytes = open(tar_path, "rb").read() + self.client.quadlets.install(tar_path) + + self.assertEqual(mock.last_request.body, original_bytes) + self.assertEqual(mock.last_request.headers["Content-Type"], "application/x-tar") + + @requests_mock.Mocker() + def test_install_single_tar_by_pathlib_path(self, mock): + """Test that a single .tar file by pathlib.Path is sent directly.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_file = os.path.join(tmpdir, "test.container") + with open(quadlet_file, "w") as f: + f.write("[Container]\nImage=alpine\n") + + tar_path = pathlib.Path(tmpdir) / "quadlet.tar" + with tarfile.open(str(tar_path), "w") as tar: + tar.add(quadlet_file, arcname="test.container") + + original_bytes = tar_path.read_bytes() + self.client.quadlets.install(tar_path) + + self.assertEqual(mock.last_request.body, original_bytes) + self.assertEqual(mock.last_request.headers["Content-Type"], "application/x-tar") + + def test_install_single_tar_not_found(self): + """Test install raises FileNotFoundError for nonexistent .tar path.""" + with self.assertRaises(FileNotFoundError): + self.client.quadlets.install("/nonexistent/path/archive.tar") + + # -- install: lists (homogeneous) ----------------------------------------- + + @requests_mock.Mocker() + def test_install_list_of_str_paths(self, mock): + """Test installing a list of string paths.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + file_a = os.path.join(tmpdir, "a.container") + with open(file_a, "w") as f: + f.write("[Container]\nImage=alpine\n") + + file_b = os.path.join(tmpdir, "Containerfile") + with open(file_b, "w") as f: + f.write("FROM alpine\n") + + self.client.quadlets.install([file_a, file_b]) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(sorted(uploaded), ["Containerfile", "a.container"]) + self.assertEqual(uploaded["a.container"], b"[Container]\nImage=alpine\n") + self.assertEqual(uploaded["Containerfile"], b"FROM alpine\n") + + @requests_mock.Mocker() + def test_install_list_of_pathlib_paths(self, mock): + """Test installing a list of pathlib.Path objects.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + file_a = pathlib.Path(tmpdir) / "a.container" + file_a.write_text("[Container]\nImage=alpine\n") + + file_b = pathlib.Path(tmpdir) / "app.yml" + file_b.write_text("services:\n web:\n image: alpine\n") + + self.client.quadlets.install([file_a, file_b]) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(sorted(uploaded), ["a.container", "app.yml"]) + self.assertEqual(uploaded["app.yml"], b"services:\n web:\n image: alpine\n") + + # -- install: mixed lists (paths, tuples) -------------------------------- + + @requests_mock.Mocker() + def test_install_mixed_str_path_and_tuple(self, mock): + """Test mixed list: string path + content tuple.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_path = os.path.join(tmpdir, "myapp.container") + with open(quadlet_path, "w") as f: + f.write("[Container]\nImage=alpine\n") + + items = [ + quadlet_path, + ("extra-config.yml", "key: value\n"), + ] + self.client.quadlets.install(items) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(sorted(uploaded), ["extra-config.yml", "myapp.container"]) + self.assertEqual(uploaded["myapp.container"], b"[Container]\nImage=alpine\n") + self.assertEqual(uploaded["extra-config.yml"], b"key: value\n") + + @requests_mock.Mocker() + def test_install_mixed_pathlib_and_tuple(self, mock): + """Test mixed list: pathlib.Path + content tuple.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_path = pathlib.Path(tmpdir) / "myapp.container" + quadlet_path.write_text("[Container]\nImage=alpine\n") + + items = [ + quadlet_path, + ("Containerfile", "FROM alpine\nCMD echo hello\n"), + ] + self.client.quadlets.install(items) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(sorted(uploaded), ["Containerfile", "myapp.container"]) + self.assertEqual(uploaded["myapp.container"], b"[Container]\nImage=alpine\n") + self.assertEqual(uploaded["Containerfile"], b"FROM alpine\nCMD echo hello\n") + + @requests_mock.Mocker() + def test_install_mixed_str_and_pathlib_paths(self, mock): + """Test mixed list: string path + pathlib.Path.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + file_a = os.path.join(tmpdir, "a.container") + with open(file_a, "w") as f: + f.write("[Container]\nImage=alpine\n") + + file_b = pathlib.Path(tmpdir) / "Containerfile" + file_b.write_text("FROM alpine\n") + + self.client.quadlets.install([file_a, file_b]) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(sorted(uploaded), ["Containerfile", "a.container"]) + self.assertEqual(uploaded["a.container"], b"[Container]\nImage=alpine\n") + self.assertEqual(uploaded["Containerfile"], b"FROM alpine\n") + + @requests_mock.Mocker() + def test_install_mixed_all_three_types(self, mock): + """Test mixed list with all three item types: str path, pathlib.Path, tuple.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + str_path = os.path.join(tmpdir, "myapp.container") + with open(str_path, "w") as f: + f.write("[Container]\nImage=alpine\n") + + pathlib_path = pathlib.Path(tmpdir) / "app.yml" + pathlib_path.write_text("services:\n web:\n image: alpine\n") + + items = [ + str_path, + pathlib_path, + ("Containerfile", "FROM alpine\nCMD echo hello\n"), + ] + self.client.quadlets.install(items) + + uploaded = _parse_multipart_files(mock.last_request) + self.assertEqual(len(uploaded), 3) + self.assertEqual( + sorted(uploaded), + ["Containerfile", "app.yml", "myapp.container"], + ) + self.assertEqual(uploaded["myapp.container"], b"[Container]\nImage=alpine\n") + self.assertEqual(uploaded["app.yml"], b"services:\n web:\n image: alpine\n") + self.assertEqual(uploaded["Containerfile"], b"FROM alpine\nCMD echo hello\n") + + def test_install_mixed_list_path_not_found(self): + """Test that FileNotFoundError is raised when one path in a list does not exist.""" + with tempfile.TemporaryDirectory() as tmpdir: + good_path = os.path.join(tmpdir, "good.container") + with open(good_path, "w") as f: + f.write("[Container]\nImage=alpine\n") + + items = [ + good_path, + "/nonexistent/bad.yml", + ("Containerfile", "FROM alpine\n"), + ] + with self.assertRaises(FileNotFoundError) as ctx: + self.client.quadlets.install(items) + + self.assertIn("bad.yml", str(ctx.exception)) + + # -- install: query params & error handling ---------------------------- + + @requests_mock.Mocker() + def test_install_with_replace(self, mock): + """Test install sends replace=True as query parameter.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_path = os.path.join(tmpdir, "test.container") + with open(quadlet_path, "w") as f: + f.write("[Container]\nImage=alpine\n") + + self.client.quadlets.install(quadlet_path, replace=True) + + self.assertEqual(mock.last_request.qs["replace"], ["true"]) + + @requests_mock.Mocker() + def test_install_with_reload_systemd_false(self, mock): + """Test install sends reload-systemd=False as query parameter.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_path = os.path.join(tmpdir, "test.container") + with open(quadlet_path, "w") as f: + f.write("[Container]\nImage=alpine\n") + + self.client.quadlets.install(quadlet_path, reload_systemd=False) + + self.assertEqual(mock.last_request.qs["reload-systemd"], ["false"]) + + @requests_mock.Mocker() + def test_install_default_params(self, mock): + """Test install sends correct default query parameters.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_path = os.path.join(tmpdir, "test.container") + with open(quadlet_path, "w") as f: + f.write("[Container]\nImage=alpine\n") + + self.client.quadlets.install(quadlet_path) + + self.assertEqual(mock.last_request.qs["replace"], ["false"]) + self.assertEqual(mock.last_request.qs["reload-systemd"], ["true"]) + self.assertTrue(mock.last_request.headers["Content-Type"].startswith("multipart/form-data")) + + @requests_mock.Mocker() + def test_install_already_exists(self, mock): + """Test install raises APIError when quadlet already exists.""" + mock.post( + tests.LIBPOD_URL + "/quadlets", + json={ + "cause": "a Quadlet with name test.container already exists, refusing to overwrite", + "message": "a Quadlet with name test.container already exists, " + "refusing to overwrite", + }, + status_code=400, + ) + + with self.assertRaises(APIError): + self.client.quadlets.install(("test.container", "[Container]\nImage=alpine\n")) + @requests_mock.Mocker() def test_delete_running_quadlet_with_force_succeeds(self, mock): """Test delete with force=True succeeds for running quadlet.""" @@ -376,6 +738,11 @@ def test_delete_all_with_partial_errors(self, mock): self.assertIn("success1.container", result) self.assertIn("success2.container", result) + def test_install_single_file_not_found(self): + """Test install raises FileNotFoundError for nonexistent file path.""" + with self.assertRaises(FileNotFoundError): + self.client.quadlets.install("/nonexistent/path/test.container") + if __name__ == '__main__': unittest.main() From 2b885704f08762fd4cb3418a0904bdc4f6bb8d4c Mon Sep 17 00:00:00 2001 From: Nicola Sella Date: Wed, 11 Mar 2026 13:44:17 +0100 Subject: [PATCH 2/2] fix comments about tar and bytes Signed-off-by: Nicola Sella --- podman/domain/quadlets.py | 14 ++++++++++---- podman/tests/unit/test_quadletsmanager.py | 20 ++++++++++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/podman/domain/quadlets.py b/podman/domain/quadlets.py index b8c3cde5..745deb90 100644 --- a/podman/domain/quadlets.py +++ b/podman/domain/quadlets.py @@ -305,8 +305,11 @@ def install( @staticmethod def _is_tar_path(item: "QuadletFileItem") -> bool: - """Return True if *item* looks like a path to a ``.tar`` archive.""" - return isinstance(item, (str, os.PathLike)) and str(item).endswith(".tar") + """Return True if *item* looks like a path to a tar archive.""" + if not isinstance(item, (str, os.PathLike)): + return False + name = str(item) + return name.endswith(".tar") or name.endswith(".tar.gz") def _prepare_install_body( self, @@ -323,7 +326,10 @@ def _prepare_install_body( for item in items: if isinstance(item, tuple): filename, content = item - result[filename] = (filename, content.encode("utf-8")) + if isinstance(content, bytes): + result[filename] = (filename, content) + else: + result[filename] = (filename, content.encode("utf-8")) elif isinstance(item, (str, os.PathLike)): fp = pathlib.Path(item) if not fp.is_file(): @@ -333,4 +339,4 @@ def _prepare_install_body( # Type alias – importable for type annotations in calling code. -QuadletFileItem = Union[tuple[str, str], str, os.PathLike] +QuadletFileItem = Union[tuple[str, Union[str, bytes]], str, os.PathLike] diff --git a/podman/tests/unit/test_quadletsmanager.py b/podman/tests/unit/test_quadletsmanager.py index 16ec24ad..9ffeea54 100644 --- a/podman/tests/unit/test_quadletsmanager.py +++ b/podman/tests/unit/test_quadletsmanager.py @@ -473,6 +473,26 @@ def test_install_single_tar_by_pathlib_path(self, mock): self.assertEqual(mock.last_request.body, original_bytes) self.assertEqual(mock.last_request.headers["Content-Type"], "application/x-tar") + @requests_mock.Mocker() + def test_install_single_tar_gz_by_string_path(self, mock): + """Test that a single .tar.gz file by string path is sent directly.""" + mock.post(tests.LIBPOD_URL + "/quadlets", json=INSTALL_REPORT, status_code=200) + + with tempfile.TemporaryDirectory() as tmpdir: + quadlet_file = os.path.join(tmpdir, "test.container") + with open(quadlet_file, "w") as f: + f.write("[Container]\nImage=alpine\n") + + tar_path = os.path.join(tmpdir, "quadlet.tar.gz") + with tarfile.open(tar_path, "w:gz") as tar: + tar.add(quadlet_file, arcname="test.container") + + original_bytes = open(tar_path, "rb").read() + self.client.quadlets.install(tar_path) + + self.assertEqual(mock.last_request.body, original_bytes) + self.assertEqual(mock.last_request.headers["Content-Type"], "application/x-tar") + def test_install_single_tar_not_found(self): """Test install raises FileNotFoundError for nonexistent .tar path.""" with self.assertRaises(FileNotFoundError):