tests/pdtest

   1 #!/usr/bin/env /usr/bin/python3
   2
   3 import os
   4 import sys
   5 from getopt import getopt
   6 from tempfile import mkstemp
   7 from subprocess import Popen, PIPE
   8 from difflib import Differ
   9
  10 DEBUG = 0
  11 VERBOSE = False
  12
  13
  14 class E_syntax(Exception):
  15     pass
  16 class E_badline(Exception):
  17     pass
  18
  19 def INFO(msg, end='\n'):
  20     if VERBOSE:
  21         print(msg, end=end)
  22         sys.stdout.flush()
  23
  24
  25 def DBG(msg):
  26     if DEBUG:
  27         print(msg)
  28
  29
  30 def ERR(msg):
  31     print(msg, file=sys.stderr)
  32
  33
  34 def usage(msg=None):
  35     if msg:
  36         print(msg.strip() + '\n')
  37     print("""Usage: testpd [-dvarslR] [test, ...]
  38   -d  Turn on debugging
  39   -v  Verbose
  40   -a  All tests
  41   -l  List all tests
  42   -s  Show test(s)
  43   -r  Run test(s)
  44   -f  Fix failed test(s)
  45   -R <directory>  Save test reports to <directory>
  46   <test>  Protocol decoder name ("i2c") and optionally test name ("i2c/icc")""")
  47     sys.exit()
  48
  49
  50 def check_tclist(tc):
  51     if 'pdlist' not in tc or not tc['pdlist']:
  52         return("No protocol decoders")
  53     if 'input' not in tc or not tc['input']:
  54         return("No input")
  55     if 'output' not in tc or not tc['output']:
  56         return("No output")
  57     for op in tc['output']:
  58         if 'match' not in op:
  59             return("No match in output")
  60
  61     return None
  62
  63
  64 def parse_testfile(path, pd, tc, op_type, op_class):
  65     DBG("Opening '%s'" % path)
  66     tclist = []
  67     for line in open(path).read().split('\n'):
  68         try:
  69             line = line.strip()
  70             if len(line) == 0 or line[0] == "#":
  71                 continue
  72             f = line.split()
  73             if not tclist and f[0] != "test":
  74                 # That can't be good.
  75                 raise E_badline
  76             key = f.pop(0)
  77             if key == 'test':
  78                 if len(f) != 1:
  79                     raise E_syntax
  80                 # new testcase
  81                 tclist.append({
  82                     'pd': pd,
  83                     'name': f[0],
  84                     'pdlist': [],
  85                     'output': [],
  86                 })
  87             elif key == 'protocol-decoder':
  88                 if len(f) < 1:
  89                     raise E_syntax
  90                 pd_spec = {
  91                     'name': f.pop(0),
  92                     'probes': [],
  93                     'options': [],
  94                 }
  95                 while len(f):
  96                     if len(f) == 1:
  97                         # Always needs <key> <value>
  98                         raise E_syntax
  99                     a, b = f[:2]
 100                     f = f[2:]
 101                     if '=' not in b:
 102                         raise E_syntax
 103                     opt, val = b.split('=')
 104                     if a == 'probe':
 105                         try:
 106                             val = int(val)
 107                         except:
 108                             raise E_syntax
 109                         pd_spec['probes'].append([opt, val])
 110                     elif a == 'option':
 111                         pd_spec['options'].append([opt, val])
 112                     else:
 113                         raise E_syntax
 114                 tclist[-1]['pdlist'].append(pd_spec)
 115             elif key == 'stack':
 116                 if len(f) < 2:
 117                     raise E_syntax
 118                 tclist[-1]['stack'] = f
 119             elif key == 'input':
 120                 if len(f) != 1:
 121                     raise E_syntax
 122                 tclist[-1]['input'] = f[0]
 123             elif key == 'output':
 124                 op_spec = {
 125                     'pd': f.pop(0),
 126                     'type': f.pop(0),
 127                 }
 128                 while len(f):
 129                     if len(f) == 1:
 130                         # Always needs <key> <value>
 131                         raise E_syntax
 132                     a, b = f[:2]
 133                     f = f[2:]
 134                     if a == 'class':
 135                         op_spec['class'] = b
 136                     elif a == 'match':
 137                         op_spec['match'] = b
 138                     else:
 139                         raise E_syntax
 140                 tclist[-1]['output'].append(op_spec)
 141             else:
 142                 raise E_badline
 143         except E_badline as e:
 144             ERR("Invalid syntax in %s: line '%s'" % (path, line))
 145             return []
 146         except E_syntax as e:
 147             ERR("Unable to parse %s: unknown line '%s'" % (path, line))
 148             return []
 149
 150     # If a specific testcase was requested, keep only that one.
 151     if tc is not None:
 152         target_tc = None
 153         for t in tclist:
 154             if t['name'] == tc:
 155                 target_tc = t
 156                 break
 157         # ...and a specific output type
 158         if op_type is not None:
 159             target_oplist = []
 160             for op in target_tc['output']:
 161                 if op['type'] == op_type:
 162                     # ...and a specific output class
 163                     if op_class is None or ('class' in op and op['class'] == op_class):
 164                         target_oplist.append(op)
 165                         DBG("match on [%s]" % str(op))
 166             target_tc['output'] = target_oplist
 167         if target_tc is None:
 168             tclist = []
 169         else:
 170             tclist = [target_tc]
 171     for t in tclist:
 172         error = check_tclist(t)
 173         if error:
 174             ERR("Error in %s: %s" % (path, error))
 175             return []
 176
 177     return tclist
 178
 179
 180 def get_tests(testnames):
 181     tests = []
 182     for testspec in testnames:
 183         # Optional testspec in the form i2c/rtc
 184         tc = op_type = op_class = None
 185         ts = testspec.strip("/").split("/")
 186         pd = ts.pop(0)
 187         if ts:
 188             tc = ts.pop(0)
 189         if ts:
 190             op_type = ts.pop(0)
 191         if ts:
 192             op_class = ts.pop(0)
 193         path = os.path.join(decoders_dir, pd)
 194         if not os.path.isdir(path):
 195             # User specified non-existent PD
 196             raise Exception("%s not found." % path)
 197         path = os.path.join(decoders_dir, pd, "test/test.conf")
 198         if not os.path.exists(path):
 199             # PD doesn't have any tests yet
 200             continue
 201         tests.append(parse_testfile(path, pd, tc, op_type, op_class))
 202
 203     return tests
 204
 205
 206 def diff_files(f1, f2):
 207     t1 = open(f1).readlines()
 208     t2 = open(f2).readlines()
 209     diff = []
 210     d = Differ()
 211     for line in d.compare(t1, t2):
 212         if line[:2] in ('- ', '+ '):
 213             diff.append(line.strip())
 214
 215     return diff
 216
 217
 218 def run_tests(tests, fix=False):
 219     errors = 0
 220     results = []
 221     cmd = os.path.join(tests_dir, 'runtc')
 222     for tclist in tests:
 223         for tc in tclist:
 224             args = [cmd]
 225             if DEBUG > 1:
 226                 args.append('-d')
 227             for pd in tc['pdlist']:
 228                 args.extend(['-P', pd['name']])
 229                 for label, probe in pd['probes']:
 230                     args.extend(['-p', "%s=%d" % (label, probe)])
 231                 for option, value in pd['options']:
 232                     args.extend(['-o', "%s=%s" % (option, value)])
 233             args.extend(['-i', os.path.join(dumps_dir, tc['input'])])
 234             for op in tc['output']:
 235                 name = "%s/%s/%s" % (tc['pd'], tc['name'], op['type'])
 236                 opargs = ['-O', "%s:%s" % (op['pd'], op['type'])]
 237                 if 'class' in op:
 238                     opargs[-1] += ":%s" % op['class']
 239                     name += "/%s" % op['class']
 240                 if VERBOSE:
 241                     dots = '.' * (60 - len(name) - 2)
 242                     INFO("%s %s " % (name, dots), end='')
 243                 results.append({
 244                     'testcase': name,
 245                 })
 246                 try:
 247                     fd, outfile = mkstemp()
 248                     os.close(fd)
 249                     opargs.extend(['-f', outfile])
 250                     DBG("Running %s" % (' '.join(args + opargs)))
 251                     p = Popen(args + opargs, stdout=PIPE, stderr=PIPE)
 252                     stdout, stderr = p.communicate()
 253                     if stdout:
 254                         results[-1]['statistics'] = stdout.decode('utf-8').strip()
 255                     if stderr:
 256                         results[-1]['error'] = stderr.decode('utf-8').strip()
 257                         errors += 1
 258                     elif p.returncode != 0:
 259                         # runtc indicated an error, but didn't output a
 260                         # message on stderr about it
 261                         results[-1]['error'] = "Unknown error: runtc %d" % p.returncode
 262                     if 'error' not in results[-1]:
 263                         match = os.path.join(decoders_dir, op['pd'], 'test', op['match'])
 264                         try:
 265                             diff = diff_error = None
 266                             if op['type'] == 'annotation':
 267                                 diff = diff_textfiles(match, outfile)
 268                             else:
 269                                 diff = ["Unsupported output type '%s'." % op['type']]
 270                         except Exception as e:
 271                             diff_error = e
 272                         if fix:
 273                             if diff or diff_error:
 274                                 copy(outfile, match)
 275                                 DBG("Wrote %s" % match)
 276                         else:
 277                             if diff:
 278                                 results[-1]['diff'] = diff
 279                             elif diff_error is not None:
 280                                 raise diff_error
 281                 except Exception as e:
 282                     results[-1]['error'] = str(e)
 283                 finally:
 284                     os.unlink(outfile)
 285                 if VERBOSE:
 286                     if 'diff' in results[-1]:
 287                         INFO("Output mismatch")
 288                     elif 'error' in results[-1]:
 289                         error = results[-1]['error']
 290                         if len(error) > 20:
 291                             error = error[:17] + '...'
 292                         INFO(error)
 293                     else:
 294                         INFO("OK")
 295                 gen_report(results[-1])
 296
 297     return results, errors
 298
 299
 300 def gen_report(result):
 301     out = []
 302     if 'error' in result:
 303         out.append("Error:")
 304         out.append(result['error'])
 305         out.append('')
 306     if 'diff' in result:
 307         out.append("Test output mismatch:")
 308         out.extend(result['diff'])
 309         out.append('')
 310     if 'statistics' in result:
 311         out.extend(["Statistics:", result['statistics']])
 312         out.append('')
 313
 314     if out:
 315         text = "Testcase: %s\n" % result['testcase']
 316         text += '\n'.join(out)
 317     else:
 318         return
 319
 320     if report_dir:
 321         filename = result['testcase'].replace('/', '_')
 322         open(os.path.join(report_dir, filename), 'w').write(text)
 323     else:
 324         print(text)
 325
 326
 327 def show_tests(tests):
 328     for tclist in tests:
 329         for tc in tclist:
 330             print("Testcase: %s/%s" % (tc['pd'], tc['name']))
 331             for pd in tc['pdlist']:
 332                 print("  Protocol decoder: %s" % pd['name'])
 333                 for label, probe in pd['probes']:
 334                     print("    Probe %s=%d" % (label, probe))
 335                 for option, value in pd['options']:
 336                     print("    Option %s=%d" % (option, value))
 337             if 'stack' in tc:
 338                 print("  Stack: %s" % ' '.join(tc['stack']))
 339             print("  Input: %s" % tc['input'])
 340             for op in tc['output']:
 341                 print("  Output:\n    Protocol decoder: %s" % op['pd'])
 342                 print("    Type: %s" % op['type'])
 343                 if 'class' in op:
 344                     print("    Class: %s" % op['class'])
 345                 print("    Match: %s" % op['match'])
 346         print()
 347
 348
 349 def list_tests(tests):
 350     for tclist in tests:
 351         for tc in tclist:
 352             for op in tc['output']:
 353                 line = "%s/%s/%s" % (tc['pd'], tc['name'], op['type'])
 354                 if 'class' in op:
 355                     line += "/%s" % op['class']
 356                 print(line)
 357
 358
 359 #
 360 # main
 361 #
 362
 363 # project root
 364 tests_dir = os.path.abspath(os.path.dirname(sys.argv[0]))
 365 base_dir = os.path.abspath(os.path.join(os.curdir, tests_dir, os.path.pardir))
 366 dumps_dir = os.path.abspath(os.path.join(base_dir, os.path.pardir, 'sigrok-dumps'))
 367 decoders_dir = os.path.abspath(os.path.join(base_dir, 'decoders'))
 368
 369 if len(sys.argv) == 1:
 370     usage()
 371
 372 opt_all = opt_run = opt_show = opt_list = opt_fix = False
 373 report_dir = None
 374 opts, args = getopt(sys.argv[1:], "dvarslfRS:")
 375 for opt, arg in opts:
 376     if opt == '-d':
 377         DEBUG += 1
 378     if opt == '-v':
 379         VERBOSE = True
 380     elif opt == '-a':
 381         opt_all = True
 382     elif opt == '-r':
 383         opt_run = True
 384     elif opt == '-s':
 385         opt_show = True
 386     elif opt == '-l':
 387         opt_list = True
 388     elif opt == '-f':
 389         opt_fix = True
 390     elif opt == '-R':
 391         report_dir = arg
 392     elif opt == '-S':
 393         dumps_dir = arg
 394
 395 if opt_run and opt_show:
 396     usage("Use either -s or -r, not both.")
 397 if args and opt_all:
 398     usage("Specify either -a or tests, not both.")
 399 if report_dir is not None and not os.path.isdir(report_dir):
 400     usage("%s is not a directory" % report_dir)
 401
 402 ret = 0
 403 try:
 404     if args:
 405         testlist = get_tests(args)
 406     elif opt_all:
 407         testlist = get_tests(os.listdir(decoders_dir))
 408     else:
 409         usage("Specify either -a or tests.")
 410
 411     if opt_run:
 412         if not os.path.isdir(dumps_dir):
 413             ERR("Could not find sigrok-dumps repository at %s" % dumps_dir)
 414             sys.exit(1)
 415         results, errors = run_tests(testlist)
 416         ret = errors
 417     elif opt_show:
 418         show_tests(testlist)
 419     elif opt_list:
 420         list_tests(testlist)
 421     elif opt_fix:
 422         run_tests(testlist, fix=True)
 423     else:
 424         usage()
 425 except Exception as e:
 426     print("Error: %s" % str(e))
 427     if DEBUG:
 428         raise
 429
 430 sys.exit(ret)
 431