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