#!/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()
Codebase Browser
llvm