Page MenuHomePhabricator

No OneTemporary

diff --git a/test/functional/setup_scripts/chronik-client_block_and_blocks.py b/test/functional/setup_scripts/chronik-client_block_and_blocks.py
index d66e42a7a..9af6e6113 100644
--- a/test/functional/setup_scripts/chronik-client_block_and_blocks.py
+++ b/test/functional/setup_scripts/chronik-client_block_and_blocks.py
@@ -1,59 +1,58 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client js library block() and blocks() functions
"""
import pathmagic # noqa
from setup_framework import SetupFramework
from test_framework.util import assert_equal
class ChronikClient_Block_Setup(SetupFramework):
def set_test_params(self):
self.num_nodes = 1
self.extra_args = [["-chronik"]]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
def run_test(self):
# Init
node = self.nodes[0]
yield True
self.log.info("Step 1: Initialized regtest chain")
assert_equal(node.getblockcount(), 200)
yield True
self.log.info("Step 2: Mine 10 more blocks")
tip = self.generate(node, 10)[-1]
assert_equal(node.getblockcount(), 210)
yield True
self.log.info("Step 3: Park the last block")
node.parkblock(node.getbestblockhash())
assert_equal(node.getblockcount(), 209)
yield True
self.log.info("Step 4: Unpark the last block")
node.unparkblock(tip)
assert_equal(node.getblockcount(), 210)
yield True
self.log.info("Step 5: invalidate the last block")
node.invalidateblock(node.getbestblockhash())
assert_equal(node.getblockcount(), 209)
yield True
self.log.info("Step 6: Reconsider the last block")
node.reconsiderblock(tip)
assert_equal(node.getblockcount(), 210)
yield True
if __name__ == "__main__":
ChronikClient_Block_Setup().main()
diff --git a/test/functional/setup_scripts/chronik-client_blockchain_info.py b/test/functional/setup_scripts/chronik-client_blockchain_info.py
index 1a161e329..4e7cf7c67 100644
--- a/test/functional/setup_scripts/chronik-client_blockchain_info.py
+++ b/test/functional/setup_scripts/chronik-client_blockchain_info.py
@@ -1,54 +1,53 @@
# Copyright (c) 2023 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client js library blockchaininfo() function
"""
import pathmagic # noqa
from setup_framework import SetupFramework
from test_framework.util import assert_equal
class ChronikClient_BlockChainInfo_Setup(SetupFramework):
def set_test_params(self):
self.num_nodes = 1
self.extra_args = [["-chronik"]]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
def run_test(self):
# Init
node = self.nodes[0]
yield True
self.log.info("Step 1: Initialized regtest chain")
assert_equal(node.getblockcount(), 200)
yield True
self.log.info("Step 2: Mine 10 more blocks")
self.generate(node, 10)
assert_equal(node.getblockcount(), 210)
yield True
self.log.info("Step 3: Mine again 10 more blocks")
tip = self.generate(node, 10)[-1]
assert_equal(node.getblockcount(), 220)
yield True
self.log.info("Step 4: Park the last block")
node.parkblock(node.getbestblockhash())
assert_equal(node.getblockcount(), 219)
yield True
self.log.info("Step 5: Unpark the last block")
node.unparkblock(tip)
assert_equal(node.getblockcount(), 220)
yield True
if __name__ == "__main__":
ChronikClient_BlockChainInfo_Setup().main()
diff --git a/test/functional/setup_scripts/chronik-client_blocktxs_and_tx_and_rawtx.py b/test/functional/setup_scripts/chronik-client_blocktxs_and_tx_and_rawtx.py
index fd6aeb532..b4a6adce8 100644
--- a/test/functional/setup_scripts/chronik-client_blocktxs_and_tx_and_rawtx.py
+++ b/test/functional/setup_scripts/chronik-client_blocktxs_and_tx_and_rawtx.py
@@ -1,57 +1,56 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client js library blockTxs(), tx(), and rawTx() functions
"""
import pathmagic # noqa
from ipc import send_ipc_message
from setup_framework import SetupFramework
from test_framework.util import assert_equal
from test_framework.wallet import MiniWallet
class ChronikClient_Block_Setup(SetupFramework):
def set_test_params(self):
self.num_nodes = 1
self.extra_args = [["-chronik"]]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
def run_test(self):
# Init
node = self.nodes[0]
wallet = MiniWallet(node)
yield True
self.log.info("Step 1: Initialized regtest chain")
assert_equal(node.getblockcount(), 200)
yield True
self.log.info("Step 2: Broadcast ten txs")
txs_and_rawtxs = {}
for x in range(10):
# Make the fee rate vary to have txs with varying amounts
tx = wallet.send_self_transfer(from_node=node, fee_rate=(x + 1) * 1000)
txs_and_rawtxs[tx["txid"]] = tx["hex"]
send_ipc_message({"txs_and_rawtxs": txs_and_rawtxs})
assert_equal(node.getblockcount(), 200)
yield True
self.log.info("Step 3: Mine a block with these txs")
self.generate(node, 1)
assert_equal(node.getblockcount(), 201)
yield True
self.log.info("Step 4: Park the last block")
node.parkblock(node.getbestblockhash())
assert_equal(node.getblockcount(), 200)
yield True
if __name__ == "__main__":
ChronikClient_Block_Setup().main()
diff --git a/test/functional/setup_scripts/chronik-client_broadcast_txs.py b/test/functional/setup_scripts/chronik-client_broadcast_txs.py
index 7277df978..3110936b4 100644
--- a/test/functional/setup_scripts/chronik-client_broadcast_txs.py
+++ b/test/functional/setup_scripts/chronik-client_broadcast_txs.py
@@ -1,135 +1,134 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client broadcastTx and broadcastTxs methods
Based on test/functional/chronik_token_broadcast_txs.py
"""
import pathmagic # noqa
from ipc import send_ipc_message
from setup_framework import SetupFramework
from test_framework.address import (
ADDRESS_ECREG_P2SH_OP_TRUE,
ADDRESS_ECREG_UNSPENDABLE,
P2SH_OP_TRUE,
SCRIPTSIG_OP_TRUE,
)
from test_framework.chronik.alp import alp_genesis, alp_opreturn, alp_send
from test_framework.messages import COutPoint, CTransaction, CTxIn, CTxOut
from test_framework.util import assert_equal
class ChronikClient_Broadcast_Setup(SetupFramework):
def set_test_params(self):
self.num_nodes = 1
self.extra_args = [["-chronik"]]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
def run_test(self):
# Init
node = self.nodes[0]
yield True
self.log.info(
"Step 1: Initialized regtest chain. Build some raw txs for chronik to broadcast"
)
# ALP txs
coinblockhash = self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0]
coinblock = node.getblock(coinblockhash)
cointx = coinblock["tx"][0]
self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE)
coinvalue = 2500000000
alp_genesis_tx = CTransaction()
alp_genesis_tx.vin = [CTxIn(COutPoint(int(cointx, 16), 0), SCRIPTSIG_OP_TRUE)]
alp_genesis_tx.vout = [
alp_opreturn(
alp_genesis(
mint_amounts=[1000, 2000, 3000, 4000, 5000, 6000],
num_batons=0,
),
),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(coinvalue - 100000, P2SH_OP_TRUE),
]
alp_genesis_tx.rehash()
alp_genesis_rawtx = alp_genesis_tx.serialize().hex()
alp_genesis_txid = self.nodes[0].decoderawtransaction(alp_genesis_rawtx)["txid"]
send_ipc_message({"alp_genesis_rawtx": alp_genesis_rawtx})
send_ipc_message({"alp_genesis_txid": alp_genesis_txid})
# An ok ALP tx to test the broadcastTxs method
ok_tx = CTransaction()
ok_tx.vin = [CTxIn(COutPoint(int(alp_genesis_txid, 16), 1), SCRIPTSIG_OP_TRUE)]
ok_tx.vout = [
alp_opreturn(alp_send(alp_genesis_txid, [1000])),
CTxOut(546, P2SH_OP_TRUE),
]
ok_tx.rehash()
ok_rawtx = ok_tx.serialize().hex()
ok_txid = self.nodes[0].decoderawtransaction(ok_rawtx)["txid"]
send_ipc_message({"ok_rawtx": ok_rawtx})
send_ipc_message({"ok_txid": ok_txid})
# ALP burn tx
burn_tx = CTransaction()
burn_tx.vin = [
CTxIn(COutPoint(int(alp_genesis_txid, 16), 2), SCRIPTSIG_OP_TRUE)
]
burn_tx.vout = [
alp_opreturn(alp_send(alp_genesis_txid, [1999])),
CTxOut(546, P2SH_OP_TRUE),
]
burn_tx.rehash()
alp_burn_rawtx = burn_tx.serialize().hex()
alp_burn_txid = self.nodes[0].decoderawtransaction(alp_burn_rawtx)["txid"]
send_ipc_message({"alp_burn_rawtx": alp_burn_rawtx})
send_ipc_message({"alp_burn_txid": alp_burn_txid})
# ALP burn 2 tx
burn2_tx = CTransaction()
burn2_tx.vin = [
CTxIn(COutPoint(int(alp_genesis_txid, 16), 3), SCRIPTSIG_OP_TRUE)
]
burn2_tx.vout = [
alp_opreturn(alp_send(alp_genesis_txid, [3001])),
CTxOut(546, P2SH_OP_TRUE),
]
burn2_tx.rehash()
alp_burn_2_rawtx = burn2_tx.serialize().hex()
alp_burn_2_txid = self.nodes[0].decoderawtransaction(alp_burn_2_rawtx)["txid"]
send_ipc_message({"alp_burn_2_rawtx": alp_burn_2_rawtx})
send_ipc_message({"alp_burn_2_txid": alp_burn_2_txid})
block_height_before_mining = node.getblockcount()
assert_equal(node.getblockcount(), 301)
yield True
self.log.info("Step 2: Mine a block with these txs")
self.generate(node, 1)
assert_equal(node.getblockcount(), block_height_before_mining + 1)
yield True
if __name__ == "__main__":
ChronikClient_Broadcast_Setup().main()
diff --git a/test/functional/setup_scripts/chronik-client_mempool_conflicts.py b/test/functional/setup_scripts/chronik-client_mempool_conflicts.py
index a1733c714..6e75301c6 100644
--- a/test/functional/setup_scripts/chronik-client_mempool_conflicts.py
+++ b/test/functional/setup_scripts/chronik-client_mempool_conflicts.py
@@ -1,114 +1,113 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client js library script endpoints
"""
import pathmagic # noqa
from ipc import send_ipc_message
from setup_framework import SetupFramework
from test_framework.address import (
ADDRESS_ECREG_P2SH_OP_TRUE,
ADDRESS_ECREG_UNSPENDABLE,
P2SH_OP_TRUE,
SCRIPTSIG_OP_TRUE,
)
from test_framework.blocktools import create_block, create_coinbase
from test_framework.messages import COutPoint, CTransaction, CTxIn, CTxOut
from test_framework.p2p import P2PDataStore
from test_framework.txtools import pad_tx
from test_framework.util import assert_equal
# Modeled on chronik_mempool_conflicts.py
class ChronikClient_Mempool_Conflict_Setup(SetupFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.num_nodes = 1
self.extra_args = [["-chronik"]]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
def run_test(self):
# Init
node = self.nodes[0]
peer = node.add_p2p_connection(P2PDataStore())
yield True
self.log.info("Step 1: New clean chain. Send subscription script to jest")
assert_equal(node.getblockcount(), 0)
yield True
self.log.info("Step 2: Broadcast some txs")
# Send some txs
coinblockhash = self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0]
coinblock = node.getblock(coinblockhash)
cointx = coinblock["tx"][0]
send_ipc_message({"cointx": cointx})
block_hashes = self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE)
coinvalue = 5000000000
tx1 = CTransaction()
tx1.vin = [CTxIn(COutPoint(int(cointx, 16), 0), SCRIPTSIG_OP_TRUE)]
tx1.vout = [
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(coinvalue - 100000, P2SH_OP_TRUE),
]
tx1.rehash()
tx1_txid = node.sendrawtransaction(tx1.serialize().hex())
send_ipc_message({"tx1_txid": tx1_txid})
tx2 = CTransaction()
tx2.vin = [
CTxIn(
COutPoint(int(tx1.hash, 16), 1),
SCRIPTSIG_OP_TRUE,
)
]
tx2.vout = [
CTxOut(546, P2SH_OP_TRUE),
CTxOut(coinvalue - 200000, P2SH_OP_TRUE),
]
tx2.rehash()
tx2_txid = node.sendrawtransaction(tx2.serialize().hex())
send_ipc_message({"tx2_txid": tx2_txid})
tx3 = CTransaction()
tx3.vin = [
CTxIn(
COutPoint(int(tx1.hash, 16), 0),
SCRIPTSIG_OP_TRUE,
),
CTxIn(COutPoint(int(tx2.hash, 16), 0), SCRIPTSIG_OP_TRUE),
]
tx3.vout = [CTxOut(546, P2SH_OP_TRUE)]
tx3.rehash()
tx3_txid = node.sendrawtransaction(tx3.serialize().hex())
send_ipc_message({"tx3_txid": tx3_txid})
assert_equal(node.getblockcount(), 101)
yield True
self.log.info("Step 3: Mine a conflicting block")
block_height = 102
conflict_tx = CTransaction()
conflict_tx.vin = [CTxIn(COutPoint(int(cointx, 16), 0), SCRIPTSIG_OP_TRUE)]
pad_tx(conflict_tx)
block = create_block(
int(block_hashes[-1], 16),
create_coinbase(block_height, b"\x03" * 33),
)
block.vtx += [conflict_tx]
block.hashMerkleRoot = block.calc_merkle_root()
block.solve()
peer.send_blocks_and_test([block], node)
yield True
if __name__ == "__main__":
ChronikClient_Mempool_Conflict_Setup().main()
diff --git a/test/functional/setup_scripts/chronik-client_script_endpoints.py b/test/functional/setup_scripts/chronik-client_script_endpoints.py
index a150268f4..a16cb3970 100644
--- a/test/functional/setup_scripts/chronik-client_script_endpoints.py
+++ b/test/functional/setup_scripts/chronik-client_script_endpoints.py
@@ -1,192 +1,191 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client js library script endpoints
"""
import pathmagic # noqa
from ipc import send_ipc_message
from setup_framework import SetupFramework
from test_framework.avatools import AvaP2PInterface, can_find_inv_in_poll
from test_framework.messages import CTransaction, CTxOut, FromHex, ToHex
from test_framework.script import OP_CHECKSIG, CScript
from test_framework.util import assert_equal
QUORUM_NODE_COUNT = 16
class ChronikClient_Block_Setup(SetupFramework):
def set_test_params(self):
self.num_nodes = 1
self.extra_args = [
[
"-chronik",
"-avaproofstakeutxodustthreshold=1000000",
"-avaproofstakeutxoconfirmations=1",
"-avacooldown=0",
"-avaminquorumstake=0",
"-avaminavaproofsnodecount=0",
"-persistavapeers=0",
"-acceptnonstdtxn=1",
]
]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
self.skip_if_no_wallet()
def run_test(self):
# Init
node = self.nodes[0]
yield True
self.log.info("Step 1. Initialized regtest chain.")
# p2pkh
# IFP address p2pkh
# Note: we use this instead of node.getnewaddress() so we don't get change
# to our p2pkh address from p2sh txs, causing chronik to give hard-to-predict
# results (txs with mixed script outputs will come up in each)
p2pkh_address = "ecregtest:qrfhcnyqnl5cgrnmlfmms675w93ld7mvvqjh9pgptw"
p2pkh_output_script = bytes.fromhex(
"76a914d37c4c809fe9840e7bfa77b86bd47163f6fb6c6088ac"
)
self.log.info(f"(p2pkh_address): {p2pkh_address}")
send_ipc_message({"p2pkh_address": p2pkh_address})
# p2sh
# use IFP address p2sh
p2sh_address = "ecregtest:prfhcnyqnl5cgrnmlfmms675w93ld7mvvq9jcw0zsn"
p2sh_output_script = bytes.fromhex(
"a914d37c4c809fe9840e7bfa77b86bd47163f6fb6c6087"
)
send_ipc_message({"p2sh_address": p2sh_address})
# p2pk
# See coinbase tx of coinbase tx output from https://explorer.e.cash/block/00000000000000002328cef155f92bf149cfbe365eecf4e428f2c11f25fcce56
pubkey = bytes.fromhex(
"047fa64f6874fb7213776b24c40bc915451b57ef7f17ad7b982561f99f7cdc7010d141b856a092ee169c5405323895e1962c6b0d7c101120d360164c9e4b3997bd"
)
send_ipc_message({"p2pk_script": pubkey.hex()})
p2pk_script_for_tx_building = CScript([pubkey, OP_CHECKSIG])
# other
# Use hex deadbeef
other_script = "deadbeef"
send_ipc_message({"other_script": other_script})
other_script_for_tx_building = bytes.fromhex(other_script)
assert_equal(node.getblockcount(), 200)
yield True
self.log.info("Step 2: Broadcast txs to a p2pk, p2pkh, and p2sh address")
# Set the number of txs you wish to broadcast
# Tested up to 100, takes 25s
# 200 goes over regtest 60s timeout
# Must be <= 200 since the JS tests match items on first page and
# this is chronik's max per-page count
# 25 takes about 10s. Default page size for chronik.
txs_broadcast = 25
send_ipc_message({"txs_broadcast": txs_broadcast})
# p2pkh
p2pkh_txids = []
for x in range(txs_broadcast):
p2pkh_txid = node.sendtoaddress(p2pkh_address, (x + 1) * 1000)
p2pkh_txids.append(p2pkh_txid)
send_ipc_message({"p2pkh_txids": p2pkh_txids})
# p2sh
p2sh_txids = []
for x in range(txs_broadcast):
p2sh_txid = node.sendtoaddress(p2sh_address, (x + 1) * 1000)
p2sh_txids.append(p2sh_txid)
send_ipc_message({"p2sh_txids": p2sh_txids})
# p2pk
p2pk_txids = []
for x in range(txs_broadcast):
tx = CTransaction()
tx.vout.append(CTxOut((x + 1) * 1000, p2pk_script_for_tx_building))
rawtx = node.fundrawtransaction(
ToHex(tx),
)["hex"]
FromHex(tx, rawtx)
rawtx = node.signrawtransactionwithwallet(ToHex(tx))["hex"]
p2pk_txid = node.sendrawtransaction(rawtx)
p2pk_txids.append(p2pk_txid)
send_ipc_message({"p2pk_txids": p2pk_txids})
# other
other_txids = []
for x in range(txs_broadcast):
tx = CTransaction()
tx.vout.append(CTxOut((x + 1) * 1000, other_script_for_tx_building))
rawtx = node.fundrawtransaction(
ToHex(tx),
)["hex"]
FromHex(tx, rawtx)
rawtx = node.signrawtransactionwithwallet(ToHex(tx))["hex"]
other_txid = node.sendrawtransaction(rawtx)
other_txids.append(other_txid)
send_ipc_message({"other_txids": other_txids})
assert_equal(node.getblockcount(), 200)
yield True
self.log.info("Step 3: Mine a block with these txs")
self.generate(node, 1)
assert_equal(node.getblockcount(), 201)
yield True
self.log.info("Step 4: Avalanche finalize a block with these txs")
# Build a fake quorum of nodes.
def get_quorum():
return [
node.add_p2p_connection(AvaP2PInterface(self, node))
for _ in range(0, QUORUM_NODE_COUNT)
]
# Pick one node from the quorum for polling.
quorum = get_quorum()
def is_quorum_established():
return node.getavalancheinfo()["ready_to_poll"] is True
self.wait_until(is_quorum_established)
blockhash = self.generate(node, 1, sync_fun=self.no_op)[0]
cb_txid = node.getblock(blockhash)["tx"][0]
assert not node.isfinalblock(blockhash)
assert not node.isfinaltransaction(cb_txid, blockhash)
def is_finalblock(blockhash):
can_find_inv_in_poll(quorum, int(blockhash, 16))
return node.isfinalblock(blockhash)
with node.assert_debug_log([f"Avalanche finalized block {blockhash}"]):
self.wait_until(lambda: is_finalblock(blockhash))
assert node.isfinaltransaction(cb_txid, blockhash)
yield True
self.log.info("Step 4: Broadcast a tx with mixed outputs")
mixed_output_tx = CTransaction()
mixed_output_tx.vout.append(CTxOut(1000000, p2pkh_output_script))
mixed_output_tx.vout.append(CTxOut(1000000, p2sh_output_script))
mixed_output_tx.vout.append(CTxOut(1000000, p2pk_script_for_tx_building))
mixed_output_tx.vout.append(CTxOut(1000000, other_script_for_tx_building))
mixed_output_rawtx = node.fundrawtransaction(
ToHex(mixed_output_tx),
)["hex"]
FromHex(mixed_output_tx, mixed_output_rawtx)
mixed_output_rawtx = node.signrawtransactionwithwallet(ToHex(mixed_output_tx))[
"hex"
]
mixed_output_txid = node.sendrawtransaction(mixed_output_rawtx)
send_ipc_message({"mixed_output_txid": mixed_output_txid})
yield True
if __name__ == "__main__":
ChronikClient_Block_Setup().main()
diff --git a/test/functional/setup_scripts/chronik-client_token_alp.py b/test/functional/setup_scripts/chronik-client_token_alp.py
index ed55c80bc..fb7ba4977 100644
--- a/test/functional/setup_scripts/chronik-client_token_alp.py
+++ b/test/functional/setup_scripts/chronik-client_token_alp.py
@@ -1,388 +1,387 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client js library endpoints that receive the Tx type
for token-specific txs
Based on test/functional/chronik_token_alp.py
"""
import pathmagic # noqa
from ipc import send_ipc_message
from setup_framework import SetupFramework
from test_framework.address import (
ADDRESS_ECREG_P2SH_OP_TRUE,
ADDRESS_ECREG_UNSPENDABLE,
P2SH_OP_TRUE,
SCRIPTSIG_OP_TRUE,
)
from test_framework.blocktools import (
create_block,
create_coinbase,
make_conform_to_ctor,
)
from test_framework.chronik.alp import (
alp_burn,
alp_genesis,
alp_mint,
alp_opreturn,
alp_send,
)
from test_framework.messages import COutPoint, CTransaction, CTxIn, CTxOut
from test_framework.p2p import P2PDataStore
from test_framework.util import assert_equal
class ChronikClientTokenAlp(SetupFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.num_nodes = 1
self.extra_args = [["-chronik"]]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
def run_test(self):
# Init
node = self.nodes[0]
peer = node.add_p2p_connection(P2PDataStore())
mocktime = 1300000000
node.setmocktime(mocktime)
yield True
coinblockhash = self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0]
coinblock = node.getblock(coinblockhash)
cointx = coinblock["tx"][0]
block_hashes = self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE)
coinvalue = 5000000000
self.log.info("Step 1: Send an ALP genesis tx")
# ALP GENESIS tx
alp_genesis_tx = CTransaction()
alp_genesis_tx.vin = [CTxIn(COutPoint(int(cointx, 16), 0), SCRIPTSIG_OP_TRUE)]
alp_genesis_tx.vout = [
alp_opreturn(
alp_genesis(
token_ticker=b"TEST",
token_name=b"Test Token",
url=b"http://example.com",
data=b"Token Data",
auth_pubkey=b"Token Pubkey",
decimals=4,
mint_amounts=[10, 20, 30, 0],
num_batons=2,
)
),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(coinvalue - 100000, P2SH_OP_TRUE),
CTxOut(5000, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
]
alp_genesis_tx_txid = node.sendrawtransaction(alp_genesis_tx.serialize().hex())
send_ipc_message({"alp_genesis_txid": alp_genesis_tx_txid})
yield True
self.log.info("Step 2: Send an ALP mint tx")
# ALP MINT tx
alp_mint_tx = CTransaction()
alp_mint_tx.vin = [
CTxIn(
COutPoint(int(alp_genesis_tx_txid, 16), 5),
SCRIPTSIG_OP_TRUE,
)
]
alp_mint_tx.vout = [
alp_opreturn(
alp_mint(
token_id=alp_genesis_tx_txid,
mint_amounts=[5, 0],
num_batons=1,
),
),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
]
alp_mint_tx_txid = node.sendrawtransaction(alp_mint_tx.serialize().hex())
send_ipc_message({"alp_mint_txid": alp_mint_tx_txid})
yield True
self.log.info("Step 3: Send an ALP send tx")
# ALP SEND tx
alp_send_tx = CTransaction()
alp_send_tx.vin = [
CTxIn(
COutPoint(int(alp_genesis_tx_txid, 16), 1),
SCRIPTSIG_OP_TRUE,
),
CTxIn(COutPoint(int(alp_mint_tx_txid, 16), 1), SCRIPTSIG_OP_TRUE),
]
alp_send_tx.vout = [
alp_opreturn(
alp_send(
token_id=alp_genesis_tx_txid,
output_amounts=[3, 12],
),
),
CTxOut(5000, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
]
alp_send_tx_txid = node.sendrawtransaction(alp_send_tx.serialize().hex())
send_ipc_message({"alp_send_txid": alp_send_tx_txid})
yield True
self.log.info("Step 4: Send another ALP genesis tx")
# Another ALP GENESIS
another_alp_genesis_tx = CTransaction()
another_alp_genesis_tx.vin = [
CTxIn(
COutPoint(int(alp_genesis_tx_txid, 16), 4),
SCRIPTSIG_OP_TRUE,
)
]
another_alp_genesis_tx.vout = [
alp_opreturn(
alp_genesis(
mint_amounts=[100],
num_batons=2,
),
),
CTxOut(5000, P2SH_OP_TRUE),
CTxOut(5000, P2SH_OP_TRUE),
CTxOut(5000, P2SH_OP_TRUE),
CTxOut(coinvalue - 200000, P2SH_OP_TRUE),
]
another_alp_genesis_tx_txid = node.sendrawtransaction(
another_alp_genesis_tx.serialize().hex()
)
send_ipc_message({"alp_genesis2_txid": another_alp_genesis_tx_txid})
yield True
self.log.info("Step 5: Send an ALP genesis + mint + send all in one")
# ALP GENESIS + MINT + SEND all in one
multi_tx = CTransaction()
multi_tx.vin = [
CTxIn(COutPoint(int(alp_send_tx_txid, 16), 1), SCRIPTSIG_OP_TRUE),
CTxIn(
COutPoint(int(another_alp_genesis_tx_txid, 16), 2),
SCRIPTSIG_OP_TRUE,
),
]
multi_tx.vout = [
alp_opreturn(
alp_genesis(
token_ticker=b"MULTI",
mint_amounts=[0xFFFF_FFFF_FFFF, 0],
num_batons=1,
),
alp_mint(
token_id=another_alp_genesis_tx_txid,
mint_amounts=[0, 5],
num_batons=0,
),
alp_burn(
token_id=alp_genesis_tx_txid,
burn_amount=1,
),
alp_send(
token_id=alp_genesis_tx_txid,
output_amounts=[0, 0, 0, 0, 2],
),
),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
]
multi_tx_txid = node.sendrawtransaction(multi_tx.serialize().hex())
send_ipc_message({"alp_multi_txid": multi_tx_txid})
yield True
self.log.info("Step 6: Send wild oversized ALP tx")
# ALP tx with all kinds of things (so big it must be mined in a block manually)
mega_tx = CTransaction()
mega_tx.vin = [
CTxIn(
COutPoint(int(another_alp_genesis_tx_txid, 16), 3),
SCRIPTSIG_OP_TRUE,
),
CTxIn(
COutPoint(int(alp_genesis_tx_txid, 16), 6),
SCRIPTSIG_OP_TRUE,
),
CTxIn(COutPoint(int(multi_tx_txid, 16), 1), SCRIPTSIG_OP_TRUE),
]
mega_tx.vout = [
alp_opreturn(
# 0: success GENESIS
alp_genesis(
token_ticker=b"ALL",
mint_amounts=[0, 7, 0, 0, 1],
num_batons=2,
),
# 1: fail GENESIS: must be first
alp_genesis(mint_amounts=[], num_batons=0),
# 2: fail MINT: Too few outputs
alp_mint(alp_genesis_tx_txid, [0, 0, 0, 0, 0, 0, 0], 99),
# 3: fail MINT: Overlapping amounts
alp_mint(alp_genesis_tx_txid, [0, 0xFFFF_FFFF_FFFF], 0),
# 4: fail MINT: Overlapping batons
alp_mint(alp_genesis_tx_txid, [0], 1),
# 5: success BURN: token ID 2
alp_burn(alp_genesis_tx_txid, 2),
# 6: success MINT: token ID 3
alp_mint(another_alp_genesis_tx_txid, [3, 0], 1),
# 7: success MINT: token ID 2
alp_mint(alp_genesis_tx_txid, [0, 0, 0, 2, 0, 0, 0], 1),
# 8: fail MINT: Duplicate token ID 2
alp_mint(alp_genesis_tx_txid, [], 0),
# 9: fail BURN: Duplicate burn token ID 2
alp_burn(alp_genesis_tx_txid, 0),
# 10: fail SEND: Too few outputs
alp_send(multi_tx_txid, [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 123]),
# 11: success SEND: token ID 4
alp_send(
multi_tx_txid,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFFFF_FFFF_FFFF],
),
# 12: fail MINT: Duplicate token ID 4
alp_mint(multi_tx_txid, [], 0),
# 13: success UNKNOWN
b"SLP2\x89",
# 14: fail BURN: Descending token type
alp_burn(multi_tx_txid, 0),
# 15: success UNKNOWN
b"SLP2\x9a",
),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(1000, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
]
mega_tx_txid = node.decoderawtransaction(mega_tx.serialize().hex())["txid"]
send_ipc_message({"alp_mega_txid": mega_tx_txid})
block_height = 102
block = create_block(
int(block_hashes[-1], 16),
create_coinbase(block_height, b"\x03" * 33),
1300000500,
)
block.vtx += [
alp_genesis_tx,
alp_send_tx,
alp_mint_tx,
another_alp_genesis_tx,
multi_tx,
mega_tx,
]
make_conform_to_ctor(block)
block.hashMerkleRoot = block.calc_merkle_root()
block.solve()
peer.send_blocks_and_test([block], node)
assert_equal(node.getblockcount(), 102)
yield True
self.log.info("Step 7: Send another ALP tx in the mempool")
# Another ALP Genesis tx with non-utf8 genesis info
# we confirm this does not crash chronik-client token info endpoint
alp_nonutf8_genesis_tx = CTransaction()
alp_nonutf8_genesis_tx.vin = [
CTxIn(
COutPoint(int(another_alp_genesis_tx_txid, 16), 4),
SCRIPTSIG_OP_TRUE,
),
CTxIn(
COutPoint(int(alp_send_tx_txid, 16), 2),
SCRIPTSIG_OP_TRUE,
),
]
# Genesis with nonstandard token ticker and name to test how the client handles this
# token_name and token_url fields from coinbase of block 833028
alp_nonutf8_genesis_tx.vout = [
alp_opreturn(
alp_genesis(
token_ticker=b"TEST",
token_name=bytes.fromhex("0304b60c048de8d9650881002834d6490000"),
url=bytes.fromhex(
"00fabe6d6d6f5486f62c703086014607f5bed91d093b092a8faf5ac882a0ccf462682a22f002"
),
data=b"Token Data",
auth_pubkey=b"Token Pubkey",
decimals=4,
mint_amounts=[10, 20, 30, 0],
num_batons=2,
)
),
CTxOut(10000, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(coinvalue - 300000, P2SH_OP_TRUE),
CTxOut(5000, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
]
alp_nonutf8_genesis_tx_txid = node.sendrawtransaction(
alp_nonutf8_genesis_tx.serialize().hex()
)
send_ipc_message({"alp_nonutf8_genesis_txid": alp_nonutf8_genesis_tx_txid})
# Another ALP Mint tx
alp_mint_two_tx = CTransaction()
alp_mint_two_tx.vin = [
CTxIn(
COutPoint(int(alp_nonutf8_genesis_tx_txid, 16), 4),
SCRIPTSIG_OP_TRUE,
),
CTxIn(
COutPoint(int(alp_mint_tx_txid, 16), 3),
SCRIPTSIG_OP_TRUE,
),
]
alp_mint_two_tx.vout = [
alp_opreturn(
alp_mint(
token_id=alp_genesis_tx_txid,
mint_amounts=[5, 0],
num_batons=1,
),
),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(546, P2SH_OP_TRUE),
CTxOut(coinvalue - 400000, P2SH_OP_TRUE),
]
alp_mint_two_tx_txid = node.sendrawtransaction(
alp_mint_two_tx.serialize().hex()
)
send_ipc_message({"alp_mint_two_txid": alp_mint_two_tx_txid})
yield True
self.log.info("Step 8: Confirm your second ALP tx")
self.generate(node, 1)
assert_equal(node.getblockcount(), 103)
yield True
if __name__ == "__main__":
ChronikClientTokenAlp().main()
diff --git a/test/functional/setup_scripts/chronik-client_websocket.py b/test/functional/setup_scripts/chronik-client_websocket.py
index 20090e792..e3be9ea0e 100644
--- a/test/functional/setup_scripts/chronik-client_websocket.py
+++ b/test/functional/setup_scripts/chronik-client_websocket.py
@@ -1,213 +1,212 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to exercise the chronik-client js library script endpoints
"""
import pathmagic # noqa
from ipc import send_ipc_message
from setup_framework import SetupFramework
from test_framework.avatools import AvaP2PInterface, can_find_inv_in_poll
from test_framework.messages import CTransaction, CTxOut, FromHex, ToHex
from test_framework.script import OP_CHECKSIG, CScript
from test_framework.util import assert_equal
QUORUM_NODE_COUNT = 16
class ChronikClient_Websocket_Setup(SetupFramework):
def set_test_params(self):
self.num_nodes = 1
self.extra_args = [
[
"-chronik",
"-avaproofstakeutxodustthreshold=1000000",
"-avaproofstakeutxoconfirmations=1",
"-avacooldown=0",
"-avaminquorumstake=0",
"-avaminavaproofsnodecount=0",
"-persistavapeers=0",
"-acceptnonstdtxn=1",
]
]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
self.skip_if_no_wallet()
def run_test(self):
# Init
node = self.nodes[0]
yield True
# p2pkh
# IFP address p2pkh
# Note: we use this instead of node.getnewaddress() so we don't get change
# to our p2pkh address from p2sh txs, causing chronik to give hard-to-predict
# results (txs with mixed script outputs will come up in each)
p2pkh_address = "ecregtest:qrfhcnyqnl5cgrnmlfmms675w93ld7mvvqjh9pgptw"
p2pkh_output_script = bytes.fromhex(
"76a914d37c4c809fe9840e7bfa77b86bd47163f6fb6c6088ac"
)
send_ipc_message({"p2pkh_address": p2pkh_address})
# p2sh
# use IFP address p2sh
p2sh_address = "ecregtest:prfhcnyqnl5cgrnmlfmms675w93ld7mvvq9jcw0zsn"
p2sh_output_script = bytes.fromhex(
"a914d37c4c809fe9840e7bfa77b86bd47163f6fb6c6087"
)
send_ipc_message({"p2sh_address": p2sh_address})
# p2pk
# See coinbase tx of coinbase tx output from https://explorer.e.cash/block/00000000000000002328cef155f92bf149cfbe365eecf4e428f2c11f25fcce56
pubkey = bytes.fromhex(
"047fa64f6874fb7213776b24c40bc915451b57ef7f17ad7b982561f99f7cdc7010d141b856a092ee169c5405323895e1962c6b0d7c101120d360164c9e4b3997bd"
)
send_ipc_message({"p2pk_script": pubkey.hex()})
p2pk_script_for_tx_building = CScript([pubkey, OP_CHECKSIG])
# other
# Use hex deadbeef
other_script = "deadbeef"
send_ipc_message({"other_script": other_script})
other_script_for_tx_building = bytes.fromhex(other_script)
# Build a fake quorum of nodes.
def get_quorum():
return [
node.add_p2p_connection(AvaP2PInterface(self, node))
for _ in range(0, QUORUM_NODE_COUNT)
]
# Pick one node from the quorum for polling.
quorum = get_quorum()
def is_quorum_established():
return node.getavalancheinfo()["ready_to_poll"] is True
def is_finalblock(blockhash):
can_find_inv_in_poll(quorum, int(blockhash, 16))
return node.isfinalblock(blockhash)
self.wait_until(is_quorum_established)
self.wait_until(lambda: is_finalblock(node.getbestblockhash()))
finalized_blockhash = self.generate(node, 1, sync_fun=self.no_op)[0]
cb_txid = node.getblock(finalized_blockhash)["tx"][0]
assert not node.isfinalblock(finalized_blockhash)
assert not node.isfinaltransaction(cb_txid, finalized_blockhash)
yield True
self.log.info("Step 1: Avalanche finalize a block")
with node.assert_debug_log(
[f"Avalanche finalized block {finalized_blockhash}"]
):
self.wait_until(lambda: is_finalblock(finalized_blockhash))
send_ipc_message({"finalized_block_blockhash": finalized_blockhash})
finalized_height = node.getblock(finalized_blockhash, 1)["height"]
send_ipc_message({"finalized_height": finalized_height})
assert node.isfinaltransaction(cb_txid, finalized_blockhash)
yield True
self.log.info("Step 2: Broadcast 1 tx to a p2pk, p2pkh, and p2sh address")
# p2pkh
p2pkh_txid = node.sendtoaddress(p2pkh_address, 1000)
send_ipc_message({"p2pkh_txid": p2pkh_txid})
# p2sh
p2sh_txid = node.sendtoaddress(p2sh_address, 1000)
send_ipc_message({"p2sh_txid": p2sh_txid})
# p2pk
p2pk_tx = CTransaction()
p2pk_tx.vout.append(CTxOut(1000, p2pk_script_for_tx_building))
p2pk_rawtx = node.fundrawtransaction(
ToHex(p2pk_tx),
)["hex"]
FromHex(p2pk_tx, p2pk_rawtx)
p2pk_rawtx = node.signrawtransactionwithwallet(ToHex(p2pk_tx))["hex"]
p2pk_txid = node.sendrawtransaction(p2pk_rawtx)
send_ipc_message({"p2pk_txid": p2pk_txid})
# other
other_tx = CTransaction()
other_tx.vout.append(CTxOut(1000, other_script_for_tx_building))
other_rawtx = node.fundrawtransaction(
ToHex(other_tx),
)["hex"]
FromHex(other_tx, other_rawtx)
other_rawtx = node.signrawtransactionwithwallet(ToHex(other_tx))["hex"]
other_txid = node.sendrawtransaction(other_rawtx)
send_ipc_message({"other_txid": other_txid})
assert_equal(node.getblockcount(), finalized_height)
yield True
self.log.info("Step 3: Mine a block with these txs")
next_blockhash = self.generate(node, 1, sync_fun=self.no_op)[0]
send_ipc_message({"next_blockhash": next_blockhash})
assert_equal(node.getblockcount(), finalized_height + 1)
yield True
self.log.info("Step 4: Park the block containing those txs")
node.parkblock(next_blockhash)
assert_equal(node.getblockcount(), finalized_height)
yield True
self.log.info("Step 5: Unpark the block containing those txs")
node.unparkblock(next_blockhash)
assert_equal(node.getblockcount(), finalized_height + 1)
yield True
self.log.info("Step 6: Invalidate the block containing those txs")
node.invalidateblock(next_blockhash)
assert_equal(node.getblockcount(), finalized_height)
yield True
self.log.info("Step 7: Reconsider the block containing those txs")
node.reconsiderblock(next_blockhash)
assert_equal(node.getblockcount(), finalized_height + 1)
yield True
self.log.info("Step 8: Finalize the block containing these txs with Avalanche")
next_cb_txid = node.getblock(next_blockhash)["tx"][0]
with node.assert_debug_log([f"Avalanche finalized block {next_blockhash}"]):
self.wait_until(lambda: is_finalblock(next_blockhash))
assert_equal(node.getblockcount(), finalized_height + 1)
assert node.isfinaltransaction(next_cb_txid, next_blockhash)
yield True
self.log.info("Step 9: Broadcast a tx with mixed outputs")
mixed_output_tx = CTransaction()
mixed_output_tx.vout.append(CTxOut(1000000, p2pkh_output_script))
mixed_output_tx.vout.append(CTxOut(1000000, p2sh_output_script))
mixed_output_tx.vout.append(CTxOut(1000000, p2pk_script_for_tx_building))
mixed_output_tx.vout.append(CTxOut(1000000, other_script_for_tx_building))
mixed_output_rawtx = node.fundrawtransaction(
ToHex(mixed_output_tx),
)["hex"]
FromHex(mixed_output_tx, mixed_output_rawtx)
mixed_output_rawtx = node.signrawtransactionwithwallet(ToHex(mixed_output_tx))[
"hex"
]
mixed_output_txid = node.sendrawtransaction(mixed_output_rawtx)
send_ipc_message({"mixed_output_txid": mixed_output_txid})
yield True
self.log.info("Step 10: Mine another block")
parked_block_hash = self.generate(node, 1)[0]
send_ipc_message({"parked_block_hash": parked_block_hash})
assert_equal(node.getblockcount(), finalized_height + 2)
yield True
if __name__ == "__main__":
ChronikClient_Websocket_Setup().main()
diff --git a/test/functional/setup_scripts/ecash-lib_base.py b/test/functional/setup_scripts/ecash-lib_base.py
index c5939be53..9bf85117e 100644
--- a/test/functional/setup_scripts/ecash-lib_base.py
+++ b/test/functional/setup_scripts/ecash-lib_base.py
@@ -1,35 +1,34 @@
# Copyright (c) 2024 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""
Setup script to test the ecash-lib js library
"""
import pathmagic # noqa
from setup_framework import SetupFramework
from test_framework.address import ADDRESS_ECREG_P2SH_OP_TRUE, ADDRESS_ECREG_UNSPENDABLE
class EcashLibSetup(SetupFramework):
def set_test_params(self):
self.num_nodes = 1
self.setup_clean_chain = True
self.extra_args = [["-chronik"]]
- self.ipc_timeout = 10
def skip_test_if_missing_module(self):
self.skip_if_no_chronik()
def run_test(self):
node = self.nodes[0]
mocktime = 1300000000
node.setmocktime(mocktime)
self.generatetoaddress(node, 1, ADDRESS_ECREG_P2SH_OP_TRUE)[0]
self.generatetoaddress(node, 100, ADDRESS_ECREG_UNSPENDABLE)
yield True
if __name__ == "__main__":
EcashLibSetup().main()

File Metadata

Mime Type
text/x-diff
Expires
Sun, Mar 2, 11:29 (1 d, 6 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
5187613
Default Alt Text
(47 KB)

Event Timeline