import pytest from chia.types.peer_info import PeerInfo from chia.util.block_tools import BlockTools from chia.util.ints import uint16 from tests.core.node_height import node_height_at_least from tests.setup_nodes import self_hostname, setup_full_node, setup_full_system, test_constants from tests.time_out_assert import time_out_assert test_constants_modified = test_constants.replace( **{ "DIFFICULTY_STARTING": 2 ** 8, "DISCRIMINANT_SIZE_BITS": 1024, "SUB_EPOCH_BLOCKS": 140, "WEIGHT_PROOF_THRESHOLD": 2, "WEIGHT_PROOF_RECENT_BLOCKS": 350, "MAX_SUB_SLOT_BLOCKS": 50, "NUM_SPS_SUB_SLOT": 32, # Must be a power of 2 "EPOCH_BLOCKS": 280, "SUB_SLOT_ITERS_STARTING": 2 ** 20, "NUMBER_ZERO_BITS_PLOT_FILTER": 5, } ) class TestSimulation: @pytest.fixture(scope="function") async def extra_node(self): b_tools = BlockTools(constants=test_constants_modified) async for _ in setup_full_node(test_constants_modified, "blockchain_test_3.db", 21240, b_tools): yield _ @pytest.fixture(scope="function") async def simulation(self): async for _ in setup_full_system(test_constants_modified): yield _ @pytest.mark.asyncio async def test_simulation_1(self, simulation, extra_node): node1, node2, _, _, _, _, _, _, _, server1 = simulation await server1.start_client(PeerInfo(self_hostname, uint16(21238))) # Use node2 to test node communication, since only node1 extends the chain. await time_out_assert(1500, node_height_at_least, True, node2, 7) async def has_compact(node1, node2): peak_height_1 = node1.full_node.blockchain.get_peak_height() headers_1 = await node1.full_node.blockchain.get_header_blocks_in_range(0, peak_height_1) peak_height_2 = node2.full_node.blockchain.get_peak_height() headers_2 = await node2.full_node.blockchain.get_header_blocks_in_range(0, peak_height_2) # Commented to speed up. # cc_eos = [False, False] # icc_eos = [False, False] # cc_sp = [False, False] # cc_ip = [False, False] has_compact = [False, False] for index, headers in enumerate([headers_1, headers_2]): for header in headers.values(): for sub_slot in header.finished_sub_slots: if sub_slot.proofs.challenge_chain_slot_proof.normalized_to_identity: # cc_eos[index] = True has_compact[index] = True if ( sub_slot.proofs.infused_challenge_chain_slot_proof is not None and sub_slot.proofs.infused_challenge_chain_slot_proof.normalized_to_identity ): # icc_eos[index] = True has_compact[index] = True if ( header.challenge_chain_sp_proof is not None and header.challenge_chain_sp_proof.normalized_to_identity ): # cc_sp[index] = True has_compact[index] = True if header.challenge_chain_ip_proof.normalized_to_identity: # cc_ip[index] = True has_compact[index] = True # return ( # cc_eos == [True, True] and icc_eos == [True, True] and cc_sp == [True, True] and cc_ip == [True, True] # ) return has_compact == [True, True] await time_out_assert(1500, has_compact, True, node1, node2) node3 = extra_node server3 = node3.full_node.server peak_height = max(node1.full_node.blockchain.get_peak_height(), node2.full_node.blockchain.get_peak_height()) await server3.start_client(PeerInfo(self_hostname, uint16(21237))) await server3.start_client(PeerInfo(self_hostname, uint16(21238))) await time_out_assert(600, node_height_at_least, True, node3, peak_height)