tx_validation_test.py

#!/usr/bin/env python3#!/usr/bin/env python3
#
# Distributed under the MIT/X11 software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
#

from test_framework.test_framework import ComparisonTestFramework
from test_framework.util import *
from test_framework.comptool import TestManager, TestInstance
from test_framework.mininode import *
from test_framework.blocktools import *
import logging
import copy
import time

'''
In this test we connect to one node over p2p, and test tx validation:
1) Add a tx to the mempool.
2) Use invalidateblock to reorg to a height below the tx's valid spend height (coinbase maturity).
3) Submit an (invalid) block that mines the tx.
'''

# Use the ComparisonTestFramework with 1 node: only use --testbinary.
class TxValidationCacheTest(ComparisonTestFramework):

    ''' Can either run this test as 1 node with expected answers, or two and compare them. 
        Change the "outcome" variable from each TestInstance object to only do the comparison. '''
    def __init__(self):
        self.num_nodes = 1

    def run_test(self):
        test = TestManager(self, self.options.tmpdir)
        test.add_all_connections(self.nodes)
        self.tip = None
        self.block_time = None
        NetworkThread().start() # Start up network handling in another thread
        test.run()

    def get_tests(self):
        if self.tip is None:
            self.tip = int ("0x" + self.nodes[0].getbestblockhash() + "L", 0)
        self.block_time = int(time.time())+1

        '''
        Create a new block with an anyone-can-spend coinbase
        '''
        block = create_block(self.tip, create_coinbase(), self.block_time)
        self.block_time += 1
        block.solve()
        # Save the coinbase for later
        self.block1 = block
        self.tip = block.sha256
        yield TestInstance([[block, True]])

        '''
        Now we need that block to mature so we can spend the coinbase.
        '''
        test = TestInstance(sync_every_block=False)
        for i in xrange(100):
            block = create_block(self.tip, create_coinbase(), self.block_time)
            block.solve()
            self.tip = block.sha256
            self.block_time += 1
            test.blocks_and_transactions.append([block, True])
        yield test

        '''
        Now construct a transaction that spends the original coinbase.
        '''
        tx1 = create_transaction(self.block1.vtx[0], 0, chr(81), 50*100000000)

        test = TestInstance(sync_every_block=True)
        test.blocks_and_transactions.append([tx1, True])
        yield test

        '''
        Now check to see what happens if we reorg back to an earlier height
        and try to mine a block with this tx (at too early a height)
        '''
        print "current block height: ", self.nodes[1].getblockcount()
        self.nodes[0].invalidateblock(self.nodes[0].getblockhash(self.nodes[0].getblockcount()-50))
        self.tip = int("0x" + self.nodes[0].getbestblockhash() + "L", 0)
        block = create_block(self.tip, create_coinbase(), self.block_time)
        block.vtx.extend([tx1])
        block.hashMerkleRoot = block.calc_merkle_root()
        block.rehash()
        block.solve()
        yield TestInstance([[block, False]])


if __name__ == '__main__':
    TxValidationCacheTest().main()