tCreate basic version of protocol class tests - electrum-personal-server - Maximally lightweight electrum server for a single user
HTML git clone https://git.parazyd.org/electrum-personal-server
DIR Log
DIR Files
DIR Refs
DIR README
---
DIR commit b38006736d604f06e3feb9e988bf20362e382c69
DIR parent a9475357565dc7779554bbad69501d3c5394a72b
HTML Author: chris-belcher <chris-belcher@users.noreply.github.com>
Date: Tue, 10 Dec 2019 20:56:54 +0000
Create basic version of protocol class tests
Diffstat:
M electrumpersonalserver/server/elec… | 32 ++++++++++++++++---------------
A test/test_electrum_protocol.py | 171 +++++++++++++++++++++++++++++++
2 files changed, 188 insertions(+), 15 deletions(-)
---
DIR diff --git a/electrumpersonalserver/server/electrumprotocol.py b/electrumpersonalserver/server/electrumprotocol.py
t@@ -72,9 +72,9 @@ def get_block_header(rpc, blockhash, raw=False):
return header
def get_current_header(rpc, raw):
- new_bestblockhash = rpc.call("getbestblockhash", [])
- header = get_block_header(rpc, new_bestblockhash, raw)
- return new_bestblockhash, header
+ bestblockhash = rpc.call("getbestblockhash", [])
+ header = get_block_header(rpc, bestblockhash, raw)
+ return bestblockhash, header
def get_block_headers_hex(rpc, start_height, count):
#read count number of headers starting from start_height
t@@ -164,6 +164,8 @@ class ElectrumProtocol(object):
query = json.loads(line)
except json.decoder.JSONDecodeError as e:
raise IOError(e)
+ if "method" not in query:
+ raise IOError("Bad client query, no \"method\"")
method = query["method"]
if method == "blockchain.transaction.get":
t@@ -200,7 +202,7 @@ class ElectrumProtocol(object):
electrum_proof["pos"], "merkle":
electrum_proof["merkle"]}
except (ValueError, JsonRpcError) as e:
- logger.info("merkle proof not found for " + txid
+ self.logger.info("merkle proof not found for " + txid
+ " sending a dummy, Electrum client should be run "
+ "with --skipmerklecheck")
#reply with a proof that the client with accept if
t@@ -213,13 +215,13 @@ class ElectrumProtocol(object):
if self.txmonitor.subscribe_address(scrhash):
history_hash = self.txmonitor.get_electrum_history_hash(scrhash)
else:
- logger.warning("Address not known to server, hash(address) = "
- + scrhash + ".\nCheck that you've imported the master "
- + "public key(s) correctly. The first three addresses of "
- + "each key are printed out on startup,\nso check that "
- + "they really are addresses you expect. In Electrum go to"
- + " Wallet -> Information to get the right master public "
- + "key.")
+ self.logger.warning("Address not known to server, hash(address)"
+ + " = " + scrhash + ".\nCheck that you've imported the "
+ + "master public key(s) correctly. The first three "
+ + "addresses of each key are printed out on startup,\nso "
+ + "check that they really are addresses you expect. In "
+ + "Electrum go to Wallet -> Information to get the right "
+ + "master public key.")
history_hash = get_status_electrum([])
self._send_response(query, history_hash)
elif method == "blockchain.scripthash.get_history":
t@@ -227,14 +229,14 @@ class ElectrumProtocol(object):
history = self.txmonitor.get_electrum_history(scrhash)
if history == None:
history = []
- logger.warning("Address history not known to server, "
+ self.logger.warning("Address history not known to server, "
+ "hash(address) = " + scrhash)
self._send_response(query, history)
elif method == "blockchain.scripthash.get_balance":
scrhash = query["params"][0]
balance = self.txmonitor.get_address_balance(scrhash)
if balance == None:
- logger.warning("Address history not known to server, "
+ self.logger.warning("Address history not known to server, "
+ "hash(address) = " + scrhash)
balance = {"confirmed": 0, "unconfirmed": 0}
self._send_response(query, balance)
t@@ -359,7 +361,7 @@ class ElectrumProtocol(object):
elif method == "mempool.get_fee_histogram":
if self.disable_mempool_fee_histogram:
result = [[0, 0]]
- logger.debug("fee histogram disabled, sending back empty "
+ self.logger.debug("fee histogram disabled, sending back empty "
+ "mempool")
else:
st = time.time()
t@@ -367,7 +369,7 @@ class ElectrumProtocol(object):
et = time.time()
MEMPOOL_WARNING_DURATION = 10 #seconds
if et - st > MEMPOOL_WARNING_DURATION:
- logger.warning("Mempool very large resulting in slow "
+ self.logger.warning("Mempool very large resulting in slow "
+ "response by server. Consider setting "
+ "`disable_mempool_fee_histogram = true`")
#algorithm copied from the relevant place in ElectrumX
DIR diff --git a/test/test_electrum_protocol.py b/test/test_electrum_protocol.py
t@@ -0,0 +1,171 @@
+
+import pytest
+import logging
+import json
+
+from electrumpersonalserver.server import (
+ TransactionMonitor,
+ JsonRpcError,
+ ElectrumProtocol,
+ get_block_header,
+ get_current_header,
+ get_block_headers_hex,
+ JsonRpcError
+)
+
+logger = logging.getLogger('ELECTRUMPERSONALSERVER-TEST')
+logger.setLevel(logging.DEBUG)
+
+def get_dummy_hash_from_height(height):
+ return str(height) + "a"*(64 - len(str(height)))
+
+def get_height_from_dummy_hash(hhash):
+ return int(hhash[:hhash.index("a")])
+
+class DummyJsonRpc(object):
+ def __init__(self):
+ self.calls = {}
+ self.blockchain_height = 100000
+
+ def call(self, method, params):
+ if method not in self.calls:
+ self.calls[method] = [0, []]
+ self.calls[method][0] += 1
+ self.calls[method][1].append(params)
+ if method == "getbestblockhash":
+ return get_dummy_hash_from_height(self.blockchain_height)
+ elif method == "getblockhash":
+ height = params[0]
+ if height > self.blockchain_height:
+ raise JsonRpcError()
+ return get_dummy_hash_from_height(height)
+ elif method == "getblockheader":
+ blockhash = params[0]
+ height = get_height_from_dummy_hash(blockhash)
+ header = {
+ "hash": blockhash,
+ "confirmations": self.blockchain_height - height + 1,
+ "height": height,
+ "version": 536870912,
+ "versionHex": "20000000",
+ "merkleroot": "aa"*32,
+ "time": height*100,
+ "mediantime": height*100,
+ "nonce": 1,
+ "bits": "207fffff",
+ "difficulty": 4.656542373906925e-10,
+ "chainwork": "000000000000000000000000000000000000000000000"
+ + "00000000000000000da",
+ "nTx": 1,
+ }
+ if height > 0:
+ header["previousblockhash"] = get_dummy_hash_from_height(
+ height - 1)
+ if height < self.blockchain_height:
+ header["nextblockhash"] = get_dummy_hash_from_height(height + 1)
+ return header
+ elif method == "gettransaction":
+ for t in self.txlist:
+ if t["txid"] == params[0]:
+ return t
+ raise JsonRpcError()
+ else:
+ raise ValueError("unknown method in dummy jsonrpc")
+
+def test_get_block_header():
+ rpc = DummyJsonRpc()
+ for height in [0, 1000]:
+ for raw in [True, False]:
+ blockhash = rpc.call("getblockhash", [height])
+ ret = get_block_header(rpc, blockhash, raw)
+ if raw:
+ assert type(ret) == dict
+ assert "hex" in ret
+ assert "height" in ret
+ assert len(ret["hex"]) == 160
+ else:
+ assert type(ret) == dict
+ assert len(ret) == 7
+
+def test_get_current_header():
+ rpc = DummyJsonRpc()
+ for raw in [True, False]:
+ ret = get_current_header(rpc, raw)
+ assert type(ret[0]) == str
+ assert len(ret[0]) == 64
+ if raw:
+ assert type(ret[1]) == dict
+ assert "hex" in ret[1]
+ assert "height" in ret[1]
+ assert len(ret[1]["hex"]) == 160
+ else:
+ assert type(ret[1]) == dict
+ assert len(ret[1]) == 7
+
+def test_get_block_headers_hex_out_of_bounds():
+ rpc = DummyJsonRpc()
+ ret = get_block_headers_hex(rpc, rpc.blockchain_height + 10, 5)
+ assert len(ret) == 2
+ assert ret[0] == ""
+ assert ret[1] == 0
+
+def test_get_block_headers_hex():
+ rpc = DummyJsonRpc()
+ count = 200
+ ret = get_block_headers_hex(rpc, 100, count)
+ assert len(ret) == 2
+ assert ret[1] == count
+ assert len(ret[0]) == count*80*2 #80 bytes per header, 2 chars per byte
+
+@pytest.mark.parametrize(
+ "invalid_json_query",
+ [
+ "{\"invalid-json\":}",
+ "{\"valid-json-no-method\": 5}"
+ ]
+)
+def test_invalid_json_query_line(invalid_json_query):
+ protocol = ElectrumProtocol(None, None, logger, None, None, None)
+ with pytest.raises(IOError) as e:
+ protocol.handle_query(invalid_json_query)
+
+class DummyTransactionMonitor(object):
+ def __init__(self):
+ self.deterministic_wallets = list(range(5))
+ self.address_history = list(range(5))
+
+ def get_electrum_history_hash(self, scrhash):
+ pass
+
+ def get_electrum_history(self, scrhash):
+ pass
+
+ def unsubscribe_all_addresses(self):
+ pass
+
+ def subscribe_address(self, scrhash):
+ pass
+
+ def get_address_balance(self, scrhash):
+ pass
+
+def create_electrum_protocol_instance(broadcast_method="own-node",
+ tor_hostport=("127.0.0.01", 9050),
+ disable_mempool_fee_histogram=False):
+ protocol = ElectrumProtocol(DummyJsonRpc(), DummyTransactionMonitor(),
+ logger, broadcast_method, tor_hostport, disable_mempool_fee_histogram)
+ sent_lines = []
+ protocol.set_send_line_fun(lambda l: sent_lines.append(json.loads(
+ l.decode())))
+ return protocol, sent_lines
+
+def test_server_ping():
+ protocol, sent_lines = create_electrum_protocol_instance()
+ idd = 1
+ protocol.handle_query(json.dumps({"method": "server.ping", "id": idd}))
+ assert len(sent_lines) == 1
+ assert sent_lines[0]["result"] == None
+ assert sent_lines[0]["id"] == idd
+
+
+