Improve tests

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,15 +38,15 @@ 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())
         )
         socket_port = container_data[0]["NetworkSettings"]["Ports"]["2375/tcp"][0][
             "HostPort"
         ]
         with local.env(DOCKER_HOST=f"tcp://localhost:{socket_port}"):
-            yield
+            yield container_id
     finally:
         if container_id:
             info(f"Removing {container_id}...")

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)