create tmpdir in master xdist worker and store lockfiles there

pull/1808/head
alpharush 2 years ago
parent 289337c236
commit 6451285f8f
  1. 4
      .github/scripts/unit_test_runner.sh
  2. 54
      tests/conftest.py

@ -2,11 +2,11 @@
# used to pass --cov=$path and --cov-append to pytest # used to pass --cov=$path and --cov-append to pytest
if [ "$1" != "" ]; then if [ "$1" != "" ]; then
pytest "$1" tests/unit/ pytest "$1" tests/unit/ -n auto
status_code=$? status_code=$?
python -m coverage report python -m coverage report
else else
pytest tests/unit/ pytest tests/unit/ -n auto
status_code=$? status_code=$?
fi fi

@ -1,17 +1,54 @@
# pylint: disable=redefined-outer-name # pylint: disable=redefined-outer-name
import os
from pathlib import Path from pathlib import Path
import tempfile
import shutil
from contextlib import contextmanager from contextlib import contextmanager
from tempfile import NamedTemporaryFile
import pytest import pytest
from filelock import FileLock from filelock import FileLock
from solc_select import solc_select from solc_select import solc_select
from slither import Slither from slither import Slither
@pytest.fixture() def pytest_configure(config):
def solc_binary_path(): """Create a temporary directory for the tests to use."""
if is_master():
config.stash["shared_directory"] = tempfile.mkdtemp()
def pytest_unconfigure(config):
"""Remove the temporary directory after the tests are done."""
if is_master():
shutil.rmtree(config.stash["shared_directory"])
def pytest_configure_node(node):
"""Configure each worker node with the shared directory."""
node.workerinput["shared_directory"] = node.config.stash["shared_directory"]
def is_master():
"""Returns True if the current process is the master process (which does not have a worker id)."""
return os.environ.get("PYTEST_XDIST_WORKER") is None
@pytest.fixture
def shared_directory(request):
"""Returns the shared directory for the current process."""
if is_master():
return request.config.stash["shared_directory"]
return request.config.workerinput["shared_directory"]
@pytest.fixture
def solc_binary_path(shared_directory):
"""
Returns the path to the solc binary for the given version.
If the binary is not installed, it will be installed.
"""
def inner(version): def inner(version):
lock = FileLock(f"{version}.lock", timeout=60) lock = FileLock(f"{shared_directory}/{version}.lock", timeout=60)
with lock: with lock:
if not solc_select.artifact_path(version).exists(): if not solc_select.artifact_path(version).exists():
print("Installing solc version", version) print("Installing solc version", version)
@ -21,18 +58,17 @@ def solc_binary_path():
return inner return inner
@pytest.fixture() @pytest.fixture
def slither_from_source(solc_binary_path): def slither_from_source(solc_binary_path):
@contextmanager @contextmanager
def inner(source_code: str, solc_version: str = "0.8.19"): def inner(source_code: str, solc_version: str = "0.8.19"):
"""Yields a Slither instance using source_code string and solc_version """Yields a Slither instance using source_code string and solc_version.
Creates a temporary file and compiles with solc_version.
Creates a temporary file and changes the solc-version temporary to solc_version.
""" """
fname = "" fname = ""
try: try:
with NamedTemporaryFile(mode="w", suffix=".sol", delete=False) as f: with tempfile.NamedTemporaryFile(mode="w", suffix=".sol", delete=False) as f:
fname = f.name fname = f.name
f.write(source_code) f.write(source_code)
solc_path = solc_binary_path(solc_version) solc_path = solc_binary_path(solc_version)

Loading…
Cancel
Save