ttest.py - obelisk - Electrum server using libbitcoin as its backend
HTML git clone https://git.parazyd.org/obelisk
DIR Log
DIR Files
DIR Refs
DIR README
DIR LICENSE
---
ttest.py (8885B)
---
1 #!/usr/bin/env python3
2 # Copyright (C) 2021 Ivan J. <parazyd@dyne.org>
3 #
4 # This file is part of obelisk
5 #
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License version 3
8 # as published by the Free Software Foundation.
9 #
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU Affero General Public License for more details.
14 #
15 # You should have received a copy of the GNU Affero General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
17 import asyncio
18 import sys
19 from logging import getLogger
20
21 from electrumobelisk.protocol import ElectrumProtocol
22
23 #
24 # See bottom of this file for test orchestration.
25 #
26
27 ENDPOINTS = {
28 "query": "tcp://testnet2.libbitcoin.net:29091",
29 "heart": "tcp://testnet2.libbitcoin.net:29092",
30 "block": "tcp://testnet2.libbitcoin.net:29093",
31 "trans": "tcp://testnet2.libbitcoin.net:29094",
32 }
33
34
35 async def test_blockchain_block_header(protocol, writer):
36 expect = "01000000c54675276e0401706aa93db6494dd7d1058b19424f23c8d7c01076da000000001c4375c8056b0ded0fa3d7fc1b5511eaf53216aed72ea95e1b5d19eccbe855f91a184a4dffff001d0336a226"
37 query = {"params": [123]}
38 res = await protocol.blockchain_block_header(writer, query)
39 if "error" in res and "result" not in res:
40 return "blockchain_block_header", False
41 if res["result"] != expect:
42 return "blockchain_block_header", False
43 return "blockchain_block_header", True
44
45
46 async def test_blockchain_block_headers(protocol, writer):
47 expect = "0100000000000000000000000000000000000000000000000000000000000000000000003ba3edfd7a7b12b27ac72c3e67768f617fc81bc3888a51323a9fb8aa4b1e5e4adae5494dffff001d1aa4ae180100000043497fd7f826957108f4a30fd9cec3aeba79972084e90ead01ea330900000000bac8b0fa927c0ac8234287e33c5f74d38d354820e24756ad709d7038fc5f31f020e7494dffff001d03e4b6720100000006128e87be8b1b4dea47a7247d5528d2702c96826c7a648497e773b800000000e241352e3bec0a95a6217e10c3abb54adfa05abb12c126695595580fb92e222032e7494dffff001d00d23534"
48 query = {"params": [123, 3]}
49 res = await protocol.blockchain_block_headers(writer, query)
50 if "error" in res and "result" not in res:
51 return "blockchain_block_headers", False
52 if res["result"]["hex"] != expect:
53 return "blockchain_block_headers", False
54 return "blockchain_block_headers", True
55
56
57 async def test_blockchain_estimatefee(protocol, writer):
58 expect = -1
59 query = {"params": []}
60 res = await protocol.blockchain_estimatefee(writer, query)
61 if "error" in res and "result" not in res:
62 return "blockchain_estimatefee", False
63 if res["result"] != expect:
64 return "blockchain_estimatefee", False
65 return "blockchain_estimatefee", True
66
67
68 async def test_blockchain_relayfee(protocol, writer):
69 expect = 0.00001
70 query = {"params": []}
71 res = await protocol.blockchain_relayfee(writer, query)
72 if "error" in res and "result" not in res:
73 return "blockchain_relayfee", False
74 if res["result"] != expect:
75 return "blockchain_relayfee", False
76 return "blockchain_relayfee", True
77
78
79 async def test_blockchain_scripthash_get_balance(protocol, writer):
80 shs = [
81 "c036b0ff3ad79662cd517cd5fe1fa0af07377b9262d16f276f11ced69aaa6921",
82 "92dd1eb7c042956d3dd9185a58a2578f61fee91347196604540838ccd0f8c08c",
83 ]
84 expect = [
85 {
86 "result": {
87 "confirmed": 0,
88 "unconfirmed": 0
89 }
90 },
91 {
92 "result": {
93 "confirmed": 831000,
94 "unconfirmed": 0
95 }
96 },
97 ]
98
99 res = []
100 for i in shs:
101 params = {"params": [i]}
102 res.append(await
103 protocol.blockchain_scripthash_get_balance(writer, params))
104
105 for i in expect:
106 if res[expect.index(i)] != i:
107 return "blockchain_scripthash_get_balance", False
108 return "blockchain_scripthash_get_balance", True
109
110
111 async def test_blockchain_scripthash_get_history(protocol, writer):
112 shs = [
113 "c036b0ff3ad79662cd517cd5fe1fa0af07377b9262d16f276f11ced69aaa6921",
114 "92dd1eb7c042956d3dd9185a58a2578f61fee91347196604540838ccd0f8c08c",
115 ]
116 expect = [
117 (
118 1936167,
119 "084eba0e08c78b63e07535b74a5a849994d49afade95d0d205e4963e3f568600",
120 ),
121 (
122 1936171,
123 "705c4f265df23726c09c5acb80f9e8a85845c17d68974d89814383855c8545a2",
124 ),
125 (
126 1936171,
127 "705c4f265df23726c09c5acb80f9e8a85845c17d68974d89814383855c8545a2",
128 ),
129 (
130 1970700,
131 "a9c3c22cc2589284288b28e802ea81723d649210d59dfa7e03af00475f4cec20",
132 ),
133 ]
134
135 res = []
136 for i in shs:
137 params = {"params": [i]}
138 data = await protocol.blockchain_scripthash_get_history(writer, params)
139 if "result" in data:
140 for j in data["result"]:
141 res.append((j["height"], j["tx_hash"]))
142
143 if expect != res:
144 return "blockchain_scripthash_get_history", False
145 return "blockchain_scripthash_get_history", True
146
147
148 async def test_blockchain_scripthash_listunspent(protocol, writer):
149 shs = [
150 "c036b0ff3ad79662cd517cd5fe1fa0af07377b9262d16f276f11ced69aaa6921",
151 "92dd1eb7c042956d3dd9185a58a2578f61fee91347196604540838ccd0f8c08c",
152 ]
153
154 expect = [
155 [],
156 [1, 731000, 1936171],
157 [1, 100000, 1970700],
158 ]
159
160 res = []
161 for i in shs:
162 params = {"params": [i]}
163 data = await protocol.blockchain_scripthash_listunspent(writer, params)
164 if "result" in data and len(data["result"]) > 0:
165 for j in data["result"]:
166 res.append([j["tx_pos"], j["value"], j["height"]])
167 else:
168 res.append([])
169
170 if res != expect:
171 return "blockchain_scripthash_listunspent", False
172 return "blockchain_scripthash_listunspent", True
173
174
175 async def test_blockchain_transaction_get(protocol, writer):
176 expect = "020000000001011caa5f4ba91ff0ab77712851c1b17943e68f28d46bb0d96cbc13cdbef53c2b87000000001716001412e6e94028ab399b67c1232383d12f1dd3fc03b5feffffff02a40111000000000017a914ff1d7f4c85c562764ca16daa11e97d10eda52ebf87a0860100000000001976a9144a0360eac874a569e82ca6b17274d90bccbcab5e88ac0247304402205392417f5ffba2c0f3a501476fb6872368b2065c53bf18b2a201691fb88cdbe5022016c68ec9e094ba2b06d4bdc6af996ac74b580ab9728c622bb5304aaff04cb6980121031092742ffdf5901ceafcccec090c58170ce1d0ec26963ef7c7a2738a415a317e0b121e00"
177 params = {
178 "params":
179 ["a9c3c22cc2589284288b28e802ea81723d649210d59dfa7e03af00475f4cec20"]
180 }
181 data = await protocol.blockchain_transaction_get(writer, params)
182
183 if "result" not in data and data["result"] != expect:
184 return "blockchain_transaction_get", False
185 return "blockchain_transaction_get", True
186
187
188 class MockWriter(asyncio.StreamWriter):
189 """Mock class for StreamWriter"""
190 def __init__(self):
191 self.mock = None
192
193 def write(self, data):
194 return True
195
196 async def drain(self):
197 return True
198
199
200 async def main():
201 test_pass = []
202 test_fail = []
203
204 log = getLogger("obelisktest")
205 protocol = ElectrumProtocol(log, "testnet", ENDPOINTS, {})
206 writer = MockWriter()
207 functions = [
208 test_blockchain_block_header,
209 test_blockchain_block_headers,
210 test_blockchain_estimatefee,
211 # test_blockchain_headers_subscribe,
212 test_blockchain_relayfee,
213 test_blockchain_scripthash_get_balance,
214 test_blockchain_scripthash_get_history,
215 # test_blockchain_scripthash_get_mempool,
216 test_blockchain_scripthash_listunspent,
217 # test_blockchain_scripthash_subscribe,
218 # test_blockchain_scripthash_unsubscribe,
219 # test_blockchain_transaction_broadcast,
220 test_blockchain_transaction_get,
221 # test_blockchain_transaction_get_merkle,
222 # test_blockchain_transaction_from_pos,
223 # test_mempool_get_fee_histogram,
224 # test_server_add_peer,
225 # test_server_banner,
226 # test_server_donation_address,
227 # test_server_features,
228 # test_server_peers_subscribe,
229 # test_server_ping,
230 # test_server_version,
231 ]
232
233 for func in functions:
234 name, result = await func(protocol, writer)
235 if result:
236 test_pass.append(name)
237 print(f"PASS: {name}")
238 else:
239 print(f"FAIL: {name}")
240 test_fail.append(name)
241
242 await protocol.stop()
243
244 print()
245 print(f"Tests passed: {len(test_pass)}")
246 print(f"Tests failed: {len(test_fail)}")
247
248 ret = 1 if len(test_fail) > 0 else 0
249 sys.exit(ret)
250
251
252 if __name__ == "__main__":
253 asyncio.run(main())