Add more assertions to both PoS and PoW functional tests

Author: alfiedotwtf

test/functional/blockprod_generate_pos_blocks.py

@@ -15,13 +15,13 @@
 #  limitations under the License.
 
 from hashlib import blake2b
+from scalecodec.base import ScaleBytes, RuntimeConfiguration, ScaleDecoder
 from test_framework.mintlayer import mintlayer_hash
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
 )
-
-import random, scalecodec
+import random
 
 GENESIS_POOL_ID = "123c4c600097c513e088b9be62069f0c74c7671c523c8e3469a1c3f14b7ea2c4"
 GENESIS_STAKE_PRIVATE_KEY = "8717e6946febd3a33ccdc3f3a27629ec80c33461c33a0fc56b4836fcedd26638"
@@ -33,52 +33,50 @@
     "e9f3cd7b78d45750afd17292031373fddb5e7a8090db51221038f5e05f29998e"
 )
 
-block_input_data_obj = scalecodec.base.RuntimeConfiguration().create_scale_object('GenerateBlockInputData')
-outpoint_obj = scalecodec.base.RuntimeConfiguration().create_scale_object('OutPoint')
-signed_transaction_obj = scalecodec.base.RuntimeConfiguration().create_scale_object('SignedTransaction')
-transaction_obj = scalecodec.base.RuntimeConfiguration().create_scale_object('TransactionV1')
+block_input_data_obj = RuntimeConfiguration().create_scale_object('GenerateBlockInputData')
+outpoint_obj = RuntimeConfiguration().create_scale_object('OutPoint')
+signed_transaction_obj = RuntimeConfiguration().create_scale_object('SignedTransaction')
+transaction_obj = RuntimeConfiguration().create_scale_object('TransactionV1')
 
 class GeneratePoSBlocksTest(BitcoinTestFramework):
     def set_test_params(self):
         self.setup_clean_chain = True
         self.num_nodes = 1
         self.extra_args = [["--chain-pos-netupgrades=true"]]
 
-    def block_height(self, n):
-        tip = self.nodes[n].chainstate_best_block_id()
-        return self.nodes[n].chainstate_block_height_in_main_chain(tip)
-
-    def assert_tip(self, expected):
-        tip = self.nodes[0].chainstate_best_block_id()
-        block = self.nodes[0].chainstate_get_block(tip)
-        assert_equal(block, expected)
+    def assert_chain(self, block, previous_tip):
+        assert_equal(block["header"]["header"]["prev_block_id"][2:], previous_tip)
 
-    def hex_to_dec_array(self, hex_string):
-        return [int(hex_string[i:i+2], 16) for i in range(0, len(hex_string), 2)]
+    def assert_height(self, expected_height, expected_block):
+        block_id = self.nodes[0].chainstate_block_id_at_height(expected_height)
+        block = self.nodes[0].chainstate_get_block(block_id)
+        assert_equal(block, expected_block)
 
-    def generate_block(self, block_input_data, transactions):
-        old_block_height = self.block_height(0)
+    def assert_pos_consensus(self, block):
+        if block["header"]["header"]["consensus_data"].get("PoS") is None:
+            raise AssertionError("Block {} was not PoS".format(block_hex))
 
-        block = self.nodes[0].blockprod_generate_block(block_input_data, transactions)
-        self.nodes[0].chainstate_submit_block(block)
+    def assert_tip(self, expected_block):
+        tip = self.nodes[0].chainstate_best_block_id()
+        block = self.nodes[0].chainstate_get_block(tip)
+        assert_equal(block, expected_block)
 
-        assert_equal(self.block_height(0), old_block_height + 1)
-        self.assert_tip(block)
+    def block_height(self, n):
+        tip = self.nodes[n].chainstate_best_block_id()
+        return self.nodes[n].chainstate_block_height_in_main_chain(tip)
 
-    def previous_block_id(self):
+    def generate_block(self, expected_height, block_input_data, transactions):
         previous_block_id = self.nodes[0].chainstate_best_block_id()
-        return self.hex_to_dec_array(previous_block_id)
+        block_hex = self.nodes[0].blockprod_generate_block(block_input_data, transactions)
+        block_hex_array = bytearray.fromhex(block_hex)
+        block = ScaleDecoder.get_decoder_class('BlockV1', ScaleBytes(block_hex_array)).decode()
 
