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.tag, section.text))
 134
 135
 136 def open_test_result(dir_name, file_name, update):
 137     mode = 'r+' if update else 'r'
 138     try:
 139         return open(os.path.join(dir_name, file_name), mode, encoding='utf-8')
 140     except FileNotFoundError:
 141         if not update:
 142             raise
 143     return open(os.path.join(dir_name, file_name), 'w+', encoding='utf-8')
 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                         default='QAPI_TEST_UPDATE' in os.environ,
 209                         help="update expected test results")
 210     parser.add_argument('tests', nargs='*', metavar='TEST', action='store')
 211     args = parser.parse_args()
 212
 213     status = 0
 214     for t in args.tests:
 215         (dir_name, base_name) = os.path.split(t)
 216         dir_name = dir_name or args.dir
 217         test_name = os.path.splitext(base_name)[0]
 218         status |= test_and_diff(test_name, dir_name, args.update)
 219
 220     sys.exit(status)
 221
 222
 223 if __name__ == '__main__':
 224     main(sys.argv)
 225     sys.exit(0)