mirror of
https://github.com/containers/podman-compose.git
synced 2025-06-19 19:27:47 +02:00
tests/unit: Add type annotations
Signed-off-by: Povilas Kanapickas <povilas@radix.lt>
This commit is contained in:
parent
ea22227625
commit
dedb081550
@ -7,7 +7,7 @@ from podman_compose import compose_exec_args
|
||||
|
||||
|
||||
class TestComposeExecArgs(unittest.TestCase):
|
||||
def test_minimal(self):
|
||||
def test_minimal(self) -> None:
|
||||
cnt = get_minimal_container()
|
||||
args = get_minimal_args()
|
||||
|
||||
@ -15,7 +15,7 @@ class TestComposeExecArgs(unittest.TestCase):
|
||||
expected = ["--interactive", "--tty", "container_name"]
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
def test_additional_env_value_equals(self):
|
||||
def test_additional_env_value_equals(self) -> None:
|
||||
cnt = get_minimal_container()
|
||||
args = get_minimal_args()
|
||||
args.env = ["key=valuepart1=valuepart2"]
|
||||
@ -31,11 +31,11 @@ class TestComposeExecArgs(unittest.TestCase):
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
|
||||
def get_minimal_container():
|
||||
def get_minimal_container() -> dict:
|
||||
return {}
|
||||
|
||||
|
||||
def get_minimal_args():
|
||||
def get_minimal_args() -> argparse.Namespace:
|
||||
return argparse.Namespace(
|
||||
T=None,
|
||||
cnt_command=None,
|
||||
|
@ -3,66 +3,67 @@
|
||||
|
||||
import io
|
||||
import unittest
|
||||
from typing import Union
|
||||
|
||||
from podman_compose import Podman
|
||||
|
||||
|
||||
class DummyReader:
|
||||
def __init__(self, data=None):
|
||||
def __init__(self, data: Union[list[bytes], None] = None):
|
||||
self.data = data or []
|
||||
|
||||
async def readuntil(self, _):
|
||||
async def readuntil(self, _: str) -> bytes:
|
||||
return self.data.pop(0)
|
||||
|
||||
def at_eof(self):
|
||||
def at_eof(self) -> bool:
|
||||
return len(self.data) == 0
|
||||
|
||||
|
||||
class TestComposeRunLogFormat(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self):
|
||||
def setUp(self) -> None:
|
||||
self.p = get_minimal_podman()
|
||||
self.buffer = io.StringIO()
|
||||
|
||||
async def test_single_line_single_chunk(self):
|
||||
async def test_single_line_single_chunk(self) -> None:
|
||||
reader = DummyReader([b'hello, world\n'])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), 'LL: hello, world\n')
|
||||
|
||||
async def test_empty(self):
|
||||
async def test_empty(self) -> None:
|
||||
reader = DummyReader([])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), '')
|
||||
|
||||
async def test_empty2(self):
|
||||
async def test_empty2(self) -> None:
|
||||
reader = DummyReader([b''])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), '')
|
||||
|
||||
async def test_empty_line(self):
|
||||
async def test_empty_line(self) -> None:
|
||||
reader = DummyReader([b'\n'])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), 'LL: \n')
|
||||
|
||||
async def test_line_split(self):
|
||||
async def test_line_split(self) -> None:
|
||||
reader = DummyReader([b'hello,', b' world\n'])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), 'LL: hello, world\n')
|
||||
|
||||
async def test_two_lines_in_one_chunk(self):
|
||||
async def test_two_lines_in_one_chunk(self) -> None:
|
||||
reader = DummyReader([b'hello\nbye\n'])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), 'LL: hello\nLL: bye\n')
|
||||
|
||||
async def test_double_blank(self):
|
||||
async def test_double_blank(self) -> None:
|
||||
reader = DummyReader([b'hello\n\n\nbye\n'])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), 'LL: hello\nLL: \nLL: \nLL: bye\n')
|
||||
|
||||
async def test_no_new_line_at_end(self):
|
||||
async def test_no_new_line_at_end(self) -> None:
|
||||
reader = DummyReader([b'hello\nbye'])
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:')
|
||||
await self.p._format_stream(reader, self.buffer, 'LL:') # type: ignore[arg-type]
|
||||
self.assertEqual(self.buffer.getvalue(), 'LL: hello\nLL: bye\n')
|
||||
|
||||
|
||||
def get_minimal_podman():
|
||||
return Podman(None)
|
||||
def get_minimal_podman() -> Podman:
|
||||
return Podman(None) # type: ignore[arg-type]
|
||||
|
@ -8,7 +8,7 @@ from podman_compose import compose_run_update_container_from_args
|
||||
|
||||
|
||||
class TestComposeRunUpdateContainerFromArgs(unittest.TestCase):
|
||||
def test_minimal(self):
|
||||
def test_minimal(self) -> None:
|
||||
cnt = get_minimal_container()
|
||||
compose = get_minimal_compose()
|
||||
args = get_minimal_args()
|
||||
@ -18,7 +18,7 @@ class TestComposeRunUpdateContainerFromArgs(unittest.TestCase):
|
||||
expected_cnt = {"name": "default_name", "tty": True}
|
||||
self.assertEqual(cnt, expected_cnt)
|
||||
|
||||
def test_additional_env_value_equals(self):
|
||||
def test_additional_env_value_equals(self) -> None:
|
||||
cnt = get_minimal_container()
|
||||
compose = get_minimal_compose()
|
||||
args = get_minimal_args()
|
||||
@ -35,7 +35,7 @@ class TestComposeRunUpdateContainerFromArgs(unittest.TestCase):
|
||||
}
|
||||
self.assertEqual(cnt, expected_cnt)
|
||||
|
||||
def test_publish_ports(self):
|
||||
def test_publish_ports(self) -> None:
|
||||
cnt = get_minimal_container()
|
||||
compose = get_minimal_compose()
|
||||
args = get_minimal_args()
|
||||
@ -51,15 +51,15 @@ class TestComposeRunUpdateContainerFromArgs(unittest.TestCase):
|
||||
self.assertEqual(cnt, expected_cnt)
|
||||
|
||||
|
||||
def get_minimal_container():
|
||||
def get_minimal_container() -> dict:
|
||||
return {}
|
||||
|
||||
|
||||
def get_minimal_compose():
|
||||
def get_minimal_compose() -> PodmanCompose:
|
||||
return PodmanCompose()
|
||||
|
||||
|
||||
def get_minimal_args():
|
||||
def get_minimal_args() -> argparse.Namespace:
|
||||
return argparse.Namespace(
|
||||
T=None,
|
||||
cnt_command=None,
|
||||
|
@ -2,14 +2,16 @@
|
||||
|
||||
import os
|
||||
import unittest
|
||||
from typing import Any
|
||||
from unittest import mock
|
||||
|
||||
from parameterized import parameterized
|
||||
|
||||
from podman_compose import PodmanCompose
|
||||
from podman_compose import container_to_args
|
||||
|
||||
|
||||
def create_compose_mock(project_name="test_project_name"):
|
||||
def create_compose_mock(project_name: str = "test_project_name") -> PodmanCompose:
|
||||
compose = mock.Mock()
|
||||
compose.project_name = project_name
|
||||
compose.dirname = "test_dirname"
|
||||
@ -19,14 +21,14 @@ def create_compose_mock(project_name="test_project_name"):
|
||||
compose.networks = {}
|
||||
compose.x_podman = {}
|
||||
|
||||
async def podman_output(*args, **kwargs):
|
||||
async def podman_output(*args: Any, **kwargs: Any) -> None:
|
||||
pass
|
||||
|
||||
compose.podman.output = mock.Mock(side_effect=podman_output)
|
||||
return compose
|
||||
|
||||
|
||||
def get_minimal_container():
|
||||
def get_minimal_container() -> dict[str, Any]:
|
||||
return {
|
||||
"name": "project_name_service_name1",
|
||||
"service_name": "service_name",
|
||||
@ -34,13 +36,13 @@ def get_minimal_container():
|
||||
}
|
||||
|
||||
|
||||
def get_test_file_path(rel_path):
|
||||
def get_test_file_path(rel_path: str) -> str:
|
||||
repo_root = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
|
||||
return os.path.realpath(os.path.join(repo_root, rel_path))
|
||||
|
||||
|
||||
class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_minimal(self):
|
||||
async def test_minimal(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -56,7 +58,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_runtime(self):
|
||||
async def test_runtime(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -75,7 +77,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_sysctl_list(self):
|
||||
async def test_sysctl_list(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -99,7 +101,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_sysctl_map(self):
|
||||
async def test_sysctl_map(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -123,7 +125,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_sysctl_wrong_type(self):
|
||||
async def test_sysctl_wrong_type(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
|
||||
@ -133,7 +135,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
cnt["sysctls"] = wrong_type
|
||||
await container_to_args(c, cnt)
|
||||
|
||||
async def test_pid(self):
|
||||
async def test_pid(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
|
||||
@ -152,7 +154,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_http_proxy(self):
|
||||
async def test_http_proxy(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -170,7 +172,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_uidmaps_extension_old_path(self):
|
||||
async def test_uidmaps_extension_old_path(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -179,7 +181,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
await container_to_args(c, cnt)
|
||||
|
||||
async def test_uidmaps_extension(self):
|
||||
async def test_uidmaps_extension(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -200,7 +202,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_gidmaps_extension(self):
|
||||
async def test_gidmaps_extension(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -221,7 +223,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_rootfs_extension(self):
|
||||
async def test_rootfs_extension(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -240,7 +242,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_no_hosts_extension(self):
|
||||
async def test_no_hosts_extension(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -258,7 +260,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_env_file_str(self):
|
||||
async def test_env_file_str(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -282,7 +284,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_env_file_str_not_exists(self):
|
||||
async def test_env_file_str_not_exists(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -291,7 +293,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
await container_to_args(c, cnt)
|
||||
|
||||
async def test_env_file_str_array_one_path(self):
|
||||
async def test_env_file_str_array_one_path(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -315,7 +317,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_env_file_str_array_two_paths(self):
|
||||
async def test_env_file_str_array_two_paths(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -344,7 +346,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_env_file_obj_required(self):
|
||||
async def test_env_file_obj_required(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -368,7 +370,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_env_file_obj_required_non_existent_path(self):
|
||||
async def test_env_file_obj_required_non_existent_path(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -377,7 +379,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
await container_to_args(c, cnt)
|
||||
|
||||
async def test_env_file_obj_optional(self):
|
||||
async def test_env_file_obj_optional(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -394,7 +396,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_gpu_count_all(self):
|
||||
async def test_gpu_count_all(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -422,7 +424,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_gpu_count_specific(self):
|
||||
async def test_gpu_count_specific(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -458,7 +460,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_gpu_device_ids_all(self):
|
||||
async def test_gpu_device_ids_all(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -492,7 +494,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_gpu_device_ids_specific(self):
|
||||
async def test_gpu_device_ids_specific(self) -> None:
|
||||
c = create_compose_mock()
|
||||
|
||||
cnt = get_minimal_container()
|
||||
@ -534,7 +536,9 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
(True, "z", ["-v", "./foo:/mnt:z"]),
|
||||
(True, "Z", ["-v", "./foo:/mnt:Z"]),
|
||||
])
|
||||
async def test_selinux_volume(self, prefer_volume, selinux_type, expected_additional_args):
|
||||
async def test_selinux_volume(
|
||||
self, prefer_volume: bool, selinux_type: str, expected_additional_args: list
|
||||
) -> None:
|
||||
c = create_compose_mock()
|
||||
c.prefer_volume_over_mount = prefer_volume
|
||||
|
||||
@ -572,7 +576,9 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
("compat_no_dash", True, "test_project_name", "test_project_name_network1"),
|
||||
("compat_dash", True, "test_project-name", "test_projectname_network1"),
|
||||
])
|
||||
async def test_network_default_name(self, name, is_compat, project_name, expected_network_name):
|
||||
async def test_network_default_name(
|
||||
self, name: str, is_compat: bool, project_name: str, expected_network_name: str
|
||||
) -> None:
|
||||
c = create_compose_mock(project_name)
|
||||
c.x_podman = {"default_net_name_compat": is_compat}
|
||||
c.networks = {'network1': {}}
|
||||
@ -591,7 +597,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_device(self):
|
||||
async def test_device(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
|
||||
@ -613,7 +619,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_cpuset(self):
|
||||
async def test_cpuset(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["cpuset"] = "0-1"
|
||||
@ -631,7 +637,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_pids_limit_container_level(self):
|
||||
async def test_pids_limit_container_level(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["pids_limit"] = 100
|
||||
@ -649,7 +655,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_pids_limit_deploy_section(self):
|
||||
async def test_pids_limit_deploy_section(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["deploy"] = {"resources": {"limits": {"pids": 100}}}
|
||||
@ -667,7 +673,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_pids_limit_both_same(self):
|
||||
async def test_pids_limit_both_same(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["pids_limit"] = 100
|
||||
@ -686,7 +692,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_pids_limit_both_different(self):
|
||||
async def test_pids_limit_both_different(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["pids_limit"] = 100
|
||||
@ -695,7 +701,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
await container_to_args(c, cnt)
|
||||
|
||||
async def test_heathcheck_string(self):
|
||||
async def test_healthcheck_string(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["healthcheck"] = {
|
||||
@ -715,7 +721,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_heathcheck_cmd_args(self):
|
||||
async def test_healthcheck_cmd_args(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["healthcheck"] = {
|
||||
@ -735,7 +741,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_heathcheck_cmd_shell(self):
|
||||
async def test_healthcheck_cmd_shell(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["healthcheck"] = {
|
||||
@ -755,7 +761,7 @@ class TestContainerToArgs(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_heathcheck_cmd_shell_error(self):
|
||||
async def test_healthcheck_cmd_shell_error(self) -> None:
|
||||
c = create_compose_mock()
|
||||
cnt = get_minimal_container()
|
||||
cnt["healthcheck"] = {
|
||||
|
@ -10,12 +10,12 @@ from tests.unit.test_container_to_args import create_compose_mock
|
||||
from tests.unit.test_container_to_args import get_minimal_container
|
||||
|
||||
|
||||
def repo_root():
|
||||
def repo_root() -> str:
|
||||
return os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
|
||||
|
||||
|
||||
class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_pass_secret_as_env_variable(self):
|
||||
async def test_pass_secret_as_env_variable(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret": {"external": "true"} # must have external or name value
|
||||
@ -43,7 +43,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_secret_as_env_external_true_has_no_name(self):
|
||||
async def test_secret_as_env_external_true_has_no_name(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret": {
|
||||
@ -74,7 +74,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_pass_secret_as_env_variable_no_external(self):
|
||||
async def test_pass_secret_as_env_variable_no_external(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret": {} # must have external or name value
|
||||
@ -124,7 +124,9 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
},
|
||||
),
|
||||
])
|
||||
async def test_secret_name(self, test_name, declared_secrets, add_to_minimal_container):
|
||||
async def test_secret_name(
|
||||
self, test_name: str, declared_secrets: dict, add_to_minimal_container: dict
|
||||
) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = declared_secrets
|
||||
|
||||
@ -136,7 +138,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
await container_to_args(c, cnt)
|
||||
self.assertIn('ERROR: undeclared secret: ', str(context.exception))
|
||||
|
||||
async def test_secret_string_no_external_name_in_declared_secrets(self):
|
||||
async def test_secret_string_no_external_name_in_declared_secrets(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {"my_secret_name": {"external": "true"}}
|
||||
cnt = get_minimal_container()
|
||||
@ -157,7 +159,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_secret_string_options_external_name_in_declared_secrets(self):
|
||||
async def test_secret_string_options_external_name_in_declared_secrets(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret_name": {
|
||||
@ -195,7 +197,9 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_secret_string_external_name_in_declared_secrets_does_not_match_secret(self):
|
||||
async def test_secret_string_external_name_in_declared_secrets_does_not_match_secret(
|
||||
self,
|
||||
) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret_name": {
|
||||
@ -213,7 +217,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
await container_to_args(c, cnt)
|
||||
self.assertIn('ERROR: Custom name/target reference ', str(context.exception))
|
||||
|
||||
async def test_secret_target_does_not_match_secret_name_secret_type_not_env(self):
|
||||
async def test_secret_target_does_not_match_secret_name_secret_type_not_env(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret_name": {
|
||||
@ -234,7 +238,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
await container_to_args(c, cnt)
|
||||
self.assertIn('ERROR: Custom name/target reference ', str(context.exception))
|
||||
|
||||
async def test_secret_target_does_not_match_secret_name_secret_type_env(self):
|
||||
async def test_secret_target_does_not_match_secret_name_secret_type_env(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret_name": {
|
||||
@ -260,7 +264,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_secret_target_matches_secret_name_secret_type_not_env(self):
|
||||
async def test_secret_target_matches_secret_name_secret_type_not_env(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"my_secret_name": {
|
||||
@ -342,8 +346,12 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
),
|
||||
])
|
||||
async def test_file_secret(
|
||||
self, test_name, declared_secrets, add_to_minimal_container, expected_volume_ref
|
||||
):
|
||||
self,
|
||||
test_name: str,
|
||||
declared_secrets: dict,
|
||||
add_to_minimal_container: dict,
|
||||
expected_volume_ref: str,
|
||||
) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = declared_secrets
|
||||
cnt = get_minimal_container()
|
||||
@ -362,7 +370,7 @@ class TestContainerToArgsSecrets(unittest.IsolatedAsyncioTestCase):
|
||||
],
|
||||
)
|
||||
|
||||
async def test_file_secret_unused_params_warning(self):
|
||||
async def test_file_secret_unused_params_warning(self) -> None:
|
||||
c = create_compose_mock()
|
||||
c.declared_secrets = {
|
||||
"file_secret": {
|
||||
|
@ -2,6 +2,7 @@ import unittest
|
||||
|
||||
from parameterized import parameterized
|
||||
|
||||
from podman_compose import PodmanCompose
|
||||
from podman_compose import get_net_args
|
||||
from tests.unit.test_container_to_args import create_compose_mock
|
||||
|
||||
@ -10,7 +11,7 @@ SERVICE_NAME = "service_name"
|
||||
CONTAINER_NAME = f"{PROJECT_NAME}_{SERVICE_NAME}_1"
|
||||
|
||||
|
||||
def get_networked_compose(num_networks=1):
|
||||
def get_networked_compose(num_networks: int = 1) -> PodmanCompose:
|
||||
compose = create_compose_mock(PROJECT_NAME)
|
||||
for network in range(num_networks):
|
||||
compose.networks[f"net{network}"] = {
|
||||
@ -30,7 +31,7 @@ def get_networked_compose(num_networks=1):
|
||||
return compose
|
||||
|
||||
|
||||
def get_minimal_container():
|
||||
def get_minimal_container() -> dict:
|
||||
return {
|
||||
"name": CONTAINER_NAME,
|
||||
"service_name": SERVICE_NAME,
|
||||
@ -39,7 +40,7 @@ def get_minimal_container():
|
||||
|
||||
|
||||
class TestGetNetArgs(unittest.TestCase):
|
||||
def test_minimal(self):
|
||||
def test_minimal(self) -> None:
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
|
||||
@ -49,7 +50,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_default_net_is_None(self):
|
||||
def test_default_net_is_None(self) -> None:
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
|
||||
@ -64,7 +65,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_one_net(self):
|
||||
def test_one_net(self) -> None:
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
container["networks"] = {"net0": {}}
|
||||
@ -75,7 +76,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_alias(self):
|
||||
def test_alias(self) -> None:
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
container["networks"] = {"net0": {}}
|
||||
@ -87,7 +88,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_aliases_on_network_scope(self):
|
||||
def test_aliases_on_network_scope(self) -> None:
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
container["networks"] = {"net0": {"aliases": ["alias1"]}}
|
||||
@ -98,7 +99,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_one_ipv4(self):
|
||||
def test_one_ipv4(self) -> None:
|
||||
ip = "192.168.0.42"
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
@ -110,7 +111,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertEqual(expected_args, args)
|
||||
|
||||
def test_one_ipv6(self):
|
||||
def test_one_ipv6(self) -> None:
|
||||
ipv6_address = "fd00:0::42"
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
@ -122,7 +123,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_one_mac(self):
|
||||
def test_one_mac(self) -> None:
|
||||
mac = "00:11:22:33:44:55"
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
@ -135,7 +136,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_one_mac_two_nets(self):
|
||||
def test_one_mac_two_nets(self) -> None:
|
||||
mac = "00:11:22:33:44:55"
|
||||
compose = get_networked_compose(num_networks=6)
|
||||
container = get_minimal_container()
|
||||
@ -153,7 +154,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
"mac_address",
|
||||
"x-podman.mac_address",
|
||||
])
|
||||
def test_mac_on_network(self, mac_attr):
|
||||
def test_mac_on_network(self, mac_attr: str) -> None:
|
||||
mac = "00:11:22:33:44:55"
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
@ -165,7 +166,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_two_nets_as_dict(self):
|
||||
def test_two_nets_as_dict(self) -> None:
|
||||
compose = get_networked_compose(num_networks=2)
|
||||
container = get_minimal_container()
|
||||
container["networks"] = {"net0": {}, "net1": {}}
|
||||
@ -177,7 +178,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_two_nets_as_list(self):
|
||||
def test_two_nets_as_list(self) -> None:
|
||||
compose = get_networked_compose(num_networks=2)
|
||||
container = get_minimal_container()
|
||||
container["networks"] = ["net0", "net1"]
|
||||
@ -189,7 +190,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_two_ipv4(self):
|
||||
def test_two_ipv4(self) -> None:
|
||||
ip0 = "192.168.0.42"
|
||||
ip1 = "192.168.1.42"
|
||||
compose = get_networked_compose(num_networks=2)
|
||||
@ -203,7 +204,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_two_ipv6(self):
|
||||
def test_two_ipv6(self) -> None:
|
||||
ip0 = "fd00:0::42"
|
||||
ip1 = "fd00:1::42"
|
||||
compose = get_networked_compose(num_networks=2)
|
||||
@ -218,7 +219,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
# custom extension; not supported by docker-compose
|
||||
def test_two_mac(self):
|
||||
def test_two_mac(self) -> None:
|
||||
mac0 = "00:00:00:00:00:01"
|
||||
mac1 = "00:00:00:00:00:02"
|
||||
compose = get_networked_compose(num_networks=2)
|
||||
@ -235,7 +236,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_mixed_mac(self):
|
||||
def test_mixed_mac(self) -> None:
|
||||
ip4_0 = "192.168.0.42"
|
||||
ip4_1 = "192.168.1.42"
|
||||
ip4_2 = "192.168.2.42"
|
||||
@ -256,7 +257,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
)
|
||||
self.assertRaisesRegex(RuntimeError, expected_exception, get_net_args, compose, container)
|
||||
|
||||
def test_mixed_config(self):
|
||||
def test_mixed_config(self) -> None:
|
||||
ip4_0 = "192.168.0.42"
|
||||
ip4_1 = "192.168.1.42"
|
||||
ip6_0 = "fd00:0::42"
|
||||
@ -297,7 +298,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
("ns:my_namespace", ["--network=ns:my_namespace"]),
|
||||
("container:my_container", ["--network=container:my_container"]),
|
||||
])
|
||||
def test_network_modes(self, network_mode, expected_args):
|
||||
def test_network_modes(self, network_mode: str, expected_args: list) -> None:
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
container["network_mode"] = network_mode
|
||||
@ -309,7 +310,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
args = get_net_args(compose, container)
|
||||
self.assertListEqual(expected_args, args)
|
||||
|
||||
def test_network_mode_invalid(self):
|
||||
def test_network_mode_invalid(self) -> None:
|
||||
compose = get_networked_compose()
|
||||
container = get_minimal_container()
|
||||
container["network_mode"] = "invalid_mode"
|
||||
@ -317,7 +318,7 @@ class TestGetNetArgs(unittest.TestCase):
|
||||
with self.assertRaises(SystemExit):
|
||||
get_net_args(compose, container)
|
||||
|
||||
def test_network__mode_service(self):
|
||||
def test_network__mode_service(self) -> None:
|
||||
compose = get_networked_compose()
|
||||
compose.container_names_by_service = {
|
||||
"service_1": ["container_1"],
|
||||
|
@ -4,7 +4,7 @@ from podman_compose import get_network_create_args
|
||||
|
||||
|
||||
class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
def test_minimal(self):
|
||||
def test_minimal(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
@ -26,7 +26,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_ipv6(self):
|
||||
def test_ipv6(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
@ -49,7 +49,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_bridge(self):
|
||||
def test_bridge(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
@ -77,7 +77,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_ipam_driver_default(self):
|
||||
def test_ipam_driver_default(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
@ -113,7 +113,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_ipam_driver(self):
|
||||
def test_ipam_driver(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
@ -151,7 +151,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_complete(self):
|
||||
def test_complete(self) -> None:
|
||||
net_desc = {
|
||||
"labels": ["label1", "label2"],
|
||||
"internal": True,
|
||||
@ -202,7 +202,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_disable_dns(self):
|
||||
def test_disable_dns(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
@ -226,7 +226,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_dns_string(self):
|
||||
def test_dns_string(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
@ -251,7 +251,7 @@ class TestGetNetworkCreateArgs(unittest.TestCase):
|
||||
args = get_network_create_args(net_desc, proj_name, net_name)
|
||||
self.assertEqual(args, expected_args)
|
||||
|
||||
def test_dns_list(self):
|
||||
def test_dns_list(self) -> None:
|
||||
net_desc = {
|
||||
"labels": [],
|
||||
"internal": False,
|
||||
|
@ -18,5 +18,5 @@ class TestIsPathGitUrl(unittest.TestCase):
|
||||
("suffix_and_prefix", "git://host.xz/path/to/repo.git", True),
|
||||
("empty_url_path", "http://#fragment", False),
|
||||
])
|
||||
def test_is_path_git_url(self, test_name, path, result):
|
||||
def test_is_path_git_url(self, test_name: str, path: str, result: bool) -> None:
|
||||
self.assertEqual(is_path_git_url(path), result)
|
||||
|
@ -1,5 +1,7 @@
|
||||
# SPDX-License-Identifier: GPL-2.0
|
||||
import unittest
|
||||
from typing import Any
|
||||
from typing import Union
|
||||
|
||||
from parameterized import parameterized
|
||||
|
||||
@ -29,7 +31,7 @@ class TestNormalizeService(unittest.TestCase):
|
||||
{"build": {"additional_contexts": ["ctx=../ctx", "ctx2=../ctx2"]}},
|
||||
),
|
||||
])
|
||||
def test_simple(self, input, expected):
|
||||
def test_simple(self, input: dict[str, Any], expected: dict[str, Any]) -> None:
|
||||
self.assertEqual(normalize_service(input), expected)
|
||||
|
||||
@parameterized.expand([
|
||||
@ -46,7 +48,9 @@ class TestNormalizeService(unittest.TestCase):
|
||||
{"build": {"context": "./sub_dir/dir-1", "dockerfile": "dockerfile-1"}},
|
||||
),
|
||||
])
|
||||
def test_normalize_service_with_sub_dir(self, input, expected):
|
||||
def test_normalize_service_with_sub_dir(
|
||||
self, input: dict[str, Any], expected: dict[str, Any]
|
||||
) -> None:
|
||||
self.assertEqual(normalize_service(input, sub_dir="./sub_dir"), expected)
|
||||
|
||||
@parameterized.expand([
|
||||
@ -60,7 +64,7 @@ class TestNormalizeService(unittest.TestCase):
|
||||
["bash", "-c", "sleep infinity"],
|
||||
),
|
||||
])
|
||||
def test_command_like(self, input, expected):
|
||||
def test_command_like(self, input: Union[list[str], str], expected: list[str]) -> None:
|
||||
for key in ['command', 'entrypoint']:
|
||||
input_service = {}
|
||||
input_service[key] = input
|
||||
|
@ -2,6 +2,7 @@
|
||||
# pylint: disable=protected-access
|
||||
|
||||
import unittest
|
||||
from typing import Any
|
||||
|
||||
from parameterized import parameterized
|
||||
|
||||
@ -66,7 +67,7 @@ class TestRecSubs(unittest.TestCase):
|
||||
]
|
||||
|
||||
@parameterized.expand(substitutions)
|
||||
def test_rec_subs(self, desc, input, expected):
|
||||
def test_rec_subs(self, desc: str, input: Any, expected: Any) -> None:
|
||||
sub_dict = {"v1": "high priority", "empty": ""}
|
||||
result = rec_subs(input, sub_dict)
|
||||
self.assertEqual(result, expected, msg=desc)
|
||||
|
@ -6,7 +6,7 @@ from podman_compose import parse_short_mount
|
||||
|
||||
|
||||
class ParseShortMountTests(unittest.TestCase):
|
||||
def test_multi_propagation(self):
|
||||
def test_multi_propagation(self) -> None:
|
||||
self.assertEqual(
|
||||
parse_short_mount("/foo/bar:/baz:U,Z", "/"),
|
||||
{
|
||||
|
Loading…
x
Reference in New Issue
Block a user