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

Changeset View

Standalone View

View Options

test/functional/chronik_script_utxos.py

Show All 23 Lines
from test_framework.txtools import pad_tx from test_framework.txtools import pad_tx
from test_framework.util import assert_equal from test_framework.util import assert_equal
class ChronikScriptUtxosTest(BitcoinTestFramework): class ChronikScriptUtxosTest(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
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())
assert_equal( assert_equal(
chronik.script('', '').utxos().err(400).msg, chronik.script("", "").utxos().err(400).msg, "400: Unknown script type: "
'400: Unknown script type: ') )
assert_equal( assert_equal(
chronik.script('foo', '').utxos().err(400).msg, chronik.script("foo", "").utxos().err(400).msg,
'400: Unknown script type: foo') "400: Unknown script type: foo",
)
assert_equal( assert_equal(
chronik.script('p2pkh', 'LILALI').utxos().err(400).msg, chronik.script("p2pkh", "LILALI").utxos().err(400).msg,
"400: Invalid hex: Invalid character 'L' at position 0") "400: Invalid hex: Invalid character 'L' at position 0",
)
assert_equal( assert_equal(
chronik.script('other', 'LILALI').utxos().err(400).msg, chronik.script("other", "LILALI").utxos().err(400).msg,
"400: Invalid hex: Invalid character 'L' at position 0") "400: Invalid hex: Invalid character 'L' at position 0",
)
assert_equal( assert_equal(
chronik.script('p2pkh', '').utxos().err(400).msg, chronik.script("p2pkh", "").utxos().err(400).msg,
'400: Invalid payload for P2PKH: Invalid length, ' + "400: Invalid payload for P2PKH: Invalid length, "
'expected 20 bytes but got 0 bytes') + "expected 20 bytes but got 0 bytes",
)
assert_equal( assert_equal(
chronik.script('p2pkh', 'aA').utxos().err(400).msg, chronik.script("p2pkh", "aA").utxos().err(400).msg,
'400: Invalid payload for P2PKH: Invalid length, ' + "400: Invalid payload for P2PKH: Invalid length, "
'expected 20 bytes but got 1 bytes') + "expected 20 bytes but got 1 bytes",
)
assert_equal( assert_equal(
chronik.script('p2sh', 'aaBB').utxos().err(400).msg, chronik.script("p2sh", "aaBB").utxos().err(400).msg,
'400: Invalid payload for P2SH: Invalid length, ' + "400: Invalid payload for P2SH: Invalid length, "
'expected 20 bytes but got 2 bytes') + "expected 20 bytes but got 2 bytes",
)
assert_equal( assert_equal(
chronik.script('p2pk', 'aaBBcc').utxos().err(400).msg, chronik.script("p2pk", "aaBBcc").utxos().err(400).msg,
'400: Invalid payload for P2PK: Invalid length, ' + "400: Invalid payload for P2PK: Invalid length, "
'expected one of [33, 65] but got 3 bytes') + "expected one of [33, 65] but got 3 bytes",
)
# Test Genesis pubkey UTXO # Test Genesis pubkey UTXO
coinvalue = 5000000000 coinvalue = 5000000000
assert_equal(chronik.script('p2pk', GENESIS_CB_PK).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes.fromhex(f'41{GENESIS_CB_PK}ac'), chronik.script("p2pk", GENESIS_CB_PK).utxos().ok(),
utxos=[pb.ScriptUtxo( pb.ScriptUtxos(
script=bytes.fromhex(f"41{GENESIS_CB_PK}ac"),
utxos=[
pb.ScriptUtxo(
outpoint=pb.OutPoint( outpoint=pb.OutPoint(
txid=bytes.fromhex(GENESIS_CB_TXID)[::-1], txid=bytes.fromhex(GENESIS_CB_TXID)[::-1],
out_idx=0, out_idx=0,
), ),
block_height=0, block_height=0,
is_coinbase=True, is_coinbase=True,
value=coinvalue, value=coinvalue,
is_final=False, is_final=False,
)])) )
],
),
)
script_type = 'p2sh' script_type = "p2sh"
payload_hex = P2SH_OP_TRUE[2:-1].hex() payload_hex = P2SH_OP_TRUE[2:-1].hex()
# Generate us a coin, creates a UTXO # Generate us a coin, creates a UTXO
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]
assert_equal(chronik.script(script_type, payload_hex).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), chronik.script(script_type, payload_hex).utxos().ok(),
utxos=[pb.ScriptUtxo( pb.ScriptUtxos(
script=bytes(P2SH_OP_TRUE),
utxos=[
pb.ScriptUtxo(
outpoint=pb.OutPoint( outpoint=pb.OutPoint(
txid=bytes.fromhex(cointx)[::-1], txid=bytes.fromhex(cointx)[::-1],
out_idx=0, out_idx=0,
), ),
block_height=1, block_height=1,
is_coinbase=True, is_coinbase=True,
value=coinvalue, value=coinvalue,
is_final=False, is_final=False,
)])) )
],
),
)
self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE) self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE)
# Make tx creating 4 UTXOs, spending the coinbase UTXO # Make tx creating 4 UTXOs, spending the coinbase UTXO
send_values = [coinvalue - 10000, 1000, 2000, 3000] send_values = [coinvalue - 10000, 1000, 2000, 3000]
tx = CTransaction() tx = CTransaction()
tx.vin = [CTxIn(outpoint=COutPoint(int(cointx, 16), 0), tx.vin = [
scriptSig=SCRIPTSIG_OP_TRUE)] CTxIn(outpoint=COutPoint(int(cointx, 16), 0), scriptSig=SCRIPTSIG_OP_TRUE)
]
tx.vout = [CTxOut(value, P2SH_OP_TRUE) for value in send_values] tx.vout = [CTxOut(value, P2SH_OP_TRUE) for value in send_values]
txid = node.sendrawtransaction(tx.serialize().hex()) txid = node.sendrawtransaction(tx.serialize().hex())
expected_utxos = [ expected_utxos = [
pb.ScriptUtxo( pb.ScriptUtxo(
outpoint=pb.OutPoint( outpoint=pb.OutPoint(
txid=bytes.fromhex(txid)[::-1], txid=bytes.fromhex(txid)[::-1],
out_idx=i, out_idx=i,
), ),
block_height=-1, block_height=-1,
is_coinbase=False, is_coinbase=False,
value=value, value=value,
is_final=False, is_final=False,
) )
for i, value in enumerate(send_values) for i, value in enumerate(send_values)
] ]
assert_equal(chronik.script(script_type, payload_hex).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), chronik.script(script_type, payload_hex).utxos().ok(),
utxos=expected_utxos)) pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), utxos=expected_utxos),
)
# Mine tx, which adds the blockheight to the UTXO # Mine tx, which adds the blockheight to the UTXO
tip = self.generatetoaddress(node, 1, ADDRESS_ECREG_UNSPENDABLE)[-1] tip = self.generatetoaddress(node, 1, ADDRESS_ECREG_UNSPENDABLE)[-1]
for expected_utxo in expected_utxos: for expected_utxo in expected_utxos:
expected_utxo.block_height = 102 expected_utxo.block_height = 102
assert_equal(chronik.script(script_type, payload_hex).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), chronik.script(script_type, payload_hex).utxos().ok(),
utxos=expected_utxos)) pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), utxos=expected_utxos),
)
# Make tx spending the 3rd UTXO, and creating 1 UTXO # Make tx spending the 3rd UTXO, and creating 1 UTXO
tx2 = CTransaction() tx2 = CTransaction()
tx2.vin = [CTxIn(outpoint=COutPoint(int(txid, 16), 3), tx2.vin = [
scriptSig=SCRIPTSIG_OP_TRUE)] CTxIn(outpoint=COutPoint(int(txid, 16), 3), scriptSig=SCRIPTSIG_OP_TRUE)
]
tx2.vout = [CTxOut(2500, P2SH_OP_TRUE)] tx2.vout = [CTxOut(2500, P2SH_OP_TRUE)]
pad_tx(tx2) pad_tx(tx2)
txid2 = node.sendrawtransaction(tx2.serialize().hex()) txid2 = node.sendrawtransaction(tx2.serialize().hex())
del expected_utxos[3] del expected_utxos[3]
expected_utxos.append(pb.ScriptUtxo( expected_utxos.append(
pb.ScriptUtxo(
outpoint=pb.OutPoint( outpoint=pb.OutPoint(
txid=bytes.fromhex(txid2)[::-1], txid=bytes.fromhex(txid2)[::-1],
out_idx=0, out_idx=0,
), ),
block_height=-1, block_height=-1,
is_coinbase=False, is_coinbase=False,
value=2500, value=2500,
is_final=False, is_final=False,
)) )
)
assert_equal(chronik.script(script_type, payload_hex).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), chronik.script(script_type, payload_hex).utxos().ok(),
utxos=expected_utxos)) pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), utxos=expected_utxos),
)
# Make tx spending a DB UTXO and a mempool UTXO # Make tx spending a DB UTXO and a mempool UTXO
tx3 = CTransaction() tx3 = CTransaction()
tx3.vin = [CTxIn(outpoint=COutPoint(int(txid, 16), 2), tx3.vin = [
scriptSig=SCRIPTSIG_OP_TRUE), CTxIn(outpoint=COutPoint(int(txid, 16), 2), scriptSig=SCRIPTSIG_OP_TRUE),
CTxIn(outpoint=COutPoint(int(txid2, 16), 0), CTxIn(outpoint=COutPoint(int(txid2, 16), 0), scriptSig=SCRIPTSIG_OP_TRUE),
scriptSig=SCRIPTSIG_OP_TRUE)] ]
pad_tx(tx3) pad_tx(tx3)
node.sendrawtransaction(tx3.serialize().hex()) node.sendrawtransaction(tx3.serialize().hex())
assert_equal(chronik.script(script_type, payload_hex).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), chronik.script(script_type, payload_hex).utxos().ok(),
utxos=expected_utxos[:2])) pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), utxos=expected_utxos[:2]),
)
# Make a tx which conflicts with tx3, by spending the same DB UTXO # Make a tx which conflicts with tx3, by spending the same DB UTXO
tx3_conflict = CTransaction() tx3_conflict = CTransaction()
tx3_conflict.vin = [CTxIn(outpoint=COutPoint(int(txid, 16), 2), tx3_conflict.vin = [
scriptSig=SCRIPTSIG_OP_TRUE)] CTxIn(outpoint=COutPoint(int(txid, 16), 2), scriptSig=SCRIPTSIG_OP_TRUE)
]
pad_tx(tx3_conflict) pad_tx(tx3_conflict)
# Mining conflicting tx returns the mempool UTXO spent by tx3 to the mempool # Mining conflicting tx returns the mempool UTXO spent by tx3 to the mempool
block = create_block(int(tip, 16), block = create_block(
create_coinbase(103, b'\x03' * 33), int(tip, 16), create_coinbase(103, b"\x03" * 33), 1300000500
1300000500) )
block.vtx += [tx3_conflict] block.vtx += [tx3_conflict]
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)
del expected_utxos[2] del expected_utxos[2]
assert_equal(chronik.script(script_type, payload_hex).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), chronik.script(script_type, payload_hex).utxos().ok(),
utxos=expected_utxos)) pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), utxos=expected_utxos),
)
# Invalidating the last block doesn't change UTXOs # Invalidating the last block doesn't change UTXOs
node.invalidateblock(block.hash) node.invalidateblock(block.hash)
assert_equal(chronik.script(script_type, payload_hex).utxos().ok(), assert_equal(
pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), chronik.script(script_type, payload_hex).utxos().ok(),
utxos=expected_utxos)) pb.ScriptUtxos(script=bytes(P2SH_OP_TRUE), utxos=expected_utxos),
)
# Invalidating the next last block returns all UTXOs back to the mempool # Invalidating the next last block returns all UTXOs back to the mempool
node.invalidateblock(tip) node.invalidateblock(tip)
for expected_utxo in expected_utxos: for expected_utxo in expected_utxos:
expected_utxo.block_height = -1 expected_utxo.block_height = -1
# Mempool UTXOs are sorted by txid:out_idx. Note: `sorted` is stable. # Mempool UTXOs are sorted by txid:out_idx. Note: `sorted` is stable.
assert_equal(list(chronik.script(script_type, payload_hex).utxos().ok().utxos), assert_equal(
sorted(expected_utxos, key=lambda utxo: utxo.outpoint.txid[::-1])) list(chronik.script(script_type, payload_hex).utxos().ok().utxos),
sorted(expected_utxos, key=lambda utxo: utxo.outpoint.txid[::-1]),
)
if __name__ == '__main__': if __name__ == "__main__":
ChronikScriptUtxosTest().main() ChronikScriptUtxosTest().main()