Make sanity tests self-contained

.github/workflows/onpush.yml

@@ -29,6 +29,4 @@ jobs:
       - name: Lint
         run: hatch run lint:check
       - name: Test
-        env:
-          RPC_URL: ${{ secrets.RPC_URL }}
         run: hatch run +py=${{ matrix.python-ver }} test:all

tests/mock_provider.py

@@ -0,0 +1,181 @@
+from collections.abc import Mapping
+from typing import cast
+
+from eth_abi import encode
+from eth_typing import ChecksumAddress
+from eth_utils import function_signature_to_4byte_selector
+from web3 import Web3
+from web3.providers.base import BaseProvider
+
+from autonity.constants import AUTONITY_CONTRACT_ADDRESS
+from autonity.contracts import (
+    accountability,
+    acu,
+    auctioneer,
+    autonity,
+    inflation_controller,
+    liquid_logic,
+    omission_accountability,
+    oracle,
+    stabilization,
+    supply_control,
+    upgrade_manager,
+)
+
+
+def make_address(index: int) -> ChecksumAddress:
+    return Web3.to_checksum_address(f"0x{index:040x}")
+
+
+CONTRACT_ADDRESSES: dict[str, ChecksumAddress] = {
+    "autonity": AUTONITY_CONTRACT_ADDRESS,
+    "accountability": make_address(1),
+    "oracle": make_address(2),
+    "acu": make_address(3),
+    "supply_control": make_address(4),
+    "stabilization": make_address(5),
+    "upgrade_manager": make_address(6),
+    "inflation_controller": make_address(7),
+    "omission_accountability": make_address(8),
+    "auctioneer": make_address(9),
+    "liquid_logic": make_address(10),
+}
+
+
+def _canonical_abi_type(item: Mapping[str, object]) -> str:
+    abi_type = cast(str, item["type"])
+    if not abi_type.startswith("tuple"):
+        return abi_type
+
+    suffix = abi_type[5:]
+    components = cast(list[dict[str, object]], item.get("components", []))
+    inner_types = ",".join(_canonical_abi_type(component) for component in components)
+    return f"({inner_types}){suffix}"
+
+
+def _default_abi_value(item: Mapping[str, object]) -> object:
+    abi_type = cast(str, item["type"])
+
+    if abi_type.endswith("]"):
+        base_type, _, suffix = abi_type.rpartition("[")
+        length_text = suffix[:-1]
+
+        base_item = dict(item)
+        base_item["type"] = base_type
+        base_value = _default_abi_value(base_item)
+
+        if length_text == "":
+            return []
+        return [base_value for _ in range(int(length_text))]
+
+    if abi_type.startswith("tuple"):
+        components = cast(list[dict[str, object]], item.get("components", []))
+        return tuple(_default_abi_value(component) for component in components)
+
+    if abi_type.startswith(("uint", "int")):
+        return 0
+    if abi_type == "bool":
+        return False
+    if abi_type == "address":
+        return make_address(0)
+    if abi_type == "string":
+        return ""
+    if abi_type == "bytes":
+        return b""
+    if abi_type.startswith("bytes"):
+        return b"\x00" * int(abi_type[5:])
+
+    raise ValueError(f"Unsupported ABI type: {abi_type}")
+
+
+def _build_selector_map(abi: list[dict[str, object]]) -> dict[str, dict[str, object]]:
+    selector_map: dict[str, dict[str, object]] = {}
+
+    for item in abi:
+        if item.get("type") != "function":
+            continue
+
+        inputs = cast(list[dict[str, object]], item.get("inputs", []))
+        input_types = ",".join(_canonical_abi_type(param) for param in inputs)
+        signature = f"{item['name']}({input_types})"
+        selector = "0x" + function_signature_to_4byte_selector(signature).hex()
+        selector_map[selector] = item
+
+    return selector_map
+
+
+class MockProvider(BaseProvider):
+    def __init__(self) -> None:
+        super().__init__()
+        self._selector_map_by_address = {
+            CONTRACT_ADDRESSES["autonity"].lower(): _build_selector_map(autonity.ABI),
+            CONTRACT_ADDRESSES["accountability"].lower(): _build_selector_map(
+                accountability.ABI
+            ),
+            CONTRACT_ADDRESSES["oracle"].lower(): _build_selector_map(oracle.ABI),
+            CONTRACT_ADDRESSES["acu"].lower(): _build_selector_map(acu.ABI),
+            CONTRACT_ADDRESSES["supply_control"].lower(): _build_selector_map(
+                supply_control.ABI
+            ),
+            CONTRACT_ADDRESSES["stabilization"].lower(): _build_selector_map(
+                stabilization.ABI
+            ),
+            CONTRACT_ADDRESSES["upgrade_manager"].lower(): _build_selector_map(
+                upgrade_manager.ABI
+            ),
+            CONTRACT_ADDRESSES["inflation_controller"].lower(): _build_selector_map(
+                inflation_controller.ABI
+            ),
+            CONTRACT_ADDRESSES["omission_accountability"].lower(): _build_selector_map(
+                omission_accountability.ABI
+            ),
+            CONTRACT_ADDRESSES["auctioneer"].lower(): _build_selector_map(
+                auctioneer.ABI
+            ),
+            CONTRACT_ADDRESSES["liquid_logic"].lower(): _build_selector_map(
+                liquid_logic.ABI
+            ),
+        }
+
+    def make_request(self, method: str, params: object) -> dict[str, object]:
+        if method == "eth_call":
+            request_params = cast(list[dict[str, str]], params)
+            tx = request_params[0]
+            to_address = tx["to"].lower()
+            call_data = tx.get("data", "0x")
+            selector = call_data[:10]
+
+            selector_map = self._selector_map_by_address[to_address]
+            function_abi = selector_map[selector]
+            outputs = cast(list[dict[str, object]], function_abi.get("outputs", []))
+
+            output_types = [_canonical_abi_type(output) for output in outputs]
+            output_values = [_default_abi_value(output) for output in outputs]
+            encoded = "0x" + encode(output_types, output_values).hex()
+
+            return {"jsonrpc": "2.0", "id": 1, "result": encoded}
+
+        if method == "eth_chainId":
+            return {"jsonrpc": "2.0", "id": 1, "result": "0x1"}
+        if method == "web3_clientVersion":
+            return {
+                "jsonrpc": "2.0",
+                "id": 1,
+                "result": "Autonity/v6.0.0/linux",
+            }
+        if method == "eth_getTransactionCount":
+            return {"jsonrpc": "2.0", "id": 1, "result": "0x0"}
+        if method == "eth_estimateGas":
+            return {"jsonrpc": "2.0", "id": 1, "result": "0x5208"}
+        if method == "eth_gasPrice":
+            return {"jsonrpc": "2.0", "id": 1, "result": "0x1"}
+        if method == "eth_maxPriorityFeePerGas":
+            return {"jsonrpc": "2.0", "id": 1, "result": "0x1"}
+        if method == "eth_getBlockByNumber":
+            return {
+                "jsonrpc": "2.0",
+                "id": 1,
+                "result": {"number": "0x1", "baseFeePerGas": "0x1"},
+            }
+
+        raise RuntimeError(f"Unexpected RPC method: {method}")

