Method test_basic

test/functional/interface_zmq.py:184–262

(self, unix = False)

        return subscribers

    def test_basic(self, unix = False):
        self.log.info(f"Running basic test with {'ipc' if unix else 'tcp'} protocol")

        # Invalid zmq arguments don't take down the node, see #17185.
        self.restart_node(0, ["-zmqpubrawtx=foo", "-zmqpubhashtx=bar"])

        address = f"tcp://127.0.0.1:{self.zmq_port_base}"

        if unix:
            # Use the shortest temp path possible since paths may have as little as 92-char limit
            socket_path = tempfile.NamedTemporaryFile().name
            address = f"ipc://{socket_path}"

        subs = self.setup_zmq_test([(topic, address) for topic in ["hashblock", "hashtx", "rawblock", "rawtx"]])

        hashblock = subs[0]
        hashtx = subs[1]
        rawblock = subs[2]
        rawtx = subs[3]

        num_blocks = 5
        self.log.info(f"Generate {num_blocks} blocks (and {num_blocks} coinbase txes)")
        genhashes = self.generatetoaddress(self.nodes[0], num_blocks, ADDRESS_BCRT1_UNSPENDABLE)

        for x in range(num_blocks):
            # Should receive the coinbase txid.
            txid = hashtx.receive()

            # Should receive the coinbase raw transaction.
            tx = tx_from_hex(rawtx.receive().hex())
            assert_equal(tx.txid_hex, txid.hex())

            # Should receive the generated raw block.
            hex = rawblock.receive()
            block = CBlock()
            block.deserialize(BytesIO(hex))
            assert block.is_valid()
            assert_equal(block.vtx[0].txid_hex, tx.txid_hex)
            assert_equal(len(block.vtx), 1)
            assert_equal(genhashes[x], hash256_reversed(hex[:80]).hex())

            # Should receive the generated block hash.
            hash = hashblock.receive().hex()
            assert_equal(genhashes[x], hash)
            # The block should only have the coinbase txid.
            assert_equal([txid.hex()], self.nodes[1].getblock(hash)["tx"])


        self.log.info("Wait for tx from second node")
        payment_tx = self.wallet.send_self_transfer(from_node=self.nodes[1])
        payment_txid = payment_tx['txid']
        self.sync_all()
        # Should receive the broadcasted txid.
        txid = hashtx.receive()
        assert_equal(payment_txid, txid.hex())

        # Should receive the broadcasted raw transaction.
        hex = rawtx.receive()