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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

#!/usr/bin/env python3
# Copyright (c) 2020 The Bitcoin Unlimited developers
"""
Tests various aspects of chained transactions, primarily `get_history` and
`get_mempool`.

`get_history` and `get_mempool` are basically the same RPC call, except that
get_mempool ignores confirmed transactions.
"""

import asyncio

from test_framework.electrumutil import (
    ElectrumTestFramework,
    script_to_scripthash,
)
from test_framework.script import CScript, OP_TRUE, OP_DROP, OP_NOP, OP_FALSE
from test_framework.util import assert_equal
from test_framework.utiltx import pad_tx
from test_framework.serialize import to_hex
from test_framework.electrumconnection import ElectrumConnection

from test_framework.utilcompat import lock, unlock
from test_framework.environment import on_bch, on_nex

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

GET_HISTORY = "blockchain.scripthash.get_history"
GET_MEMPOOL = "blockchain.scripthash.get_mempool"


class ElectrumMempoolChain(ElectrumTestFramework):
    async def run_test(self):
        n = self.nodes[0]

        await self.bootstrap_p2p()
        cli = ElectrumConnection()
        try:
            await cli.connect()
            coinbases = await self.mine_blocks(cli, n, 100)
            await self.test_blockheight_unconfirmed(n, cli, coinbases.pop(0))
            await self.test_chain_to_from_one_scripthash(n, cli, coinbases.pop(0))
        finally:
            cli.disconnect()

    async def test_blockheight_unconfirmed(self, n, cli, coinbase):
        """
        Check that unconfirmed transactions are correctly provided.

        Note that height is different for unconfirmed with confirmed parents
        and unconfirmed with unconfirmed parents.
        """
        # Another unique anyone-can-spend scriptpubkey
        scriptpubkey = CScript([OP_FALSE, OP_DROP, OP_NOP])
        scripthash = script_to_scripthash(lock(scriptpubkey))

        # There should exist any history for scripthash
        assert_equal(0, len(await cli.call(GET_HISTORY, scripthash)))
        assert_equal(0, len(await cli.call(GET_MEMPOOL, scripthash)))

        # Send a chain of three txs. We expect that:
        # tx1: height 0,  signaling unconfirmed with confirmed parents
        # tx2: height -1, signaling unconfirmed with unconfirmed parents
        # tx3: height -1, signaling unconfirmed with unconfirmed parents
        fee = 200
        tx1 = self.create_transaction(
            coinbase,
            n=0,
            value=coinbase.vout[0].n_value - fee,
            sig=anyonecanspend_unlocking_script(),
            out=lock(scriptpubkey),
        )
        pad_tx(tx1)
        tx2 = self.create_transaction(
            tx1,
            n=0,
            value=tx1.vout[0].n_value - fee,
            sig=unlock(scriptpubkey),
            out=lock(scriptpubkey),
        )
        pad_tx(tx2)
        tx3 = self.create_transaction(
            tx2,
            n=0,
            value=tx2.vout[0].n_value - fee,
            sig=unlock(scriptpubkey),
            out=lock(scriptpubkey),
        )
        pad_tx(tx3)

        for tx in [tx1, tx2, tx3]:
            await cli.call("blockchain.transaction.broadcast", to_hex(tx))
        await self.wait_for_mempool_count(cli, n, count=3)

        res = await cli.call(GET_HISTORY, scripthash)
        print("RES: ", res)
        assert_equal(3, len(res))
        assert_equal(res, await cli.call(GET_MEMPOOL, scripthash))

        def get_tx(txhash):
            for tx in res:
                if tx["tx_hash"] == txhash:
                    return tx
            raise AssertionError("tx not in result")

        assert_equal(0, get_tx(tx1.get_rpc_hex_id())["height"])
        assert_equal(-1, get_tx(tx2.get_rpc_hex_id())["height"])
        assert_equal(-1, get_tx(tx3.get_rpc_hex_id())["height"])

        # Confirm tx1, see that
        # tx1: gets tipheight
        # tx2: gets height 0, tx3 keeps height -1
        await self.mine_blocks(cli, n, 1, [tx1])
        for call in [GET_HISTORY, GET_MEMPOOL]:
            res = await cli.call(call, scripthash)

            if call == GET_HISTORY:
                assert_equal(n.getblockcount(), get_tx(tx1.get_rpc_hex_id())["height"])
            else:
                assert (tx["tx_hash"] != tx1.get_rpc_hex_id() for tx in res)
            assert_equal(0, get_tx(tx2.get_rpc_hex_id())["height"])
            assert_equal(-1, get_tx(tx3.get_rpc_hex_id())["height"])

        # cleanup mempool for next test
        await self.mine_blocks(cli, n, 1, [tx2, tx3])
        assert len(n.getrawtxpool()) == 0

    async def test_chain_to_from_one_scripthash(self, n, cli, coinbase):
        """
        Creates a tx chain where the same scripthash is both funder and spender
        """
        scriptpubkey = CScript([OP_TRUE, OP_TRUE, OP_DROP, OP_DROP])
        scripthash = script_to_scripthash(lock(scriptpubkey))
        assert_equal(0, len(await cli.call(GET_HISTORY, scripthash)))
        assert_equal(0, len(await cli.call(GET_MEMPOOL, scripthash)))

        def has_tx(res, txhash):
            for tx in res:
                if tx["tx_hash"] == txhash:
                    return True
            return False

        chain_length = 25
        tx_chain = self._create_tx_chain(coinbase, scriptpubkey, chain_length)

        # Check mempool
        assert len(n.getrawtxpool()) == 0
        await self.p2p.send_txs_and_test(tx_chain, n)
        await self.wait_for_mempool_count(cli, n, count=len(tx_chain))

        res = await cli.call(GET_HISTORY, scripthash)
        assert_equal(len(tx_chain), len(res))
        assert all(has_tx(res, tx.get_rpc_hex_id()) for tx in tx_chain)

        res_mempool = await cli.call(GET_MEMPOOL, scripthash)
        assert_equal(res, res_mempool)

        # Check when confirmed in a block
        await self.mine_blocks(cli, n, 1, tx_chain)
        res = await cli.call(GET_HISTORY, scripthash)
        assert_equal(len(tx_chain), len(res))
        assert all(has_tx(res, tx.get_rpc_hex_id()) for tx in tx_chain)
        assert_equal(0, len(await cli.call(GET_MEMPOOL, scripthash)))

    def _create_tx_chain(self, coinbase, scriptpubkey, chain_len):
        tx_chain = [coinbase]
        fee = 300
        for _ in range(chain_len):
            prev_tx = tx_chain[-1]

            amount = prev_tx.vout[0].n_value - fee

            if len(tx_chain) == 1:
                # first tx is a coinbase spend
                unlocking_script = anyonecanspend_unlocking_script()
            else:
                unlocking_script = unlock(scriptpubkey)

            tx = self.create_transaction(
                prev_tx,
                n=0,
                value=amount,
                sig=unlocking_script,
                out=lock(scriptpubkey),
            )
            pad_tx(tx)
            tx_chain.append(tx)

        tx_chain.pop(0)  # the initial unspent is not part of the chain
        return tx_chain


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