ut_tx_deserialize.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. #!/usr/bin/env python3
  2. """
  3. test/unit_tests_d/ut_tx_deserialize: TX deserialization unit test for the MMGen suite
  4. """
  5. import os,json
  6. from mmgen.common import *
  7. from ..include.common import *
  8. from mmgen.protocol import init_coin
  9. from mmgen.tx import MMGenTX,DeserializedTX
  10. from mmgen.rpc import rpc_init
  11. from mmgen.daemon import CoinDaemon
  12. class unit_test(object):
  13. def _get_core_repo_root(self):
  14. self.core_repo_root = os.getenv('CORE_REPO_ROOT')
  15. if not self.core_repo_root:
  16. die(1,'The environmental variable CORE_REPO_ROOT must be set before running this test')
  17. def run_test(self,name,ut):
  18. async def test_tx(txhex,desc,n):
  19. def has_nonstandard_outputs(outputs):
  20. for o in outputs:
  21. t = o['scriptPubKey']['type']
  22. if t in ('nonstandard','pubkey','nulldata'):
  23. return True
  24. return False
  25. d = await g.rpc.call('decoderawtransaction',txhex)
  26. if has_nonstandard_outputs(d['vout']): return False
  27. dt = DeserializedTX(txhex)
  28. if opt.verbose:
  29. Msg('\n====================================================')
  30. Msg_r('.' if opt.quiet else '{:>3}) {}\n'.format(n,desc))
  31. if opt.verbose:
  32. Pmsg(d)
  33. Msg('----------------------------------------------------')
  34. Pmsg(dt)
  35. # metadata
  36. assert dt['txid'] == d['txid'],'TXID does not match'
  37. assert dt['lock_time'] == d['locktime'],'Locktime does not match'
  38. assert dt['version'] == d['version'],'Version does not match'
  39. # inputs
  40. a,b = d['vin'],dt['txins']
  41. for i in range(len(a)):
  42. assert a[i]['txid'] == b[i]['txid'],'TxID of input {} does not match'.format(i)
  43. assert a[i]['vout'] == b[i]['vout'],'vout of input {} does not match'.format(i)
  44. assert a[i]['sequence'] == int(b[i]['nSeq'],16),(
  45. 'nSeq of input {} does not match'.format(i))
  46. if 'txinwitness' in a[i]:
  47. assert a[i]['txinwitness'] == b[i]['witness'],(
  48. 'witness of input {} does not match'.format(i))
  49. # outputs
  50. a,b = d['vout'],dt['txouts']
  51. for i in range(len(a)):
  52. A = a[i]['scriptPubKey']['addresses'][0]
  53. B = b[i]['address']
  54. fs = 'address of output {} does not match\nA: {}\nB: {}'
  55. assert A == B, fs.format(i,A,B)
  56. A = a[i]['value']
  57. B = b[i]['amount']
  58. fs = 'value of output {} does not match\nA: {}\nB: {}'
  59. assert A == B, fs.format(i,A,B)
  60. A = a[i]['scriptPubKey']['hex']
  61. B = b[i]['scriptPubKey']
  62. fs = 'scriptPubKey of output {} does not match\nA: {}\nB: {}'
  63. assert A == B, fs.format(i,A,B)
  64. return True
  65. def print_info(fn,extra_desc):
  66. if opt.names:
  67. Msg_r('{} {} ({}){}'.format(
  68. purple('Testing'),
  69. cyan(name),
  70. extra_desc,
  71. '' if opt.quiet else '\n'))
  72. else:
  73. Msg_r('Testing transactions from {!r}'.format(fn))
  74. if not opt.quiet: Msg('')
  75. async def test_core_vectors():
  76. self._get_core_repo_root()
  77. fn_b = 'src/test/data/tx_valid.json'
  78. fn = os.path.join(self.core_repo_root,fn_b)
  79. data = json.loads(open(fn).read())
  80. print_info(fn_b,'Core test vectors')
  81. n = 1
  82. for e in data:
  83. if type(e[0]) == list:
  84. await rpc_init()
  85. await test_tx(e[1],desc,n)
  86. n += 1
  87. else:
  88. desc = e[0]
  89. Msg('OK')
  90. async def test_mmgen_txs():
  91. fns = ( ('btc',False,'test/ref/0B8D5A[15.31789,14,tl=1320969600].rawtx'),
  92. ('btc',True,'test/ref/0C7115[15.86255,14,tl=1320969600].testnet.rawtx'),
  93. # ('bch',False,'test/ref/460D4D-BCH[10.19764,tl=1320969600].rawtx')
  94. )
  95. print_info('test/ref/*rawtx','MMGen reference transactions')
  96. for n,(coin,tn,fn) in enumerate(fns):
  97. init_coin(coin,tn)
  98. g.proto.daemon_data_dir = 'test/daemons/' + g.coin.lower()
  99. g.rpc_port = CoinDaemon(coin + ('','_tn')[tn],test_suite=True).rpc_port
  100. await rpc_init()
  101. await test_tx(MMGenTX(fn).hex,fn,n+1)
  102. init_coin('btc',False)
  103. g.rpc_port = CoinDaemon('btc',test_suite=True).rpc_port
  104. await rpc_init()
  105. Msg('OK')
  106. start_test_daemons('btc','btc_tn') # ,'bch')
  107. run_session(test_mmgen_txs(),do_rpc_init=False)
  108. run_session(test_core_vectors(),do_rpc_init=False)
  109. stop_test_daemons('btc','btc_tn') # ,'bch')
  110. return True