qa/tasks/mgr/dashboard: Imroved JSON validation

Refactored `OsdTest` to make use of `self.assertSchema()`

Signed-off-by: Sebastian Wagner <sebastian.wagner@suse.com>

qa/tasks/mgr/dashboard/helper.py

@@ -7,8 +7,11 @@ import logging
 import os
 import subprocess
 import sys
+from collections import namedtuple
 
 import requests
+import six
+
 from ..mgr_test_case import MgrTestCase
 
 
@@ -117,6 +120,12 @@ class DashboardTestCase(MgrTestCase):
         body = self._resp.json()
         self.assertEqual(body, data)
 
+    def assertSchema(self, data, schema):
+        try:
+            return _validate_json(data, schema)
+        except _ValError as e:
+            self.assertEqual(data, str(e))
+
     def assertBody(self, body):
         self.assertEqual(self._resp.text, body)
 
@@ -148,3 +157,74 @@ class DashboardTestCase(MgrTestCase):
         out = cls.ceph_cluster.mon_manager.raw_cluster_cmd('mon_status')
         j = json.loads(out)
         return [mon['name'] for mon in j['monmap']['mons']]
+
+
+class JLeaf(namedtuple('JLeaf', ['typ', 'none'])):
+    def __new__(cls, typ, none=False):
+        if typ == str:
+            typ = six.string_types
+        return super(JLeaf, cls).__new__(cls, typ, none)
+
+
+JList = namedtuple('JList', ['elem_typ'])
+
+JTuple = namedtuple('JList', ['elem_typs'])
+
+
+class JObj(namedtuple('JObj', ['sub_elems', 'allow_unknown'])):
+    def __new__(cls, sub_elems, allow_unknown=False):
+        """
+        :type sub_elems: dict[str, JAny | JLeaf | JList | JObj]
+        :type allow_unknown: bool
+        :return:
+        """
+        return super(JObj, cls).__new__(cls, sub_elems, allow_unknown)
+
+
+JAny = namedtuple('JAny', ['none'])
+
+
+class _ValError(Exception):
+    def __init__(self, msg, path):
+        path_str = ''.join('[{}]'.format(repr(p)) for p in path)
+        super(_ValError, self).__init__('In `input{}`: {}'.format(path_str, msg))
+
+
+def _validate_json(val, schema, path=[]):
+    """
+    >>> d = {'a': 1, 'b': 'x', 'c': range(10)}
+    ... ds = JObj({'a': JLeaf(int), 'b': JLeaf(str), 'c': JList(JLeaf(int))})
+    ... _validate_json(d, ds)
+    True
+    """
+    if isinstance(schema, JAny):
+        if not schema.none and val is None:
+            raise _ValError('val is None', path)
+        return True
+    if isinstance(schema, JLeaf):
+        if schema.none and val is None:
+            return True
+        if not isinstance(val, schema.typ):
+            raise _ValError('val not of type {}'.format(schema.typ), path)
+        return True
+    if isinstance(schema, JList):
+        return all(_validate_json(e, schema.elem_typ, path + [i]) for i, e in enumerate(val))
+    if isinstance(schema, JTuple):
+        return all(_validate_json(val[i], typ, path + [i])
+                   for i, typ in enumerate(schema.elem_typs))
+    if isinstance(schema, JObj):
+        missing_keys = set(schema.sub_elems.keys()).difference(set(val.keys()))
+        if missing_keys:
+            raise _ValError('missing keys: {}'.format(missing_keys), path)
+        unknown_keys = set(val.keys()).difference(set(schema.sub_elems.keys()))
+        if not schema.allow_unknown and unknown_keys:
+            raise _ValError('unknown keys: {}'.format(unknown_keys), path)
+        return all(
+            _validate_json(val[sub_elem_name], sub_elem, path + [sub_elem_name])
+            for sub_elem_name, sub_elem in schema.sub_elems.items()
+        )
+
+    assert False, str(path)
+
+
+

qa/tasks/mgr/dashboard/test_osd.py

@@ -2,15 +2,13 @@
 
 from __future__ import absolute_import
 
-from .helper import DashboardTestCase, authenticate
+from .helper import DashboardTestCase, authenticate, JObj, JAny, JList, JLeaf, JTuple
 
 
 class OsdTest(DashboardTestCase):
 
     def assert_in_and_not_none(self, data, properties):
-        for prop in properties:
-            self.assertIn(prop, data)
-            self.assertIsNotNone(data[prop])
+        self.assertSchema(data, JObj({p: JAny(none=False) for p in properties}, allow_unknown=True))
 
     @authenticate
     def test_list(self):
@@ -25,6 +23,8 @@ class OsdTest(DashboardTestCase):
         self.assert_in_and_not_none(data['stats'], ['numpg', 'stat_bytes_used', 'stat_bytes',
                                                     'op_r', 'op_w'])
         self.assert_in_and_not_none(data['stats_history'], ['op_out_bytes', 'op_in_bytes'])
+        self.assertSchema(data['stats_history']['op_out_bytes'],
+                          JList(JTuple([JLeaf(int), JLeaf(float)])))
 
     @authenticate
     def test_details(self):