1
0
mirror of https://github.com/Tecnativa/docker-socket-proxy synced 2025-01-03 04:10:55 +00:00

Improve tests

This commit is contained in:
João Marques 2020-12-02 14:25:50 +00:00
parent 91c06522f3
commit 3cf4ee6d66
2 changed files with 62 additions and 58 deletions

View File

@ -18,18 +18,6 @@ def docker_image():
return IMAGE_NAME 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 @contextmanager
def proxy(**env_vars): def proxy(**env_vars):
"""A context manager that starts the proxy with the specified env. """A context manager that starts the proxy with the specified env.
@ -39,7 +27,7 @@ def proxy(**env_vars):
""" """
container_id = None container_id = None
env_list = [f"--env={key}={value}" for key, value in env_vars.items()] 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: try:
container_id = docker( container_id = docker(
"container", "container",
@ -50,7 +38,7 @@ def proxy(**env_vars):
"--volume=/var/run/docker.sock:/var/run/docker.sock", "--volume=/var/run/docker.sock:/var/run/docker.sock",
*env_list, *env_list,
IMAGE_NAME, IMAGE_NAME,
) ).strip()
container_data = json.loads( container_data = json.loads(
docker("container", "inspect", container_id.strip()) docker("container", "inspect", container_id.strip())
) )
@ -58,7 +46,7 @@ def proxy(**env_vars):
"HostPort" "HostPort"
] ]
with local.env(DOCKER_HOST=f"tcp://localhost:{socket_port}"): with local.env(DOCKER_HOST=f"tcp://localhost:{socket_port}"):
yield yield container_id
finally: finally:
if container_id: if container_id:
info(f"Removing {container_id}...") info(f"Removing {container_id}...")

View File

@ -1,62 +1,78 @@
import logging import logging
import pytest import pytest
from conftest import proxy
from plumbum import ProcessExecutionError from plumbum import ProcessExecutionError
from plumbum.cmd import docker from plumbum.cmd import docker
from .conftest import proxy
logger = logging.getLogger() logger = logging.getLogger()
def test_default_permissions(sleeping_container): def _check_permissions(allowed_calls, forbidden_calls):
allowed_calls = (("version",),) for args in allowed_calls:
forbidden_calls = ( docker(*args)
("pull", "alpine"), for args in forbidden_calls:
("--rm", "alpine", "--name", sleeping_container), with pytest.raises(ProcessExecutionError):
("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:
docker(*args) 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(): def test_container_permissions():
with _docker_proxy(CONTAINERS=1) as (docker, test_container): with proxy(CONTAINERS=1) as test_container:
_query_docker_with_proxy("logs", test_container, allowed=True) allowed_calls = [
_query_docker_with_proxy("inspect", test_container, allowed=True) ("logs", test_container),
_query_docker_with_proxy("wait", test_container, allowed=False) ("inspect", test_container),
_query_docker_with_proxy("run", "--rm", "alpine", allowed=False) ]
_query_docker_with_proxy("rm", "-f", test_container, allowed=False) forbidden_calls = [
_query_docker_with_proxy("restart", test_container, allowed=False) ("wait", test_container),
("run", "--rm", "alpine"),
("rm", "-f", test_container),
("restart", test_container),
]
_check_permissions(allowed_calls, forbidden_calls)
def test_post_permissions(): def test_post_permissions():
with _docker_proxy(POST=1) as (docker, test_container): with proxy(POST=1) as test_container:
_query_docker_with_proxy("rm", "-f", test_container, allowed=False) allowed_calls = []
_query_docker_with_proxy("pull", "alpine", allowed=False) forbidden_calls = [
_query_docker_with_proxy("run", "--rm", "alpine", allowed=False) ("rm", "-f", test_container),
_query_docker_with_proxy("network", "create", "foobar", allowed=False) ("pull", "alpine"),
("run", "--rm", "alpine"),
("network", "create", "foobar"),
]
_check_permissions(allowed_calls, forbidden_calls)
def test_network_post_permissions(): def test_network_post_permissions():
with _docker_proxy(POST=1, NETWORKS=1) as (docker, test_container): with proxy(POST=1, NETWORKS=1):
_query_docker_with_proxy("network", "ls", allowed=True) allowed_calls = [
_query_docker_with_proxy("network", "create", "foo", allowed=True) ("network", "ls"),
_query_docker_with_proxy("network", "rm", "foo", allowed=True) ("network", "create", "foo"),
("network", "rm", "foo"),
]
forbidden_calls = []
_check_permissions(allowed_calls, forbidden_calls)