-WIP-electrum-btcp/lib/tests/test_transaction.py

207 lines
10 KiB
Python

import unittest
from lib import transaction
import pprint
unsigned_blob = '01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65bf38633b424eb4031000000005701ff4c53ff0488b21e03ef2afea18000000089689bff23e1e7fb2f161daa37270a97a3d8c2e537584b2d304ecb47b86d21fc021b010d3bd425f8cf2e04824bfdf1f1f5ff1d51fadd9a41f9e3fb8dd3403b1bfe00000000ffffffff0140420f00000000001976a914230ac37834073a42146f11ef8414ae929feaafc388ac00000000'
signed_blob = '01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65bf38633b424eb4031000000006c493046022100a82bbc57a0136751e5433f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7de89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d985012102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae35cdb84d2f6ffffffff0140420f00000000001976a914230ac37834073a42146f11ef8414ae929feaafc388ac00000000'
class TestBCDataStream(unittest.TestCase):
def test_compact_size(self):
s = transaction.BCDataStream()
values = [0, 1, 252, 253, 2**16-1, 2**16, 2**32-1, 2**32, 2**64-1]
for v in values:
s.write_compact_size(v)
with self.assertRaises(transaction.SerializationError):
s.write_compact_size(-1)
self.assertEquals(s.input.encode('hex'),
'0001fcfdfd00fdfffffe00000100feffffffffff0000000001000000ffffffffffffffffff')
for v in values:
self.assertEquals(s.read_compact_size(), v)
with self.assertRaises(IndexError):
s.read_compact_size()
def test_string(self):
s = transaction.BCDataStream()
with self.assertRaises(transaction.SerializationError):
s.read_string()
msgs = ['Hello', ' ', 'World', '', '!']
for msg in msgs:
s.write_string(msg)
for msg in msgs:
self.assertEquals(s.read_string(), msg)
with self.assertRaises(transaction.SerializationError):
s.read_string()
def test_bytes(self):
s = transaction.BCDataStream()
s.write('foobar')
self.assertEquals(s.read_bytes(3), 'foo')
self.assertEquals(s.read_bytes(2), 'ba')
self.assertEquals(s.read_bytes(4), 'r')
self.assertEquals(s.read_bytes(1), '')
class TestTransaction(unittest.TestCase):
def test_tx_unsigned(self):
expected = {
'inputs': [{
'address': '1446oU3z268EeFgfcwJv6X2VBXHfoYxfuD',
'is_coinbase': False,
'num_sig': 1,
'prevout_hash': '3140eb24b43386f35ba69e3875eb6c93130ac66201d01c58f598defc949a5c2a',
'prevout_n': 0,
'pubkeys': ['02e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae35cdb84d2f6'],
'scriptSig': '01ff4c53ff0488b21e03ef2afea18000000089689bff23e1e7fb2f161daa37270a97a3d8c2e537584b2d304ecb47b86d21fc021b010d3bd425f8cf2e04824bfdf1f1f5ff1d51fadd9a41f9e3fb8dd3403b1bfe00000000',
'sequence': 4294967295,
'signatures': [None],
'x_pubkeys': ['ff0488b21e03ef2afea18000000089689bff23e1e7fb2f161daa37270a97a3d8c2e537584b2d304ecb47b86d21fc021b010d3bd425f8cf2e04824bfdf1f1f5ff1d51fadd9a41f9e3fb8dd3403b1bfe00000000']}],
'lockTime': 0,
'outputs': [{
'address': '14CHYaaByjJZpx4oHBpfDMdqhTyXnZ3kVs',
'prevout_n': 0,
'scriptPubKey': '76a914230ac37834073a42146f11ef8414ae929feaafc388ac',
'type': 'address',
'value': 1000000}],
'version': 1
}
tx = transaction.Transaction(unsigned_blob)
self.assertEquals(tx.deserialize(), expected)
self.assertEquals(tx.deserialize(), None)
self.assertEquals(tx.as_dict(), {'hex': unsigned_blob, 'complete': False})
self.assertEquals(tx.get_outputs(), [('14CHYaaByjJZpx4oHBpfDMdqhTyXnZ3kVs', 1000000)])
self.assertEquals(tx.get_output_addresses(), ['14CHYaaByjJZpx4oHBpfDMdqhTyXnZ3kVs'])
self.assertTrue(tx.has_address('14CHYaaByjJZpx4oHBpfDMdqhTyXnZ3kVs'))
self.assertTrue(tx.has_address('1446oU3z268EeFgfcwJv6X2VBXHfoYxfuD'))
self.assertFalse(tx.has_address('1CQj15y1N7LDHp7wTt28eoD1QhHgFgxECH'))
self.assertEquals(tx.inputs_to_sign(), set(x_pubkey for i in expected['inputs'] for x_pubkey in i['x_pubkeys']))
self.assertEquals(tx.serialize(), unsigned_blob)
tx.update_signatures(signed_blob)
self.assertEquals(tx.raw, signed_blob)
tx.update(unsigned_blob)
tx.raw = None
blob = str(tx)
self.assertEquals(transaction.deserialize(blob), expected)
def test_tx_signed(self):
expected = {
'inputs': [{
'address': '1446oU3z268EeFgfcwJv6X2VBXHfoYxfuD',
'is_coinbase': False,
'num_sig': 1,
'prevout_hash': '3140eb24b43386f35ba69e3875eb6c93130ac66201d01c58f598defc949a5c2a',
'prevout_n': 0,
'pubkeys': ['02e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae35cdb84d2f6'],
'scriptSig': '493046022100a82bbc57a0136751e5433f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7de89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d985012102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae35cdb84d2f6',
'sequence': 4294967295,
'signatures': ['3046022100a82bbc57a0136751e5433f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7de89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d985'],
'x_pubkeys': ['02e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae35cdb84d2f6']}],
'lockTime': 0,
'outputs': [{
'address': '14CHYaaByjJZpx4oHBpfDMdqhTyXnZ3kVs',
'prevout_n': 0,
'scriptPubKey': '76a914230ac37834073a42146f11ef8414ae929feaafc388ac',
'type': 'address',
'value': 1000000}],
'version': 1
}
tx = transaction.Transaction(signed_blob)
self.assertEquals(tx.deserialize(), expected)
self.assertEquals(tx.deserialize(), None)
self.assertEquals(tx.as_dict(), {'hex': signed_blob, 'complete': True})
self.assertEquals(tx.inputs_to_sign(), set())
self.assertEquals(tx.serialize(), signed_blob)
tx.update_signatures(signed_blob)
def test_errors(self):
with self.assertRaises(TypeError):
transaction.Transaction.pay_script(output_type=None, addr='')
with self.assertRaises(BaseException):
transaction.parse_xpub('')
def test_parse_xpub(self):
res = transaction.parse_xpub('fe4e13b0f311a55b8a5db9a32e959da9f011b131019d4cebe6141b9e2c93edcbfc0954c358b062a9f94111548e50bde5847a3096b8b7872dcffadb0e9579b9017b01000200')
self.assertEquals(res, ('04ee98d63800824486a1cf5b4376f2f574d86e0a3009a6448105703453f3368e8e1d8d090aaecdd626a45cc49876709a3bbb6dc96a4311b3cac03e225df5f63dfc', '19h943e4diLc68GXW7G75QNe2KWuMu7BaJ'))
res = transaction.parse_xpub('fd007d260305ef27224bbcf6cf5238d2b3638b5a78d5')
self.assertEquals(res, (None, '1CQj15y1N7LDHp7wTt28eoD1QhHgFgxECH'))
def test_sign_tx(self):
tx = transaction.Transaction(unsigned_blob)
tx.deserialize()
x_pubkey = 'ff0488b21e03ef2afea18000000089689bff23e1e7fb2f161daa37270a97a3d8c2e537584b2d304ecb47b86d21fc021b010d3bd425f8cf2e04824bfdf1f1f5ff1d51fadd9a41f9e3fb8dd3403b1bfe00000000'
privkey = 'L187zmkzzGgf9QdB23MrZvwJ52WoZuQHtkddjmePtbVjXxicJND2'
tx.sign(keypairs={x_pubkey: privkey})
self.assertEquals(tx.serialize(), signed_blob)
tx.sign(keypairs={x_pubkey: privkey})
self.assertEquals(tx.serialize(), signed_blob)
def test_sweep(self):
privkeys = ['5HuH1SHoSVrgtPEwew9JzVAHGoKyp47x564mBCTgVmUT2Me1Q18']
unspent = [
{
"height": 371447,
"tx_hash": "8e4d173db094786cc128b0c12eebc2200c0d8bfc3ad04ba39f487222d18bae3c",
"tx_pos": 0,
"value": 599995800
}
]
to_address = '1JtBahwvii2pRkBmb4QMfcJQux1rk3Jkbq'
network = NetworkMock(unspent)
tx = transaction.Transaction.sweep(privkeys, network, to_address, fee=5000)
result = transaction.deserialize(tx.serialize())
expected = {
'inputs': [{
'address': '1t28kmZypcPQrunmJk212dcPGPxbBtB6Y',
'is_coinbase': False,
'num_sig': 1,
'prevout_hash': '8e4d173db094786cc128b0c12eebc2200c0d8bfc3ad04ba39f487222d18bae3c',
'prevout_n': 0,
'pubkeys': ['047b9f9014f8d0d6f24dcaf5681b6ab185bd821e0fcce29d84e0452845baf1b2dbe332a7cd4dbdab786adda6d71b2188298c756b265c63de2794f7317b71a7ac02'],
'scriptSig': '48304502203f1ff200490d18bcb802c7cf7ba4264727b089f1db6746a62997285b5ac77969022100e495591ea5111bb23a782984736f32942570fda781b7ed085fc8c88a9756aaac0141047b9f9014f8d0d6f24dcaf5681b6ab185bd821e0fcce29d84e0452845baf1b2dbe332a7cd4dbdab786adda6d71b2188298c756b265c63de2794f7317b71a7ac02',
'sequence': 4294967295,
'signatures': ['304502203f1ff200490d18bcb802c7cf7ba4264727b089f1db6746a62997285b5ac77969022100e495591ea5111bb23a782984736f32942570fda781b7ed085fc8c88a9756aaac'],
'x_pubkeys': ['047b9f9014f8d0d6f24dcaf5681b6ab185bd821e0fcce29d84e0452845baf1b2dbe332a7cd4dbdab786adda6d71b2188298c756b265c63de2794f7317b71a7ac02']}],
'lockTime': 0,
'outputs': [{'address': '1JtBahwvii2pRkBmb4QMfcJQux1rk3Jkbq',
'prevout_n': 0,
'scriptPubKey': '76a914c4282f6060b811ee695ebb2068b8788213451d6a88ac',
'type': 'address',
'value': 599990800}],
'version': 1}
self.assertEquals(result, expected)
network = NetworkMock([])
tx = transaction.Transaction.sweep(privkeys, network, to_address, fee=5000)
self.assertEquals(tx, None)
privkeys = []
tx = transaction.Transaction.sweep(privkeys, network, to_address, fee=5000)
self.assertEquals(tx, None)
class NetworkMock(object):
def __init__(self, unspent):
self.unspent = unspent
def synchronous_get(self, arg):
return self.unspent