Parameter type checking

Author: technige

neo4j/v1/api.py

@@ -21,12 +21,11 @@
 
 from collections import deque
 from random import random
-from threading import RLock
 from time import time, sleep
 from warnings import warn
 
 from neo4j.exceptions import ProtocolError, ServiceUnavailable
-from neo4j.compat import urlparse
+from neo4j.compat import urlparse, ustr, string, integer
 from neo4j.exceptions import CypherError, TransientError
 from neo4j.config import default_config
 
@@ -41,6 +40,9 @@
 RETRY_DELAY_MULTIPLIER = 2.0
 RETRY_DELAY_JITTER_FACTOR = 0.2
 
+INT64_MIN = -(2 ** 63)
+INT64_MAX = (2 ** 63) - 1
+
 
 def last_bookmark(b0, b1):
     """ Return the latest of two bookmarks by looking for the maximum
@@ -786,12 +788,30 @@ def fix_statement(statement):
     return statement
 
 
+def coerce_parameters(x):
+    if x is None:
+        return None
+    elif isinstance(x, bool):
+        return x
+    elif isinstance(x, integer):
+        if INT64_MIN <= x <= INT64_MAX:
+            return x
+        raise ValueError("Integer out of bounds (64-bit signed integer values only)")
+    elif isinstance(x, float):
+        return x
+    elif isinstance(x, string):
+        return ustr(x)
+    elif isinstance(x, (bytes, bytearray)):  # the order is important here - bytes must be checked after string
+        return x
+    elif isinstance(x, list):
+        return list(map(coerce_parameters, x))
+    elif isinstance(x, dict):
+        return {ustr(key): coerce_parameters(value) for key, value in x.items()}
+    else:
+        raise TypeError("Parameters of type {} are not supported".format(type(x).__name__))
+
+
 def fix_parameters(parameters=None, **kwparameters):
-    params_in = parameters or {}
-    params_in.update(kwparameters)
-    params_out = {}
-    for key, value in params_in.items():
-        if isinstance(key, bytes):
-            key = key.decode("UTF-8")
-        params_out[key] = value
-    return params_out
+    p = parameters or {}
+    p.update(kwparameters)
+    return coerce_parameters(p)

test/unit/test_api.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+
+# Copyright (c) 2002-2017 "Neo Technology,"
+# Network Engine for Objects in Lund AB [http://neotechnology.com]
+#
+# This file is part of Neo4j.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from unittest import TestCase
+from uuid import uuid4
+
+from neo4j.v1.api import coerce_parameters
+
+
+class ParameterTypeTestCase(TestCase):
+    def test_should_allow_none(self):
+        self.assertIsNone(coerce_parameters(None))
+
+    def test_should_allow_boolean(self):
+        self.assertTrue(coerce_parameters(True))
+        self.assertFalse(coerce_parameters(False))
+
+    def test_should_allow_integer(self):
+        self.assertEqual(coerce_parameters(0), 0)
+        self.assertEqual(coerce_parameters(0x7F), 0x7F)
+        self.assertEqual(coerce_parameters(0x7FFF), 0x7FFF)
+        self.assertEqual(coerce_parameters(0x7FFFFFFF), 0x7FFFFFFF)
+        self.assertEqual(coerce_parameters(0x7FFFFFFFFFFFFFFF), 0x7FFFFFFFFFFFFFFF)
+
+    def test_should_disallow_oversized_integer(self):
+        with self.assertRaises(ValueError):
+            coerce_parameters(0x10000000000000000)
+        with self.assertRaises(ValueError):
+            coerce_parameters(-0x10000000000000000)
+
+    def test_should_allow_float(self):
+        self.assertEqual(coerce_parameters(0.0), 0.0)
+        self.assertEqual(coerce_parameters(3.1415926), 3.1415926)
+
+    def test_should_allow_string(self):
+        self.assertEqual(coerce_parameters(u""), u"")
+        self.assertEqual(coerce_parameters(u"hello, world"), u"hello, world")
+
+    def test_should_allow_bytes(self):
+        self.assertEqual(coerce_parameters(bytearray()), bytearray())
+        self.assertEqual(coerce_parameters(bytearray([1, 2, 3])), bytearray([1, 2, 3]))
+
+    def test_should_allow_list(self):
+        self.assertEqual(coerce_parameters([]), [])
+        self.assertEqual(coerce_parameters([1, 2, 3]), [1, 2, 3])
+
+    def test_should_allow_dict(self):
+        self.assertEqual(coerce_parameters({}), {})
+        self.assertEqual(coerce_parameters({u"one": 1, u"two": 1, u"three": 1}), {u"one": 1, u"two": 1, u"three": 1})
+        self.assertEqual(coerce_parameters(
+            {u"list": [1, 2, 3, [4, 5, 6]], u"dict": {u"a": 1, u"b": 2}}),
+            {u"list": [1, 2, 3, [4, 5, 6]], u"dict": {u"a": 1, u"b": 2}})
+
+    def test_should_disallow_object(self):
+        with self.assertRaises(TypeError):
+            coerce_parameters(object())
+        with self.assertRaises(TypeError):
+            coerce_parameters(uuid4())