Differential D1961 Diff 5681 test/functional/bip68-sequence.py

Changeset View

Standalone View

test/functional/bip68-sequence.py

#!/usr/bin/env python3		#!/usr/bin/env python3
# Copyright (c) 2014-2016 The Bitcoin Core developers		# Copyright (c) 2014-2016 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying		# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.		# file COPYING or http://www.opensource.org/licenses/mit-license.php.

#		#
# Test BIP68 implementation		# Test BIP68 implementation
#		#

from test_framework.test_framework import BitcoinTestFramework		from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import *		from test_framework.util import *
from test_framework.script import *		from test_framework.script import *
from test_framework.mininode import *		from test_framework.mininode import *
from test_framework.blocktools import *		from test_framework.blocktools import *
		from test_framework.txtools import pad_tx, pad_raw_tx

SEQUENCE_LOCKTIME_DISABLE_FLAG = (1 << 31)		SEQUENCE_LOCKTIME_DISABLE_FLAG = (1 << 31)
SEQUENCE_LOCKTIME_TYPE_FLAG = (1 << 22) # this means use time (0 means height)		SEQUENCE_LOCKTIME_TYPE_FLAG = (1 << 22) # this means use time (0 means height)
SEQUENCE_LOCKTIME_GRANULARITY = 9 # this is a bit-shift		SEQUENCE_LOCKTIME_GRANULARITY = 9 # this is a bit-shift
SEQUENCE_LOCKTIME_MASK = 0x0000ffff		SEQUENCE_LOCKTIME_MASK = 0x0000ffff

# RPC error for non-BIP68 final transactions		# RPC error for non-BIP68 final transactions
NOT_FINAL_ERROR = "64: non-BIP68-final"		NOT_FINAL_ERROR = "64: non-BIP68-final"


class BIP68Test(BitcoinTestFramework):		class BIP68Test(BitcoinTestFramework):
		def add_options(self, parser):
		parser.add_argument("--magnetic-anomaly-time", dest="magnetic_anomaly_time", type=long, default=0,
		help="Test Magnetic Anomaly activation at a particular time (0 for never, 1 for startup)")

def set_test_params(self):		def set_test_params(self):
self.num_nodes = 2		self.num_nodes = 2
self.extra_args = [["-blockprioritypercentage=0"],		self.extra_args = [["-blockprioritypercentage=0"],
["-blockprioritypercentage=0", "-acceptnonstdtxn=0"]]		["-blockprioritypercentage=0", "-acceptnonstdtxn=0"]]

		activation_time = self.options.magnetic_anomaly_time
		if activation_time != 0:
		for args in self.extra_args:
		args.append(
		"-magneticanomalyactivationtime={}".format(activation_time))

def run_test(self):		def run_test(self):
self.relayfee = self.nodes[0].getnetworkinfo()["relayfee"]		self.relayfee = self.nodes[0].getnetworkinfo()["relayfee"]

# Generate some coins		# Generate some coins
self.nodes[0].generate(110)		self.nodes[0].generate(110)

self.log.info("Running test disable flag")		self.log.info("Running test disable flag")
self.test_disable_flag()		self.test_disable_flag()
Show All 36 Lines	def test_disable_flag(self):

# Check that the disable flag disables relative locktime.		# Check that the disable flag disables relative locktime.
# If sequence locks were used, this would require 1 block for the		# If sequence locks were used, this would require 1 block for the
# input to mature.		# input to mature.
sequence_value = SEQUENCE_LOCKTIME_DISABLE_FLAG \| 1		sequence_value = SEQUENCE_LOCKTIME_DISABLE_FLAG \| 1
tx1.vin = [		tx1.vin = [
CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]), nSequence=sequence_value)]		CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]), nSequence=sequence_value)]
tx1.vout = [CTxOut(value, CScript([b'a']))]		tx1.vout = [CTxOut(value, CScript([b'a']))]
		pad_tx(tx1)

tx1_signed = self.nodes[0].signrawtransaction(ToHex(tx1))["hex"]		tx1_signed = self.nodes[0].signrawtransaction(ToHex(tx1))["hex"]
tx1_id = self.nodes[0].sendrawtransaction(tx1_signed)		tx1_id = self.nodes[0].sendrawtransaction(tx1_signed)
tx1_id = int(tx1_id, 16)		tx1_id = int(tx1_id, 16)

# This transaction will enable sequence-locks, so this transaction should		# This transaction will enable sequence-locks, so this transaction should
# fail		# fail
tx2 = CTransaction()		tx2 = CTransaction()
tx2.nVersion = 2		tx2.nVersion = 2
sequence_value = sequence_value & 0x7fffffff		sequence_value = sequence_value & 0x7fffffff
tx2.vin = [CTxIn(COutPoint(tx1_id, 0), nSequence=sequence_value)]		tx2.vin = [CTxIn(COutPoint(tx1_id, 0), nSequence=sequence_value)]
tx2.vout = [CTxOut(int(value - self.relayfee * COIN), CScript([b'a']))]		tx2.vout = [CTxOut(int(value - self.relayfee * COIN), CScript([b'a']))]
		pad_tx(tx2)
tx2.rehash()		tx2.rehash()

assert_raises_rpc_error(-26, NOT_FINAL_ERROR,		assert_raises_rpc_error(-26, NOT_FINAL_ERROR,
self.nodes[0].sendrawtransaction, ToHex(tx2))		self.nodes[0].sendrawtransaction, ToHex(tx2))

