feat(testing.pytest): add subprocess decorator
This commit is contained in:
parent
1ea3b3a24d
commit
43cdf21d4b
6 changed files with 122 additions and 3 deletions
14
src/byteb4rb1e/utils/testing/pytest/__init__.py
Normal file
14
src/byteb4rb1e/utils/testing/pytest/__init__.py
Normal file
|
|
@ -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
|
||||||
|
|
||||||
47
src/byteb4rb1e/utils/testing/pytest/decorators.py
Normal file
47
src/byteb4rb1e/utils/testing/pytest/decorators.py
Normal file
|
|
@ -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
|
||||||
|
|
||||||
|
|
@ -4,11 +4,11 @@ from typing import Tuple
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
from byteb4rb1e.utils.testing.pytest import get_current_test
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def current_test() -> Tuple[Path, str]:
|
def current_test() -> Tuple[Path, str]:
|
||||||
"""
|
"""
|
||||||
"""
|
"""
|
||||||
suite_path, case_name = os.getenv('PYTEST_CURRENT_TEST').split('::', 1)
|
return get_current_test()
|
||||||
case_name = case_name.split(' ', 1)[0]
|
|
||||||
return Path(suite_path).resolve(), case_name
|
|
||||||
|
|
|
||||||
33
tests/integration/byteb4rb1e/utils/testing/pytest/test_.py
Normal file
33
tests/integration/byteb4rb1e/utils/testing/pytest/test_.py
Normal file
|
|
@ -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()
|
||||||
|
|
@ -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()
|
||||||
|
|
@ -1,5 +1,8 @@
|
||||||
|
import pytest
|
||||||
|
|
||||||
from byteb4rb1e.utils.urllib.request import PkgHandler
|
from byteb4rb1e.utils.urllib.request import PkgHandler
|
||||||
|
|
||||||
|
|
||||||
class TestPkgHandler:
|
class TestPkgHandler:
|
||||||
"""
|
"""
|
||||||
"""
|
"""
|
||||||
|
|
@ -7,3 +10,4 @@ class TestPkgHandler:
|
||||||
"""
|
"""
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue