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