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

Code before:
    new_mi1 = mempool_manager.get_mempool_item(mi1.spend_bundle_name)
    assert new_mi1 is not None
    latest_singleton_lineage1 = new_mi1.bundle_coin_spends[version1_id].latest_singleton_lineage
    assert latest_singleton_lineage1 is not None
    assert latest_singleton_lineage1.coin_id == new_latest_lineage_id
    new_mi2 = mempool_manager.get_mempool_item(mi2.spend_bundle_name)
    assert new_mi2 is not None
    latest_singleton_lineage2 = new_mi2.bundle_coin_spends[version2_id].latest_singleton_lineage
    assert latest_singleton_lineage2 is not None
    assert latest_singleton_lineage2.coin_id == new_latest_lineage_id


@pytest.mark.anyio
@pytest.mark.parametrize("optimized_path", [True, False])
async def test_new_peak_txs_added(optimized_path: bool) -> None:
    """
    Covers the scenario where new peak returns added transactions as a result
    of retrying potential transactions.
    """
    coins = TestCoins([TEST_COIN], {})
    mempool_manager = await setup_mempool(coins)
    # Add an item that should go to the pending cache
    assert mempool_manager.peak is not None
    asserted_abs_height = mempool_manager.peak.height + 1
    sb, sb_name, result = await generate_and_add_spendbundle(
        mempool_manager, [[ConditionOpcode.ASSERT_HEIGHT_ABSOLUTE, asserted_abs_height]]
    )
    _, status, error = result
    assert status == MempoolInclusionStatus.PENDING
    assert error == Err.ASSERT_HEIGHT_ABSOLUTE_FAILED
    # Advance the mempool beyond the asserted height to retry the test item
    if optimized_path:
        spent_coins: Optional[list[bytes32]] = []
        new_peak_info = await mempool_manager.new_peak(
            create_test_block_record(height=uint32(asserted_abs_height)), spent_coins
        )
        # We're not there yet (needs to be higher, not equal)
        assert mempool_manager.get_mempool_item(sb_name) is None
        assert new_peak_info.items == []
    else:
        spent_coins = None
    new_peak_info = await mempool_manager.new_peak(
        create_test_block_record(height=uint32(asserted_abs_height + 1)), spent_coins
    )
    # The item gets retried successfully now
    mi = mempool_manager.get_mempool_item(sb_name)
    assert mi is not None
    assert new_peak_info.items == [NewPeakItem(sb_name, sb, mi.conds)]


Review comment (suggestion):
I think it would be great to paramaterize this test on the condition as well, to cover `ASSERT_HEIGHT_RELATIVE`, `ASSERT_SECONDS_ABSOLUTE` and `ASSERT_SECONDS_RELATIVE`

Code after:
    new_mi1 = mempool_manager.get_mempool_item(mi1.spend_bundle_name)
    assert new_mi1 is not None
    latest_singleton_lineage1 = new_mi1.bundle_coin_spends[version1_id].latest_singleton_lineage
    assert latest_singleton_lineage1 is not None
    assert latest_singleton_lineage1.coin_id == new_latest_lineage_id
    new_mi2 = mempool_manager.get_mempool_item(mi2.spend_bundle_name)
    assert new_mi2 is not None
    latest_singleton_lineage2 = new_mi2.bundle_coin_spends[version2_id].latest_singleton_lineage
    assert latest_singleton_lineage2 is not None
    assert latest_singleton_lineage2.coin_id == new_latest_lineage_id


@pytest.mark.anyio
@pytest.mark.parametrize(
    "condition_and_error",
    [
        (ConditionOpcode.ASSERT_HEIGHT_RELATIVE, Err.ASSERT_HEIGHT_RELATIVE_FAILED),
        (ConditionOpcode.ASSERT_HEIGHT_ABSOLUTE, Err.ASSERT_HEIGHT_ABSOLUTE_FAILED),
    ],
)
@pytest.mark.parametrize("optimized_path", [True, False])
async def test_new_peak_txs_added(condition_and_error: tuple[ConditionOpcode, Err], optimized_path: bool) -> None:
    """
    Tests that deferred transactions because of time-lock are retried once the
    time-lock allows them to be reconsidered.
    """
    coins = TestCoins([TEST_COIN], {})
    mempool_manager = await setup_mempool(coins)
    # Add an item that should go to the pending cache
    assert mempool_manager.peak is not None
    condition_height = mempool_manager.peak.height + 1
    condition, expected_error = condition_and_error
    sb, sb_name, result = await generate_and_add_spendbundle(mempool_manager, [[condition, condition_height]])
    _, status, error = result
    assert status == MempoolInclusionStatus.PENDING
    assert error == expected_error
    # Advance the mempool beyond the asserted height to retry the test item
    if optimized_path:
        spent_coins: Optional[list[bytes32]] = []
        new_peak_info = await mempool_manager.new_peak(
            create_test_block_record(height=uint32(condition_height)), spent_coins
        )
        # We're not there yet (needs to be higher, not equal)
        assert mempool_manager.get_mempool_item(sb_name, include_pending=False) is None
        assert new_peak_info.items == []
    else:
        spent_coins = None
    new_peak_info = await mempool_manager.new_peak(
        create_test_block_record(height=uint32(condition_height + 1)), spent_coins
    )
    # The item gets retried successfully now