Changeset View
Changeset View
Standalone View
Standalone View
test/functional/chronik_block_txs.py
Show All 23 Lines | |||||
from test_framework.test_framework import BitcoinTestFramework | from test_framework.test_framework import BitcoinTestFramework | ||||
from test_framework.util import assert_equal | from test_framework.util import assert_equal | ||||
class ChronikBlockTxsTest(BitcoinTestFramework): | class ChronikBlockTxsTest(BitcoinTestFramework): | ||||
def set_test_params(self): | def set_test_params(self): | ||||
self.setup_clean_chain = True | self.setup_clean_chain = True | ||||
self.num_nodes = 1 | self.num_nodes = 1 | ||||
self.extra_args = [['-chronik']] | self.extra_args = [["-chronik"]] | ||||
self.rpc_timeout = 240 | self.rpc_timeout = 240 | ||||
def skip_test_if_missing_module(self): | def skip_test_if_missing_module(self): | ||||
self.skip_if_no_chronik() | self.skip_if_no_chronik() | ||||
def run_test(self): | def run_test(self): | ||||
from test_framework.chronik.client import ChronikClient, pb | from test_framework.chronik.client import ChronikClient, pb | ||||
from test_framework.chronik.test_data import genesis_cb_tx | from test_framework.chronik.test_data import genesis_cb_tx | ||||
node = self.nodes[0] | node = self.nodes[0] | ||||
node.setmocktime(1300000000) | node.setmocktime(1300000000) | ||||
chronik = ChronikClient('127.0.0.1', node.chronik_port) | chronik = ChronikClient("127.0.0.1", node.chronik_port) | ||||
peer = node.add_p2p_connection(P2PDataStore()) | peer = node.add_p2p_connection(P2PDataStore()) | ||||
# Not a valid hash or height | # Not a valid hash or height | ||||
assert_equal(chronik.block_txs('1234f').err(400).msg, | assert_equal( | ||||
'400: Not a hash or height: 1234f') | chronik.block_txs("1234f").err(400).msg, "400: Not a hash or height: 1234f" | ||||
assert_equal(chronik.block_txs('00' * 31).err(400).msg, | ) | ||||
f'400: Not a hash or height: {"00"*31}') | assert_equal( | ||||
assert_equal( | chronik.block_txs("00" * 31).err(400).msg, | ||||
chronik.block_txs('01').err(400).msg, | f'400: Not a hash or height: {"00"*31}', | ||||
'400: Not a hash or height: 01') | ) | ||||
assert_equal(chronik.block_txs('12345678901').err(400).msg, | assert_equal( | ||||
'400: Not a hash or height: 12345678901') | chronik.block_txs("01").err(400).msg, "400: Not a hash or height: 01" | ||||
) | |||||
assert_equal( | assert_equal( | ||||
chronik.block_txs('00' * 32, page=0, page_size=201).err(400).msg, | chronik.block_txs("12345678901").err(400).msg, | ||||
'400: Requested block tx page size 201 is too big, maximum is 200') | "400: Not a hash or height: 12345678901", | ||||
assert_equal( | ) | ||||
chronik.block_txs('00' * 32, page=0, page_size=0).err(400).msg, | |||||
'400: Requested block tx page size 0 is too small, minimum is 1') | assert_equal( | ||||
assert_equal( | chronik.block_txs("00" * 32, page=0, page_size=201).err(400).msg, | ||||
chronik.block_txs('00' * 32, page=0, page_size=2**32).err(400).msg, | "400: Requested block tx page size 201 is too big, maximum is 200", | ||||
'400: Invalid param page_size: 4294967296, ' + | ) | ||||
'number too large to fit in target type') | assert_equal( | ||||
assert_equal( | chronik.block_txs("00" * 32, page=0, page_size=0).err(400).msg, | ||||
chronik.block_txs('00' * 32, page=2**32, page_size=1).err(400).msg, | "400: Requested block tx page size 0 is too small, minimum is 1", | ||||
'400: Invalid param page: 4294967296, ' + | ) | ||||
'number too large to fit in target type') | assert_equal( | ||||
chronik.block_txs("00" * 32, page=0, page_size=2**32).err(400).msg, | |||||
"400: Invalid param page_size: 4294967296, " | |||||
+ "number too large to fit in target type", | |||||
) | |||||
assert_equal( | |||||
chronik.block_txs("00" * 32, page=2**32, page_size=1).err(400).msg, | |||||
"400: Invalid param page: 4294967296, " | |||||
+ "number too large to fit in target type", | |||||
) | |||||
assert_equal( | assert_equal( | ||||
chronik.block_txs(GENESIS_BLOCK_HASH, page=2**32 - 1, page_size=200).ok(), | chronik.block_txs(GENESIS_BLOCK_HASH, page=2**32 - 1, page_size=200).ok(), | ||||
pb.TxHistoryPage(txs=[], num_pages=1, num_txs=1), | pb.TxHistoryPage(txs=[], num_pages=1, num_txs=1), | ||||
) | ) | ||||
assert_equal( | assert_equal( | ||||
chronik.block_txs(GENESIS_BLOCK_HASH).ok(), | chronik.block_txs(GENESIS_BLOCK_HASH).ok(), | ||||
pb.TxHistoryPage( | pb.TxHistoryPage( | ||||
txs=[genesis_cb_tx()], | txs=[genesis_cb_tx()], | ||||
num_pages=1, | num_pages=1, | ||||
num_txs=1, | num_txs=1, | ||||
), | ), | ||||
) | ) | ||||
coinblockhash = self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0] | coinblockhash = self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0] | ||||
coinblock = node.getblock(coinblockhash) | coinblock = node.getblock(coinblockhash) | ||||
cointx = coinblock['tx'][0] | cointx = coinblock["tx"][0] | ||||
tip = self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE)[-1] | tip = self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE)[-1] | ||||
coinvalue = 5000000000 | coinvalue = 5000000000 | ||||
tx1 = CTransaction() | tx1 = CTransaction() | ||||
tx1.vin = [CTxIn(outpoint=COutPoint(int(cointx, 16), 0), | tx1.vin = [ | ||||
scriptSig=SCRIPTSIG_OP_TRUE)] | CTxIn(outpoint=COutPoint(int(cointx, 16), 0), scriptSig=SCRIPTSIG_OP_TRUE) | ||||
] | |||||
tx1.vout = [ | tx1.vout = [ | ||||
CTxOut(coinvalue - 10000, P2SH_OP_TRUE), | CTxOut(coinvalue - 10000, P2SH_OP_TRUE), | ||||
CTxOut(1000, CScript([OP_RETURN, b'test'])), | CTxOut(1000, CScript([OP_RETURN, b"test"])), | ||||
] | ] | ||||
tx1.rehash() | tx1.rehash() | ||||
tx2 = CTransaction() | tx2 = CTransaction() | ||||
tx2.vin = [CTxIn(outpoint=COutPoint(int(tx1.hash, 16), 0), | tx2.vin = [ | ||||
scriptSig=SCRIPTSIG_OP_TRUE)] | CTxIn(outpoint=COutPoint(int(tx1.hash, 16), 0), scriptSig=SCRIPTSIG_OP_TRUE) | ||||
] | |||||
tx2.vout = [ | tx2.vout = [ | ||||
CTxOut(3000, CScript([OP_RETURN, b'test'])), | CTxOut(3000, CScript([OP_RETURN, b"test"])), | ||||
CTxOut(coinvalue - 20000, P2SH_OP_TRUE), | CTxOut(coinvalue - 20000, P2SH_OP_TRUE), | ||||
] | ] | ||||
tx2.rehash() | tx2.rehash() | ||||
tx_coinbase = create_coinbase(102, b'\x03' * 33) | tx_coinbase = create_coinbase(102, b"\x03" * 33) | ||||
block = create_block(int(tip, 16), | block = create_block(int(tip, 16), tx_coinbase, 1300000500) | ||||
tx_coinbase, | |||||
1300000500) | |||||
block.vtx += [tx1, tx2] | block.vtx += [tx1, tx2] | ||||
make_conform_to_ctor(block) | make_conform_to_ctor(block) | ||||
block.hashMerkleRoot = block.calc_merkle_root() | block.hashMerkleRoot = block.calc_merkle_root() | ||||
block.solve() | block.solve() | ||||
peer.send_blocks_and_test([block], node) | peer.send_blocks_and_test([block], node) | ||||
block_metadata = pb.BlockMetadata( | block_metadata = pb.BlockMetadata( | ||||
height=102, | height=102, | ||||
hash=bytes.fromhex(block.hash)[::-1], | hash=bytes.fromhex(block.hash)[::-1], | ||||
timestamp=1300000500, | timestamp=1300000500, | ||||
) | ) | ||||
proto_coinbase_tx = pb.Tx( | proto_coinbase_tx = pb.Tx( | ||||
txid=bytes.fromhex(tx_coinbase.hash)[::-1], | txid=bytes.fromhex(tx_coinbase.hash)[::-1], | ||||
version=1, | version=1, | ||||
inputs=[ | inputs=[ | ||||
pb.TxInput( | pb.TxInput( | ||||
prev_out=pb.OutPoint(txid=bytes(32), out_idx=0xffffffff), | prev_out=pb.OutPoint(txid=bytes(32), out_idx=0xFFFFFFFF), | ||||
input_script=bytes(tx_coinbase.vin[0].scriptSig), | input_script=bytes(tx_coinbase.vin[0].scriptSig), | ||||
sequence_no=0xffffffff, | sequence_no=0xFFFFFFFF, | ||||
), | ), | ||||
], | ], | ||||
outputs=[ | outputs=[ | ||||
pb.TxOutput( | pb.TxOutput( | ||||
value=coinvalue, | value=coinvalue, | ||||
output_script=bytes(tx_coinbase.vout[0].scriptPubKey), | output_script=bytes(tx_coinbase.vout[0].scriptPubKey), | ||||
), | ), | ||||
pb.TxOutput( | pb.TxOutput( | ||||
Show All 24 Lines | def run_test(self): | ||||
output_script=bytes(P2SH_OP_TRUE), | output_script=bytes(P2SH_OP_TRUE), | ||||
spent_by=pb.SpentBy( | spent_by=pb.SpentBy( | ||||
txid=bytes.fromhex(tx2.hash)[::-1], | txid=bytes.fromhex(tx2.hash)[::-1], | ||||
input_idx=0, | input_idx=0, | ||||
), | ), | ||||
), | ), | ||||
pb.TxOutput( | pb.TxOutput( | ||||
value=1000, | value=1000, | ||||
output_script=bytes(CScript([OP_RETURN, b'test'])), | output_script=bytes(CScript([OP_RETURN, b"test"])), | ||||
), | ), | ||||
], | ], | ||||
lock_time=0, | lock_time=0, | ||||
size=len(tx1.serialize()), | size=len(tx1.serialize()), | ||||
block=block_metadata, | block=block_metadata, | ||||
) | ) | ||||
proto_tx2 = pb.Tx( | proto_tx2 = pb.Tx( | ||||
txid=bytes.fromhex(tx2.hash)[::-1], | txid=bytes.fromhex(tx2.hash)[::-1], | ||||
version=1, | version=1, | ||||
inputs=[ | inputs=[ | ||||
pb.TxInput( | pb.TxInput( | ||||
prev_out=pb.OutPoint(txid=bytes.fromhex(tx1.hash)[::-1], out_idx=0), | prev_out=pb.OutPoint(txid=bytes.fromhex(tx1.hash)[::-1], out_idx=0), | ||||
input_script=bytes(SCRIPTSIG_OP_TRUE), | input_script=bytes(SCRIPTSIG_OP_TRUE), | ||||
output_script=bytes(P2SH_OP_TRUE), | output_script=bytes(P2SH_OP_TRUE), | ||||
value=coinvalue - 10000, | value=coinvalue - 10000, | ||||
sequence_no=0, | sequence_no=0, | ||||
), | ), | ||||
], | ], | ||||
outputs=[ | outputs=[ | ||||
pb.TxOutput( | pb.TxOutput( | ||||
value=3000, | value=3000, | ||||
output_script=bytes(CScript([OP_RETURN, b'test'])), | output_script=bytes(CScript([OP_RETURN, b"test"])), | ||||
), | ), | ||||
pb.TxOutput( | pb.TxOutput( | ||||
value=coinvalue - 20000, | value=coinvalue - 20000, | ||||
output_script=bytes(P2SH_OP_TRUE), | output_script=bytes(P2SH_OP_TRUE), | ||||
), | ), | ||||
], | ], | ||||
lock_time=0, | lock_time=0, | ||||
size=len(tx2.serialize()), | size=len(tx2.serialize()), | ||||
block=block_metadata, | block=block_metadata, | ||||
) | ) | ||||
sorted_tx1, sorted_tx2 = sorted( | sorted_tx1, sorted_tx2 = sorted( | ||||
[proto_tx1, proto_tx2], key=lambda tx: tx.txid[::-1]) | [proto_tx1, proto_tx2], key=lambda tx: tx.txid[::-1] | ||||
) | |||||
for page, tx in enumerate([proto_coinbase_tx, sorted_tx1, sorted_tx2]): | for page, tx in enumerate([proto_coinbase_tx, sorted_tx1, sorted_tx2]): | ||||
assert_equal( | assert_equal( | ||||
chronik.block_txs(block.hash, page=page, page_size=1).ok(), | chronik.block_txs(block.hash, page=page, page_size=1).ok(), | ||||
pb.TxHistoryPage( | pb.TxHistoryPage( | ||||
txs=[tx], | txs=[tx], | ||||
num_pages=3, | num_pages=3, | ||||
num_txs=3, | num_txs=3, | ||||
Show All 25 Lines | def run_test(self): | ||||
num_txs=3, | num_txs=3, | ||||
), | ), | ||||
) | ) | ||||
node.invalidateblock(block.hash) | node.invalidateblock(block.hash) | ||||
chronik.block_txs(block.hash).err(404) | chronik.block_txs(block.hash).err(404) | ||||
if __name__ == '__main__': | if __name__ == "__main__": | ||||
ChronikBlockTxsTest().main() | ChronikBlockTxsTest().main() |