60 lines
2.8 KiB
Python
60 lines
2.8 KiB
Python
"""Keep proxy integration failures and interrupts inside their resource envelope."""
|
|
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
import subprocess
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
spec = importlib.util.spec_from_file_location("proxyRunner", Path(__file__).with_name("verifyHTTP2Proxy.py"))
|
|
runnerModule = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(runnerModule)
|
|
|
|
|
|
class ContainerCleanupTests(unittest.TestCase):
|
|
def testFailedAndInterruptedStagesRemoveOnlyTheirContainer(self):
|
|
failures = (subprocess.TimeoutExpired("docker", 1), KeyboardInterrupt(), subprocess.CompletedProcess([], 7))
|
|
for failure in failures:
|
|
with self.subTest(failure=type(failure).__name__):
|
|
calls = []
|
|
|
|
def execute(command, **kwargs):
|
|
calls.append(command)
|
|
if command[1] == "run":
|
|
if isinstance(failure, BaseException):
|
|
raise failure
|
|
return failure
|
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
|
|
runner = runnerModule.ContainerRunner("/fake/docker", io.StringIO(), time.monotonic() + 10)
|
|
with patch.object(runnerModule.subprocess, "run", side_effect=execute):
|
|
with self.assertRaises((subprocess.TimeoutExpired, KeyboardInterrupt, RuntimeError)):
|
|
runner.container("owned-proxy-check", ["nginx:test"])
|
|
self.assertEqual(calls[1:], [["/fake/docker", "rm", "--force", "owned-proxy-check"]])
|
|
self.assertEqual(runner.cleaned, ["owned-proxy-check"])
|
|
|
|
def testCleanupFailureRemainsAnError(self):
|
|
runner = runnerModule.ContainerRunner("/fake/docker", io.StringIO(), time.monotonic() + 10)
|
|
with patch.object(runnerModule.subprocess, "run", side_effect=[
|
|
subprocess.CompletedProcess([], 0),
|
|
subprocess.CompletedProcess([], 1, "", "Docker daemon unavailable"),
|
|
]):
|
|
with self.assertRaisesRegex(RuntimeError, "could not clean up"):
|
|
runner.container("owned-proxy-check", ["nginx:test"])
|
|
self.assertEqual(runner.cleaned, [])
|
|
|
|
def testDeadlinePreventsStartingAnotherContainer(self):
|
|
runner = runnerModule.ContainerRunner("/fake/docker", io.StringIO(), time.monotonic() - 1)
|
|
with patch.object(runnerModule.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, "", "")) as execute:
|
|
with self.assertRaises(TimeoutError):
|
|
runner.container("not-started", ["nginx:test"])
|
|
self.assertEqual(execute.call_args.args[0], ["/fake/docker", "rm", "--force", "not-started"])
|
|
self.assertEqual(execute.call_count, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|