diff --git a/Sources/toolkit/bhamon_development_toolkit/python/pytest_output_handler.py b/Sources/toolkit/bhamon_development_toolkit/python/pytest_output_handler.py index 40ccb99..8992416 100644 --- a/Sources/toolkit/bhamon_development_toolkit/python/pytest_output_handler.py +++ b/Sources/toolkit/bhamon_development_toolkit/python/pytest_output_handler.py @@ -23,14 +23,21 @@ def __init__(self, scope: PytestScope) -> None: status_collection = [ "passed", "failed", "skipped", "error", "xpassed", "xfailed" ] status_collection_regex = "|".join(x.upper() for x in status_collection) + self._interrupt_regex = re.compile(r"^!+ Interrupted: (?P[^=]*) !+$") self._test_result_regex = re.compile(r"^(?P.*)\s+(?P" + status_collection_regex + r")\s+\[\s*[0-9]+%\]$") def process_stdout_line(self, line: str) -> None: - self._handle_line_as_test_result(line.rstrip()) + line = line.rstrip() + + self._handle_line_as_interrupt(line) + self._handle_line_as_test_result(line) def process_stderr_line(self, line: str) -> None: + line = line.rstrip() + + self._handle_line_as_interrupt(line) self._handle_line_as_test_result(line.rstrip()) @@ -42,6 +49,12 @@ def process_stderr_end(self) -> None: pass + def _handle_line_as_interrupt(self, line: str) -> None: + interrupt_match = re.search(self._interrupt_regex, line) + if interrupt_match is not None: + logger.error("Interrupted: %s", interrupt_match.group("message")) + + def _handle_line_as_test_result(self, line: str) -> None: test_result = self._parse_test_result(line) if test_result is not None: diff --git a/Sources/toolkit/bhamon_development_toolkit/python/pytest_runner.py b/Sources/toolkit/bhamon_development_toolkit/python/pytest_runner.py index 54a50d7..b644bfd 100644 --- a/Sources/toolkit/bhamon_development_toolkit/python/pytest_runner.py +++ b/Sources/toolkit/bhamon_development_toolkit/python/pytest_runner.py @@ -130,7 +130,7 @@ def _check_exit_code(self, exit_code: Optional[int]) -> None: if exit_code == 1: # Tests were collected and run but some of the tests failed return if exit_code == 2: # Test execution was interrupted by the user - raise KeyboardInterrupt("Pytest execution was interrupted") + raise RuntimeError("Pytest execution was interrupted") if exit_code == 3: # Internal error happened while executing tests raise RuntimeError("Pytest internal error") if exit_code == 4: # pytest command line usage error diff --git a/Tests/toolkit/bhamon_development_toolkit_tests/python/test_pytest_runner_with_integration.py b/Tests/toolkit/bhamon_development_toolkit_tests/python/test_pytest_runner_with_integration.py index ce010d2..ae269ce 100644 --- a/Tests/toolkit/bhamon_development_toolkit_tests/python/test_pytest_runner_with_integration.py +++ b/Tests/toolkit/bhamon_development_toolkit_tests/python/test_pytest_runner_with_integration.py @@ -1,5 +1,8 @@ +# cspell:words caplog + """ Integration tests for PytestRunner """ +import logging import os import sys @@ -66,6 +69,32 @@ async def test_run_with_failure(tmpdir): await pytest_runner.run(all_scopes, run_identifier, result_directory, working_directory = working_directory) +@pytest.mark.asyncio +async def test_run_with_syntax_error(tmpdir, caplog): + python_executable = sys.executable + process_runner = ProcessRunner(ProcessSpawner(is_console = True)) + pytest_runner = PytestRunner(process_runner, python_executable) + + test_directory = os.path.join(tmpdir, "Tests") + + os.makedirs(test_directory) + with open(os.path.join(test_directory, "test_my_module.py"), mode = "w", encoding = "utf-8") as test_file: + test_file.write("def") + + all_scopes = [ PytestScope("All", "Tests", None) ] + run_identifier = "my-run-identifier" + result_directory = os.path.join(tmpdir, "TestResults") + working_directory = str(tmpdir) + + with pytest.raises(RuntimeError): + await pytest_runner.run(all_scopes, run_identifier, result_directory, working_directory = working_directory) + + error_messages = [ record for record in caplog.records if record.levelno == logging.ERROR ] + + assert len(error_messages) == 1 + assert error_messages[0].message == "Interrupted: 1 error during collection" + + @pytest.mark.asyncio async def test_run_with_no_tests(tmpdir): python_executable = sys.executable