Reject unsupported values before changing store data

jsonstore.py

@@ -7,19 +7,14 @@
 import json
 import os
 import stat
-import sys
 from collections import OrderedDict
 from copy import deepcopy
 from tempfile import mkstemp
 
 __all__ = ["JsonStore"]
 
 STRING_TYPES = (str,)
-INT_TYPES = (int,)
-if sys.version_info < (3,):
-    STRING_TYPES += (unicode,)
-    INT_TYPES += (long,)
-VALUE_TYPES = (bool, int, float, type(None)) + INT_TYPES
+VALUE_TYPES = (bool, int, float, type(None))
 
 
 class JsonStore(object):
@@ -113,7 +108,6 @@ def _verify_object(cls, obj, parents=None):
         Raise an exception if the object is not suitable for assignment.
 
         """
-        # pylint: disable=unicode-builtin,long-builtin
         if isinstance(obj, (dict, list)):
             if parents is None:
                 parents = []
@@ -131,7 +125,8 @@ def _verify_object(cls, obj, parents=None):
             for o in obj:
                 cls._verify_object(o, parents)
         else:
-            return cls._valid_value(obj)
+            if not cls._valid_value(obj):
+                raise TypeError("unsupported JSON value type: %s" % type(obj).__name__)
 
     @classmethod
     def _valid_value(cls, value):
@@ -163,7 +158,9 @@ def _canonical_key(cls, key):
 
     def __setattr__(self, attr, value):
         self._verify_object(value)
-        self._data[attr] = deepcopy(value)
+        value = deepcopy(value)
+        self._verify_object(value)
+        self._data[attr] = value
         self._do_auto_commit()
 
     def __delattr__(self, attr):
@@ -189,11 +186,13 @@ def __setitem__(self, key, value):
         steps = self._canonical_key(key)
         path, step = steps[:-1], steps[-1]
         self._verify_object(value)
+        value = deepcopy(value)
+        self._verify_object(value)
         container = self.__get_obj(path)
         if isinstance(container, dict) and not self._valid_string(step):
             raise TypeError("%s is a dict and %s is not a string" % (path, step))
         try:
-            container[step] = deepcopy(value)
+            container[step] = value
         except (IndexError, TypeError) as e:
             raise type(e)("unable to set %s from %s: %s" % (step, path, e))
         self._do_auto_commit()

test_jsonstore.py

@@ -125,6 +125,64 @@ def assign(value):
             self.assertRaises(TypeError, assign(None for i in range(2)))
             self.assertRaises(TypeError, assign({1: 1}))
 
+    def test_invalid_values_leave_store_unchanged(self):
+        values = (object(), set(), b"bytes", {"nested": object()},
+                  [1, {"nested": set()}], (1, object()), {1: "non-string key"})
+        for auto_commit in (False, True):
+            for method in ("attribute", "item", "nested item", "slice"):
+                for value in values:
+                    with self.subTest(auto_commit=auto_commit, method=method,
+                                      value_type=type(value).__name__):
+                        self.store.value = "original"
+                        self.store.nested = {"key": "original"}
+                        self.store.items = [1, 2]
+                        store = JsonStore(self._store_file, auto_commit=auto_commit)
+                        before = Path(self._store_file).read_bytes()
+                        with self.assertRaises(TypeError):
+                            if method == "attribute":
+                                store.value = value
+                            elif method == "item":
+                                store["value"] = value
+                            elif method == "nested item":
+                                store["nested.key"] = value
+                            else:
+                                store["items", :] = value
+                        self.assertEqual(store.value, "original")
+                        self.assertEqual(store.nested, {"key": "original"})
+                        self.assertEqual(store.items, [1, 2])
+                        self.assertEqual(Path(self._store_file).read_bytes(), before)
+                        with store:
+                            store.good = 1
+                        self.assertEqual(JsonStore(self._store_file).good, 1)
+
+    def test_invalid_value_cannot_poison_new_key(self):
+        for auto_commit in (False, True):
+            for assign in (setattr, lambda store, key, value: store.__setitem__(key, value)):
+                with self.subTest(auto_commit=auto_commit, assign=assign):
+                    store = JsonStore(self._store_file, auto_commit=auto_commit)
+                    with self.assertRaises(TypeError):
+                        assign(store, "bad", object())
+                    self.assertNotIn("bad", store)
+                    with store:
+                        store.good = 1
+                    self.assertEqual(JsonStore(self._store_file).good, 1)
+
+    def test_deepcopy_result_is_validated_before_assignment(self):
+        class InvalidCopy(str):
+            def __deepcopy__(self, memo):
+                return object()
+
+        class InvalidValue:
+            def __deepcopy__(self, memo):
+                return "valid-looking copy"
+
+        for assign in (setattr, lambda store, key, value: store.__setitem__(key, value)):
+            for value in (InvalidCopy("value"), InvalidValue()):
+                with self.subTest(assign=assign, value_type=type(value).__name__):
+                    with self.assertRaises(TypeError):
+                        assign(self.store, "bad", value)
+                    self.assertNotIn("bad", self.store)
+
     def test_assign_bad_keys(self):
         value = 1
         # the root object is a dict, so a string key is needed