diff --git a/jsonstore.py b/jsonstore.py index d1c6ddd..3c390e3 100644 --- a/jsonstore.py +++ b/jsonstore.py @@ -7,7 +7,6 @@ import json import os import stat -import sys from collections import OrderedDict from copy import deepcopy from tempfile import mkstemp @@ -15,11 +14,7 @@ __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() diff --git a/test_jsonstore.py b/test_jsonstore.py index 89e339a..9dfca13 100644 --- a/test_jsonstore.py +++ b/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