From 875d49fe10b6ba2dd3ec82a45ea58d304419ab2b Mon Sep 17 00:00:00 2001
From: Alex Pyrgiotis <alex.p@freedom.press>
Date: Tue, 9 Apr 2024 13:47:00 +0300
Subject: [PATCH] tests: Add termination tests for containers

Add termination-related tests for containers. To achieve this, we need
to make a change to the container isolation provider. More specifically,
we need to make the isolation provider yield control to the caller only
when the container is up and running. Failure to do so may lead to
lingering processes.
---
 tests/isolation_provider/test_container.py | 43 +++++++++++++++++++++-
 1 file changed, 42 insertions(+), 1 deletion(-)

diff --git a/tests/isolation_provider/test_container.py b/tests/isolation_provider/test_container.py
index a6de19b..33bd8b4 100644
--- a/tests/isolation_provider/test_container.py
+++ b/tests/isolation_provider/test_container.py
@@ -1,12 +1,17 @@
 import itertools
 import json
+import os
+import subprocess
+import time
 from typing import Any, Dict
 
 import pytest
 from pytest_mock import MockerFixture
 
 from dangerzone.document import Document
+from dangerzone.isolation_provider import base
 from dangerzone.isolation_provider.container import Container
+from dangerzone.isolation_provider.qubes import is_qubes_native_conversion
 
 # XXX Fixtures used in abstract Test class need to be imported regardless
 from .. import (
@@ -17,7 +22,7 @@ from .. import (
     sanitized_text,
     uncommon_text,
 )
-from .base import IsolationProviderTest
+from .base import IsolationProviderTermination, IsolationProviderTest
 
 
 @pytest.fixture
@@ -25,5 +30,41 @@ def provider() -> Container:
     return Container()
 
 
+class ContainerWait(Container):
+    """Container isolation provider that blocks until the container has started."""
+
+    def exec_container(self, *args, **kwargs):  # type: ignore [no-untyped-def]
+        # Check every 100ms if a container with the expected name has showed up.
+        # Else, closing the file descriptors may not work.
+        name = kwargs["name"]
+        runtime = self.get_runtime()
+        p = super().exec_container(*args, **kwargs)
+        for i in range(50):
+            containers = subprocess.run(
+                [runtime, "ps"], capture_output=True
+            ).stdout.decode()
+            if name in containers:
+                return p
+            time.sleep(0.1)
+
+        raise RuntimeError(f"Container {name} did not start within 5 seconds")
+
+
+@pytest.fixture
+def provider_wait() -> ContainerWait:
+    return ContainerWait()
+
+
 class TestContainer(IsolationProviderTest):
     pass
+
+
+@pytest.mark.skipif(
+    os.environ.get("DUMMY_CONVERSION", False),
+    reason="cannot run for dummy conversions",
+)
+@pytest.mark.skipif(
+    is_qubes_native_conversion(), reason="Qubes native conversion is enabled"
+)
+class TestContainerTermination(IsolationProviderTermination):
+    pass