protocol.py: new init_proto() factory function

This commit is contained in:
The MMGen Project 2020-05-14 16:40:45 +00:00
commit 8a4fbab48d
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
10 changed files with 54 additions and 52 deletions

View file

@ -697,8 +697,8 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
else:
mmtype = MMGenAddrType(al_mmtype,on_fail='raise')
from .protocol import CoinProtocol
base_coin = CoinProtocol(al_coin or 'BTC',testnet=False).base_coin
from .protocol import init_proto
base_coin = init_proto(al_coin or 'BTC',testnet=False).base_coin
return base_coin,mmtype,tn
def check_coin_mismatch(base_coin): # die if addrfile coin doesn't match g.coin

View file

@ -428,13 +428,13 @@ class CoinInfo(object):
@classmethod
def verify_core_coin_data(cls,quiet=False,verbose=False):
from mmgen.protocol import CoinProtocol
from mmgen.protocol import CoinProtocol,init_proto
for network in ('mainnet','testnet'):
for coin in CoinProtocol.core_coins:
e = cls.get_entry(coin,network)
if e:
proto = CoinProtocol(coin,testnet=network=='testnet')
proto = init_proto(coin,testnet=network=='testnet')
cdata = (network,coin,e,type(proto).__name__,verbose)
if not quiet:
msg('Verifying {} {}'.format(coin.upper(),network))

View file

@ -353,8 +353,8 @@ class CoinDaemon(Daemon):
me.desc = 'test suite daemon'
rel_datadir = os.path.join('test','daemons',daemon_id)
else:
from .protocol import CoinProtocol
me.datadir = CoinProtocol(daemon_id,False).daemon_data_dir
from .protocol import init_proto
me.datadir = init_proto(daemon_id,False).daemon_data_dir
if test_suite:
me.datadir = os.path.abspath(os.path.join(os.getcwd(),rel_datadir))

View file

@ -113,7 +113,7 @@ exit_if_mswin('autosigning')
import mmgen.tx
from .txsign import txsign
from .protocol import CoinProtocol,init_coin
from .protocol import init_proto,init_coin
from .rpc import rpc_init
if g.test_suite:
@ -134,7 +134,7 @@ async def check_daemons_running():
coins = ['BTC']
for coin in coins:
g.proto = CoinProtocol(coin,g.testnet)
g.proto = init_proto(coin,g.testnet)
if g.proto.sign_mode == 'daemon':
if g.test_suite:
g.proto.daemon_data_dir = 'test/daemons/' + coin.lower()

View file

@ -87,14 +87,14 @@ def init_term_and_color():
init_color(num_colors=('auto',256)[bool(g.force_256_color)])
def override_globals_from_cfg_file(ucfg):
from .protocol import CoinProtocol
from .protocol import CoinProtocol,init_proto
for d in ucfg.parse():
val = d.value
if d.name in g.cfg_file_opts:
ns = d.name.split('_')
if ns[0] in CoinProtocol.coins:
nse,tn = (ns[2:],True) if len(ns) > 2 and ns[1] == 'testnet' else (ns[1:],False)
cls = CoinProtocol(ns[0],tn)
cls = init_proto(ns[0],tn)
attr = '_'.join(nse)
else:
cls = g
@ -259,11 +259,11 @@ def init(opts_data,add_opts=[],opt_filter=None,parse_only=False):
g.network = 'testnet' if g.testnet else 'mainnet'
from .protocol import init_genonly_altcoins,CoinProtocol
from .protocol import init_genonly_altcoins,init_proto
altcoin_trust_level = init_genonly_altcoins(g.coin)
# g.testnet is finalized, so we can set g.proto
g.proto = CoinProtocol(g.coin,g.testnet)
g.proto = init_proto(g.coin,g.testnet)
# this could have been set from long opts
if g.daemon_data_dir:
@ -293,7 +293,7 @@ def init(opts_data,add_opts=[],opt_filter=None,parse_only=False):
if g.bob or g.alice:
g.testnet = True
g.regtest = True
g.proto = CoinProtocol(g.coin,g.testnet)
g.proto = init_proto(g.coin,g.testnet)
g.rpc_host = 'localhost'
g.data_dir = os.path.join(g.data_dir_root,'regtest',g.coin.lower(),('alice','bob')[g.bob])
from .regtest import MMGenRegtest
@ -491,7 +491,7 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
def chk_coin(key,val,desc):
from .protocol import CoinProtocol
opt_is_in_list(val.lower(),list(CoinProtocol.coins.keys()),'coin')
opt_is_in_list(val.lower(),CoinProtocol.coins,'coin')
def chk_rbf(key,val,desc):
if not g.proto.cap('rbf'):

