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

Add Balancer test, move requests to RPC Client class

Browse Source
This commit is contained in:
Thales Lima
2024-07-22 03:47:58 +02:00
committed by tvinagre
parent 3ccf0a8a1f
commit 19bf222e8e
3 changed files with 61 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
testing/runner.py
View File

@@ -14,7 +14,7 @@ from tycho_client.models import Blockchain, EVMBlock
from tycho_client.tycho_adapter import TychoPoolStateStreamAdapter
from evm import get_token_balance, get_block_header
from tycho import TychoRunner
from tycho import TychoRunner, TychoRPCClient
class TestResult:
@@ -49,6 +49,7 @@ class TestRunner:
self.config = load_config(config_path)
self.base_dir = os.path.dirname(config_path)
self.tycho_runner = TychoRunner(with_binary_logs)
self.tycho_rpc_client = TychoRPCClient()
self.db_url = db_url
self._chain = Blockchain.ethereum
@@ -82,8 +83,8 @@ class TestRunner:
def validate_state(self, expected_state: dict, stop_block: int) -> TestResult:
"""Validate the current protocol state against the expected state."""
protocol_components = self.tycho_runner.get_protocol_components()
protocol_states = self.tycho_runner.get_protocol_state()
protocol_components = self.tycho_rpc_client.get_protocol_components()
protocol_states = self.tycho_rpc_client.get_protocol_state()
components = {
component["id"]: component
for component in protocol_components["protocol_components"]
@@ -142,7 +143,7 @@ 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 = self.tycho_runner.get_contract_state()
contract_states = self.tycho_rpc_client.get_contract_state()
simulation_failures = self.simulate_get_amount_out(
token_balances,
stop_block,

62
testing/tycho.py
View File

@@ -26,6 +26,38 @@ def get_binary_path():
binary_path = get_binary_path()
class TychoRPCClient:
def __init__(self, rpc_url: str = "http://0.0.0.0:4242"):
self.rpc_url = rpc_url
def get_protocol_components(self) -> dict:
"""Retrieve protocol components from the RPC server."""
url = self.rpc_url + "/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()
def get_protocol_state(self) -> dict:
"""Retrieve protocol state from the RPC server."""
url = self.rpc_url + "/v1/ethereum/protocol_state"
headers = {"accept": "application/json", "Content-Type": "application/json"}
data = {}
response = requests.post(url, headers=headers, json=data)
return response.json()
def get_contract_state(self) -> dict:
"""Retrieve contract state from the RPC server."""
url = self.rpc_url + "/v1/ethereum/contract_state"
headers = {"accept": "application/json", "Content-Type": "application/json"}
data = {}
response = requests.post(url, headers=headers, json=data)
return response.json()
class TychoRunner:
def __init__(self, with_binary_logs: bool = False):
self.with_binary_logs = with_binary_logs
@@ -142,36 +174,6 @@ class TychoRunner:
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 = {}
response = requests.post(url, headers=headers, json=data)
return response.json()
@staticmethod
def get_contract_state() -> dict:
"""Retrieve contract state from the RPC server."""
url = "http://0.0.0.0:4242/v1/ethereum/contract_state"
headers = {"accept": "application/json", "Content-Type": "application/json"}
data = {}
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."""