mmgen-wallet/test/include/common.py

#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
# Copyright (C)2013-2021 The MMGen Project <mmgen@tuta.io>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""
common.py: Shared routines and data for the MMGen test suites
"""

class TestSuiteException(Exception): pass
class TestSuiteFatalException(Exception): pass

import os
from mmgen.common import *
from mmgen.devtools import *

def strip_ansi_escapes(s):
	import re
	return re.sub('\x1b\[[;0-9]+?m','',s)

ascii_uc   = ''.join(map(chr,list(range(65,91))))   # 26 chars
ascii_lc   = ''.join(map(chr,list(range(97,123))))  # 26 chars
lat_accent = ''.join(map(chr,list(range(192,383)))) # 191 chars, L,S
ru_uc = ''.join(map(chr,list(range(1040,1072)))) # 32 chars
gr_uc = ''.join(map(chr,list(range(913,930)) + list(range(931,940)))) # 26 chars (930 is ctrl char)
gr_uc_w_ctrl = ''.join(map(chr,list(range(913,940)))) # 27 chars, L,C
lat_cyr_gr = lat_accent[:130:5] + ru_uc + gr_uc # 84 chars
ascii_cyr_gr = ascii_uc + ru_uc + gr_uc # 84 chars

utf8_text      = '[α-$ample UTF-8 text-ω]' * 10   # 230 chars, L,N,P,S,Z
utf8_combining = '[α-$ámple UTF-8 téxt-ω]' * 10   # L,N,P,S,Z,M
utf8_ctrl      = '[α-$ample\nUTF-8\ntext-ω]' * 10 # L,N,P,S,Z,C

text_jp = '必要なのは、信用ではなく暗号化された証明に基づく電子取引システムであり、これにより希望する二者が信用できる第三者機関を介さずに直接取引できるよう' # 72 chars ('W'ide)
text_zh = '所以，我們非常需要這樣一種電子支付系統，它基於密碼學原理而不基於信用，使得任何達成一致的雙方，能夠直接進行支付，從而不需要協力廠商仲介的參與。。' # 72 chars ('F'ull + 'W'ide)

sample_text = 'The Times 03/Jan/2009 Chancellor on brink of second bailout for banks'
sample_mn = {
	'mmgen': { # 'able': 0, 'youth': 1625, 'after' == 'afternoon'[:5]
		'mn': 'able cast forgive master funny gaze after afternoon million paint moral youth',
		'hex': '0005685ab4e94cbe3b228cf92112bc5f',
	},
	'bip39': { # len('sun') < uniq_ss_len
		'mn': 'vessel ladder alter error federal sibling chat ability sun glass valve picture',
		'hex': 'f30f8c1da665478f49b001d94c5fc452',
	},
	'xmrseed': {
		'mn': 'viewpoint donuts ardent template unveil agile meant unafraid urgent athlete rustled mime azure jaded hawk baby jagged haystack baby jagged haystack ramped oncoming point template',
		'hex': 'e8164dda6d42bd1e261a3406b2038dcbddadbeefdeadbeefdeadbeefdeadbe0f',
	},
}

ref_kafile_pass = 'kafile password'
ref_kafile_hash_preset = '1'

def ts_urandom(n):
	assert n < 70, f'{n} > len(sample_text) (69 bytes)'
	return sample_text.encode()[:n] if g.test_suite_deterministic else os.urandom(n)

def getrandnum(n):
	return int(ts_urandom(n).hex(),16)

def getrandhex(n):
	return ts_urandom(n).hex()

def getrandnum_range(nbytes,rn_max):
	while True:
		rn = int(ts_urandom(nbytes).hex(),16)
		if rn < rn_max:
			return rn

def getrandstr(num_chars,no_space=False):
	n,m = (94,33) if no_space else (95,32)
	return ''.join([chr(i%n+m) for i in list(os.urandom(num_chars))])

def get_data_dir():
	return os.path.join('test','data_dir' + ('','-α')[bool(os.getenv('MMGEN_DEBUG_UTF8'))])

# Windows uses non-UTF8 encodings in filesystem, so use raw bytes here
def cleandir(d,do_msg=False):
	d_enc = d.encode()

	try:    files = os.listdir(d_enc)
	except: return

	from shutil import rmtree
	if do_msg:
		gmsg(f'Cleaning directory {d!r}')
	for f in files:
		try:
			os.unlink(os.path.join(d_enc,f))
		except:
			rmtree(os.path.join(d_enc,f),ignore_errors=True)

def mk_tmpdir(d):
	try: os.mkdir(d,0o755)
	except OSError as e:
		if e.errno != 17: raise
	else:
		vmsg(f'Created directory {d!r}')

def get_tmpfile(cfg,fn):
	return os.path.join(cfg['tmpdir'],fn)

def write_to_file(fn,data,binary=False):
	write_data_to_file( fn,
						data,
						quiet = True,
						binary = binary,
						ignore_opt_outdir = True )

def write_to_tmpfile(cfg,fn,data,binary=False):
	write_to_file(  os.path.join(cfg['tmpdir'],fn), data=data, binary=binary )

def read_from_file(fn,binary=False):
	from mmgen.util import get_data_from_file
	return get_data_from_file(fn,quiet=True,binary=binary)

def read_from_tmpfile(cfg,fn,binary=False):
	return read_from_file(os.path.join(cfg['tmpdir'],fn),binary=binary)

def joinpath(*args,**kwargs):
	return os.path.join(*args,**kwargs)

def ok():
	if opt.profile: return
	if opt.verbose or opt.exact_output:
		gmsg('OK')
	else: msg(' OK')

def cmp_or_die(s,t,desc=None):
	if s != t:
		raise TestSuiteFatalException(
			(f'For {desc}:\n' if desc else '') +
			f'ERROR: recoded data:\n{t!r}\ndiffers from original data:\n{s!r}'
		)

def init_coverage():
	coverdir = os.path.join('test','trace')
	acc_file = os.path.join('test','trace.acc')
	try: os.mkdir(coverdir,0o755)
	except: pass
	return coverdir,acc_file

devnull_fh = open(('/dev/null','null.out')[g.platform == 'win'],'w')
def silence():
	if not (opt.verbose or getattr(opt,'exact_output',None)):
		g.stdout = g.stderr = devnull_fh

def end_silence():
	if not (opt.verbose or getattr(opt,'exact_output',None)):
		g.stdout = sys.stdout
		g.stderr = sys.stderr

def omsg(s):
	sys.stderr.write(s + '\n')
def omsg_r(s):
	sys.stderr.write(s)
	sys.stderr.flush()

def imsg(s):
	if opt.verbose or getattr(opt,'exact_output',None):
		omsg(s)
def imsg_r(s):
	if opt.verbose or getattr(opt,'exact_output',None):
		omsg_r(s)

def iqmsg(s):
	if not opt.quiet:
		omsg(s)
def iqmsg_r(s):
	if not opt.quiet:
		omsg_r(s)

def oqmsg(s):
	if not (opt.verbose or getattr(opt,'exact_output',None)):
		omsg(s)
def oqmsg_r(s):
	if not (opt.verbose or getattr(opt,'exact_output',None)):
		omsg_r(s)

def start_test_daemons(*network_ids,remove_datadir=False):
	if not opt.no_daemon_autostart:
		return test_daemons_ops(*network_ids,op='start',remove_datadir=remove_datadir)

def stop_test_daemons(*network_ids):
	if not opt.no_daemon_stop:
		return test_daemons_ops(*network_ids,op='stop')

def restart_test_daemons(*network_ids,remove_datadir=False):
	if not stop_test_daemons(*network_ids):
		return False
	return start_test_daemons(*network_ids,remove_datadir=remove_datadir)

def test_daemons_ops(*network_ids,op,remove_datadir=False):
	if not opt.no_daemon_autostart:
		from mmgen.daemon import CoinDaemon
		silent = not opt.verbose and not getattr(opt,'exact_output',False)
		ret = False
		for network_id in network_ids:
			d = CoinDaemon(network_id,test_suite=True,daemon_id=g.daemon_id)
			if remove_datadir:
				d.stop(silent=True)
				d.remove_datadir()
			ret = d.cmd(op,silent=silent)
		return ret