@@ -20,12 +20,9 @@ from .base import Base, decodeScriptPubKey
class Completed(Base, TxBase.Completed):
fn_fee_unit = 'satoshi'
- def decode_tx_usr_data(self):
+ def get_tx_usr_data(self):
if o := self.data_output:
- try:
- return o.data.decode()
- except UnicodeDecodeError:
- pass
+ return o.data
# check signature and witness data
def check_sigs(self): # return True if sigs found, False otherwise; raise exception on error
@@ -47,7 +47,7 @@ class Memo:
}
@classmethod
- def is_partial_memo(cls, s):
+ def is_partial_memo(cls, bytes_data):
import re
ops = {
'swap': ('SWAP', 's', '='),
@@ -60,7 +60,7 @@ class Memo:
'misc': ('BOND', 'UNBOND', 'LEAVE', 'MIGRATE', 'NOOP', 'DONATE', 'RESERVE'),
pat = r'^(' + '|'.join('|'.join(pats) for pats in ops.values()) + r'):\S\S+'
- return bool(re.search(pat, str(s)))
+ return bool(re.search(pat.encode(), bytes_data))
def parse(cls, s):
@@ -70,10 +70,11 @@ class Completed(Base):
return cls
def check_swap_memo(self):
- if text := self.decode_tx_usr_data():
+ if data := self.get_tx_usr_data():
from ..swap.proto.thorchain.memo import Memo
- if Memo.is_partial_memo(text):
+ if Memo.is_partial_memo(data):
from ..protocol import init_proto
+ text = data.decode('ascii')
p = Memo.parse(text)
assert p.function == 'SWAP', f'‘{p.function}’: unsupported function in swap memo ‘{text}’'
assert p.chain == p.asset, f'{p.chain} != {p.asset}: chain/asset mismatch in swap memo ‘{text}’'
@@ -74,9 +74,9 @@ class TxInfo:
if tx.is_swap:
from ..swap.proto.thorchain.memo import Memo, proto_name
- text = tx.data_output.data.decode()
- p = Memo.parse(text)
+ data = tx.data_output.data
+ p = Memo.parse(data.decode('ascii'))
yield ' {} {}\n'.format(magenta('DEX Protocol:'), blue(proto_name))
yield ' Swap: {}\n'.format(orange(f'{tx.proto.coin} => {p.asset}'))
yield ' Dest: {}{}\n'.format(
@@ -222,7 +222,7 @@ class unit_tests:
'BOND:xz',
):
vmsg(f' pass: {vec}')
- assert Memo.is_partial_memo(vec), vec
+ assert Memo.is_partial_memo(vec.encode('ascii')), vec
for vec in (
'=',
@@ -232,7 +232,7 @@ class unit_tests:
'SWAP:a',
vmsg(f' fail: {vec}')
- assert not Memo.is_partial_memo(vec), vec
+ assert not Memo.is_partial_memo(vec.encode('ascii')), vec
vmsg('\nTesting error handling:')