From 3cf4ee6d6648e30069e006aa715edf5822213356 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Marques?= Date: Wed, 2 Dec 2020 14:25:50 +0000 Subject: [PATCH] Improve tests --- tests/conftest.py | 18 ++------ tests/test_service.py | 102 ++++++++++++++++++++++++------------------ 2 files changed, 62 insertions(+), 58 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 0b76054..69a779b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -18,18 +18,6 @@ def docker_image(): return IMAGE_NAME -@pytest.fixture() -def sleeping_container(): - """Launch a test container that will last alive as long as the test.""" - try: - container = docker( - "container", "run", "--rm", "--detach", "alpine", "sleep", "3600" - ).strip() - yield container - finally: - docker("container", "rm", "--force", container) - - @contextmanager def proxy(**env_vars): """A context manager that starts the proxy with the specified env. @@ -39,7 +27,7 @@ def proxy(**env_vars): """ container_id = None env_list = [f"--env={key}={value}" for key, value in env_vars.items()] - info(f"Starting {IMAGE_NAME} container with: {env_vars.join(' ')}") + info(f"Starting {IMAGE_NAME} container with: {env_list}") try: container_id = docker( "container", @@ -50,7 +38,7 @@ def proxy(**env_vars): "--volume=/var/run/docker.sock:/var/run/docker.sock", *env_list, IMAGE_NAME, - ) + ).strip() container_data = json.loads( docker("container", "inspect", container_id.strip()) ) @@ -58,7 +46,7 @@ def proxy(**env_vars): "HostPort" ] with local.env(DOCKER_HOST=f"tcp://localhost:{socket_port}"): - yield + yield container_id finally: if container_id: info(f"Removing {container_id}...") diff --git a/tests/test_service.py b/tests/test_service.py index 6857b10..3b4d95b 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1,62 +1,78 @@ import logging import pytest +from conftest import proxy from plumbum import ProcessExecutionError from plumbum.cmd import docker -from .conftest import proxy - logger = logging.getLogger() -def test_default_permissions(sleeping_container): - allowed_calls = (("version",),) - forbidden_calls = ( - ("pull", "alpine"), - ("--rm", "alpine", "--name", sleeping_container), - ("logs", sleeping_container), - ("wait", sleeping_container), - ("rm", "-f", sleeping_container), - ("restart", sleeping_container), - ("network", "ls"), - ("config", "ls"), - ("service", "ls"), - ("stack", "ls"), - ("secret", "ls"), - ("plugin", "ls"), - ("info",), - ("system", "info"), - ("build", "."), - ("swarm", "init"), - ) - with proxy(): - for args in allowed_calls: +def _check_permissions(allowed_calls, forbidden_calls): + for args in allowed_calls: + docker(*args) + for args in forbidden_calls: + with pytest.raises(ProcessExecutionError): docker(*args) - for args in forbidden_calls: - with pytest.raises(ProcessExecutionError): - docker(*args) + + +def test_default_permissions(): + with proxy() as test_container: + allowed_calls = (("version",),) + forbidden_calls = ( + ("pull", "alpine"), + ("--rm", "alpine", "--name", test_container), + ("logs", test_container), + ("wait", test_container), + ("rm", "-f", test_container), + ("restart", test_container), + ("network", "ls"), + ("config", "ls"), + ("service", "ls"), + ("stack", "ls"), + ("secret", "ls"), + ("plugin", "ls"), + ("info",), + ("system", "info"), + ("build", "."), + ("swarm", "init"), + ) + _check_permissions(allowed_calls, forbidden_calls) def test_container_permissions(): - with _docker_proxy(CONTAINERS=1) as (docker, test_container): - _query_docker_with_proxy("logs", test_container, allowed=True) - _query_docker_with_proxy("inspect", test_container, allowed=True) - _query_docker_with_proxy("wait", test_container, allowed=False) - _query_docker_with_proxy("run", "--rm", "alpine", allowed=False) - _query_docker_with_proxy("rm", "-f", test_container, allowed=False) - _query_docker_with_proxy("restart", test_container, allowed=False) + with proxy(CONTAINERS=1) as test_container: + allowed_calls = [ + ("logs", test_container), + ("inspect", test_container), + ] + forbidden_calls = [ + ("wait", test_container), + ("run", "--rm", "alpine"), + ("rm", "-f", test_container), + ("restart", test_container), + ] + _check_permissions(allowed_calls, forbidden_calls) def test_post_permissions(): - with _docker_proxy(POST=1) as (docker, test_container): - _query_docker_with_proxy("rm", "-f", test_container, allowed=False) - _query_docker_with_proxy("pull", "alpine", allowed=False) - _query_docker_with_proxy("run", "--rm", "alpine", allowed=False) - _query_docker_with_proxy("network", "create", "foobar", allowed=False) + with proxy(POST=1) as test_container: + allowed_calls = [] + forbidden_calls = [ + ("rm", "-f", test_container), + ("pull", "alpine"), + ("run", "--rm", "alpine"), + ("network", "create", "foobar"), + ] + _check_permissions(allowed_calls, forbidden_calls) def test_network_post_permissions(): - with _docker_proxy(POST=1, NETWORKS=1) as (docker, test_container): - _query_docker_with_proxy("network", "ls", allowed=True) - _query_docker_with_proxy("network", "create", "foo", allowed=True) - _query_docker_with_proxy("network", "rm", "foo", allowed=True) + with proxy(POST=1, NETWORKS=1): + allowed_calls = [ + ("network", "ls"), + ("network", "create", "foo"), + ("network", "rm", "foo"), + ] + forbidden_calls = [] + _check_permissions(allowed_calls, forbidden_calls)