proto.btc.tx: support OP_RETURN data outputs
From mmgen-txcreate --help:
A single DATA_SPEC argument may be given on the command line to create an
OP_RETURN data output with a zero spend amount. This is the preferred way to
embed data in the blockchain. DATA_SPEC may be of the form "data":DATA or
"hexdata":DATA
More info:
$ mmgen-txcreate --help
Testing:
$ test/modtest.py -v tx.op_return_data
$ test/cmdtest.py -ne swap
This commit is contained in:
parent
1f1e0a1186
commit
8fd463ecfe
GPG key ID:
3F8B1861E32B7DA2
16 changed files with 389 additions and 34 deletions
|
|
@ -1 +1 @@
|
|||
January 2025
|
||||
February 2025
|
||||
|
|
|
|||
|
|
@ -1 +1 @@
|
|||
15.1.dev13
|
||||
15.1.dev14
|
||||
|
|
|
|||
|
|
@ -22,7 +22,7 @@ class help_notes:
|
|||
|
||||
def txcreate_args(self):
|
||||
return (
|
||||
'[ADDR,AMT ...] ADDR <change addr, addrlist ID or addr type>'
|
||||
'[ADDR,AMT ... | DATA_SPEC] ADDR <change addr, addrlist ID or addr type>'
|
||||
if self.proto.base_proto == 'Bitcoin' else
|
||||
'ADDR,AMT')
|
||||
|
||||
|
|
@ -212,7 +212,15 @@ will be displayed in a menu, with the user prompted to select one. In the
|
|||
second form, the user specifies the Seed ID as well, allowing the script to
|
||||
select the transaction’s change output or single output without prompting.
|
||||
See EXAMPLES below.
|
||||
"""
|
||||
|
||||
A single DATA_SPEC argument may also be given on the command line to create
|
||||
an OP_RETURN data output with a zero spend amount. This is the preferred way
|
||||
to embed data in the blockchain. DATA_SPEC may be of the form "data":DATA
|
||||
or "hexdata":DATA. In the first form, DATA is a string in your system’s native
|
||||
encoding, typically UTF-8. In the second, DATA is a hexadecimal string (with
|
||||
the leading ‘0x’ omitted) encoding the binary data to be embedded. In both
|
||||
cases, the resulting byte string must not exceed {bl} bytes in length.
|
||||
""".format(bl=self.proto.max_op_return_data_len)
|
||||
if self.proto.base_proto == 'Bitcoin' else """
|
||||
The transaction output is specified in the form ADDRESS,AMOUNT.
|
||||
""")
|
||||
|
|
|
|||
|
|
@ -51,6 +51,7 @@ class mainnet(CoinProtocol.Secp256k1): # chainparams.cpp
|
|||
max_halvings = 64
|
||||
start_subsidy = 50
|
||||
max_int = 0xffffffff
|
||||
max_op_return_data_len = 80
|
||||
|
||||
coin_cfg_opts = (
|
||||
'ignore_daemon_version',
|
||||
|
|
|
|||
|
|
@ -14,10 +14,16 @@ proto.btc.tx.base: Bitcoin base transaction class
|
|||
|
||||
from collections import namedtuple
|
||||
|
||||
from ....addr import CoinAddr
|
||||
from ....tx import base as TxBase
|
||||
from ....obj import MMGenList, HexStr
|
||||
from ....obj import MMGenList, HexStr, ListItemAttr
|
||||
from ....util import msg, make_chksum_6, die, pp_fmt
|
||||
|
||||
from .op_return_data import OpReturnData
|
||||
|
||||
def data2scriptPubKey(data):
|
||||
return '6a' + '{:02x}'.format(len(data)) + data.hex() # OP_RETURN data
|
||||
|
||||
def addr2scriptPubKey(proto, addr):
|
||||
|
||||
def decode_addr(proto, addr):
|
||||
|
|
@ -45,6 +51,15 @@ def decodeScriptPubKey(proto, s):
|
|||
elif len(s) == 44 and s[:4] == proto.witness_vernum_hex + '14':
|
||||
return ret('witness_v0_keyhash', 'bech32', proto.pubhash2bech32addr(bytes.fromhex(s[4:])), None)
|
||||
|
||||
elif s[:2] == '6a': # OP_RETURN
|
||||
# range 1-80 == hex 2-160, plus 4 for opcode byte + push byte
|
||||
if 6 <= len(s) <= (proto.max_op_return_data_len * 2) + 6: # 2-160 -> 6-166
|
||||
return ret('nulldata', None, None, s[4:]) # return data in hex format
|
||||
else:
|
||||
raise ValueError('{}: OP_RETURN data bytes length not in range 1-{}'.format(
|
||||
len(s[4:]) // 2,
|
||||
proto.max_op_return_data_len))
|
||||
|
||||
else:
|
||||
raise NotImplementedError(f'Unrecognized scriptPubKey ({s})')
|
||||
|
||||
|
|
@ -159,6 +174,10 @@ class Base(TxBase.Base):
|
|||
rel_fee_disp = 'sat/byte'
|
||||
_deserialized = None
|
||||
|
||||
class Output(TxBase.Base.Output): # output contains either addr or data, but not both
|
||||
addr = ListItemAttr(CoinAddr, include_proto=True) # ImmutableAttr in parent cls
|
||||
data = ListItemAttr(OpReturnData, include_proto=True, typeconv=True) # type None in parent cls
|
||||
|
||||
class InputList(TxBase.Base.InputList):
|
||||
|
||||
# Lexicographical Indexing of Transaction Inputs and Outputs
|
||||
|
|
@ -176,7 +195,9 @@ class Base(TxBase.Base):
|
|||
def sort_func(a):
|
||||
return (
|
||||
int.to_bytes(a.amt.to_unit('satoshi'), 8, 'big')
|
||||
+ bytes.fromhex(addr2scriptPubKey(self.parent.proto, a.addr)))
|
||||
+ bytes.fromhex(
|
||||
addr2scriptPubKey(self.parent.proto, a.addr) if a.addr else
|
||||
data2scriptPubKey(a.data)))
|
||||
self.sort(key=sort_func)
|
||||
|
||||
def has_segwit_inputs(self):
|
||||
|
|
@ -226,8 +247,10 @@ class Base(TxBase.Base):
|
|||
# 8 (amt) + scriptlen_byte + script_bytes
|
||||
# script_bytes:
|
||||
# ADDR: p2pkh: 25, p2sh: 23, bech32: 22
|
||||
# DATA: opcode_byte ('6a') + push_byte + nulldata_bytes
|
||||
return sum(
|
||||
{'p2pkh':34, 'p2sh':32, 'bech32':31}[o.addr.addr_fmt]
|
||||
{'p2pkh':34, 'p2sh':32, 'bech32':31}[o.addr.addr_fmt] if o.addr else
|
||||
(11 + len(o.data))
|
||||
for o in self.outputs)
|
||||
|
||||
# https://github.com/bitcoin/bips/blob/master/bip-0141.mediawiki
|
||||
|
|
@ -328,8 +351,8 @@ class Base(TxBase.Base):
|
|||
|
||||
check_equal(
|
||||
'outputs',
|
||||
sorted((o['addr'], o['amt']) for o in dtx.txouts),
|
||||
sorted((o.addr, o.amt) for o in self.outputs))
|
||||
sorted((o['addr'] or o['data'], o['amt']) for o in dtx.txouts),
|
||||
sorted((o.addr or o.data.hex(), o.amt) for o in self.outputs))
|
||||
|
||||
if str(self.txid) != make_chksum_6(bytes.fromhex(dtx.unsigned_hex)).upper():
|
||||
do_error(f'MMGen TxID ({self.txid}) does not match serialized transaction data!')
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ proto.btc.tx.info: Bitcoin transaction info class
|
|||
|
||||
from ....tx.info import TxInfo
|
||||
from ....util import fmt, die
|
||||
from ....color import red, green, pink
|
||||
from ....color import red, green, pink, blue
|
||||
from ....addr import MMGenID
|
||||
|
||||
class TxInfo(TxInfo):
|
||||
|
|
@ -79,17 +79,20 @@ class TxInfo(TxInfo):
|
|||
'raw': lambda: io
|
||||
}[sort]
|
||||
|
||||
def data_disp(data):
|
||||
return f'OP_RETURN data ({len(data)} bytes)'
|
||||
|
||||
if terse:
|
||||
iwidth = max(len(str(int(e.amt))) for e in io)
|
||||
addr_w = max(len(e.addr.views[vp1]) for f in (tx.inputs, tx.outputs) for e in f)
|
||||
addr_w = max((len(e.addr.views[vp1]) if e.addr else len(data_disp(e.data))) for f in (tx.inputs, tx.outputs) for e in f)
|
||||
for n, e in enumerate(io_sorted()):
|
||||
yield '{:3} {} {} {} {}\n'.format(
|
||||
n+1,
|
||||
e.addr.fmt(vp1, width=addr_w, color=True),
|
||||
get_mmid_fmt(e, is_input),
|
||||
e.addr.fmt(vp1, width=addr_w, color=True) if e.addr else blue(data_disp(e.data).ljust(addr_w)),
|
||||
get_mmid_fmt(e, is_input) if e.addr else ''.ljust(max_mmwid),
|
||||
e.amt.fmt(iwidth=iwidth, color=True),
|
||||
tx.dcoin)
|
||||
if have_bch:
|
||||
if have_bch and e.addr:
|
||||
yield '{:3} [{}]\n'.format('', e.addr.hl(vp2, color=False))
|
||||
else:
|
||||
col1_w = len(str(len(io))) + 1
|
||||
|
|
@ -105,8 +108,11 @@ class TxInfo(TxInfo):
|
|||
if have_bch:
|
||||
yield ('', '', f'[{e.addr.hl(vp2, color=False)}]')
|
||||
else:
|
||||
yield (n+1, 'address:', f'{e.addr.hl(vp1)} {mmid_fmt}')
|
||||
if have_bch:
|
||||
yield (
|
||||
n + 1,
|
||||
'address:',
|
||||
(f'{e.addr.hl(vp1)} {mmid_fmt}' if e.addr else e.data.hl(add_label=True)))
|
||||
if have_bch and e.addr:
|
||||
yield ('', '', f'[{e.addr.hl(vp2, color=False)}]')
|
||||
if e.comment:
|
||||
yield ('', 'comment:', e.comment.hl())
|
||||
|
|
|
|||
|
|
@ -23,6 +23,15 @@ class New(Base, TxBase.New):
|
|||
fee_fail_fs = 'Network fee estimation for {c} confirmations failed ({t})'
|
||||
no_chg_msg = 'Warning: Change address will be deleted as transaction produces no change'
|
||||
|
||||
def process_data_output_arg(self, arg):
|
||||
if any(arg.startswith(pfx) for pfx in ('data:', 'hexdata:')):
|
||||
if hasattr(self, '_have_op_return_data'):
|
||||
die(1, 'Transaction may have at most one OP_RETURN data output!')
|
||||
self._have_op_return_data = True
|
||||
from .op_return_data import OpReturnData
|
||||
OpReturnData(self.proto, arg) # test data for validity
|
||||
return arg
|
||||
|
||||
@property
|
||||
def relay_fee(self):
|
||||
kb_fee = self.proto.coin_amt(self.rpc.cached['networkinfo']['relayfee'])
|
||||
|
|
@ -134,7 +143,7 @@ class New(Base, TxBase.New):
|
|||
'sequence': e.sequence
|
||||
} for e in self.inputs]
|
||||
|
||||
outputs_dict = {e.addr:e.amt for e in self.outputs}
|
||||
outputs_dict = dict((e.addr, e.amt) if e.addr else ('data', e.data.hex()) for e in self.outputs)
|
||||
|
||||
ret = await self.rpc.call('createrawtransaction', inputs_list, outputs_dict)
|
||||
|
||||
|
|
|
|||
71
mmgen/proto/btc/tx/op_return_data.py
Executable file
71
mmgen/proto/btc/tx/op_return_data.py
Executable file
|
|
@ -0,0 +1,71 @@
|
|||
#!/usr/bin/env python3
|
||||
#
|
||||
# MMGen Wallet, a terminal-based cryptocurrency wallet
|
||||
# Copyright (C)2013-2024 The MMGen Project <mmgen@tuta.io>
|
||||
# Licensed under the GNU General Public License, Version 3:
|
||||
# https://www.gnu.org/licenses
|
||||
# Public project repositories:
|
||||
# https://github.com/mmgen/mmgen-wallet
|
||||
# https://gitlab.com/mmgen/mmgen-wallet
|
||||
|
||||
"""
|
||||
proto.btc.tx.op_return_data: Bitcoin OP_RETURN data class
|
||||
"""
|
||||
|
||||
from ....obj import InitErrors
|
||||
|
||||
class OpReturnData(bytes, InitErrors):
|
||||
|
||||
def __new__(cls, proto, data_spec):
|
||||
|
||||
desc = 'OpReturnData initializer'
|
||||
|
||||
assert isinstance(data_spec, str), f'{desc} must be a string'
|
||||
|
||||
if data_spec.startswith('hexdata:'):
|
||||
hexdata = data_spec[8:]
|
||||
from ....util import is_hex_str
|
||||
assert is_hex_str(hexdata), f'{hexdata!r}: {desc} hexdata not in hexadecimal format'
|
||||
assert not len(hexdata) % 2, f'{len(hexdata)}: {desc} hexdata of non-even length'
|
||||
ret = bytes.fromhex(hexdata)
|
||||
elif data_spec.startswith('data:'):
|
||||
try:
|
||||
ret = data_spec[5:].encode('utf8')
|
||||
except:
|
||||
raise ValueError(f'{desc} string be UTF-8 encoded')
|
||||
else:
|
||||
raise ValueError(f'{desc} string must start with ‘data:’ or ‘hexdata:’')
|
||||
|
||||
assert 1 <= len(ret) <= proto.max_op_return_data_len, (
|
||||
f'{len(ret)}: {desc} string encoded byte length not in range 1-{proto.max_op_return_data_len}')
|
||||
|
||||
return bytes.__new__(cls, ret)
|
||||
|
||||
def __repr__(self):
|
||||
'return an initialization string'
|
||||
ret = str(self)
|
||||
return ('hexdata:' if self.display_hex else 'data:') + ret
|
||||
|
||||
def __str__(self):
|
||||
'return something suitable for display to the user'
|
||||
self.display_hex = True
|
||||
try:
|
||||
ret = self.decode('utf8')
|
||||
except:
|
||||
return self.hex()
|
||||
else:
|
||||
import unicodedata
|
||||
for ch in ret:
|
||||
if ch == '\n' or unicodedata.category(ch)[0] in ('C', 'M'): # see MMGenLabel
|
||||
return self.hex()
|
||||
self.display_hex = False
|
||||
return ret
|
||||
|
||||
def hl(self, add_label=False):
|
||||
'colorize and optionally label the result of str()'
|
||||
from ....color import blue, pink
|
||||
ret = str(self)
|
||||
if add_label:
|
||||
return blue('OP_RETURN data' + (' (hex): ' if self.display_hex else ': ')) + pink(ret)
|
||||
else:
|
||||
return pink(ret)
|
||||
|
|
@ -55,7 +55,8 @@ class MMGenTxIO(MMGenListItem):
|
|||
str(self.mmid.mmtype) if self.mmid else
|
||||
'B' if self.addr.addr_fmt == 'bech32' else
|
||||
'S' if self.addr.addr_fmt == 'p2sh' else
|
||||
None)
|
||||
None
|
||||
) if self.addr else None
|
||||
|
||||
class MMGenTxIOList(list, MMGenObject):
|
||||
|
||||
|
|
@ -97,6 +98,7 @@ class Base(MMGenObject):
|
|||
|
||||
class Output(MMGenTxIO):
|
||||
is_chg = ListItemAttr(bool, typeconv=False)
|
||||
data = ListItemAttr(None, typeconv=False) # placeholder
|
||||
|
||||
class InputList(MMGenTxIOList):
|
||||
desc = 'transaction inputs'
|
||||
|
|
|
|||
|
|
@ -24,10 +24,18 @@ import os, json
|
|||
|
||||
from ..util import ymsg, make_chksum_6, die
|
||||
from ..obj import MMGenObject, HexStr, MMGenTxID, CoinTxID, MMGenTxComment
|
||||
from ..rpc import json_encoder
|
||||
|
||||
class txdata_json_encoder(json.JSONEncoder):
|
||||
def default(self, o):
|
||||
if type(o).__name__.endswith('Amt'):
|
||||
return str(o)
|
||||
elif type(o).__name__ == 'OpReturnData':
|
||||
return repr(o)
|
||||
else:
|
||||
return json.JSONEncoder.default(self, o)
|
||||
|
||||
def json_dumps(data):
|
||||
return json.dumps(data, separators = (',', ':'), cls=json_encoder)
|
||||
return json.dumps(data, separators = (',', ':'), cls=txdata_json_encoder)
|
||||
|
||||
def get_proto_from_coin_id(tx, coin_id, chain):
|
||||
coin, tokensym = coin_id.split(':') if ':' in coin_id else (coin_id, None)
|
||||
|
|
|
|||
|
|
@ -103,7 +103,7 @@ class New(Base):
|
|||
|
||||
def check_dup_addrs(self, io_desc):
|
||||
assert io_desc in ('inputs', 'outputs')
|
||||
addrs = [e.addr for e in getattr(self, io_desc)]
|
||||
addrs = [e.addr for e in getattr(self, io_desc) if e.addr]
|
||||
if len(addrs) != len(set(addrs)):
|
||||
die(2, f'{addrs}: duplicate address in transaction {io_desc}')
|
||||
|
||||
|
|
@ -161,12 +161,18 @@ class New(Base):
|
|||
return False
|
||||
return True
|
||||
|
||||
def add_output(self, coinaddr, amt, is_chg=None):
|
||||
self.outputs.append(self.Output(self.proto, addr=coinaddr, amt=amt, is_chg=is_chg))
|
||||
def add_output(self, coinaddr, amt, is_chg=False, data=None):
|
||||
self.outputs.append(self.Output(self.proto, addr=coinaddr, amt=amt, is_chg=is_chg, data=data))
|
||||
|
||||
def process_data_output_arg(self, arg):
|
||||
return None
|
||||
|
||||
def parse_cmd_arg(self, arg_in, ad_f, ad_w):
|
||||
|
||||
_pa = namedtuple('parsed_txcreate_cmdline_arg', ['arg', 'mmid', 'coin_addr', 'amt'])
|
||||
_pa = namedtuple('parsed_txcreate_cmdline_arg', ['arg', 'mmid', 'coin_addr', 'amt', 'data'])
|
||||
|
||||
if data := self.process_data_output_arg(arg_in):
|
||||
return _pa(arg_in, None, None, None, data)
|
||||
|
||||
arg, amt = arg_in.split(',', 1) if ',' in arg_in else (arg_in, None)
|
||||
|
||||
|
|
@ -182,7 +188,7 @@ class New(Base):
|
|||
else:
|
||||
die(2, f'{arg_in}: invalid command-line argument')
|
||||
|
||||
return _pa(arg, mmid, coin_addr, amt)
|
||||
return _pa(arg, mmid, coin_addr, amt, None)
|
||||
|
||||
async def process_cmd_args(self, cmd_args, ad_f, ad_w):
|
||||
|
||||
|
|
@ -208,17 +214,20 @@ class New(Base):
|
|||
|
||||
parsed_args = [self.parse_cmd_arg(a, ad_f, ad_w) for a in cmd_args]
|
||||
|
||||
chg_args = [a for a in parsed_args if not (a.amt and a.coin_addr)]
|
||||
chg_args = [a for a in parsed_args if not ((a.amt and a.coin_addr) or a.data)]
|
||||
|
||||
if len(chg_args) > 1:
|
||||
desc = 'requested' if self.chg_autoselected else 'listed'
|
||||
die(2, f'ERROR: More than one change address {desc} on command line')
|
||||
|
||||
for a in parsed_args:
|
||||
self.add_output(
|
||||
coinaddr = a.coin_addr or (await get_autochg_addr(a.arg, parsed_args)).addr,
|
||||
amt = self.proto.coin_amt(a.amt or '0'),
|
||||
is_chg = not a.amt)
|
||||
if a.data:
|
||||
self.add_output(None, self.proto.coin_amt('0'), data=a.data)
|
||||
else:
|
||||
self.add_output(
|
||||
coinaddr = a.coin_addr or (await get_autochg_addr(a.arg, parsed_args)).addr,
|
||||
amt = self.proto.coin_amt(a.amt or '0'),
|
||||
is_chg = not a.amt)
|
||||
|
||||
if self.chg_idx is None:
|
||||
die(2,
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ cmd_groups_dfl = {
|
|||
'autosign_automount': ('CmdTestAutosignAutomount', {'modname': 'automount'}),
|
||||
'autosign_eth': ('CmdTestAutosignETH', {'modname': 'automount_eth'}),
|
||||
'regtest': ('CmdTestRegtest', {}),
|
||||
'swap': ('CmdTestSwap', {}),
|
||||
# 'chainsplit': ('CmdTestChainsplit', {}),
|
||||
'ethdev': ('CmdTestEthdev', {}),
|
||||
'xmrwallet': ('CmdTestXMRWallet', {}),
|
||||
|
|
@ -235,6 +236,7 @@ cfgs = { # addr_idx_lists (except 31, 32, 33, 34) must contain exactly 8 address
|
|||
'32': {}, # ref_tx
|
||||
'33': {}, # ref_tx
|
||||
'34': {}, # ref_tx
|
||||
'37': {}, # swap
|
||||
'38': {}, # autosign_clean
|
||||
'39': {}, # xmr_autosign
|
||||
'40': {}, # cfgfile
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ import time
|
|||
|
||||
from .ct_autosign import CmdTestAutosignThreaded
|
||||
from .ct_regtest import CmdTestRegtestBDBWallet, rt_pw
|
||||
from ..include.common import cfg
|
||||
from ..include.common import cfg, gr_uc
|
||||
|
||||
class CmdTestAutosignAutomount(CmdTestAutosignThreaded, CmdTestRegtestBDBWallet):
|
||||
'automounted transacting operations via regtest mode'
|
||||
|
|
@ -78,7 +78,7 @@ class CmdTestAutosignAutomount(CmdTestAutosignThreaded, CmdTestRegtestBDBWallet)
|
|||
|
||||
self.opts.append('--alice')
|
||||
|
||||
def _alice_txcreate(self, chg_addr, opts=[], exit_val=0, expect_str=None):
|
||||
def _alice_txcreate(self, chg_addr, opts=[], exit_val=0, expect_str=None, data_arg=None):
|
||||
|
||||
def do_return():
|
||||
if expect_str:
|
||||
|
|
@ -94,6 +94,7 @@ class CmdTestAutosignAutomount(CmdTestAutosignThreaded, CmdTestRegtestBDBWallet)
|
|||
'mmgen-txcreate',
|
||||
opts
|
||||
+ ['--alice', '--autosign']
|
||||
+ ([data_arg] if data_arg else [])
|
||||
+ [f'{self.burn_addr},1.23456', f'{sid}:{chg_addr}'],
|
||||
exit_val = exit_val or None)
|
||||
|
||||
|
|
@ -109,7 +110,9 @@ class CmdTestAutosignAutomount(CmdTestAutosignThreaded, CmdTestRegtestBDBWallet)
|
|||
return do_return()
|
||||
|
||||
def alice_txcreate1(self):
|
||||
return self._alice_txcreate(chg_addr='C:5')
|
||||
return self._alice_txcreate(
|
||||
chg_addr = 'C:5',
|
||||
data_arg = 'data:'+gr_uc[:24])
|
||||
|
||||
def alice_txcreate2(self):
|
||||
return self._alice_txcreate(chg_addr='L:5')
|
||||
|
|
|
|||
|
|
@ -1170,6 +1170,9 @@ class CmdTestRegtest(CmdTestBase, CmdTestShared):
|
|||
t.expect(f'Mined {num_blocks} block')
|
||||
return t
|
||||
|
||||
def _do_cli(self, cmd_args, decode_json=False):
|
||||
return self._do_mmgen_regtest(['cli'] + cmd_args, decode_json=decode_json)
|
||||
|
||||
def _do_mmgen_regtest(self, cmd_args, decode_json=False):
|
||||
ret = self.spawn(
|
||||
'mmgen-regtest',
|
||||
|
|
|
|||
153
test/cmdtest_d/ct_swap.py
Executable file
153
test/cmdtest_d/ct_swap.py
Executable file
|
|
@ -0,0 +1,153 @@
|
|||
#!/usr/bin/env python3
|
||||
#
|
||||
# MMGen Wallet, a terminal-based cryptocurrency wallet
|
||||
# Copyright (C)2013-2024 The MMGen Project <mmgen@tuta.io>
|
||||
# Licensed under the GNU General Public License, Version 3:
|
||||
# https://www.gnu.org/licenses
|
||||
# Public project repositories:
|
||||
# https://github.com/mmgen/mmgen-wallet
|
||||
# https://gitlab.com/mmgen/mmgen-wallet
|
||||
|
||||
"""
|
||||
test.cmdtest_d.ct_swap: asset swap tests for the cmdtest.py test suite
|
||||
"""
|
||||
|
||||
from .ct_regtest import CmdTestRegtest, rt_data, dfl_wcls, rt_pw
|
||||
|
||||
rtFundAmt = rtFee = None # ruff
|
||||
|
||||
sample1 = '=:ETH.ETH:0x86d526d6624AbC0178cF7296cD538Ecc080A95F1:0/1/0'
|
||||
sample2 = '00010203040506'
|
||||
|
||||
class CmdTestSwap(CmdTestRegtest):
|
||||
bdb_wallet = True
|
||||
networks = ('btc',)
|
||||
tmpdir_nums = [37]
|
||||
|
||||
cmd_group_in = (
|
||||
('setup', 'regtest (Bob and Alice) mode setup'),
|
||||
('subgroup.init_bob', []),
|
||||
('subgroup.fund_bob', ['init_bob']),
|
||||
('subgroup.data', ['init_bob']),
|
||||
('stop', 'stopping regtest daemon'),
|
||||
)
|
||||
cmd_subgroups = {
|
||||
'init_bob': (
|
||||
'creating Bob’s MMGen wallet and tracking wallet',
|
||||
('walletgen_bob', 'wallet generation (Bob)'),
|
||||
('addrgen_bob', 'address generation (Bob)'),
|
||||
('addrimport_bob', 'importing Bob’s addresses'),
|
||||
),
|
||||
'fund_bob': (
|
||||
'funding Bob’s wallet',
|
||||
('fund_bob', 'funding Bob’s wallet'),
|
||||
('bob_bal1', 'Bob’s balance'),
|
||||
),
|
||||
'data': (
|
||||
'OP_RETURN data operations',
|
||||
('data_tx1_create', 'Creating a transaction with OP_RETURN data (hex-encoded ascii)'),
|
||||
('data_tx1_sign', 'Signing the transaction'),
|
||||
('data_tx1_send', 'Sending the transaction'),
|
||||
('data_tx1_chk', 'Checking the sent transaction'),
|
||||
('generate3', 'Generate 3 blocks'),
|
||||
('data_tx2_do', 'Creating and sending a transaction with OP_RETURN data (binary)'),
|
||||
('data_tx2_chk', 'Checking the sent transaction'),
|
||||
('generate3', 'Generate 3 blocks'),
|
||||
),
|
||||
}
|
||||
|
||||
def __init__(self, trunner, cfgs, spawn):
|
||||
super().__init__(trunner, cfgs, spawn)
|
||||
gldict = globals()
|
||||
for k in rt_data:
|
||||
gldict[k] = rt_data[k]['btc']
|
||||
|
||||
@property
|
||||
def sid(self):
|
||||
return self._user_sid('bob')
|
||||
|
||||
def addrgen_bob(self):
|
||||
return self.addrgen('bob', mmtypes=['S', 'B'])
|
||||
|
||||
def addrimport_bob(self):
|
||||
return self.addrimport('bob', mmtypes=['S', 'B'])
|
||||
|
||||
def fund_bob(self):
|
||||
return self.fund_wallet('bob', 'B', rtFundAmt)
|
||||
|
||||
def data_tx1_create(self):
|
||||
return self._data_tx_create('1', 'B:2', 'B:3', 'data', sample1)
|
||||
|
||||
def _data_tx_create(self, src, dest, chg, pfx, sample):
|
||||
t = self.spawn(
|
||||
'mmgen-txcreate',
|
||||
['-d', self.tmpdir, '-B', '--bob', f'{self.sid}:{dest},1', f'{self.sid}:{chg}', f'{pfx}:{sample}'])
|
||||
return self.txcreate_ui_common(t, menu=[], inputs='1', interactive_fee='3s')
|
||||
|
||||
def data_tx1_sign(self):
|
||||
return self._data_tx_sign()
|
||||
|
||||
def _data_tx_sign(self):
|
||||
fn = self.get_file_with_ext('rawtx')
|
||||
t = self.spawn('mmgen-txsign', ['-d', self.tmpdir, '--bob', fn])
|
||||
t.view_tx('v')
|
||||
t.passphrase(dfl_wcls.desc, rt_pw)
|
||||
t.do_comment(None)
|
||||
t.expect('(Y/n): ', 'y')
|
||||
t.written_to_file('Signed transaction')
|
||||
return t
|
||||
|
||||
def data_tx1_send(self):
|
||||
return self._data_tx_send()
|
||||
|
||||
def _data_tx_send(self):
|
||||
fn = self.get_file_with_ext('sigtx')
|
||||
t = self.spawn('mmgen-txsend', ['-q', '-d', self.tmpdir, '--bob', fn])
|
||||
t.expect('view: ', 'n')
|
||||
t.expect('(y/N): ', '\n')
|
||||
t.expect('to confirm: ', 'YES\n')
|
||||
t.written_to_file('Sent transaction')
|
||||
return t
|
||||
|
||||
def data_tx1_chk(self):
|
||||
return self._data_tx_chk(sample1.encode().hex())
|
||||
|
||||
def data_tx2_do(self):
|
||||
return self._data_tx_do('2', 'B:4', 'B:5', 'hexdata', sample2, 'v')
|
||||
|
||||
def data_tx2_chk(self):
|
||||
return self._data_tx_chk(sample2)
|
||||
|
||||
def _data_tx_do(self, src, dest, chg, pfx, sample, view):
|
||||
t = self.user_txdo(
|
||||
user = 'bob',
|
||||
fee = rtFee[0],
|
||||
outputs_cl = [f'{self.sid}:{dest},1', f'{self.sid}:{chg}', f'{pfx}:{sample}'],
|
||||
outputs_list = src,
|
||||
add_comment = 'Transaction with OP_RETURN data',
|
||||
return_early = True)
|
||||
|
||||
t.view_tx(view)
|
||||
if view == 'v':
|
||||
t.expect(sample)
|
||||
t.expect('amount:')
|
||||
t.passphrase(dfl_wcls.desc, rt_pw)
|
||||
t.written_to_file('Signed transaction')
|
||||
self._do_confirm_send(t)
|
||||
t.expect('Transaction sent')
|
||||
return t
|
||||
|
||||
def _data_tx_chk(self, sample):
|
||||
mp = self._get_mempool(do_msg=True)
|
||||
assert len(mp) == 1
|
||||
self.write_to_tmpfile('data_tx1_id', mp[0]+'\n')
|
||||
tx_hex = self._do_cli(['getrawtransaction', mp[0]])
|
||||
tx = self._do_cli(['decoderawtransaction', tx_hex], decode_json=True)
|
||||
v0 = tx['vout'][0]
|
||||
assert v0['scriptPubKey']['hex'] == f'6a{(len(sample) // 2):02x}{sample}'
|
||||
assert v0['scriptPubKey']['type'] == 'nulldata'
|
||||
assert v0['value'] == "0.00000000"
|
||||
return 'ok'
|
||||
|
||||
def generate3(self):
|
||||
return self.generate(3)
|
||||
|
|
@ -10,7 +10,7 @@ from mmgen.tx import CompletedTX, UnsignedTX
|
|||
from mmgen.tx.file import MMGenTxFile
|
||||
from mmgen.cfg import Config
|
||||
|
||||
from ..include.common import cfg, qmsg, vmsg
|
||||
from ..include.common import cfg, qmsg, vmsg, gr_uc
|
||||
|
||||
async def do_txfile_test(desc, fns, cfg=cfg, check=False):
|
||||
qmsg(f'\n Testing CompletedTX initializer ({desc})')
|
||||
|
|
@ -103,3 +103,60 @@ class unit_tests:
|
|||
)
|
||||
ut.process_bad_data(bad_data)
|
||||
return True
|
||||
|
||||
def op_return_data(self, name, ut, desc='OpReturnData class'):
|
||||
from mmgen.proto.btc.tx.op_return_data import OpReturnData
|
||||
vecs = [
|
||||
'data:=:ETH.ETH:0x86d526d6624AbC0178cF7296cD538Ecc080A95F1:0/1/0',
|
||||
'hexdata:3d3a4554482e4554483a30783836643532366436363234416243303137'
|
||||
'38634637323936634435333845636330383041393546313a302f312f30',
|
||||
'hexdata:00010203040506',
|
||||
'data:a\n',
|
||||
'data:a\tb',
|
||||
'data:' + gr_uc[:24],
|
||||
]
|
||||
|
||||
assert OpReturnData(cfg._proto, vecs[0]) == OpReturnData(cfg._proto, vecs[1])
|
||||
|
||||
for vec in vecs:
|
||||
d = OpReturnData(cfg._proto, vec)
|
||||
assert d == OpReturnData(cfg._proto, repr(d)) # repr() must return a valid initializer
|
||||
assert isinstance(d, bytes)
|
||||
assert isinstance(str(d), str)
|
||||
vmsg('-' * 80)
|
||||
vmsg(vec)
|
||||
vmsg(repr(d))
|
||||
vmsg(d.hl())
|
||||
vmsg(d.hl(add_label=True))
|
||||
|
||||
bad_data = [
|
||||
'data:',
|
||||
'hexdata:',
|
||||
'data:' + ('x' * 81),
|
||||
'hexdata:' + ('deadbeef' * 20) + 'ee',
|
||||
'hex:0abc',
|
||||
'da:xyz',
|
||||
'hexdata:xyz',
|
||||
'hexdata:abcde',
|
||||
b'data:abc',
|
||||
]
|
||||
|
||||
def bad(n):
|
||||
return lambda: OpReturnData(cfg._proto, bad_data[n])
|
||||
|
||||
vmsg('-' * 80)
|
||||
vmsg('Testing error handling:')
|
||||
|
||||
ut.process_bad_data((
|
||||
('bad1', 'AssertionError', 'not in range', bad(0)),
|
||||
('bad2', 'AssertionError', 'not in range', bad(1)),
|
||||
('bad3', 'AssertionError', 'not in range', bad(2)),
|
||||
('bad4', 'AssertionError', 'not in range', bad(3)),
|
||||
('bad5', 'ValueError', 'must start', bad(4)),
|
||||
('bad6', 'ValueError', 'must start', bad(5)),
|
||||
('bad7', 'AssertionError', 'not in hex', bad(6)),
|
||||
('bad8', 'AssertionError', 'even', bad(7)),
|
||||
('bad9', 'AssertionError', 'a string', bad(8)),
|
||||
), pfx='')
|
||||
|
||||
return True
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue