diff --git a/src/byteb4rb1e/utils/testing/pytest/__init__.py b/src/byteb4rb1e/utils/testing/pytest/__init__.py new file mode 100644 index 0000000..87e7c10 --- /dev/null +++ b/src/byteb4rb1e/utils/testing/pytest/__init__.py @@ -0,0 +1,14 @@ +import os +from pathlib import Path +from typing import Tuple + + +def get_current_test() -> Tuple[Path, str]: + current_test_env = os.getenv("PYTEST_CURRENT_TEST") + if current_test_env is None: + raise RuntimeError("PYTEST_CURRENT_TEST not set. Must be run under pytest.") + + suite_path, case_name = current_test_env.split('::', 1) + case_name = case_name.split(' ', 1)[0] + return Path(suite_path).resolve(), case_name + diff --git a/src/byteb4rb1e/utils/testing/pytest/decorators.py b/src/byteb4rb1e/utils/testing/pytest/decorators.py new file mode 100644 index 0000000..bf1066b --- /dev/null +++ b/src/byteb4rb1e/utils/testing/pytest/decorators.py @@ -0,0 +1,47 @@ +from functools import wraps +from pathlib import Path +import os +import subprocess +import sys + +from byteb4rb1e.utils.testing.pytest import get_current_test + + +def run_in_subprocess_once(): + """ + A decorator that reruns th test in a subprocess if not already inside one. + Requires pytest to be installed and test to be run by pytest. + + For what? Anything that can't be done in a thread-safe manner, e.g. modifying PYTHON_PATH + """ + def decorator(test_func): + @wraps(test_func) + def wrapper(*args, **kwargs): + if os.environ.get("XPYTEST_INSIDE_SUBPROCESS") == "1": + return test_func(*args, **kwargs) + + suite_path, case_name = get_current_test() + + cmd = [ + sys.executable, + "-m", "pytest", + f"{suite_path}::{case_name}", + ] + + result = subprocess.run( + cmd, + env={**os.environ, "XPYTEST_INSIDE_SUBPROCESS": "1"}, + capture_output=True, + text=True, + ) + + if result.returncode != 0: + print(' '.join(cmd)) + print("==== Subprocess stdout ====") + print(result.stdout) + print("==== Subprocess stderr ====") + print(result.stderr) + raise AssertionError(f"Subprocess test failed with exit code {result.returncode}") + return wrapper + return decorator + diff --git a/src/byteb4rb1e/utils/testing/pytest/fixtures.py b/src/byteb4rb1e/utils/testing/pytest/fixtures.py index 7415a22..dc87091 100644 --- a/src/byteb4rb1e/utils/testing/pytest/fixtures.py +++ b/src/byteb4rb1e/utils/testing/pytest/fixtures.py @@ -4,11 +4,11 @@ from typing import Tuple import pytest +from byteb4rb1e.utils.testing.pytest import get_current_test + @pytest.fixture def current_test() -> Tuple[Path, str]: """ """ - suite_path, case_name = os.getenv('PYTEST_CURRENT_TEST').split('::', 1) - case_name = case_name.split(' ', 1)[0] - return Path(suite_path).resolve(), case_name + return get_current_test() diff --git a/tests/integration/byteb4rb1e/utils/testing/pytest/test_.py b/tests/integration/byteb4rb1e/utils/testing/pytest/test_.py new file mode 100644 index 0000000..57a0410 --- /dev/null +++ b/tests/integration/byteb4rb1e/utils/testing/pytest/test_.py @@ -0,0 +1,33 @@ +import os +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.pytest + +from byteb4rb1e.utils.testing.pytest import get_current_test +from byteb4rb1e.utils.testing.pytest.decorators import run_in_subprocess_once + + +class Test_get_current_test: + """ + """ + + def test_default(self): + """ + """ + os.environ['PYTEST_CURRENT_TEST'] = 'foo::bar (something)' + + result = get_current_test() + + assert isinstance(result[0], Path) + assert str(result[0].name) == 'foo' + + assert result[1] == 'bar' + + def test_invalid(self): + """ + """ + del os.environ['PYTEST_CURRENT_TEST'] + with pytest.raises(RuntimeError): + get_current_test() diff --git a/tests/integration/byteb4rb1e/utils/testing/pytest/test_decorators.py b/tests/integration/byteb4rb1e/utils/testing/pytest/test_decorators.py new file mode 100644 index 0000000..7dd20fe --- /dev/null +++ b/tests/integration/byteb4rb1e/utils/testing/pytest/test_decorators.py @@ -0,0 +1,21 @@ +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.pytest + +from byteb4rb1e.utils.testing.pytest.decorators import run_in_subprocess_once + + +@run_in_subprocess_once() +def test_run_in_subprocess_once(tmp_path): + marker = tmp_path / "executed_in_subprocess.txt" + + if marker.exists(): + raise AssertionError("Marker file exists before test logic ran (shouldn't happen in parent process)") + + # Create proof of execution + marker.write_text("Subprocess was here.") + + # Now assert it + assert marker.exists() diff --git a/tests/unit/byteb4rb1e/utils/urllib/test_request.py b/tests/unit/byteb4rb1e/utils/urllib/test_request.py index 28548ef..74cf40d 100644 --- a/tests/unit/byteb4rb1e/utils/urllib/test_request.py +++ b/tests/unit/byteb4rb1e/utils/urllib/test_request.py @@ -1,5 +1,8 @@ +import pytest + from byteb4rb1e.utils.urllib.request import PkgHandler + class TestPkgHandler: """ """ @@ -7,3 +10,4 @@ class TestPkgHandler: """ """ pass +