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

Code before:
                self.log.debug("Wait for another, but time out, since the fee threshold is set now")
                template7 = await wait_next_template(template6, stack, ctx, waitoptions)
                assert template7 is None

                self.log.debug("interruptWait should abort the current wait")
                async def wait_for_block():
                    new_waitoptions = self.capnp_modules['mining'].BlockWaitOptions()
                    new_waitoptions.timeout = waitoptions.timeout * 60 # 1 minute wait
                    new_waitoptions.feeThreshold = 1
                    template7 = await wait_next_template(template6, stack, ctx, new_waitoptions)
                    assert template7 is None
                await wait_and_do(wait_for_block(), template6.interruptWait())

                self.log.debug("Use absurdly large reserved weight to force an empty template")
                opts.blockReservedWeight = MAX_BLOCK_WEIGHT
                empty_template = await create_block_template(mining, stack, ctx, opts)
                assert empty_template is not None
                block = await self.parse_and_deserialize_block(empty_template, ctx)
                assert_equal(len(block.vtx), 1)

                self.log.debug("Enforce minimum reserved weight for IPC clients too")
                opts.blockReservedWeight = 0
                try:
                    await mining.createNewBlock(opts)
                    raise AssertionError("createNewBlock unexpectedly succeeded")
                except Exception as e:
                    assert "must be at least" in str(e)

                # Restore opts
                opts.blockReservedWeight = 4000

            current_block_height = self.nodes[0].getchaintips()[0]["height"]
            check_opts = self.capnp_modules['mining'].BlockCheckOptions()
            async with destroying((await mining.createNewBlock(opts)).result, ctx) as template:
                block = await self.parse_and_deserialize_block(template, ctx)
                coinbase = await self.parse_and_deserialize_coinbase_tx(template, ctx)
                balance = miniwallet.get_balance()
                coinbase.vout[0].scriptPubKey = miniwallet.get_output_script()
                coinbase.vout[0].nValue = COIN
                block.vtx[0] = coinbase
                block.hashMerkleRoot = block.calc_merkle_root()
                original_version = block.nVersion
                self.log.debug("Submit a block with a bad version")
                block.nVersion = 0
                block.solve()
                check = await mining.checkBlock(block.serialize(), check_opts)
                assert_equal(check.result, False)
                assert_equal(check.reason, "bad-version(0x00000000)")
                submitted = (await template.submitSolution(ctx, block.nVersion, block.nTime, block.nNonce, coinbase.serialize())).result
                assert_equal(submitted, False)
                self.log.debug("Submit a valid block")

Review comment (bug):
In commit "mining: enforce minimum reserved weight for IPC" (d3db5285989f7b3fef7436a37ce928065be2804e)

Might be nice to make exception more specific to expected error like:

```diff
--- a/test/functional/interface_ipc.py
+++ b/test/functional/interface_ipc.py
@@ -278,8 +278,9 @@ class IPCInterfaceTest(BitcoinTestFramework):
                 try:
                     await mining.createNewBlock(opts)
                     raise AssertionError("createNewBlock unexpectedly succeeded")
-                except Exception as e:
-                    assert "must be at least" in str(e)
+                except capnp.lib.capnp.KjException as e:
+                    assert_equal(e.description, "remote exception: std::exception: block_reserved_weight (0) must be at least 2000 weight units")
+                    assert_equal(e.type, "FAILED")
 
                 # Restore opts
                 opts.blockReservedWeight = 4000
```


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_test(self):
        self.run_echo_test()
        self.run_mining_test()

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