common.py 7.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254
  1. #!/usr/bin/env python3
  2. #
  3. # mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
  4. # Copyright (C)2013-2022 The MMGen Project <mmgen@tuta.io>
  5. #
  6. # This program is free software: you can redistribute it and/or modify
  7. # it under the terms of the GNU General Public License as published by
  8. # the Free Software Foundation, either version 3 of the License, or
  9. # (at your option) any later version.
  10. #
  11. # This program is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14. # GNU General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU General Public License
  17. # along with this program. If not, see <http://www.gnu.org/licenses/>.
  18. """
  19. common.py: Shared routines and data for the MMGen test suites
  20. """
  21. class TestSuiteException(Exception): pass
  22. class TestSuiteFatalException(Exception): pass
  23. import os
  24. from subprocess import run,PIPE
  25. from mmgen.common import *
  26. from mmgen.devtools import *
  27. from mmgen.fileutil import write_data_to_file,get_data_from_file
  28. def strip_ansi_escapes(s):
  29. import re
  30. return re.sub('\x1b' + r'\[[;0-9]+?m','',s)
  31. ascii_uc = ''.join(map(chr,list(range(65,91)))) # 26 chars
  32. ascii_lc = ''.join(map(chr,list(range(97,123)))) # 26 chars
  33. lat_accent = ''.join(map(chr,list(range(192,383)))) # 191 chars, L,S
  34. ru_uc = ''.join(map(chr,list(range(1040,1072)))) # 32 chars
  35. gr_uc = ''.join(map(chr,list(range(913,930)) + list(range(931,940)))) # 26 chars (930 is ctrl char)
  36. gr_uc_w_ctrl = ''.join(map(chr,list(range(913,940)))) # 27 chars, L,C
  37. lat_cyr_gr = lat_accent[:130:5] + ru_uc + gr_uc # 84 chars
  38. ascii_cyr_gr = ascii_uc + ru_uc + gr_uc # 84 chars
  39. utf8_text = '[α-$ample UTF-8 text-ω]' * 10 # 230 chars, L,N,P,S,Z
  40. utf8_combining = '[α-$ámple UTF-8 téxt-ω]' * 10 # L,N,P,S,Z,M
  41. utf8_ctrl = '[α-$ample\nUTF-8\ntext-ω]' * 10 # L,N,P,S,Z,C
  42. text_jp = '必要なのは、信用ではなく暗号化された証明に基づく電子取引システムであり、これにより希望する二者が信用できる第三者機関を介さずに直接取引できるよう' # 72 chars ('W'ide)
  43. text_zh = '所以,我們非常需要這樣一種電子支付系統,它基於密碼學原理而不基於信用,使得任何達成一致的雙方,能夠直接進行支付,從而不需要協力廠商仲介的參與。。' # 72 chars ('F'ull + 'W'ide)
  44. sample_text = 'The Times 03/Jan/2009 Chancellor on brink of second bailout for banks'
  45. sample_mn = {
  46. 'mmgen': { # 'able': 0, 'youth': 1625, 'after' == 'afternoon'[:5]
  47. 'mn': 'able cast forgive master funny gaze after afternoon million paint moral youth',
  48. 'hex': '0005685ab4e94cbe3b228cf92112bc5f',
  49. },
  50. 'bip39': { # len('sun') < uniq_ss_len
  51. 'mn': 'vessel ladder alter error federal sibling chat ability sun glass valve picture',
  52. 'hex': 'f30f8c1da665478f49b001d94c5fc452',
  53. },
  54. 'xmrseed': {
  55. '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',
  56. 'hex': 'e8164dda6d42bd1e261a3406b2038dcbddadbeefdeadbeefdeadbeefdeadbe0f',
  57. },
  58. }
  59. ref_kafile_pass = 'kafile password'
  60. ref_kafile_hash_preset = '1'
  61. def getrand(n):
  62. if g.test_suite_deterministic:
  63. from mmgen.crypto import fake_urandom
  64. return fake_urandom(n)
  65. else:
  66. return os.urandom(n)
  67. def getrandnum(n):
  68. return int(getrand(n).hex(),16)
  69. def getrandhex(n):
  70. return getrand(n).hex()
  71. def getrandnum_range(nbytes,rn_max):
  72. while True:
  73. rn = int(getrand(nbytes).hex(),16)
  74. if rn < rn_max:
  75. return rn
  76. def getrandstr(num_chars,no_space=False):
  77. n,m = (94,33) if no_space else (95,32)
  78. return ''.join( chr(i % n + m) for i in list(getrand(num_chars)) )
  79. def get_data_dir():
  80. return os.path.join('test','data_dir' + ('','-α')[bool(os.getenv('MMGEN_DEBUG_UTF8'))])
  81. # Windows uses non-UTF8 encodings in filesystem, so use raw bytes here
  82. def cleandir(d,do_msg=False):
  83. d_enc = d.encode()
  84. try: files = os.listdir(d_enc)
  85. except: return
  86. from shutil import rmtree
  87. if do_msg:
  88. gmsg(f'Cleaning directory {d!r}')
  89. for f in files:
  90. try:
  91. os.unlink(os.path.join(d_enc,f))
  92. except:
  93. rmtree(os.path.join(d_enc,f),ignore_errors=True)
  94. def mk_tmpdir(d):
  95. try: os.mkdir(d,0o755)
  96. except OSError as e:
  97. if e.errno != 17:
  98. raise
  99. else:
  100. vmsg(f'Created directory {d!r}')
  101. def get_tmpfile(cfg,fn):
  102. return os.path.join(cfg['tmpdir'],fn)
  103. def write_to_file(fn,data,binary=False):
  104. write_data_to_file(
  105. fn,
  106. data,
  107. quiet = True,
  108. binary = binary,
  109. ignore_opt_outdir = True )
  110. def write_to_tmpfile(cfg,fn,data,binary=False):
  111. write_to_file( os.path.join(cfg['tmpdir'],fn), data=data, binary=binary )
  112. def read_from_file(fn,binary=False):
  113. return get_data_from_file(fn,quiet=True,binary=binary)
  114. def read_from_tmpfile(cfg,fn,binary=False):
  115. return read_from_file(os.path.join(cfg['tmpdir'],fn),binary=binary)
  116. def joinpath(*args,**kwargs):
  117. return os.path.join(*args,**kwargs)
  118. def ok():
  119. if opt.profile:
  120. return
  121. if opt.verbose or opt.exact_output:
  122. gmsg('OK')
  123. else:
  124. msg(' OK')
  125. def cmp_or_die(s,t,desc=None):
  126. if s != t:
  127. die( 'TestSuiteFatalException',
  128. (f'For {desc}:\n' if desc else '') +
  129. f'ERROR: recoded data:\n{t!r}\ndiffers from original data:\n{s!r}'
  130. )
  131. def init_coverage():
  132. coverdir = os.path.join('test','trace')
  133. acc_file = os.path.join('test','trace.acc')
  134. try: os.mkdir(coverdir,0o755)
  135. except: pass
  136. return coverdir,acc_file
  137. def silence():
  138. if not (opt.verbose or opt.exact_output):
  139. devnull_fn = ('/dev/null','null.out')[g.platform == 'win']
  140. g.stdout = g.stderr = open(devnull_fn,'w')
  141. def end_silence():
  142. if not (opt.verbose or opt.exact_output):
  143. g.stdout.close()
  144. g.stdout = sys.stdout
  145. g.stderr = sys.stderr
  146. def omsg(s):
  147. sys.stderr.write(s + '\n')
  148. def omsg_r(s):
  149. sys.stderr.write(s)
  150. sys.stderr.flush()
  151. def imsg(s):
  152. if opt.verbose or opt.exact_output:
  153. omsg(s)
  154. def imsg_r(s):
  155. if opt.verbose or opt.exact_output:
  156. omsg_r(s)
  157. def iqmsg(s):
  158. if not opt.quiet:
  159. omsg(s)
  160. def iqmsg_r(s):
  161. if not opt.quiet:
  162. omsg_r(s)
  163. def oqmsg(s):
  164. if not (opt.verbose or opt.exact_output):
  165. omsg(s)
  166. def oqmsg_r(s):
  167. if not (opt.verbose or opt.exact_output):
  168. omsg_r(s)
  169. def end_msg(t):
  170. omsg(green(
  171. 'All requested tests finished OK' +
  172. ('' if g.test_suite_deterministic else f', elapsed time: {t//60:02d}:{t%60:02d}')
  173. ))
  174. def start_test_daemons(*network_ids,remove_datadir=False):
  175. if not opt.no_daemon_autostart:
  176. return test_daemons_ops(*network_ids,op='start',remove_datadir=remove_datadir)
  177. def stop_test_daemons(*network_ids,force=False,remove_datadir=False):
  178. if force or not opt.no_daemon_stop:
  179. return test_daemons_ops(*network_ids,op='stop',remove_datadir=remove_datadir)
  180. def restart_test_daemons(*network_ids,remove_datadir=False):
  181. if not stop_test_daemons(*network_ids):
  182. return False
  183. return start_test_daemons(*network_ids,remove_datadir=remove_datadir)
  184. def test_daemons_ops(*network_ids,op,remove_datadir=False):
  185. if not opt.no_daemon_autostart:
  186. from mmgen.daemon import CoinDaemon
  187. silent = not (opt.verbose or opt.exact_output)
  188. ret = False
  189. for network_id in network_ids:
  190. d = CoinDaemon(network_id,test_suite=True,daemon_id=g.daemon_id)
  191. if remove_datadir:
  192. d.stop(silent=True)
  193. d.remove_datadir()
  194. ret = d.cmd(op,silent=silent)
  195. return ret
  196. tested_solc_ver = '0.8.7'
  197. def check_solc_ver():
  198. cmd = 'python3 scripts/create-token.py --check-solc-version'
  199. try:
  200. cp = run(cmd.split(),check=False,stdout=PIPE)
  201. except Exception as e:
  202. die(4,f'Unable to execute {cmd!r}: {e}')
  203. res = cp.stdout.decode().strip()
  204. if cp.returncode == 0:
  205. omsg(
  206. orange(f'Found supported solc version {res}') if res == tested_solc_ver else
  207. yellow(f'WARNING: solc version ({res}) does not match tested version ({tested_solc_ver})')
  208. )
  209. return True
  210. else:
  211. omsg(yellow('Warning: Solidity compiler (solc) could not be executed or has unsupported version'))
  212. omsg(res)
  213. return False