← Watch feed
tests: split out toy_server from test_interface.py
Public commit record
60/100 · Adequate
What the developer wrote
Authored by SomberNight
tests: split out toy_server from test_interface.py
✓ Specific, descriptive subject✓ Names a concrete action or component✓ Mentions testing or verification! No meaningful explanatory body
What changed, and why it matters
This commit is a simple housekeeping change in the project's test code. It moves a helper mock server (used only for automated testing) out of one test file and into a new dedicated file. There is no change to the actual Electrum wallet software that users run, and no security issue is present.
Recommended action
No action required. This is a benign test refactoring.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit refactors test-only code by extracting the ToyServer and ToyServerSession classes from tests/test_interface.py into a new tests/toyserver.py module. The moved code is a local/regtest-only mock Electrum server used by unit tests. No functional changes were made to production code, network protocol handling, cryptography, or wallet logic.
Changed components
tests/test_interface.pytests/toyserver.pyInspect captured patch +271 / −258
diff --git a/tests/test_interface.py b/tests/test_interface.py
index 528fdc6..6087a00 100644
--- a/tests/test_interface.py
+++ b/tests/test_interface.py
@@ -1,26 +1,20 @@
import asyncio
-import collections
from functools import partial
-from typing import Optional, Sequence, Iterable, Mapping, List, Set, Dict
import aiorpcx
from aiorpcx import RPCError
-import electrum
from electrum.interface import ServerAddr, Interface, PaddedRSTransport
from electrum import util, blockchain
from electrum.util import OldTaskGroup, bfh
-from electrum.logging import Logger
from electrum.simple_config import SimpleConfig
from electrum.transaction import Transaction
-from electrum import constants
from electrum.wallet import Abstract_Wallet
from electrum.blockchain import Blockchain
-from electrum.bitcoin import script_to_scripthash
-from electrum.synchronizer import history_status
from . import ElectrumTestCase
from . import restore_wallet_from_text__for_unittest
+from .toyserver import ToyServer, ToyServerSession
class TestServerAddr(ElectrumTestCase):
@@ -99,257 +93,6 @@ class MockNetwork:
return self.blockchain().height()
-# regtest chain:
-_BLOCK_HEADERS: List[bytes] = [
- bfh("0100000000000000000000000000000000000000000000000000000000000000000000003ba3edfd7a7b12b27ac72c3e67768f617fc81bc3888a51323a9fb8aa4b1e5e4adae5494dffff7f2002000000"),
- bfh("0000002006226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f186c8dfd970a4545f79916bc1d75c9d00432f57c89209bf3bb115b7612848f509c25f45bffff7f2000000000"),
- bfh("00000020686bdfc6a3db73d5d93e8c9663a720a26ecb1ef20eb05af11b36cdbc57c19f7ebf2cbf153013a1c54abaf70e95198fcef2f3059cc6b4d0f7e876808e7d24d11cc825f45bffff7f2000000000"),
- bfh("00000020122baa14f3ef54985ae546d1611559e3f487bd2a0f46e8dbb52fbacc9e237972e71019d7feecd9b8596eca9a67032c5f4641b23b5d731dc393e37de7f9c2f299e725f45bffff7f2000000000"),
- bfh("00000020f8016f7ef3a17d557afe05d4ea7ab6bde1b2247b7643896c1b63d43a1598b747a3586da94c71753f27c075f57f44faf913c31177a0957bbda42e7699e3a2141aed25f45bffff7f2001000000"),
- bfh("000000201d589c6643c1d121d73b0573e5ee58ab575b8fdf16d507e7e915c5fbfbbfd05e7aee1d692d1615c3bdf52c291032144ce9e3b258a473c17c745047f3431ff8e2ee25f45bffff7f2000000000"),
- bfh("00000020b833ed46eea01d4c980f59feee44a66aa1162748b6801029565d1466790c405c3a141ce635cbb1cd2b3a4fcdd0a3380517845ba41736c82a79cab535d31128066526f45bffff7f2001000000"),
-]
-
-class ToyServer:
- """Electrum Server backend"""
-
- def __init__(self):
- self.sessions = set() # type: Set[ToyServerSession]
- self._block_headers = _BLOCK_HEADERS[:]
- self._txid_to_block_height = collections.defaultdict(int) # type: dict[str, int]
- self.sh_to_funding_txids = collections.defaultdict(set) # type: dict[str, set[str]]
- self.sh_to_spending_txids = collections.defaultdict(set) # type: dict[str, set[str]]
- self.txs = {} # type: dict[str, bytes]
-
- def get_session_by_name(self, client_name: str) -> 'ToyServerSession':
- found_sessions = [
- session for session in self.sessions
- if session.client_name == client_name]
- if len(found_sessions) > 1:
- raise Exception("multiple sessions reusing client_name")
- elif len(found_sessions) == 0:
- raise Exception("no session with given client_name")
- else:
- assert len(found_sessions) == 1
- return found_sessions[0]
-
- @property
- def cur_height(self) -> int:
- """chain tip"""
- return len(self._block_headers) - 1
-
- def get_block_header(self, height: int) -> bytes:
- return self._block_headers[height]
-
- def calc_sh_history(self, sh: str) -> Sequence[tuple[str, int]]:
- txids = self.sh_to_funding_txids[sh] | self.sh_to_spending_txids[sh]
- hist = []
- for txid in txids:
- bh = self._txid_to_block_height[txid]
- hist.append((txid, bh))
- hist.sort(key=lambda x: x[1]) # FIXME put mempool txs last
- return hist
-
- def _add_tx(self, tx: Transaction) -> set[str]:
- """Returns touched scripthashes."""
- txid = tx.txid()
- assert txid
- self.txs[txid] = bfh(str(tx))
- touched_sh = set()
- # update sh_to_funding_txids
- for txout in tx.outputs():
- sh = script_to_scripthash(txout.scriptpubkey)
- self.sh_to_funding_txids[sh].add(txid)
- touched_sh.add(sh)
- # update sh_to_spending_txids
- for txin in tx.inputs():
- if parent_tx_raw := self.txs.get(txin.prevout.txid.hex()):
- parent_tx = Transaction(parent_tx_raw)
- ptxout = parent_tx.outputs()[txin.prevout.out_idx]
- sh = script_to_scripthash(ptxout.scriptpubkey)
- self.sh_to_spending_txids[sh].add(txid)
- touched_sh.add(sh)
- return touched_sh
-
- async def mempool_add_tx(self, tx: Transaction) -> None:
- touched_sh = self._add_tx(tx)
- for session in self.sessions:
- await session.server_send_notifications(touched_sh=touched_sh)
-
- async def mine_block(self, *, txs: Iterable[Transaction] = None):
- if txs is None:
- txs = []
- # new header
- prev_header = self._block_headers[-1]
- prev_blockhash = blockchain.hash_raw_header(prev_header)
- new_header = blockchain.serialize_header({
- 'version': 99999, # don't care
- 'prev_block_hash': prev_blockhash,
- 'merkle_root': 'deadbeef' * 8, # don't care
- 'timestamp': 1_500_000_000, # don't care
- 'bits': 0x1d00ffff, # don't care
- 'nonce': 1, # don't care
- })
- self._block_headers.append(new_header)
- # process txs
- touched_sh = set()
- for tx in txs:
- txid = tx.txid()
- assert txid
- self._txid_to_block_height[txid] = self.cur_height
- touched_sh |= self._add_tx(tx)
- for session in self.sessions:
- await session.server_send_notifications(touched_sh=touched_sh, height_changed=True)
-
-
-class ToyServerSession(aiorpcx.RPCSession, Logger):
- """Server-side representation of a single electrum-protocol session."""
-
- def __init__(self, *args, toyserver: ToyServer, **kwargs):
- aiorpcx.RPCSession.__init__(self, *args, **kwargs)
- Logger.__init__(self)
- self.svr = toyserver
- self.logger.debug(f'connection from {self.remote_address()}')
- self.subbed_headers = False
- self.notified_height = None # type: Optional[int]
- self.subbed_scripthashes = set() # type: set[str]
- self._method_counts = collections.defaultdict(int) # type: dict[str, int]
- self.client_name = None
- self.svr.sessions.add(self)
-
- async def connection_lost(self):
- await super().connection_lost()
- self.logger.debug(f'{self.remote_address()} disconnected')
- self.svr.sessions.discard(self)
-
- async def handle_request(self, request):
- handlers = {
- 'server.version': self._handle_server_version,
- 'server.features': self._handle_server_features,
- 'blockchain.estimatefee': self._handle_estimatefee,
- 'blockchain.headers.subscribe': self._handle_headers_subscribe,
- 'blockchain.block.header': self._handle_block_header,
- 'blockchain.block.headers': self._handle_block_headers,
- 'blockchain.scripthash.subscribe': self._handle_scripthash_subscribe,
- 'blockchain.scripthash.get_history': self._handle_scripthash_get_history,
- 'blockchain.transaction.get': self._handle_transaction_get,
- 'blockchain.transaction.broadcast': self._handle_transaction_broadcast,
- 'blockchain.transaction.get_merkle': self._handle_transaction_get_merkle,
- 'mempool.get_info': self._handle_mempool_get_info,
- 'server.ping': self._handle_ping,
- }
- handler = handlers.get(request.method)
- self._method_counts[request.method] += 1
- coro = aiorpcx.handler_invocation(handler, request)()
- return await coro
-
- async def _handle_server_version(self, client_name='', protocol_version=None, *args, **kwargs):
- self.client_name = client_name
- return ['toyserver/0.1', '1.6']
-
- async def _handle_server_features(self) -> dict:
- return {
- 'genesis_hash': constants.net.GENESIS,
- 'hosts': {"14.3.140.101": {"tcp_port": 51001, "ssl_port": 51002}},
- 'protocol_max': '1.6',
- 'protocol_min': '1.6',
- 'pruning': None,
- 'server_version': 'ElectrumX 1.19.0',
- 'hash_function': 'sha256',
- }
-
- async def _handle_estimatefee(self, number, mode=None):
- return 0.00001000
-
- async def _handle_mempool_get_info(self):
- return {
- "mempoolminfee": 0.00001000,
- "minrelaytxfee": 0.00001000,
- "incrementalrelayfee": 0.00001000,
- }
-
- def _get_headersub_result(self):
- height = self.svr.cur_height
- return {'hex': self.svr.get_block_header(height).hex(), 'height': height}
-
- async def _handle_headers_subscribe(self):
- self.subbed_headers = True
- return self._get_headersub_result()
-
- async def _handle_block_header(self, height):
- return self.svr.get_block_header(height).hex()
-
- async def _handle_block_headers(self, start_height, count):
- cur_height = self.svr.cur_height
- assert start_height <= cur_height, (start_height, cur_height)
- last_height = min(start_height+count-1, cur_height) # [start_height, last_height]
- count = last_height - start_height + 1
- headers = list(self.svr.get_block_header(idx).hex() for idx in range(start_height, last_height + 1))
- return {'headers': headers, 'count': count, 'max': 2016}
-
- async def _handle_ping(self):
- return None
-
- async def _handle_transaction_get(self, tx_hash: str, verbose=False):
- assert not verbose
- rawtx = self.svr.txs.get(tx_hash)
- if rawtx is None:
- DAEMON_ERROR = 2
- raise RPCError(DAEMON_ERROR, f'daemon error: unknown txid={tx_hash}')
- return rawtx.hex()
-
- async def _handle_transaction_get_merkle(self, tx_hash: str, height: int) -> dict:
- # Fake stuff. Client will ignore it due to config.NETWORK_SKIPMERKLECHECK
- return {
- "merkle":
- [
- "713d6c7e6ce7bbea708d61162231eaa8ecb31c4c5dd84f81c20409a90069cb24",
- "03dbaec78d4a52fbaf3c7aa5d3fccd9d8654f323940716ddf5ee2e4bda458fde",
- "e670224b23f156c27993ac3071940c0ff865b812e21e0a162fe7a005d6e57851",
- "369a1619a67c3108a8850118602e3669455c70cdcdb89248b64cc6325575b885",
- "4756688678644dcb27d62931f04013254a62aeee5dec139d1aac9f7b1f318112",
- "7b97e73abc043836fd890555bfce54757d387943a6860e5450525e8e9ab46be5",
- "61505055e8b639b7c64fd58bce6fc5c2378b92e025a02583303f69930091b1c3",
- "27a654ff1895385ac14a574a0415d3bbba9ec23a8774f22ec20d53dd0b5386ff",
- "5312ed87933075e60a9511857d23d460a085f3b6e9e5e565ad2443d223cfccdc",
- "94f60b14a9f106440a197054936e6fb92abbd69d6059b38fdf79b33fc864fca0",
- "2d64851151550e8c4d337f335ee28874401d55b358a66f1bafab2c3e9f48773d"
- ],
- "block_height": height,
- "pos": 710,
- }
-
- async def _handle_transaction_broadcast(self, raw_tx: str) -> str:
- tx = Transaction(raw_tx)
- txid = tx.txid()
- await self.svr.mempool_add_tx(tx) # TODO don't await, just queue up? this sends notifs before response, lol
- return txid
-
- async def _handle_scripthash_subscribe(self, sh: str) -> Optional[str]:
- self.subbed_scripthashes.add(sh)
- hist = self.svr.calc_sh_history(sh)
- return history_status(hist)
-
- async def _handle_scripthash_get_history(self, sh: str) -> Sequence[dict]:
- hist_tuples = self.svr.calc_sh_history(sh)
- hist_dicts = [{"height": height, "tx_hash": txid} for (txid, height) in hist_tuples]
- for hist_dict in hist_dicts: # add "fee" key for mempool txs
- if hist_dict["height"] in (0, -1,):
- hist_dict["fee"] = 0
- return hist_dicts
-
- async def server_send_notifications(self, *, touched_sh: Iterable[str], height_changed: bool = False) -> None:
- if height_changed and self.subbed_headers and self.notified_height != self.svr.cur_height:
- self.notified_height = self.svr.cur_height
- args = (self._get_headersub_result(),)
- await self.send_notification('blockchain.headers.subscribe', args)
- touched_sh = set(sh for sh in touched_sh if sh in self.subbed_scripthashes)
- for sh in touched_sh:
- hist = self.svr.calc_sh_history(sh)
- args = (sh, history_status(hist))
- await self.send_notification("blockchain.scripthash.subscribe", args)
-
-
class TestInterface(ElectrumTestCase):
REGTEST = True
diff --git a/tests/toyserver.py b/tests/toyserver.py
new file mode 100644
index 0000000..7dfce4a
--- /dev/null
+++ b/tests/toyserver.py
@@ -0,0 +1,270 @@
+# Copyright (C) 2026 The Electrum developers
+# Distributed under the MIT software license, see the accompanying
+# file LICENCE or http://www.opensource.org/licenses/mit-license.php
+
+import collections
+from typing import Optional, Sequence, Iterable, List, Set
+
+import aiorpcx
+from aiorpcx import RPCError
+
+from electrum import blockchain
+from electrum.util import bfh
+from electrum.logging import Logger
+from electrum.transaction import Transaction
+from electrum import constants
+from electrum.bitcoin import script_to_scripthash
+from electrum.synchronizer import history_status
+
+
+# regtest chain:
+_BLOCK_HEADERS: List[bytes] = [
+ bfh("0100000000000000000000000000000000000000000000000000000000000000000000003ba3edfd7a7b12b27ac72c3e67768f617fc81bc3888a51323a9fb8aa4b1e5e4adae5494dffff7f2002000000"),
+ bfh("0000002006226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f186c8dfd970a4545f79916bc1d75c9d00432f57c89209bf3bb115b7612848f509c25f45bffff7f2000000000"),
+ bfh("00000020686bdfc6a3db73d5d93e8c9663a720a26ecb1ef20eb05af11b36cdbc57c19f7ebf2cbf153013a1c54abaf70e95198fcef2f3059cc6b4d0f7e876808e7d24d11cc825f45bffff7f2000000000"),
+ bfh("00000020122baa14f3ef54985ae546d1611559e3f487bd2a0f46e8dbb52fbacc9e237972e71019d7feecd9b8596eca9a67032c5f4641b23b5d731dc393e37de7f9c2f299e725f45bffff7f2000000000"),
+ bfh("00000020f8016f7ef3a17d557afe05d4ea7ab6bde1b2247b7643896c1b63d43a1598b747a3586da94c71753f27c075f57f44faf913c31177a0957bbda42e7699e3a2141aed25f45bffff7f2001000000"),
+ bfh("000000201d589c6643c1d121d73b0573e5ee58ab575b8fdf16d507e7e915c5fbfbbfd05e7aee1d692d1615c3bdf52c291032144ce9e3b258a473c17c745047f3431ff8e2ee25f45bffff7f2000000000"),
+ bfh("00000020b833ed46eea01d4c980f59feee44a66aa1162748b6801029565d1466790c405c3a141ce635cbb1cd2b3a4fcdd0a3380517845ba41736c82a79cab535d31128066526f45bffff7f2001000000"),
+]
+
+
+class ToyServer:
+ """Electrum Server backend"""
+
+ def __init__(self):
+ self.sessions = set() # type: Set[ToyServerSession]
+ self._block_headers = _BLOCK_HEADERS[:]
+ self._txid_to_block_height = collections.defaultdict(int) # type: dict[str, int]
+ self.sh_to_funding_txids = collections.defaultdict(set) # type: dict[str, set[str]]
+ self.sh_to_spending_txids = collections.defaultdict(set) # type: dict[str, set[str]]
+ self.txs = {} # type: dict[str, bytes]
+
+ def get_session_by_name(self, client_name: str) -> 'ToyServerSession':
+ found_sessions = [
+ session for session in self.sessions
+ if session.client_name == client_name]
+ if len(found_sessions) > 1:
+ raise Exception("multiple sessions reusing client_name")
+ elif len(found_sessions) == 0:
+ raise Exception("no session with given client_name")
+ else:
+ assert len(found_sessions) == 1
+ return found_sessions[0]
+
+ @property
+ def cur_height(self) -> int:
+ """chain tip"""
+ return len(self._block_headers) - 1
+
+ def get_block_header(self, height: int) -> bytes:
+ return self._block_headers[height]
+
+ def calc_sh_history(self, sh: str) -> Sequence[tuple[str, int]]:
+ txids = self.sh_to_funding_txids[sh] | self.sh_to_spending_txids[sh]
+ hist = []
+ for txid in txids:
+ bh = self._txid_to_block_height[txid]
+ hist.append((txid, bh))
+ hist.sort(key=lambda x: x[1]) # FIXME put mempool txs last
+ return hist
+
+ def _add_tx(self, tx: Transaction) -> set[str]:
+ """Returns touched scripthashes."""
+ txid = tx.txid()
+ assert txid
+ self.txs[txid] = bfh(str(tx))
+ touched_sh = set()
+ # update sh_to_funding_txids
+ for txout in tx.outputs():
+ sh = script_to_scripthash(txout.scriptpubkey)
+ self.sh_to_funding_txids[sh].add(txid)
+ touched_sh.add(sh)
+ # update sh_to_spending_txids
+ for txin in tx.inputs():
+ if parent_tx_raw := self.txs.get(txin.prevout.txid.hex()):
+ parent_tx = Transaction(parent_tx_raw)
+ ptxout = parent_tx.outputs()[txin.prevout.out_idx]
+ sh = script_to_scripthash(ptxout.scriptpubkey)
+ self.sh_to_spending_txids[sh].add(txid)
+ touched_sh.add(sh)
+ return touched_sh
+
+ async def mempool_add_tx(self, tx: Transaction) -> None:
+ touched_sh = self._add_tx(tx)
+ for session in self.sessions:
+ await session.server_send_notifications(touched_sh=touched_sh)
+
+ async def mine_block(self, *, txs: Iterable[Transaction] = None):
+ if txs is None:
+ txs = []
+ # new header
+ prev_header = self._block_headers[-1]
+ prev_blockhash = blockchain.hash_raw_header(prev_header)
+ new_header = blockchain.serialize_header({
+ 'version': 99999, # don't care
+ 'prev_block_hash': prev_blockhash,
+ 'merkle_root': 'deadbeef' * 8, # don't care
+ 'timestamp': 1_500_000_000, # don't care
+ 'bits': 0x1d00ffff, # don't care
+ 'nonce': 1, # don't care
+ })
+ self._block_headers.append(new_header)
+ # process txs
+ touched_sh = set()
+ for tx in txs:
+ txid = tx.txid()
+ assert txid
+ self._txid_to_block_height[txid] = self.cur_height
+ touched_sh |= self._add_tx(tx)
+ for session in self.sessions:
+ await session.server_send_notifications(touched_sh=touched_sh, height_changed=True)
+
+
+class ToyServerSession(aiorpcx.RPCSession, Logger):
+ """Server-side representation of a single electrum-protocol session."""
+
+ def __init__(self, *args, toyserver: ToyServer, **kwargs):
+ aiorpcx.RPCSession.__init__(self, *args, **kwargs)
+ Logger.__init__(self)
+ self.svr = toyserver
+ self.logger.debug(f'connection from {self.remote_address()}')
+ self.subbed_headers = False
+ self.notified_height = None # type: Optional[int]
+ self.subbed_scripthashes = set() # type: set[str]
+ self._method_counts = collections.defaultdict(int) # type: dict[str, int]
+ self.client_name = None
+ self.svr.sessions.add(self)
+
+ async def connection_lost(self):
+ await super().connection_lost()
+ self.logger.debug(f'{self.remote_address()} disconnected')
+ self.svr.sessions.discard(self)
+
+ async def handle_request(self, request):
+ handlers = {
+ 'server.version': self._handle_server_version,
+ 'server.features': self._handle_server_features,
+ 'blockchain.estimatefee': self._handle_estimatefee,
+ 'blockchain.headers.subscribe': self._handle_headers_subscribe,
+ 'blockchain.block.header': self._handle_block_header,
+ 'blockchain.block.headers': self._handle_block_headers,
+ 'blockchain.scripthash.subscribe': self._handle_scripthash_subscribe,
+ 'blockchain.scripthash.get_history': self._handle_scripthash_get_history,
+ 'blockchain.transaction.get': self._handle_transaction_get,
+ 'blockchain.transaction.broadcast': self._handle_transaction_broadcast,
+ 'blockchain.transaction.get_merkle': self._handle_transaction_get_merkle,
+ 'mempool.get_info': self._handle_mempool_get_info,
+ 'server.ping': self._handle_ping,
+ }
+ handler = handlers.get(request.method)
+ self._method_counts[request.method] += 1
+ coro = aiorpcx.handler_invocation(handler, request)()
+ return await coro
+
+ async def _handle_server_version(self, client_name='', protocol_version=None, *args, **kwargs):
+ self.client_name = client_name
+ return ['toyserver/0.1', '1.6']
+
+ async def _handle_server_features(self) -> dict:
+ return {
+ 'genesis_hash': constants.net.GENESIS,
+ 'hosts': {"14.3.140.101": {"tcp_port": 51001, "ssl_port": 51002}},
+ 'protocol_max': '1.6',
+ 'protocol_min': '1.6',
+ 'pruning': None,
+ 'server_version': 'ElectrumX 1.19.0',
+ 'hash_function': 'sha256',
+ }
+
+ async def _handle_estimatefee(self, number, mode=None):
+ return 0.00001000
+
+ async def _handle_mempool_get_info(self):
+ return {
+ "mempoolminfee": 0.00001000,
+ "minrelaytxfee": 0.00001000,
+ "incrementalrelayfee": 0.00001000,
+ }
+
+ def _get_headersub_result(self):
+ height = self.svr.cur_height
+ return {'hex': self.svr.get_block_header(height).hex(), 'height': height}
+
+ async def _handle_headers_subscribe(self):
+ self.subbed_headers = True
+ return self._get_headersub_result()
+
+ async def _handle_block_header(self, height):
+ return self.svr.get_block_header(height).hex()
+
+ async def _handle_block_headers(self, start_height, count):
+ cur_height = self.svr.cur_height
+ assert start_height <= cur_height, (start_height, cur_height)
+ last_height = min(start_height+count-1, cur_height) # [start_height, last_height]
+ count = last_height - start_height + 1
+ headers = list(self.svr.get_block_header(idx).hex() for idx in range(start_height, last_height + 1))
+ return {'headers': headers, 'count': count, 'max': 2016}
+
+ async def _handle_ping(self):
+ return None
+
+ async def _handle_transaction_get(self, tx_hash: str, verbose=False):
+ assert not verbose
+ rawtx = self.svr.txs.get(tx_hash)
+ if rawtx is None:
+ DAEMON_ERROR = 2
+ raise RPCError(DAEMON_ERROR, f'daemon error: unknown txid={tx_hash}')
+ return rawtx.hex()
+
+ async def _handle_transaction_get_merkle(self, tx_hash: str, height: int) -> dict:
+ # Fake stuff. Client will ignore it due to config.NETWORK_SKIPMERKLECHECK
+ return {
+ "merkle":
+ [
+ "713d6c7e6ce7bbea708d61162231eaa8ecb31c4c5dd84f81c20409a90069cb24",
+ "03dbaec78d4a52fbaf3c7aa5d3fccd9d8654f323940716ddf5ee2e4bda458fde",
+ "e670224b23f156c27993ac3071940c0ff865b812e21e0a162fe7a005d6e57851",
+ "369a1619a67c3108a8850118602e3669455c70cdcdb89248b64cc6325575b885",
+ "4756688678644dcb27d62931f04013254a62aeee5dec139d1aac9f7b1f318112",
+ "7b97e73abc043836fd890555bfce54757d387943a6860e5450525e8e9ab46be5",
+ "61505055e8b639b7c64fd58bce6fc5c2378b92e025a02583303f69930091b1c3",
+ "27a654ff1895385ac14a574a0415d3bbba9ec23a8774f22ec20d53dd0b5386ff",
+ "5312ed87933075e60a9511857d23d460a085f3b6e9e5e565ad2443d223cfccdc",
+ "94f60b14a9f106440a197054936e6fb92abbd69d6059b38fdf79b33fc864fca0",
+ "2d64851151550e8c4d337f335ee28874401d55b358a66f1bafab2c3e9f48773d"
+ ],
+ "block_height": height,
+ "pos": 710,
+ }
+
+ async def _handle_transaction_broadcast(self, raw_tx: str) -> str:
+ tx = Transaction(raw_tx)
+ txid = tx.txid()
+ await self.svr.mempool_add_tx(tx) # TODO don't await, just queue up? this sends notifs before response, lol
+ return txid
+
+ async def _handle_scripthash_subscribe(self, sh: str) -> Optional[str]:
+ self.subbed_scripthashes.add(sh)
+ hist = self.svr.calc_sh_history(sh)
+ return history_status(hist)
+
+ async def _handle_scripthash_get_history(self, sh: str) -> Sequence[dict]:
+ hist_tuples = self.svr.calc_sh_history(sh)
+ hist_dicts = [{"height": height, "tx_hash": txid} for (txid, height) in hist_tuples]
+ for hist_dict in hist_dicts: # add "fee" key for mempool txs
+ if hist_dict["height"] in (0, -1,):
+ hist_dict["fee"] = 0
+ return hist_dicts
+
+ async def server_send_notifications(self, *, touched_sh: Iterable[str], height_changed: bool = False) -> None:
+ if height_changed and self.subbed_headers and self.notified_height != self.svr.cur_height:
+ self.notified_height = self.svr.cur_height
+ args = (self._get_headersub_result(),)
+ await self.send_notification('blockchain.headers.subscribe', args)
+ touched_sh = set(sh for sh in touched_sh if sh in self.subbed_scripthashes)
+ for sh in touched_sh:
+ hist = self.svr.calc_sh_history(sh)
+ args = (sh, history_status(hist))
+ await self.send_notification("blockchain.scripthash.subscribe", args)
+
Risk score
Our methodology →Why this scored 15/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.