Now supports both Python 2.7 and 3.5. Polished code from feedback on prvious commit

Author: RagnarW

test/tck/configure_feature_files.py

@@ -1,7 +1,5 @@
 import os
 import tarfile
-import urllib2
-
 
 def clean_up():
     dir_path = (os.path.dirname(os.path.realpath(__file__)))
@@ -13,18 +11,28 @@ def clean_up():
 
 def set_up():
     dir_path = (os.path.dirname(os.path.realpath(__file__)))
-    feature_url = "https://s3-eu-west-1.amazonaws.com/remoting.neotechnology.com/driver-compliance/tck.tar.gz"
-    file_name = feature_url.split('/')[-1]
-    tar = open(file_name, 'w')
-    response = urllib2.urlopen(feature_url)
-    block_sz = 1024
-    while True:
-        buffer = response.read(block_sz)
-        if not buffer:
-            break
-        tar.write(buffer)
-    tar.close()
+    url = "https://s3-eu-west-1.amazonaws.com/remoting.neotechnology.com/driver-compliance/tck.tar.gz"
+    file_name = url.split('/')[-1]
+    _download_tar(url,file_name)
+
     tar = tarfile.open(file_name)
     tar.extractall(dir_path)
     tar.close()
     os.remove(file_name)
+
+
+def _download_tar(url, file_name):
+    try:
+        import urllib2
+        tar = open(file_name, 'w')
+        response = urllib2.urlopen(url)
+        block_sz = 1024
+        while True:
+            buffer = response.read(block_sz)
+            if not buffer:
+                break
+            tar.write(buffer)
+        tar.close()
+    except ImportError:
+        from urllib import request
+        request.urlretrieve(url, file_name)

test/tck/environment.py

@@ -15,8 +15,8 @@ def after_scenario(context, scenario):
     for step in scenario.steps:
         if step.status == 'failed':
             logging.error("Scenario :'%s' at step: '%s' failed! ", scenario.name, step.name)
-            logging.debug("Expected result: %s", tck_util.as_cypger_text(context.expected))
-            logging.debug("Actual result: %s", tck_util.as_cypger_text(context.results))
+            logging.debug("Expected result: %s", tck_util.as_cypher_text(context.expected))
+            logging.debug("Actual result: %s", tck_util.as_cypher_text(context.results))
         if step.status == 'skipped':
             logging.warn("Scenario :'%s' at step: '%s' was skipped! ", scenario.name, step.name)
         if step.status == 'passed':

test/tck/steps/bolt_type_steps.py

@@ -2,8 +2,6 @@
 
 from test.tck import tck_util
 
-from neo4j.v1.typesystem import Node, Relationship, Path
-
 use_step_matcher("re")
 
 
@@ -13,19 +11,19 @@ def step_impl(context):
     # check if running
 
 
-@given("a value (?P<Input>.+) of type (?P<BoltType>.+)")
-def step_impl(context, Input, BoltType):
-    context.expected = tck_util.get_bolt_value(BoltType, Input)
+@given("a value (?P<input>.+) of type (?P<bolt_type>.+)")
+def step_impl(context, input, bolt_type):
+    context.expected = tck_util.get_bolt_value(bolt_type, input)
 
 
-@given("a value  of type (?P<BoltType>.+)")
-def step_impl(context, BoltType):
-    context.expected = tck_util.get_bolt_value(BoltType, u' ')
+@given("a value  of type (?P<bolt_type>.+)")
+def step_impl(context, bolt_type):
+    context.expected = tck_util.get_bolt_value(bolt_type, u' ')
 
 
-@given("a list value (?P<Input>.+) of type (?P<BoltType>.+)")
-def step_impl(context, Input, BoltType):
-    context.expected = tck_util.get_list_from_feature_file(Input, BoltType)
+@given("a list value (?P<input>.+) of type (?P<bolt_type>.+)")
+def step_impl(context, input, bolt_type):
+    context.expected = tck_util.get_list_from_feature_file(input, bolt_type)
 
 
 @given("an empty list L")
@@ -43,14 +41,14 @@ def step_impl(context, size):
     context.expected = tck_util.get_random_string(int(size))
 
 
-@given("a List of size (?P<size>\d+) and type (?P<Type>.+)")
-def step_impl(context, size, Type):
-    context.expected = tck_util.get_list_of_random_type(int(size), Type)
+@given("a List of size (?P<size>\d+) and type (?P<type>.+)")
+def step_impl(context, size, type):
+    context.expected = tck_util.get_list_of_random_type(int(size), type)
 
 
-@given("a Map of size (?P<size>\d+) and type (?P<Type>.+)")
-def step_impl(context, size, Type):
-    context.expected = tck_util.get_dict_of_random_type(int(size), Type)
+@given("a Map of size (?P<size>\d+) and type (?P<type>.+)")
+def step_impl(context, size, type):
+    context.expected = tck_util.get_dict_of_random_type(int(size), type)
 
 
 @step("adding a table of lists to the list L")
@@ -68,7 +66,7 @@ def step_impl(context):
 @step("adding a table of values to the map M")
 def step_impl(context):
     for row in context.table:
-        context.M['a' + str(len(context.M))] = tck_util.get_bolt_value(row[0], row[1])
+        context.M['a%d' % len(context.M)] = tck_util.get_bolt_value(row[0], row[1])
 
 
 @step("adding map M to list L")
@@ -79,40 +77,40 @@ def step_impl(context):
 @when("adding a table of lists to the map M")
 def step_impl(context):
     for row in context.table:
