2238dcc393
This is an ongoing series of commits that are reformatting our Python code. Reformatting is done with `black` (23.1.0). If you end up having problems merging this commit because you have made changes to a python file, the best way to handle that is to run `git checkout --ours <yourfile>` and then reformat it with black. RFC: https://discourse.llvm.org/t/rfc-document-and-standardize-python-code-style Differential revision: https://reviews.llvm.org/D151460
234 lines
8.1 KiB
Python
Executable File
234 lines
8.1 KiB
Python
Executable File
#!/usr/bin/env python
|
|
"""
|
|
Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
|
|
See https://llvm.org/LICENSE.txt for license information.
|
|
SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
|
|
|
|
Provides classes used by the test results reporting infrastructure
|
|
within the LLDB test suite.
|
|
|
|
|
|
Tests the process_control module.
|
|
"""
|
|
|
|
|
|
# System imports.
|
|
import os
|
|
import os.path
|
|
import unittest
|
|
import sys
|
|
import threading
|
|
|
|
# Our imports.
|
|
from test_runner import process_control
|
|
|
|
|
|
class TestInferiorDriver(process_control.ProcessDriver):
|
|
def __init__(self, soft_terminate_timeout=None):
|
|
super(TestInferiorDriver, self).__init__(
|
|
soft_terminate_timeout=soft_terminate_timeout
|
|
)
|
|
self.started_event = threading.Event()
|
|
self.started_event.clear()
|
|
|
|
self.completed_event = threading.Event()
|
|
self.completed_event.clear()
|
|
|
|
self.was_timeout = False
|
|
self.returncode = None
|
|
self.output = None
|
|
|
|
def write(self, content):
|
|
# We'll swallow this to keep tests non-noisy.
|
|
# Uncomment the following line if you want to see it.
|
|
# sys.stdout.write(content)
|
|
pass
|
|
|
|
def on_process_started(self):
|
|
self.started_event.set()
|
|
|
|
def on_process_exited(self, command, output, was_timeout, exit_status):
|
|
self.returncode = exit_status
|
|
self.was_timeout = was_timeout
|
|
self.output = output
|
|
self.returncode = exit_status
|
|
self.completed_event.set()
|
|
|
|
|
|
class ProcessControlTests(unittest.TestCase):
|
|
@classmethod
|
|
def _suppress_soft_terminate(cls, command):
|
|
# Do the right thing for your platform here.
|
|
# Right now only POSIX-y systems are reporting
|
|
# soft terminate support, so this is set up for
|
|
# those.
|
|
helper = process_control.ProcessHelper.process_helper()
|
|
signals = helper.soft_terminate_signals()
|
|
if signals is not None:
|
|
for signum in helper.soft_terminate_signals():
|
|
command.extend(["--ignore-signal", str(signum)])
|
|
|
|
@classmethod
|
|
def inferior_command(cls, ignore_soft_terminate=False, options=None):
|
|
# Base command.
|
|
script_name = "{}/inferior.py".format(os.path.dirname(__file__))
|
|
if not os.path.exists(script_name):
|
|
raise Exception(
|
|
"test inferior python script not found: {}".format(script_name)
|
|
)
|
|
command = [sys.executable, script_name]
|
|
|
|
if ignore_soft_terminate:
|
|
cls._suppress_soft_terminate(command)
|
|
|
|
# Handle options as string or list.
|
|
if isinstance(options, str):
|
|
command.extend(options.split())
|
|
elif isinstance(options, list):
|
|
command.extend(options)
|
|
|
|
# Return full command.
|
|
return command
|
|
|
|
|
|
class ProcessControlNoTimeoutTests(ProcessControlTests):
|
|
"""Tests the process_control module."""
|
|
|
|
def test_run_completes(self):
|
|
"""Test that running completes and gets expected stdout/stderr."""
|
|
driver = TestInferiorDriver()
|
|
driver.run_command(self.inferior_command())
|
|
self.assertTrue(driver.completed_event.wait(5), "process failed to complete")
|
|
self.assertEqual(driver.returncode, 0, "return code does not match")
|
|
|
|
def test_run_completes_with_code(self):
|
|
"""Test that running completes and gets expected stdout/stderr."""
|
|
driver = TestInferiorDriver()
|
|
driver.run_command(self.inferior_command(options="-r10"))
|
|
self.assertTrue(driver.completed_event.wait(5), "process failed to complete")
|
|
self.assertEqual(driver.returncode, 10, "return code does not match")
|
|
|
|
|
|
class ProcessControlTimeoutTests(ProcessControlTests):
|
|
def test_run_completes(self):
|
|
"""Test that running completes and gets expected return code."""
|
|
driver = TestInferiorDriver()
|
|
timeout_seconds = 5
|
|
driver.run_command_with_timeout(
|
|
self.inferior_command(), "{}s".format(timeout_seconds), False
|
|
)
|
|
self.assertTrue(
|
|
driver.completed_event.wait(2 * timeout_seconds),
|
|
"process failed to complete",
|
|
)
|
|
self.assertEqual(driver.returncode, 0)
|
|
|
|
def _soft_terminate_works(self, with_core):
|
|
# Skip this test if the platform doesn't support soft ti
|
|
helper = process_control.ProcessHelper.process_helper()
|
|
if not helper.supports_soft_terminate():
|
|
self.skipTest("soft terminate not supported by platform")
|
|
|
|
driver = TestInferiorDriver()
|
|
timeout_seconds = 5
|
|
|
|
driver.run_command_with_timeout(
|
|
# Sleep twice as long as the timeout interval. This
|
|
# should force a timeout.
|
|
self.inferior_command(options="--sleep {}".format(timeout_seconds * 2)),
|
|
"{}s".format(timeout_seconds),
|
|
with_core,
|
|
)
|
|
|
|
# We should complete, albeit with a timeout.
|
|
self.assertTrue(
|
|
driver.completed_event.wait(2 * timeout_seconds),
|
|
"process failed to complete",
|
|
)
|
|
|
|
# Ensure we received a timeout.
|
|
self.assertTrue(driver.was_timeout, "expected to end with a timeout")
|
|
|
|
self.assertTrue(
|
|
helper.was_soft_terminate(driver.returncode, with_core),
|
|
(
|
|
"timeout didn't return expected returncode "
|
|
"for soft terminate with core: {}"
|
|
).format(driver.returncode),
|
|
)
|
|
|
|
def test_soft_terminate_works_core(self):
|
|
"""Driver uses soft terminate (with core request) when process times out."""
|
|
self._soft_terminate_works(True)
|
|
|
|
def test_soft_terminate_works_no_core(self):
|
|
"""Driver uses soft terminate (no core request) when process times out."""
|
|
self._soft_terminate_works(False)
|
|
|
|
def test_hard_terminate_works(self):
|
|
"""Driver falls back to hard terminate when soft terminate is ignored."""
|
|
|
|
driver = TestInferiorDriver(soft_terminate_timeout=2.0)
|
|
timeout_seconds = 1
|
|
|
|
driver.run_command_with_timeout(
|
|
# Sleep much longer than the timeout interval,forcing a
|
|
# timeout. Do whatever is needed to have the inferior
|
|
# ignore soft terminate calls.
|
|
self.inferior_command(ignore_soft_terminate=True, options="--never-return"),
|
|
"{}s".format(timeout_seconds),
|
|
True,
|
|
)
|
|
|
|
# We should complete, albeit with a timeout.
|
|
self.assertTrue(driver.completed_event.wait(60), "process failed to complete")
|
|
|
|
# Ensure we received a timeout.
|
|
self.assertTrue(driver.was_timeout, "expected to end with a timeout")
|
|
|
|
helper = process_control.ProcessHelper.process_helper()
|
|
self.assertTrue(
|
|
helper.was_hard_terminate(driver.returncode),
|
|
(
|
|
"timeout didn't return expected returncode "
|
|
"for hard teriminate: {} ({})"
|
|
).format(driver.returncode, driver.output),
|
|
)
|
|
|
|
def test_inferior_exits_with_live_child_shared_handles(self):
|
|
"""inferior exit detected when inferior children are live with shared
|
|
stdout/stderr handles.
|
|
"""
|
|
# Requires review D13362 or equivalent to be implemented.
|
|
self.skipTest("http://reviews.llvm.org/D13362")
|
|
|
|
driver = TestInferiorDriver()
|
|
|
|
# Create the inferior (I1), and instruct it to create a child (C1)
|
|
# that shares the stdout/stderr handles with the inferior.
|
|
# C1 will then loop forever.
|
|
driver.run_command_with_timeout(
|
|
self.inferior_command(
|
|
options="--launch-child-share-handles --return-code 3"
|
|
),
|
|
"5s",
|
|
False,
|
|
)
|
|
|
|
# We should complete without a timetout. I1 should end
|
|
# immediately after launching C1.
|
|
self.assertTrue(driver.completed_event.wait(5), "process failed to complete")
|
|
|
|
# Ensure we didn't receive a timeout.
|
|
self.assertFalse(driver.was_timeout, "inferior should have completed normally")
|
|
|
|
self.assertEqual(
|
|
driver.returncode,
|
|
3,
|
|
"expected inferior process to end with expected returncode",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|