tests/test_sanity.py

@@ -1,31 +1,80 @@
 # type: ignore
 
-import os
+from dataclasses import is_dataclass
 from enum import IntEnum
 from inspect import isclass, signature
 from typing import Callable, List
 
 from eth_typing import ChecksumAddress
 from hexbytes import HexBytes
-from web3 import Web3, HTTPProvider
-from web3.exceptions import ContractLogicError, ContractPanicError
+from web3 import Web3
 from web3.contract.contract import ContractFunction, ContractEvent
 
 import autonity
+from autonity import factory
+from autonity.constants import AUTONITY_CONTRACT_VERSION
 from autonity.contracts.accountability import BaseSlashingRates, Factors
+from autonity.contracts.autonity import Config, Contracts, Policy, Protocol
 from autonity.factory import LiquidLogic
+from tests.mock_provider import CONTRACT_ADDRESSES, MockProvider, make_address
 
 BINDINGS = [attr for attr in autonity.__dict__.values() if isinstance(attr, Callable)]
 
+
+def _fake_config(_: Web3) -> Config:
+    return Config(
+        policy=Policy(
+            0,
+            0,
+            0,
+            0,
+            0,
+            0,
+            0,
+            0,
+            make_address(11),
+            make_address(12),
+            0,
+            0,
+        ),
+        contracts=Contracts(
+            CONTRACT_ADDRESSES["accountability"],
+            CONTRACT_ADDRESSES["oracle"],
+            CONTRACT_ADDRESSES["acu"],
+            CONTRACT_ADDRESSES["supply_control"],
+            CONTRACT_ADDRESSES["stabilization"],
+            CONTRACT_ADDRESSES["upgrade_manager"],
+            CONTRACT_ADDRESSES["inflation_controller"],
+            CONTRACT_ADDRESSES["omission_accountability"],
+            CONTRACT_ADDRESSES["auctioneer"],
+        ),
+        protocol=Protocol(
+            make_address(13),
+            0,
+            0,
+            0,
+            0,
+            0,
+            0,
+            0,
+        ),
+        contract_version=AUTONITY_CONTRACT_VERSION,
+    )
+
+
+factory._config.cache_clear()
+factory._config = _fake_config
+
+
 TEST_INPUTS = {
     bool: True,
     int: 1,
     str: "",
-    ChecksumAddress: "0x0123456789abcDEF0123456789abCDef01234567",
+    ChecksumAddress: make_address(0),
     HexBytes: HexBytes(val=""),
     List[int]: [1],
     List[str]: [""],
-    List[ChecksumAddress]: ["0x0123456789abcDEF0123456789abCDef01234567"],
+    List[ChecksumAddress]: [make_address(0)],
     BaseSlashingRates: BaseSlashingRates(0, 0, 0),
     Factors: Factors(0, 0, 0),
 }