-        context.M['a' + str(len(context.M))] = tck_util.get_list_from_feature_file(row[1], row[0])
+        context.M['a%d' % len(context.M)] = tck_util.get_list_from_feature_file(row[1], row[0])
 
 
 @step("adding a copy of map M to map M")
 def step_impl(context):
-    context.M['a' + str(len(context.M))] = context.M.copy()
+    context.M['a%d' % len(context.M)] = context.M.copy()
 
 
 @when("the driver asks the server to echo this value back")
 def step_impl(context):
     context.results = {}
-    context.results["as_string"] = tck_util.send_string("RETURN " + tck_util.as_cypger_text(context.expected))
+    context.results["as_string"] = tck_util.send_string("RETURN " + tck_util.as_cypher_text(context.expected))
     context.results["as_parameters"] = tck_util.send_parameters("RETURN {input}", {'input': context.expected})
 
 
 @when("the driver asks the server to echo this list back")
 def step_impl(context):
     context.expected = context.L
     context.results = {}
-    context.results["as_string"] = tck_util.send_string("RETURN " + tck_util.as_cypger_text(context.expected))
+    context.results["as_string"] = tck_util.send_string("RETURN " + tck_util.as_cypher_text(context.expected))
     context.results["as_parameters"] = tck_util.send_parameters("RETURN {input}", {'input': context.expected})
 
 
 @when("the driver asks the server to echo this map back")
 def step_impl(context):
     context.expected = context.M
     context.results = {}
-    context.results["as_string"] = tck_util.send_string("RETURN " + tck_util.as_cypger_text(context.expected))
+    context.results["as_string"] = tck_util.send_string("RETURN " + tck_util.as_cypher_text(context.expected))
     context.results["as_parameters"] = tck_util.send_parameters("RETURN {input}", {'input': context.expected})
 
 
 @then("the result returned from the server should be a single record with a single value")
 def step_impl(context):
-    assert len(context.results) > 0
+    assert context.results
     for result in context.results.values():
         assert len(result) == 1
         assert len(result[0]) == 1
@@ -123,18 +121,4 @@ def step_impl(context):
     assert len(context.results) > 0
     for result in context.results.values():
         result_value = result[0].values()[0]
-        assert result_value == context.expected
-
-
-@step("the node value given in the result should be the same as what was sent")
-def step_impl(context):
-    assert len(context.results) > 0
-    for result in context.results.values():
-        result_value = result[0].values()[0]
-        assert result_value == context.expected
-        assert result_value.labels == context.expected.labels
-        assert result_value.keys() == context.expected.keys()
-        assert result_value.values() == context.expected.values()
-        assert result_value.items() == context.expected.items()
-        assert len(result_value) == len(context.expected)
-        assert iter(result_value) == iter(context.expected)
+        assert result_value == context.expected

test/tck/tck_util.py

@@ -1,7 +1,6 @@
 import string
 import random
-
-import sys
+from neo4j.v1 import compat
 
 from neo4j.v1 import GraphDatabase
 
@@ -28,40 +27,40 @@ def get_bolt_value(type, value):
     if type == 'Float':
         return float(value)
     if type == 'String':
-        return value
+        return to_unicode(value)
     if type == 'Null':
         return None
     if type == 'Boolean':
         return bool(value)
-    raise ValueError('No such type : %s', type)
+    raise ValueError('No such type : %s' % type)
 
 
-def as_cypger_text(expected):
+def as_cypher_text(expected):
     if expected is None:
         return "Null"
-    if isinstance(expected, unicode):
-        return '"' + expected + '"'
+    if isinstance(expected, (str, compat.string)):
+            return '"' + expected + '"'
     if isinstance(expected, float):
         return repr(expected).replace('+', '')
     if isinstance(expected, list):
         l = u'['
         for i, val in enumerate(expected):
-            l += as_cypger_text(val)
+            l += as_cypher_text(val)
             if i < len(expected)-1:
                 l+= u','
         l += u']'
         return l
     if isinstance(expected, dict):
         d = u'{'
         for i, (key, val) in enumerate(expected.items()):
-            d += unicode(key) + ':'
-            d += as_cypger_text(val)
+            d += to_unicode(key) + ':'
+            d += as_cypher_text(val)
             if i < len(expected.items())-1:
                 d+= u','
         d += u'}'
         return d
     else:
-        return unicode(expected)
+        return to_unicode(expected)
 
 
 def get_list_from_feature_file(string_list, bolt_type):
@@ -89,7 +88,7 @@ def get_none():
 
     if type == 'Integer':
         fu = random.randint
-        args = [-sys.maxint - 1, sys.maxint]
+        args = [-9223372036854775808, 9223372036854775808]
     elif type == 'Float':
         fu = random.random
         args = []
@@ -103,7 +102,7 @@ def get_none():
         fu = get_random_bool
         args = []
     else:
-        raise ValueError('No such type : %s', type)
+        raise ValueError('No such type : %s' % type)
     return (fu, args)
 
 
@@ -114,7 +113,11 @@ def get_list_of_random_type(size, type):
 
 def get_dict_of_random_type(size, type):
     fu, args = _get_random_func(type)
-    map = {}
-    for i in range(size):
-        map['a' + str(i)] = fu(*args)
-    return map
+    return {'a%d' % i: fu(*args) for i in range(size)}
+
+def to_unicode(val):
+    try:
+        return unicode(val)
+    except NameError:
+        return str(val)
+