topotest: implement json_cmp function

Implemented a JSON compare function that tells you when a specific
subset of items exist or not inside a JSON dataset.

More details can be found in the function docstring or in the test file
lib/test_json.py.

diff --git a/tests/topotests/lib/test/__init__.py b/tests/topotests/lib/test/__init__.py
new file mode 100644 (file)
index 0000000..e69de29

diff --git a/tests/topotests/lib/test/test_json.py b/tests/topotests/lib/test/test_json.py
new file mode 100755 (executable)
index 0000000..87bf48d
--- /dev/null
+++ b/tests/topotests/lib/test/test_json.py
@@ -0,0 +1,253 @@
+#!/usr/bin/env python
+
+#
+# test_json.py
+# Tests for library function: json_cmp().
+#
+# Copyright (c) 2017 by
+# Network Device Education Foundation, Inc. ("NetDEF")
+#
+# Permission to use, copy, modify, and/or distribute this software
+# for any purpose with or without fee is hereby granted, provided
+# that the above copyright notice and this permission notice appear
+# in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
+# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
+# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
+# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
+# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
+# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
+# OF THIS SOFTWARE.
+#
+
+"""
+Tests for the json_cmp() function.
+"""
+
+import os
+import sys
+import pytest
+
+# Save the Current Working Directory to find lib files.
+CWD = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(CWD, '../../'))
+
+# pylint: disable=C0413
+from lib.topotest import json_cmp
+
+def test_json_intersect_true():
+    "Test simple correct JSON intersections"
+
+    dcomplete = {
+        'i1': 'item1',
+        'i2': 'item2',
+        'i3': 'item3',
+        'i100': 'item4',
+    }
+
+    dsub1 = {
+        'i1': 'item1',
+        'i3': 'item3',
+    }
+    dsub2 = {
+        'i1': 'item1',
+        'i2': 'item2',
+    }
+    dsub3 = {
+        'i100': 'item4',
+        'i2': 'item2',
+    }
+    dsub4 = {
+        'i50': None,
+        'i100': 'item4',
+    }
+
+    assert json_cmp(dcomplete, dsub1) is None
+    assert json_cmp(dcomplete, dsub2) is None
+    assert json_cmp(dcomplete, dsub3) is None
+    assert json_cmp(dcomplete, dsub4) is None
+
+def test_json_intersect_false():
+    "Test simple incorrect JSON intersections"
+
+    dcomplete = {
+        'i1': 'item1',
+        'i2': 'item2',
+        'i3': 'item3',
+        'i100': 'item4',
+    }
+
+    # Incorrect value for 'i1'
+    dsub1 = {
+        'i1': 'item3',
+        'i3': 'item3',
+    }
+    # Non-existing key 'i5'
+    dsub2 = {
+        'i1': 'item1',
+        'i5': 'item2',
+    }
+    # Key should not exist
+    dsub3 = {
+        'i100': None,
+    }
+
+    assert json_cmp(dcomplete, dsub1) is not None
+    assert json_cmp(dcomplete, dsub2) is not None
+    assert json_cmp(dcomplete, dsub3) is not None
+
+def test_json_intersect_multilevel_true():
+    "Test multi level correct JSON intersections"
+
+    dcomplete = {
+        'i1': 'item1',
+        'i2': 'item2',
+        'i3': {
+            'i100': 'item100',
+        },
+        'i4': {
+            'i41': {
+                'i411': 'item411',
+            },
+            'i42': {
+                'i421': 'item421',
+                'i422': 'item422',
+            }
+        }
+    }
+
+    dsub1 = {
+        'i1': 'item1',
+        'i3': {
+            'i100': 'item100',
+        },
+        'i10': None,
+    }
+    dsub2 = {
+        'i1': 'item1',
+        'i2': 'item2',
+        'i3': {},
+    }
+    dsub3 = {
+        'i2': 'item2',
+        'i4': {
+            'i41': {
+                'i411': 'item411',
+            },
+            'i42': {
+                'i422': 'item422',
+                'i450': None,
+            }
+        }
+    }
+    dsub4 = {
+        'i2': 'item2',
+        'i4': {
+            'i41': {},
+            'i42': {
+                'i450': None,
+            }
+        }
+    }
+    dsub5 = {
+        'i2': 'item2',
+        'i3': {
+            'i100': 'item100',
+        },
+        'i4': {
+            'i42': {
+                'i450': None,
+            }
+        }
+    }
+
+    assert json_cmp(dcomplete, dsub1) is None
+    assert json_cmp(dcomplete, dsub2) is None
+    assert json_cmp(dcomplete, dsub3) is None
+    assert json_cmp(dcomplete, dsub4) is None
+    assert json_cmp(dcomplete, dsub5) is None
+
+def test_json_intersect_multilevel_false():
+    "Test multi level incorrect JSON intersections"
+
+    dcomplete = {
+        'i1': 'item1',
+        'i2': 'item2',
+        'i3': {
+            'i100': 'item100',
+        },
+        'i4': {
+            'i41': {
+                'i411': 'item411',
+            },
+            'i42': {
+                'i421': 'item421',
+                'i422': 'item422',
+            }
+        }
+    }
+
+    # Incorrect sub-level value
+    dsub1 = {
+        'i1': 'item1',
+        'i3': {
+            'i100': 'item00',
+        },
+        'i10': None,
+    }
+    # Inexistent sub-level
+    dsub2 = {
+        'i1': 'item1',
+        'i2': 'item2',
+        'i3': None,
+    }
+    # Inexistent sub-level value
+    dsub3 = {
+        'i1': 'item1',
+        'i3': {
+            'i100': None,
+        },
+    }
+    # Inexistent sub-sub-level value
+    dsub4 = {
+        'i4': {
+            'i41': {
+                'i412': 'item412',
+            },
+            'i42': {
+                'i421': 'item421',
+            }
+        }
+    }
+    # Invalid sub-sub-level value
+    dsub5 = {
+        'i4': {
+            'i41': {
+                'i411': 'item411',
+            },
+            'i42': {
+                'i421': 'item420000',
+            }
+        }
+    }
+    # sub-sub-level should be value
+    dsub6 = {
+        'i4': {
+            'i41': {
+                'i411': 'item411',
+            },
+            'i42': 'foobar',
+        }
+    }
+
+    assert json_cmp(dcomplete, dsub1) is not None
+    assert json_cmp(dcomplete, dsub2) is not None
+    assert json_cmp(dcomplete, dsub3) is not None
+    assert json_cmp(dcomplete, dsub4) is not None
+    assert json_cmp(dcomplete, dsub5) is not None
+    assert json_cmp(dcomplete, dsub6) is not None
+
+if __name__ == '__main__':
+    sys.exit(pytest.main())

