rostrum 14.0.1

An efficient implementation of Electrum Server with token support
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

#!/usr/bin/env python3
# Copyright (c) 2024 The Bitcoin Unlimited developers

"""
This tests that rostrum can connect and communicate with the node via p2p protocol
"""

import asyncio
from test_framework.script import CScript, OP_NOP
from test_framework.electrumutil import (
    ElectrumTestFramework,
    get_txid_from_idem,
)
from test_framework.environment import on_bch, on_nex
from test_framework.electrumconnection import ElectrumConnection
from test_framework.utiltx import pad_tx
from test_framework.utilcompat import lock

if on_bch():
    from test_framework.bch.blocktools import create_transaction
    from test_framework.bch.script import anyonecanspend_unlocking_script
elif on_nex():
    from test_framework.nex.blocktools import create_transaction
    from test_framework.nex.script import anyonecanspend_unlocking_script
else:
    raise NotImplementedError()

TX_BROADCAST = "blockchain.transaction.broadcast"


class FeatureP2PTests(ElectrumTestFramework):
    async def run_test(self):
        await self.bootstrap_p2p()
        cli = ElectrumConnection()
        await cli.connect()
        try:
            coinbases = await self.mine_blocks(cli, self.nodes[0], 101)
            await self.test_p2p_broadcast(cli, self.nodes[0], coinbases.pop(0))

        finally:
            cli.disconnect()

    async def test_p2p_broadcast(self, cli, n, utxo):
        scriptpubkey = CScript([OP_NOP])

        tx = create_transaction(
            utxo,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=lock(scriptpubkey),
            value=utxo.vout[0].n_value - 1000,
        )
        pad_tx(tx)

        n = self.nodes[0]

        # Mempool is empty before broadcast
        await self.wait_for_mempool_count(cli, count=0)

        broadcast_opt = {"p2p_only": True}

        txid = await get_txid_from_idem(
            n, await cli.call(TX_BROADCAST, tx.to_hex(), broadcast_opt)
        )

        # If the node sees p2p broadcated tx; it means p2p connection is good.
        await self.wait_for_mempool_tx(cli, txid)


if __name__ == "__main__":
    asyncio.run(FeatureP2PTests().main())