common.py 7.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230
  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 mmgen.common import *
  25. from mmgen.devtools import *
  26. def strip_ansi_escapes(s):
  27. import re
  28. return re.sub('\x1b' + r'\[[;0-9]+?m','',s)
  29. ascii_uc = ''.join(map(chr,list(range(65,91)))) # 26 chars
  30. ascii_lc = ''.join(map(chr,list(range(97,123)))) # 26 chars
  31. lat_accent = ''.join(map(chr,list(range(192,383)))) # 191 chars, L,S
  32. ru_uc = ''.join(map(chr,list(range(1040,1072)))) # 32 chars
  33. gr_uc = ''.join(map(chr,list(range(913,930)) + list(range(931,940)))) # 26 chars (930 is ctrl char)
  34. gr_uc_w_ctrl = ''.join(map(chr,list(range(913,940)))) # 27 chars, L,C
  35. lat_cyr_gr = lat_accent[:130:5] + ru_uc + gr_uc # 84 chars
  36. ascii_cyr_gr = ascii_uc + ru_uc + gr_uc # 84 chars
  37. utf8_text = '[α-$ample UTF-8 text-ω]' * 10 # 230 chars, L,N,P,S,Z
  38. utf8_combining = '[α-$ámple UTF-8 téxt-ω]' * 10 # L,N,P,S,Z,M
  39. utf8_ctrl = '[α-$ample\nUTF-8\ntext-ω]' * 10 # L,N,P,S,Z,C
  40. text_jp = '必要なのは、信用ではなく暗号化された証明に基づく電子取引システムであり、これにより希望する二者が信用できる第三者機関を介さずに直接取引できるよう' # 72 chars ('W'ide)
  41. text_zh = '所以,我們非常需要這樣一種電子支付系統,它基於密碼學原理而不基於信用,使得任何達成一致的雙方,能夠直接進行支付,從而不需要協力廠商仲介的參與。。' # 72 chars ('F'ull + 'W'ide)
  42. sample_text = 'The Times 03/Jan/2009 Chancellor on brink of second bailout for banks'
  43. sample_mn = {
  44. 'mmgen': { # 'able': 0, 'youth': 1625, 'after' == 'afternoon'[:5]
  45. 'mn': 'able cast forgive master funny gaze after afternoon million paint moral youth',
  46. 'hex': '0005685ab4e94cbe3b228cf92112bc5f',
  47. },
  48. 'bip39': { # len('sun') < uniq_ss_len
  49. 'mn': 'vessel ladder alter error federal sibling chat ability sun glass valve picture',
  50. 'hex': 'f30f8c1da665478f49b001d94c5fc452',
  51. },
  52. 'xmrseed': {
  53. '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',
  54. 'hex': 'e8164dda6d42bd1e261a3406b2038dcbddadbeefdeadbeefdeadbeefdeadbe0f',
  55. },
  56. }
  57. ref_kafile_pass = 'kafile password'
  58. ref_kafile_hash_preset = '1'
  59. def getrand(n):
  60. if g.test_suite_deterministic:
  61. from mmgen.crypto import fake_urandom
  62. return fake_urandom(n)
  63. else:
  64. return os.urandom(n)
  65. def getrandnum(n):
  66. return int(getrand(n).hex(),16)
  67. def getrandhex(n):
  68. return getrand(n).hex()
  69. def getrandnum_range(nbytes,rn_max):
  70. while True:
  71. rn = int(getrand(nbytes).hex(),16)
  72. if rn < rn_max:
  73. return rn
  74. def getrandstr(num_chars,no_space=False):
  75. n,m = (94,33) if no_space else (95,32)
  76. return ''.join( chr(i % n + m) for i in list(getrand(num_chars)) )
  77. def get_data_dir():
  78. return os.path.join('test','data_dir' + ('','-α')[bool(os.getenv('MMGEN_DEBUG_UTF8'))])
  79. # Windows uses non-UTF8 encodings in filesystem, so use raw bytes here
  80. def cleandir(d,do_msg=False):
  81. d_enc = d.encode()
  82. try: files = os.listdir(d_enc)
  83. except: return
  84. from shutil import rmtree
  85. if do_msg:
  86. gmsg(f'Cleaning directory {d!r}')
  87. for f in files:
  88. try:
  89. os.unlink(os.path.join(d_enc,f))
  90. except:
  91. rmtree(os.path.join(d_enc,f),ignore_errors=True)
  92. def mk_tmpdir(d):
  93. try: os.mkdir(d,0o755)
  94. except OSError as e:
  95. if e.errno != 17: raise
  96. else:
  97. vmsg(f'Created directory {d!r}')
  98. def get_tmpfile(cfg,fn):
  99. return os.path.join(cfg['tmpdir'],fn)
  100. def write_to_file(fn,data,binary=False):
  101. write_data_to_file( fn,
  102. data,
  103. quiet = True,
  104. binary = binary,
  105. ignore_opt_outdir = True )
  106. def write_to_tmpfile(cfg,fn,data,binary=False):
  107. write_to_file( os.path.join(cfg['tmpdir'],fn), data=data, binary=binary )
  108. def read_from_file(fn,binary=False):
  109. from mmgen.util import get_data_from_file
  110. return get_data_from_file(fn,quiet=True,binary=binary)
  111. def read_from_tmpfile(cfg,fn,binary=False):
  112. return read_from_file(os.path.join(cfg['tmpdir'],fn),binary=binary)
  113. def joinpath(*args,**kwargs):
  114. return os.path.join(*args,**kwargs)
  115. def ok():
  116. if opt.profile:
  117. return
  118. if opt.verbose or opt.exact_output:
  119. gmsg('OK')
  120. else:
  121. msg(' OK')
  122. def cmp_or_die(s,t,desc=None):
  123. if s != t:
  124. raise TestSuiteFatalException(
  125. (f'For {desc}:\n' if desc else '') +
  126. f'ERROR: recoded data:\n{t!r}\ndiffers from original data:\n{s!r}'
  127. )
  128. def init_coverage():
  129. coverdir = os.path.join('test','trace')
  130. acc_file = os.path.join('test','trace.acc')
  131. try: os.mkdir(coverdir,0o755)
  132. except: pass
  133. return coverdir,acc_file
  134. devnull_fh = open(('/dev/null','null.out')[g.platform == 'win'],'w')
  135. def silence():
  136. if not (opt.verbose or getattr(opt,'exact_output',None)):
  137. g.stdout = g.stderr = devnull_fh
  138. def end_silence():
  139. if not (opt.verbose or getattr(opt,'exact_output',None)):
  140. g.stdout = sys.stdout
  141. g.stderr = sys.stderr
  142. def omsg(s):
  143. sys.stderr.write(s + '\n')
  144. def omsg_r(s):
  145. sys.stderr.write(s)
  146. sys.stderr.flush()
  147. def imsg(s):
  148. if opt.verbose or getattr(opt,'exact_output',None):
  149. omsg(s)
  150. def imsg_r(s):
  151. if opt.verbose or getattr(opt,'exact_output',None):
  152. omsg_r(s)
  153. def iqmsg(s):
  154. if not opt.quiet:
  155. omsg(s)
  156. def iqmsg_r(s):
  157. if not opt.quiet:
  158. omsg_r(s)
  159. def oqmsg(s):
  160. if not (opt.verbose or getattr(opt,'exact_output',None)):
  161. omsg(s)
  162. def oqmsg_r(s):
  163. if not (opt.verbose or getattr(opt,'exact_output',None)):
  164. omsg_r(s)
  165. def end_msg(t):
  166. omsg(green(
  167. 'All requested tests finished OK' +
  168. ('' if g.test_suite_deterministic else f', elapsed time: {t//60:02d}:{t%60:02d}')
  169. ))
  170. def start_test_daemons(*network_ids,remove_datadir=False):
  171. if not opt.no_daemon_autostart:
  172. return test_daemons_ops(*network_ids,op='start',remove_datadir=remove_datadir)
  173. def stop_test_daemons(*network_ids):
  174. if not opt.no_daemon_stop:
  175. return test_daemons_ops(*network_ids,op='stop')
  176. def restart_test_daemons(*network_ids,remove_datadir=False):
  177. if not stop_test_daemons(*network_ids):
  178. return False
  179. return start_test_daemons(*network_ids,remove_datadir=remove_datadir)
  180. def test_daemons_ops(*network_ids,op,remove_datadir=False):
  181. if not opt.no_daemon_autostart:
  182. from mmgen.daemon import CoinDaemon
  183. silent = not opt.verbose and not getattr(opt,'exact_output',False)
  184. ret = False
  185. for network_id in network_ids:
  186. d = CoinDaemon(network_id,test_suite=True,daemon_id=g.daemon_id)
  187. if remove_datadir:
  188. d.stop(silent=True)
  189. d.remove_datadir()
  190. ret = d.cmd(op,silent=silent)
  191. return ret