test.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())