diff --git a/podman/domain/containers_create.py b/podman/domain/containers_create.py index 12b756fc..b6a5ddff 100644 --- a/podman/domain/containers_create.py +++ b/podman/domain/containers_create.py @@ -17,7 +17,7 @@ logger = logging.getLogger("podman.containers") -NAMED_VOLUME_PATTERN = re.compile(r'[a-zA-Z0-9][a-zA-Z0-9_.-]*') +NAMED_VOLUME_PATTERN = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9_.-]*") class CreateMixin: # pylint: disable=too-few-public-methods @@ -375,7 +375,9 @@ def create( payload = api.prepare_body(payload) response = self.client.post( - "/containers/create", headers={"content-type": "application/json"}, data=payload + "/containers/create", + headers={"content-type": "application/json"}, + data=payload, ) response.raise_for_status(not_found=ImageNotFound) @@ -383,6 +385,48 @@ def create( return self.get(container_id) + @staticmethod + def _convert_env_list_to_dict(env_list): + """Convert a list of environment variables to a dictionary. + + Args: + env_list (List[str]): List of environment variables in the format ["KEY=value"] + + Returns: + Dict[str, str]: Dictionary of environment variables + + Raises: + ValueError: If any environment variable is not in the correct format + """ + if not isinstance(env_list, list): + raise TypeError(f"Expected list, got {type(env_list).__name__}") + + env_dict = {} + + for env_var in env_list: + if not isinstance(env_var, str): + raise TypeError( + f"Environment variable must be a string, " + f"got {type(env_var).__name__}: {repr(env_var)}" + ) + + # Handle empty strings + if not env_var.strip(): + raise ValueError(f"Environment variable cannot be empty") + if "=" not in env_var: + raise ValueError( + f"Environment variable '{env_var}' is not in the correct format. " + "Expected format: 'KEY=value'" + ) + key, value = env_var.split("=", 1) # Split on first '=' only + + # Validate key is not empty + if not key.strip(): + raise ValueError(f"Environment variable has empty key: '{env_var}'") + + env_dict[key] = value + return env_dict + # pylint: disable=too-many-locals,too-many-statements,too-many-branches @staticmethod def _render_payload(kwargs: MutableMapping[str, Any]) -> dict[str, Any]: @@ -410,6 +454,23 @@ def _render_payload(kwargs: MutableMapping[str, Any]) -> dict[str, Any]: with suppress(KeyError): del args[key] + # Handle environment variables + environment = args.pop("environment", None) + if environment is not None: + if isinstance(environment, list): + try: + environment = CreateMixin._convert_env_list_to_dict(environment) + except ValueError as e: + raise ValueError( + "Failed to convert environment variables list to dictionary. " + f"Error: {str(e)}" + ) from e + elif not isinstance(environment, dict): + raise TypeError( + "Environment variables must be provided as either a dictionary " + "or a list of strings in the format ['KEY=value']" + ) + # These keywords are not supported for various reasons. unsupported_keys = set(args.keys()).intersection( ( @@ -466,9 +527,9 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: try: return int(size) except ValueError as bad_size: - mapping = {'b': 0, 'k': 1, 'm': 2, 'g': 3} - mapping_regex = ''.join(mapping.keys()) - search = re.search(rf'^(\d+)([{mapping_regex}])$', size.lower()) + mapping = {"b": 0, "k": 1, "m": 2, "g": 3} + mapping_regex = "".join(mapping.keys()) + search = re.search(rf"^(\d+)([{mapping_regex}])$", size.lower()) if search: return int(search.group(1)) * (1024 ** mapping[search.group(2)]) raise TypeError( @@ -497,7 +558,7 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: "dns_search": pop("dns_search"), "dns_server": pop("dns"), "entrypoint": pop("entrypoint"), - "env": pop("environment"), + "env": environment, "env_host": pop("env_host"), # TODO document, podman only "expose": {}, "groups": pop("group_add"), @@ -607,7 +668,7 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: if _k in bool_options and v is True: options.append(option_name) elif _k in regular_options: - options.append(f'{option_name}={v}') + options.append(f"{option_name}={v}") elif _k in simple_options: options.append(v) @@ -709,12 +770,12 @@ def parse_host_port(_container_port, _protocol, _host): for item in args.pop("volumes", {}).items(): key, value = item - extended_mode = value.get('extended_mode', []) + extended_mode = value.get("extended_mode", []) if not isinstance(extended_mode, list): raise ValueError("'extended_mode' value should be a list") options = extended_mode - mode = value.get('mode') + mode = value.get("mode") if mode is not None: if not isinstance(mode, str): raise ValueError("'mode' value should be a str") @@ -729,10 +790,10 @@ def parse_host_port(_container_port, _protocol, _host): params["volumes"].append(volume) else: mount_point = { - "destination": value['bind'], + "destination": value["bind"], "options": options, "source": key, - "type": 'bind', + "type": "bind", } params["mounts"].append(mount_point) diff --git a/podman/tests/integration/test_container_create.py b/podman/tests/integration/test_container_create.py index 815590eb..eac8200a 100644 --- a/podman/tests/integration/test_container_create.py +++ b/podman/tests/integration/test_container_create.py @@ -103,6 +103,44 @@ def test_container_extra_hosts(self): for hosts_entry in formatted_hosts: self.assertIn(hosts_entry, logs) + def test_container_environment_variables(self): + """Test environment variables passed to the container.""" + with self.subTest("Check environment variables as dictionary"): + env_dict = {"MY_VAR": "123", "ANOTHER_VAR": "456"} + container = self.client.containers.create( + self.alpine_image, command=["env"], environment=env_dict + ) + self.containers.append(container) + + container_env = container.attrs.get('Config', {}).get('Env', []) + for key, value in env_dict.items(): + self.assertIn(f"{key}={value}", container_env) + + container.start() + container.wait() + logs = b"\n".join(container.logs()).decode() + + for key, value in env_dict.items(): + self.assertIn(f"{key}={value}", logs) + + with self.subTest("Check environment variables as list"): + env_list = ["MY_VAR=123", "ANOTHER_VAR=456"] + container = self.client.containers.create( + self.alpine_image, command=["env"], environment=env_list + ) + self.containers.append(container) + + container_env = container.attrs.get('Config', {}).get('Env', []) + for env in env_list: + self.assertIn(env, container_env) + + container.start() + container.wait() + logs = b"\n".join(container.logs()).decode() + + for env in env_list: + self.assertIn(env, logs) + def _test_memory_limit(self, parameter_name, host_config_name, set_mem_limit=False): """Base for tests which checks memory limits""" memory_limit_tests = [ diff --git a/podman/tests/unit/test_containersmanager.py b/podman/tests/unit/test_containersmanager.py index 47b30482..28e20d33 100644 --- a/podman/tests/unit/test_containersmanager.py +++ b/podman/tests/unit/test_containersmanager.py @@ -8,12 +8,13 @@ # Python < 3.10 from collections.abc import Iterator -from unittest.mock import DEFAULT, patch, MagicMock +from unittest.mock import DEFAULT, MagicMock, patch import requests_mock from podman import PodmanClient, tests from podman.domain.containers import Container +from podman.domain.containers_create import CreateMixin from podman.domain.containers_manager import ContainersManager from podman.errors import ImageNotFound, NotFound @@ -64,7 +65,8 @@ def test_get(self, mock): "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd" ) self.assertEqual( - actual.id, "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd" + actual.id, + "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", ) @requests_mock.Mocker() @@ -104,10 +106,12 @@ def test_list(self, mock): self.assertIsInstance(actual, list) self.assertEqual( - actual[0].id, "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd" + actual[0].id, + "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", ) self.assertEqual( - actual[1].id, "6dc84cc0a46747da94e4c1571efcc01a756b4017261440b4b8985d37203c3c03" + actual[1].id, + "6dc84cc0a46747da94e4c1571efcc01a756b4017261440b4b8985d37203c3c03", ) @requests_mock.Mocker() @@ -132,10 +136,12 @@ def test_list_filtered(self, mock): self.assertIsInstance(actual, list) self.assertEqual( - actual[0].id, "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd" + actual[0].id, + "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", ) self.assertEqual( - actual[1].id, "6dc84cc0a46747da94e4c1571efcc01a756b4017261440b4b8985d37203c3c03" + actual[1].id, + "6dc84cc0a46747da94e4c1571efcc01a756b4017261440b4b8985d37203c3c03", ) @requests_mock.Mocker() @@ -148,10 +154,12 @@ def test_list_no_filters(self, mock): self.assertIsInstance(actual, list) self.assertEqual( - actual[0].id, "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd" + actual[0].id, + "87e1325c82424e49a00abdd4de08009eb76c7de8d228426a9b8af9318ced5ecd", ) self.assertEqual( - actual[1].id, "6dc84cc0a46747da94e4c1571efcc01a756b4017261440b4b8985d37203c3c03" + actual[1].id, + "6dc84cc0a46747da94e4c1571efcc01a756b4017261440b4b8985d37203c3c03", ) @requests_mock.Mocker() @@ -228,8 +236,8 @@ def test_create_parse_host_port(self, mock): json=FIRST_CONTAINER, ) - port_str = {'2233': 3333} - port_str_protocol = {'2244/tcp': 3344} + port_str = {"2233": 3333} + port_str_protocol = {"2244/tcp": 3344} port_int = {2255: 3355} ports = {**port_str, **port_str_protocol, **port_int} self.client.containers.create("fedora", "/usr/bin/ls", ports=ports) @@ -237,23 +245,23 @@ def test_create_parse_host_port(self, mock): self.client.containers.client.post.assert_called() expected_ports = [ { - 'container_port': 2233, - 'host_port': 3333, - 'protocol': 'tcp', + "container_port": 2233, + "host_port": 3333, + "protocol": "tcp", }, { - 'container_port': 2244, - 'host_port': 3344, - 'protocol': 'tcp', + "container_port": 2244, + "host_port": 3344, + "protocol": "tcp", }, { - 'container_port': 2255, - 'host_port': 3355, - 'protocol': 'tcp', + "container_port": 2255, + "host_port": 3355, + "protocol": "tcp", }, ] - actual_ports = json.loads(self.client.containers.client.post.call_args[1]['data'])[ - 'portmappings' + actual_ports = json.loads(self.client.containers.client.post.call_args[1]["data"])[ + "portmappings" ] self.assertEqual(expected_ports, actual_ports) @@ -313,6 +321,127 @@ def test_create_unknown_key(self): with self.assertRaises(TypeError): self.client.containers.create("fedora", "/usr/bin/ls", unknown_key=100.0) + @requests_mock.Mocker() + def test_create_convert_env_list_to_dict(self, mock): + env_list1 = ["FOO=foo", "BAR=bar"] + # Test valid list + converted_dict1 = {"FOO": "foo", "BAR": "bar"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list1), converted_dict1) + + # Test empty string + env_list2 = ["FOO=foo", ""] + self.assertRaises(ValueError, CreateMixin._convert_env_list_to_dict, env_list2) + + # Test non iterable + env_list3 = ["FOO=foo", None] + self.assertRaises(TypeError, CreateMixin._convert_env_list_to_dict, env_list3) + + # Test iterable with non string element + env_list4 = ["FOO=foo", []] + self.assertRaises(TypeError, CreateMixin._convert_env_list_to_dict, env_list4) + + # Test empty list + env_list5 = [] + converted_dict5 = {} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list5), converted_dict5) + + # Test single valid environment variable + env_list6 = ["SINGLE=value"] + converted_dict6 = {"SINGLE": "value"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list6), converted_dict6) + + # Test environment variable with empty value + env_list7 = ["EMPTY="] + converted_dict7 = {"EMPTY": ""} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list7), converted_dict7) + + # Test environment variable with multiple equals signs + env_list8 = ["URL=https://example.com/path?param=value"] + converted_dict8 = {"URL": "https://example.com/path?param=value"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list8), converted_dict8) + + # Test environment variable with spaces in value + env_list9 = ["MESSAGE=Hello World", "PATH=/usr/local/bin:/usr/bin"] + converted_dict9 = {"MESSAGE": "Hello World", "PATH": "/usr/local/bin:/usr/bin"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list9), converted_dict9) + + # Test environment variable with special characters + env_list10 = ["SPECIAL=!@#$%^&*()_+-=[]{}|;':\",./<>?"] + converted_dict10 = {"SPECIAL": "!@#$%^&*()_+-=[]{}|;':\",./<>?"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list10), converted_dict10) + + # Test environment variable with numeric values + env_list11 = ["PORT=8080", "TIMEOUT=30"] + converted_dict11 = {"PORT": "8080", "TIMEOUT": "30"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list11), converted_dict11) + + # Test environment variable with boolean-like values + env_list12 = ["DEBUG=true", "VERBOSE=false", "ENABLED=1", "DISABLED=0"] + converted_dict12 = { + "DEBUG": "true", + "VERBOSE": "false", + "ENABLED": "1", + "DISABLED": "0", + } + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list12), converted_dict12) + + # Test environment variable with whitespace in key (should preserve) + env_list13 = [" SPACED_KEY =value", "KEY= spaced_value "] + converted_dict13 = {" SPACED_KEY ": "value", "KEY": " spaced_value "} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list13), converted_dict13) + + # Test missing equals sign + env_list14 = ["FOO=foo", "INVALID"] + self.assertRaises(ValueError, CreateMixin._convert_env_list_to_dict, env_list14) + + # Test environment variable with only equals sign (empty key) + env_list15 = ["FOO=foo", "=value"] + self.assertRaises(ValueError, CreateMixin._convert_env_list_to_dict, env_list15) + + # Test environment variable with only whitespace key + env_list16 = ["FOO=foo", " =value"] + self.assertRaises(ValueError, CreateMixin._convert_env_list_to_dict, env_list16) + + # Test whitespace-only string + env_list17 = ["FOO=foo", " "] + self.assertRaises(ValueError, CreateMixin._convert_env_list_to_dict, env_list17) + + # Test various non-string types in list + env_list18 = ["FOO=foo", 123] + self.assertRaises(TypeError, CreateMixin._convert_env_list_to_dict, env_list18) + + env_list19 = ["FOO=foo", {"key": "value"}] + self.assertRaises(TypeError, CreateMixin._convert_env_list_to_dict, env_list19) + + env_list20 = ["FOO=foo", True] + self.assertRaises(TypeError, CreateMixin._convert_env_list_to_dict, env_list20) + + # Test duplicate keys (last one should win) + env_list21 = ["KEY=first", "KEY=second", "OTHER=value"] + converted_dict21 = {"KEY": "second", "OTHER": "value"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list21), converted_dict21) + + # Test very long environment variable + long_value = "x" * 1000 + env_list22 = [f"LONG_VAR={long_value}"] + converted_dict22 = {"LONG_VAR": long_value} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list22), converted_dict22) + + # Test environment variable with newlines and tabs + env_list23 = ["MULTILINE=line1\nline2\ttabbed"] + converted_dict23 = {"MULTILINE": "line1\nline2\ttabbed"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list23), converted_dict23) + + # Test environment variable with unicode characters + env_list24 = ["UNICODE=こんにちは", "EMOJI=🚀🌟"] + converted_dict24 = {"UNICODE": "こんにちは", "EMOJI": "🚀🌟"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list24), converted_dict24) + + # Test case sensitivity + env_list25 = ["path=/usr/bin", "PATH=/usr/local/bin"] + converted_dict25 = {"path": "/usr/bin", "PATH": "/usr/local/bin"} + self.assertEqual(CreateMixin._convert_env_list_to_dict(env_list25), converted_dict25) + @requests_mock.Mocker() def test_run_detached(self, mock): mock.post( @@ -375,7 +504,7 @@ def test_run(self, mock): actual = self.client.containers.run("fedora", "/usr/bin/ls") self.assertIsInstance(actual, bytes) - self.assertEqual(actual, b'This is a unittest - line 1This is a unittest - line 2') + self.assertEqual(actual, b"This is a unittest - line 1This is a unittest - line 2") # iter() cannot be reset so subtests used to create new instance with self.subTest("Stream results"): @@ -388,5 +517,5 @@ def test_run(self, mock): self.assertEqual(next(actual), b"This is a unittest - line 2") -if __name__ == '__main__': +if __name__ == "__main__": unittest.main()