Update tests to align with mutation testing runs

pyproject.toml

@@ -162,6 +162,7 @@ test = [
     "filelock>=3.15.1,<4",
     "requests",
     "requests-cache>=1.2.1,<2",
+    "portalocker",
 ]
 
 lint = [

tests/json_infra/conftest.py

@@ -5,6 +5,8 @@
 from typing import Final, Optional, Set
 
 import git
+import portalocker
+import pytest
 import requests_cache
 from _pytest.config import Config
 from _pytest.config.argparsing import Parser
@@ -214,19 +216,8 @@ def __exit__(
 fixture_lock = StashKey[Optional[FileLock]]()
 
 
-def pytest_sessionstart(session: Session) -> None:  # noqa: U100
-    if get_xdist_worker_id(session) != "master":
-        return
-
-    lock_path = session.config.rootpath.joinpath("tests/fixtures/.lock")
-    stash = session.stash
-    lock_file = FileLock(str(lock_path), timeout=0)
-    lock_file.acquire()
-
-    assert fixture_lock not in stash
-    stash[fixture_lock] = lock_file
-
-    with _FixturesDownloader(session.config.rootpath) as downloader:
+def download_fixtures(root: Path) -> None:
+    with _FixturesDownloader(root) as downloader:
         for _, props in TEST_FIXTURES.items():
             fixture_path = props["fixture_path"]
 
@@ -243,15 +234,70 @@ def pytest_sessionstart(session: Session) -> None:  # noqa: U100
                 )
 
 
+def pytest_sessionstart(session: Session) -> None:  # noqa: U100
+    lock_path = session.config.rootpath.joinpath("tests/fixtures/.lock")
+
+    # use portalocker for mutmut runs
+    if os.environ.get("MUTANT_UNDER_TEST"):
+        shared_lock = portalocker.Lock(lock_path, flags=portalocker.LOCK_SH)
+        shared_lock.acquire()
+        session.stash['mutmut_shared_lock'] = shared_lock
+
+        all_fixtures_ready = all(
+            os.path.exists(props["fixture_path"])
+            for props in TEST_FIXTURES.values()
+        )
+        if not all_fixtures_ready:
+            shared_lock.release()
+            with portalocker.Lock(lock_path, flags=portalocker.LOCK_EX):
+                all_fixtures_ready = all(
+                    os.path.exists(props["fixture_path"])
+                    for props in TEST_FIXTURES.values()
+                )
+                if not all_fixtures_ready:
+                    download_fixtures(session.config.rootpath)
+            shared_lock.acquire()
+            session.stash['mutmut_shared_lock'] = shared_lock
+        return
+
+    if get_xdist_worker_id(session) != "master":
+        return
+
+    stash = session.stash
+    lock_file = FileLock(str(lock_path), timeout=0)
+    lock_file.acquire()
+
+    assert fixture_lock not in stash
+    stash[fixture_lock] = lock_file
+
+    download_fixtures(session.config.rootpath)
+
+
 def pytest_sessionfinish(
     session: Session, exitstatus: int  # noqa: U100
 ) -> None:
     del exitstatus
+
+    if os.environ.get("MUTANT_UNDER_TEST"):
+        shared_lock = session.stash.get('mutmut_shared_lock', None)
+        if shared_lock:
+            shared_lock.release()
+        return
+
     if get_xdist_worker_id(session) != "master":
         return
 
-    lock_file = session.stash[fixture_lock]
-    session.stash[fixture_lock] = None
+    if fixture_lock in session.stash:
+        lock_file = session.stash[fixture_lock]
+        session.stash[fixture_lock] = None
+
+        assert lock_file is not None
+        lock_file.release()
+
 
-    assert lock_file is not None
-    lock_file.release()
+# This is required explicitly becuase when the source does not have any
+# mutable code, mutmut does not run the forced fail condition.
+@pytest.fixture(autouse=True)
+def mutmut_forced_fail() -> None:
+    if os.environ.get("MUTANT_UNDER_TEST") == "fail":
+        pytest.fail("Forced fail for mutmut sanity check")

tests/json_infra/test_blockchain_tests.py

@@ -1,4 +1,4 @@
-from typing import Callable, Dict
+from typing import Any, Callable, Dict
 
 import pytest
 
@@ -10,13 +10,41 @@
     run_blockchain_st_test,
 )
 