# Setting the version back down to 1 should disable the sequence lock,		# Setting the version back down to 1 should disable the sequence lock,
# so this should be accepted.		# so this should be accepted.
tx2.nVersion = 1		tx2.nVersion = 1
▲ Show 20 Lines • Show All 151 Lines • ▼ Show 20 Lines	def test_sequence_lock_unconfirmed_inputs(self):
sequence_value \|= SEQUENCE_LOCKTIME_TYPE_FLAG		sequence_value \|= SEQUENCE_LOCKTIME_TYPE_FLAG

tx = CTransaction()		tx = CTransaction()
tx.nVersion = 2		tx.nVersion = 2
tx.vin = [		tx.vin = [
CTxIn(COutPoint(orig_tx.sha256, 0), nSequence=sequence_value)]		CTxIn(COutPoint(orig_tx.sha256, 0), nSequence=sequence_value)]
tx.vout = [		tx.vout = [
CTxOut(int(orig_tx.vout[0].nValue - fee_multiplier * node.calculate_fee(tx)), CScript([b'a']))]		CTxOut(int(orig_tx.vout[0].nValue - fee_multiplier * node.calculate_fee(tx)), CScript([b'a']))]
		pad_tx(tx)
tx.rehash()		tx.rehash()

if (orig_tx.hash in node.getrawmempool()):		if (orig_tx.hash in node.getrawmempool()):
# sendrawtransaction should fail if the tx is in the mempool		# sendrawtransaction should fail if the tx is in the mempool
assert_raises_rpc_error(-26, NOT_FINAL_ERROR,		assert_raises_rpc_error(-26, NOT_FINAL_ERROR,
node.sendrawtransaction, ToHex(tx))		node.sendrawtransaction, ToHex(tx))
else:		else:
# sendrawtransaction should succeed if the tx is not in the mempool		# sendrawtransaction should succeed if the tx is not in the mempool
▲ Show 20 Lines • Show All 123 Lines • ▼ Show 20 Lines	def test_bip68_not_consensus(self):
tx2.nVersion = 1		tx2.nVersion = 1
tx2.vin = [CTxIn(COutPoint(tx1.sha256, 0), nSequence=0)]		tx2.vin = [CTxIn(COutPoint(tx1.sha256, 0), nSequence=0)]
tx2.vout = [		tx2.vout = [
CTxOut(int(tx1.vout[0].nValue - self.relayfee * COIN), CScript([b'a']))]		CTxOut(int(tx1.vout[0].nValue - self.relayfee * COIN), CScript([b'a']))]

# sign tx2		# sign tx2
tx2_raw = self.nodes[0].signrawtransaction(ToHex(tx2))["hex"]		tx2_raw = self.nodes[0].signrawtransaction(ToHex(tx2))["hex"]
tx2 = FromHex(tx2, tx2_raw)		tx2 = FromHex(tx2, tx2_raw)
		pad_tx(tx2)
tx2.rehash()		tx2.rehash()

self.nodes[0].sendrawtransaction(ToHex(tx2))		self.nodes[0].sendrawtransaction(ToHex(tx2))

# Now make an invalid spend of tx2 according to BIP68		# Now make an invalid spend of tx2 according to BIP68
sequence_value = 100 # 100 block relative locktime		sequence_value = 100 # 100 block relative locktime

tx3 = CTransaction()		tx3 = CTransaction()
tx3.nVersion = 2		tx3.nVersion = 2
tx3.vin = [CTxIn(COutPoint(tx2.sha256, 0), nSequence=sequence_value)]		tx3.vin = [CTxIn(COutPoint(tx2.sha256, 0), nSequence=sequence_value)]
tx3.vout = [		tx3.vout = [
CTxOut(int(tx2.vout[0].nValue - self.relayfee * COIN), CScript([b'a']))]		CTxOut(int(tx2.vout[0].nValue - self.relayfee * COIN), CScript([b'a']))]
		pad_tx(tx3)
tx3.rehash()		tx3.rehash()

assert_raises_rpc_error(-26, NOT_FINAL_ERROR,		assert_raises_rpc_error(-26, NOT_FINAL_ERROR,
self.nodes[0].sendrawtransaction, ToHex(tx3))		self.nodes[0].sendrawtransaction, ToHex(tx3))

# make a block that violates bip68; ensure that the tip updates		# make a block that violates bip68; ensure that the tip updates
tip = int(self.nodes[0].getbestblockhash(), 16)		tip = int(self.nodes[0].getbestblockhash(), 16)
block = create_block(		block = create_block(
tip, create_coinbase(self.nodes[0].getblockcount() + 1))		tip, create_coinbase(self.nodes[0].getblockcount() + 1))
block.nVersion = 3		block.nVersion = 3
		if self.options.magnetic_anomaly_time == 0:
block.vtx.extend([tx1, tx2, tx3])		block.vtx.extend([tx1, tx2, tx3])
		else:
		block.vtx.extend(
		sorted([tx1, tx2, tx3], key=lambda tx: tx.get_id()))
block.hashMerkleRoot = block.calc_merkle_root()		block.hashMerkleRoot = block.calc_merkle_root()
block.rehash()		block.rehash()
block.solve()		block.solve()

self.nodes[0].submitblock(ToHex(block))		self.nodes[0].submitblock(ToHex(block))
assert_equal(self.nodes[0].getbestblockhash(), block.hash)		assert_equal(self.nodes[0].getbestblockhash(), block.hash)

def activateCSV(self):		def activateCSV(self):
Show All 33 Lines