mmgen/mmgen-wallet
1
Fork 0
Code Releases 28 Wiki Activity

fixes and cleanups

Browse source
This commit is contained in:
The MMGen Project 2025-03-04 09:51:05 +00:00
commit 809856c07d
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
7 changed files with 59 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

3
mmgen/amt.py
View file

@ -48,7 +48,7 @@ class CoinAmt(Decimal, Hilite, InitErrors): # abstract class
try:
if from_unit:
assert from_unit in cls.units, f'{from_unit!r}: unrecognized coin unit for {cls.__name__}'
assert type(num) is int, 'value is not an integer'
assert isinstance(num, int), 'value is not an integer'
me = Decimal.__new__(cls, num * getattr(cls, from_unit))
elif from_decimal:
assert isinstance(num, Decimal), f'number must be of type Decimal, not {type(num).__name__})'
@ -157,6 +157,7 @@ class CoinAmt(Decimal, Hilite, InitErrors): # abstract class
self.method_not_implemented()
def is_coin_amt(proto, num, from_unit=None, from_decimal=False):
assert proto.coin_amt, 'proto.coin_amt is None! Did you call init_proto() with ‘need_amt’?'
return get_obj(proto.coin_amt, num=num, from_unit=from_unit, from_decimal=from_decimal, silent=True, return_bool=True)
class BTCAmt(CoinAmt):

19
mmgen/swap/proto/thorchain/memo.py
View file

