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

import contextlib
import re
import os
import subprocess
import time

from .portseed import electrum_monitoring_port
from .environment import network, Network, node, node_supports, NodeFeature, Node


class TestNode:
    # pylint: disable=too-many-arguments,too-many-positional-arguments
    def __init__(
        self,
        rpc,
        datadir,
        process: subprocess.Popen,
        node_number: int,
        rostrum_process: subprocess.Popen | None,
    ):
        self.rpc = rpc
        self.datadir = datadir
        self.process = process
        self.node_number = node_number
        self.rostrum_process = rostrum_process

    def getrawtxpool(self, *args):
        if network() == Network.BCH:
            if len(args) == 2 and args[0] is False and args[1] == "id":
                # Defaults for BCH, but bch does not take the "id" arg.
                return self.rpc.getrawmempool()
            return self.rpc.getrawmempool(*args)
        if network() == Network.NEX:
            return self.rpc.getrawtxpool(*args)
        raise NotImplementedError()

    def getrawmempool(self, _):
        raise NotImplementedError("Use 'getrawtxpool'")

    def signrawtransactionwithwallet(self, _):
        raise NotImplementedError("Use 'signrawtransaction'")

    def signrawtransaction(self, *args):
        if node() == Node.BCHN:
            return self.rpc.signrawtransactionwithwallet(*args)
        return self.rpc.signrawtransaction(*args)

    def monitoring_url(self) -> str:
        return f"http://127.0.0.1:{electrum_monitoring_port(self.node_number)}"

    def __getattr__(self, name):
        """
        Assume anything not implemented here is an rpc call and pass it on
        """
        return getattr(self.rpc, name)

    def _raise_assertion_error(self, error):
        assert False, error

    @contextlib.contextmanager
    def assert_debug_log(self, *, expected_msgs, unexpected_msgs):
        if node_supports(node(), NodeFeature.RPC_LOGLINE):
            self.rpc.logline("flush log")
        debug_log = os.path.join(self.datadir, "regtest", "debug.log")
        with open(debug_log, encoding="utf-8") as dl:
            dl.seek(0, 2)
            prev_size = dl.tell()
        try:
            yield
        finally:
            # Travis test framework is so erratic that to make this reliable
            # but generally quick we need to try a bunch of time looking for
            # the appropriate log
            missing_expected = []
            for _ in range(0, 90):
                if node_supports(node(), NodeFeature.RPC_LOGLINE):
                    self.rpc.logline("flush log")
                with open(debug_log, encoding="utf-8") as dl:
                    dl.seek(prev_size)
                    log = dl.read()
                print_log = " - " + "\n - ".join(log.splitlines())
                for unexpected in unexpected_msgs:
                    if (
                        re.search(re.escape(unexpected), log, flags=re.MULTILINE)
                        is not None
                    ):
                        self._raise_assertion_error(
                            f'Unexpected message "{unexpected}" matched log:\n\n{print_log}\n\n'
                        )

                missing_expected = []
                for expected_msg in expected_msgs:
                    if (
                        re.search(re.escape(expected_msg), log, flags=re.MULTILINE)
                        is None
                    ):
                        missing_expected.append(expected_msg)

                if not missing_expected:
                    break
                time.sleep(1)

            if missing_expected:
                self._raise_assertion_error(
                    f"Expected messages {missing_expected} do not partially match log:\n\n{print_log}\n\n"
                )