add filter test

This commit is contained in:
nolash 2021-07-16 21:48:30 +02:00
parent 1929050e97
commit c35957275c
Signed by: lash
GPG Key ID: 21D2E7BB88C2A746
3 changed files with 104 additions and 1 deletions

View File

@ -0,0 +1,3 @@
eth_tester==0.5.0b3
py-evm==0.3.0a20
rlp==2.0.1

View File

@ -4,11 +4,13 @@ import unittest
# external imports # external imports
from chainsyncer.unittest.db import ChainSyncerDb from chainsyncer.unittest.db import ChainSyncerDb
from chainqueue.unittest.db import ChainQueueDb from chainqueue.unittest.db import ChainQueueDb
from chainlib.eth.unittest.ethtester import EthTesterCase
class TestBase(unittest.TestCase): class TestBase(EthTesterCase):
def setUp(self): def setUp(self):
super(TestBase, self).setUp()
self.db_chainsyncer = ChainSyncerDb() self.db_chainsyncer = ChainSyncerDb()
self.session_chainsyncer = self.db_chainsyncer.bind_session() self.session_chainsyncer = self.db_chainsyncer.bind_session()
@ -20,3 +22,4 @@ class TestBase(unittest.TestCase):
self.db_chainsyncer.release_session(self.session_chainsyncer) self.db_chainsyncer.release_session(self.session_chainsyncer)
self.session_chainqueue.commit() self.session_chainqueue.commit()
self.db_chainqueue.release_session(self.session_chainqueue) self.db_chainqueue.release_session(self.session_chainqueue)
super(TestBase, self).tearDown()

97
tests/test_filter.py Normal file
View File

@ -0,0 +1,97 @@
# standard imports
import unittest
import logging
import os
# external imports
from potaahto.symbols import snake_and_camel
from hexathon import (
strip_0x,
)
from chainqueue.adapters.eth import EthAdapter
from chainqueue.unittest.db import (
db_config,
dsn_from_config,
)
from chainqueue.sql.backend import SQLBackend
from chainqueue.enum import is_alive
from chainqueue.sql.query import get_tx
from chainlib.eth.gas import (
RPCGasOracle,
Gas,
)
from chainlib.eth.nonce import (
RPCNonceOracle,
)
from chainlib.eth.tx import (
TxFormat,
raw,
unpack,
receipt,
Tx,
)
from chainlib.eth.block import (
block_by_hash,
Block,
)
from chainqueue.sql.state import (
set_sent,
set_reserved,
set_ready,
)
# local imports
from chaind_eth.filter import StateFilter
# test imports
from tests.chaind_eth_base import TestBase
logging.basicConfig(level=logging.DEBUG)
logg = logging.getLogger()
class TestFilter(TestBase):
def test_filter(self):
gas_oracle = RPCGasOracle(conn=self.rpc)
nonce_oracle = RPCNonceOracle(self.accounts[0], conn=self.rpc)
c = Gas(self.chain_spec, signer=self.signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
(tx_hash, tx_raw_rlp_signed) = c.create(self.accounts[0], self.accounts[1], 1024, tx_format=TxFormat.RLP_SIGNED)
o = raw(tx_raw_rlp_signed)
self.rpc.do(o)
o = receipt(tx_hash)
rcpt = self.rpc.do(o)
o = block_by_hash(rcpt['block_hash'])
block_src = self.rpc.do(o)
block = Block(block_src)
dsn = dsn_from_config(db_config)
backend = SQLBackend(dsn, debug=bool(os.environ.get('DATABASE_DEBUG')))
adapter = EthAdapter(backend)
tx_raw_rlp_signed_bytes = bytes.fromhex(strip_0x(tx_raw_rlp_signed))
adapter.add(self.chain_spec, tx_raw_rlp_signed_bytes, session=self.session_chainqueue)
set_ready(self.chain_spec, tx_hash, session=self.session_chainqueue)
set_reserved(self.chain_spec, tx_hash, session=self.session_chainqueue)
set_sent(self.chain_spec, tx_hash, session=self.session_chainqueue)
tx_src = unpack(tx_raw_rlp_signed_bytes, self.chain_spec)
tx = Tx(tx_src, block=block, rcpt=rcpt)
tx_repr = get_tx(self.chain_spec, tx_hash, session=self.session_chainqueue)
assert is_alive(tx_repr['status'])
fltr = StateFilter(self.chain_spec)
fltr.filter(self.rpc, block, tx, session=self.session_chainqueue)
tx_repr = get_tx(self.chain_spec, tx_hash, session=self.session_chainqueue)
assert not is_alive(tx_repr['status'])
if __name__ == '__main__':
unittest.main()