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" % (' '.join(args + opargs)))
 248                     p = Popen(args + opargs, stdout=PIPE, stderr=PIPE)
 249                     stdout, stderr = p.communicate()
 250                     if stdout:
 251                         results[-1]['statistics'] = stdout.decode('utf-8').strip()
 252                     if stderr:
 253                         results[-1]['error'] = stderr.decode('utf-8').strip()
 254                         errors += 1
 255                     elif p.returncode != 0:
 256                         # runtc indicated an error, but didn't output a
 257                         # message on stderr about it
 258                         results[-1]['error'] = "Unknown error: runtc %d" % p.returncode
 259                     # Only bother with the diff if it all worked.
 260                     if 'error' not in results[-1]:
 261                         match = "%s/%s/test/%s" % (decoders_dir, op['pd'], op['match'])
 262                         diff = diff_files(match, outfile)
 263                         if diff:
 264                             results[-1]['diff'] = diff
 265                 except Exception as e:
 266                     results[-1]['error'] = str(e)
 267                 finally:
 268                     os.unlink(outfile)
 269                 if VERBOSE:
 270                     if 'diff' in results[-1]:
 271                         INFO("Output mismatch")
 272                     elif 'error' in results[-1]:
 273                         error = results[-1]['error']
 274                         if len(error) > 20:
 275                             error = error[:17] + '...'
 276                         INFO(error)
 277                     else:
 278                         INFO("OK")
 279                 gen_report(results[-1])
 280
 281     return results, errors
 282
 283
 284 def gen_report(result):
 285     out = []
 286     if 'error' in result:
 287         out.append("Error:")
 288         out.append(result['error'])
 289         out.append('')
 290     if 'diff' in result:
 291         out.append("Test output mismatch:")
 292         out.extend(result['diff'])
 293         out.append('')
 294     if 'statistics' in result:
 295         out.extend(["Statistics:", result['statistics']])
 296         out.append('')
 297
 298     if out:
 299         text = "Testcase: %s\n" % result['testcase']
 300         text += '\n'.join(out)
 301     else:
 302         return
 303
 304     if report_dir:
 305         filename = result['testcase'].replace('/', '_')
 306         open(os.path.join(report_dir, filename), 'w').write(text)
 307     else:
 308         print(text)
 309
 310
 311 def show_tests(tests):
 312     for tclist in tests:
 313         for tc in tclist:
 314             print("Testcase: %s/%s" % (tc['pd'], tc['name']))
 315             for pd in tc['pdlist']:
 316                 print("  Protocol decoder: %s" % pd['name'])
 317                 for label, probe in pd['probes']:
 318                     print("    Probe %s=%d" % (label, probe))
 319                 for option, value in pd['options']:
 320                     print("    Option %s=%d" % (option, value))
 321             if 'stack' in tc:
 322                 print("  Stack: %s" % ' '.join(tc['stack']))
 323             print("  Input: %s" % tc['input'])
 324             for op in tc['output']:
 325                 print("  Output:\n    Protocol decoder: %s" % op['pd'])
 326                 print("    Type: %s" % op['type'])
 327                 if 'class' in op:
 328                     print("    Class: %s" % op['class'])
 329                 print("    Match: %s" % op['match'])
 330         print()
 331
 332
 333 def list_tests(tests):
 334     for tclist in tests:
 335         for tc in tclist:
 336             for op in tc['output']:
 337                 line = "%s/%s/%s" % (tc['pd'], tc['name'], op['type'])
 338                 if 'class' in op:
 339                     line += "/%s" % op['class']
 340                 print(line)
 341
 342
 343 #
 344 # main
 345 #
 346
 347 # project root
 348 tests_dir = os.path.abspath(os.path.dirname(sys.argv[0]))
 349 base_dir = os.path.abspath(os.path.join(os.curdir, tests_dir, os.path.pardir))
 350 dumps_dir = os.path.abspath(os.path.join(base_dir, os.path.pardir, 'sigrok-dumps'))
 351 decoders_dir = os.path.abspath(os.path.join(base_dir, 'decoders'))
 352
 353 if len(sys.argv) == 1:
 354     usage()
 355
 356 opt_all = opt_run = opt_show = opt_list = False
 357 report_dir = None
 358 opts, args = getopt(sys.argv[1:], "dvarslRS:")
 359 for opt, arg in opts:
 360     if opt == '-d':
 361         DEBUG = True
 362     if opt == '-v':
 363         VERBOSE = True
 364     elif opt == '-a':
 365         opt_all = True
 366     elif opt == '-r':
 367         opt_run = True
 368     elif opt == '-s':
 369         opt_show = True
 370     elif opt == '-l':
 371         opt_list = True
 372     elif opt == '-R':
 373         report_dir = arg
 374     elif opt == '-S':
 375         dumps_dir = arg
 376
 377 if opt_run and opt_show:
 378     usage("Use either -s or -r, not both.")
 379 if args and opt_all:
 380     usage("Specify either -a or tests, not both.")
 381 if report_dir is not None and not os.path.isdir(report_dir):
 382     usage("%s is not a directory" % report_dir)
 383
 384 ret = 0
 385 try:
 386     if args:
 387         testlist = get_tests(args)
 388     elif opt_all:
 389         testlist = get_tests(os.listdir(decoders_dir))
 390     else:
 391         usage("Specify either -a or tests.")
 392
 393     if opt_run:
 394         if not os.path.isdir(dumps_dir):
 395             ERR("Could not find sigrok-dumps repository at %s" % dumps_dir)
 396             sys.exit(1)
 397         results, errors = run_tests(testlist)
 398         ret = errors
 399     elif opt_show:
 400         show_tests(testlist)
 401     elif opt_list:
 402         list_tests(testlist)
 403     else:
 404         usage()
 405 except Exception as e:
 406     print("Error: %s" % str(e))
 407     if DEBUG:
 408         raise
 409
 410 sys.exit(ret)
 411