138 lines
5.3 KiB
Python
138 lines
5.3 KiB
Python
# SPDX-FileCopyrightText: 2026 Bart Broere
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""End-to-end test of result streaming, speaking Bolt directly for control over PULL.
|
|
|
|
Usage: python tests/streaming_test.py [host] [port] [user] [password] [graph]
|
|
The graph must exist. Requires an importable boltage.
|
|
"""
|
|
|
|
import socket
|
|
import struct
|
|
import sys
|
|
|
|
from boltage.packstream import Packer, Structure, unpack
|
|
|
|
host = sys.argv[1] if len(sys.argv) > 1 else "localhost"
|
|
port = int(sys.argv[2]) if len(sys.argv) > 2 else 7687
|
|
user = sys.argv[3] if len(sys.argv) > 3 else "postgres"
|
|
password = sys.argv[4] if len(sys.argv) > 4 else "postgres"
|
|
graph = sys.argv[5] if len(sys.argv) > 5 else "neo4j"
|
|
|
|
HELLO, RESET, RUN, BEGIN, COMMIT, DISCARD, PULL = 0x01, 0x0F, 0x10, 0x11, 0x12, 0x2F, 0x3F
|
|
SUCCESS, RECORD, FAILURE = 0x70, 0x71, 0x7F
|
|
RANGE = "UNWIND range(1, 2500) AS i RETURN i"
|
|
|
|
|
|
def check(label, condition):
|
|
print(("ok " if condition else "FAIL ") + label)
|
|
if not condition:
|
|
check.failed = True
|
|
|
|
|
|
check.failed = False
|
|
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
self.sock = socket.create_connection((host, port), timeout=30)
|
|
self.sock.sendall(b"\x60\x60\xB0\x17" + bytes([0, 0, 4, 4]) + bytes(12))
|
|
assert self.read(4) == bytes([0, 0, 4, 4])
|
|
self.packer = Packer(element_ids=False)
|
|
self.request(HELLO, {"user_agent": "boltage-test", "scheme": "basic", "principal": user,
|
|
"credentials": password})
|
|
|
|
def read(self, n):
|
|
data = b""
|
|
while len(data) < n:
|
|
chunk = self.sock.recv(n - len(data))
|
|
if not chunk:
|
|
raise ConnectionError("closed by server")
|
|
data += chunk
|
|
return data
|
|
|
|
def message(self):
|
|
data = b""
|
|
while True:
|
|
size = struct.unpack(">H", self.read(2))[0]
|
|
if size == 0:
|
|
return unpack(data)
|
|
data += self.read(size)
|
|
|
|
def request(self, tag, *fields):
|
|
"""Send a message; return its records and the final SUCCESS or FAILURE."""
|
|
data = self.packer.pack(Structure(tag, list(fields)))
|
|
self.sock.sendall(struct.pack(">H", len(data)) + data + b"\x00\x00")
|
|
records = []
|
|
while True:
|
|
reply = self.message()
|
|
if reply.tag != RECORD:
|
|
return records, reply
|
|
records.append(reply.fields[0])
|
|
|
|
def run(self, query):
|
|
return self.request(RUN, query, {}, {"db": graph})[1]
|
|
|
|
|
|
|
|
c = Client()
|
|
|
|
check("RUN reports the columns", c.run(RANGE).fields[0]["fields"] == ["i"])
|
|
records, end = c.request(PULL, {"n": 10})
|
|
check("PULL n sends n records", [r[0] for r in records] == list(range(1, 11)))
|
|
check("PULL n reports has_more", end.tag == SUCCESS and end.fields[0].get("has_more") is True)
|
|
records, end = c.request(PULL, {"n": 1500})
|
|
check("PULL n crosses a batch boundary", [r[0] for r in records] == list(range(11, 1511)))
|
|
check("has_more while rows remain", end.fields[0].get("has_more") is True)
|
|
records, end = c.request(PULL, {"n": -1})
|
|
check("PULL all sends the rest", [r[0] for r in records] == list(range(1511, 2501)))
|
|
check("final summary", "has_more" not in end.fields[0] and "bookmark" in end.fields[0])
|
|
|
|
c.run("UNWIND range(1, 2000) AS i RETURN i")
|
|
records, end = c.request(PULL, {"n": 1000})
|
|
check("exactly one batch leaves has_more", len(records) == 1000 and end.fields[0].get("has_more") is True)
|
|
records, end = c.request(PULL, {"n": 1000})
|
|
check("exactly two batches end the result", len(records) == 1000 and "has_more" not in end.fields[0])
|
|
|
|
c.run("UNWIND range(1, 1000) AS i RETURN i")
|
|
records, end = c.request(PULL, {"n": 1000})
|
|
check("a result of exactly one batch ends", len(records) == 1000 and "has_more" not in end.fields[0])
|
|
|
|
c.run(RANGE)
|
|
c.request(PULL, {"n": 10})
|
|
records, end = c.request(DISCARD, {"n": -1})
|
|
check("DISCARD mid-result ends it", end.tag == SUCCESS and not records and "has_more" not in end.fields[0])
|
|
|
|
c.run(RANGE)
|
|
c.request(PULL, {"n": 10})
|
|
check("RUN replaces an unfinished result", c.run("RETURN 2 AS x").tag == SUCCESS)
|
|
records, _ = c.request(PULL, {"n": -1})
|
|
check("the new result is read", records == [[2]])
|
|
|
|
c.run(RANGE)
|
|
c.request(PULL, {"n": 10})
|
|
check("BEGIN after an unfinished result", c.request(BEGIN, {"db": graph})[1].tag == SUCCESS)
|
|
first = c.run("UNWIND range(1, 2500) AS i RETURN i").fields[0]["qid"]
|
|
second = c.run("UNWIND range(5001, 7500) AS i RETURN i").fields[0]["qid"]
|
|
values = []
|
|
for _ in range(1500):
|
|
values.append(c.request(PULL, {"n": 1, "qid": first})[0][0][0])
|
|
values.append(c.request(PULL, {"n": 1, "qid": second})[0][0][0])
|
|
check("two results in a transaction, pulled alternately",
|
|
values[:2] == [1, 5001] and values[-2:] == [1500, 6500])
|
|
check("COMMIT with results open", c.request(COMMIT, {})[1].tag == SUCCESS)
|
|
|
|
c.run("UNWIND range(1, 1500) AS i RETURN 1 / (1500 - i) AS v")
|
|
records, end = c.request(PULL, {"n": -1})
|
|
check("error in a later batch fails the PULL", end.tag == FAILURE and len(records) >= 1000)
|
|
check("RESET recovers", c.request(RESET)[1].tag == SUCCESS)
|
|
check("no transaction left aborted", c.run("RETURN 1 AS x").tag == SUCCESS)
|
|
c.request(PULL, {"n": -1})
|
|
|
|
c.request(BEGIN, {"db": graph})
|
|
c.run("RETURN 1 / 0 AS v")
|
|
check("error in a transaction's first batch keeps it failed until RESET", c.request(RESET)[1].tag == SUCCESS)
|
|
c.run("RETURN 1 AS x")
|
|
check("usable after that RESET", c.request(PULL, {"n": -1})[0] == [[1]])
|
|
|
|
sys.exit(1 if check.failed else 0)
|