forked from LBRYCommunity/lbry-sdk
289 lines
11 KiB
Python
289 lines
11 KiB
Python
from binascii import hexlify, unhexlify
|
|
from itertools import cycle
|
|
from twisted.trial import unittest
|
|
from twisted.internet import defer
|
|
|
|
from torba.coin.bitcoinsegwit import MainNetLedger as ledger_class
|
|
from torba.constants import CENT, COIN
|
|
|
|
|
|
NULL_HASH = b'\x00'*32
|
|
FEE_PER_BYTE = 50
|
|
FEE_PER_CHAR = 200000
|
|
|
|
|
|
def get_output(amount=CENT, pubkey_hash=NULL_HASH):
|
|
return ledger_class.transaction_class() \
|
|
.add_outputs([ledger_class.transaction_class.output_class.pay_pubkey_hash(amount, pubkey_hash)]) \
|
|
.outputs[0]
|
|
|
|
|
|
def get_input(amount=CENT, pubkey_hash=NULL_HASH):
|
|
return ledger_class.transaction_class.input_class.spend(get_output(amount, pubkey_hash))
|
|
|
|
|
|
def get_transaction(txo=None):
|
|
return ledger_class.transaction_class() \
|
|
.add_inputs([get_input()]) \
|
|
.add_outputs([txo or ledger_class.transaction_class.output_class.pay_pubkey_hash(CENT, NULL_HASH)])
|
|
|
|
|
|
class TestSizeAndFeeEstimation(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.ledger = ledger_class({'db': ledger_class.database_class(':memory:')})
|
|
|
|
def test_output_size_and_fee(self):
|
|
txo = get_output()
|
|
self.assertEqual(txo.size, 46)
|
|
self.assertEqual(txo.get_fee(self.ledger), 46 * FEE_PER_BYTE)
|
|
|
|
def test_input_size_and_fee(self):
|
|
txi = get_input()
|
|
self.assertEqual(txi.size, 148)
|
|
self.assertEqual(txi.get_fee(self.ledger), 148 * FEE_PER_BYTE)
|
|
|
|
def test_transaction_size_and_fee(self):
|
|
tx = get_transaction()
|
|
self.assertEqual(tx.size, 204)
|
|
self.assertEqual(tx.base_size, tx.size - tx.inputs[0].size - tx.outputs[0].size)
|
|
self.assertEqual(tx.get_base_fee(self.ledger), FEE_PER_BYTE * tx.base_size)
|
|
|
|
|
|
class TestTransactionSerialization(unittest.TestCase):
|
|
|
|
def test_genesis_transaction(self):
|
|
raw = unhexlify(
|
|
'01000000010000000000000000000000000000000000000000000000000000000000000000ffffffff4d04'
|
|
'ffff001d0104455468652054696d65732030332f4a616e2f32303039204368616e63656c6c6f72206f6e20'
|
|
'6272696e6b206f66207365636f6e64206261696c6f757420666f722062616e6b73ffffffff0100f2052a01'
|
|
'000000434104678afdb0fe5548271967f1a67130b7105cd6a828e03909a67962e0ea1f61deb649f6bc3f4c'
|
|
'ef38c4f35504e51ec112de5c384df7ba0b8d578a4c702b6bf11d5fac00000000'
|
|
)
|
|
tx = ledger_class.transaction_class(raw)
|
|
self.assertEqual(tx.version, 1)
|
|
self.assertEqual(tx.locktime, 0)
|
|
self.assertEqual(len(tx.inputs), 1)
|
|
self.assertEqual(len(tx.outputs), 1)
|
|
|
|
coinbase = tx.inputs[0]
|
|
self.assertTrue(coinbase.txo_ref.is_null, NULL_HASH)
|
|
self.assertEqual(coinbase.txo_ref.position, 0xFFFFFFFF)
|
|
self.assertEqual(coinbase.sequence, 4294967295)
|
|
self.assertIsNotNone(coinbase.coinbase)
|
|
self.assertIsNone(coinbase.script)
|
|
self.assertEqual(
|
|
coinbase.coinbase[8:],
|
|
b'The Times 03/Jan/2009 Chancellor on brink of second bailout for banks'
|
|
)
|
|
|
|
out = tx.outputs[0]
|
|
self.assertEqual(out.amount, 5000000000)
|
|
self.assertEqual(out.position, 0)
|
|
self.assertTrue(out.script.is_pay_pubkey)
|
|
self.assertFalse(out.script.is_pay_pubkey_hash)
|
|
self.assertFalse(out.script.is_pay_script_hash)
|
|
|
|
tx._reset()
|
|
self.assertEqual(tx.raw, raw)
|
|
|
|
def test_coinbase_transaction(self):
|
|
raw = unhexlify(
|
|
'01000000010000000000000000000000000000000000000000000000000000000000000000ffffffff4e03'
|
|
'1f5a070473319e592f4254432e434f4d2f4e59412ffabe6d6dcceb2a9d0444c51cabc4ee97a1a000036ca0'
|
|
'cb48d25b94b78c8367d8b868454b0100000000000000c0309b21000008c5f8f80000ffffffff0291920b5d'
|
|
'0000000017a914e083685a1097ce1ea9e91987ab9e94eae33d8a13870000000000000000266a24aa21a9ed'
|
|
'e6c99265a6b9e1d36c962fda0516b35709c49dc3b8176fa7e5d5f1f6197884b400000000'
|
|
)
|
|
tx = ledger_class.transaction_class(raw)
|
|
self.assertEqual(tx.version, 1)
|
|
self.assertEqual(tx.locktime, 0)
|
|
self.assertEqual(len(tx.inputs), 1)
|
|
self.assertEqual(len(tx.outputs), 2)
|
|
|
|
coinbase = tx.inputs[0]
|
|
self.assertTrue(coinbase.txo_ref.is_null)
|
|
self.assertEqual(coinbase.txo_ref.position, 0xFFFFFFFF)
|
|
self.assertEqual(coinbase.sequence, 4294967295)
|
|
self.assertIsNotNone(coinbase.coinbase)
|
|
self.assertIsNone(coinbase.script)
|
|
self.assertEqual(coinbase.coinbase[9:22], b'/BTC.COM/NYA/')
|
|
|
|
out = tx.outputs[0]
|
|
self.assertEqual(out.amount, 1561039505)
|
|
self.assertEqual(out.position, 0)
|
|
self.assertFalse(out.script.is_pay_pubkey)
|
|
self.assertFalse(out.script.is_pay_pubkey_hash)
|
|
self.assertTrue(out.script.is_pay_script_hash)
|
|
self.assertFalse(out.script.is_return_data)
|
|
|
|
out1 = tx.outputs[1]
|
|
self.assertEqual(out1.amount, 0)
|
|
self.assertEqual(out1.position, 1)
|
|
self.assertEqual(
|
|
hexlify(out1.script.values['data']),
|
|
b'aa21a9ede6c99265a6b9e1d36c962fda0516b35709c49dc3b8176fa7e5d5f1f6197884b4'
|
|
)
|
|
self.assertTrue(out1.script.is_return_data)
|
|
self.assertFalse(out1.script.is_pay_pubkey)
|
|
self.assertFalse(out1.script.is_pay_pubkey_hash)
|
|
self.assertFalse(out1.script.is_pay_script_hash)
|
|
|
|
tx._reset()
|
|
self.assertEqual(tx.raw, raw)
|
|
|
|
|
|
class TestTransactionSigning(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.ledger = ledger_class({'db': ledger_class.database_class(':memory:')})
|
|
return self.ledger.db.start()
|
|
|
|
@defer.inlineCallbacks
|
|
def test_sign(self):
|
|
account = self.ledger.account_class.from_dict(
|
|
self.ledger, {
|
|
"seed": "carbon smart garage balance margin twelve chest sword "
|
|
"toast envelope bottom stomach absent"
|
|
|
|
}
|
|
)
|
|
|
|
yield account.ensure_address_gap()
|
|
address1, address2 = yield account.receiving.get_addresses(2)
|
|
pubkey_hash1 = self.ledger.address_to_hash160(address1)
|
|
pubkey_hash2 = self.ledger.address_to_hash160(address2)
|
|
|
|
tx_class = ledger_class.transaction_class
|
|
|
|
tx = tx_class() \
|
|
.add_inputs([tx_class.input_class.spend(get_output(2*COIN, pubkey_hash1))]) \
|
|
.add_outputs([tx_class.output_class.pay_pubkey_hash(int(1.9*COIN), pubkey_hash2)]) \
|
|
|
|
yield tx.sign([account])
|
|
|
|
print(hexlify(tx.inputs[0].script.values['signature']))
|
|
self.assertEqual(
|
|
hexlify(tx.inputs[0].script.values['signature']),
|
|
b'304402205a1df8cd5d2d2fa5934b756883d6c07e4f83e1350c740992d47a12422'
|
|
b'226aaa202200098ac8675827aea2b0d6f0e49566143a95d523e311d342172cd99e2021e47cb01'
|
|
)
|
|
|
|
|
|
class TransactionIOBalancing(unittest.TestCase):
|
|
|
|
@defer.inlineCallbacks
|
|
def setUp(self):
|
|
self.ledger = ledger_class({'db': ledger_class.database_class(':memory:')})
|
|
yield self.ledger.db.start()
|
|
self.account = self.ledger.account_class.from_dict(
|
|
self.ledger, {
|
|
"seed": "carbon smart garage balance margin twelve chest sword "
|
|
"toast envelope bottom stomach absent"
|
|
}
|
|
)
|
|
|
|
addresses = yield self.account.ensure_address_gap()
|
|
self.pubkey_hash = [self.ledger.address_to_hash160(a) for a in addresses]
|
|
self.hash_cycler = cycle(self.pubkey_hash)
|
|
|
|
def txo(self, amount, address=None):
|
|
return get_output(int(amount*COIN), address or next(self.hash_cycler))
|
|
|
|
def txi(self, txo):
|
|
return ledger_class.transaction_class.input_class.spend(txo)
|
|
|
|
def tx(self, inputs, outputs):
|
|
return ledger_class.transaction_class.create(inputs, outputs, [self.account], self.account)
|
|
|
|
@defer.inlineCallbacks
|
|
def create_utxos(self, amounts):
|
|
utxos = [self.txo(amount) for amount in amounts]
|
|
|
|
self.funding_tx = ledger_class.transaction_class() \
|
|
.add_inputs([self.txi(self.txo(sum(amounts)+0.1))]) \
|
|
.add_outputs(utxos)
|
|
|
|
save_tx = 'insert'
|
|
for utxo in utxos:
|
|
yield self.ledger.db.save_transaction_io(
|
|
save_tx, self.funding_tx, 1, True,
|
|
self.ledger.hash160_to_address(utxo.script.values['pubkey_hash']),
|
|
utxo.script.values['pubkey_hash'], ''
|
|
)
|
|
save_tx = 'update'
|
|
|
|
defer.returnValue(utxos)
|
|
|
|
@staticmethod
|
|
def inputs(tx):
|
|
return [round(i.amount/COIN, 2) for i in tx.inputs]
|
|
|
|
@staticmethod
|
|
def outputs(tx):
|
|
return [round(o.amount/COIN, 2) for o in tx.outputs]
|
|
|
|
@defer.inlineCallbacks
|
|
def test_basic_use_cases(self):
|
|
self.ledger.fee_per_byte = int(.01*CENT)
|
|
|
|
# available UTXOs for filling missing inputs
|
|
utxos = yield self.create_utxos([
|
|
1, 1, 3, 5, 10
|
|
])
|
|
|
|
# pay 3 coins (3.02 w/ fees)
|
|
tx = yield self.tx(
|
|
[], # inputs
|
|
[self.txo(3)] # outputs
|
|
)
|
|
# best UTXO match is 5 (as UTXO 3 will be short 0.02 to cover fees)
|
|
self.assertEqual(self.inputs(tx), [5])
|
|
# a change of 1.98 is added to reach balance
|
|
self.assertEqual(self.outputs(tx), [3, 1.98])
|
|
|
|
yield self.ledger.release_outputs(utxos)
|
|
|
|
# pay 2.98 coins (3.00 w/ fees)
|
|
tx = yield self.tx(
|
|
[], # inputs
|
|
[self.txo(2.98)] # outputs
|
|
)
|
|
# best UTXO match is 3 and no change is needed
|
|
self.assertEqual(self.inputs(tx), [3])
|
|
self.assertEqual(self.outputs(tx), [2.98])
|
|
|
|
yield self.ledger.release_outputs(utxos)
|
|
|
|
# supplied input and output, but input is not enough to cover output
|
|
tx = yield self.tx(
|
|
[self.txi(self.txo(10))], # inputs
|
|
[self.txo(11)] # outputs
|
|
)
|
|
# additional input is chosen (UTXO 3)
|
|
self.assertEqual([10, 3], self.inputs(tx))
|
|
# change is now needed to consume extra input
|
|
self.assertEqual([11, 1.96], self.outputs(tx))
|
|
|
|
yield self.ledger.release_outputs(utxos)
|
|
|
|
# liquidating a UTXO
|
|
tx = yield self.tx(
|
|
[self.txi(self.txo(10))], # inputs
|
|
[] # outputs
|
|
)
|
|
self.assertEqual([10], self.inputs(tx))
|
|
# missing change added to consume the amount
|
|
self.assertEqual([9.98], self.outputs(tx))
|
|
|
|
yield self.ledger.release_outputs(utxos)
|
|
|
|
# liquidating at a loss, requires adding extra inputs
|
|
tx = yield self.tx(
|
|
[self.txi(self.txo(0.01))], # inputs
|
|
[] # outputs
|
|
)
|
|
# UTXO 1 is added to cover some of the fee
|
|
self.assertEqual([0.01, 1], self.inputs(tx))
|
|
# change is now needed to consume extra input
|
|
self.assertEqual([0.97], self.outputs(tx))
|