tests/qapi-schema/test-qapi.py

   1 #!/usr/bin/env python3
   2 #
   3 # QAPI parser test harness
   4 #
   5 # Copyright (c) 2013 Red Hat Inc.
   6 #
   7 # Authors:
   8 #  Markus Armbruster <armbru@redhat.com>
   9 #
  10 # This work is licensed under the terms of the GNU GPL, version 2 or later.
  11 # See the COPYING file in the top-level directory.
  12 #
  13
  14
  15 import argparse
  16 import difflib
  17 import os
  18 import sys
  19 from io import StringIO
  20
  21 from qapi.error import QAPIError
  22 from qapi.schema import QAPISchema, QAPISchemaVisitor
  23
  24
  25 class QAPISchemaTestVisitor(QAPISchemaVisitor):
  26
  27     def visit_module(self, name):
  28         print('module %s' % name)
  29
  30     def visit_include(self, name, info):
  31         print('include %s' % name)
  32
  33     def visit_enum_type(self, name, info, ifcond, features, members, prefix):
  34         print('enum %s' % name)
  35         if prefix:
  36             print('    prefix %s' % prefix)
  37         for m in members:
  38             print('    member %s' % m.name)
  39             self._print_if(m.ifcond, indent=8)
  40         self._print_if(ifcond)
  41         self._print_features(features)
  42
  43     def visit_array_type(self, name, info, ifcond, element_type):
  44         if not info:
  45             return              # suppress built-in arrays
  46         print('array %s %s' % (name, element_type.name))
  47         self._print_if(ifcond)
  48
  49     def visit_object_type(self, name, info, ifcond, features,
  50                           base, members, variants):
  51         print('object %s' % name)
  52         if base:
  53             print('    base %s' % base.name)
  54         for m in members:
  55             print('    member %s: %s optional=%s'
  56                   % (m.name, m.type.name, m.optional))
  57             self._print_if(m.ifcond, 8)
  58             self._print_features(m.features, indent=8)
  59         self._print_variants(variants)
  60         self._print_if(ifcond)
  61         self._print_features(features)
  62
  63     def visit_alternate_type(self, name, info, ifcond, features, variants):
  64         print('alternate %s' % name)
  65         self._print_variants(variants)
  66         self._print_if(ifcond)
  67         self._print_features(features)
  68
  69     def visit_command(self, name, info, ifcond, features,
  70                       arg_type, ret_type, gen, success_response, boxed,
  71                       allow_oob, allow_preconfig, coroutine):
  72         print('command %s %s -> %s'
  73               % (name, arg_type and arg_type.name,
  74                  ret_type and ret_type.name))
  75         print('    gen=%s success_response=%s boxed=%s oob=%s preconfig=%s%s'
  76               % (gen, success_response, boxed, allow_oob, allow_preconfig,
  77                  " coroutine=True" if coroutine else ""))
  78         self._print_if(ifcond)
  79         self._print_features(features)
  80
  81     def visit_event(self, name, info, ifcond, features, arg_type, boxed):
  82         print('event %s %s' % (name, arg_type and arg_type.name))
  83         print('    boxed=%s' % boxed)
  84         self._print_if(ifcond)
  85         self._print_features(features)
  86
  87     @staticmethod
  88     def _print_variants(variants):
  89         if variants:
  90             print('    tag %s' % variants.tag_member.name)
  91             for v in variants.variants:
  92                 print('    case %s: %s' % (v.name, v.type.name))
  93                 QAPISchemaTestVisitor._print_if(v.ifcond, indent=8)
  94
  95     @staticmethod
  96     def _print_if(ifcond, indent=4):
  97         # TODO Drop this hack after replacing OrderedDict by plain
  98         # dict (requires Python 3.7)
  99         def _massage(subcond):
 100             if isinstance(subcond, str):
 101                 return subcond
 102             if isinstance(subcond, list):
 103                 return [_massage(val) for val in subcond]
 104             return {key: _massage(val) for key, val in subcond.items()}
 105
 106         if ifcond.is_present():
 107             print('%sif %s' % (' ' * indent, _massage(ifcond.ifcond)))
 108
 109     @classmethod
 110     def _print_features(cls, features, indent=4):
 111         if features:
 112             for f in features:
 113                 print('%sfeature %s' % (' ' * indent, f.name))
 114                 cls._print_if(f.ifcond, indent + 4)
 115
 116
 117 def test_frontend(fname):
 118     schema = QAPISchema(fname)
 119     schema.visit(QAPISchemaTestVisitor())
 120
 121     for doc in schema.docs:
 122         if doc.symbol:
 123             print('doc symbol=%s' % doc.symbol)
 124         else:
 125             print('doc freeform')
 126         print('    body=\n%s' % doc.body.text)
 127         for arg, section in doc.args.items():
 128             print('    arg=%s\n%s' % (arg, section.text))
 129         for feat, section in doc.features.items():
 130             print('    feature=%s\n%s' % (feat, section.text))
 131         for section in doc.sections:
 132             print('    section=%s\n%s' % (section.name, section.text))
 133
 134
 135 def open_test_result(dir_name, file_name, update):
 136     mode = 'r+' if update else 'r'
 137     try:
 138         fp = open(os.path.join(dir_name, file_name), mode)
 139     except FileNotFoundError:
 140         if not update:
 141             raise
 142         fp = open(os.path.join(dir_name, file_name), 'w+')
 143     return fp
 144
 145
 146 def test_and_diff(test_name, dir_name, update):
 147     sys.stdout = StringIO()
 148     try:
 149         test_frontend(os.path.join(dir_name, test_name + '.json'))
 150     except QAPIError as err:
 151         errstr = str(err) + '\n'
 152         if dir_name:
 153             errstr = errstr.replace(dir_name + '/', '')
 154         actual_err = errstr.splitlines(True)
 155     else:
 156         actual_err = []
 157     finally:
 158         actual_out = sys.stdout.getvalue().splitlines(True)
 159         sys.stdout.close()
 160         sys.stdout = sys.__stdout__
 161
 162     try:
 163         outfp = open_test_result(dir_name, test_name + '.out', update)
 164         errfp = open_test_result(dir_name, test_name + '.err', update)
 165         expected_out = outfp.readlines()
 166         expected_err = errfp.readlines()
 167     except OSError as err:
 168         print("%s: can't open '%s': %s"
 169               % (sys.argv[0], err.filename, err.strerror),
 170               file=sys.stderr)
 171         return 2
 172
 173     if actual_out == expected_out and actual_err == expected_err:
 174         return 0
 175
 176     print("%s %s" % (test_name, 'UPDATE' if update else 'FAIL'),
 177           file=sys.stderr)
 178     out_diff = difflib.unified_diff(expected_out, actual_out, outfp.name)
 179     err_diff = difflib.unified_diff(expected_err, actual_err, errfp.name)
 180     sys.stdout.writelines(out_diff)
 181     sys.stdout.writelines(err_diff)
 182
 183     if not update:
 184         return 1
 185
 186     try:
 187         outfp.truncate(0)
 188         outfp.seek(0)
 189         outfp.writelines(actual_out)
 190         errfp.truncate(0)
 191         errfp.seek(0)
 192         errfp.writelines(actual_err)
 193     except OSError as err:
 194         print("%s: can't write '%s': %s"
 195               % (sys.argv[0], err.filename, err.strerror),
 196               file=sys.stderr)
 197         return 2
 198
 199     return 0
 200
 201
 202 def main(argv):
 203     parser = argparse.ArgumentParser(
 204         description='QAPI schema tester')
 205     parser.add_argument('-d', '--dir', action='store', default='',
 206                         help="directory containing tests")
 207     parser.add_argument('-u', '--update', action='store_true',
 208                         help="update expected test results")
 209     parser.add_argument('tests', nargs='*', metavar='TEST', action='store')
 210     args = parser.parse_args()
 211
 212     status = 0
 213     for t in args.tests:
 214         (dir_name, base_name) = os.path.split(t)
 215         dir_name = dir_name or args.dir
 216         test_name = os.path.splitext(base_name)[0]
 217         status |= test_and_diff(test_name, dir_name, args.update)
 218
 219     exit(status)
 220
 221
 222 if __name__ == '__main__':
 223     main(sys.argv)
 224     exit(0)