obelisk

Electrum server using libbitcoin as its backend
git clone https://git.parazyd.org/obelisk
Log | Files | Refs | README | LICENSE

test.py (8877B)


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