forked from LBRYCommunity/lbry-sdk
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
import logging
|
|
import asyncio
|
|
from torba.testcase import IntegrationTestCase
|
|
from torba.client.constants import COIN
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
log.setLevel(logging.INFO)
|
|
|
|
|
|
class BasicTransactionTests(IntegrationTestCase):
|
|
|
|
VERBOSITY = logging.WARN
|
|
|
|
async def test_stressing(self):
|
|
await self.blockchain.generate(1000)
|
|
await self.assertBalance(self.account, '0.0')
|
|
address1 = await self.account.receiving.get_or_create_usable_address()
|
|
hash1 = self.ledger.address_to_hash160(address1)
|
|
|
|
txids = await asyncio.gather(*(
|
|
self.blockchain.send_to_address(address1, 100)
|
|
for _ in range(10)
|
|
))
|
|
|
|
await asyncio.wait([
|
|
self.on_transaction_id(txid)
|
|
for txid in txids
|
|
])
|
|
|
|
await self.assertBalance(self.account, '1000.0')
|
|
|
|
tasks = []
|
|
for _ in range(10):
|
|
tx = await self.ledger.transaction_class.create(
|
|
[],
|
|
[self.ledger.transaction_class.output_class.pay_pubkey_hash(1*COIN, hash1)],
|
|
[self.account], self.account
|
|
)
|
|
await self.broadcast(tx)
|
|
tasks.append(asyncio.create_task(self.ledger.wait(tx)))
|
|
|
|
await asyncio.wait(tasks)
|
|
|
|
await self.assertBalance(self.account, '999.99876')
|
|
|
|
async def test_sending_and_receiving(self):
|
|
account1, account2 = self.account, self.wallet.generate_account(self.ledger)
|
|
await self.ledger.subscribe_account(account2)
|
|
|
|
await self.assertBalance(account1, '0.0')
|
|
await self.assertBalance(account2, '0.0')
|
|
|
|
sendtxids = []
|
|
for i in range(5):
|
|
address1 = await account1.receiving.get_or_create_usable_address()
|
|
sendtxid = await self.blockchain.send_to_address(address1, 1.1)
|
|
sendtxids.append(sendtxid)
|
|
await self.on_transaction_id(sendtxid) # mempool
|
|
await self.blockchain.generate(1)
|
|
await asyncio.wait([ # confirmed
|
|
self.on_transaction_id(txid) for txid in sendtxids
|
|
])
|
|
|
|
await self.assertBalance(account1, '5.5')
|
|
await self.assertBalance(account2, '0.0')
|
|
|
|
address2 = await account2.receiving.get_or_create_usable_address()
|
|
hash2 = self.ledger.address_to_hash160(address2)
|
|
tx = await self.ledger.transaction_class.create(
|
|
[],
|
|
[self.ledger.transaction_class.output_class.pay_pubkey_hash(2*COIN, hash2)],
|
|
[account1], account1
|
|
)
|
|
await self.broadcast(tx)
|
|
await self.ledger.wait(tx) # mempool
|
|
await self.blockchain.generate(1)
|
|
await self.ledger.wait(tx) # confirmed
|
|
|
|
await self.assertBalance(account1, '3.499802')
|
|
await self.assertBalance(account2, '2.0')
|
|
|
|
utxos = await self.account.get_utxos()
|
|
tx = await self.ledger.transaction_class.create(
|
|
[self.ledger.transaction_class.input_class.spend(utxos[0])],
|
|
[],
|
|
[account1], account1
|
|
)
|
|
await self.broadcast(tx)
|
|
await self.ledger.wait(tx) # mempool
|
|
await self.blockchain.generate(1)
|
|
await self.ledger.wait(tx) # confirmed
|
|
|
|
txs = await account1.get_transactions()
|
|
tx = txs[1]
|
|
self.assertEqual(round(tx.inputs[0].txo_ref.txo.amount/COIN, 1), 1.1)
|
|
self.assertEqual(round(tx.inputs[1].txo_ref.txo.amount/COIN, 1), 1.1)
|
|
self.assertEqual(round(tx.outputs[0].amount/COIN, 1), 2.0)
|
|
self.assertEqual(tx.outputs[0].get_address(self.ledger), address2)
|
|
self.assertEqual(tx.outputs[0].is_change, False)
|
|
self.assertEqual(tx.outputs[1].is_change, True)
|