-    def pack_transaction(self, transaction):
-        transaction_encoded = signed_transaction_obj.encode(transaction).to_hex()[2:]
-        transaction_id = scalecodec.ScaleBytes(
-            mintlayer_hash(transaction_obj.encode(transaction["transaction"]).data)
-        ).to_hex()[2:]
+        self.nodes[0].chainstate_submit_block(block_hex)
 
-        return (transaction_encoded, transaction_id)
-
-    def genesis_pool_id(self):
-        return self.hex_to_dec_array(GENESIS_POOL_ID)
+        self.assert_tip(block_hex)
+        self.assert_height(expected_height, block_hex)
+        self.assert_pos_consensus(block)
+        self.assert_chain(block, previous_block_id)
 
     def generate_pool_id(self, transaction_id):
         kernel_input_outpoint = outpoint_obj.encode({
@@ -88,8 +86,6 @@ def generate_pool_id(self, transaction_id):
             "index": 0,
         }).to_hex()[2:]
 
-        outpoint = self.hex_to_dec_array(kernel_input_outpoint)
-
         # Include PoolId pre-image suffix of [0, 0, 0, 0]
         blake2b_hasher = blake2b()
         blake2b_hasher.update(bytes.fromhex(kernel_input_outpoint))
@@ -98,6 +94,42 @@ def generate_pool_id(self, transaction_id):
         # Truncate output to match Rust's split()
         return self.hex_to_dec_array(blake2b_hasher.hexdigest()[:64])
 
+    def genesis_pool_id(self):
+        return self.hex_to_dec_array(GENESIS_POOL_ID)
+
+    def hex_to_dec_array(self, hex_string):
+        return [int(hex_string[i:i+2], 16) for i in range(0, len(hex_string), 2)]
+
+    def new_stake_keys(self):
+        new_stake_private_key_hex = self.nodes[0].test_functions_new_private_key()
+        new_stake_private_key = self.stake_private_key(new_stake_private_key_hex[2:])
+
+        new_stake_public_key_hex = self.nodes[0].test_functions_public_key_from_private_key(new_stake_private_key_hex)
+        new_stake_public_key = self.stake_public_key(new_stake_public_key_hex[2:])
+
+        return (new_stake_private_key, new_stake_public_key)
+
+    def new_vrf_keys(self):
+        new_vrf_private_key_hex = self.nodes[0].test_functions_new_vrf_private_key()
+        new_vrf_private_key = self.vrf_private_key(new_vrf_private_key_hex[2:])
+
+        new_vrf_public_key_hex = self.nodes[0].test_functions_vrf_public_key_from_private_key(new_vrf_private_key_hex)
+        new_vrf_public_key = self.vrf_public_key(new_vrf_public_key_hex[2:])
+
+        return (new_vrf_private_key, new_vrf_public_key)
+
+    def pack_transaction(self, transaction):
+        transaction_encoded = signed_transaction_obj.encode(transaction).to_hex()[2:]
+        transaction_id = ScaleBytes(
+            mintlayer_hash(transaction_obj.encode(transaction["transaction"]).data)
+        ).to_hex()[2:]
+
+        return (transaction_encoded, transaction_id)
+
+    def previous_block_id(self):
+        previous_block_id = self.nodes[0].chainstate_best_block_id()
+        return self.hex_to_dec_array(previous_block_id)
+
     def stake_private_key(self, stake_private_key):
         return {
             "key": {
@@ -134,24 +166,6 @@ def vrf_public_key(self, vrf_public_key):
             },
         }
 
-    def new_stake_keys(self):
-        new_stake_private_key_hex = self.nodes[0].test_functions_new_private_key()
-        new_stake_private_key = self.stake_private_key(new_stake_private_key_hex[2:])
-
-        new_stake_public_key_hex = self.nodes[0].test_functions_public_key_from_private_key(new_stake_private_key_hex)
-        new_stake_public_key = self.stake_public_key(new_stake_public_key_hex[2:])
-
-        return (new_stake_private_key, new_stake_public_key)
-
-    def new_vrf_keys(self):
-        new_vrf_private_key_hex = self.nodes[0].test_functions_new_vrf_private_key()
-        new_vrf_private_key = self.vrf_private_key(new_vrf_private_key_hex[2:])
-
-        new_vrf_public_key_hex = self.nodes[0].test_functions_vrf_public_key_from_private_key(new_vrf_private_key_hex)
-        new_vrf_public_key = self.vrf_public_key(new_vrf_public_key_hex[2:])
-
-        return (new_vrf_private_key, new_vrf_public_key)
-
     def run_test(self):
         #
         # Transfer Genesis UTXO to AnyoneCanSpend
