Liquidity.Party/tycho-protocol-sdk
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: Misc fixes around byte encoding.

Browse Source 
Also initialize TychoDB for each test case individually.
This commit is contained in:
kayibal
2024-08-07 13:42:13 +02:00
committed by tvinagre
parent 9df366b29d
commit 9f4503e2a9
24 changed files with 32 additions and 2485 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
testing/src/runner/__init__.py
View File

1
testing/src/runner/adapter_handler.py
View File

@@ -1,4 +1,3 @@
import os
import subprocess
from typing import Optional

2
testing/src/runner/cli.py
View File

@@ -1,5 +1,5 @@
import argparse
from .runner import TestRunner
from runner import TestRunner
def main() -> None:

29
testing/src/runner/runner.py
View File

@@ -10,6 +10,7 @@ from pathlib import Path
import yaml
from protosim_py.evm.decoders import ThirdPartyPoolTychoDecoder
from protosim_py.evm.storage import TychoDBSingleton
from protosim_py.models import EVMBlock
from pydantic import BaseModel
from tycho_client.dto import (
@@ -24,12 +25,11 @@ from tycho_client.dto import (
Snapshot,
)
from tycho_client.rpc_client import TychoRPCClient
from tycho_client.stream import TychoStream
from .adapter_handler import AdapterContractHandler
from .evm import get_token_balance, get_block_header
from .tycho import TychoRunner
from .utils import build_snapshot_message, token_factory
from adapter_handler import AdapterContractHandler
from evm import get_token_balance, get_block_header
from tycho import TychoRunner
from utils import build_snapshot_message, token_factory
class TestResult:
@@ -109,14 +109,10 @@ class TestRunner:
def validate_state(self, expected_state: dict, stop_block: int) -> TestResult:
"""Validate the current protocol state against the expected state."""
protocol_components: list[
ProtocolComponent
] = self.tycho_rpc_client.get_protocol_components(
protocol_components = self.tycho_rpc_client.get_protocol_components(
ProtocolComponentsParams(protocol_system="test_protocol")
)
protocol_states: list[
ResponseProtocolState
] = self.tycho_rpc_client.get_protocol_state(
protocol_states = self.tycho_rpc_client.get_protocol_state(
ProtocolStateParams(protocol_system="test_protocol")
)
components_by_id = {
@@ -175,9 +171,9 @@ class TestRunner:
None,
)
if state:
balance_hex = state.balances.get(token, "0x0")
balance_hex = state.balances.get(token, HexBytes("0x00"))
else:
balance_hex = "0x0"
balance_hex = HexBytes("0x00")
tycho_balance = int(balance_hex)
token_balances[comp_id][token] = tycho_balance
@@ -188,9 +184,9 @@ class TestRunner:
f"Balance mismatch for {comp_id}:{token} at block {stop_block}: got {node_balance} "
f"from rpc call and {tycho_balance} from Substreams"
)
contract_states: list[
ResponseAccount
] = self.tycho_rpc_client.get_contract_state(ContractStateParams())
contract_states = self.tycho_rpc_client.get_contract_state(
ContractStateParams()
)
filtered_components = [
pc
for pc in protocol_components
@@ -227,6 +223,7 @@ class TestRunner:
protocol_components: list[ProtocolComponent],
contract_states: list[ResponseAccount],
) -> dict[str, list[SimulationFailure]]:
TychoDBSingleton.initialize()
protocol_type_names = self.config["protocol_type_names"]
block_header = get_block_header(block_number)

1
testing/src/runner/tycho.py
View File

@@ -4,7 +4,6 @@ import threading
import time
import psycopg2
import requests
from psycopg2 import sql
import os

29
testing/src/runner/utils.py
View File

@@ -1,7 +1,7 @@
from logging import getLogger
from typing import Union
from protosim_py.evm.pool_state import ThirdPartyPool
from eth_utils import to_checksum_address
from protosim_py.models import EthereumToken
from tycho_client.dto import (
ResponseProtocolState,
@@ -12,7 +12,6 @@ from tycho_client.dto import (
HexBytes,
TokensParams,
PaginationParams,
ResponseToken,
)
from tycho_client.rpc_client import TychoRPCClient
@@ -43,16 +42,18 @@ def build_snapshot_message(
def token_factory(rpc_client: TychoRPCClient) -> callable(HexBytes):
_client = rpc_client
_token_cache: dict[HexBytes, EthereumToken] = {}
_token_cache: dict[str, EthereumToken] = {}
def factory(addresses: Union[HexBytes, list[HexBytes]]) -> list[EthereumToken]:
if not isinstance(addresses, list):
addresses = [addresses]
def factory(requested_addresses: Union[str, list[str]]) -> list[EthereumToken]:
if not isinstance(requested_addresses, list):
requested_addresses = [to_checksum_address(requested_addresses)]
else:
requested_addresses = [to_checksum_address(a) for a in requested_addresses]
response = dict()
to_fetch = []
for address in addresses:
for address in requested_addresses:
if address in _token_cache:
response[address] = _token_cache[address]
else:
@@ -63,11 +64,17 @@ def token_factory(rpc_client: TychoRPCClient) -> callable(HexBytes):
params = TokensParams(token_addresses=to_fetch, pagination=pagination)
tokens = _client.get_tokens(params)
for token in tokens:
eth_token = EthereumToken(**token.dict())
address = to_checksum_address(token.address)
eth_token = EthereumToken(
symbol=token.symbol,
address=address,
decimals=token.decimals,
gas=token.gas,
)
response[token.address] = eth_token
_token_cache[token.address] = eth_token
response[address] = eth_token
_token_cache[address] = eth_token
return [response[address] for address in addresses]
return [response[address] for address in requested_addresses]
return factory

3
testing/tycho-client/MANIFEST.in
View File

@@ -1,3 +0,0 @@
include wheels/*.whl
include tycho_client/assets/*
include tycho_client/bins/*

41
testing/tycho-client/README.md
View File

@@ -1,41 +0,0 @@
# Tycho Adapter
This repository contains the Tycho Adapter, a tool that allows you to interact with the Tycho API.
## Installation
### Prerequisites
- Python 3.9
- Access to PropellerHead's private PyPi repository (CodeArtifact)
### Install with pip
```shell
# Access to PropellerHead's private PyPi repository (CodeArtifact)
aws codeartifact login --tool pip --repository protosim --domain propeller
# Create conda environment
conda create -n tycho pip python=3.9
# Activate environment
conda activate tycho
# Install packages
pip install -r requirements.txt
```
## Usage
```python
from tycho_client.decoders import ThirdPartyPoolTychoDecoder
from tycho_client.models import Blockchain
from tycho_client.tycho_adapter import TychoPoolStateStreamAdapter
decoder = ThirdPartyPoolTychoDecoder(
"MyProtocolSwapAdapter.evm.runtime", minimum_gas=0, hard_limit=False
)
stream_adapter = TychoPoolStateStreamAdapter(
tycho_url="0.0.0.0:4242",
protocol="my_protocol",
decoder=decoder,
blockchain=Blockchain.ethereum,
)
```

6
testing/tycho-client/requirements.txt
View File

@@ -1,6 +0,0 @@
requests==2.32.2
eth-abi==2.2.0
eth-typing==2.3.0
eth-utils==1.9.5
hexbytes==0.3.1
pydantic==2.8.2

36
testing/tycho-client/setup.py
View File

@@ -1,36 +0,0 @@
from setuptools import setup, find_packages
def read_requirements():
with open("requirements.txt") as req:
content = req.read()
requirements = content.split("\n")
return [req for req in requirements if req and not req.startswith("#")]
setup(
name="tycho-client",
version="0.1.0",
author="Propeller Heads",
description="A package for interacting with the Tycho API.",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
packages=find_packages(),
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
],
python_requires="~=3.9",
install_requires=[
"requests==2.32.2",
"eth-abi==2.2.0",
"eth-typing==2.3.0",
"eth-utils==1.9.5",
"hexbytes==0.3.1",
"pydantic==2.8.2",
"protosim_py==0.4.11",
],
package_data={"tycho-client": ["../wheels/*", "./assets/*", "./bins/*"]},
include_package_data=True,
)

0
testing/tycho-client/tycho_client/__init__.py
View File

211
testing/tycho-client/tycho_client/adapter_contract.py
View File

@@ -1,211 +0,0 @@
import logging
import time
from decimal import Decimal
from fractions import Fraction
from typing import Any, Union, NamedTuple
import eth_abi
from eth_abi.exceptions import DecodingError
from eth_typing import HexStr
from eth_utils import keccak
from eth_utils.abi import collapse_if_tuple
from hexbytes import HexBytes
from protosim_py import (
SimulationEngine,
SimulationParameters,
SimulationResult,
StateUpdate,
)
from .constants import EXTERNAL_ACCOUNT
from .models import Address, EthereumToken, EVMBlock, Capability
from .utils import load_abi, maybe_coerce_error
log = logging.getLogger(__name__)
TStateOverwrites = dict[Address, dict[int, int]]
class Trade(NamedTuple):
"""
Trade represents a simple trading operation with fields:
received_amount: Amount received from the trade
gas_used: Amount of gas used in the transaction
price: Price at which the trade was executed
"""
received_amount: float
gas_used: float
price: float
class ProtoSimResponse:
def __init__(self, return_value: Any, simulation_result: SimulationResult):
self.return_value = return_value
self.simulation_result = simulation_result
class ProtoSimContract:
def __init__(self, address: Address, abi_name: str, engine: SimulationEngine):
self.abi = load_abi(abi_name)
self.address = address
self.engine = engine
self._default_tx_env = dict(
caller=EXTERNAL_ACCOUNT, to=self.address, value=0, overrides={}
)
functions = [f for f in self.abi if f["type"] == "function"]
self._functions = {f["name"]: f for f in functions}
if len(self._functions) != len(functions):
raise ValueError(
f"ProtoSimContract does not support overloaded function names! "
f"Encountered while loading {abi_name}."
)
def _encode_input(self, fname: str, args: list) -> bytearray:
func = self._functions[fname]
types = [collapse_if_tuple(t) for t in func["inputs"]]
selector = keccak(text=f"{fname}({','.join(types)})")[:4]
return bytearray(selector + eth_abi.encode(types, args))
def _decode_output(self, fname: str, encoded: list[int]) -> Any:
func = self._functions[fname]
types = [collapse_if_tuple(t) for t in func["outputs"]]
return eth_abi.decode(types, bytearray(encoded))
def call(
self,
fname: str,
*args: list[Union[int, str, bool, bytes]],
block_number,
timestamp: int = None,
overrides: TStateOverwrites = None,
caller: Address = EXTERNAL_ACCOUNT,
value: int = 0,
) -> ProtoSimResponse:
call_data = self._encode_input(fname, *args)
params = SimulationParameters(
data=call_data,
to=self.address,
block_number=block_number,
timestamp=timestamp or int(time.time()),
overrides=overrides or {},
caller=caller,
value=value,
)
sim_result = self._simulate(params)
try:
output = self._decode_output(fname, sim_result.result)
except DecodingError:
log.warning("Failed to decode output")
output = None
return ProtoSimResponse(output, sim_result)
def _simulate(self, params: SimulationParameters) -> "SimulationResult":
"""Run simulation and handle errors.
It catches a RuntimeError:
- if it's ``Execution reverted``, re-raises a RuntimeError
with a Tenderly link added
- if it's ``Out of gas``, re-raises a RecoverableSimulationException
- otherwise it just re-raises the original error.
"""
try:
simulation_result = self.engine.run_sim(params)
return simulation_result
except RuntimeError as err:
try:
coerced_err = maybe_coerce_error(err, self, params.gas_limit)
except Exception:
log.exception("Couldn't coerce error. Re-raising the original one.")
raise err
msg = str(coerced_err)
if "Revert!" in msg:
raise type(coerced_err)(msg, repr(self)) from err
else:
raise coerced_err
class AdapterContract(ProtoSimContract):
"""
The AdapterContract provides an interface to interact with the protocols implemented
by third parties using the `propeller-protocol-lib`.
"""
def __init__(self, address: Address, engine: SimulationEngine):
super().__init__(address, "ISwapAdapter", engine)
def price(
self,
pair_id: HexStr,
sell_token: EthereumToken,
buy_token: EthereumToken,
amounts: list[int],
block: EVMBlock,
overwrites: TStateOverwrites = None,
) -> list[Fraction]:
args = [HexBytes(pair_id), sell_token.address, buy_token.address, amounts]
res = self.call(
"price",
args,
block_number=block.id,
timestamp=int(block.ts.timestamp()),
overrides=overwrites,
)
return list(map(lambda x: Fraction(*x), res.return_value[0]))
def swap(
self,
pair_id: HexStr,
sell_token: EthereumToken,
buy_token: EthereumToken,
is_buy: bool,
amount: Decimal,
block: EVMBlock,
overwrites: TStateOverwrites = None,
) -> tuple[Trade, dict[str, StateUpdate]]:
args = [
HexBytes(pair_id),
sell_token.address,
buy_token.address,
int(is_buy),
amount,
]
res = self.call(
"swap",
args,
block_number=block.id,
timestamp=int(block.ts.timestamp()),
overrides=overwrites,
)
amount, gas, price = res.return_value[0]
return Trade(amount, gas, Fraction(*price)), res.simulation_result.state_updates
def get_limits(
self,
pair_id: HexStr,
sell_token: EthereumToken,
buy_token: EthereumToken,
block: EVMBlock,
overwrites: TStateOverwrites = None,
) -> tuple[int, int]:
args = [HexBytes(pair_id), sell_token.address, buy_token.address]
res = self.call(
"getLimits",
args,
block_number=block.id,
timestamp=int(block.ts.timestamp()),
overrides=overwrites,
)
return res.return_value[0]
def get_capabilities(
self, pair_id: HexStr, sell_token: EthereumToken, buy_token: EthereumToken
) -> set[Capability]:
args = [HexBytes(pair_id), sell_token.address, buy_token.address]
res = self.call("getCapabilities", args, block_number=1)
return set(map(Capability, res.return_value[0]))
def min_gas_usage(self) -> int:
res = self.call("minGasUsage", [], block_number=1)
return res.return_value[0]

BIN
testing/tycho-client/tycho_client/assets/ERC20.bin
View File

Binary file not shown.

78
testing/tycho-client/tycho_client/assets/IERC20.sol
View File

@@ -1,78 +0,0 @@
// SPDX-License-Identifier: MIT
// OpenZeppelin Contracts (last updated v4.9.0) (token/ERC20/IERC20.sol)
pragma solidity ^0.8.19;
/**
* @dev Interface of the ERC20 standard as defined in the EIP.
*/
interface IERC20 {
/**
* @dev Emitted when `value` tokens are moved from one account (`from`) to
* another (`to`).
*
* Note that `value` may be zero.
*/
event Transfer(address indexed from, address indexed to, uint256 value);
/**
* @dev Emitted when the allowance of a `spender` for an `owner` is set by
* a call to {approve}. `value` is the new allowance.
*/
event Approval(address indexed owner, address indexed spender, uint256 value);
/**
* @dev Returns the amount of tokens in existence.
*/
function totalSupply() external view returns (uint256);
/**
* @dev Returns the amount of tokens owned by `account`.
*/
function balanceOf(address account) external view returns (uint256);
/**
* @dev Moves `amount` tokens from the caller's account to `to`.
*
* Returns a boolean value indicating whether the operation succeeded.
*
* Emits a {Transfer} event.
*/
function transfer(address to, uint256 amount) external returns (bool);
/**
* @dev Returns the remaining number of tokens that `spender` will be
* allowed to spend on behalf of `owner` through {transferFrom}. This is
* zero by default.
*
* This value changes when {approve} or {transferFrom} are called.
*/
function allowance(address owner, address spender) external view returns (uint256);
/**
* @dev Sets `amount` as the allowance of `spender` over the caller's tokens.
*
* Returns a boolean value indicating whether the operation succeeded.
*
* IMPORTANT: Beware that changing an allowance with this method brings the risk
* that someone may use both the old and the new allowance by unfortunate
* transaction ordering. One possible solution to mitigate this race
* condition is to first reduce the spender's allowance to 0 and set the
* desired value afterwards:
* https://github.com/ethereum/EIPs/issues/20#issuecomment-263524729
*
* Emits an {Approval} event.
*/
function approve(address spender, uint256 amount) external returns (bool);
/**
* @dev Moves `amount` tokens from `from` to `to` using the
* allowance mechanism. `amount` is then deducted from the caller's
* allowance.
*
* Returns a boolean value indicating whether the operation succeeded.
*
* Emits a {Transfer} event.
*/
function transferFrom(address from, address to, uint256 amount) external returns (bool);
}

250
testing/tycho-client/tycho_client/assets/ISwapAdapter.abi
View File

@@ -1,250 +0,0 @@
[
{
"inputs": [
{
"internalType": "uint256",
"name": "limit",
"type": "uint256"
}
],
"name": "LimitExceeded",
"type": "error"
},
{
"inputs": [
{
"internalType": "string",
"name": "reason",
"type": "string"
}
],
"name": "NotImplemented",
"type": "error"
},
{
"inputs": [
{
"internalType": "string",
"name": "reason",
"type": "string"
}
],
"name": "Unavailable",
"type": "error"
},
{
"inputs": [
{
"internalType": "bytes32",
"name": "poolId",
"type": "bytes32"
},
{
"internalType": "contract IERC20",
"name": "sellToken",
"type": "address"
},
{
"internalType": "contract IERC20",
"name": "buyToken",
"type": "address"
}
],
"name": "getCapabilities",
"outputs": [
{
"internalType": "enum ISwapAdapterTypes.Capability[]",
"name": "capabilities",
"type": "uint8[]"
}
],
"stateMutability": "nonpayable",
"type": "function"
},
{
"inputs": [
{
"internalType": "bytes32",
"name": "poolId",
"type": "bytes32"
},
{
"internalType": "contract IERC20",
"name": "sellToken",
"type": "address"
},
{
"internalType": "contract IERC20",
"name": "buyToken",
"type": "address"
}
],
"name": "getLimits",
"outputs": [
{
"internalType": "uint256[]",
"name": "limits",
"type": "uint256[]"
}
],
"stateMutability": "nonpayable",
"type": "function"
},
{
"inputs": [
{
"internalType": "uint256",
"name": "offset",
"type": "uint256"
},
{
"internalType": "uint256",
"name": "limit",
"type": "uint256"
}
],
"name": "getPoolIds",
"outputs": [
{
"internalType": "bytes32[]",
"name": "ids",
"type": "bytes32[]"
}
],
"stateMutability": "nonpayable",
"type": "function"
},
{
"inputs": [
{
"internalType": "bytes32",
"name": "poolId",
"type": "bytes32"
}
],
"name": "getTokens",
"outputs": [
{
"internalType": "contract IERC20[]",
"name": "tokens",
"type": "address[]"
}
],
"stateMutability": "nonpayable",
"type": "function"
},
{
"inputs": [
{
"internalType": "bytes32",
"name": "poolId",
"type": "bytes32"
},
{
"internalType": "contract IERC20",
"name": "sellToken",
"type": "address"
},
{
"internalType": "contract IERC20",
"name": "buyToken",
"type": "address"
},
{
"internalType": "uint256[]",
"name": "specifiedAmounts",
"type": "uint256[]"
}
],
"name": "price",
"outputs": [
{
"components": [
{
"internalType": "uint256",
"name": "numerator",
"type": "uint256"
},
{
"internalType": "uint256",
"name": "denominator",
"type": "uint256"
}
],
"internalType": "struct ISwapAdapterTypes.Fraction[]",
"name": "prices",
"type": "tuple[]"
}
],
"stateMutability": "view",
"type": "function"
},
{
"inputs": [
{
"internalType": "bytes32",
"name": "poolId",
"type": "bytes32"
},
{
"internalType": "contract IERC20",
"name": "sellToken",
"type": "address"
},
{
"internalType": "contract IERC20",
"name": "buyToken",
"type": "address"
},
{
"internalType": "enum ISwapAdapterTypes.OrderSide",
"name": "side",
"type": "uint8"
},
{
"internalType": "uint256",
"name": "specifiedAmount",
"type": "uint256"
}
],
"name": "swap",
"outputs": [
{
"components": [
{
"internalType": "uint256",
"name": "calculatedAmount",
"type": "uint256"
},
{
"internalType": "uint256",
"name": "gasUsed",
"type": "uint256"
},
{
"components": [
{
"internalType": "uint256",
"name": "numerator",
"type": "uint256"
},
{
"internalType": "uint256",
"name": "denominator",
"type": "uint256"
}
],
"internalType": "struct ISwapAdapterTypes.Fraction",
"name": "price",
"type": "tuple"
}
],
"internalType": "struct ISwapAdapterTypes.Trade",
"name": "trade",
"type": "tuple"
}
],
"stateMutability": "nonpayable",
"type": "function"
}
]

363
testing/tycho-client/tycho_client/assets/mocked_ERC20.sol
View File

@@ -1,363 +0,0 @@
// SPDX-License-Identifier: MIT
// OpenZeppelin Contracts (last updated v4.9.0) (token/ERC20/ERC20.sol)
pragma solidity ^0.8.19;
import "./IERC20.sol";
/**
* @dev Provides information about the current execution context, including the
* sender of the transaction and its data. While these are generally available
* via msg.sender and msg.data, they should not be accessed in such a direct
* manner, since when dealing with meta-transactions the account sending and
* paying for execution may not be the actual sender (as far as an application
* is concerned).
*
* This contract is only required for intermediate, library-like contracts.
*/
abstract contract Context {
function _msgSender() internal view virtual returns (address) {
return msg.sender;
}
function _msgData() internal view virtual returns (bytes calldata) {
return msg.data;
}
}
/**
* @dev Interface for the optional metadata functions from the ERC20 standard.
*
* _Available since v4.1._
*/
interface IERC20Metadata is IERC20 {
/**
* @dev Returns the name of the token.
*/
function name() external view returns (string memory);
/**
* @dev Returns the symbol of the token.
*/
function symbol() external view returns (string memory);
/**
* @dev Returns the decimals places of the token.
*/
function decimals() external view returns (uint8);
}
/**
* @dev Implementation of the {IERC20} interface.
*
* This implementation is agnostic to the way tokens are created. This means
* that a supply mechanism has to be added in a derived contract using {_mint}.
*
* TIP: For a detailed writeup see our guide
* https://forum.openzeppelin.com/t/how-to-implement-erc20-supply-mechanisms/226[How
* to implement supply mechanisms].
*
* The default value of {decimals} is 18. To change this, you should override
* this function so it returns a different value.
*
* We have followed general OpenZeppelin Contracts guidelines: functions revert
* instead returning `false` on failure. This behavior is nonetheless
* conventional and does not conflict with the expectations of ERC20
* applications.
*
* Additionally, an {Approval} event is emitted on calls to {transferFrom}.
* This allows applications to reconstruct the allowance for all accounts just
* by listening to said events. Other implementations of the EIP may not emit
* these events, as it isn't required by the specification.
*
* Finally, the non-standard {decreaseAllowance} and {increaseAllowance}
* functions have been added to mitigate the well-known issues around setting
* allowances. See {IERC20-approve}.
*/
contract ERC20 is Context, IERC20, IERC20Metadata {
mapping(address => uint256) private _balances;
mapping(address => mapping(address => uint256)) private _allowances;
uint256 private _totalSupply;
string private _name;
string private _symbol;
uint8 private _decimals;
/**
* @dev Sets the values for {name}, {symbol} and {decimals}.
*
* All three of these values are immutable: they can only be set once during
* construction.
*/
constructor(string memory name_, string memory symbol_, uint8 decimals_) {
_name = name_;
_symbol = symbol_;
_decimals = decimals_;
}
/**
* @dev Returns the name of the token.
*/
function name() public view virtual returns (string memory) {
return _name;
}
/**
* @dev Returns the symbol of the token, usually a shorter version of the
* name.
*/
function symbol() public view virtual returns (string memory) {
return _symbol;
}
/**
* @dev Returns the number of decimals used to get its user representation.
* For example, if `decimals` equals `2`, a balance of `505` tokens should
* be displayed to a user as `5.05` (`505 / 10 ** 2`).
*
* Tokens usually opt for a value of 18, imitating the relationship between
* Ether and Wei. This is the default value returned by this function, unless
* it's overridden.
*
* NOTE: This information is only used for _display_ purposes: it in
* no way affects any of the arithmetic of the contract, including
* {IERC20-balanceOf} and {IERC20-transfer}.
*/
function decimals() public view virtual returns (uint8) {
return _decimals;
}
/**
* @dev See {IERC20-totalSupply}.
*/
function totalSupply() public view virtual returns (uint256) {
return _totalSupply;
}
/**
* @dev See {IERC20-balanceOf}.
*/
function balanceOf(address account) public view virtual returns (uint256) {
return _balances[account];
}
/**
* @dev See {IERC20-transfer}.
*
* Requirements:
*
* - `to` cannot be the zero address.
* - the caller must have a balance of at least `amount`.
*/
function transfer(address to, uint256 amount) public virtual returns (bool) {
address owner = _msgSender();
_transfer(owner, to, amount);
return true;
}
/**
* @dev See {IERC20-allowance}.
*/
function allowance(address owner, address spender) public view virtual returns (uint256) {
return _allowances[owner][spender];
}
/**
* @dev See {IERC20-approve}.
*
* NOTE: If `amount` is the maximum `uint256`, the allowance is not updated on
* `transferFrom`. This is semantically equivalent to an infinite approval.
*
* Requirements:
*
* - `spender` cannot be the zero address.
*/
function approve(address spender, uint256 amount) public virtual returns (bool) {
address owner = _msgSender();
_approve(owner, spender, amount);
return true;
}
/**
* @dev See {IERC20-transferFrom}.
*
* Emits an {Approval} event indicating the updated allowance. This is not
* required by the EIP. See the note at the beginning of {ERC20}.
*
* NOTE: Does not update the allowance if the current allowance
* is the maximum `uint256`.
*
* Requirements:
*
* - `from` and `to` cannot be the zero address.
* - `from` must have a balance of at least `amount`.
* - the caller must have allowance for ``from``'s tokens of at least
* `amount`.
*/
function transferFrom(address from, address to, uint256 amount) public virtual returns (bool) {
address spender = _msgSender();
_spendAllowance(from, spender, amount);
_transfer(from, to, amount);
return true;
}
/**
* @dev Atomically increases the allowance granted to `spender` by the caller.
*
* This is an alternative to {approve} that can be used as a mitigation for
* problems described in {IERC20-approve}.
*
* Emits an {Approval} event indicating the updated allowance.
*
* Requirements:
*
* - `spender` cannot be the zero address.
*/
function increaseAllowance(address spender, uint256 addedValue) public virtual returns (bool) {
address owner = _msgSender();
_approve(owner, spender, allowance(owner, spender) + addedValue);
return true;
}
/**
* @dev Atomically decreases the allowance granted to `spender` by the caller.
*
* This is an alternative to {approve} that can be used as a mitigation for
* problems described in {IERC20-approve}.
*
* Emits an {Approval} event indicating the updated allowance.
*
* Requirements:
*
* - `spender` cannot be the zero address.
* - `spender` must have allowance for the caller of at least
* `subtractedValue`.
*/
function decreaseAllowance(address spender, uint256 subtractedValue) public virtual returns (bool) {
address owner = _msgSender();
uint256 currentAllowance = allowance(owner, spender);
require(currentAllowance >= subtractedValue, "ERC20: decreased allowance below zero");
unchecked {
_approve(owner, spender, currentAllowance - subtractedValue);
}
return true;
}
/**
* @dev Moves `amount` of tokens from `from` to `to`.
*
* This internal function is equivalent to {transfer}, and can be used to
* e.g. implement automatic token fees, slashing mechanisms, etc.
*
* Emits a {Transfer} event.
*
* NOTE: This function is not virtual, {_update} should be overridden instead.
*/
function _transfer(address from, address to, uint256 amount) internal {
require(from != address(0), "ERC20: transfer from the zero address");
require(to != address(0), "ERC20: transfer to the zero address");
_update(from, to, amount);
}
/**
* @dev Transfers `amount` of tokens from `from` to `to`, or alternatively mints (or burns) if `from` (or `to`) is
* the zero address. All customizations to transfers, mints, and burns should be done by overriding this function.
*
* Emits a {Transfer} event.
*/
function _update(address from, address to, uint256 amount) internal virtual {
if (from == address(0)) {
_totalSupply += amount;
} else {
uint256 fromBalance = _balances[from];
require(fromBalance >= amount, "ERC20: transfer amount exceeds balance");
unchecked {
// Overflow not possible: amount <= fromBalance <= totalSupply.
_balances[from] = fromBalance - amount;
}
}
if (to == address(0)) {
unchecked {
// Overflow not possible: amount <= totalSupply or amount <= fromBalance <= totalSupply.
_totalSupply -= amount;
}
} else {
unchecked {
// Overflow not possible: balance + amount is at most totalSupply, which we know fits into a uint256.
_balances[to] += amount;
}
}
emit Transfer(from, to, amount);
}
/**
* @dev Creates `amount` tokens and assigns them to `account`, by transferring it from address(0).
* Relies on the `_update` mechanism
*
* Emits a {Transfer} event with `from` set to the zero address.
*
* NOTE: This function is not virtual, {_update} should be overridden instead.
*/
function _mint(address account, uint256 amount) internal {
require(account != address(0), "ERC20: mint to the zero address");
_update(address(0), account, amount);
}
/**
* @dev Destroys `amount` tokens from `account`, by transferring it to address(0).
* Relies on the `_update` mechanism.
*
* Emits a {Transfer} event with `to` set to the zero address.
*
* NOTE: This function is not virtual, {_update} should be overridden instead
*/
function _burn(address account, uint256 amount) internal {
require(account != address(0), "ERC20: burn from the zero address");
_update(account, address(0), amount);
}
/**
* @dev Sets `amount` as the allowance of `spender` over the `owner` s tokens.
*
* This internal function is equivalent to `approve`, and can be used to
* e.g. set automatic allowances for certain subsystems, etc.
*
* Emits an {Approval} event.
*
* Requirements:
*
* - `owner` cannot be the zero address.
* - `spender` cannot be the zero address.
*/
function _approve(address owner, address spender, uint256 amount) internal virtual {
require(owner != address(0), "ERC20: approve from the zero address");
require(spender != address(0), "ERC20: approve to the zero address");
_allowances[owner][spender] = amount;
emit Approval(owner, spender, amount);
}
/**
* @dev Updates `owner` s allowance for `spender` based on spent `amount`.
*
* Does not update the allowance amount in case of infinite allowance.
* Revert if not enough allowance is available.
*
* Might emit an {Approval} event.
*/
function _spendAllowance(address owner, address spender, uint256 amount) internal virtual {
uint256 currentAllowance = allowance(owner, spender);
if (currentAllowance != type(uint256).max) {
require(currentAllowance >= amount, "ERC20: insufficient allowance");
unchecked {
_approve(owner, spender, currentAllowance - amount);
}
}
}
}

12
testing/tycho-client/tycho_client/constants.py
View File

@@ -1,12 +0,0 @@
from pathlib import Path
from typing import Final
ASSETS_FOLDER = Path(__file__).parent / "assets"
TYCHO_CLIENT_FOLDER = Path(__file__).parent / "bins"
TYCHO_CLIENT_LOG_FOLDER = TYCHO_CLIENT_FOLDER / "logs"
EXTERNAL_ACCOUNT: Final[str] = "0xf847a638E44186F3287ee9F8cAF73FF4d4B80784"
"""This is a dummy address used as a transaction sender"""
UINT256_MAX: Final[int] = 2**256 - 1
MAX_BALANCE: Final[int] = UINT256_MAX // 2
"""0.5 of the maximal possible balance to avoid overflow errors"""

175
testing/tycho-client/tycho_client/decoders.py
View File

@@ -1,175 +0,0 @@
import time
from decimal import Decimal
from logging import getLogger
from typing import Any
import eth_abi
from eth_utils import keccak
from protosim_py import SimulationEngine, SimulationParameters, AccountInfo
from .constants import EXTERNAL_ACCOUNT
from .exceptions import TychoDecodeError
from .models import EVMBlock, EthereumToken
from .pool_state import ThirdPartyPool
from .tycho_db import TychoDBSingleton
from .utils import decode_tycho_exchange, get_code_for_address
log = getLogger(__name__)
class ThirdPartyPoolTychoDecoder:
"""ThirdPartyPool decoder for protocol messages from the Tycho feed"""
def __init__(self, adapter_contract: str, minimum_gas: int, trace: bool):
self.adapter_contract = adapter_contract
self.minimum_gas = minimum_gas
self.trace = trace
def decode_snapshot(
self,
snapshot: dict[str, Any],
block: EVMBlock,
tokens: dict[str, EthereumToken],
) -> tuple[dict[str, ThirdPartyPool], list[str]]:
pools = {}
failed_pools = []
for snap in snapshot.values():
try:
pool = self.decode_pool_state(snap, block, tokens)
pools[pool.id_] = pool
except TychoDecodeError as e:
log.error(f"Failed to decode third party snapshot: {e}")
failed_pools.append(snap["component"]["id"])
continue
return pools, failed_pools
def decode_pool_state(
self, snap: dict, block: EVMBlock, tokens: dict[str, EthereumToken]
) -> ThirdPartyPool:
component = snap["component"]
exchange, _ = decode_tycho_exchange(component["protocol_system"])
try:
tokens = tuple(tokens[t] for t in component["tokens"])
except KeyError as e:
raise TychoDecodeError("Unsupported token", pool_id=component["id"])
balances = self.decode_balances(snap, tokens)
optional_attributes = self.decode_optional_attributes(component, snap, block.id)
return ThirdPartyPool(
id_=optional_attributes.pop("pool_id", component["id"]),
tokens=tokens,
balances=balances,
block=block,
spot_prices={},
trading_fee=Decimal("0"),
exchange=exchange,
adapter_contract_name=self.adapter_contract,
minimum_gas=self.minimum_gas,
trace=self.trace,
**optional_attributes,
)
@staticmethod
def decode_optional_attributes(component, snap, block_number):
# Handle optional state attributes
attributes = snap["state"]["attributes"]
balance_owner = attributes.get("balance_owner")
stateless_contracts = {}
static_attributes = snap["component"]["static_attributes"]
pool_id = static_attributes.get("pool_id") or component["id"]
index = 0
while f"stateless_contract_addr_{index}" in static_attributes:
encoded_address = static_attributes[f"stateless_contract_addr_{index}"]
decoded = bytes.fromhex(
encoded_address[2:]
if encoded_address.startswith("0x")
else encoded_address
).decode("utf-8")
if decoded.startswith("call"):
address = ThirdPartyPoolTychoDecoder.get_address_from_call(
block_number, decoded
)
else:
address = decoded
code = static_attributes.get(
f"stateless_contract_code_{index}"
) or get_code_for_address(address)
stateless_contracts[address] = code
index += 1
index = 0
while f"stateless_contract_addr_{index}" in attributes:
address = attributes[f"stateless_contract_addr_{index}"]
code = attributes.get(
f"stateless_contract_code_{index}"
) or get_code_for_address(address)
stateless_contracts[address] = code
index += 1
return {
"balance_owner": balance_owner,
"pool_id": pool_id,
"stateless_contracts": stateless_contracts,
}
@staticmethod
def get_address_from_call(block_number, decoded):
db = TychoDBSingleton.get_instance()
engine = SimulationEngine.new_with_tycho_db(db=db)
engine.init_account(
address="0x0000000000000000000000000000000000000000",
account=AccountInfo(balance=0, nonce=0),
mocked=False,
permanent_storage=None,
)
selector = keccak(text=decoded.split(":")[-1])[:4]
sim_result = engine.run_sim(
SimulationParameters(
data=bytearray(selector),
to=decoded.split(":")[1],
block_number=block_number,
timestamp=int(time.time()),
overrides={},
caller=EXTERNAL_ACCOUNT,
value=0,
)
)
address = eth_abi.decode(["address"], bytearray(sim_result.result))
return address[0]
@staticmethod
def decode_balances(snap, tokens):
balances = {}
for addr, balance in snap["state"]["balances"].items():
checksum_addr = addr
token = next(t for t in tokens if t.address == checksum_addr)
balances[token.address] = token.from_onchain_amount(
int(balance, 16) # balances are big endian encoded
)
return balances
@staticmethod
def apply_update(
pool: ThirdPartyPool,
pool_update: dict[str, Any],
balance_updates: dict[str, Any],
block: EVMBlock,
) -> ThirdPartyPool:
# check for and apply optional state attributes
attributes = pool_update.get("updated_attributes")
if attributes:
# TODO: handle balance_owner and stateless_contracts updates
pass
for addr, balance_msg in balance_updates.items():
token = [t for t in pool.tokens if t.address == addr][0]
balance = int(balance_msg["balance"], 16) # balances are big endian encoded
pool.balances[token.address] = token.from_onchain_amount(balance)
pool.block = block
# we clear simulation cache and overwrites on the pool and trigger a recalculation of spot prices
pool.clear_all_cache()
return pool

59
testing/tycho-client/tycho_client/exceptions.py
View File

@@ -1,59 +0,0 @@
from decimal import Decimal
class TychoDecodeError(Exception):
def __init__(self, msg: str, pool_id: str):
super().__init__(msg)
self.pool_id = pool_id
class APIRequestError(Exception):
pass
class TradeSimulationException(Exception):
def __init__(self, message, pool_id: str):
self.pool_id = pool_id
super().__init__(message)
class RecoverableSimulationException(TradeSimulationException):
"""Marks that the simulation could not fully fulfill the requested order.
Provides a partial trade that is valid but does not fully fulfill the conditions
requested.
Parameters
----------
message
Error message
pool_id
ID of a pool that caused the error
partial_trade
A tuple of (bought_amount, gas_used, new_pool_state, sold_amount)
"""
def __init__(
self,
message,
pool_id: str,
partial_trade: tuple[Decimal, int, "ThirdPartyPool", Decimal] = None,
):
super().__init__(message, pool_id)
self.partial_trade = partial_trade
class OutOfGas(RecoverableSimulationException):
"""This exception indicates that the underlying VM **likely** ran out of gas.
It is not easy to judge whether it was really due to out of gas, as the details
of the SC being called might be hiding this. E.g. out of gas may happen while
calling an external contract, which might show as the external call failing, although
it was due to a lack of gas.
"""
pass
class TychoClientException(Exception):
pass

127
testing/tycho-client/tycho_client/models.py
View File

@@ -1,127 +0,0 @@
import datetime
from decimal import Decimal, localcontext, Context, ROUND_FLOOR, InvalidOperation
from enum import Enum, IntEnum, auto
from fractions import Fraction
from logging import getLogger
from typing import Union
from pydantic import BaseModel, Field, PrivateAttr
Address = str
log = getLogger(__name__)
class Blockchain(Enum):
ethereum = "ethereum"
arbitrum = "arbitrum"
polygon = "polygon"
zksync = "zksync"
class EVMBlock(BaseModel):
id: int
ts: datetime.datetime = Field(default_factory=datetime.datetime.utcnow)
hash_: str
class EthereumToken(BaseModel):
symbol: str
address: str
decimals: int
gas: Union[int, list[int]] = 29000
_hash: int = PrivateAttr(default=None)
def to_onchain_amount(self, amount: Union[float, Decimal, str]) -> int:
"""Converts floating-point numerals to an integer, by shifting right by the
token's maximum amount of decimals (e.g.: 1.000000 becomes 1000000).
For the reverse operation please see self.from_onchain_amount
"""
if not isinstance(amount, Decimal):
log.warning(f"Expected variable of type Decimal. Got {type(amount)}.")
with localcontext(Context(rounding=ROUND_FLOOR, prec=256)):
amount = Decimal(str(amount)) * (10**self.decimals)
try:
amount = amount.quantize(Decimal("1.0"))
except InvalidOperation:
log.error(
f"Quantize failed for {self.symbol}, {amount}, {self.decimals}"
)
return int(amount)
def from_onchain_amount(
self, onchain_amount: Union[int, Fraction], quantize: bool = True
) -> Decimal:
"""Converts an Integer to a quantized decimal, by shifting left by the token's
maximum amount of decimals (e.g.: 1000000 becomes 1.000000 for a 6-decimal token
For the reverse operation please see self.to_onchain_amount
If the onchain_amount is too low, then using quantize can underflow without
raising and the offchain amount returned is 0.
See _decimal.Decimal.quantize docstrings for details.
Quantize is needed for UniswapV2.
"""
with localcontext(Context(rounding=ROUND_FLOOR, prec=256)):
if isinstance(onchain_amount, Fraction):
return (
Decimal(onchain_amount.numerator)
/ Decimal(onchain_amount.denominator)
/ Decimal(10**self.decimals)
).quantize(Decimal(f"{1 / 10 ** self.decimals}"))
if quantize is True:
try:
amount = (
Decimal(str(onchain_amount)) / 10**self.decimals
).quantize(Decimal(f"{1 / 10 ** self.decimals}"))
except InvalidOperation:
amount = Decimal(str(onchain_amount)) / Decimal(10**self.decimals)
else:
amount = Decimal(str(onchain_amount)) / Decimal(10**self.decimals)
return amount
def __repr__(self):
return self.symbol
def __str__(self):
return self.symbol
def __eq__(self, other) -> bool:
# this is faster than calling custom __hash__, due to cache check
return other.address == self.address
def __hash__(self) -> int:
if self._hash is None:
# caching the hash saves time during graph search
self._hash = hash(self.address)
return self._hash
class DatabaseType(Enum):
# Make call to the node each time it needs a storage (unless cached from a previous call).
rpc_reader = "rpc_reader"
# Connect to Tycho and cache the whole state of a target contract, the state is continuously updated by Tycho.
# To use this we need Tycho to be configured to index the target contract state.
tycho = "tycho"
class Capability(IntEnum):
SellSide = auto()
BuySide = auto()
PriceFunction = auto()
FeeOnTransfer = auto()
ConstantPrice = auto()
TokenBalanceIndependent = auto()
ScaledPrice = auto()
HardLimits = auto()
MarginalPrice = auto()
class SynchronizerState(Enum):
started = "started"
ready = "ready"
stale = "stale"
delayed = "delayed"
advanced = "advanced"
ended = "ended"

347
testing/tycho-client/tycho_client/pool_state.py
View File

@@ -1,347 +0,0 @@
import functools
import itertools
from collections import defaultdict
from copy import deepcopy
from decimal import Decimal
from fractions import Fraction
from logging import getLogger
from typing import Optional, cast, TypeVar, Annotated
from eth_typing import HexStr
from protosim_py import SimulationEngine, AccountInfo
from pydantic import BaseModel, PrivateAttr, Field
from .adapter_contract import AdapterContract
from .constants import MAX_BALANCE, EXTERNAL_ACCOUNT
from .exceptions import RecoverableSimulationException
from .models import EVMBlock, Capability, Address, EthereumToken
from .utils import (
create_engine,
get_contract_bytecode,
frac_to_decimal,
ERC20OverwriteFactory,
)
ADAPTER_ADDRESS = "0xA2C5C98A892fD6656a7F39A2f63228C0Bc846270"
log = getLogger(__name__)
TPoolState = TypeVar("TPoolState", bound="ThirdPartyPool")
class ThirdPartyPool(BaseModel):
id_: str
tokens: tuple[EthereumToken, ...]
balances: dict[Address, Decimal]
block: EVMBlock
spot_prices: dict[tuple[EthereumToken, EthereumToken], Decimal]
trading_fee: Decimal
exchange: str
minimum_gas: int
_engine: SimulationEngine = PrivateAttr(default=None)
adapter_contract_name: str
"""The adapters contract name. Used to look up the byte code for the adapter."""
_adapter_contract: AdapterContract = PrivateAttr(default=None)
stateless_contracts: dict[str, bytes] = {}
"""The address to bytecode map of all stateless contracts used by the protocol for simulations."""
capabilities: set[Capability] = Field(default_factory=lambda: {Capability.SellSide})
"""The supported capabilities of this pool."""
balance_owner: Optional[str] = None
"""The contract address for where protocol balances are stored (i.e. a vault contract).
If given, balances will be overwritten here instead of on the pool contract during simulations."""
block_lasting_overwrites: defaultdict[
Address,
Annotated[dict[int, int], Field(default_factory=lambda: defaultdict[dict])],
] = Field(default_factory=lambda: defaultdict(dict))
"""Storage overwrites that will be applied to all simulations. They will be cleared
when ``clear_all_cache`` is called, i.e. usually at each block. Hence the name."""
trace: bool = False
def __init__(self, **data):
super().__init__(**data)
self._set_engine(data.get("engine", None))
self.balance_owner = data.get("balance_owner", None)
self._adapter_contract = AdapterContract(ADAPTER_ADDRESS, self._engine)
self._set_capabilities()
if len(self.spot_prices) == 0:
self._set_spot_prices()
def _set_engine(self, engine: Optional[SimulationEngine]):
"""Set instance's simulation engine. If no engine given, make a default one.
If engine is already set, this is a noop.
The engine will have the specified adapter contract mocked, as well as the
tokens used by the pool.
Parameters
----------
engine
Optional simulation engine instance.
"""
if self._engine is not None:
return
else:
engine = create_engine([t.address for t in self.tokens], trace=self.trace)
engine.init_account(
address="0x0000000000000000000000000000000000000000",
account=AccountInfo(balance=0, nonce=0),
mocked=False,
permanent_storage=None,
)
engine.init_account(
address="0x0000000000000000000000000000000000000004",
account=AccountInfo(balance=0, nonce=0),
mocked=False,
permanent_storage=None,
)
engine.init_account(
address=ADAPTER_ADDRESS,
account=AccountInfo(
balance=MAX_BALANCE,
nonce=0,
code=get_contract_bytecode(self.adapter_contract_name),
),
mocked=False,
permanent_storage=None,
)
for addr, bytecode in self.stateless_contracts.items():
engine.init_account(
address=addr,
account=AccountInfo(balance=0, nonce=0, code=bytecode),
mocked=False,
permanent_storage=None,
)
self._engine = engine
def _set_spot_prices(self):
"""Set the spot prices for this pool.
We currently require the price function capability for now.
"""
self._ensure_capability(Capability.PriceFunction)
for t0, t1 in itertools.permutations(self.tokens, 2):
sell_amount = t0.to_onchain_amount(
self.get_sell_amount_limit(t0, t1) * Decimal("0.01")
)
frac = self._adapter_contract.price(
cast(HexStr, self.id_),
t0,
t1,
[sell_amount],
block=self.block,
overwrites=self.block_lasting_overwrites,
)[0]
if Capability.ScaledPrice in self.capabilities:
self.spot_prices[(t0, t1)] = frac_to_decimal(frac)
else:
scaled = frac * Fraction(10**t0.decimals, 10**t1.decimals)
self.spot_prices[(t0, t1)] = frac_to_decimal(scaled)
def _ensure_capability(self, capability: Capability):
"""Ensures the protocol/adapter implement a certain capability."""
if capability not in self.capabilities:
raise NotImplemented(f"{capability} not available!")
def _set_capabilities(self):
"""Sets capabilities of the pool."""
capabilities = []
for t0, t1 in itertools.permutations(self.tokens, 2):
capabilities.append(
self._adapter_contract.get_capabilities(cast(HexStr, self.id_), t0, t1)
)
max_capabilities = max(map(len, capabilities))
self.capabilities = functools.reduce(set.intersection, capabilities)
if len(self.capabilities) < max_capabilities:
log.warning(
f"Pool {self.id_} hash different capabilities depending on the token pair!"
)
def get_amount_out(
self: TPoolState,
sell_token: EthereumToken,
sell_amount: Decimal,
buy_token: EthereumToken,
) -> tuple[Decimal, int, TPoolState]:
# if the pool has a hard limit and the sell amount exceeds that, simulate and
# raise a partial trade
if Capability.HardLimits in self.capabilities:
sell_limit = self.get_sell_amount_limit(sell_token, buy_token)
if sell_amount > sell_limit:
partial_trade = self._get_amount_out(sell_token, sell_limit, buy_token)
raise RecoverableSimulationException(
"Sell amount exceeds sell limit",
repr(self),
partial_trade + (sell_limit,),
)
return self._get_amount_out(sell_token, sell_amount, buy_token)
def _get_amount_out(
self: TPoolState,
sell_token: EthereumToken,
sell_amount: Decimal,
buy_token: EthereumToken,
) -> tuple[Decimal, int, TPoolState]:
trade, state_changes = self._adapter_contract.swap(
cast(HexStr, self.id_),
sell_token,
buy_token,
False,
sell_token.to_onchain_amount(sell_amount),
block=self.block,
overwrites=self._get_overwrites(sell_token, buy_token),
)
new_state = self._duplicate()
for address, state_update in state_changes.items():
for slot, value in state_update.storage.items():
new_state.block_lasting_overwrites[address][slot] = value
new_price = frac_to_decimal(trade.price)
if new_price != Decimal(0):
new_state.spot_prices = {
(sell_token, buy_token): new_price,
(buy_token, sell_token): Decimal(1) / new_price,
}
buy_amount = buy_token.from_onchain_amount(trade.received_amount)
return buy_amount, trade.gas_used, new_state
def _get_overwrites(
self, sell_token: EthereumToken, buy_token: EthereumToken, **kwargs
) -> dict[Address, dict[int, int]]:
"""Get an overwrites dictionary to use in a simulation.
The returned overwrites include block-lasting overwrites set on the instance
level, and token-specific overwrites that depend on passed tokens.
"""
token_overwrites = self._get_token_overwrites(sell_token, buy_token, **kwargs)
return _merge(self.block_lasting_overwrites, token_overwrites)
def _get_token_overwrites(
self, sell_token: EthereumToken, buy_token: EthereumToken, max_amount=None
) -> dict[Address, dict[int, int]]:
"""Creates overwrites for a token.
Funds external account with enough tokens to execute swaps. Also creates a
corresponding approval to the adapter contract.
If the protocol reads its own token balances, the balances for the underlying
pool contract will also be overwritten.
"""
res = []
if Capability.TokenBalanceIndependent not in self.capabilities:
res = [self._get_balance_overwrites()]
# avoids recursion if using this method with get_sell_amount_limit
if max_amount is None:
max_amount = sell_token.to_onchain_amount(
self.get_sell_amount_limit(sell_token, buy_token)
)
overwrites = ERC20OverwriteFactory(sell_token)
overwrites.set_balance(max_amount, EXTERNAL_ACCOUNT)
overwrites.set_allowance(
allowance=max_amount, owner=EXTERNAL_ACCOUNT, spender=ADAPTER_ADDRESS
)
res.append(overwrites.get_protosim_overwrites())
# we need to merge the dictionaries because balance overwrites may target
# the same token address.
res = functools.reduce(_merge, res)
return res
def _get_balance_overwrites(self) -> dict[Address, dict[int, int]]:
balance_overwrites = {}
address = self.balance_owner or self.id_
for t in self.tokens:
overwrites = ERC20OverwriteFactory(t)
overwrites.set_balance(
t.to_onchain_amount(self.balances[t.address]), address
)
balance_overwrites.update(overwrites.get_protosim_overwrites())
return balance_overwrites
def _duplicate(self: type["ThirdPartyPool"]) -> "ThirdPartyPool":
"""Make a new instance identical to self that shares the same simulation engine.
Note that the new and current state become coupled in a way that they must
simulate the same block. This is fine, see
https://datarevenue.atlassian.net/browse/ROC-1301
Not naming this method _copy to not confuse with Pydantic's .copy method.
"""
return type(self)(
exchange=self.exchange,
adapter_contract_name=self.adapter_contract_name,
block=self.block,
id_=self.id_,
tokens=self.tokens,
spot_prices=self.spot_prices.copy(),
trading_fee=self.trading_fee,
block_lasting_overwrites=deepcopy(self.block_lasting_overwrites),
engine=self._engine,
balances=self.balances,
minimum_gas=self.minimum_gas,
balance_owner=self.balance_owner,
stateless_contracts=self.stateless_contracts,
)
def get_sell_amount_limit(
self, sell_token: EthereumToken, buy_token: EthereumToken
) -> Decimal:
"""
Retrieves the sell amount of the given token.
For pools with more than 2 tokens, the sell limit is obtain for all possible buy token
combinations and the minimum is returned.
"""
limit = self._adapter_contract.get_limits(
cast(HexStr, self.id_),
sell_token,
buy_token,
block=self.block,
overwrites=self._get_overwrites(
sell_token, buy_token, max_amount=MAX_BALANCE // 100
),
)[0]
return sell_token.from_onchain_amount(limit)
def clear_all_cache(self):
self._engine.clear_temp_storage()
self.block_lasting_overwrites = defaultdict(dict)
self._set_spot_prices()
def _merge(a: dict, b: dict, path=None):
"""
Merges two dictionaries (a and b) deeply. This means it will traverse and combine
their nested dictionaries too if present.
Parameters:
a (dict): The first dictionary to merge.
b (dict): The second dictionary to merge into the first one.
path (list, optional): An internal parameter used during recursion
to keep track of the ancestry of nested dictionaries.
Returns:
a (dict): The merged dictionary which includes all key-value pairs from `b`
added into `a`. If they have nested dictionaries with same keys, those are also merged.
On key conflicts, preference is given to values from b.
"""
if path is None:
path = []
for key in b:
if key in a:
if isinstance(a[key], dict) and isinstance(b[key], dict):
_merge(a[key], b[key], path + [str(key)])
else:
a[key] = b[key]
return a

345
testing/tycho-client/tycho_client/tycho_adapter.py
View File

@@ -1,345 +0,0 @@
import asyncio
import json
import platform
import time
from asyncio.subprocess import STDOUT, PIPE
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from http.client import HTTPException
from logging import getLogger
from typing import Any, Optional, Dict
import requests
from protosim_py import AccountUpdate, AccountInfo, BlockHeader
from .constants import TYCHO_CLIENT_LOG_FOLDER, TYCHO_CLIENT_FOLDER
from .decoders import ThirdPartyPoolTychoDecoder
from .exceptions import APIRequestError, TychoClientException
from .models import Blockchain, EVMBlock, EthereumToken, SynchronizerState, Address
from .pool_state import ThirdPartyPool
from .tycho_db import TychoDBSingleton
from .utils import create_engine
log = getLogger(__name__)
class TokenLoader:
def __init__(
self,
tycho_url: str,
blockchain: Blockchain,
min_token_quality: Optional[int] = 0,
):
self.tycho_url = tycho_url
self.blockchain = blockchain
self.min_token_quality = min_token_quality
self.endpoint = "/v1/{}/tokens"
self._token_limit = 10000
def get_tokens(self) -> dict[str, EthereumToken]:
"""Loads all tokens from Tycho RPC"""
url = self.tycho_url + self.endpoint.format(self.blockchain.value)
page = 0
start = time.monotonic()
all_tokens = []
while data := self._get_all_with_pagination(
url=url,
page=page,
limit=self._token_limit,
params={"min_quality": self.min_token_quality},
):
all_tokens.extend(data)
page += 1
if len(data) < self._token_limit:
break
log.info(f"Loaded {len(all_tokens)} tokens in {time.monotonic() - start:.2f}s")
formatted_tokens = dict()
for token in all_tokens:
formatted = EthereumToken(**token)
formatted_tokens[formatted.address] = formatted
return formatted_tokens
def get_token_subset(self, addresses: list[str]) -> dict[str, EthereumToken]:
"""Loads a subset of tokens from Tycho RPC"""
url = self.tycho_url + self.endpoint.format(self.blockchain.value)
page = 0
start = time.monotonic()
all_tokens = []
while data := self._get_all_with_pagination(
url=url,
page=page,
limit=self._token_limit,
params={"min_quality": self.min_token_quality, "addresses": addresses},
):
all_tokens.extend(data)
page += 1
if len(data) < self._token_limit:
break
log.info(f"Loaded {len(all_tokens)} tokens in {time.monotonic() - start:.2f}s")
formatted_tokens = dict()
for token in all_tokens:
formatted = EthereumToken(**token)
formatted_tokens[formatted.address] = formatted
return formatted_tokens
@staticmethod
def _get_all_with_pagination(
url: str, params: Optional[Dict] = None, page: int = 0, limit: int = 50
) -> Dict:
if params is None:
params = {}
params["pagination"] = {"page": page, "page_size": limit}
r = requests.post(url, json=params)
try:
r.raise_for_status()
except HTTPException as e:
log.error(f"Request status {r.status_code} with content {r.json()}")
raise APIRequestError("Failed to load token configurations")
return r.json()["tokens"]
@dataclass(repr=False)
class BlockProtocolChanges:
block: EVMBlock
pool_states: dict[Address, ThirdPartyPool]
"""All updated pools"""
removed_pools: set[Address]
deserialization_time: float
"""The time it took to deserialize the pool states from the tycho feed message"""
class TychoPoolStateStreamAdapter:
def __init__(
self,
tycho_url: str,
protocol: str,
decoder: ThirdPartyPoolTychoDecoder,
blockchain: Blockchain,
min_tvl: Optional[Decimal] = 10,
min_token_quality: Optional[int] = 0,
include_state=True,
):
"""
:param tycho_url: URL to connect to Tycho DB
:param protocol: Name of the protocol that you're testing
:param blockchain: Blockchain enum
:param min_tvl: Minimum TVL to consider a pool
:param min_token_quality: Minimum token quality to consider a token
:param include_state: Include state in the stream
"""
self.min_token_quality = min_token_quality
self.tycho_url = tycho_url
self.min_tvl = min_tvl
self.tycho_client = None
self.protocol = f"vm:{protocol}"
self._include_state = include_state
self._blockchain = blockchain
self._decoder = decoder
# Create engine
# TODO: This should be initialized outside the adapter?
TychoDBSingleton.initialize(tycho_http_url=self.tycho_url)
self._engine = create_engine([], trace=False)
# Loads tokens from Tycho
self._tokens: dict[str, EthereumToken] = TokenLoader(
tycho_url=f"http://{self.tycho_url}",
blockchain=self._blockchain,
min_token_quality=self.min_token_quality,
).get_tokens()
async def start(self):
"""Start the tycho-client Rust binary through subprocess"""
# stdout=PIPE means that the output is piped directly to this Python process
# stderr=STDOUT combines the stderr and stdout streams
bin_path = self._get_binary_path()
cmd = [
"--log-folder",
str(TYCHO_CLIENT_LOG_FOLDER),
"--tycho-url",
self.tycho_url,
"--min-tvl",
str(self.min_tvl),
]
if not self._include_state:
cmd.append("--no-state")
cmd.append("--exchange")
cmd.append(self.protocol)
log.debug(f"Starting tycho-client binary at {bin_path}. CMD: {cmd}")
self.tycho_client = await asyncio.create_subprocess_exec(
str(bin_path), *cmd, stdout=PIPE, stderr=STDOUT, limit=2**64
)
@staticmethod
def _get_binary_path():
"""Determines the correct binary path based on the OS and architecture."""
os_name = platform.system()
if os_name == "Linux":
architecture = platform.machine()
if architecture == "aarch64":
return TYCHO_CLIENT_FOLDER / "tycho-client-linux-arm64"
else:
return TYCHO_CLIENT_FOLDER / "tycho-client-linux-x64"
elif os_name == "Darwin":
architecture = platform.machine()
if architecture == "arm64":
return TYCHO_CLIENT_FOLDER / "tycho-client-mac-arm64"
else:
return TYCHO_CLIENT_FOLDER / "tycho-client-mac-x64"
else:
raise ValueError(f"Unsupported OS: {os_name}")
def __aiter__(self):
return self
async def __anext__(self) -> BlockProtocolChanges:
if self.tycho_client.stdout.at_eof():
raise StopAsyncIteration
line = await self.tycho_client.stdout.readline()
try:
if not line:
exit_code = await self.tycho_client.wait()
if exit_code == 0:
# Clean exit, handle accordingly, possibly without raising an error
log.debug("Tycho client exited cleanly.")
raise StopAsyncIteration
else:
line = f"Tycho client failed with exit code: {exit_code}"
# Non-zero exit code, handle accordingly, possibly by raising an error
raise TychoClientException(line)
msg = json.loads(line.decode("utf-8"))
except (json.JSONDecodeError, TychoClientException):
# Read the last 10 lines from the log file available under TYCHO_CLIENT_LOG_FOLDER
# and raise an exception with the last 10 lines
error_msg = f"Invalid JSON output on tycho. Original line: {line}."
with open(TYCHO_CLIENT_LOG_FOLDER / "dev_logs.log", "r") as f:
lines = f.readlines()
last_lines = lines[-10:]
error_msg += f" Tycho logs: {last_lines}"
log.exception(error_msg)
raise Exception("Tycho-client failed.")
return self.process_tycho_message(msg)
@staticmethod
def build_snapshot_message(
protocol_components: dict, protocol_states: dict, contract_states: dict
) -> dict[str, ThirdPartyPool]:
vm_states = {state["address"]: state for state in contract_states["accounts"]}
states = {}
for component in protocol_components["protocol_components"]:
pool_id = component["id"]
states[pool_id] = {"component": component}
for state in protocol_states["states"]:
pool_id = state["component_id"]
if pool_id not in states:
log.debug(f"{pool_id} was present in snapshot but not in components")
continue
states[pool_id]["state"] = state
snapshot = {"vm_storage": vm_states, "states": states}
return snapshot
def process_tycho_message(self, msg) -> BlockProtocolChanges:
self._validate_sync_states(msg)
state_msg = msg["state_msgs"][self.protocol]
block = EVMBlock(
id=msg["block"]["id"],
ts=datetime.fromtimestamp(msg["block"]["timestamp"]),
hash_=msg["block"]["hash"],
)
return self.process_snapshot(block, state_msg["snapshot"])
def process_snapshot(
self, block: EVMBlock, state_msg: dict
) -> BlockProtocolChanges:
start = time.monotonic()
removed_pools = set()
decoded_count = 0
failed_count = 0
self._process_vm_storage(state_msg["vm_storage"], block)
# decode new components
decoded_pools, failed_pools = self._decoder.decode_snapshot(
state_msg["states"], block, self._tokens
)
decoded_count += len(decoded_pools)
failed_count += len(failed_pools)
decoded_pools = {
p.id_: p for p in decoded_pools.values()
} # remap pools to their pool ids
deserialization_time = time.monotonic() - start
total = decoded_count + failed_count
log.debug(
f"Received {total} snapshots. n_decoded: {decoded_count}, n_failed: {failed_count}"
)
if failed_count > 0:
log.info(f"Could not to decode {failed_count}/{total} pool snapshots")
return BlockProtocolChanges(
block=block,
pool_states=decoded_pools,
removed_pools=removed_pools,
deserialization_time=round(deserialization_time, 3),
)
def _validate_sync_states(self, msg):
try:
sync_state = msg["sync_states"][self.protocol]
log.info(f"Received sync state for {self.protocol}: {sync_state}")
if not sync_state["status"] != SynchronizerState.ready.value:
raise ValueError("Tycho-indexer is not synced")
except KeyError:
raise ValueError("Invalid message received from tycho-client.")
def _process_vm_storage(self, storage: dict[str, Any], block: EVMBlock):
vm_updates = []
for storage_update in storage.values():
address = storage_update["address"]
balance = int(storage_update["native_balance"], 16)
code = bytearray.fromhex(storage_update["code"][2:])
# init accounts
self._engine.init_account(
address=address,
account=AccountInfo(balance=balance, nonce=0, code=code),
mocked=False,
permanent_storage=None,
)
# apply account updates
slots = {int(k, 16): int(v, 16) for k, v in storage_update["slots"].items()}
vm_updates.append(
AccountUpdate(
address=address,
chain=storage_update["chain"],
slots=slots,
balance=balance,
code=code,
change="Update",
)
)
block_header = BlockHeader(block.id, block.hash_, int(block.ts.timestamp()))
TychoDBSingleton.get_instance().update(vm_updates, block_header)

48
testing/tycho-client/tycho_client/tycho_db.py
View File

@@ -1,48 +0,0 @@
from protosim_py import TychoDB
class TychoDBSingleton:
"""
A singleton wrapper around the TychoDB class.
This class ensures that there is only one instance of TychoDB throughout the lifetime of the program,
avoiding the overhead of creating multiple instances.
"""
_instance = None
@classmethod
def initialize(cls, tycho_http_url: str):
"""
Initialize the TychoDB instance with the given URLs.
Parameters
----------
tycho_http_url : str
The URL of the Tycho HTTP server.
"""
cls._instance = TychoDB(tycho_http_url=tycho_http_url)
@classmethod
def get_instance(cls) -> TychoDB:
"""
Retrieve the singleton instance of TychoDB.
If the TychoDB instance does not exist, it creates a new one.
If it already exists, it returns the existing instance.
Returns
-------
TychoDB
The singleton instance of TychoDB.
"""
if cls._instance is None:
raise ValueError(
"TychoDB instance not initialized. Call initialize() first."
)
return cls._instance
@classmethod
def clear_instance(cls):
cls._instance = None

354
testing/tycho-client/tycho_client/utils.py
View File

@@ -1,354 +0,0 @@
import json
import os
from decimal import Decimal
from fractions import Fraction
from functools import lru_cache
from logging import getLogger
from pathlib import Path
from typing import Final, Any
import eth_abi
from eth_typing import HexStr
from hexbytes import HexBytes
from protosim_py import SimulationEngine, AccountInfo
import requests
from web3 import Web3
from .constants import EXTERNAL_ACCOUNT, MAX_BALANCE, ASSETS_FOLDER
from .exceptions import OutOfGas
from .models import Address, EthereumToken
from .tycho_db import TychoDBSingleton
log = getLogger(__name__)
def decode_tycho_exchange(exchange: str) -> (str, bool):
# removes vm prefix if present, returns True if vm prefix was present (vm protocol) or False if native protocol
return (exchange.split(":")[1], False) if "vm:" in exchange else (exchange, True)
def create_engine(
mocked_tokens: list[Address], trace: bool = False
) -> SimulationEngine:
"""Create a simulation engine with a mocked ERC20 contract at given addresses.
Parameters
----------
mocked_tokens
A list of addresses at which a mocked ERC20 contract should be inserted.
trace
Whether to trace calls, only meant for debugging purposes, might print a lot of
data to stdout.
"""
db = TychoDBSingleton.get_instance()
engine = SimulationEngine.new_with_tycho_db(db=db, trace=trace)
for t in mocked_tokens:
info = AccountInfo(
balance=0, nonce=0, code=get_contract_bytecode(ASSETS_FOLDER / "ERC20.bin")
)
engine.init_account(
address=t, account=info, mocked=True, permanent_storage=None
)
engine.init_account(
address=EXTERNAL_ACCOUNT,
account=AccountInfo(balance=MAX_BALANCE, nonce=0, code=None),
mocked=False,
permanent_storage=None,
)
return engine
class ERC20OverwriteFactory:
def __init__(self, token: EthereumToken):
"""
Initialize the ERC20OverwriteFactory.
Parameters:
token: The token object.
"""
self._token = token
self._overwrites = dict()
self._balance_slot: Final[int] = 0
self._allowance_slot: Final[int] = 1
self._total_supply_slot: Final[int] = 2
def set_balance(self, balance: int, owner: Address):
"""
Set the balance for a given owner.
Parameters:
balance: The balance value.
owner: The owner's address.
"""
storage_index = get_storage_slot_at_key(HexStr(owner), self._balance_slot)
self._overwrites[storage_index] = balance
log.log(
5,
f"Override balance: token={self._token.address} owner={owner}"
f"value={balance} slot={storage_index}",
)
def set_allowance(self, allowance: int, spender: Address, owner: Address):
"""
Set the allowance for a given spender and owner.
Parameters:
allowance: The allowance value.
spender: The spender's address.
owner: The owner's address.
"""
storage_index = get_storage_slot_at_key(
HexStr(spender),
get_storage_slot_at_key(HexStr(owner), self._allowance_slot),
)
self._overwrites[storage_index] = allowance
log.log(
5,
f"Override allowance: token={self._token.address} owner={owner}"
f"spender={spender} value={allowance} slot={storage_index}",
)
def set_total_supply(self, supply: int):
"""
Set the total supply of the token.
Parameters:
supply: The total supply value.
"""
self._overwrites[self._total_supply_slot] = supply
log.log(
5, f"Override total supply: token={self._token.address} supply={supply}"
)
def get_protosim_overwrites(self) -> dict[Address, dict[int, int]]:
"""
Get the overwrites dictionary of previously collected values.
Returns:
dict[Address, dict]: A dictionary containing the token's address
and the overwrites.
"""
# Protosim returns lowercase addresses in state updates returned from simulation
return {self._token.address.lower(): self._overwrites}
def get_geth_overwrites(self) -> dict[Address, dict[int, int]]:
"""
Get the overwrites dictionary of previously collected values.
Returns:
dict[Address, dict]: A dictionary containing the token's address
and the overwrites.
"""
formatted_overwrites = {
HexBytes(key).hex(): "0x" + HexBytes(val).hex().lstrip("0x").zfill(64)
for key, val in self._overwrites.items()
}
code = "0x" + get_contract_bytecode(ASSETS_FOLDER / "ERC20.bin").hex()
return {self._token.address: {"stateDiff": formatted_overwrites, "code": code}}
def get_storage_slot_at_key(key: Address, mapping_slot: int) -> int:
"""Get storage slot index of a value stored at a certain key in a mapping
Parameters
----------
key
Key in a mapping. This function is meant to work with ethereum addresses
and accepts only strings.
mapping_slot
Storage slot at which the mapping itself is stored. See the examples for more
explanation.
Returns
-------
slot
An index of a storage slot where the value at the given key is stored.
Examples
--------
If a mapping is declared as a first variable in solidity code, its storage slot
is 0 (e.g. ``balances`` in our mocked ERC20 contract). Here's how to compute
a storage slot where balance of a given account is stored::
get_storage_slot_at_key("0xC63135E4bF73F637AF616DFd64cf701866BB2628", 0)
For nested mappings, we need to apply the function twice. An example of this is
``allowances`` in ERC20. It is a mapping of form:
``dict[owner, dict[spender, value]]``. In our mocked ERC20 contract, ``allowances``
is a second variable, so it is stored at slot 1. Here's how to get a storage slot
where an allowance of ``0xspender`` to spend ``0xowner``'s money is stored::
get_storage_slot_at_key("0xspender", get_storage_slot_at_key("0xowner", 1)))
See Also
--------
`Solidity Storage Layout documentation
<https://docs.soliditylang.org/en/v0.8.13/internals/layout_in_storage.html#mappings-and-dynamic-arrays>`_
"""
key_bytes = bytes.fromhex(key[2:]).rjust(32, b"\0")
mapping_slot_bytes = int.to_bytes(mapping_slot, 32, "big")
slot_bytes = Web3.keccak(key_bytes + mapping_slot_bytes)
return int.from_bytes(slot_bytes, "big")
@lru_cache
def get_contract_bytecode(path: str) -> bytes:
"""Load contract bytecode from a file given an absolute path"""
with open(path, "rb") as fh:
code = fh.read()
return code
def frac_to_decimal(frac: Fraction) -> Decimal:
return Decimal(frac.numerator) / Decimal(frac.denominator)
def load_abi(name_or_path: str) -> dict:
if os.path.exists(abspath := os.path.abspath(name_or_path)):
path = abspath
else:
path = f"{os.path.dirname(os.path.abspath(__file__))}/assets/{name_or_path}.abi"
try:
with open(os.path.abspath(path)) as f:
abi: dict = json.load(f)
except FileNotFoundError:
search_dir = f"{os.path.dirname(os.path.abspath(__file__))}/assets/"
# List all files in search dir and subdirs suggest them to the user in an error message
available_files = []
for dirpath, dirnames, filenames in os.walk(search_dir):
for filename in filenames:
# Make paths relative to search_dir
relative_path = os.path.relpath(
os.path.join(dirpath, filename), search_dir
)
available_files.append(relative_path.replace(".abi", ""))
raise FileNotFoundError(
f"File {name_or_path} not found. "
f"Did you mean one of these? {', '.join(available_files)}"
)
return abi
# https://docs.soliditylang.org/en/latest/control-structures.html#panic-via-assert-and-error-via-require
solidity_panic_codes = {
0: "GenericCompilerPanic",
1: "AssertionError",
17: "ArithmeticOver/Underflow",
18: "ZeroDivisionError",
33: "UnkownEnumMember",
34: "BadStorageByteArrayEncoding",
51: "EmptyArray",
0x32: "OutOfBounds",
0x41: "OutOfMemory",
0x51: "BadFunctionPointer",
}
def parse_solidity_error_message(data) -> str:
data_bytes = HexBytes(data)
error_string = f"Failed to decode: {data}"
# data is encoded as Error(string)
if data_bytes[:4] == HexBytes("0x08c379a0"):
(error_string,) = eth_abi.decode(["string"], data_bytes[4:])
return error_string
elif data_bytes[:4] == HexBytes("0x4e487b71"):
(error_code,) = eth_abi.decode(["uint256"], data_bytes[4:])
return solidity_panic_codes.get(error_code, f"Panic({error_code})")
# old solidity: revert 'some string' case
try:
(error_string,) = eth_abi.decode(["string"], data_bytes)
return error_string
except Exception:
pass
# some custom error maybe it is with string?
try:
(error_string,) = eth_abi.decode(["string"], data_bytes[4:])
return error_string
except Exception:
pass
try:
(error_string,) = eth_abi.decode(["string"], data_bytes[4:])
return error_string
except Exception:
pass
return error_string
def maybe_coerce_error(
err: RuntimeError, pool_state: Any, gas_limit: int = None
) -> Exception:
details = err.args[0]
# we got bytes as data, so this was a revert
if details.data.startswith("0x"):
err = RuntimeError(
f"Revert! Reason: {parse_solidity_error_message(details.data)}"
)
# we have gas information, check if this likely an out of gas err.
if gas_limit is not None and details.gas_used is not None:
# if we used up 97% or more issue a OutOfGas error.
usage = details.gas_used / gas_limit
if usage >= 0.97:
return OutOfGas(
f"SimulationError: Likely out-of-gas. "
f"Used: {usage * 100:.2f}% of gas limit. "
f"Original error: {err}",
repr(pool_state),
)
elif "OutOfGas" in details.data:
if gas_limit is not None:
usage = details.gas_used / gas_limit
usage_msg = f"Used: {usage * 100:.2f}% of gas limit. "
else:
usage_msg = ""
return OutOfGas(
f"SimulationError: out-of-gas. {usage_msg}Original error: {details.data}",
repr(pool_state),
)
return err
def exec_rpc_method(url, method, params, timeout=240) -> dict:
payload = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1}
headers = {"Content-Type": "application/json"}
r = requests.post(url, data=json.dumps(payload), headers=headers, timeout=timeout)
if r.status_code >= 400:
raise RuntimeError(
"RPC failed: status_code not ok. (method {}: {})".format(
method, r.status_code
)
)
data = r.json()
if "result" in data:
return data["result"]
elif "error" in data:
raise RuntimeError(
"RPC failed with Error {} - {}".format(data["error"], method)
)
def get_code_for_address(address: str, connection_string: str = None):
if connection_string is None:
connection_string = os.getenv("RPC_URL")
if connection_string is None:
raise EnvironmentError("RPC_URL environment variable is not set")
method = "eth_getCode"
params = [address, "latest"]
try:
code = exec_rpc_method(connection_string, method, params)
return bytes.fromhex(code[2:])
except RuntimeError as e:
print(f"Error fetching code for address {address}: {e}")
return None