diff --git a/tests/topotests/lib/topotest.py b/tests/topotests/lib/topotest.py
index d0b377b33b110bfb8d34dd8385bb89dcc2ea95cc..f484366750fbacab3153018d7699998dcbecacfd 100644 (file)
--- a/tests/topotests/lib/topotest.py
+++ b/tests/topotests/lib/topotest.py
@@ -42,6 +42,45 @@ from mininet.link import Intf
 
 from time import sleep
 
+
+def json_cmp(d1, d2, reason=False):
+    """
+    JSON compare function. Receives two parameters:
+    * `d1`: json value
+    * `d2`: json subset which we expect
+
+    Returns `None` when all keys that `d1` has matches `d2`,
+    otherwise a string containing what failed.
+
+    Note: key absence can be tested by adding a key with value `None`.
+    """
+    squeue = [(d1, d2)]
+    for s in squeue:
+        nd1, nd2 = s
+        s1, s2 = set(nd1), set(nd2)
+
+        # Expect all required fields to exist.
+        s2_req = set([key for key in nd2 if nd2[key] is not None])
+        diff = s2_req - s1
+        if diff != set({}):
+            return 'expected keys "{}" in "{}"'.format(diff, str(nd1))
+
+        for key in s2.intersection(s1):
+            # Test for non existence of key in d2
+            if nd2[key] is None:
+                return '"{}" should not exist in "{}"'.format(key, str(nd1))
+            # If nd1 key is a dict, we have to recurse in it later.
+            if isinstance(nd2[key], type({})):
+                squeue.append((nd1[key], nd2[key]))
+                continue
+            # Compare JSON values
+            if nd1[key] != nd2[key]:
+                return '"{}" value is different ("{}" != "{}")'.format(
+                    key, str(nd1[key]), str(nd2[key])
+                )
+
+    return None
+
 def run_and_expect(func, what, count=20, wait=3):
     """
     Run `func` and compare the result with `what`. Do it for `count` times