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