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

feat(testing): add a script for Tycho integration testing

Browse Source
This commit is contained in:
Florian Pellissier
2024-05-16 09:42:28 +02:00
parent 226ec98cf8
commit 8cc526527e
8 changed files with 490 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -12,6 +12,8 @@ target/
.idea
*.log
__pycache__
substreams/ethereum-template/Cargo.lock
.DS_Store

15
docker-compose.yaml Normal file
View File

@@ -0,0 +1,15 @@
version: '3.1'
services:
db:
image: ghcr.io/dbsystel/postgresql-partman:15-5
restart: "always"
environment:
POSTGRESQL_PASSWORD: mypassword
POSTGRESQL_DATABASE: tycho_indexer_0
POSTGRESQL_USERNAME: postgres
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
volumes:
postgres_data:

3
substreams/ethereum-curve/substreams.yaml
View File

@@ -64,3 +64,6 @@ modules:
mode: deltas # This is the key property that simplifies `BalanceChange` handling
output:
type: proto:tycho.evm.v1.BlockContractChanges
params:
map_components: "address=bebc44782c7db0a1a60cb6fe97d0b483032ff1c7&tx_hash=20793bbf260912aae189d5d261ff003c9b9166da8191d8f9d63ff1c7722f3ac6&tokens[]=6b175474e89094c44da98b954eedeac495271d0f&tokens[]=a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&tokens[]=dac17f958d2ee523a2206206994597c13d831ec7,address=dc24316b9ae028f1497c275eb9192a3ea0f67022&tx_hash=fac67ecbd423a5b915deff06045ec9343568edaec34ae95c43d35f2c018afdaa&tokens[]=EeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE&tokens[]=ae7ab96520DE3A18E5e111B5EaAb095312D7fE84,address=d51a44d3fae010294c616388b506acda1bfaae46&tx_hash=dafb6385ed988ce8aacecfe1d97b38ea5e60b1ebce74d2423f71ddd621680138&tokens[]=dAC17F958D2ee523a2206206994597C13D831ec7&tokens[]=2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599&tokens[]=C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"

108
substreams/ethereum-curve/test_assets.yaml Normal file
View File

