lbry-sdk/tests/integration/test_reconnect.py

33 lines
1.5 KiB
Python
Raw Normal View History

2018-10-15 06:45:21 +02:00
from orchstr8.testcase import IntegrationTestCase
2018-10-11 05:07:38 +02:00
from torba.constants import COIN
class ReconnectTests(IntegrationTestCase):
VERBOSE = False
async def test_connection_drop_still_receives_events_after_reconnected(self):
2018-10-15 06:45:21 +02:00
address1 = await self.account.receiving.get_or_create_usable_address()
2018-10-11 05:07:38 +02:00
self.ledger.network.client.connectionLost()
sendtxid = await self.blockchain.send_to_address(address1, 1.1337)
await self.on_transaction_id(sendtxid) # mempool
await self.blockchain.generate(1)
await self.on_transaction_id(sendtxid) # confirmed
self.assertEqual(round(await self.get_balance(self.account)/COIN, 4), 1.1337)
# is it real? are we rich!? let me see this tx...
d = self.ledger.network.get_transaction(sendtxid)
# what's that smoke on my ethernet cable? oh no!
self.ledger.network.client.connectionLost()
with self.assertRaisesRegex(TimeoutError, 'Connection dropped'):
2018-10-15 06:45:21 +02:00
await d
2018-10-11 05:07:38 +02:00
# rich but offline? no way, no water, let's retry
with self.assertRaisesRegex(ConnectionError, 'connection is not available'):
2018-10-15 06:45:21 +02:00
await self.ledger.network.get_transaction(sendtxid)
2018-10-11 05:07:38 +02:00
# * goes to pick some water outside... * time passes by and another donation comes in
sendtxid = await self.blockchain.send_to_address(address1, 42)
await self.blockchain.generate(1)
# omg, the burned cable still works! torba is fire proof!
2018-10-15 06:45:21 +02:00
await self.ledger.network.get_transaction(sendtxid)
2018-10-11 05:07:38 +02:00