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

Code before:
        # 10000 that's 300000
        (make_test_conds(seconds_absolute=20000, seconds_relative=20000), expect(seconds=30000)),
        # Same thing but without the absolute seconds
        (make_test_conds(seconds_relative=20000), expect(seconds=30000)),
    ],
)
def test_compute_assert_height(conds: SpendBundleConditions, expected: TimelockConditions) -> None:
    coin_id = TEST_COIN.name()

    confirmed_height = uint32(12)
    coin_records = {coin_id: CoinRecord(TEST_COIN, confirmed_height, uint32(0), False, uint64(10000))}

    assert compute_assert_height(coin_records, conds) == expected


def spend_bundle_from_conditions(
    conditions: list[list[Any]], coin: Coin = TEST_COIN, aggsig: G2Element = G2Element()
) -> SpendBundle:
    solution = SerializedProgram.to(conditions)
    coin_spend = make_spend(coin, IDENTITY_PUZZLE, solution)
    return SpendBundle([coin_spend], aggsig)


async def add_spendbundle(
    mempool_manager: MempoolManager, sb: SpendBundle, sb_name: bytes32
) -> tuple[uint64, MempoolInclusionStatus, Optional[Err]]:
    sbc = await mempool_manager.pre_validate_spendbundle(sb, sb_name)
    ret = await mempool_manager.add_spend_bundle(sb, sbc, sb_name, TEST_HEIGHT)
    invariant_check_mempool(mempool_manager.mempool)
    return ret.cost, ret.status, ret.error


async def generate_and_add_spendbundle(
    mempool_manager: MempoolManager,
    conditions: list[list[Any]],
    coin: Coin = TEST_COIN,
    aggsig: G2Element = G2Element(),
) -> tuple[SpendBundle, bytes32, tuple[uint64, MempoolInclusionStatus, Optional[Err]]]:
    sb = spend_bundle_from_conditions(conditions, coin, aggsig)
    sb_name = sb.name()
    result = await add_spendbundle(mempool_manager, sb, sb_name)
    return (sb, sb_name, result)


def make_bundle_spends_map_and_fee(
    spend_bundle: SpendBundle, conds: SpendBundleConditions
) -> tuple[dict[bytes32, BundleCoinSpend], uint64]:
    bundle_coin_spends: dict[bytes32, BundleCoinSpend] = {}

    spend_conditions = {bytes32(spend.coin_id): spend for spend in conds.spends}


Review comment (bug):
This is unrelated to the patch, it resulted from an unrelated change below to `add_spendbundle`'s failure path. I've addressed this as well in the addendum.

Code after:
        # 10000 that's 300000
        (make_test_conds(seconds_absolute=20000, seconds_relative=20000), expect(seconds=30000)),
        # Same thing but without the absolute seconds
        (make_test_conds(seconds_relative=20000), expect(seconds=30000)),
    ],
)
def test_compute_assert_height(conds: SpendBundleConditions, expected: TimelockConditions) -> None:
    coin_id = TEST_COIN.name()

    confirmed_height = uint32(12)
    coin_records = {coin_id: CoinRecord(TEST_COIN, confirmed_height, uint32(0), False, uint64(10000))}

    assert compute_assert_height(coin_records, conds) == expected


def spend_bundle_from_conditions(
    conditions: list[list[Any]], coin: Coin = TEST_COIN, aggsig: G2Element = G2Element()
) -> SpendBundle:
    solution = SerializedProgram.to(conditions)
    coin_spend = make_spend(coin, IDENTITY_PUZZLE, solution)
    return SpendBundle([coin_spend], aggsig)


async def add_spendbundle(
    mempool_manager: MempoolManager, sb: SpendBundle, sb_name: bytes32
) -> tuple[Optional[uint64], MempoolInclusionStatus, Optional[Err]]:
    sbc = await mempool_manager.pre_validate_spendbundle(sb, sb_name)
    ret = await mempool_manager.add_spend_bundle(sb, sbc, sb_name, TEST_HEIGHT)
    invariant_check_mempool(mempool_manager.mempool)
    return ret.cost, ret.status, ret.error


async def generate_and_add_spendbundle(
    mempool_manager: MempoolManager,
    conditions: list[list[Any]],
    coin: Coin = TEST_COIN,
    aggsig: G2Element = G2Element(),
) -> tuple[SpendBundle, bytes32, tuple[Optional[uint64], MempoolInclusionStatus, Optional[Err]]]:
    sb = spend_bundle_from_conditions(conditions, coin, aggsig)
    sb_name = sb.name()
    result = await add_spendbundle(mempool_manager, sb, sb_name)
    return (sb, sb_name, result)


def make_bundle_spends_map_and_fee(
    spend_bundle: SpendBundle, conds: SpendBundleConditions
) -> tuple[dict[bytes32, BundleCoinSpend], uint64]:
    bundle_coin_spends: dict[bytes32, BundleCoinSpend] = {}

    spend_conditions = {bytes32(spend.coin_id): spend for spend in conds.spends}