a353e1c683
All checks were successful
CI / Lint, PostgreSQL tests, and package (push) Successful in 12s
81 lines
3.5 KiB
Python
81 lines
3.5 KiB
Python
import io
|
|
import os
|
|
import select
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from car_tracker import container_entrypoint
|
|
|
|
|
|
class ContainerEntrypointTests(unittest.TestCase):
|
|
def test_requires_a_command(self):
|
|
with patch("sys.argv", ["entrypoint"]), patch("sys.stderr", new_callable=io.StringIO):
|
|
self.assertEqual(container_entrypoint.main(), 3)
|
|
|
|
def test_missing_display_has_distinct_safe_error(self):
|
|
error = FileNotFoundError(2, "missing", "/private/display")
|
|
with patch("sys.argv", ["entrypoint", "command"]), \
|
|
patch("sys.stderr", new_callable=io.StringIO) as stderr, \
|
|
patch.object(container_entrypoint.subprocess, "Popen", side_effect=error):
|
|
self.assertEqual(container_entrypoint.main(), 3)
|
|
self.assertIn("Virtual display executable could not be found", stderr.getvalue())
|
|
self.assertNotIn("/private/display", stderr.getvalue())
|
|
|
|
def test_missing_command_has_distinct_safe_error_and_cleans_display(self):
|
|
display = Mock()
|
|
error = FileNotFoundError(2, "missing", "/private/command")
|
|
with patch("sys.argv", ["entrypoint", "/private/command"]), \
|
|
patch("sys.stderr", new_callable=io.StringIO) as stderr, \
|
|
patch.object(container_entrypoint.subprocess, "Popen", side_effect=[display, error]), \
|
|
patch.object(container_entrypoint.select, "select", return_value=([3], [], [])), \
|
|
patch.object(container_entrypoint.os, "read", return_value=b"0\n"):
|
|
self.assertEqual(container_entrypoint.main(), 3)
|
|
self.assertIn("Container command could not be found", stderr.getvalue())
|
|
self.assertNotIn("/private/command", stderr.getvalue())
|
|
display.terminate.assert_called_once()
|
|
display.wait.assert_called_once()
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("RUN_BROWSER_TESTS") == "1", "Xvfb container tests are opt-in")
|
|
class VirtualDisplayTests(unittest.TestCase):
|
|
def test_child_exit_code_is_preserved(self):
|
|
result = subprocess.run([
|
|
sys.executable, "-m", "car_tracker.container_entrypoint",
|
|
sys.executable, "-c", "raise SystemExit(7)",
|
|
], capture_output=True, text=True, timeout=20)
|
|
self.assertEqual(result.returncode, 7, result.stderr)
|
|
|
|
def test_sigterm_waits_for_child_finalization(self):
|
|
command = """
|
|
import signal, sys, time
|
|
def stop(signum, frame):
|
|
print('cleanup-start', flush=True)
|
|
time.sleep(1)
|
|
print('cleanup-done', flush=True)
|
|
sys.exit(130)
|
|
signal.signal(signal.SIGTERM, stop)
|
|
print('ready', flush=True)
|
|
while True:
|
|
signal.pause()
|
|
"""
|
|
process = subprocess.Popen([
|
|
sys.executable, "-m", "car_tracker.container_entrypoint",
|
|
sys.executable, "-c", command,
|
|
], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
try:
|
|
ready, _, _ = select.select([process.stdout], [], [], 15)
|
|
self.assertTrue(ready, "Child did not become ready")
|
|
self.assertEqual(process.stdout.readline().strip(), "ready")
|
|
process.send_signal(signal.SIGTERM)
|
|
stdout, stderr = process.communicate(timeout=15)
|
|
self.assertEqual(process.returncode, 130, stderr)
|
|
self.assertIn("cleanup-start", stdout)
|
|
self.assertIn("cleanup-done", stdout)
|
|
finally:
|
|
if process.poll() is None:
|
|
process.terminate()
|
|
process.communicate(timeout=15)
|