Added unit test for new 4.1 metadata

Author: technige

tests/unit/io/conftest.py

@@ -19,9 +19,13 @@
 # limitations under the License.
 
 
+from collections import deque
+from struct import unpack as struct_unpack
+
 import pytest
 
 from neo4j.io._common import MessageInbox
+from neo4j.packstream import UnpackableBuffer, Unpacker
 
 
 class FakeSocket:
@@ -54,6 +58,83 @@ def pop_message(self):
         return self.messages.pop()
 
 
+class FakeSocket2:
+
+    def __init__(self, address=None, on_send=None):
+        self.address = address
+        self.recv_buffer = bytearray()
+        self._messages = MessageInbox(self, on_error=print)
+        self.on_send = on_send
+
+    def getsockname(self):
+        return "127.0.0.1", 0xFFFF
+
+    def getpeername(self):
+        return self.address
+
+    def recv_into(self, buffer, nbytes):
+        data = self.recv_buffer[:nbytes]
+        actual = len(data)
+        buffer[:actual] = data
+        self.recv_buffer = self.recv_buffer[actual:]
+        return actual
+
+    def sendall(self, data):
+        if callable(self.on_send):
+            self.on_send(data)
+
+    def close(self):
+        return
+
+    def inject(self, data):
+        self.recv_buffer += data
+
+    def pop_chunk(self):
+        chunk_size, = struct_unpack(">H", self.recv_buffer[:2])
+        print("CHUNK SIZE %r" % chunk_size)
+        end = 2 + chunk_size
+        chunk_data, self.recv_buffer = self.recv_buffer[2:end], self.recv_buffer[end:]
+        return chunk_data
+
+    def pop_message(self):
+        data = bytearray()
+        while True:
+            chunk = self.pop_chunk()
+            print("CHUNK %r" % chunk)
+            if chunk:
+                data.extend(chunk)
+            elif data:
+                break       # end of message
+            else:
+                continue    # NOOP
+        header = data[0]
+        n_fields = header % 0x10
+        tag = data[1]
+        buffer = UnpackableBuffer(data[2:])
+        unpacker = Unpacker(buffer)
+        fields = [unpacker.unpack() for _ in range(n_fields)]
+        return tag, fields
+
+
+class FakeSocketPair:
+
+    def __init__(self, address):
+        self.client = FakeSocket2(address)
+        self.server = FakeSocket2()
+        self.client.on_send = self.server.inject
+        self.server.on_send = self.client.inject
+
+
 @pytest.fixture
 def fake_socket():
     return FakeSocket
+
+
+@pytest.fixture
+def fake_socket_2():
+    return FakeSocket2
+
+
+@pytest.fixture
+def fake_socket_pair():
+    return FakeSocketPair

tests/unit/io/test_class_bolt4x1.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+
+# Copyright (c) 2002-2020 "Neo4j,"
+# Neo4j Sweden AB [http://neo4j.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.
+
+
+import pytest
+
+from neo4j.io._bolt4 import Bolt4x1
+from neo4j.conf import PoolConfig
+
+
+def test_conn_timed_out(fake_socket):
+    address = ("127.0.0.1", 7687)
+    max_connection_lifetime = 0
+    connection = Bolt4x1(address, fake_socket(address), max_connection_lifetime)
+    assert connection.timedout() is True
+
+
+def test_conn_not_timed_out_if_not_enabled(fake_socket):
+    address = ("127.0.0.1", 7687)
+    max_connection_lifetime = -1
+    connection = Bolt4x1(address, fake_socket(address), max_connection_lifetime)
+    assert connection.timedout() is False
+
+
+def test_conn_not_timed_out(fake_socket):
+    address = ("127.0.0.1", 7687)
+    max_connection_lifetime = 999999999
+    connection = Bolt4x1(address, fake_socket(address), max_connection_lifetime)
+    assert connection.timedout() is False
+
+
+def test_db_extra_in_begin(fake_socket):
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.begin(db="something")
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x11"
+    assert len(fields) == 1
+    assert fields[0] == {"db": "something"}
+
+
+def test_db_extra_in_run(fake_socket):
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.run("", {}, db="something")
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x10"
+    assert len(fields) == 3
+    assert fields[0] == ""
+    assert fields[1] == {}
+    assert fields[2] == {"db": "something"}
+
+
+def test_n_extra_in_discard(fake_socket):
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.discard(n=666)
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x2F"
+    assert len(fields) == 1
+    assert fields[0] == {"n": 666}
+
+
+@pytest.mark.parametrize(
+    "test_input, expected",
+    [
+        (666, {"n": -1, "qid": 666}),
+        (-1, {"n": -1}),
+    ]
+)
+def test_qid_extra_in_discard(fake_socket, test_input, expected):
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.discard(qid=test_input)
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x2F"
+    assert len(fields) == 1
+    assert fields[0] == expected
+
+
+@pytest.mark.parametrize(
+    "test_input, expected",
+    [
+        (777, {"n": 666, "qid": 777}),
+        (-1, {"n": 666}),
+    ]
+)
+def test_n_and_qid_extras_in_discard(fake_socket, test_input, expected):
+    # python -m pytest tests/unit/io/test_class_bolt4x0.py -s -k test_n_and_qid_extras_in_discard
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.discard(n=666, qid=test_input)
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x2F"
+    assert len(fields) == 1
+    assert fields[0] == expected
+
+
+@pytest.mark.parametrize(
+    "test_input, expected",
+    [
+        (666, {"n": 666}),
+        (-1, {"n": -1}),
+    ]
+)
+def test_n_extra_in_pull(fake_socket, test_input, expected):
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.pull(n=test_input)
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x3F"
+    assert len(fields) == 1
+    assert fields[0] == expected
+
+
+@pytest.mark.parametrize(
+    "test_input, expected",
+    [
+        (777, {"n": -1, "qid": 777}),
+        (-1, {"n": -1}),
+    ]
+)
+def test_qid_extra_in_pull(fake_socket, test_input, expected):
+    # python -m pytest tests/unit/io/test_class_bolt4x0.py -s -k test_qid_extra_in_pull
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.pull(qid=test_input)
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x3F"
+    assert len(fields) == 1
+    assert fields[0] == expected
+
+
+def test_n_and_qid_extras_in_pull(fake_socket):
+    address = ("127.0.0.1", 7687)
+    socket = fake_socket(address)
+    connection = Bolt4x1(address, socket, PoolConfig.max_connection_lifetime)
+    connection.pull(n=666, qid=777)
+    connection.send_all()
+    tag, fields = socket.pop_message()
+    assert tag == b"\x3F"
+    assert len(fields) == 1
+    assert fields[0] == {"n": 666, "qid": 777}
+
+
+def test_hello_passes_routing_metadata(fake_socket_pair):
+    address = ("127.0.0.1", 7687)
+    sockets = fake_socket_pair(address)
+    # TODO helper method for encoding messages
+    sockets.server.sendall(b"\x00\x03\xB1\x70\xA0\x00\x00")
+    connection = Bolt4x1(address, sockets.client, PoolConfig.max_connection_lifetime,
+                         routing_context={"foo": "bar"})
+    connection.hello()
+    tag, fields = sockets.server.pop_message()
+    assert tag == 0x01
+    assert len(fields) == 1
+    assert fields[0]["routing"] == {"foo": "bar"}