mirror of
https://github.com/ElementsProject/elements.git
synced 2026-08-13 12:33:42 +02:00
134 lines
5.2 KiB
Python
Executable file
134 lines
5.2 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Test getnewblockhex
|
|
"""
|
|
from io import BytesIO
|
|
|
|
from test_framework.messages import CBlock
|
|
from test_framework.p2p import (
|
|
P2PDataStore,
|
|
)
|
|
from test_framework.test_framework import BitcoinTestFramework
|
|
from test_framework.util import (
|
|
assert_equal,
|
|
)
|
|
|
|
class GetNewBlockHexTest(BitcoinTestFramework):
|
|
def set_test_params(self):
|
|
self.setup_clean_chain = False
|
|
self.num_nodes = 1
|
|
|
|
def run_test(self):
|
|
self.log.info("Starting test...")
|
|
self.bootstrap_p2p()
|
|
|
|
node = self.nodes[0]
|
|
|
|
height = node.getblockcount()
|
|
assert_equal(height, 200)
|
|
|
|
self.log.info("Call getnewblockhex with no args")
|
|
hex = node.getnewblockhex()
|
|
(height, hash) = self.mine_block(hex)
|
|
assert_equal(height, 201)
|
|
|
|
self.log.info("Call getnewblockhex with single string commitment")
|
|
hex = node.getnewblockhex(0, None, "1234")
|
|
assert "6a021234" in hex
|
|
(height, hash) = self.mine_block(hex)
|
|
assert_equal(height, 202)
|
|
block = node.getblock(hash, 2)
|
|
vout = block["tx"][0]["vout"]
|
|
assert_equal(vout[0]["scriptPubKey"]["hex"], "6a021234")
|
|
|
|
self.log.info("Call getnewblockhex with single string commitment with spaces")
|
|
# ParseHex only validates hex chars, so spaces skipped
|
|
hex = node.getnewblockhex(0, None, "1234 5678")
|
|
assert "6a0412345678" in hex
|
|
(height, hash) = self.mine_block(hex)
|
|
assert_equal(height, 203)
|
|
block = node.getblock(hash, 2)
|
|
vout = block["tx"][0]["vout"]
|
|
assert_equal(vout[0]["scriptPubKey"]["hex"], "6a0412345678")
|
|
|
|
self.log.info("Call getnewblockhex with single commitment")
|
|
hex = node.getnewblockhex(0, None, ["1234"])
|
|
assert "6a021234" in hex
|
|
(height, hash) = self.mine_block(hex)
|
|
assert_equal(height, 204)
|
|
block = node.getblock(hash, 2)
|
|
vout = block["tx"][0]["vout"]
|
|
assert_equal(vout[0]["scriptPubKey"]["hex"], "6a021234")
|
|
|
|
self.log.info("Call getnewblockhex with multiple commitments")
|
|
hex = node.getnewblockhex(0, None, ["1234", "deadbeef"])
|
|
assert "6a021234" in hex
|
|
assert "6a04deadbeef" in hex
|
|
(height, hash) = self.mine_block(hex)
|
|
assert_equal(height, 205)
|
|
block = node.getblock(hash, 2)
|
|
vout = block["tx"][0]["vout"]
|
|
assert_equal(vout[0]["scriptPubKey"]["hex"], "6a021234")
|
|
assert_equal(vout[1]["scriptPubKey"]["hex"], "6a04deadbeef")
|
|
|
|
hex = node.getnewblockhex(0, None, ["1234", "dead", "cafe", "babe"])
|
|
assert "6a021234" in hex
|
|
assert "6a02dead" in hex
|
|
assert "6a02cafe" in hex
|
|
assert "6a02babe" in hex
|
|
(height, hash) = self.mine_block(hex)
|
|
assert_equal(height, 206)
|
|
block = node.getblock(hash, 2)
|
|
vout = block["tx"][0]["vout"]
|
|
assert_equal(vout[0]["scriptPubKey"]["hex"], "6a021234")
|
|
assert_equal(vout[1]["scriptPubKey"]["hex"], "6a02dead")
|
|
assert_equal(vout[2]["scriptPubKey"]["hex"], "6a02cafe")
|
|
assert_equal(vout[3]["scriptPubKey"]["hex"], "6a02babe")
|
|
|
|
self.log.info("Done.")
|
|
|
|
def mine_block(self, hex):
|
|
"""Mine and submit a block from a given hex template.
|
|
|
|
Returns a tuple of the new chain height and the block hash."""
|
|
|
|
b = bytes.fromhex(hex)
|
|
block = CBlock()
|
|
block.deserialize(BytesIO(b))
|
|
block.solve()
|
|
self.send_blocks([block], success=True)
|
|
height = self.nodes[0].getblockcount()
|
|
return (height, block.hash)
|
|
|
|
def bootstrap_p2p(self, timeout=10):
|
|
"""Add a P2P connection to the node.
|
|
|
|
Helper to connect and wait for version handshake."""
|
|
self.helper_peer = self.nodes[0].add_p2p_connection(P2PDataStore())
|
|
# We need to wait for the initial getheaders from the peer before we
|
|
# start populating our blockstore. If we don't, then we may run ahead
|
|
# to the next subtest before we receive the getheaders. We'd then send
|
|
# an INV for the next block and receive two getheaders - one for the
|
|
# IBD and one for the INV. We'd respond to both and could get
|
|
# unexpectedly disconnected if the DoS score for that error is 50.
|
|
self.helper_peer.wait_for_getheaders(timeout=timeout)
|
|
|
|
def reconnect_p2p(self, timeout=60):
|
|
"""Tear down and bootstrap the P2P connection to the node.
|
|
|
|
The node gets disconnected several times in this test. This helper
|
|
method reconnects the p2p and restarts the network thread."""
|
|
self.nodes[0].disconnect_p2ps()
|
|
self.bootstrap_p2p(timeout=timeout)
|
|
|
|
def send_blocks(self, blocks, success=True, reject_reason=None, force_send=False, reconnect=False, timeout=960):
|
|
"""Sends blocks to test node. Syncs and verifies that tip has advanced to most recent block.
|
|
|
|
Call with success = False if the tip shouldn't advance to the most recent block."""
|
|
self.helper_peer.send_blocks_and_test(blocks, self.nodes[0], success=success, reject_reason=reject_reason, force_send=force_send, timeout=timeout, expect_disconnect=reconnect)
|
|
|
|
if reconnect:
|
|
self.reconnect_p2p(timeout=timeout)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
GetNewBlockHexTest(__file__).main()
|