Property-based tests with Hypothesis

.gitignore

@@ -4,3 +4,4 @@ build
 dist
 *.egg-info/
 doc/_build
+.hypothesis/

.travis.yml

@@ -12,6 +12,7 @@ python:
 - pypy3
 install:
 - travis_retry pip install -r requirements.txt
+- travis_retry pip install -r requirements-dev.txt
 - travis_retry pip install coveralls
 script:
 - coverage run --source=jsonpointer tests.py
@@ -22,8 +23,6 @@ addons:
   apt:
     packages:
     - pandoc
-before_deploy:
-- pip install -r requirements-dev.txt
 deploy:
   provider: pypi
   user: skoegl

jsonpatch.py

@@ -281,8 +281,7 @@ def from_diff(cls, src, dst, optimization=True):
         True
         """
 
-        builder = DiffBuilder()
-        builder._compare_values('', None, src, dst)
+        builder = DiffBuilder(src, dst)
         ops = list(builder.execute())
         return cls(ops)
 
@@ -625,134 +624,75 @@ def apply(self, obj):
 
 class DiffBuilder(object):
 
-    def __init__(self):
-        self.index_storage = [{}, {}]
-        self.index_storage2 = [[], []]
-        self.__root = root = []
-        root[:] = [root, root, None]
+    def __init__(self, src, dst):
+        self.src = src
+        self.dst = dst
 
-    def store_index(self, value, index, st):
-        try:
-            storage = self.index_storage[st]
-            stored = storage.get(value)
-            if stored is None:
-                storage[value] = [index]
-            else:
-                storage[value].append(index)
+        self.ops = []
+        self.index_changes = {}
 
-        except TypeError:
-            self.index_storage2[st].append((value, index))
+    def insert(self, op):
+        self.ops.append(op)
 
-    def take_index(self, value, st):
-        try:
-            stored = self.index_storage[st].get(value)
-            if stored:
-                return stored.pop()
+    def execute(self):
+        self._compare_values('', None, self.src, self.dst)
+        return [op.operation for op in self.ops]
 
-        except TypeError:
-            storage = self.index_storage2[st]
-            for i in range(len(storage)-1, -1, -1):
-                if storage[i][0] == value:
-                    return storage.pop(i)[1]
+    def _item_added(self, path, key, item, is_sequence):
+        if is_sequence:
+            key = self.get_index_change(path, key)
 
-    def insert(self, op):
-        root = self.__root
-        last = root[0]
-        last[1] = root[0] = [last, root, op]
-        return root[0]
-
-    def remove(self, index):
-        link_prev, link_next, _ = index
-        link_prev[1] = link_next
-        link_next[0] = link_prev
-        index[:] = []
-
-    def iter_from(self, start):
-        root = self.__root
-        curr = start[1]
-        while curr is not root:
-            yield curr[2]
-            curr = curr[1]
+        new_op = AddOperation({
+            'op': 'add',
+            'path': _path_join(path, key),
+            'value': item,
+        })
+        self.insert(new_op)
 
-    def __iter__(self):
-        root = self.__root
-        curr = root[1]
-        while curr is not root:
-            yield curr[2]
-            curr = curr[1]
+        if is_sequence:
+            self.add_index_change(path, key, +1)
 
-    def execute(self):
-        root = self.__root
-        curr = root[1]
-        while curr is not root:
-            if curr[1] is not root:
-                op_first, op_second = curr[2], curr[1][2]
-                if op_first.location == op_second.location and \
-                        type(op_first) == RemoveOperation and \
-                        type(op_second) == AddOperation:
-                    yield ReplaceOperation({
-                        'op': 'replace',
-                        'path': op_second.location,
-                        'value': op_second.operation['value'],
-                    }).operation
-                    curr = curr[1][1]
-                    continue
+    def _item_moved(self, path, key, item, newkey):
+        """ The `item` was moved from `key` to `newkey`
+
+        [a, b, c, d, e]
+         0  1, 2, 3, 4
+
+        case 1: move 1 to 3
+        [a, c, d, b, e]
+         0  1, 2, 3, 4
+
+        """
+
+        key = self.get_index_change(path, key)
+        newkey = self.get_index_change(path, newkey)
+
+        if key == newkey:
+            return
+
+        new_op = MoveOperation({
+            'op': 'move',
+            'from': _path_join(path, key),
+            'path': _path_join(path, newkey),
+        })
+        self.insert(new_op)
+
+        self.add_index_change(path, key+1, -1)
+        self.add_index_change(path, newkey+1, +1)
+
+
+    def _item_removed(self, path, key, item, is_sequence):
+        if is_sequence:
+            key = self.get_index_change(path, key)
 
-            yield curr[2].operation
-            curr = curr[1]
-
-    def _item_added(self, path, key, item):
-        index = self.take_index(item, _ST_REMOVE)
-        if index is not None:
-            op = index[2]
-            if type(op.key) == int:
-                for v in self.iter_from(index):
-                    op.key = v._on_undo_remove(op.path, op.key)
-
-            self.remove(index)
-            if op.location != _path_join(path, key):
-                new_op = MoveOperation({
-                    'op': 'move',
-                    'from': op.location,
-                    'path': _path_join(path, key),
-                })
-                self.insert(new_op)
-        else:
-            new_op = AddOperation({
-                'op': 'add',
-                'path': _path_join(path, key),
-                'value': item,
-            })
-            new_index = self.insert(new_op)
-            self.store_index(item, new_index, _ST_ADD)
-
-    def _item_removed(self, path, key, item):
         new_op = RemoveOperation({
             'op': 'remove',
             'path': _path_join(path, key),
         })
-        index = self.take_index(item, _ST_ADD)
-        new_index = self.insert(new_op)
-        if index is not None:
-            op = index[2]
-            if type(op.key) == int:
-                for v in self.iter_from(index):
-                    op.key = v._on_undo_add(op.path, op.key)
-
-            self.remove(index)
-            if new_op.location != op.location:
-                new_op = MoveOperation({
-                    'op': 'move',
-                    'from': new_op.location,
-                    'path': op.location,
-                })
-                new_index[2] = new_op
+        self.insert(new_op)
 
-            else:
-                self.remove(new_index)
-
-        else:
-            self.store_index(item, new_index, _ST_REMOVE)
+        if is_sequence:
+            self.add_index_change(path, key, -1)
 
     def _item_replaced(self, path, key, item):
         self.insert(ReplaceOperation({
@@ -768,10 +708,10 @@ def _compare_dicts(self, path, src, dst):
         removed_keys = src_keys - dst_keys
 
         for key in removed_keys:
-            self._item_removed(path, str(key), src[key])
+            self._item_removed(path, str(key), src[key], False)
 
         for key in added_keys:
-            self._item_added(path, str(key), dst[key])
+            self._item_added(path, str(key), dst[key], False)
 
         for key in src_keys & dst_keys:
             self._compare_values(path, key, src[key], dst[key])
@@ -783,6 +723,7 @@ def _compare_lists(self, path, src, dst):
         for key in range(max_len):
             if key < min_len:
                 old, new = src[key], dst[key]
+
                 if old == new:
                     continue
 
@@ -795,14 +736,14 @@ def _compare_lists(self, path, src, dst):
                     self._compare_lists(_path_join(path, key), old, new)
 
                 else:
-                    self._item_removed(path, key, old)
-                    self._item_added(path, key, new)
+                    self._item_removed(path, key, old, True)
+                    self._item_added(path, key, new, True)
 
             elif len_src > len_dst:
-                self._item_removed(path, len_dst, src[key])
+                self._item_removed(path, len_dst, src[key], True)
 
             else:
-                self._item_added(path, key, dst[key])
+                self._item_added(path, key, dst[key], True)
 
     def _compare_values(self, path, key, src, dst):
         if src == dst:
@@ -819,6 +760,28 @@ def _compare_values(self, path, key, src, dst):
         else:
             self._item_replaced(path, key, dst)
 
+    def add_index_change(self, path, key, change):
+        """ Store the index change of a certain path
+
+        [a, b, c]
+         0, 1, 2
+
+        When inserting d at position 1, the indexes of b and c change
+        [a, d, b, c]
+         0, 1, 2, 3
+
+        add_index_change('/', 1, +1)
+        """
+
+        path_changes = self.index_changes.get(path, {})
+        key_change = path_changes.get(key, 0)
+        key_change = key_change + change
+        path_changes[key] = key_change
+
+    def get_index_change(self, path, key):
+        path_changes = self.index_changes.get(path, {})
+        return path_changes.get(key, 0) + key
+
 
 def _path_join(path, key):
     if key is None:

requirements-dev.txt

@@ -1,2 +1,3 @@
 wheel
 pypandoc
+hypothesis>=3; python_version=="2.7" or python_version>"3.3"

tests.py

@@ -9,6 +9,18 @@
 import jsonpatch
 import jsonpointer
 import sys
+import string
+
+try:
+    import hypothesis
+    from hypothesis import strategies as st
+except ImportError:
+    hypothesis = None
+
+try:
+    str = unicode  # Python 2
+except NameError:
+    pass
 
 
 class ApplyPatchTestCase(unittest.TestCase):
@@ -608,6 +620,30 @@ def test_replace_missing(self):
         self.assertRaises(jsonpatch.JsonPatchConflict, jsonpatch.apply_patch, src, patch_obj)
 
 
+if hypothesis is not None:
+    json_st = st.recursive(
+        st.one_of([
+            st.none(),
+            st.booleans(),
+            st.floats(allow_nan=False),
+            st.text(string.printable)]),
+        lambda children:
+        st.lists(children)
+        | st.dictionaries(st.text(string.printable), children))
+
+    class RoundtripTests(unittest.TestCase):
+        @hypothesis.example({}, {str('%20'): None})
+        @hypothesis.example({str('%20'): None}, {})
+        @hypothesis.given(json_st, json_st)
+        def test_roundtrip(self, src, dst):
+            patch = jsonpatch.JsonPatch.from_diff(src, dst, False)
+            hypothesis.note('Patch: %s' % (patch,))
+            res = patch.apply(src)
+            message = '{src} + {patch} resulted in {res}; {dst} was expected'.format(
+                src=repr(src), patch=repr(patch), res=repr(res), dst=repr(dst))
+            self.assertEqual(res, dst, message)
+
+
 if __name__ == '__main__':
     modules = ['jsonpatch']
 
@@ -622,6 +658,8 @@ def get_suite():
         suite.addTest(unittest.makeSuite(InvalidInputTests))
         suite.addTest(unittest.makeSuite(ConflictTests))
         suite.addTest(unittest.makeSuite(OptimizationTests))
+        if hypothesis is not None:
+            suite.addTest(unittest.makeSuite(RoundtripTests))
         return suite