chainqueue/tests/test_entry.py

64 lines
1.6 KiB
Python
Raw Normal View History

2022-03-11 20:38:12 +01:00
# standard imports
import os
import logging
import unittest
# external imports
from hexathon import add_0x
# local imports
from chainqueue import QueueEntry
# test imports
from tests.base_shep import TestShepBase
logging.basicConfig(level=logging.DEBUG)
logg = logging.getLogger()
2022-03-12 09:48:19 +01:00
class TestEntry(TestShepBase):
2022-03-11 20:38:12 +01:00
def test_entry_get(self):
2022-03-11 22:49:23 +01:00
tx_hash_one = add_0x(os.urandom(32).hex())
2022-03-11 20:38:12 +01:00
signed_tx = add_0x(os.urandom(128).hex())
nonce = 42
entry = QueueEntry(self.store, tx_hash_one)
2022-03-12 09:48:19 +01:00
entry.create(signed_tx)
2022-03-11 20:38:12 +01:00
2022-03-11 22:49:23 +01:00
tx_hash_two = add_0x(os.urandom(32).hex())
2022-03-11 20:38:12 +01:00
signed_tx = add_0x(os.urandom(128).hex())
nonce = 42
entry = QueueEntry(self.store, tx_hash_two)
2022-03-12 09:48:19 +01:00
entry.create(signed_tx)
2022-03-11 20:38:12 +01:00
2022-03-12 14:48:40 +01:00
txs = self.store.by_state()
2022-03-11 20:38:12 +01:00
self.assertEqual(len(txs), 2)
entry = QueueEntry(self.store, tx_hash_one)
entry.load()
entry.sent()
2022-03-12 14:48:40 +01:00
txs = self.store.by_state()
2022-03-11 20:38:12 +01:00
self.assertEqual(len(txs), 1)
2022-03-12 14:48:40 +01:00
txs = self.store.by_state(state=self.store.IN_NETWORK)
2022-03-11 20:38:12 +01:00
self.assertEqual(len(txs), 1)
entry.succeed(0)
2022-03-12 14:48:40 +01:00
txs = self.store.by_state()
2022-03-11 20:38:12 +01:00
self.assertEqual(len(txs), 1)
entry = QueueEntry(self.store, tx_hash_two)
entry.load()
entry.sent()
2022-03-12 14:48:40 +01:00
txs = self.store.by_state(state=self.store.IN_NETWORK)
2022-03-11 20:38:12 +01:00
self.assertEqual(len(txs), 2)
2022-03-12 14:48:40 +01:00
txs = self.store.by_state(state=self.store.IN_NETWORK, strict=True)
2022-03-11 20:43:00 +01:00
self.assertEqual(len(txs), 1)
2022-03-11 20:38:12 +01:00
if __name__ == '__main__':
unittest.main()