@@ -0,0 +1,108 @@
substreams_yaml_path: ./substreams.yaml
protocol_type_names:
- "curve_pool"
- "crypto_swap_ng"
- "tricrypto"
tests:
- name: test_3pool_creation
start_block: 10809470
stop_block: 10809480
expected_state:
protocol_components:
- id: "0xbebc44782c7db0a1a60cb6fe97d0b483032ff1c7"
tokens:
- "0xdac17f958d2ee523a2206206994597c13d831ec7"
- "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"
- "0x6b175474e89094c44da98b954eedeac495271d0f"
static_attributes:
creation_tx: "0x20793bbf260912aae189d5d261ff003c9b9166da8191d8f9d63ff1c7722f3ac6"
- name: test_steth_creation
start_block: 11592550
stop_block: 11592553
expected_state:
protocol_components:
- id: "0xdc24316b9ae028f1497c275eb9192a3ea0f67022"
tokens:
- "0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE"
- "0xae7ab96520DE3A18E5e111B5EaAb095312D7fE84"
static_attributes:
creation_tx: "0xfac67ecbd423a5b915deff06045ec9343568edaec34ae95c43d35f2c018afdaa"
- name: test_crypto_swap_ng_factory_plain_pool_creation
start_block: 19355220
stop_block: 19355225
expected_state:
protocol_components:
- id: "0xeeda34a377dd0ca676b9511ee1324974fa8d980d"
tokens:
- "0xd9a442856c234a39a81a089c06451ebaa4306a72"
- "0x7f39c581f595b53c5cb19bd0b3f8da6c935e2ca0"
static_attributes:
creation_tx: "0x0e2bad5695d4ff8ebbaf668674a24bdcc4843da44933d947f2a454fd731da3c1"
- name: test_crypto_swap_ng_factory_meta_pool_creation
start_block: 19216042
stop_block: 19216045
expected_state:
protocol_components:
- id: "0xef484de8C07B6e2d732A92B5F78e81B38f99f95E"
tokens:
- "0x865377367054516e17014CcdED1e7d814EDC9ce4"
- "0xA5588F7cdf560811710A2D82D3C9c99769DB1Dcb"
static_attributes:
creation_tx: "0x3cfeecae1b43086ee5705f89b803e21eb0492d7d5db06c229586db8fc72f5665"
- name: test_metapool_factory_metapool_creation
start_block: 18028600
stop_block: 18028610
expected_state:
protocol_components:
- id: "0x61fA2c947e523F9ABfb8d7e2903A5D5218C119a7"
tokens:
- "0x6c3ea9036406852006290770BEdFcAbA0e23A0e8"
- "0x3175Df0976dFA876431C2E9eE6Bc45b65d3473CC"
static_attributes:
creation_tx: "0x78454fd29fad021842288092aaeee5eff9ab877e006ac372d084e5e95ca2bd2c"
- name: test_metapool_factory_plainpool_creation
start_block: 18808555
stop_block: 18808577
expected_state:
protocol_components:
- id: "0xf2DCf6336D8250754B4527f57b275b19c8D5CF88"
tokens:
- "0xe9633C52f4c8B7BDeb08c4A7fE8a5c1B84AFCf67"
- "0x77E06c9eCCf2E797fd462A92B6D7642EF85b0A44"
static_attributes:
creation_tx: "0xeb34c90d352f18ffcfe78b7e393e155f0314acf06c54d1ac9996e4ee5a9b4742"
- id: "0x3f67dc2AdBA4B1beB6A48c30AB3AFb1c1440d35B"
tokens:
- "0xe9633C52f4c8B7BDeb08c4A7fE8a5c1B84AFCf67"
- "0x77E06c9eCCf2E797fd462A92B6D7642EF85b0A44"
static_attributes:
creation_tx: "0x455559b43afaf429c15c1d807fd7f5dd47be30f6411a854499f719b944f4c024"
- name: test_cryptopool_factory_creation
start_block: 19162590
stop_block: 19162633
expected_state:
protocol_components:
- id: "0x71db3764d6841d8b01dc27c0fd4a66a8a34b2be0"
tokens:
- "0x04c154b66cb340f3ae24111cc767e0184ed00cc6"
- "0x4591dbff62656e7859afe5e45f6f47d3669fbb28"
static_attributes:
creation_tx: "0xa89c09a7e0dfd84f3a294b8df4f33cc4a623e6d52deee357457afe2591ea596f"
- id: "0x6c9Fe53cC13b125d6476E5Ce2b76983bd5b7A112"
tokens:
- "0x35fA164735182de50811E8e2E824cFb9B6118ac2"
- "0xf951E335afb289353dc249e82926178EaC7DEd78"
static_attributes:
creation_tx: "0xa5b13d50c56242f7994b8e1339032bb4c6f9ac3af3054d4eae3ce9e32e3c1a50"
- name: test_tricrypto_factory_creation
start_block: 17371455
stop_block: 17371457
expected_state:
protocol_components:
- id: "0x7F86Bf177Dd4F3494b841a37e810A34dD56c829B"
tokens:
- "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
- "0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599"
- "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" #TODO: NEEDS TO BE NATIVE
static_attributes:
creation_tx: "0x2bd59c19f993b83729fb23498f897a58567c6f0b3ee2f00613ba515a7b19fe23"

24
testing/cli.py Normal file
View File

@@ -0,0 +1,24 @@
import argparse
from runner import TestRunner
def main() -> None:
parser = argparse.ArgumentParser(
description="Run indexer within a specified range of blocks"
)
parser.add_argument(
"--test_yaml_path", type=str, help="Path to the test configuration YAML file."
)
parser.add_argument(
"--with_binary_logs",
action="store_true",
help="Flag to activate logs from Tycho.",
)
args = parser.parse_args()
test_runner = TestRunner(args.test_yaml_path, args.with_binary_logs)
test_runner.run_tests()
if __name__ == "__main__":
main()

24
testing/evm.py Normal file
View File

