Class ExecuteTest

app/execute_test.py:52–116

class ExecuteTest(unittest.TestCase):

    def test_process_with_result_child_completed(self):
        process = execute.ProcessWithResult(target=do_nothing, daemon=True)
        process.start()
        process.join()
        result = process.result()
        self.assertTrue(result.was_successful())
        self.assertEqual(
            execute.ProcessResult(return_value=None, exception=None), result)

    def test_process_with_result_child_not_completed(self):
        process = execute.ProcessWithResult(target=sleep_1_second, daemon=True)
        process.start()
        # Get the result before the child process has completed.
        self.assertIsNone(process.result())

        # Clean up the running child process.
        process.kill()

    def test_process_with_result_child_exception(self):
        # Silence stderr while the child exception is being raised to avoid
        # polluting the terminal output.
        with silence_stderr():
            process = execute.ProcessWithResult(target=raise_exception,
                                                daemon=True)
            process.start()
            process.join()
        result = process.result()
        self.assertFalse(result.was_successful())
        self.assertEqual(
            execute.ProcessResult(return_value=None, exception=mock.ANY),
            result)
        self.assertEqual('Child exception', str(result.exception))

    def test_process_with_result_return_value(self):
        process = execute.ProcessWithResult(target=return_string, daemon=True)
        process.start()
        process.join()
        result = process.result()
        self.assertTrue(result.was_successful())
        self.assertEqual(
            execute.ProcessResult(return_value='Done!', exception=None), result)

    def test_execute_with_timeout_and_timeout_reached(self):
        with self.assertRaises(TimeoutError):
            execute.with_timeout(sleep_1_second, timeout_in_seconds=0.5)

    def test_execute_with_timeout_return_value(self):
        return_value = execute.with_timeout(return_string,
                                            timeout_in_seconds=0.5)
        self.assertEqual('Done!', return_value)

    def test_execute_with_timeout_child_exception(self):
        with silence_stderr():
            with self.assertRaises(Exception) as ctx:
                execute.with_timeout(raise_exception, timeout_in_seconds=0.5)
        self.assertEqual('Child exception', str(ctx.exception))