Repository: Chia-Network/chia-blockchain
File: chia/_tests/core/full_node/test_full_node.py (Python)

Code before:
    assert block1.transactions_info == block2.transactions_info, "Mismatch in transactions_info"
    assert block1.transactions_generator == block2.transactions_generator, "Mismatch in transactions_generator"
    assert block1.transactions_generator_ref_list == block2.transactions_generator_ref_list

    # Final assertion to check the entire block
    assert block1 == block2, "The entire block objects are not identical"
    return True


@pytest.mark.anyio
@pytest.mark.parametrize(
    "condition_and_error",
    [
        (ConditionOpcode.ASSERT_HEIGHT_RELATIVE, "ASSERT_HEIGHT_RELATIVE_FAILED"),
        (ConditionOpcode.ASSERT_HEIGHT_ABSOLUTE, "ASSERT_HEIGHT_ABSOLUTE_FAILED"),
    ],
)
async def test_peak_post_processing_added_tx(
    condition_and_error: tuple[ConditionOpcode, str],
    blockchain_constants: ConsensusConstants,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """
    Tests peak post processing with regards to pending transactions due of
    time-lock to make sure they get retried once the time-lock allows them to
    be reconsidered.
    """
    async with setup_simulators_and_wallets(1, 0, blockchain_constants) as new:
        full_node_api = new.simulators[0].peer_api
        bt = new.bt
        wallet = WalletTool(test_constants)
        ph = wallet.get_new_puzzlehash()
        blocks = bt.get_consecutive_blocks(
            3, guarantee_transaction_block=True, farmer_reward_puzzle_hash=ph, pool_reward_puzzle_hash=ph
        )
        for block in blocks:
            await full_node_api.full_node.add_block(block)
        peak = full_node_api.full_node.blockchain.get_peak()
        assert peak is not None
        current_height = peak.height
        # Create a transaction with a height condition that makes it pending
        coin = blocks[-1].get_included_reward_coins()[0]
        condition, expected_error = condition_and_error
        if condition == ConditionOpcode.ASSERT_HEIGHT_RELATIVE:
            condition_height = 1
        else:
            condition_height = current_height + 1
        condition_dic = {condition: [ConditionWithArgs(condition, [int_to_bytes(condition_height)])]}
        sb = wallet.generate_signed_transaction(uint64(42), ph, coin, condition_dic)
        sb_name = sb.name()
        # Send the transaction

Review comment (performance):
I think this comment should more directly say what *feature* is being tested and maybe which *cases* are covered. It's the `PendingTXCache`, right?

Code after:
    assert block1.transactions_info == block2.transactions_info, "Mismatch in transactions_info"
    assert block1.transactions_generator == block2.transactions_generator, "Mismatch in transactions_generator"
    assert block1.transactions_generator_ref_list == block2.transactions_generator_ref_list

    # Final assertion to check the entire block
    assert block1 == block2, "The entire block objects are not identical"
    return True


@pytest.mark.anyio
@pytest.mark.parametrize(
    "condition, error",
    [
        (ConditionOpcode.ASSERT_HEIGHT_RELATIVE, "ASSERT_HEIGHT_RELATIVE_FAILED"),
        (ConditionOpcode.ASSERT_HEIGHT_ABSOLUTE, "ASSERT_HEIGHT_ABSOLUTE_FAILED"),
    ],
)
async def test_pending_tx_cache_retry_on_new_peak(
    condition: ConditionOpcode, error: str, blockchain_constants: ConsensusConstants, caplog: pytest.LogCaptureFixture
) -> None:
    """
    Covers PendingTXCache items that are placed there due to unmet relative or
    absolute height conditions, to make sure those items get retried at peak
    post processing when those conditions are met.
    """
    async with setup_simulators_and_wallets(1, 0, blockchain_constants) as new:
        full_node_api = new.simulators[0].peer_api
        bt = new.bt
        wallet = WalletTool(test_constants)
        ph = wallet.get_new_puzzlehash()
        blocks = bt.get_consecutive_blocks(
            3, guarantee_transaction_block=True, farmer_reward_puzzle_hash=ph, pool_reward_puzzle_hash=ph
        )
        for block in blocks:
            await full_node_api.full_node.add_block(block)
        peak = full_node_api.full_node.blockchain.get_peak()
        assert peak is not None
        current_height = peak.height
        # Create a transaction with a height condition that makes it pending
        coin = blocks[-1].get_included_reward_coins()[0]
        if condition == ConditionOpcode.ASSERT_HEIGHT_RELATIVE:
            condition_height = 1
        else:
            condition_height = current_height + 1
        condition_dic = {condition: [ConditionWithArgs(condition, [int_to_bytes(condition_height)])]}
        sb = wallet.generate_signed_transaction(uint64(42), ph, coin, condition_dic)
        sb_name = sb.name()
        # Send the transaction
        res = await full_node_api.send_transaction(SendTransaction(sb))
        assert res is not None
        assert ProtocolMessageTypes(res.type) == ProtocolMessageTypes.transaction_ack