+# angry mutant cases are tests that cannot be run for mutation testing
+ANGRY_MUTANT_CASES = (
+    "Callcode1024OOG",
+    "Call1024OOG",
+    "CallRecursiveBombPreCall",
+    "CallRecursiveBomb1",
+    "ABAcalls2",
+    "CallRecursiveBombLog2",
+    "CallRecursiveBomb0",
+    "ABAcalls1",
+    "CallRecursiveBomb2",
+    "CallRecursiveBombLog",
+)
+
+
+def is_angry_mutant(test_case: Any) -> bool:
+    return any(case in str(test_case) for case in ANGRY_MUTANT_CASES)
+
+
+def get_marked_blockchain_test_cases(fork_name: str) -> list:
+    """Get blockchain test cases with angry mutant marking for the given fork."""
+    return [
+        pytest.param(tc, marks=pytest.mark.angry_mutant)
+        if is_angry_mutant(tc)
+        else tc
+        for tc in fetch_blockchain_tests(fork_name)
+    ]
+
 
 def _generate_test_function(fork_name: str) -> Callable:
     @pytest.mark.fork(fork_name)
     @pytest.mark.json_blockchain_tests
     @pytest.mark.parametrize(
         "blockchain_test_case",
-        fetch_blockchain_tests(fork_name),
+        get_marked_blockchain_test_cases(fork_name),
         ids=idfn,
     )
     def test_func(blockchain_test_case: Dict) -> None:

tests/json_infra/test_state_tests.py

@@ -1,18 +1,47 @@
-from typing import Callable, Dict
+from typing import Any, Callable, Dict
 
 import pytest
 
 from . import FORKS
 from .helpers.load_state_tests import fetch_state_tests, idfn, run_state_test
 
+# angry mutant cases are tests that cannot be run for mutation testing
+ANGRY_MUTANT_CASES = (
+    "Callcode1024OOG",
+    "Call1024OOG",
+    "CallRecursiveBombPreCall",
+    "CallRecursiveBombLog2",
+    "CallRecursiveBomb2",
+    "ABAcalls1",
+    "CallRecursiveBomb0_OOG_atMaxCallDepth",
+    "ABAcalls2",
+    "CallRecursiveBomb0",
+    "CallRecursiveBomb1",
+    "CallRecursiveBombLog",
+)
+
+
+def is_angry_mutant(test_case: Any) -> bool:
+    return any(case in str(test_case) for case in ANGRY_MUTANT_CASES)
+
+
+def get_marked_state_test_cases(fork_name: str) -> list:
+    """Get state test cases with angry mutant marking for the given fork."""
+    return [
+        pytest.param(tc, marks=pytest.mark.angry_mutant)
+        if is_angry_mutant(tc)
+        else tc
+        for tc in fetch_state_tests(fork_name)
+    ]
+
 
 def _generate_test_function(fork_name: str) -> Callable:
     @pytest.mark.fork(fork_name)
     @pytest.mark.evm_tools
     @pytest.mark.json_state_tests
     @pytest.mark.parametrize(
         "state_test_case",
-        fetch_state_tests(fork_name),
+        get_marked_state_test_cases(fork_name),
         ids=idfn,
     )
     def test_func(state_test_case: Dict) -> None:

vulture_whitelist.py

@@ -112,3 +112,7 @@
 
 
 _children  # unused attribute (src/ethereum_spec_tools/docc.py:751)
+
+# tests/conftest.py
+# used for mutation testing
+mutmut_forced_fail