@@ -35,15 +84,13 @@ def pytest_generate_tests(metafunc):
     if "test_input" not in metafunc.fixturenames:
         return
 
-    w3 = Web3(HTTPProvider(os.environ["RPC_URL"]))
+    w3 = Web3(MockProvider())
     test_inputs = []
     ids = []
 
     for binding in BINDINGS:
         if binding is LiquidLogic:
-            aut = autonity.Autonity(w3)
-            validator = aut.get_validator(aut.get_validators()[0])
-            contract = binding(w3, validator.liquid_state_contract)
+            contract = binding(w3, CONTRACT_ADDRESSES["liquid_logic"])
         else:
             contract = binding(w3)
 
@@ -84,17 +131,28 @@ def _get_arg_value(type_):
                 for param in signature(type_).parameters.values()
             ]
             return type_(*inputs)
+        if is_dataclass(type_):
+            inputs = [
+                _get_arg_value(param.annotation)
+                for param in signature(type_).parameters.values()
+            ]
+            return type_(*inputs)
     return TEST_INPUTS[type_]
 
 
 def test_bindings_with_arbitrary_inputs(test_input):
     binding, args = test_input
-    try:
-        return_value = binding(*args)
-        assert return_value is not None
-        if isinstance(return_value, ContractFunction):
-            assert return_value.build_transaction()
-    except (ContractLogicError, ContractPanicError):
-        # The contract execution doesn't have to be successful,
-        # only creating the Web3.py contract function instance does
-        pass
+    return_value = binding(*args)
+    assert return_value is not None
+
+    if isinstance(return_value, ContractFunction):
+        built_transaction = return_value.build_transaction(
+            {
+                "from": make_address(14),
+                "nonce": 0,
+                "gas": 210000,
+                "gasPrice": 1,
+                "chainId": 1,
+            }
+        )
+        assert built_transaction["data"].startswith("0x")