View file

@ -69,6 +69,19 @@ def _b58chk_decode(s):
finfo = namedtuple('fork_info',['height','hash','name','replayable'])
class CoinProtocol(MMGenObject):
proto_info = namedtuple('proto_info',['mainnet','testnet','trust_level']) # trust levels: see altcoin.py
coins = {
'btc': proto_info('Bitcoin', 'BitcoinTestnet', 5),
'bch': proto_info('BitcoinCash', 'BitcoinCashTestnet', 5),
'ltc': proto_info('Litecoin', 'LitecoinTestnet', 5),
'eth': proto_info('Ethereum', 'EthereumTestnet', 4),
'etc': proto_info('EthereumClassic', 'EthereumClassicTestnet', 4),
'zec': proto_info('Zcash', 'ZcashTestnet', 2),
'xmr': proto_info('Monero', 'MoneroTestnet', 4)
}
core_coins = tuple(coins.keys()) # coins may be added by init_genonly_altcoins(), so save
# chainparams.cpp
class BitcoinProtocol(MMGenObject):
name = 'bitcoin'
@ -118,7 +131,7 @@ class BitcoinProtocol(MMGenObject):
@staticmethod
def get_protocol_by_chain(chain):
return CoinProtocol(g.coin,{'mainnet':False,'testnet':True,'regtest':True}[chain])
return init_proto(g.coin,{'mainnet':False,'testnet':True,'regtest':True}[chain])
def cap(self,s): return s in self.caps
@ -425,32 +438,18 @@ class MoneroProtocol(DummyWIF,BitcoinProtocolAddrgen):
class MoneroTestnetProtocol(MoneroProtocol):
addr_ver_bytes = { '35': 'monero', '3f': 'monero_sub' }
class CoinProtocol(MMGenObject):
pi = namedtuple('proto_info',['main_cls','test_cls','trust_level']) # trust levels: see altcoin.py
coins = {
'btc': pi(BitcoinProtocol,BitcoinTestnetProtocol,5),
'bch': pi(BitcoinCashProtocol,BitcoinCashTestnetProtocol,5),
'ltc': pi(LitecoinProtocol,LitecoinTestnetProtocol,5),
'eth': pi(EthereumProtocol,EthereumTestnetProtocol,4),
'etc': pi(EthereumClassicProtocol,EthereumClassicTestnetProtocol,4),
'zec': pi(ZcashProtocol,ZcashTestnetProtocol,2),
'xmr': pi(MoneroProtocol,MoneroTestnetProtocol,4)
}
core_coins = tuple(coins.keys())
def __new__(cls,coin,testnet):
coin = coin.lower()
assert type(testnet) == bool
m = "{}: not a valid coin for network {}\nSupported coins: {}"
assert coin in cls.coins, m.format(coin.upper(),g.network.upper(),' '.join(cls.list_coins()))
proto = cls.coins[coin][testnet]
if hasattr(proto,'bech32_hrps'):
proto.bech32_hrp = proto.bech32_hrps[('testnet','regtest')[g.regtest]]
return proto()
@classmethod
def list_coins(cls):
return [c.upper() for c in cls.coins]
def init_proto(coin,testnet):
coin = coin.lower()
assert type(testnet) == bool, type(testnet)
if coin not in CoinProtocol.coins:
raise ValueError(
'{}: not a valid coin for network {}\nSupported coins: {}'.format(
coin.upper(),g.network.upper(),
' '.join(c.upper() for c in CoinProtocol.coins) ))
proto = globals()[CoinProtocol.coins[coin][testnet] + 'Protocol']
if hasattr(proto,'bech32_hrps'):
proto.bech32_hrp = proto.bech32_hrps[('testnet','regtest')[g.regtest]]
return proto()
def init_genonly_altcoins(usr_coin=None):
"""
@ -519,13 +518,13 @@ def make_init_genonly_altcoins_str(data):
out += make_proto(e,testnet=True)
tn_coins = [e.symbol for e in data['testnet']]
fs = "CoinProtocol.coins['{}'] = ({}Protocol,{})\n"
fs = "CoinProtocol.coins['{}'] = CoinProtocol.proto_info('{}',{},{})\n"
for e in data['mainnet']:
proto,coin = e.name,e.symbol
if proto[0] in '0123456789': proto = 'X_'+proto
if proto+'Protocol' in globals(): continue
if coin.lower() in CoinProtocol.coins: continue
out += fs.format(coin.lower(),proto,('None',proto+'TestnetProtocol')[coin in tn_coins])
out += fs.format(coin.lower(),proto,('None',f"'{proto}Testnet'")[coin in tn_coins],e.trust_level)
return out
def init_coin(coin,testnet=None):
@ -533,5 +532,5 @@ def init_coin(coin,testnet=None):
g.testnet = testnet
g.network = ('mainnet','testnet')[g.testnet]
g.coin = coin.upper()
g.proto = CoinProtocol(g.coin,g.testnet)
g.proto = init_proto(g.coin,g.testnet)
return g.proto

View file

@ -270,8 +270,8 @@ class MMGenRegtest(MMGenObject):
def fork(self,coin): # currently disabled
from .protocol import CoinProtocol
forks = CoinProtocol(coin,False).forks
from .protocol import init_proto
forks = init_proto(coin,False).forks
if not [f for f in forks if f[2] == g.coin.lower() and f[3] == True]:
die(1,"Coin {} is not a replayable fork of coin {}".format(g.coin,coin))

View file

@ -1195,7 +1195,10 @@ class tool_api(
"""The available coins"""
from .protocol import CoinProtocol
from .altcoin import CoinInfo
return sorted(set(CoinProtocol.list_coins() + [c.symbol for c in CoinInfo.get_supported_coins(g.network)]))
return sorted(set(
[c.upper() for c in CoinProtocol.coins]
+ [c.symbol for c in CoinInfo.get_supported_coins(g.network)]
))
@property
def coin(self):

View file

@ -421,7 +421,7 @@ elif not b and hasattr(arg2,'read'):
dump_test(a,ag,arg2)
elif a and b and type(arg2) == int:
if opt.all:
from mmgen.protocol import init_genonly_altcoins,CoinProtocol
from mmgen.protocol import CoinProtocol,init_genonly_altcoins
init_genonly_altcoins()
for coin in ci.external_tests[g.network][b.desc]:
if coin.lower() not in CoinProtocol.coins:

View file

@ -25,7 +25,7 @@ from decimal import Decimal
from mmgen.globalvars import g
from mmgen.opts import opt
from mmgen.util import die,gmsg,write_data_to_file
from mmgen.protocol import CoinProtocol
from mmgen.protocol import init_proto
from mmgen.addr import AddrList
from mmgen.wallet import MMGenWallet
from ..include.common import *
@ -344,7 +344,7 @@ class TestSuiteRegtest(TestSuiteBase,TestSuiteShared):
x='' if g.debug_utf8 else ''))
if mmtype == g.proto.mmtypes[0] and user == 'bob':
psave = g.proto
g.proto = CoinProtocol(g.coin,True)
g.proto = init_proto(g.coin,True)
self._add_comments_to_addr_file(addrfile,addrfile,use_labels=True)
g.proto = psave
t = self.spawn( 'mmgen-addrimport',
@ -568,7 +568,7 @@ class TestSuiteRegtest(TestSuiteBase,TestSuiteShared):
sid,self.altcoin_pfx,id_str,addr_range,x='' if g.debug_utf8 else '')
addrfile = get_file_with_ext(self._user_dir(user),ext,no_dot=True)
psave = g.proto
g.proto = CoinProtocol(g.coin,True)
g.proto = init_proto(g.coin,True)
silence()
addr = AddrList(addrfile).data[idx].addr
end_silence()
@ -810,7 +810,7 @@ class TestSuiteRegtest(TestSuiteBase,TestSuiteShared):
"MMGen address '{}' not found in tracking wallet".format(addr))
def alice_add_label_badaddr4(self):
addr = CoinProtocol(g.coin,True).pubhash2addr('00'*20,False) # testnet zero address
addr = init_proto(g.coin,True).pubhash2addr('00'*20,False) # testnet zero address
return self.alice_add_label_badaddr(addr,
"Address '{}' not found in tracking wallet".format(addr))