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

fix(balancer): Get all balancer tests working.

Browse Source
This commit is contained in:
kayibal
2024-08-01 12:03:01 +02:00
parent a6cff51bf6
commit 75b66a85af
8 changed files with 187 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -85,7 +85,6 @@ class TestRunner:
else:
print(f"❗️ {test['name']} failed: {result.message}")
def validate_state(self, expected_state: dict, stop_block: int) -> TestResult:
"""Validate the current protocol state against the expected state."""
protocol_components = self.tycho_rpc_client.get_protocol_components()
@@ -153,7 +152,8 @@ class TestRunner:
contract_states = self.tycho_rpc_client.get_contract_state()
filtered_components = {'protocol_components': [pc for pc in protocol_components["protocol_components"] if
pc["id"] in [c["id"].lower() for c in
expected_state["protocol_components"] if c["skip_simulation"] is False]]}
expected_state["protocol_components"] if
c.get("skip_simulation", False) is False]]}
simulation_failures = self.simulate_get_amount_out(
stop_block,
protocol_states,
@@ -198,7 +198,7 @@ class TestRunner:
self.adapters_src, "out", f"{self.config['adapter_contract']}.sol",
f"{self.config['adapter_contract']}.evm.runtime"
)
decoder = ThirdPartyPoolTychoDecoder(adapter_contract, 0, False)
decoder = ThirdPartyPoolTychoDecoder(adapter_contract, 0)
stream_adapter = TychoPoolStateStreamAdapter(
tycho_url="0.0.0.0:4242",
protocol=protocol,

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

@@ -20,10 +20,9 @@ log = getLogger(__name__)
class ThirdPartyPoolTychoDecoder:
"""ThirdPartyPool decoder for protocol messages from the Tycho feed"""
def __init__(self, adapter_contract: str, minimum_gas: int, hard_limit: bool):
def __init__(self, adapter_contract: str, minimum_gas: int):
self.adapter_contract = adapter_contract
self.minimum_gas = minimum_gas
self.hard_limit = hard_limit
def decode_snapshot(
self,
@@ -68,8 +67,7 @@ class ThirdPartyPoolTychoDecoder:
exchange=exchange,
adapter_contract_name=self.adapter_contract,
minimum_gas=self.minimum_gas,
hard_sell_limit=self.hard_limit,
trace=True,
trace=False,
**optional_attributes,
)
@@ -77,15 +75,16 @@ class ThirdPartyPoolTychoDecoder:
def decode_optional_attributes(component, snap, block_number):
# Handle optional state attributes
attributes = snap["state"]["attributes"]
pool_id = attributes.get("pool_id") or component["id"]
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')
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:
@@ -94,13 +93,13 @@ class ThirdPartyPoolTychoDecoder:
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
index += 1
return {
"balance_owner": balance_owner,
"pool_id": pool_id,

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

@@ -26,10 +26,10 @@ log = getLogger(__name__)
class TokenLoader:
def __init__(
self,
tycho_url: str,
blockchain: Blockchain,
min_token_quality: Optional[int] = 0,
self,
tycho_url: str,
blockchain: Blockchain,
min_token_quality: Optional[int] = 0,
):
self.tycho_url = tycho_url
self.blockchain = blockchain
@@ -45,10 +45,10 @@ class TokenLoader:
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},
url=url,
page=page,
limit=self._token_limit,
params={"min_quality": self.min_token_quality},
):
all_tokens.extend(data)
page += 1
@@ -73,10 +73,10 @@ class TokenLoader:
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},
url=url,
page=page,
limit=self._token_limit,
params={"min_quality": self.min_token_quality, "addresses": addresses},
):
all_tokens.extend(data)
page += 1
@@ -95,7 +95,7 @@ class TokenLoader:
@staticmethod
def _get_all_with_pagination(
url: str, params: Optional[Dict] = None, page: int = 0, limit: int = 50
url: str, params: Optional[Dict] = None, page: int = 0, limit: int = 50
) -> Dict:
if params is None:
params = {}
@@ -122,14 +122,14 @@ class BlockProtocolChanges:
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,
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
@@ -238,7 +238,7 @@ class TychoPoolStateStreamAdapter:
@staticmethod
def build_snapshot_message(
protocol_components: dict, protocol_states: dict, contract_states: dict
protocol_components: dict, protocol_states: dict, contract_states: dict
) -> dict[str, ThirdPartyPool]:
vm_states = {state["address"]: state for state in contract_states["accounts"]}
states = {}
@@ -248,7 +248,7 @@ class TychoPoolStateStreamAdapter:
for state in protocol_states["states"]:
pool_id = state["component_id"]
if pool_id not in states:
log.warning(f"State for pool {pool_id} not found in components")
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}
@@ -269,7 +269,7 @@ class TychoPoolStateStreamAdapter:
return self.process_snapshot(block, state_msg["snapshot"])
def process_snapshot(
self, block: EVMBlock, state_msg: dict
self, block: EVMBlock, state_msg: dict
) -> BlockProtocolChanges:
start = time.monotonic()
removed_pools = set()