kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Reinstate basic tx test

Browse Source
This commit is contained in:
nolash 2021-03-20 22:58:48 +01:00
parent 2eaaedb0f0
commit 06ddfb4fe8
Signed by untrusted user who does not match committer: lash
GPG Key ID: 21D2E7BB88C2A746
3 changed files with 105 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
apps/cic-eth/cic_eth/eth/erc20.py
View File

@ -448,70 +448,69 @@ def cache_approve_data(
return (tx_hash_hex, cache_id)
# TODO: Move to dedicated metadata package
class ExtendedTx:
_default_decimals = 6
def __init__(self, tx_hash, chain_spec):
self._chain_spec = chain_spec
self.chain = str(chain_spec)
self.hash = tx_hash
self.sender = None
self.sender_label = None
self.recipient = None
self.recipient_label = None
self.source_token_value = 0
self.destination_token_value = 0
self.source_token = ZERO_ADDRESS
self.destination_token = ZERO_ADDRESS
self.source_token_symbol = ''
self.destination_token_symbol = ''
self.source_token_decimals = ExtendedTx._default_decimals
self.destination_token_decimals = ExtendedTx._default_decimals
self.status = TxStatus.PENDING.name
self.status_code = TxStatus.PENDING.value
def set_actors(self, sender, recipient, trusted_declarator_addresses=None):
self.sender = sender
self.recipient = recipient
if trusted_declarator_addresses != None:
self.sender_label = translate_address(sender, trusted_declarator_addresses, self.chain)
self.recipient_label = translate_address(recipient, trusted_declarator_addresses, self.chain)
def set_tokens(self, source, source_value, destination=None, destination_value=None):
c = RpcClient(self._chain_spec)
registry = safe_registry(c.w3)
if destination == None:
destination = source
if destination_value == None:
destination_value = source_value
st = registry.get_address(self._chain_spec, source)
dt = registry.get_address(self._chain_spec, destination)
self.source_token = source
self.source_token_symbol = st.symbol()
self.source_token_decimals = st.decimals()
self.source_token_value = source_value
self.destination_token = destination
self.destination_token_symbol = dt.symbol()
self.destination_token_decimals = dt.decimals()
self.destination_token_value = destination_value
def set_status(self, n):
if n:
self.status = TxStatus.ERROR.name
else:
self.status = TxStatus.SUCCESS.name
self.status_code = n
def to_dict(self):
o = {}
for attr in dir(self):
if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'set_status', 'to_dict']:
continue
o[attr] = getattr(self, attr)
return o
#class ExtendedTx:
#
# _default_decimals = 6
#
# def __init__(self, tx_hash, chain_spec):
# self._chain_spec = chain_spec
# self.chain = str(chain_spec)
# self.hash = tx_hash
# self.sender = None
# self.sender_label = None
# self.recipient = None
# self.recipient_label = None
# self.source_token_value = 0
# self.destination_token_value = 0
# self.source_token = ZERO_ADDRESS
# self.destination_token = ZERO_ADDRESS
# self.source_token_symbol = ''
# self.destination_token_symbol = ''
# self.source_token_decimals = ExtendedTx._default_decimals
# self.destination_token_decimals = ExtendedTx._default_decimals
# self.status = TxStatus.PENDING.name
# self.status_code = TxStatus.PENDING.value
#
#
# def set_actors(self, sender, recipient, trusted_declarator_addresses=None):
# self.sender = sender
# self.recipient = recipient
# if trusted_declarator_addresses != None:
# self.sender_label = translate_address(sender, trusted_declarator_addresses, self.chain)
# self.recipient_label = translate_address(recipient, trusted_declarator_addresses, self.chain)
#
#
# def set_tokens(self, source, source_value, destination=None, destination_value=None):
# c = RpcClient(self._chain_spec)
# registry = safe_registry(c.w3)
# if destination == None:
# destination = source
# if destination_value == None:
# destination_value = source_value
# st = registry.get_address(self._chain_spec, source)
# dt = registry.get_address(self._chain_spec, destination)
# self.source_token = source
# self.source_token_symbol = st.symbol()
# self.source_token_decimals = st.decimals()
# self.source_token_value = source_value
# self.destination_token = destination
# self.destination_token_symbol = dt.symbol()
# self.destination_token_decimals = dt.decimals()
# self.destination_token_value = destination_value
#
#
# def set_status(self, n):
# if n:
# self.status = TxStatus.ERROR.name
# else:
# self.status = TxStatus.SUCCESS.name
# self.status_code = n
#
#
# def to_dict(self):
# o = {}
# for attr in dir(self):
# if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'set_status', 'to_dict']:
# continue
# o[attr] = getattr(self, attr)
# return o

30
apps/cic-eth/tests/unit/eth/test_raw.py Normal file
View File

@ -0,0 +1,30 @@
# external imports
from chainlib.eth.gas import (
Gas,
RPCGasOracle,
)
from chainlib.eth.tx import (
TxFormat,
unpack,
)
from chainlib.eth.nonce import RPCNonceOracle
from chainlib.connection import RPCConnection
from hexathon import strip_0x
def test_unpack(
default_chain_spec,
eth_rpc,
eth_signer,
agent_roles,
):
chain_id = default_chain_spec.chain_id()
rpc = RPCConnection.connect(default_chain_spec, 'default')
nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], eth_rpc)
gas_oracle = RPCGasOracle(eth_rpc)
c = Gas(signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_id=default_chain_spec.chain_id())
(tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
tx = unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), chain_id=chain_id)
assert tx_hash_hex == tx['hash']

9
apps/cic-eth/tests/util/gas.py Normal file
View File

@ -0,0 +1,9 @@
class StaticGasOracle:
def __init__(self, price, limit):
self.price = price
self.limit = limit
def get_gas(self):
return (self.price, self.limit)