mirror of
https://github.com/lexmount/moli.git
synced 2026-09-29 08:01:36 +00:00
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import signal
|
|
import sys
|
|
import unittest
|
|
|
|
from moli_cdp_smoke.assertions import SmokeError
|
|
from moli_cdp_smoke.groups.external_process import run_external_json_process
|
|
from moli_cdp_smoke.process import terminate_process_tree
|
|
|
|
|
|
class ExternalProcessTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_successful_json_process_extends_results(self) -> None:
|
|
expected = {"name": "external-process-probe", "ok": True}
|
|
payload = json.dumps({"ok": True, "results": [expected]})
|
|
results: list[dict[str, object]] = []
|
|
|
|
await run_external_json_process(
|
|
"test",
|
|
[sys.executable, "-c", f"print({payload!r})"],
|
|
results,
|
|
timeout_seconds=5,
|
|
)
|
|
|
|
self.assertEqual(results, [expected])
|
|
|
|
@unittest.skipUnless(os.name == "posix", "process-group regression is POSIX-specific")
|
|
async def test_timeout_kills_descendants_holding_output_pipes(self) -> None:
|
|
descendant = "import time; time.sleep(4)"
|
|
parent = (
|
|
"import subprocess, sys, time; "
|
|
f"subprocess.Popen([sys.executable, '-c', {descendant!r}], "
|
|
"stdout=sys.stdout, stderr=sys.stderr); "
|
|
"print('descendant-started', flush=True); "
|
|
"time.sleep(30)"
|
|
)
|
|
loop = asyncio.get_running_loop()
|
|
started_at = loop.time()
|
|
|
|
with self.assertRaisesRegex(SmokeError, "test smoke timed out") as raised:
|
|
await run_external_json_process(
|
|
"test",
|
|
[sys.executable, "-c", parent],
|
|
[],
|
|
timeout_seconds=1,
|
|
)
|
|
|
|
elapsed = loop.time() - started_at
|
|
self.assertIn("descendant-started", str(raised.exception))
|
|
self.assertLess(
|
|
elapsed,
|
|
2.5,
|
|
"timeout cleanup waited for the descendant to close inherited pipes",
|
|
)
|
|
|
|
@unittest.skipUnless(os.name == "posix", "signal regression is POSIX-specific")
|
|
async def test_service_cleanup_escalates_to_sigkill_with_a_bound(self) -> None:
|
|
process = await asyncio.create_subprocess_exec(
|
|
sys.executable,
|
|
"-c",
|
|
(
|
|
"import signal, time; "
|
|
"signal.signal(signal.SIGTERM, signal.SIG_IGN); "
|
|
"print('ready', flush=True); "
|
|
"time.sleep(30)"
|
|
),
|
|
stdout=asyncio.subprocess.PIPE,
|
|
start_new_session=True,
|
|
)
|
|
self.assertIsNotNone(process.stdout)
|
|
self.assertEqual(await process.stdout.readline(), b"ready\n")
|
|
|
|
stopped = await terminate_process_tree(
|
|
process,
|
|
terminate_timeout_seconds=0.05,
|
|
kill_timeout_seconds=1,
|
|
)
|
|
|
|
self.assertTrue(stopped)
|
|
self.assertEqual(process.returncode, -signal.SIGKILL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|