test: Run pep-8 on touched test

Can be reviewed with --ignore-all-space
This commit is contained in:
MarcoFalke 2021-06-10 13:56:05 +02:00
parent fab7e99c2a
commit fa7d71f270
No known key found for this signature in database
GPG Key ID: CE2B75697E69A548

View File

@ -20,6 +20,7 @@ MAX_REPLACEMENT_LIMIT = 100
def txToHex(tx): def txToHex(tx):
return tx.serialize().hex() return tx.serialize().hex()
def make_utxo(node, amount, confirmed=True, scriptPubKey=DUMMY_P2WPKH_SCRIPT): def make_utxo(node, amount, confirmed=True, scriptPubKey=DUMMY_P2WPKH_SCRIPT):
"""Create a txout with a given amount and scriptPubKey """Create a txout with a given amount and scriptPubKey
@ -28,12 +29,12 @@ def make_utxo(node, amount, confirmed=True, scriptPubKey=DUMMY_P2WPKH_SCRIPT):
confirmed - txouts created will be confirmed in the blockchain; confirmed - txouts created will be confirmed in the blockchain;
unconfirmed otherwise. unconfirmed otherwise.
""" """
fee = 1*COIN fee = 1 * COIN
while node.getbalance() < satoshi_round((amount + fee)/COIN): while node.getbalance() < satoshi_round((amount + fee) / COIN):
node.generate(COINBASE_MATURITY) node.generate(COINBASE_MATURITY)
new_addr = node.getnewaddress() new_addr = node.getnewaddress()
txid = node.sendtoaddress(new_addr, satoshi_round((amount+fee)/COIN)) txid = node.sendtoaddress(new_addr, satoshi_round((amount + fee) / COIN))
tx1 = node.getrawtransaction(txid, 1) tx1 = node.getrawtransaction(txid, 1)
txid = int(txid, 16) txid = int(txid, 16)
i, _ = next(filter(lambda vout: new_addr == vout[1]['scriptPubKey']['address'], enumerate(tx1['vout']))) i, _ = next(filter(lambda vout: new_addr == vout[1]['scriptPubKey']['address'], enumerate(tx1['vout'])))
@ -80,7 +81,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
self.skip_if_no_wallet() self.skip_if_no_wallet()
def run_test(self): def run_test(self):
make_utxo(self.nodes[0], 1*COIN) make_utxo(self.nodes[0], 1 * COIN)
# Ensure nodes are synced # Ensure nodes are synced
self.sync_all() self.sync_all()
@ -122,7 +123,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_simple_doublespend(self): def test_simple_doublespend(self):
"""Simple doublespend""" """Simple doublespend"""
tx0_outpoint = make_utxo(self.nodes[0], int(1.1*COIN)) tx0_outpoint = make_utxo(self.nodes[0], int(1.1 * COIN))
# make_utxo may have generated a bunch of blocks, so we need to sync # make_utxo may have generated a bunch of blocks, so we need to sync
# before we can spend the coins generated, or else the resulting # before we can spend the coins generated, or else the resulting
@ -164,14 +165,14 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_doublespend_chain(self): def test_doublespend_chain(self):
"""Doublespend of a long chain""" """Doublespend of a long chain"""
initial_nValue = 50*COIN initial_nValue = 50 * COIN
tx0_outpoint = make_utxo(self.nodes[0], initial_nValue) tx0_outpoint = make_utxo(self.nodes[0], initial_nValue)
prevout = tx0_outpoint prevout = tx0_outpoint
remaining_value = initial_nValue remaining_value = initial_nValue
chain_txids = [] chain_txids = []
while remaining_value > 10*COIN: while remaining_value > 10 * COIN:
remaining_value -= 1*COIN remaining_value -= 1 * COIN
tx = CTransaction() tx = CTransaction()
tx.vin = [CTxIn(prevout, nSequence=0)] tx.vin = [CTxIn(prevout, nSequence=0)]
tx.vout = [CTxOut(remaining_value, CScript([1, OP_DROP] * 15 + [1]))] tx.vout = [CTxOut(remaining_value, CScript([1, OP_DROP] * 15 + [1]))]
@ -204,10 +205,10 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_doublespend_tree(self): def test_doublespend_tree(self):
"""Doublespend of a big tree of transactions""" """Doublespend of a big tree of transactions"""
initial_nValue = 50*COIN initial_nValue = 50 * COIN
tx0_outpoint = make_utxo(self.nodes[0], initial_nValue) tx0_outpoint = make_utxo(self.nodes[0], initial_nValue)
def branch(prevout, initial_value, max_txs, tree_width=5, fee=0.0001*COIN, _total_txs=None): def branch(prevout, initial_value, max_txs, tree_width=5, fee=0.0001 * COIN, _total_txs=None):
if _total_txs is None: if _total_txs is None:
_total_txs = [0] _total_txs = [0]
if _total_txs[0] >= max_txs: if _total_txs[0] >= max_txs:
@ -238,7 +239,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
_total_txs=_total_txs): _total_txs=_total_txs):
yield x yield x
fee = int(0.0001*COIN) fee = int(0.0001 * COIN)
n = MAX_REPLACEMENT_LIMIT n = MAX_REPLACEMENT_LIMIT
tree_txs = list(branch(tx0_outpoint, initial_nValue, n, fee=fee)) tree_txs = list(branch(tx0_outpoint, initial_nValue, n, fee=fee))
assert_equal(len(tree_txs), n) assert_equal(len(tree_txs), n)
@ -266,8 +267,8 @@ class ReplaceByFeeTest(BitcoinTestFramework):
# Try again, but with more total transactions than the "max txs # Try again, but with more total transactions than the "max txs
# double-spent at once" anti-DoS limit. # double-spent at once" anti-DoS limit.
for n in (MAX_REPLACEMENT_LIMIT+1, MAX_REPLACEMENT_LIMIT*2): for n in (MAX_REPLACEMENT_LIMIT + 1, MAX_REPLACEMENT_LIMIT * 2):
fee = int(0.0001*COIN) fee = int(0.0001 * COIN)
tx0_outpoint = make_utxo(self.nodes[0], initial_nValue) tx0_outpoint = make_utxo(self.nodes[0], initial_nValue)
tree_txs = list(branch(tx0_outpoint, initial_nValue, n, fee=fee)) tree_txs = list(branch(tx0_outpoint, initial_nValue, n, fee=fee))
assert_equal(len(tree_txs), n) assert_equal(len(tree_txs), n)
@ -285,7 +286,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_replacement_feeperkb(self): def test_replacement_feeperkb(self):
"""Replacement requires fee-per-KB to be higher""" """Replacement requires fee-per-KB to be higher"""
tx0_outpoint = make_utxo(self.nodes[0], int(1.1*COIN)) tx0_outpoint = make_utxo(self.nodes[0], int(1.1 * COIN))
tx1a = CTransaction() tx1a = CTransaction()
tx1a.vin = [CTxIn(tx0_outpoint, nSequence=0)] tx1a.vin = [CTxIn(tx0_outpoint, nSequence=0)]
@ -297,7 +298,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
# rejected. # rejected.
tx1b = CTransaction() tx1b = CTransaction()
tx1b.vin = [CTxIn(tx0_outpoint, nSequence=0)] tx1b.vin = [CTxIn(tx0_outpoint, nSequence=0)]
tx1b.vout = [CTxOut(int(0.001*COIN), CScript([b'a'*999000]))] tx1b.vout = [CTxOut(int(0.001 * COIN), CScript([b'a' * 999000]))]
tx1b_hex = txToHex(tx1b) tx1b_hex = txToHex(tx1b)
# This will raise an exception due to insufficient fee # This will raise an exception due to insufficient fee
@ -305,8 +306,8 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_spends_of_conflicting_outputs(self): def test_spends_of_conflicting_outputs(self):
"""Replacements that spend conflicting tx outputs are rejected""" """Replacements that spend conflicting tx outputs are rejected"""
utxo1 = make_utxo(self.nodes[0], int(1.2*COIN)) utxo1 = make_utxo(self.nodes[0], int(1.2 * COIN))
utxo2 = make_utxo(self.nodes[0], 3*COIN) utxo2 = make_utxo(self.nodes[0], 3 * COIN)
tx1a = CTransaction() tx1a = CTransaction()
tx1a.vin = [CTxIn(utxo1, nSequence=0)] tx1a.vin = [CTxIn(utxo1, nSequence=0)]
@ -345,8 +346,8 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_new_unconfirmed_inputs(self): def test_new_unconfirmed_inputs(self):
"""Replacements that add new unconfirmed inputs are rejected""" """Replacements that add new unconfirmed inputs are rejected"""
confirmed_utxo = make_utxo(self.nodes[0], int(1.1*COIN)) confirmed_utxo = make_utxo(self.nodes[0], int(1.1 * COIN))
unconfirmed_utxo = make_utxo(self.nodes[0], int(0.1*COIN), False) unconfirmed_utxo = make_utxo(self.nodes[0], int(0.1 * COIN), False)
tx1 = CTransaction() tx1 = CTransaction()
tx1.vin = [CTxIn(confirmed_utxo)] tx1.vin = [CTxIn(confirmed_utxo)]
@ -368,13 +369,13 @@ class ReplaceByFeeTest(BitcoinTestFramework):
# transactions # transactions
# Start by creating a single transaction with many outputs # Start by creating a single transaction with many outputs
initial_nValue = 10*COIN initial_nValue = 10 * COIN
utxo = make_utxo(self.nodes[0], initial_nValue) utxo = make_utxo(self.nodes[0], initial_nValue)
fee = int(0.0001*COIN) fee = int(0.0001 * COIN)
split_value = int((initial_nValue-fee)/(MAX_REPLACEMENT_LIMIT+1)) split_value = int((initial_nValue - fee) / (MAX_REPLACEMENT_LIMIT + 1))
outputs = [] outputs = []
for _ in range(MAX_REPLACEMENT_LIMIT+1): for _ in range(MAX_REPLACEMENT_LIMIT + 1):
outputs.append(CTxOut(split_value, CScript([1]))) outputs.append(CTxOut(split_value, CScript([1])))
splitting_tx = CTransaction() splitting_tx = CTransaction()
@ -386,7 +387,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
txid = int(txid, 16) txid = int(txid, 16)
# Now spend each of those outputs individually # Now spend each of those outputs individually
for i in range(MAX_REPLACEMENT_LIMIT+1): for i in range(MAX_REPLACEMENT_LIMIT + 1):
tx_i = CTransaction() tx_i = CTransaction()
tx_i.vin = [CTxIn(COutPoint(txid, i), nSequence=0)] tx_i.vin = [CTxIn(COutPoint(txid, i), nSequence=0)]
tx_i.vout = [CTxOut(split_value - fee, DUMMY_P2WPKH_SCRIPT)] tx_i.vout = [CTxOut(split_value - fee, DUMMY_P2WPKH_SCRIPT)]
@ -396,9 +397,9 @@ class ReplaceByFeeTest(BitcoinTestFramework):
# Now create doublespend of the whole lot; should fail. # Now create doublespend of the whole lot; should fail.
# Need a big enough fee to cover all spending transactions and have # Need a big enough fee to cover all spending transactions and have
# a higher fee rate # a higher fee rate
double_spend_value = (split_value-100*fee)*(MAX_REPLACEMENT_LIMIT+1) double_spend_value = (split_value - 100 * fee) * (MAX_REPLACEMENT_LIMIT + 1)
inputs = [] inputs = []
for i in range(MAX_REPLACEMENT_LIMIT+1): for i in range(MAX_REPLACEMENT_LIMIT + 1):
inputs.append(CTxIn(COutPoint(txid, i), nSequence=0)) inputs.append(CTxIn(COutPoint(txid, i), nSequence=0))
double_tx = CTransaction() double_tx = CTransaction()
double_tx.vin = inputs double_tx.vin = inputs
@ -417,7 +418,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_opt_in(self): def test_opt_in(self):
"""Replacing should only work if orig tx opted in""" """Replacing should only work if orig tx opted in"""
tx0_outpoint = make_utxo(self.nodes[0], int(1.1*COIN)) tx0_outpoint = make_utxo(self.nodes[0], int(1.1 * COIN))
# Create a non-opting in transaction # Create a non-opting in transaction
tx1a = CTransaction() tx1a = CTransaction()
@ -438,7 +439,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
# This will raise an exception # This will raise an exception
assert_raises_rpc_error(-26, "txn-mempool-conflict", self.nodes[0].sendrawtransaction, tx1b_hex, 0) assert_raises_rpc_error(-26, "txn-mempool-conflict", self.nodes[0].sendrawtransaction, tx1b_hex, 0)
tx1_outpoint = make_utxo(self.nodes[0], int(1.1*COIN)) tx1_outpoint = make_utxo(self.nodes[0], int(1.1 * COIN))
# Create a different non-opting in transaction # Create a different non-opting in transaction
tx2a = CTransaction() tx2a = CTransaction()
@ -466,7 +467,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
tx3a = CTransaction() tx3a = CTransaction()
tx3a.vin = [CTxIn(COutPoint(tx1a_txid, 0), nSequence=0xffffffff), tx3a.vin = [CTxIn(COutPoint(tx1a_txid, 0), nSequence=0xffffffff),
CTxIn(COutPoint(tx2a_txid, 0), nSequence=0xfffffffd)] CTxIn(COutPoint(tx2a_txid, 0), nSequence=0xfffffffd)]
tx3a.vout = [CTxOut(int(0.9*COIN), CScript([b'c'])), CTxOut(int(0.9*COIN), CScript([b'd']))] tx3a.vout = [CTxOut(int(0.9 * COIN), CScript([b'c'])), CTxOut(int(0.9 * COIN), CScript([b'd']))]
tx3a_hex = txToHex(tx3a) tx3a_hex = txToHex(tx3a)
tx3a_txid = self.nodes[0].sendrawtransaction(tx3a_hex, 0) tx3a_txid = self.nodes[0].sendrawtransaction(tx3a_hex, 0)
@ -494,7 +495,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
# correctly used by replacement logic # correctly used by replacement logic
# 1. Check that feeperkb uses modified fees # 1. Check that feeperkb uses modified fees
tx0_outpoint = make_utxo(self.nodes[0], int(1.1*COIN)) tx0_outpoint = make_utxo(self.nodes[0], int(1.1 * COIN))
tx1a = CTransaction() tx1a = CTransaction()
tx1a.vin = [CTxIn(tx0_outpoint, nSequence=0)] tx1a.vin = [CTxIn(tx0_outpoint, nSequence=0)]
@ -505,14 +506,14 @@ class ReplaceByFeeTest(BitcoinTestFramework):
# Higher fee, but the actual fee per KB is much lower. # Higher fee, but the actual fee per KB is much lower.
tx1b = CTransaction() tx1b = CTransaction()
tx1b.vin = [CTxIn(tx0_outpoint, nSequence=0)] tx1b.vin = [CTxIn(tx0_outpoint, nSequence=0)]
tx1b.vout = [CTxOut(int(0.001*COIN), CScript([b'a'*740000]))] tx1b.vout = [CTxOut(int(0.001 * COIN), CScript([b'a' * 740000]))]
tx1b_hex = txToHex(tx1b) tx1b_hex = txToHex(tx1b)
# Verify tx1b cannot replace tx1a. # Verify tx1b cannot replace tx1a.
assert_raises_rpc_error(-26, "insufficient fee", self.nodes[0].sendrawtransaction, tx1b_hex, 0) assert_raises_rpc_error(-26, "insufficient fee", self.nodes[0].sendrawtransaction, tx1b_hex, 0)
# Use prioritisetransaction to set tx1a's fee to 0. # Use prioritisetransaction to set tx1a's fee to 0.
self.nodes[0].prioritisetransaction(txid=tx1a_txid, fee_delta=int(-0.1*COIN)) self.nodes[0].prioritisetransaction(txid=tx1a_txid, fee_delta=int(-0.1 * COIN))
# Now tx1b should be able to replace tx1a # Now tx1b should be able to replace tx1a
tx1b_txid = self.nodes[0].sendrawtransaction(tx1b_hex, 0) tx1b_txid = self.nodes[0].sendrawtransaction(tx1b_hex, 0)
@ -520,7 +521,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
assert tx1b_txid in self.nodes[0].getrawmempool() assert tx1b_txid in self.nodes[0].getrawmempool()
# 2. Check that absolute fee checks use modified fee. # 2. Check that absolute fee checks use modified fee.
tx1_outpoint = make_utxo(self.nodes[0], int(1.1*COIN)) tx1_outpoint = make_utxo(self.nodes[0], int(1.1 * COIN))
tx2a = CTransaction() tx2a = CTransaction()
tx2a.vin = [CTxIn(tx1_outpoint, nSequence=0)] tx2a.vin = [CTxIn(tx1_outpoint, nSequence=0)]
@ -539,7 +540,7 @@ class ReplaceByFeeTest(BitcoinTestFramework):
assert_raises_rpc_error(-26, "insufficient fee", self.nodes[0].sendrawtransaction, tx2b_hex, 0) assert_raises_rpc_error(-26, "insufficient fee", self.nodes[0].sendrawtransaction, tx2b_hex, 0)
# Now prioritise tx2b to have a higher modified fee # Now prioritise tx2b to have a higher modified fee
self.nodes[0].prioritisetransaction(txid=tx2b.hash, fee_delta=int(0.1*COIN)) self.nodes[0].prioritisetransaction(txid=tx2b.hash, fee_delta=int(0.1 * COIN))
# tx2b should now be accepted # tx2b should now be accepted
tx2b_txid = self.nodes[0].sendrawtransaction(tx2b_hex, 0) tx2b_txid = self.nodes[0].sendrawtransaction(tx2b_hex, 0)
@ -549,11 +550,11 @@ class ReplaceByFeeTest(BitcoinTestFramework):
def test_rpc(self): def test_rpc(self):
us0 = self.nodes[0].listunspent()[0] us0 = self.nodes[0].listunspent()[0]
ins = [us0] ins = [us0]
outs = {self.nodes[0].getnewaddress() : Decimal(1.0000000)} outs = {self.nodes[0].getnewaddress(): Decimal(1.0000000)}
rawtx0 = self.nodes[0].createrawtransaction(ins, outs, 0, True) rawtx0 = self.nodes[0].createrawtransaction(ins, outs, 0, True)
rawtx1 = self.nodes[0].createrawtransaction(ins, outs, 0, False) rawtx1 = self.nodes[0].createrawtransaction(ins, outs, 0, False)
json0 = self.nodes[0].decoderawtransaction(rawtx0) json0 = self.nodes[0].decoderawtransaction(rawtx0)
json1 = self.nodes[0].decoderawtransaction(rawtx1) json1 = self.nodes[0].decoderawtransaction(rawtx1)
assert_equal(json0["vin"][0]["sequence"], 4294967293) assert_equal(json0["vin"][0]["sequence"], 4294967293)
assert_equal(json1["vin"][0]["sequence"], 4294967295) assert_equal(json1["vin"][0]["sequence"], 4294967295)
@ -561,8 +562,8 @@ class ReplaceByFeeTest(BitcoinTestFramework):
frawtx2a = self.nodes[0].fundrawtransaction(rawtx2, {"replaceable": True}) frawtx2a = self.nodes[0].fundrawtransaction(rawtx2, {"replaceable": True})
frawtx2b = self.nodes[0].fundrawtransaction(rawtx2, {"replaceable": False}) frawtx2b = self.nodes[0].fundrawtransaction(rawtx2, {"replaceable": False})
json0 = self.nodes[0].decoderawtransaction(frawtx2a['hex']) json0 = self.nodes[0].decoderawtransaction(frawtx2a['hex'])
json1 = self.nodes[0].decoderawtransaction(frawtx2b['hex']) json1 = self.nodes[0].decoderawtransaction(frawtx2b['hex'])
assert_equal(json0["vin"][0]["sequence"], 4294967293) assert_equal(json0["vin"][0]["sequence"], 4294967293)
assert_equal(json1["vin"][0]["sequence"], 4294967294) assert_equal(json1["vin"][0]["sequence"], 4294967294)
@ -623,5 +624,6 @@ class ReplaceByFeeTest(BitcoinTestFramework):
assert_equal(True, self.nodes[0].getmempoolentry(optin_parent_tx['txid'])['bip125-replaceable']) assert_equal(True, self.nodes[0].getmempoolentry(optin_parent_tx['txid'])['bip125-replaceable'])
assert_raises_rpc_error(-26, 'txn-mempool-conflict', self.nodes[0].sendrawtransaction, replacement_child_tx["hex"], 0) assert_raises_rpc_error(-26, 'txn-mempool-conflict', self.nodes[0].sendrawtransaction, replacement_child_tx["hex"], 0)
if __name__ == '__main__': if __name__ == '__main__':
ReplaceByFeeTest().main() ReplaceByFeeTest().main()