Page MenuHomePhabricator
Differential D13878 Diff 40252 test/functional/chronik_script_confirmed_txs.py

Changeset View

Standalone View

View Options

test/functional/chronik_script_confirmed_txs.py

Show All 17 Lines
from test_framework.txtools import pad_tx from test_framework.txtools import pad_tx
from test_framework.util import assert_equal, iter_chunks from test_framework.util import assert_equal, iter_chunks
class ChronikScriptConfirmedTxsTest(BitcoinTestFramework): class ChronikScriptConfirmedTxsTest(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]
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())
mocktime = 1300000000 mocktime = 1300000000
node.setmocktime(mocktime) node.setmocktime(mocktime)
assert_equal( assert_equal(
chronik.script('', '').confirmed_txs().err(400).msg, chronik.script("", "").confirmed_txs().err(400).msg,
'400: Unknown script type: ') "400: Unknown script type: ",
)
assert_equal(
chronik.script("foo", "").confirmed_txs().err(400).msg,
"400: Unknown script type: foo",
)
assert_equal(
chronik.script("p2pkh", "LILALI").confirmed_txs().err(400).msg,
"400: Invalid hex: Invalid character 'L' at position 0",
)
assert_equal( assert_equal(
chronik.script('foo', '').confirmed_txs().err(400).msg, chronik.script("other", "LILALI").confirmed_txs().err(400).msg,
'400: Unknown script type: foo') "400: Invalid hex: Invalid character 'L' at position 0",
)
assert_equal( assert_equal(
chronik.script('p2pkh', 'LILALI').confirmed_txs().err(400).msg, chronik.script("p2pkh", "").confirmed_txs().err(400).msg,
"400: Invalid hex: Invalid character 'L' at position 0") "400: Invalid payload for P2PKH: Invalid length, "
+ "expected 20 bytes but got 0 bytes",
)
assert_equal( assert_equal(
chronik.script('other', 'LILALI').confirmed_txs().err(400).msg, chronik.script("p2pkh", "aA").confirmed_txs().err(400).msg,
"400: Invalid hex: Invalid character 'L' at position 0") "400: Invalid payload for P2PKH: Invalid length, "
assert_equal( + "expected 20 bytes but got 1 bytes",
chronik.script('p2pkh', '').confirmed_txs().err(400).msg, )
'400: Invalid payload for P2PKH: Invalid length, ' + assert_equal(
'expected 20 bytes but got 0 bytes') chronik.script("p2sh", "aaBB").confirmed_txs().err(400).msg,
assert_equal( "400: Invalid payload for P2SH: Invalid length, "
chronik.script('p2pkh', 'aA').confirmed_txs().err(400).msg, + "expected 20 bytes but got 2 bytes",
'400: Invalid payload for P2PKH: Invalid length, ' + )
'expected 20 bytes but got 1 bytes') assert_equal(
assert_equal( chronik.script("p2pk", "aaBBcc").confirmed_txs().err(400).msg,
chronik.script('p2sh', 'aaBB').confirmed_txs().err(400).msg, "400: Invalid payload for P2PK: Invalid length, "
'400: Invalid payload for P2SH: Invalid length, ' + + "expected one of [33, 65] but got 3 bytes",
'expected 20 bytes but got 2 bytes') )
assert_equal(
chronik.script('p2pk', 'aaBBcc').confirmed_txs().err(400).msg, assert_equal(
'400: Invalid payload for P2PK: Invalid length, ' + chronik.script("p2pk", GENESIS_CB_PK)
'expected one of [33, 65] but got 3 bytes') .confirmed_txs(page=0, page_size=201)
.err(400)
assert_equal( .msg,
chronik.script( "400: Requested page size 201 is too big, maximum is 200",
'p2pk', GENESIS_CB_PK).confirmed_txs( )
page=0, page_size=201).err(400).msg, assert_equal(
'400: Requested page size 201 is too big, maximum is 200') chronik.script("p2pk", GENESIS_CB_PK)
assert_equal( .confirmed_txs(page=0, page_size=0)
chronik.script( .err(400)
'p2pk', GENESIS_CB_PK).confirmed_txs( .msg,
page=0, page_size=0).err(400).msg, "400: Requested page size 0 is too small, minimum is 1",
'400: Requested page size 0 is too small, minimum is 1') )
assert_equal( assert_equal(
chronik.script( chronik.script("p2pk", GENESIS_CB_PK)
'p2pk', GENESIS_CB_PK).confirmed_txs( .confirmed_txs(page=0, page_size=2**32)
page=0, page_size=2**32).err(400).msg, .err(400)
'400: Invalid param page_size: 4294967296, ' + .msg,
'number too large to fit in target type') "400: Invalid param page_size: 4294967296, "
assert_equal( + "number too large to fit in target type",
chronik.script( )
'p2pk', GENESIS_CB_PK).confirmed_txs( assert_equal(
page=2**32, page_size=1).err(400).msg, chronik.script("p2pk", GENESIS_CB_PK)
'400: Invalid param page: 4294967296, ' + .confirmed_txs(page=2**32, page_size=1)
'number too large to fit in target type') .err(400)
.msg,
"400: Invalid param page: 4294967296, "
+ "number too large to fit in target type",
)
# Handle overflow gracefully on 32-bit # Handle overflow gracefully on 32-bit
assert_equal( assert_equal(
chronik.script('p2pk', GENESIS_CB_PK) chronik.script("p2pk", GENESIS_CB_PK)
.confirmed_txs(page=2**32 - 1, page_size=200) .confirmed_txs(page=2**32 - 1, page_size=200)
.ok(), .ok(),
pb.TxHistoryPage(num_pages=1, num_txs=1)) pb.TxHistoryPage(num_pages=1, num_txs=1),
)
genesis_db_script_history = chronik.script( genesis_db_script_history = (
'p2pk', GENESIS_CB_PK).confirmed_txs().ok() chronik.script("p2pk", GENESIS_CB_PK).confirmed_txs().ok()
assert_equal(genesis_db_script_history, )
pb.TxHistoryPage(txs=[genesis_cb_tx()], assert_equal(
num_pages=1, genesis_db_script_history,
num_txs=1)) pb.TxHistoryPage(txs=[genesis_cb_tx()], num_pages=1, num_txs=1),
)
script_type = 'p2sh' script_type = "p2sh"
payload_hex = P2SH_OP_TRUE[2:-1].hex() payload_hex = P2SH_OP_TRUE[2:-1].hex()
# Generate 101 blocks to some address and verify pages # Generate 101 blocks to some address and verify pages
blockhashes = self.generatetoaddress(node, 101, ADDRESS_ECREG_P2SH_OP_TRUE) blockhashes = self.generatetoaddress(node, 101, ADDRESS_ECREG_P2SH_OP_TRUE)
def check_confirmed_txs(txs, *, page_size=25): def check_confirmed_txs(txs, *, page_size=25):
pages = list(iter_chunks(txs, page_size)) pages = list(iter_chunks(txs, page_size))
for page_num, page_txs in enumerate(pages): for page_num, page_txs in enumerate(pages):
script_history = chronik.script( script_history = (
script_type, payload_hex).confirmed_txs( chronik.script(script_type, payload_hex)
page_num, page_size).ok() .confirmed_txs(page_num, page_size)
.ok()
)
for tx_idx, entry in enumerate(page_txs): for tx_idx, entry in enumerate(page_txs):
script_tx = script_history.txs[tx_idx] script_tx = script_history.txs[tx_idx]
if 'txid' in entry: if "txid" in entry:
assert_equal(script_tx.txid[::-1].hex(), entry['txid']) assert_equal(script_tx.txid[::-1].hex(), entry["txid"])
if 'block' in entry: if "block" in entry:
block_height, block_hash = entry['block'] block_height, block_hash = entry["block"]
assert_equal(script_tx.block, pb.BlockMetadata( assert_equal(
script_tx.block,
pb.BlockMetadata(
hash=bytes.fromhex(block_hash)[::-1], hash=bytes.fromhex(block_hash)[::-1],
height=block_height, height=block_height,
timestamp=script_tx.block.timestamp, timestamp=script_tx.block.timestamp,
)) ),
)
txs = [{'block': (i + 1, blockhash)} for i, blockhash in enumerate(blockhashes)] txs = [{"block": (i + 1, blockhash)} for i, blockhash in enumerate(blockhashes)]
check_confirmed_txs(txs) check_confirmed_txs(txs)
check_confirmed_txs(txs, page_size=200) check_confirmed_txs(txs, page_size=200)
# Undo last block & check history # Undo last block & check history
node.invalidateblock(blockhashes[-1]) node.invalidateblock(blockhashes[-1])
check_confirmed_txs(txs[:-1]) check_confirmed_txs(txs[:-1])
check_confirmed_txs(txs[:-1], page_size=200) check_confirmed_txs(txs[:-1], page_size=200)
# Create 1 block manually # Create 1 block manually
coinbase_tx = create_coinbase(101) coinbase_tx = create_coinbase(101)
coinbase_tx.vout[0].scriptPubKey = P2SH_OP_TRUE coinbase_tx.vout[0].scriptPubKey = P2SH_OP_TRUE
coinbase_tx.rehash() coinbase_tx.rehash()
block = create_block(int(blockhashes[-2], 16), block = create_block(int(blockhashes[-2], 16), coinbase_tx, mocktime + 1000)
coinbase_tx,
mocktime + 1000)
block.solve() block.solve()
peer.send_blocks_and_test([block], node) peer.send_blocks_and_test([block], node)
blockhashes[-1] = block.hash blockhashes[-1] = block.hash
txs = [{'block': (i + 1, blockhash)} for i, blockhash in enumerate(blockhashes)] txs = [{"block": (i + 1, blockhash)} for i, blockhash in enumerate(blockhashes)]
check_confirmed_txs(txs) check_confirmed_txs(txs)
check_confirmed_txs(txs, page_size=200) check_confirmed_txs(txs, page_size=200)
# Generate 900 more blocks and verify # Generate 900 more blocks and verify
# Total of 1001 txs for this script (a page in the DB is 1000 entries long) # Total of 1001 txs for this script (a page in the DB is 1000 entries long)
blockhashes += self.generatetoaddress(node, 900, ADDRESS_ECREG_P2SH_OP_TRUE) blockhashes += self.generatetoaddress(node, 900, ADDRESS_ECREG_P2SH_OP_TRUE)
txs = [{'block': (i + 1, blockhash)} for i, blockhash in enumerate(blockhashes)] txs = [{"block": (i + 1, blockhash)} for i, blockhash in enumerate(blockhashes)]
page_sizes = [1, 5, 7, 25, 111, 200] page_sizes = [1, 5, 7, 25, 111, 200]
for page_size in page_sizes: for page_size in page_sizes:
check_confirmed_txs(txs, page_size=page_size) check_confirmed_txs(txs, page_size=page_size)
coinvalue = 5000000000 coinvalue = 5000000000
cointxids = [] cointxids = []
for coinblockhash in blockhashes[:10]: for coinblockhash in blockhashes[:10]:
coinblock = node.getblock(coinblockhash) coinblock = node.getblock(coinblockhash)
cointxids.append(coinblock['tx'][0]) cointxids.append(coinblock["tx"][0])
mempool_txids = [] mempool_txids = []
for cointxid in cointxids: for cointxid in cointxids:
tx = CTransaction() tx = CTransaction()
tx.nVersion = 1 tx.nVersion = 1
tx.vin = [CTxIn(outpoint=COutPoint(int(cointxid, 16), 0), tx.vin = [
scriptSig=SCRIPTSIG_OP_TRUE)] CTxIn(
outpoint=COutPoint(int(cointxid, 16), 0),
scriptSig=SCRIPTSIG_OP_TRUE,
)
]
tx.vout = [CTxOut(coinvalue - 1000, P2SH_OP_TRUE)] tx.vout = [CTxOut(coinvalue - 1000, P2SH_OP_TRUE)]
pad_tx(tx) pad_tx(tx)
txid = node.sendrawtransaction(tx.serialize().hex()) txid = node.sendrawtransaction(tx.serialize().hex())
mempool_txids.append(txid) mempool_txids.append(txid)
# confirmed-txs completely unaffected by mempool txs # confirmed-txs completely unaffected by mempool txs
for page_size in page_sizes: for page_size in page_sizes:
check_confirmed_txs(txs, page_size=page_size) check_confirmed_txs(txs, page_size=page_size)
# Mine mempool txs, now they're in confirmed-txs # Mine mempool txs, now they're in confirmed-txs
newblockhash = self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0] newblockhash = self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0]
txs.append({'block': (1002, newblockhash)}) txs.append({"block": (1002, newblockhash)})
txs += [{'block': (1002, newblockhash), 'txid': txid} txs += [
for txid in sorted(mempool_txids)] {"block": (1002, newblockhash), "txid": txid}
for txid in sorted(mempool_txids)
]
for page_size in page_sizes: for page_size in page_sizes:
check_confirmed_txs(txs, page_size=page_size) check_confirmed_txs(txs, page_size=page_size)
if __name__ == '__main__': if __name__ == "__main__":
ChronikScriptConfirmedTxsTest().main() ChronikScriptConfirmedTxsTest().main()