From 789b44fe908c58037f67a2c925ccb2ab9762b7dd Mon Sep 17 00:00:00 2001 From: Nigel Small Date: Wed, 29 Nov 2017 10:57:40 +0000 Subject: [PATCH] Parameter type checking --- neo4j/v1/api.py | 40 +++++++++++++++++------ test/unit/test_api.py | 76 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 10 deletions(-) create mode 100644 test/unit/test_api.py diff --git a/neo4j/v1/api.py b/neo4j/v1/api.py index 652b7154..4b351192 100644 --- a/neo4j/v1/api.py +++ b/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) diff --git a/test/unit/test_api.py b/test/unit/test_api.py new file mode 100644 index 00000000..87188e46 --- /dev/null +++ b/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())