Manage Anvil via pytest plugin

.github/workflows/pr.yml

@@ -69,4 +69,4 @@ jobs:
           uv run ape compile
           uv run python tools/json_filter.py
       - name: Run tests
-        run: ANVIL_LOG_FILE=/tmp/secret-escrow-anvil.log bash tests/run_anvil_test.sh tests/ -v
+        run: uv run ape test --network ethereum:local:foundry tests/

Makefile

@@ -1,10 +1,5 @@
 .PHONY: install setup update format lint compile test
 
-ANVIL_HOST ?= 127.0.0.1
-ANVIL_PORT ?= 8545
-ANVIL_LOG_FILE ?= /tmp/secret-escrow-anvil.log
-ANVIL_STARTUP_TIMEOUT_SECONDS ?= 30
-
 install:
 	uv sync --frozen --no-install-project --all-extras
 	uv run pre-commit install
@@ -27,6 +22,5 @@ lint:
 compile:
 	uv run ape compile
 
-test:
-	@ANVIL_HOST=$(ANVIL_HOST) ANVIL_PORT=$(ANVIL_PORT) ANVIL_LOG_FILE=$(ANVIL_LOG_FILE) ANVIL_STARTUP_TIMEOUT_SECONDS=$(ANVIL_STARTUP_TIMEOUT_SECONDS) \
-		bash tests/run_anvil_test.sh tests/ $(ARG)
+test: compile
+	uv run ape test --network ethereum:local:foundry tests/ ${ARG}

ape-config.yaml

@@ -1,4 +1,4 @@
-name: Secret-Escrow
+name: Linked-ST
 
 plugins:
   - name: solidity
@@ -14,9 +14,8 @@ compile:
 
 solidity:
   version: 0.8.34
-  evm_version: berlin
-  import_remapping:
-    - OpenZeppelin/openzeppelin-contracts@4.9.3=@openzeppelin
+  evm_version: osaka
+  via_ir: true
 
 ethereum:
   default_network: local
@@ -28,9 +27,9 @@ foundry:
   host: http://127.0.0.1:8545
   base_fee: 0
   priority_fee: 0
-  evm_version: berlin
-  request_timeout: 60
-  process_attempts: 10
+  evm_version: osaka
+  request_timeout: 5
+  process_attempts: 3
 
 test:
   mnemonic: test test test test test test test test test test test junk

config.py

@@ -1,19 +1,7 @@
 import os
 
 WEB3_HTTP_PROVIDER = os.environ.get("WEB3_HTTP_PROVIDER") or "http://localhost:8545"
-WEB3_REQUEST_RETRY_COUNT = (
-    int(os.environ.get("WEB3_REQUEST_RETRY_COUNT"))
-    if os.environ.get("WEB3_REQUEST_RETRY_COUNT")
-    else 3
-)
-WEB3_REQUEST_WAIT_TIME = (
-    int(os.environ.get("WEB3_REQUEST_WAIT_TIME"))
-    if os.environ.get("WEB3_REQUEST_WAIT_TIME")
-    else 3
-)
-
-CHAIN_ID = int(os.environ.get("CHAIN_ID")) if os.environ.get("CHAIN_ID") else 2017
-
-TX_GAS_LIMIT = (
-    int(os.environ.get("TX_GAS_LIMIT")) if os.environ.get("TX_GAS_LIMIT") else 6000000
-)
+WEB3_REQUEST_RETRY_COUNT = int(os.environ.get("WEB3_REQUEST_RETRY_COUNT") or 3)
+WEB3_REQUEST_WAIT_TIME = int(os.environ.get("WEB3_REQUEST_WAIT_TIME") or 3)
+CHAIN_ID = int(os.environ.get("CHAIN_ID") or 2017)
+TX_GAS_LIMIT = int(os.environ.get("TX_GAS_LIMIT") or 6000000)

pyproject.toml

@@ -29,12 +29,12 @@ exclude-newer = "7 days"
 
 [tool.pytest.ini_options]
 addopts = ["--network=ethereum:local:foundry"]
-pythonpath = ["."]
+pythonpath = [".", "tests"]
 
 [tool.ruff]
 line-length = 88
 indent-width = 4
-target-version = "py311"
+target-version = "py314"
 exclude = [".venv/*"]
 
 [tool.ruff.format]

tests/anvil_manager.py

