test assert time passed conditions

tests/test_blockchain_transactions.py

@@ -1,4 +1,5 @@
 import asyncio
+import time
 from typing import Optional
 import pytest
 from clvm.casts import int_to_bytes
@@ -11,6 +12,7 @@ from src.protocols import full_node_protocol
 from src.types.full_block import FullBlock
 from src.types.hashable.SpendBundle import SpendBundle
 from src.util.ConsensusError import Err
+from src.util.ints import uint64
 from tests.setup_nodes import setup_two_nodes, test_constants, bt
 from tests.wallet_tools import WalletTool
 
@@ -481,4 +483,75 @@ class TestBlockchainTransactions:
             next_block, next_block.body.fees_coin.amount
         )
 
+        assert error is None
+
+    @pytest.mark.asyncio
+    async def test_assert_time_exceeds(self, two_nodes):
+
+        num_blocks = 10
+        wallet_a = WalletTool()
+        coinbase_puzzlehash = wallet_a.get_new_puzzlehash()
+        wallet_receiver = WalletTool()
+        receiver_puzzlehash = wallet_receiver.get_new_puzzlehash()
+
+        # Farm blocks
+        blocks = bt.get_consecutive_blocks(
+            test_constants, num_blocks, [], 10, b"", coinbase_puzzlehash
+        )
+        full_node_1, full_node_2, server_1, server_2 = two_nodes
+
+        for block in blocks:
+            async for _ in full_node_1.block(full_node_protocol.Block(block)):
+                pass
+
+        # Coinbase that gets spent
+        block1 = blocks[1]
+
+        # This condition requires block1 coinbase to be spent after 3 seconds from now
+        current_time_plus3 = uint64(int(time.time() * 1000) + 3000)
+        block1_cvp = ConditionVarPair(
+            ConditionOpcode.ASSERT_TIME_EXCEEDS, int_to_bytes(current_time_plus3), None
+        )
+        block1_dic = {block1_cvp.opcode: [block1_cvp]}
+        block1_spend_bundle = wallet_a.generate_signed_transaction(
+            1000, receiver_puzzlehash, block1.body.coinbase, block1_dic
+        )
+
+        # program that will be sent to early
+        program = best_solution_program(block1_spend_bundle)
+        aggsig = block1_spend_bundle.aggregated_signature
+
+        # Create another block that includes our transaction
+        dic_h = {11: (program, aggsig)}
+        invalid_new_blocks = bt.get_consecutive_blocks(
+            test_constants, 1, blocks, 10, b"", coinbase_puzzlehash, dic_h
+        )
+
+        # Try to validate that block before 3 sec
+        next_block = invalid_new_blocks[11]
+        error = await full_node_1.blockchain.validate_transactions(
+            next_block, next_block.body.fees_coin.amount
+        )
+
+        assert error is Err.ASSERT_TIME_EXCEEDS_FAILED
+
+        # wait 3 sec to pass
+        await asyncio.sleep(3.1)
+
+        dic_h = {12: (program, aggsig)}
+        valid_new_blocks = bt.get_consecutive_blocks(
+            test_constants, 2, blocks[:11], 10, b"", coinbase_puzzlehash, dic_h
+        )
+
+        for block in valid_new_blocks:
+            async for _ in full_node_1.block(full_node_protocol.Block(block)):
+                pass
+
+        # Try to validate that block after 3 sec have passed
+        next_block = valid_new_blocks[12]
+
+        error = await full_node_1.blockchain.validate_transactions(
+            next_block, next_block.body.fees_coin.amount
+        )
+
         assert error is None