@@ -0,0 +1,24 @@
from web3 import Web3
def get_token_balance(rpc_url, token_address, wallet_address, block_number):
web3 = Web3(Web3.HTTPProvider(rpc_url))
if not web3.isConnected():
raise ConnectionError("Failed to connect to the Ethereum node")
erc20_abi = [
{
"constant": True,
"inputs": [{"name": "_owner", "type": "address"}],
"name": "balanceOf",
"outputs": [{"name": "balance", "type": "uint256"}],
"type": "function",
}
]
contract = web3.eth.contract(address=token_address, abi=erc20_abi)
balance = contract.functions.balanceOf(wallet_address).call(
block_identifier=block_number
)
return balance

137
testing/runner.py Normal file
View File

@@ -0,0 +1,137 @@
import os
from pathlib import Path
import shutil
import subprocess
import yaml
from tycho import TychoRunner
class TestResult:
def __init__(self, success: bool, message: str = None):
self.success = success
self.message = message
@classmethod
def Passed(cls):
return cls(success=True)
@classmethod
def Failed(cls, message: str):
return cls(success=False, message=message)
def load_config(yaml_path: str) -> dict:
"""Load YAML configuration from a specified file path."""
with open(yaml_path, "r") as file:
return yaml.safe_load(file)
class TestRunner:
def __init__(self, config_path: str, with_binary_logs: bool):
self.config = load_config(config_path)
self.base_dir = os.path.dirname(config_path)
self.tycho_runner = TychoRunner(with_binary_logs)
def run_tests(self) -> None:
"""Run all tests specified in the configuration."""
print(f"Running tests ...")
for test in self.config["tests"]:
spkg_path = self.build_spkg(
os.path.join(self.base_dir, self.config["substreams_yaml_path"]),
lambda data: self.update_initial_block(data, test["start_block"]),
)
self.tycho_runner.run_tycho(
spkg_path,
test["start_block"],
test["stop_block"],
self.config["protocol_type_names"],
)
result = self.tycho_runner.run_with_rpc_server(
self.validate_state, test["expected_state"]
)
if result.success:
print(f"✅ {test['name']} passed.")
else:
print(f"❗️ {test['name']} failed: {result.message}")
self.tycho_runner.empty_database(
"postgres://postgres:mypassword@localhost:5432"
)
def validate_state(self, expected_state: dict) -> TestResult:
"""Validate the current protocol state against the expected state."""
protocol_components = self.tycho_runner.get_protocol_components()
components = {
component["id"]: component
for component in protocol_components["protocol_components"]
}
try:
for expected_component in expected_state.get("protocol_components", []):
comp_id = expected_component["id"].lower()
if comp_id not in components:
return TestResult.Failed(
f"'{comp_id}' not found in protocol components."
)
component = components[comp_id]
for key, value in expected_component.items():
if key not in component:
return TestResult.Failed(
f"Missing '{key}' in component '{comp_id}'."
)
if isinstance(value, list):
if set(map(str.lower, value)) != set(
map(str.lower, component[key])
):
return TestResult.Failed(
f"List mismatch for key '{key}': {value} != {component[key]}"
)
elif value is not None and value.lower() != component[key]:
return TestResult.Failed(
f"Value mismatch for key '{key}': {value} != {component[key]}"
)
return TestResult.Passed()
except Exception as e:
return TestResult.Failed(str(e))
@staticmethod
def build_spkg(yaml_file_path: str, modify_func: callable) -> str:
"""Build a Substreams package with modifications to the YAML file."""
backup_file_path = f"{yaml_file_path}.backup"
shutil.copy(yaml_file_path, backup_file_path)
with open(yaml_file_path, "r") as file:
data = yaml.safe_load(file)
modify_func(data)
spkg_name = f"{yaml_file_path.rsplit('/', 1)[0]}/{data['package']['name'].replace('_', '-', 1)}-{data['package']['version']}.spkg"
with open(yaml_file_path, "w") as file:
yaml.dump(data, file, default_flow_style=False)
try:
result = subprocess.run(
["substreams", "pack", yaml_file_path], capture_output=True, text=True
)
if result.returncode != 0:
print("Substreams pack command failed:", result.stderr)
except Exception as e:
print(f"Error running substreams pack command: {e}")
shutil.copy(backup_file_path, yaml_file_path)
Path(backup_file_path).unlink()
return spkg_name
@staticmethod
def update_initial_block(data: dict, start_block: int) -> None:
"""Update the initial block for all modules in the configuration data."""
for module in data["modules"]:
module["initialBlock"] = start_block

