tests/pdtest

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