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.
This commit is contained in:
Alex Pyrgiotis 2024-04-09 13:47:00 +03:00
parent fec7609547
commit 875d49fe10
No known key found for this signature in database
GPG key ID: B6C15EBA0357C9AA

View file

@ -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