From 94ade27715d5cac43af455f9762a41429a78dd52 Mon Sep 17 00:00:00 2001 From: Nicola Sella Date: Mon, 8 Jun 2026 18:25:39 +0200 Subject: [PATCH] Fix healthcheck kwargs for container.create() The health_* kwargs were being passed as snake_case in the JSON payload, but /libpod/containers/create endpoint expects nested objects. Assemble kwargs into proper structures: - healthconfig: {Test, Interval, Timeout, Retries, StartPeriod} - startupHealthConfig: {Test, Interval, Timeout, Retries, Successes} - healthLogDestination, healthMaxLogCount, healthMaxLogSize (camelCase) - health_on_failure string mapped to health_check_on_failure_action int Also, it's possible to pass duration values as go-style strings ("30s", "1m", "500ms") and integers as nanoseconds. Fixes: https://redhat.atlassian.net/browse/RHEL-140716 Signed-off-by: Nicola Sella --- podman/api/__init__.py | 4 + podman/api/parse_utils.py | 63 +++++ podman/domain/containers_create.py | 141 ++++++++-- .../integration/test_container_create.py | 47 ++++ podman/tests/unit/test_containersmanager.py | 247 +++++++++++++----- podman/tests/unit/test_parse_utils.py | 110 ++++++++ 6 files changed, 514 insertions(+), 98 deletions(-) diff --git a/podman/api/__init__.py b/podman/api/__init__.py index 77cbfb72..b7eb6ba6 100644 --- a/podman/api/__init__.py +++ b/podman/api/__init__.py @@ -4,8 +4,10 @@ from podman.api.api_versions import VERSION, COMPATIBLE_VERSION from podman.api.http_utils import encode_auth_header, prepare_body, prepare_filters from podman.api.parse_utils import ( + HEALTH_ON_FAILURE_ACTION, decode_header, frames, + prepare_duration_ns, parse_repository, prepare_cidr, prepare_timestamp, @@ -22,11 +24,13 @@ 'APIClient', 'COMPATIBLE_VERSION', 'DEFAULT_CHUNK_SIZE', + 'HEALTH_ON_FAILURE_ACTION', 'VERSION', 'create_tar', 'decode_header', 'encode_auth_header', 'frames', + 'prepare_duration_ns', 'parse_repository', 'prepare_body', 'prepare_cidr', diff --git a/podman/api/parse_utils.py b/podman/api/parse_utils.py index 0294712d..c240d270 100644 --- a/podman/api/parse_utils.py +++ b/podman/api/parse_utils.py @@ -3,6 +3,7 @@ import base64 import ipaddress import json +import re import struct from datetime import datetime, timezone from typing import Any, Optional, Union @@ -11,6 +12,68 @@ from podman.api.client import APIResponse from .output_utils import demux_output +DURATION_UNITS_NS = { + "ns": 1, + "us": 1_000, + "µs": 1_000, + "ms": 1_000_000, + "s": 1_000_000_000, + "m": 60_000_000_000, + "h": 3_600_000_000_000, +} + +HEALTH_ON_FAILURE_ACTION = { + "none": 0, + # value 1 is "invalid" in the Go enum (HealthCheckOnFailureActionInvalid) + # and should not be used directly + "kill": 2, + "restart": 3, + "stop": 4, +} + + +def prepare_duration_ns(value: Union[int, str, None]) -> Union[int, None]: + """Returns nanoseconds from given input. + + Accepts: + - None: returns None + - int: returned as-is (assumed nanoseconds) + - str: parsed as Go duration (e.g. "30s", "1m", "500ms", "1h30m", "0") + + Raises: + TypeError: if value is not int, str, or None + ValueError: if the string cannot be parsed as a Go duration + """ + if value is None: + return None + if isinstance(value, int): + return value + if not isinstance(value, str): + raise TypeError(f"Duration must be int or str, got {type(value).__name__}") + + total_ns = 0 + remaining = value.strip() + if not remaining: + raise ValueError("Empty duration string") + + if remaining == "0": + return 0 + + pattern = re.compile(r"(\d+)(ns|µs|us|ms|s|m|h)") + pos = 0 + while pos < len(remaining): + match = pattern.match(remaining, pos) + if not match: + raise ValueError( + f"Invalid duration format: {value!r}. " + f"Use Go-style durations like '30s', '1m', '500ms', '1h30m'." + ) + num_str, unit = match.groups() + total_ns += int(num_str) * DURATION_UNITS_NS[unit] + pos = match.end() + + return total_ns + def parse_repository(name: str) -> tuple[str, Optional[str]]: """Parse repository image name from tag. diff --git a/podman/domain/containers_create.py b/podman/domain/containers_create.py index 7ccdf7f4..78d65179 100644 --- a/podman/domain/containers_create.py +++ b/podman/domain/containers_create.py @@ -10,6 +10,7 @@ import requests from podman import api +from podman.api.parse_utils import HEALTH_ON_FAILURE_ACTION, prepare_duration_ns from podman.domain.containers import Container from podman.domain.images import Image from podman.domain.pods import Pod @@ -91,38 +92,48 @@ def create( group_add (list[str]): List of additional group names and/or IDs that the container process will run as. healthcheck (dict[str,Any]): Specify a test to perform to check that the - container is healthy. + container is healthy. Mutually exclusive with fine-grained health_* kwargs. health_check_on_failure_action (int): Specify an action if a healthcheck fails. + Mutually exclusive with health_on_failure. health_cmd (str): set a healthcheck command for the container ('None' disables the existing healthcheck) - health_interval (str): set an interval for the healthcheck (a value of disable results - in no automatic timer setup)(Changing this setting resets timer.) (default "30s") - health_log_destination (str): set the destination of the HealthCheck log. Directory - path, local or events_logger (local use container state file)(Warning: Changing + health_interval (str or int): set an interval for the healthcheck. Accepts a + Go-style duration string (e.g. "30s", "1m") or nanoseconds as int. + (a value of disable results in no automatic timer setup) + (Changing this setting resets timer.) (default "30s") + health_log_destination (str): set the destination of the HealthCheck log. Directory + path, local or events_logger (local use container state file) (Warning: Changing this setting may cause the loss of previous logs.) (default "local") health_max_log_count (int): set maximum number of attempts in the HealthCheck log file. ('0' value means an infinite number of attempts in the log file) (default 5) health_max_logs_size (int): set maximum length in characters of stored HealthCheck log. ('0' value means an infinite log length) (default 500) - health_on_failure (str): action to take once the container turns unhealthy - (default "none") + health_on_failure (str): action to take once the container turns unhealthy. + One of: "none" (0), "kill" (2), "restart" (3), "stop" (4). Value 1 is + reserved as "invalid" in the Go enum. Mutually exclusive with + health_check_on_failure_action. (default "none") health_retries (int): the number of retries allowed before a healthcheck is considered to be unhealthy (default 3) - health_start_period (str): the initialization time needed for a container to bootstrap + health_start_period (str or int): the initialization time needed for a container to + bootstrap. Accepts a Go-style duration string or nanoseconds as int. (default "0s") - health_startup_cmd (str): Set a startup healthcheck command for the container - health_startup_interval (str): Set an interval for the startup healthcheck. Changing - this setting resets the timer, depending on the state of the container. - (default "30s") - health_startup_retries (int): Set the maximum number of retries before the startup + health_startup_cmd (str): set a startup healthcheck command for the container + health_startup_interval (str or int): set an interval for the startup healthcheck. + Accepts a Go-style duration string or nanoseconds as int. Changing this setting + resets the timer, depending on the state of the container. (default "30s") + health_startup_retries (int): set the maximum number of retries before the startup healthcheck will restart the container - health_startup_success (int): Set the number of consecutive successes before the + health_startup_success (int): set the number of consecutive successes before the startup healthcheck is marked as successful and the normal healthcheck begins (0 indicates any success will start the regular healthcheck) - health_startup_timeout (str): Set the maximum amount of time that the startup - healthcheck may take before it is considered failed (default "30s") - health_timeout (str): the maximum time allowed to complete the healthcheck before an - interval is considered failed (default "30s") + health_startup_timeout (str or int): set the maximum amount of time that the startup + healthcheck may take before it is considered failed. Accepts a Go-style duration + string or nanoseconds as int. (default "30s") + health_timeout (str or int): the maximum time allowed to complete the healthcheck + before an interval is considered failed. Accepts a Go-style duration string or + nanoseconds as int. (default "30s") + no_healthcheck (bool): disable healthchecks for this container. Mutually exclusive + with health_* kwargs. hostname (str): Optional hostname for the container. init (bool): Run an init inside the container that forwards signals and reaps processes init_path (str): Path to the docker-init binary @@ -807,7 +818,8 @@ def parse_host_port(_container_port, _protocol, _host): params["restart_tries"] = args["restart_policy"].get("MaximumRetryCount") args.pop("restart_policy") - health_commands_data = { + # Assemble fine-grained health kwargs into nested API objects. + health_kwargs = { "health_cmd", "health_interval", "health_log_destination", @@ -823,15 +835,88 @@ def parse_host_port(_container_port, _protocol, _host): "health_startup_timeout", "health_timeout", } - # the healthcheck section of parameters accepted can be either no_healthcheck or a series - # of healthcheck parameters - if args.get("no_healthcheck"): - if conflicts := set(args) & health_commands_data: - raise ValueError(f"Cannot set {conflicts.pop()} when no_healthcheck is True") - params["no_healthcheck"] = args.pop("no_healthcheck", None) - else: - for hc in set(args) & health_commands_data: - params[hc] = args.pop(hc, None) + present_health_kwargs = set(args) & health_kwargs + no_healthcheck = args.pop("no_healthcheck", None) + + if no_healthcheck and present_health_kwargs: + raise ValueError(f"Cannot set {min(present_health_kwargs)} when no_healthcheck is True") + + if params.get("healthconfig") and present_health_kwargs: + raise ValueError( + "Cannot combine 'healthcheck' dict with fine-grained health_* kwargs. " + "Use one or the other." + ) + + if no_healthcheck: + params["healthconfig"] = {"Test": ["NONE"]} + elif present_health_kwargs: + health_cmd = args.pop("health_cmd", None) + health_interval = args.pop("health_interval", None) + health_timeout = args.pop("health_timeout", None) + health_retries = args.pop("health_retries", None) + health_start_period = args.pop("health_start_period", None) + + healthconfig = {} + if health_cmd is not None: + healthconfig["Test"] = ["CMD-SHELL", health_cmd] + if health_interval is not None: + healthconfig["Interval"] = prepare_duration_ns(health_interval) + if health_timeout is not None: + healthconfig["Timeout"] = prepare_duration_ns(health_timeout) + if health_retries is not None: + healthconfig["Retries"] = int(health_retries) + if health_start_period is not None: + healthconfig["StartPeriod"] = prepare_duration_ns(health_start_period) + if healthconfig: + params["healthconfig"] = healthconfig + + startup_cmd = args.pop("health_startup_cmd", None) + startup_interval = args.pop("health_startup_interval", None) + startup_timeout = args.pop("health_startup_timeout", None) + startup_retries = args.pop("health_startup_retries", None) + startup_success = args.pop("health_startup_success", None) + + startup_config = {} + if startup_cmd is not None: + startup_config["Test"] = ["CMD-SHELL", startup_cmd] + if startup_interval is not None: + startup_config["Interval"] = prepare_duration_ns(startup_interval) + if startup_timeout is not None: + startup_config["Timeout"] = prepare_duration_ns(startup_timeout) + if startup_retries is not None: + startup_config["Retries"] = int(startup_retries) + if startup_success is not None: + startup_config["Successes"] = int(startup_success) + if startup_config: + params["startupHealthConfig"] = startup_config + + health_log_dest = args.pop("health_log_destination", None) + if health_log_dest is not None: + params["healthLogDestination"] = health_log_dest + + health_max_count = args.pop("health_max_log_count", None) + if health_max_count is not None: + params["healthMaxLogCount"] = int(health_max_count) + + health_max_size = args.pop("health_max_logs_size", None) + if health_max_size is not None: + params["healthMaxLogSize"] = int(health_max_size) + + health_on_failure = args.pop("health_on_failure", None) + if health_on_failure is not None: + if params.get("health_check_on_failure_action") is not None: + raise ValueError( + "Cannot set both 'health_on_failure' and " + "'health_check_on_failure_action'. Use one or the other." + ) + action = HEALTH_ON_FAILURE_ACTION.get(health_on_failure.lower()) + if action is None: + valid = ", ".join(sorted(HEALTH_ON_FAILURE_ACTION)) + raise ValueError( + f"Invalid health_on_failure value: {health_on_failure!r}. " + f"Must be one of: {valid}" + ) + params["health_check_on_failure_action"] = action params["resource_limits"]["pids"] = {"limit": args.pop("pids_limit", None)} diff --git a/podman/tests/integration/test_container_create.py b/podman/tests/integration/test_container_create.py index 3da51879..1744d7d7 100644 --- a/podman/tests/integration/test_container_create.py +++ b/podman/tests/integration/test_container_create.py @@ -262,6 +262,53 @@ def test_container_healthchecks(self): container = self.client.containers.create(self.alpine_image, **parameters) self.containers.append(container) + def test_container_healthcheck_fine_grained_kwargs(self): + """Test fine-grained health kwargs produce correct healthcheck config.""" + container = self.client.containers.create( + self.alpine_image, + health_cmd="echo healthy", + health_interval="10s", + health_timeout="5s", + health_retries=3, + health_start_period="2s", + health_on_failure="restart", + ) + self.containers.append(container) + + inspect = container.inspect() + hc = inspect['Config']['Healthcheck'] + self.assertEqual(hc['Test'], ['CMD-SHELL', 'echo healthy']) + self.assertEqual(hc['Interval'], 10_000_000_000) + self.assertEqual(hc['Timeout'], 5_000_000_000) + self.assertEqual(hc['Retries'], 3) + self.assertEqual(hc['StartPeriod'], 2_000_000_000) + + def test_container_healthcheck_on_failure_none(self): + """Test health_on_failure='none' (default action) works.""" + container = self.client.containers.create( + self.alpine_image, + health_cmd="echo healthy", + health_interval="10s", + health_on_failure="none", + ) + self.containers.append(container) + + inspect = container.inspect() + hc = inspect['Config']['Healthcheck'] + self.assertEqual(hc['Test'], ['CMD-SHELL', 'echo healthy']) + + def test_container_healthcheck_no_healthcheck(self): + """Test no_healthcheck=True disables healthchecks.""" + container = self.client.containers.create( + self.alpine_image, + no_healthcheck=True, + ) + self.containers.append(container) + + inspect = container.inspect() + hc = inspect['Config']['Healthcheck'] + self.assertEqual(hc['Test'], ['NONE']) + def test_container_mem_limit(self): """Test passing memory limit""" self._test_memory_limit('mem_limit', 'Memory') diff --git a/podman/tests/unit/test_containersmanager.py b/podman/tests/unit/test_containersmanager.py index f1c1b1a3..3a737022 100644 --- a/podman/tests/unit/test_containersmanager.py +++ b/podman/tests/unit/test_containersmanager.py @@ -679,8 +679,6 @@ def test_run(self, mock): @requests_mock.Mocker() def test_run_404(self, mock): - # mock the first POST to return 404, - # then the second POST (after pulling the image) to return 201 mock.post( tests.LIBPOD_URL + "/containers/create", [ @@ -744,8 +742,8 @@ def test_run_404(self, mock): self.assertEqual(mock.call_count, 5) @requests_mock.Mocker() - def test_create_all_healthcheck_parameters(self, mock): - """Test that all healthcheck parameters are correctly passed to the API.""" + def test_create_healthcheck_nested_payload(self, mock): + """Test that fine-grained health kwargs produce nested healthconfig/startupHealthConfig.""" mock_response = MagicMock() mock_response.json = lambda: { "Id": "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", @@ -758,41 +756,125 @@ def test_create_all_healthcheck_parameters(self, mock): json=FIRST_CONTAINER, ) - # Test all 14 healthcheck parameters with realistic values - healthcheck_params = { - "health_cmd": "curl -f http://localhost:8080/health", - "health_interval": "30s", - "health_log_destination": "local", - "health_max_log_count": 5, - "health_max_logs_size": 500, - "health_on_failure": "restart", - "health_retries": 3, - "health_start_period": "60s", - "health_startup_cmd": "curl -f http://localhost:8080/ready", - "health_startup_interval": "10s", - "health_startup_retries": 5, - "health_startup_success": 2, - "health_startup_timeout": "45s", - "health_timeout": "30s", + self.client.containers.create( + "fedora", + "/usr/bin/ls", + health_cmd="curl -f http://localhost:8080/health", + health_interval="30s", + health_timeout="10s", + health_retries=3, + health_start_period="1m", + ) + + actual_data = json.loads(self.client.containers.client.post.call_args[1]["data"]) + + self.assertEqual( + actual_data["healthconfig"], + { + "Test": ["CMD-SHELL", "curl -f http://localhost:8080/health"], + "Interval": 30_000_000_000, + "Timeout": 10_000_000_000, + "Retries": 3, + "StartPeriod": 60_000_000_000, + }, + ) + + @requests_mock.Mocker() + def test_create_startup_healthcheck_nested_payload(self, mock): + """Test that startup health kwargs produce nested startupHealthConfig.""" + mock_response = MagicMock() + mock_response.json = lambda: { + "Id": "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", + "Size": 1024, } + self.client.containers.client.post = MagicMock(return_value=mock_response) + mock.get( + tests.LIBPOD_URL + + "/containers/87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd/json", + json=FIRST_CONTAINER, + ) - self.client.containers.create("fedora", "/usr/bin/ls", **healthcheck_params) - self.client.containers.client.post.assert_called() + self.client.containers.create( + "fedora", + "/usr/bin/ls", + health_startup_cmd="curl -f http://localhost:8080/ready", + health_startup_interval="10s", + health_startup_timeout="45s", + health_startup_retries=5, + health_startup_success=2, + ) - # Verify all healthcheck parameters are in the request payload actual_data = json.loads(self.client.containers.client.post.call_args[1]["data"]) - for param_name, expected_value in healthcheck_params.items(): - self.assertIn(param_name, actual_data, f"Parameter {param_name} not found in request") - self.assertEqual( - actual_data[param_name], - expected_value, - f"Parameter {param_name} has incorrect value", - ) + self.assertEqual( + actual_data["startupHealthConfig"], + { + "Test": ["CMD-SHELL", "curl -f http://localhost:8080/ready"], + "Interval": 10_000_000_000, + "Timeout": 45_000_000_000, + "Retries": 5, + "Successes": 2, + }, + ) + + @requests_mock.Mocker() + def test_create_health_scalar_fields(self, mock): + """Test healthLogDestination, healthMaxLogCount, healthMaxLogSize mapping.""" + mock_response = MagicMock() + mock_response.json = lambda: { + "Id": "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", + "Size": 1024, + } + self.client.containers.client.post = MagicMock(return_value=mock_response) + mock.get( + tests.LIBPOD_URL + + "/containers/87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd/json", + json=FIRST_CONTAINER, + ) + + self.client.containers.create( + "fedora", + "/usr/bin/ls", + health_cmd="true", + health_log_destination="events_logger", + health_max_log_count=10, + health_max_logs_size=1024, + ) + + actual_data = json.loads(self.client.containers.client.post.call_args[1]["data"]) + + self.assertEqual(actual_data["healthLogDestination"], "events_logger") + self.assertEqual(actual_data["healthMaxLogCount"], 10) + self.assertEqual(actual_data["healthMaxLogSize"], 1024) @requests_mock.Mocker() - def test_create_no_healthcheck_validation(self, mock): - """Test no_healthcheck validation and functionality.""" + def test_create_health_on_failure_mapping(self, mock): + """Test health_on_failure string maps to health_check_on_failure_action int.""" + mock_response = MagicMock() + mock_response.json = lambda: { + "Id": "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", + "Size": 1024, + } + self.client.containers.client.post = MagicMock(return_value=mock_response) + mock.get( + tests.LIBPOD_URL + + "/containers/87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd/json", + json=FIRST_CONTAINER, + ) + + self.client.containers.create( + "fedora", + "/usr/bin/ls", + health_cmd="true", + health_on_failure="restart", + ) + + actual_data = json.loads(self.client.containers.client.post.call_args[1]["data"]) + self.assertEqual(actual_data["health_check_on_failure_action"], 3) + + @requests_mock.Mocker() + def test_create_no_healthcheck(self, mock): + """Test no_healthcheck=True produces healthconfig with Test=NONE.""" mock_response = MagicMock() mock_response.json = lambda: { "Id": "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", @@ -805,54 +887,79 @@ def test_create_no_healthcheck_validation(self, mock): json=FIRST_CONTAINER, ) - # Test that no_healthcheck=True works without conflicts self.client.containers.create("fedora", "/usr/bin/ls", no_healthcheck=True) - self.client.containers.client.post.assert_called() actual_data = json.loads(self.client.containers.client.post.call_args[1]["data"]) - self.assertIn("no_healthcheck", actual_data) - self.assertTrue(actual_data["no_healthcheck"]) - - # Test that setting healthcheck parameters with no_healthcheck=True raises ValueError - healthcheck_params = [ - "health_cmd", - "health_interval", - "health_log_destination", - "health_max_log_count", - "health_max_logs_size", - "health_on_failure", - "health_retries", - "health_start_period", - "health_startup_cmd", - "health_startup_interval", - "health_startup_retries", - "health_startup_success", - "health_startup_timeout", - "health_timeout", - ] + self.assertEqual(actual_data["healthconfig"], {"Test": ["NONE"]}) - # Test each parameter individually with no_healthcheck=True - for param in healthcheck_params: - with self.subTest(param=param): - kwargs = {"no_healthcheck": True, param: "test_value"} - with self.assertRaises(ValueError) as context: - self.client.containers.create("fedora", "/usr/bin/ls", **kwargs) + def test_create_no_healthcheck_conflicts_with_health_kwargs(self): + """Test that no_healthcheck=True + health kwargs raises ValueError.""" + with self.assertRaises(ValueError) as ctx: + self.client.containers.create( + "fedora", "/usr/bin/ls", no_healthcheck=True, health_cmd="true" + ) + self.assertIn("no_healthcheck is True", str(ctx.exception)) - expected_message = f"Cannot set {param} when no_healthcheck is True" - self.assertEqual(str(context.exception), expected_message) + def test_create_healthcheck_dict_conflicts_with_health_kwargs(self): + """Test that healthcheck dict + health_cmd raises ValueError.""" + with self.assertRaises(ValueError) as ctx: + self.client.containers.create( + "fedora", + "/usr/bin/ls", + healthcheck={"Test": ["CMD-SHELL", "true"]}, + health_cmd="curl localhost", + ) + self.assertIn("Cannot combine", str(ctx.exception)) + + def test_create_health_on_failure_conflicts_with_int(self): + """Test that health_on_failure + health_check_on_failure_action raises ValueError.""" + with self.assertRaises(ValueError) as ctx: + self.client.containers.create( + "fedora", + "/usr/bin/ls", + health_cmd="true", + health_on_failure="restart", + health_check_on_failure_action=3, + ) + self.assertIn("Cannot set both", str(ctx.exception)) - # Test multiple healthcheck parameters with no_healthcheck=True - with self.assertRaises(ValueError) as context: + def test_create_health_on_failure_invalid_value(self): + """Test that invalid health_on_failure value raises ValueError.""" + with self.assertRaises(ValueError) as ctx: self.client.containers.create( "fedora", "/usr/bin/ls", - no_healthcheck=True, - health_cmd="test", - health_interval="30s", + health_cmd="true", + health_on_failure="invalid_action", ) - # Should raise for the first conflicting parameter found - self.assertIn("Cannot set", str(context.exception)) - self.assertIn("when no_healthcheck is True", str(context.exception)) + self.assertIn("Invalid health_on_failure value", str(ctx.exception)) + + @requests_mock.Mocker() + def test_create_health_duration_as_int(self, mock): + """Test that int duration values are passed through as nanoseconds.""" + mock_response = MagicMock() + mock_response.json = lambda: { + "Id": "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", + "Size": 1024, + } + self.client.containers.client.post = MagicMock(return_value=mock_response) + mock.get( + tests.LIBPOD_URL + + "/containers/87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd/json", + json=FIRST_CONTAINER, + ) + + self.client.containers.create( + "fedora", + "/usr/bin/ls", + health_cmd="true", + health_interval=5_000_000_000, + health_timeout=2_000_000_000, + ) + + actual_data = json.loads(self.client.containers.client.post.call_args[1]["data"]) + self.assertEqual(actual_data["healthconfig"]["Interval"], 5_000_000_000) + self.assertEqual(actual_data["healthconfig"]["Timeout"], 2_000_000_000) if __name__ == "__main__": diff --git a/podman/tests/unit/test_parse_utils.py b/podman/tests/unit/test_parse_utils.py index 61d7da61..8b2a3b9d 100644 --- a/podman/tests/unit/test_parse_utils.py +++ b/podman/tests/unit/test_parse_utils.py @@ -116,5 +116,115 @@ def test_stream_helper_with_decode(self) -> None: self.assertDictEqual(json.loads(expected), actual) # type: ignore[arg-type] +class PrepareDurationNsTestCase(unittest.TestCase): + """Test prepare_duration_ns utility.""" + + def test_none_returns_none(self): + self.assertIsNone(api.prepare_duration_ns(None)) + + def test_int_passthrough(self): + self.assertEqual(api.prepare_duration_ns(0), 0) + self.assertEqual(api.prepare_duration_ns(1), 1) + self.assertEqual(api.prepare_duration_ns(30_000_000_000), 30_000_000_000) + + def test_seconds(self): + self.assertEqual(api.prepare_duration_ns("1s"), 1_000_000_000) + self.assertEqual(api.prepare_duration_ns("30s"), 30_000_000_000) + self.assertEqual(api.prepare_duration_ns("120s"), 120_000_000_000) + + def test_minutes(self): + self.assertEqual(api.prepare_duration_ns("1m"), 60_000_000_000) + self.assertEqual(api.prepare_duration_ns("5m"), 300_000_000_000) + + def test_hours(self): + self.assertEqual(api.prepare_duration_ns("1h"), 3_600_000_000_000) + self.assertEqual(api.prepare_duration_ns("2h"), 7_200_000_000_000) + + def test_milliseconds(self): + self.assertEqual(api.prepare_duration_ns("1ms"), 1_000_000) + self.assertEqual(api.prepare_duration_ns("500ms"), 500_000_000) + + def test_microseconds_us(self): + self.assertEqual(api.prepare_duration_ns("1us"), 1_000) + self.assertEqual(api.prepare_duration_ns("100us"), 100_000) + + def test_microseconds_mu(self): + self.assertEqual(api.prepare_duration_ns("1µs"), 1_000) + self.assertEqual(api.prepare_duration_ns("100µs"), 100_000) + + def test_nanoseconds(self): + self.assertEqual(api.prepare_duration_ns("1ns"), 1) + self.assertEqual(api.prepare_duration_ns("999ns"), 999) + + def test_compound_duration(self): + self.assertEqual(api.prepare_duration_ns("1h30m"), 5_400_000_000_000) + self.assertEqual(api.prepare_duration_ns("1m30s"), 90_000_000_000) + self.assertEqual(api.prepare_duration_ns("2h30m15s"), 9_015_000_000_000) + self.assertEqual(api.prepare_duration_ns("1s500ms"), 1_500_000_000) + + def test_bare_zero(self): + self.assertEqual(api.prepare_duration_ns("0"), 0) + + def test_fractional_raises(self): + with self.assertRaises(ValueError): + api.prepare_duration_ns("1.5s") + with self.assertRaises(ValueError): + api.prepare_duration_ns("0.5s") + + def test_whitespace_stripped(self): + self.assertEqual(api.prepare_duration_ns(" 30s "), 30_000_000_000) + self.assertEqual(api.prepare_duration_ns(" 1m "), 60_000_000_000) + + def test_invalid_format_raises(self): + with self.assertRaises(ValueError): + api.prepare_duration_ns("invalid") + + def test_empty_string_raises(self): + with self.assertRaises(ValueError): + api.prepare_duration_ns("") + + def test_whitespace_only_raises(self): + with self.assertRaises(ValueError): + api.prepare_duration_ns(" ") + + def test_number_without_unit_raises(self): + with self.assertRaises(ValueError): + api.prepare_duration_ns("30") + + def test_unit_without_number_raises(self): + with self.assertRaises(ValueError): + api.prepare_duration_ns("s") + + def test_wrong_type_raises(self): + with self.assertRaises(TypeError): + api.prepare_duration_ns(3.14) # type: ignore + with self.assertRaises(TypeError): + api.prepare_duration_ns([30]) # type: ignore + + +class HealthOnFailureActionTestCase(unittest.TestCase): + """Test HEALTH_ON_FAILURE_ACTION mapping.""" + + def test_none_maps_to_zero(self): + self.assertEqual(api.HEALTH_ON_FAILURE_ACTION["none"], 0) + + def test_kill_maps_to_two(self): + self.assertEqual(api.HEALTH_ON_FAILURE_ACTION["kill"], 2) + + def test_restart_maps_to_three(self): + self.assertEqual(api.HEALTH_ON_FAILURE_ACTION["restart"], 3) + + def test_stop_maps_to_four(self): + self.assertEqual(api.HEALTH_ON_FAILURE_ACTION["stop"], 4) + + def test_invalid_value_one_not_exposed(self): + self.assertNotIn(1, api.HEALTH_ON_FAILURE_ACTION.values()) + + def test_all_values_are_ints(self): + for key, value in api.HEALTH_ON_FAILURE_ACTION.items(): + self.assertIsInstance(key, str) + self.assertIsInstance(value, int) + + if __name__ == '__main__': unittest.main()