Repository: bitcoin/bitcoin
File: test/functional/interface_ipc.py (Python)

Code before:
            assert_equal(newblockref.height, current_block_height + 1)
            self.log.debug("Wait for timeout")
            oldblockref = (await mining.waitTipChanged(ctx, newblockref.hash, timeout)).result
            assert_equal(len(newblockref.hash), block_hash_size)
            assert_equal(oldblockref.hash, newblockref.hash)
            assert_equal(oldblockref.height, newblockref.height)

            async with AsyncExitStack() as stack:
                opts = self.capnp_modules['mining'].BlockCreateOptions()
                opts.useMempool = True
                opts.blockReservedWeight = 4000
                opts.coinbaseOutputMaxAdditionalSigops = 0

                self.log.debug("createNewBlock() should wait if tip is still updating")
                self.disconnect_nodes(0, 1)
                node1_block_hash = self.generate(self.nodes[1], 1, sync_fun=self.no_op)[0]
                header = from_hex(CBlockHeader(), self.nodes[1].getblockheader(node1_block_hash, False))
                header_only_peer = self.nodes[0].add_p2p_connection(P2PInterface())
                header_only_peer.send_and_ping(msg_headers([header]))
                start = time.time()
                async with destroying((await mining.createNewBlock(opts)).result, ctx):
                    pass
                # Lower-bound only: a heavily loaded CI host might still exceed 0.9s
                # even without the cooldown, so this can miss regressions but avoids
                # spurious failures.
                assert_greater_than_or_equal(time.time() - start, 0.9)
                header_only_peer.peer_disconnect()
                self.connect_nodes(0, 1)
                self.sync_all()

                self.log.debug("Create a template")
                template = await create_block_template(mining, stack, ctx, opts)

                self.log.debug("Test some inspectors of Template")
                header = (await template.getBlockHeader(ctx)).result
                assert_equal(len(header), block_header_size)
                block = await self.parse_and_deserialize_block(template, ctx)
                current_tip = self.nodes[0].getbestblockhash()
                assert_equal(ser_uint256(block.hashPrevBlock), ser_uint256(int(current_tip, 16)))
                assert_greater_than_or_equal(len(block.vtx), 1)
                txfees = await template.getTxFees(ctx)
                assert_equal(len(txfees.result), 0)
                txsigops = await template.getTxSigops(ctx)
                assert_equal(len(txsigops.result), 0)
                coinbase_data = BytesIO((await template.getCoinbaseRawTx(ctx)).result)
                coinbase = CTransaction()
                coinbase.deserialize(coinbase_data)
                assert_equal(coinbase.vin[0].prevout.hash, 0)

                self.log.debug("Wait for a new template")
                waitoptions = self.capnp_modules['mining'].BlockWaitOptions()

Review comment (performance):
In commit "mining: createNewBlock() waits if tip is updating" (9817657d463ff1d8502a5edf0641e88b9b1d4853)

Since this check uses a delay, if there is a way you could move this test to a separate test function (not `run_mining_test`) i think that would be very good.

I'm concerned about interface_ipc test getting harder to maintain and slower to run, and would like to move towards using separate test methods like #34284, so we can still have fast update/test cycles by being able to run specific tests.

Code after:
#!/usr/bin/env python3
# Copyright (c) The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test the IPC (multiprocess) interface."""
import asyncio

from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal
from test_framework.ipc_util import (
    load_capnp_modules,
    make_capnp_init_ctx,
)

# Test may be skipped and not have capnp installed
try:
    import capnp  # type: ignore[import] # noqa: F401
except ModuleNotFoundError:
    pass


class IPCInterfaceTest(BitcoinTestFramework):

    def skip_test_if_missing_module(self):
        self.skip_if_no_ipc()
        self.skip_if_no_py_capnp()

    def set_test_params(self):
        self.num_nodes = 1

    def setup_nodes(self):
        self.extra_init = [{"ipcbind": True}]
        super().setup_nodes()
        # Use this function to also load the capnp modules (we cannot use set_test_params for this,
        # as it is being called before knowing whether capnp is available).
        self.capnp_modules = load_capnp_modules(self.config)

    def run_echo_test(self):
        self.log.info("Running echo test")
        async def async_routine():
            ctx, init = await make_capnp_init_ctx(self)
            self.log.debug("Create Echo proxy object")
            echo = init.makeEcho(ctx).result
            self.log.debug("Test a few invocations of echo")
            for s in ["hallo", "", "haha"]:
                result_eval = (await echo.echo(ctx, s)).result
                assert_equal(s, result_eval)
            self.log.debug("Destroy the Echo object")
            echo.destroy(ctx)
        asyncio.run(capnp.run(async_routine()))

    def run_mining_test(self):
        self.log.info("Running mining test")
        block_hash_size = 32

        async def async_routine():
            ctx, init = await make_capnp_init_ctx(self)
            self.log.debug("Create Mining proxy object")
            mining = init.makeMining(ctx).result
            self.log.debug("Test simple inspectors")
            assert (await mining.isTestChain(ctx)).result
            assert not (await mining.isInitialBlockDownload(ctx)).result
            blockref = await mining.getTip(ctx)
            assert blockref.hasResult
            assert_equal(len(blockref.result.hash), block_hash_size)
            current_block_height = self.nodes[0].getchaintips()[0]["height"]
            assert_equal(blockref.result.height, current_block_height)

        asyncio.run(capnp.run(async_routine()))

    def run_deprecated_mining_test(self):
        self.log.info("Running deprecated mining interface test")
        async def async_routine():
            ctx, init = await make_capnp_init_ctx(self)
            self.log.debug("Calling deprecated makeMiningOld2 should raise an error")
            try:
                await init.makeMiningOld2()
                raise AssertionError("makeMiningOld2 unexpectedly succeeded")
            except capnp.KjException as e:
                assert_equal(e.description, "remote exception: std::exception: Old mining interface (@2) not supported. Please update your client!")
                assert_equal(e.type, "FAILED")
        asyncio.run(capnp.run(async_routine()))

    def run_test(self):
        self.run_echo_test()
        self.run_mining_test()
        self.run_deprecated_mining_test()

if __name__ == '__main__':
    IPCInterfaceTest(__file__).main()