rpc.py 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. #!/usr/bin/env python3
  2. #
  3. # MMGen Wallet, a terminal-based cryptocurrency wallet
  4. # Copyright (C)2013-2025 The MMGen Project <mmgen@tuta.io>
  5. # Licensed under the GNU General Public License, Version 3:
  6. # https://www.gnu.org/licenses
  7. # Public project repositories:
  8. # https://github.com/mmgen/mmgen-wallet
  9. # https://gitlab.com/mmgen/mmgen-wallet
  10. """
  11. test.cmdtest_d.httpd.thornode.rpc: Thornode RPC HTTP server
  12. """
  13. import re, json
  14. from wsgiref.util import request_uri
  15. from . import ThornodeServer
  16. class ThornodeRPCServer(ThornodeServer):
  17. name = 'thornode RPC server'
  18. def make_response_body(self, method, environ):
  19. req_str = request_uri(environ)
  20. if re.search(r'/bank/balances/(\S+)', req_str):
  21. data = {
  22. 'result': [
  23. {'denom': 'foocoin', 'amount': 321321321321},
  24. {'denom': 'rune', 'amount': 987654321321},
  25. {'denom': 'barcoin', 'amount': 123123123123},
  26. ]}
  27. elif m := re.search(r'/auth/accounts/(\S+)', req_str):
  28. data = {
  29. 'result': {
  30. 'value': {
  31. 'address': m[1],
  32. 'pub_key': 'PubKeySecp256k1{0000}',
  33. 'account_number': '1234',
  34. 'sequence': '333444'
  35. }}}
  36. elif m := re.search(r'/tx$', req_str):
  37. assert method == 'POST'
  38. txid = environ['wsgi.input'].read(71).decode().removeprefix('hash=0x').upper()
  39. data = {
  40. 'result': {
  41. 'hash': txid,
  42. 'height': '21298600',
  43. 'index': 2,
  44. 'tx_result': {
  45. 'gas_used': '173222',
  46. 'events': [],
  47. 'codespace': ''
  48. },
  49. 'tx': 'MHgwMGZvb2Jhcg=='
  50. }
  51. }
  52. elif m := re.search(r'/check_tx$', req_str):
  53. assert method == 'POST'
  54. data = {
  55. 'result': {
  56. 'code': 0,
  57. 'data': '',
  58. 'log': '',
  59. 'info': '',
  60. 'gas_wanted': '-1',
  61. 'gas_used': '53774',
  62. 'events': [],
  63. 'codespace': ''
  64. }
  65. }
  66. elif m := re.search(r'/broadcast_tx_sync$', req_str):
  67. assert method == 'POST'
  68. txhex = environ['wsgi.input'].read(24).decode().removeprefix('tx=0x').upper()
  69. if txhex.startswith('0A540A52'):
  70. data = {
  71. 'result': {
  72. 'code': 0,
  73. 'codespace': '',
  74. 'data': '',
  75. 'hash': '14463C716CF08A814868DB779156BCD85A1DF8EE49E924900A74482E9DEE132D',
  76. 'log': ''
  77. }
  78. }
  79. else:
  80. raise ValueError(f'‘{req_str}’: malformed query path')
  81. return json.dumps(data).encode()