-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- revamped pytest isolation module, now running the test suite
a lot more cleanly while isolating the collection phase to work around any in-memory state pollution;
- Loading branch information
1 parent
406d98c
commit 81fb131
Showing
2 changed files
with
121 additions
and
72 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,66 +1,68 @@ | ||
| import os | ||
| import pytest | ||
| import sys | ||
| from pathlib import Path | ||
|
|
||
|
|
||
| class IsolatePlugin: | ||
| """Pytest plugin to isolate test collection, so that if a test's collection pollutes the in-memory | ||
| state, it doesn't affect the execution of other tests.""" | ||
|
|
||
| def __init__(self): | ||
| self._is_child = False | ||
| self._test_failed = False | ||
| class IsolateItem(pytest.Item): | ||
| def __init__(self, *args, **kwargs): | ||
| super().__init__(*args, **kwargs) | ||
|
|
||
| def pytest_ignore_collect(self, path, config): | ||
| if self._is_child: | ||
| return True # only one test module per child process | ||
| def runtest(self): | ||
| assert False, "should never execute" | ||
|
|
||
| if not str(path).endswith('.py'): | ||
| return False # only fork for test modules, not directories, etc. | ||
| def run_forked(self): | ||
| # adapted from pytest-forked | ||
| import marshal | ||
| import pytest_forked as ptf | ||
| import _pytest | ||
| import py | ||
|
|
||
| # FIXME manage stdout/stderr to avoid children clobbering output, | ||
| # FIXME properly report executed tests | ||
| ihook = self.ihook | ||
| ihook.pytest_runtest_logstart(nodeid=self.nodeid, location=self.location) | ||
|
|
||
| if (pid := os.fork()): | ||
| pid, status = os.waitpid(pid, 0) | ||
| if status: | ||
| if os.WIFSIGNALED(status): | ||
| exitstatus = os.WTERMSIG(status) + 128 | ||
| else: | ||
| exitstatus = os.WEXITSTATUS(status) | ||
| else: | ||
| exitstatus = 0 | ||
| def runforked(): | ||
| module = pytest.Module.from_parent(parent=self.parent, path=self.path) | ||
| reports = list() | ||
| for it in module.collect(): | ||
| reports.extend(ptf.forked_run_report(it)) | ||
| print(reports) | ||
| return marshal.dumps([ptf.serialize_report(x) for x in reports]) | ||
|
|
||
| if exitstatus not in (pytest.ExitCode.OK, pytest.ExitCode.NO_TESTS_COLLECTED): | ||
| self._test_failed = True | ||
| ff = py.process.ForkedFunc(runforked) | ||
| result = ff.waitfinish() | ||
|
|
||
| return True | ||
| if result.retval is not None: | ||
| reports = [_pytest.runner.TestReport(**r) for r in marshal.loads(result.retval)] | ||
| else: | ||
| self._is_child = True | ||
| return False | ||
| reports = [ptf.report_process_crash(self, result)] | ||
|
|
||
| def pytest_collectreport(self, report): | ||
| if self._is_child and report.failed and report.nodeid.endswith('.py'): | ||
| self._test_failed = True | ||
| for r in reports: | ||
| ihook.pytest_runtest_logreport(report=r) | ||
|
|
||
| def pytest_runtest_logreport(self, report): | ||
| if self._is_child and report.failed: | ||
| self._test_failed = True | ||
| ihook.pytest_runtest_logfinish(nodeid=self.nodeid, location=self.location) | ||
|
|
||
| def pytest_unconfigure(self, config): | ||
| if self._is_child: | ||
| os._exit(self.get_exit_code()) | ||
|
|
||
| def get_exit_code(self): | ||
| # FIXME this error handling is very simplistic, extend to other cases | ||
| return pytest.ExitCode.TESTS_FAILED if self._test_failed else pytest.ExitCode.OK | ||
| class IsolateModule(pytest.Module): | ||
| def __init__(self, *args, **kwargs): | ||
| super().__init__(*args, **kwargs) | ||
|
|
||
| def collect(self): | ||
| yield IsolateItem.from_parent(parent=self, name="(module)") | ||
|
|
||
| if __name__ == "__main__": | ||
| plugin = IsolatePlugin() | ||
| exitcode = pytest.main(sys.argv[1:] + ['--forked'], plugins=[plugin]) | ||
| if exitcode in (pytest.ExitCode.OK, pytest.ExitCode.NO_TESTS_COLLECTED): | ||
| exitcode = plugin.get_exit_code() | ||
|
|
||
| sys.exit(exitcode) | ||
| class IsolatePlugin: | ||
| """Pytest plugin to isolate test collection, so that if a test's collection pollutes the in-memory | ||
| state, it doesn't affect the execution of other tests.""" | ||
|
|
||
| @pytest.hookimpl(tryfirst=True) | ||
| def pytest_pycollect_makemodule(self, module_path, parent): | ||
| return IsolateModule.from_parent(parent, path=module_path) | ||
|
|
||
| @pytest.hookimpl(tryfirst=True) | ||
| def pytest_runtestloop(self, session): | ||
| for item in session.items: | ||
| item.run_forked() | ||
| return True | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| sys.exit(pytest.main(sys.argv[1:], plugins=[IsolatePlugin()])) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters