Add support for initial pin settings.
[sigrok-test.git] / decoder / pdtest
1 #!/usr/bin/env python3
2 ##
3 ## This file is part of the sigrok-test 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 import re
24 from getopt import getopt
25 from tempfile import mkstemp
26 from subprocess import Popen, PIPE
27 from difflib import Differ
28 from hashlib import md5
29 from shutil import copy
30
31 DEBUG = 0
32 VERBOSE = False
33
34
35 class E_syntax(Exception):
36     pass
37 class E_badline(Exception):
38     pass
39
40 def INFO(msg, end='\n'):
41     if VERBOSE:
42         print(msg, end=end)
43         sys.stdout.flush()
44
45
46 def DBG(msg):
47     if DEBUG:
48         print(msg)
49
50
51 def ERR(msg):
52     print(msg, file=sys.stderr)
53
54
55 def usage(msg=None):
56     if msg:
57         print(msg.strip() + '\n')
58     print("""Usage: testpd [-dvalsrfcR] [<test1> <test2> ...]
59   -d  Turn on debugging
60   -v  Verbose
61   -a  All tests
62   -l  List test(s)
63   -s  Show test(s)
64   -r  Run test(s)
65   -f  Fix failed test(s) / create initial output for new test(s)
66   -c  Report decoder code coverage
67   -R <directory>  Save test reports to <directory>
68   <test>  Protocol decoder name ("i2c") and optionally test name ("i2c/rtc")""")
69     sys.exit()
70
71
72 def check_tclist(tc):
73     if 'pdlist' not in tc or not tc['pdlist']:
74         return("No protocol decoders")
75     if 'input' not in tc or not tc['input']:
76         return("No input")
77     if 'output' not in tc or not tc['output']:
78         return("No output")
79     for op in tc['output']:
80         if 'match' not in op:
81             return("No match in output")
82
83     return None
84
85
86 def parse_testfile(path, pd, tc, op_type, op_class):
87     DBG("Opening '%s'" % path)
88     tclist = []
89     for line in open(path).read().split('\n'):
90         try:
91             line = line.strip()
92             if len(line) == 0 or line[0] == "#":
93                 continue
94             f = line.split()
95             if not tclist and f[0] != "test":
96                 # That can't be good.
97                 raise E_badline
98             key = f.pop(0)
99             if key == 'test':
100                 if len(f) != 1:
101                     raise E_syntax
102                 # new testcase
103                 tclist.append({
104                     'pd': pd,
105                     'name': f[0],
106                     'pdlist': [],
107                     'output': [],
108                 })
109             elif key == 'protocol-decoder':
110                 if len(f) < 1:
111                     raise E_syntax
112                 pd_spec = {
113                     'name': f.pop(0),
114                     'channels': [],
115                     'options': [],
116                     'initial_pins': [],
117                 }
118                 while len(f):
119                     if len(f) == 1:
120                         # Always needs <key> <value>
121                         raise E_syntax
122                     a, b = f[:2]
123                     f = f[2:]
124                     if '=' not in b:
125                         raise E_syntax
126                     opt, val = b.split('=')
127                     if a == 'channel':
128                         try:
129                             val = int(val)
130                         except:
131                             raise E_syntax
132                         pd_spec['channels'].append([opt, val])
133                     elif a == 'option':
134                         pd_spec['options'].append([opt, val])
135                     elif a == 'initial_pin':
136                         try:
137                             val = int(val)
138                         except:
139                             raise E_syntax
140                         pd_spec['initial_pins'].append([opt, val])
141                     else:
142                         raise E_syntax
143                 tclist[-1]['pdlist'].append(pd_spec)
144             elif key == 'stack':
145                 if len(f) < 2:
146                     raise E_syntax
147                 tclist[-1]['stack'] = f
148             elif key == 'input':
149                 if len(f) != 1:
150                     raise E_syntax
151                 tclist[-1]['input'] = f[0]
152             elif key == 'output':
153                 op_spec = {
154                     'pd': f.pop(0),
155                     'type': f.pop(0),
156                 }
157                 while len(f):
158                     if len(f) == 1:
159                         # Always needs <key> <value>
160                         raise E_syntax
161                     a, b = f[:2]
162                     f = f[2:]
163                     if a == 'class':
164                         op_spec['class'] = b
165                     elif a == 'match':
166                         op_spec['match'] = b
167                     else:
168                         raise E_syntax
169                 tclist[-1]['output'].append(op_spec)
170             else:
171                 raise E_badline
172         except E_badline as e:
173             ERR("Invalid syntax in %s: line '%s'" % (path, line))
174             return []
175         except E_syntax as e:
176             ERR("Unable to parse %s: unknown line '%s'" % (path, line))
177             return []
178
179     # If a specific testcase was requested, keep only that one.
180     if tc is not None:
181         target_tc = None
182         for t in tclist:
183             if t['name'] == tc:
184                 target_tc = t
185                 break
186         # ...and a specific output type
187         if op_type is not None:
188             target_oplist = []
189             for op in target_tc['output']:
190                 if op['type'] == op_type:
191                     # ...and a specific output class
192                     if op_class is None or ('class' in op and op['class'] == op_class):
193                         target_oplist.append(op)
194                         DBG("match on [%s]" % str(op))
195             target_tc['output'] = target_oplist
196         if target_tc is None:
197             tclist = []
198         else:
199             tclist = [target_tc]
200     for t in tclist:
201         error = check_tclist(t)
202         if error:
203             ERR("Error in %s: %s" % (path, error))
204             return []
205
206     return tclist
207
208
209 def get_tests(testnames):
210     tests = {}
211     for testspec in testnames:
212         # Optional testspec in the form pd/testcase/type/class
213         tc = op_type = op_class = None
214         ts = testspec.strip("/").split("/")
215         pd = ts.pop(0)
216         tests[pd] = []
217         if ts:
218             tc = ts.pop(0)
219         if ts:
220             op_type = ts.pop(0)
221         if ts:
222             op_class = ts.pop(0)
223         path = os.path.join(tests_dir, pd)
224         if not os.path.isdir(path):
225             # User specified non-existent PD
226             raise Exception("%s not found." % path)
227         path = os.path.join(tests_dir, pd, "test.conf")
228         if not os.path.exists(path):
229             # PD doesn't have any tests yet
230             continue
231         tests[pd].append(parse_testfile(path, pd, tc, op_type, op_class))
232
233     return tests
234
235
236 def diff_text(f1, f2):
237     t1 = open(f1).readlines()
238     t2 = open(f2).readlines()
239     diff = []
240     d = Differ()
241     for line in d.compare(t1, t2):
242         if line[:2] in ('- ', '+ '):
243             diff.append(line.strip())
244
245     return diff
246
247
248 def compare_binary(f1, f2):
249     h1 = md5()
250     h1.update(open(f1, 'rb').read())
251     h2 = md5()
252     h2.update(open(f2, 'rb').read())
253     if h1.digest() == h2.digest():
254         result = None
255     else:
256         result = ["Binary output does not match."]
257
258     return result
259
260
261 # runtc's stdout can have lines like:
262 # coverage: lines=161 missed=2 coverage=99%
263 def parse_stats(text):
264     stats = {}
265     for line in text.strip().split('\n'):
266         fields = line.split()
267         key = fields.pop(0).strip(':')
268         if key not in stats:
269             stats[key] = []
270         stats[key].append({})
271         for f in fields:
272             k, v = f.split('=')
273             stats[key][-1][k] = v
274
275     return stats
276
277
278 # take result set of all tests in a PD, and summarize which lines
279 # were not covered by any of the tests.
280 def coverage_sum(cvglist):
281     lines = 0
282     missed = 0
283     missed_lines = {}
284     for record in cvglist:
285         lines = int(record['lines'])
286         missed += int(record['missed'])
287         if 'missed_lines' not in record:
288             continue
289         for linespec in record['missed_lines'].split(','):
290             if linespec not in missed_lines:
291                 missed_lines[linespec] = 1
292             else:
293                 missed_lines[linespec] += 1
294
295     # keep only those lines that didn't show up in every non-summary record
296     final_missed = []
297     for linespec in missed_lines:
298         if missed_lines[linespec] != len(cvglist):
299             continue
300         final_missed.append(linespec)
301
302     return lines, final_missed
303
304
305 def run_tests(tests, fix=False):
306     errors = 0
307     results = []
308     cmd = [os.path.join(runtc_dir, 'runtc')]
309     if opt_coverage:
310         fd, coverage = mkstemp()
311         os.close(fd)
312         cmd.extend(['-c', coverage])
313     else:
314         coverage = None
315     for pd in sorted(tests.keys()):
316         pd_cvg = []
317         for tclist in tests[pd]:
318             for tc in tclist:
319                 args = cmd[:]
320                 if DEBUG > 1:
321                     args.append('-d')
322                 # Set up PD stack for this test.
323                 for spd in tc['pdlist']:
324                     args.extend(['-P', spd['name']])
325                     for label, channel in spd['channels']:
326                         args.extend(['-p', "%s=%d" % (label, channel)])
327                     for option, value in spd['options']:
328                         args.extend(['-o', "%s=%s" % (option, value)])
329                     for label, initial_pin in spd['initial_pins']:
330                         args.extend(['-N', "%s=%d" % (label, initial_pin)])
331                 args.extend(['-i', os.path.join(dumps_dir, tc['input'])])
332                 for op in tc['output']:
333                     name = "%s/%s/%s" % (pd, tc['name'], op['type'])
334                     opargs = ['-O', "%s:%s" % (op['pd'], op['type'])]
335                     if 'class' in op:
336                         opargs[-1] += ":%s" % op['class']
337                         name += "/%s" % op['class']
338                     if VERBOSE:
339                         dots = '.' * (77 - len(name) - 2)
340                         INFO("%s %s " % (name, dots), end='')
341                     results.append({
342                         'testcase': name,
343                     })
344                     try:
345                         fd, outfile = mkstemp()
346                         os.close(fd)
347                         opargs.extend(['-f', outfile])
348                         DBG("Running %s" % (' '.join(args + opargs)))
349                         p = Popen(args + opargs, stdout=PIPE, stderr=PIPE)
350                         stdout, stderr = p.communicate()
351                         if stdout:
352                             # statistics and coverage data on stdout
353                             results[-1].update(parse_stats(stdout.decode('utf-8')))
354                         if stderr:
355                             results[-1]['error'] = stderr.decode('utf-8').strip()
356                             errors += 1
357                         elif p.returncode != 0:
358                             # runtc indicated an error, but didn't output a
359                             # message on stderr about it
360                             results[-1]['error'] = "Unknown error: runtc %d" % p.returncode
361                         if 'error' not in results[-1]:
362                             matchfile = os.path.join(tests_dir, op['pd'], op['match'])
363                             DBG("Comparing with %s" % matchfile)
364                             try:
365                                 diff = diff_error = None
366                                 if op['type'] in ('annotation', 'python'):
367                                     diff = diff_text(matchfile, outfile)
368                                 elif op['type'] == 'binary':
369                                     diff = compare_binary(matchfile, outfile)
370                                 else:
371                                     diff = ["Unsupported output type '%s'." % op['type']]
372                             except Exception as e:
373                                 diff_error = e
374                             if fix:
375                                 if diff or diff_error:
376                                     copy(outfile, matchfile)
377                                     DBG("Wrote %s" % matchfile)
378                             else:
379                                 if diff:
380                                     results[-1]['diff'] = diff
381                                 elif diff_error is not None:
382                                     raise diff_error
383                     except Exception as e:
384                         results[-1]['error'] = str(e)
385                     finally:
386                         if coverage:
387                             results[-1]['coverage_report'] = coverage
388                         os.unlink(outfile)
389                     if op['type'] == 'exception' and 'error' in results[-1]:
390                         # filter out the exception we were looking for
391                         reg = "^Error: srd: %s:" % op['match']
392                         if re.match(reg, results[-1]['error']):
393                             # found it, not an error
394                             results[-1].pop('error')
395                             errors -= 1
396                     if VERBOSE:
397                         if 'diff' in results[-1]:
398                             INFO("Output mismatch")
399                         elif 'error' in results[-1]:
400                             error = results[-1]['error']
401                             if len(error) > 20:
402                                 error = error[:17] + '...'
403                             INFO(error)
404                         elif 'coverage' in results[-1]:
405                             # report coverage of this PD
406                             for record in results[-1]['coverage']:
407                                 # but not others used in the stack
408                                 # as part of the test.
409                                 if record['scope'] == pd:
410                                     INFO(record['coverage'])
411                                     break
412                         else:
413                             INFO("OK")
414                     gen_report(results[-1])
415                     if coverage:
416                         os.unlink(coverage)
417                         # only keep track of coverage records for this PD,
418                         # not others in the stack just used for testing.
419                         for cvg in results[-1]['coverage']:
420                             if cvg['scope'] == pd:
421                                 pd_cvg.append(cvg)
422         if opt_coverage and len(pd_cvg) > 1:
423             # report total coverage of this PD, across all the tests
424             # that were done on it.
425             total_lines, missed_lines = coverage_sum(pd_cvg)
426             pd_coverage = 100 - (float(len(missed_lines)) / total_lines * 100)
427             if VERBOSE:
428                 dots = '.' * (54 - len(pd) - 2)
429                 INFO("%s total %s %d%%" % (pd, dots, pd_coverage))
430             if report_dir:
431                 # generate a missing lines list across all the files in
432                 # the PD
433                 files = {}
434                 for entry in missed_lines:
435                     filename, line = entry.split(':')
436                     if filename not in files:
437                         files[filename] = []
438                     files[filename].append(line)
439                 text = ''
440                 for filename in sorted(files.keys()):
441                     line_list = ','.join(sorted(files[filename], key=int))
442                     text += "%s: %s\n" % (filename, line_list)
443                 open(os.path.join(report_dir, pd + "_total"), 'w').write(text)
444
445
446     return results, errors
447
448 def get_run_tests_error_diff_counts(results):
449     """Get error and diff counters from run_tests() results."""
450     errs = 0
451     diffs = 0
452     for result in results:
453         if 'error' in result:
454             errs += 1
455         if 'diff' in result:
456             diffs += 1
457     return errs, diffs
458
459
460 def gen_report(result):
461     out = []
462     if 'error' in result:
463         out.append("Error:")
464         out.append(result['error'])
465         out.append('')
466     if 'diff' in result:
467         out.append("Test output mismatch:")
468         out.extend(result['diff'])
469         out.append('')
470     if 'coverage_report' in result:
471         out.append(open(result['coverage_report'], 'r').read())
472         out.append('')
473
474     if out:
475         text = "Testcase: %s\n" % result['testcase']
476         text += '\n'.join(out)
477     else:
478         return
479
480     if report_dir:
481         filename = result['testcase'].replace('/', '_')
482         open(os.path.join(report_dir, filename), 'w').write(text)
483     else:
484         print(text)
485
486
487 def show_tests(tests):
488     for pd in sorted(tests.keys()):
489         for tclist in tests[pd]:
490             for tc in tclist:
491                 print("Testcase: %s/%s" % (tc['pd'], tc['name']))
492                 for pd in tc['pdlist']:
493                     print("  Protocol decoder: %s" % pd['name'])
494                     for label, channel in pd['channels']:
495                         print("    Channel %s=%d" % (label, channel))
496                     for option, value in pd['options']:
497                         print("    Option %s=%s" % (option, value))
498                     for label, initial_pin in pd['initial_pins']:
499                         print("    Initial pin %s=%d" % (label, initial_pin))
500                 if 'stack' in tc:
501                     print("  Stack: %s" % ' '.join(tc['stack']))
502                 print("  Input: %s" % tc['input'])
503                 for op in tc['output']:
504                     print("  Output:\n    Protocol decoder: %s" % op['pd'])
505                     print("    Type: %s" % op['type'])
506                     if 'class' in op:
507                         print("    Class: %s" % op['class'])
508                     print("    Match: %s" % op['match'])
509             print()
510
511
512 def list_tests(tests):
513     for pd in sorted(tests.keys()):
514         for tclist in tests[pd]:
515             for tc in tclist:
516                 for op in tc['output']:
517                     line = "%s/%s/%s" % (tc['pd'], tc['name'], op['type'])
518                     if 'class' in op:
519                         line += "/%s" % op['class']
520                     print(line)
521
522
523 #
524 # main
525 #
526
527 # project root
528 runtc_dir = os.path.abspath(os.path.dirname(sys.argv[0]))
529 base_dir = os.path.abspath(os.path.join(os.curdir, runtc_dir, os.path.pardir))
530 dumps_dir = os.path.abspath(os.path.join(base_dir, os.path.pardir, 'sigrok-dumps'))
531 tests_dir = os.path.abspath(os.path.join(runtc_dir, 'test'))
532
533 if len(sys.argv) == 1:
534     usage()
535
536 opt_all = opt_run = opt_show = opt_list = opt_fix = opt_coverage = False
537 report_dir = None
538 try:
539     opts, args = getopt(sys.argv[1:], "dvarslfcR:S:")
540 except Exception as e:
541     usage('error while parsing command line arguments: {}'.format(e))
542 for opt, arg in opts:
543     if opt == '-d':
544         DEBUG += 1
545     if opt == '-v':
546         VERBOSE = True
547     elif opt == '-a':
548         opt_all = True
549     elif opt == '-r':
550         opt_run = True
551     elif opt == '-s':
552         opt_show = True
553     elif opt == '-l':
554         opt_list = True
555     elif opt == '-f':
556         opt_fix = True
557     elif opt == '-c':
558         opt_coverage = True
559     elif opt == '-R':
560         report_dir = arg
561     elif opt == '-S':
562         dumps_dir = arg
563
564 if opt_run and opt_show:
565     usage("Use either -s or -r, not both.")
566 if args and opt_all:
567     usage("Specify either -a or tests, not both.")
568 if report_dir is not None and not os.path.isdir(report_dir):
569     usage("%s is not a directory" % report_dir)
570
571 ret = 0
572 try:
573     if args:
574         testlist = get_tests(args)
575     elif opt_all or opt_list:
576         testlist = get_tests(os.listdir(tests_dir))
577     else:
578         usage("Specify either -a or tests.")
579
580     if opt_run:
581         if not os.path.isdir(dumps_dir):
582             ERR("Could not find sigrok-dumps repository at %s" % dumps_dir)
583             sys.exit(1)
584         results, errors = run_tests(testlist, fix=opt_fix)
585         ret = 0
586         errs, diffs = get_run_tests_error_diff_counts(results)
587         if errs:
588             ret = 1
589         elif diffs:
590             ret = 2
591     elif opt_show:
592         show_tests(testlist)
593     elif opt_list:
594         list_tests(testlist)
595     elif opt_fix:
596         run_tests(testlist, fix=True)
597     else:
598         usage()
599 except Exception as e:
600     print("Error: %s" % str(e))
601     if DEBUG:
602         raise
603
604 sys.exit(ret)