@@ -0,0 +1,159 @@
+"""
+Copyright BOOSTRY Co., Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+
+You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+import atexit
+import os
+import socket
+import subprocess
+import time
+from pathlib import Path
+
+import pytest
+from local_anvil_config import local_anvil_command
+
+_ANVIL_PROCESS: subprocess.Popen[str] | None = None
+
+
+def _anvil_host() -> str:
+    return os.environ.get("ANVIL_HOST", "127.0.0.1")
+
+
+def _anvil_port() -> int:
+    return int(os.environ.get("ANVIL_PORT", "8545"))
+
+
+def _anvil_log_file() -> Path:
+    return Path(os.environ.get("ANVIL_LOG_FILE", "/tmp/ibet-wst-anvil.log"))
+
+
+def _anvil_startup_timeout() -> int:
+    return int(os.environ.get("ANVIL_STARTUP_TIMEOUT_SECONDS", "30"))
+
+
+def _listener_pid() -> str | None:
+    port = str(_anvil_port())
+    result = subprocess.run(
+        ["lsof", "-tiTCP:" + port, "-sTCP:LISTEN"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    pid = result.stdout.strip().splitlines()
+    return pid[0] if pid else None
+
+
+def _listener_command(pid: str) -> str:
+    result = subprocess.run(
+        ["ps", "-p", pid, "-o", "command="],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return result.stdout.strip()
+
+
+def _is_port_open() -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.settimeout(1)
+        return sock.connect_ex((_anvil_host(), _anvil_port())) == 0
+
+
+def _wait_for_anvil() -> None:
+    deadline = time.time() + _anvil_startup_timeout()
+    while time.time() < deadline:
+        if _is_port_open():
+            return
+        time.sleep(1)
+
+    log_tail = ""
+    log_file = _anvil_log_file()
+    if log_file.exists():
+        log_tail = "\n" + "".join(log_file.read_text().splitlines(keepends=True)[-50:])
+    raise RuntimeError(
+        f"Anvil did not become ready within {_anvil_startup_timeout()}s.{log_tail}"
+    )
+
+
+def stop_managed_anvil() -> None:
+    """
+    Stop the managed Anvil process if it is running.
+    """
+
+    global _ANVIL_PROCESS
+
+    if _ANVIL_PROCESS is None:
+        return
+
+    _ANVIL_PROCESS.terminate()
+    try:
+        _ANVIL_PROCESS.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        _ANVIL_PROCESS.kill()
+        _ANVIL_PROCESS.wait(timeout=5)
+
+    _ANVIL_PROCESS = None
+
+
+def ensure_anvil_running() -> None:
+    """
+    Ensure that an Anvil process is running and ready to accept connections.
+    """
+
+    global _ANVIL_PROCESS
+
+    # Check if a process is already listening on the Anvil port
+    existing_pid = _listener_pid()
+    if existing_pid is not None:
+        existing_command = _listener_command(existing_pid)
+        if "anvil" not in existing_command:
+            raise RuntimeError(
+                f"Port {_anvil_port()} is already in use by a non-anvil process: {existing_command}"
+            )
+        return
+
+    # Start a new anvil process
+    log_file = _anvil_log_file()
+    log_file.parent.mkdir(parents=True, exist_ok=True)
+    log_handle = log_file.open("w")
+    _ANVIL_PROCESS = subprocess.Popen(
+        local_anvil_command(host=_anvil_host(), port=_anvil_port()),
+        stdout=log_handle,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    _wait_for_anvil()
+    atexit.register(stop_managed_anvil)
+
+
+@pytest.hookimpl(tryfirst=True)
+def pytest_configure():
+    """
+    Pytest hook to ensure Anvil is running before any tests are executed.
+    """
+
+    ensure_anvil_running()
+
+
+@pytest.hookimpl(trylast=True)
+def pytest_unconfigure():
+    """
+    Pytest hook to stop the managed Anvil process after all tests have completed.
+    """
+
+    stop_managed_anvil()

tests/conftest.py

@@ -3,6 +3,8 @@
 import pytest
 from pydantic import BaseModel, ConfigDict
 
+pytest_plugins = ("anvil_manager",)
+
 
 class Users(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)

tests/local_anvil_config.py

@@ -0,0 +1,78 @@
+"""
+Copyright BOOSTRY Co., Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+from typing import Any
+from urllib.parse import urlparse
+
+from ape import networks
+
+from config import CHAIN_ID
+
+
+def _normalized_derivation_path(value: str) -> str:
+    return value.replace("{}", "").rstrip("/") + "/"
+
+
+def _provider_host(host: str | None = None, port: int | None = None) -> str | None:
+    if host is None and port is None:
+        return None
+
+    resolved_host = host or "127.0.0.1"
+    resolved_port = port or 8545
+    return f"http://{resolved_host}:{resolved_port}"
+
+
+def local_foundry_provider(host: str | None = None, port: int | None = None):
+    """Get the local Foundry provider with optional host and port overrides."""
+    provider_settings: dict[str, Any] = {}
+    if configured_host := _provider_host(host, port):
+        provider_settings["host"] = configured_host
+
+    return networks.ethereum.local.get_provider(  # type: ignore
+        "foundry", provider_settings=provider_settings
+    )
+
+
+def local_anvil_command(host: str | None = None, port: int | None = None) -> list[str]:
+    """Generate the command to start a local Anvil instance with optional host and port overrides."""
+    provider = local_foundry_provider(host=host, port=port)
+    parsed_uri = urlparse(provider.uri)
+
+    command = [
+        provider.anvil_bin,
+        "--host",
+        parsed_uri.hostname or "127.0.0.1",
+        "--port",
+        str(parsed_uri.port or 8545),
+        "--chain-id",
+        str(CHAIN_ID),
+        "--hardfork",
+        str(provider.settings.evm_version),
+        "--block-base-fee-per-gas",
+        str(provider.settings.base_fee),
+        "--mnemonic",
+        provider.mnemonic,
+        "--accounts",
+        str(provider.number_of_accounts),
+        "--balance",
+        str(provider.initial_balance),
+        "--derivation-path",
+        _normalized_derivation_path(str(provider.test_config.hd_path)),
+        "--steps-tracing",
+    ]
+    return command