@@ -227,7 +241,7 @@ def run_test(self):
         }
 
         (transfer_transaction_encoded, transfer_transaction_id) = self.pack_transaction(transfer_transaction)
-        self.generate_block(block_input_data, [transfer_transaction_encoded])
+        self.generate_block(1, block_input_data, [transfer_transaction_encoded])
 
         #
         # Create the new stake pool using the AnyoneCanSpend transaction
@@ -306,7 +320,7 @@ def run_test(self):
         }
 
         (create_new_pool_transaction_encoded, create_new_pool_transaction_id) = self.pack_transaction(create_new_pool_transaction)
-        self.generate_block(block_input_data, [create_new_pool_transaction_encoded])
+        self.generate_block(2, block_input_data, [create_new_pool_transaction_encoded])
 
         #
         # Stake many blocks with the new stake pool
@@ -335,7 +349,7 @@ def run_test(self):
             ],
         }
 
-        for _ in range(1, random.randint(1,100)):
+        for i in range(0, random.randint(10,100)):
             block_input_data = block_input_data_obj.encode({
                 "PoS": {
                 "stake_private_key": new_stake_private_key,
@@ -350,7 +364,7 @@ def run_test(self):
                 }
             }).to_hex()[2:]
 
-            self.generate_block(block_input_data, [])
+            self.generate_block(3 + i, block_input_data, [])
 
             kernel_input_outpoint = {
                 "id": {

test/functional/blockprod_generate_pow_blocks.py

@@ -14,6 +14,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+from scalecodec.base import ScaleBytes, RuntimeConfiguration, ScaleDecoder
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
@@ -28,23 +29,42 @@ def set_test_params(self):
         self.setup_clean_chain = True
         self.num_nodes = 1
 
-    def block_height(self, n):
-        tip = self.nodes[n].chainstate_best_block_id()
-        return self.nodes[n].chainstate_block_height_in_main_chain(tip)
+    def assert_chain(self, block, previous_tip):
+        assert_equal(block["header"]["header"]["prev_block_id"][2:], previous_tip)
+
+    def assert_height(self, expected_height, expected_block):
+        block_id = self.nodes[0].chainstate_block_id_at_height(expected_height)
+        block = self.nodes[0].chainstate_get_block(block_id)
+        assert_equal(block, expected_block)
+
+    def assert_pow_consensus(self, block):
+        if block["header"]["header"]["consensus_data"].get("PoW") is None:
+            raise AssertionError("Block {} was not PoS".format(block_hex))
 
     def assert_tip(self, expected):
         tip = self.nodes[0].chainstate_best_block_id()
         block = self.nodes[0].chainstate_get_block(tip)
         assert_equal(block, expected)
 
-    def generate_block(self, block_input_data, transactions):
-        old_block_height = self.block_height(0)
+    def block_height(self, n):
+        tip = self.nodes[n].chainstate_best_block_id()
+        return self.nodes[n].chainstate_block_height_in_main_chain(tip)
+
+    def generate_block(self, expected_height, block_input_data, transactions):
+        previous_block_id = self.nodes[0].chainstate_best_block_id()
+        block_hex = self.nodes[0].blockprod_generate_block(block_input_data, transactions)
+        block_hex_array = bytearray.fromhex(block_hex)
+        block = ScaleDecoder.get_decoder_class('BlockV1', ScaleBytes(block_hex_array)).decode()
+
+        self.nodes[0].chainstate_submit_block(block_hex)
 
-        block = self.nodes[0].blockprod_generate_block(block_input_data, transactions)
-        self.nodes[0].chainstate_submit_block(block)
+        self.assert_tip(block_hex)
+        self.assert_height(expected_height, block_hex)
+        self.assert_pow_consensus(block)
+        self.assert_chain(block, previous_block_id)
 
-        assert_equal(self.block_height(0), old_block_height + 1)
-        self.assert_tip(block)
+    def hex_to_dec_array(self, hex_string):
+        return [int(hex_string[i:i+2], 16) for i in range(0, len(hex_string), 2)]
 
     def run_test(self):
         block_input_data = block_input_data_obj.encode(
@@ -55,8 +75,8 @@ def run_test(self):
             }
         ).to_hex()[2:]
 
-        for i in range(random.randint(100, 1000)):
-            self.nodes[0].blockprod_generate_block(block_input_data, [])
+        for i in range(random.randint(10, 100)):
+            self.generate_block(i + 1, block_input_data, [])
 
 if __name__ == '__main__':
     GeneratePoWBlocksTest().main()