@ -12,6 +12,8 @@
swap.proto.thorchain.memo: THORChain swap protocol memo class
"""
from ....util import die
from . import name as proto_name
class Memo:
@ -65,14 +67,13 @@ class Memo:
All fields are validated, excluding address (cannot validate, since network is unknown)
"""
from collections import namedtuple
from ....exception import SwapMemoParseError
from ....util import is_int
def get_item(desc):
try:
return fields.pop(0)
except IndexError as e:
raise SwapMemoParseError(f'malformed {proto_name} memo (missing {desc} field)') from e
except IndexError:
die('SwapMemoParseError', f'malformed {proto_name} memo (missing {desc} field)')
def get_id(data, item, desc):
if item in data:
@ -80,12 +81,12 @@ class Memo:
rev_data = {v:k for k,v in data.items()}
if item in rev_data:
return rev_data[item]
raise SwapMemoParseError(f'{item!r}: unrecognized {proto_name} {desc} abbreviation')
die('SwapMemoParseError', f'{item!r}: unrecognized {proto_name} {desc} abbreviation')
fields = str(s).split(':')
if len(fields) < 4:
raise SwapMemoParseError('memo must contain at least 4 comma-separated fields')
die('SwapMemoParseError', 'memo must contain at least 4 comma-separated fields')
function = get_id(cls.function_abbrevs, get_item('function'), 'function')
@ -98,15 +99,15 @@ class Memo:
try:
limit, interval, quantity = lsq.split('/')
except ValueError as e:
raise SwapMemoParseError(f'malformed memo (failed to parse {desc} field) [{lsq}]') from e
except ValueError:
die('SwapMemoParseError', f'malformed memo (failed to parse {desc} field) [{lsq}]')
for n in (limit, interval, quantity):
if not is_int(n):
raise SwapMemoParseError(f'malformed memo (non-integer in {desc} field [{lsq}])')
die('SwapMemoParseError', f'malformed memo (non-integer in {desc} field [{lsq}])')
if fields:
raise SwapMemoParseError('malformed memo (unrecognized extra data)')
die('SwapMemoParseError', 'malformed memo (unrecognized extra data)')
ret = namedtuple(
'parsed_memo',

16
mmgen/swap/proto/thorchain/midgard.py
View file

@ -58,20 +58,23 @@ class Midgard:
c = self.in_amt.to_unit('satoshi'))
self.result = self.rpc.get(self.get_str)
self.data = json.loads(self.result.content)
if not 'expiry' in self.data:
from ....util import pp_fmt, die
die(2, pp_fmt(self.data))
def format_quote(self):
from ....util import make_timestr, pp_fmt, die
from ....util import make_timestr
from ....util2 import format_elapsed_hr
from ....color import blue, cyan, pink, orange
from . import name
d = self.data
if not 'expiry' in d:
die(2, pp_fmt(d))
tx = self.tx
in_coin = tx.send_proto.coin
out_coin = tx.recv_proto.coin
in_amt = self.in_amt
out_amt = tx.recv_proto.coin_amt(int(d['expected_amount_out']), from_unit='satoshi')
min_in_amt = tx.send_proto.coin_amt(int(d['recommended_min_amount_in']), from_unit='satoshi')
gas_unit = {
'satsperbyte': 'sat/byte',
@ -82,16 +85,17 @@ class Midgard:
fees_pct_disp = str(fees['total_bps'] / 100) + '%'
slip_pct_disp = str(fees['slippage_bps'] / 100) + '%'
hdr = f'SWAP QUOTE (source: {self.rpc.host})'
return f"""
{cyan(hdr)}
Protocol: {blue(name)}
Direction: {orange(f'{in_coin} => {out_coin}')}
Vault address: {cyan(d['inbound_address'])}
Quote expires: {pink(elapsed_disp)} [{make_timestr(d['expiry'])}]
Amount in: {self.in_amt.hl()} {in_coin}
Amount in: {in_amt.hl()} {in_coin}
Expected amount out: {out_amt.hl()} {out_coin}
Rate: {(out_amt / self.in_amt).hl()} {out_coin}/{in_coin}
Reverse rate: {(self.in_amt / out_amt).hl()} {in_coin}/{out_coin}
Rate: {(out_amt / in_amt).hl()} {out_coin}/{in_coin}
Reverse rate: {(in_amt / out_amt).hl()} {in_coin}/{out_coin}
Recommended minimum in amount: {min_in_amt.hl()} {in_coin}
Recommended fee: {pink(d['recommended_gas_rate'])} {pink(gas_unit)}
Fees:

3
mmgen/tx/new_swap.py
View file

@ -17,9 +17,6 @@ from .new import New
class NewSwap(New):
desc = 'swap transaction'
async def process_swap_cmdline_args(self, cmd_args, addrfiles):
raise NotImplementedError(f'Swap not implemented for protocol {self.proto.__name__}')
def update_vault_output(self, amt):
import importlib
sp = importlib.import_module(f'mmgen.swap.proto.{self.swap_proto}')

20
test/cmdtest_d/ct_swap.py
View file

@ -317,10 +317,10 @@ class CmdTestSwap(CmdTestRegtest, CmdTestAutosignThreaded):
return self.addrimport('bob', mmtypes=['C'], proto=self.protos[2])
def fund_bob_send(self):
return self._fund_bob(2, 'C', '500')
return self._fund_bob(2, 'C', '5')
def bob_bal_send(self):
return self._bob_bal(2, '500')
return self._bob_bal(2, '5')
def setup_recv_coin(self):
return self._setup(proto=self.protos[1], remove_datadir=False)
@ -332,10 +332,10 @@ class CmdTestSwap(CmdTestRegtest, CmdTestAutosignThreaded):
return self._addrimport_bob(1)
def fund_bob_recv1(self):
return self._fund_bob(1, 'S', '500')
return self._fund_bob(1, 'S', '5')
def fund_bob_recv2(self):
return self._fund_bob(1, 'B', '500')
return self._fund_bob(1, 'B', '5')
def addrgen_bob_recv_subwallet(self):
return self._addrgen_bob(1, ['C', 'B'], subseed_idx='29L')
@ -343,7 +343,7 @@ class CmdTestSwap(CmdTestRegtest, CmdTestAutosignThreaded):
def addrimport_bob_recv_subwallet(self):
return self._subwallet_addrimport('bob', '29L', ['C', 'B'], proto=self.protos[1])
def fund_bob_recv_subwallet(self, proto_idx=1, amt='500'):
def fund_bob_recv_subwallet(self, proto_idx=1, amt='5'):
coin_arg = f'--coin={self.protos[proto_idx].coin}'
t = self.spawn('mmgen-tool', ['--bob', coin_arg, 'listaddresses'])
addr = [s for s in strip_ansi_escapes(t.read()).splitlines() if 'C:1 No' in s][0].split()[3]
@ -351,7 +351,7 @@ class CmdTestSwap(CmdTestRegtest, CmdTestAutosignThreaded):
return t
def bob_bal_recv(self):
return self._bob_bal(1, '1500')
return self._bob_bal(1, '15')
def _swaptxcreate_ui_common(
self,
@ -487,7 +487,7 @@ class CmdTestSwap(CmdTestRegtest, CmdTestAutosignThreaded):
def swaptxsign1_create(self):
self.get_file_with_ext('rawtx', delete_all=True)
return self._swaptxcreate_ui_common(
self._swaptxcreate(['LTC', '5.4321', f'{self.sid}:S:2', 'BCH', f'{self.sid}:C:2']))
self._swaptxcreate(['LTC', '4.321', f'{self.sid}:S:2', 'BCH', f'{self.sid}:C:2']))
def swaptxsign1(self):
return self._swaptxsign()
@ -605,17 +605,17 @@ class CmdTestSwap(CmdTestRegtest, CmdTestAutosignThreaded):
return self._generate_for_proto(2)
def swap_bal1(self):
return self._bob_bal(1, '1494.56784238')
return self._bob_bal(1, '10.67894238')
def swap_bal2(self):
return self._bob_bal(1, '1382.79038152')
return self._bob_bal(1, '8.90148152')
def swap_bal3(self):
return self._bob_bal(0, '999.99990407')
def swaptxsign1_do(self):
return self._swaptxcreate_ui_common(
self._swaptxcreate(['LTC', '111.777444', f'{self.sid}:B:2', 'BCH', f'{self.sid}:C:2'], action='txdo'),
self._swaptxcreate(['LTC', '1.777444', f'{self.sid}:B:2', 'BCH', f'{self.sid}:C:2'], action='txdo'),
sign_and_send = True,
file_desc = 'Sent transaction')

2
test/include/unit_test.py
View file

@ -144,7 +144,7 @@ class UnitTestHelpers:
asyncio.run(ret)
except Exception as e:
exc = type(e).__name__
emsg = e.args[0]
emsg = e.args[0] if e.args else '(unspecified error)'
cfg._util.vmsg(f' {exc:{exc_w}} [{emsg}]')
assert exc == exc_chk, m_exc.format(exc, exc_chk)
assert re.search(emsg_chk, emsg), m_err.format(emsg, emsg_chk)

48
test/modtest_d/ut_tx.py
View file

@ -180,23 +180,25 @@ class unit_tests:
proto = init_proto(cfg, coin)
addr = make_burn_addr(proto, addrtype)
vmsg('\nTesting memo initialization:')
m = Memo(proto, addr)
vmsg(f'str(memo): {m}')
vmsg(f'repr(memo): {m!r}')
if True:
vmsg('\nTesting memo initialization:')
m = Memo(proto, addr)
vmsg(f'str(memo): {m}')
vmsg(f'repr(memo): {m!r}')
vmsg('\nTesting memo parsing:')
p = Memo.parse(m)
from pprint import pformat
vmsg(pformat(p._asdict()))
assert p.proto == 'THORChain'
assert p.function == 'SWAP'
assert p.chain == coin.upper()
assert p.asset == coin.upper()
assert p.address == addr.views[addr.view_pref]
assert p.trade_limit == 0
assert p.stream_interval == 1
assert p.stream_quantity == 0 # auto
p = Memo.parse(m)
vmsg('\nTesting memo parsing:')
from pprint import pformat
vmsg(pformat(p._asdict()))
assert p.proto == 'THORChain'
assert p.function == 'SWAP'
assert p.chain == coin.upper()
assert p.asset == coin.upper()
assert p.address == addr.views[addr.view_pref]
assert p.trade_limit == 0
assert p.stream_interval == 1
assert p.stream_quantity == 0 # auto
vmsg('\nTesting is_partial_memo():')
for vec in (
@ -230,13 +232,13 @@ class unit_tests:
return lambda: Memo.parse(s)
ut.process_bad_data((
('bad1', 'SwapMemoParseError', 'must contain', bad('x')),
('bad2', 'SwapMemoParseError', 'must contain', bad('y:z:x')),
('bad3', 'SwapMemoParseError', 'function abbrev', bad('z:l:foobar:0/1/0')),
('bad4', 'SwapMemoParseError', 'asset abbrev', bad('=:x:foobar:0/1/0')),
('bad5', 'SwapMemoParseError', 'failed to parse', bad('=:l:foobar:n')),
('bad6', 'SwapMemoParseError', 'non-integer', bad('=:l:foobar:x/1/0')),
('bad7', 'SwapMemoParseError', 'extra', bad('=:l:foobar:0/1/0:x')),
('bad1', 'SwapMemoParseError', 'must contain', bad('x')),
('bad2', 'SwapMemoParseError', 'must contain', bad('y:z:x')),
('bad3', 'SwapMemoParseError', 'function abbrev', bad('z:l:foobar:0/1/0')),
('bad4', 'SwapMemoParseError', 'asset abbrev', bad('=:x:foobar:0/1/0')),
('bad5', 'SwapMemoParseError', 'failed to parse', bad('=:l:foobar:n')),
('bad6', 'SwapMemoParseError', 'non-integer', bad('=:l:foobar:x/1/0')),
('bad7', 'SwapMemoParseError', 'extra', bad('=:l:foobar:0/1/0:x')),
), pfx='')
return True