175
testing/tycho.py Normal file
View File

@@ -0,0 +1,175 @@
import signal
import threading
import time
import requests
import subprocess
import os
import psycopg2
from psycopg2 import sql
binary_path = "./testing/tycho-indexer"
class TychoRunner:
def __init__(self, with_binary_logs: bool = False):
self.with_binary_logs = with_binary_logs
def run_tycho(
self,
spkg_path: str,
start_block: int,
end_block: int,
protocol_type_names: list,
) -> None:
"""Run the Tycho indexer with the specified SPKG and block range."""
env = os.environ.copy()
env["RUST_LOG"] = "info"
try:
process = subprocess.Popen(
[
binary_path,
"run",
"--spkg",
spkg_path,
"--module",
"map_protocol_changes",
"--protocol-type-names",
",".join(protocol_type_names),
"--start-block",
str(start_block),
"--stop-block",
str(end_block + 2),
], # TODO: +2 is a hack to make up for the cache in the index side.
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
env=env,
)
if self.with_binary_logs:
with process.stdout:
for line in iter(process.stdout.readline, ""):
if line:
print(line.strip())
with process.stderr:
for line in iter(process.stderr.readline, ""):
if line:
print(line.strip())
process.wait()
except Exception as e:
print(f"Error running Tycho indexer: {e}")
def run_with_rpc_server(self, func: callable, *args, **kwargs):
"""
Run a function with Tycho RPC running in background.
This function is a wrapper around a target function. It starts Tycho RPC as a background task, executes the target function and stops Tycho RPC.
"""
stop_event = threading.Event()
process = None
def run_rpc_server():
nonlocal process
try:
env = os.environ.copy()
env["RUST_LOG"] = "info"
process = subprocess.Popen(
[binary_path, "rpc"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
env=env,
)
# Read remaining stdout and stderr
if self.with_binary_logs:
for output in process.stdout:
if output:
print(output.strip())
for error_output in process.stderr:
if error_output:
print(error_output.strip())
process.wait()
if process.returncode != 0:
print("Command failed with return code:", process.returncode)
except Exception as e:
print(f"An error occurred while running the command: {e}")
finally:
if process and process.poll() is None:
process.terminate()
process.wait()
# Start the RPC server in a separate thread
rpc_thread = threading.Thread(target=run_rpc_server)
rpc_thread.start()
time.sleep(3) # Wait for the RPC server to start
try:
# Run the provided function
result = func(*args, **kwargs)
return result
finally:
stop_event.set()
if process and process.poll() is None:
process.send_signal(signal.SIGINT)
if rpc_thread.is_alive():
rpc_thread.join()
@staticmethod
def get_protocol_components() -> dict:
"""Retrieve protocol components from the RPC server."""
url = "http://0.0.0.0:4242/v1/ethereum/protocol_components"
headers = {"accept": "application/json", "Content-Type": "application/json"}
data = {"protocol_system": "test_protocol"}
response = requests.post(url, headers=headers, json=data)
return response.json()
@staticmethod
def get_protocol_state() -> dict:
"""Retrieve protocol state from the RPC server."""
url = "http://0.0.0.0:4242/v1/ethereum/protocol_state"
headers = {"accept": "application/json", "Content-Type": "application/json"}
data = {
"protocolSystem": "string",
"version": {"block": {"chain": "ethereum", "number": 0}},
}
response = requests.post(url, headers=headers, json=data)
return response.json()
@staticmethod
def empty_database(db_url: str) -> None:
"""Drop and recreate the Tycho indexer database."""
try:
conn = psycopg2.connect(db_url)
conn.autocommit = True
cursor = conn.cursor()
cursor.execute(
sql.SQL("DROP DATABASE IF EXISTS {}").format(
sql.Identifier("tycho_indexer_0")
)
)
cursor.execute(
sql.SQL("CREATE DATABASE {}").format(sql.Identifier("tycho_indexer_0"))
)
except psycopg2.Error as e:
print(f"Database error: {e}")
finally:
if cursor:
cursor.close()
if conn:
conn.close()