Page Menu
Home
Phabricator
Search
Configure Global Search
Log In
Files
F13115591
No One
Temporary
Actions
View File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Flag For Later
Size
47 KB
Subscribers
None
View Options
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
Details
Attached
Mime Type
text/x-diff
Expires
Sun, Mar 2, 11:29 (1 d, 8 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
5187613
Default Alt Text
(47 KB)
Attached To
rABC Bitcoin ABC
Event Timeline
Log In to Comment