# SPDX-FileCopyrightText: 2026 Bart Broere # SPDX-License-Identifier: GPL-3.0-or-later """End-to-end smoke test for a running boltage bridge, using the official Neo4j driver. Usage: python tests/smoke_test.py [bolt-uri] [user] [password] Requires: pip install neo4j """ import sys import neo4j from neo4j.graph import Node, Path, Relationship uri = sys.argv[1] if len(sys.argv) > 1 else "neo4j://localhost:7687" user = sys.argv[2] if len(sys.argv) > 2 else "postgres" password = sys.argv[3] if len(sys.argv) > 3 else "postgres" GRAPH = "boltage_smoke" def check(label, condition): print(("ok " if condition else "FAIL ") + label) if not condition: check.failed = True check.failed = False with neo4j.GraphDatabase.driver(uri, auth=(user, password)) as driver: driver.verify_connectivity() with driver.session(database="system") as s: s.run(f"DROP DATABASE {GRAPH} IF EXISTS").consume() s.run(f"CREATE DATABASE {GRAPH}").consume() names = [r["name"] for r in s.run("SHOW DATABASES")] check("SHOW DATABASES lists system and new graph", "system" in names and GRAPH in names) comp = s.run("CALL dbms.components() YIELD name, versions, edition").single() check("dbms.components", comp["name"] == "Neo4j Kernel") with driver.session(database=GRAPH) as s: s.run("CREATE (a:Person {name: $a, age: 30})-[:KNOWS {since: 2020}]->(b:Person {name: $b})", a="Ann", b="Bob's").consume() s.run("CREATE (:City {name: 'Utrecht', pop: 1.5})").consume() rec = s.run("MATCH (n:Person) WHERE n.name = $name RETURN n, n.age AS age", name="Ann").single() check("node returned as Node", isinstance(rec["n"], Node) and rec["n"]["name"] == "Ann") check("labels mapped", set(rec["n"].labels) == {"Person"}) check("alias column", rec["age"] == 30) rec = s.run("MATCH p=(a)-[r:KNOWS]->(b) RETURN p, r, b.name").single() check("path returned as Path", isinstance(rec["p"], Path) and len(rec["p"]) == 1) check("relationship returned", isinstance(rec["r"], Relationship) and rec["r"].type == "KNOWS") check("unaliased column name", rec.keys() == ["p", "r", "b.name"] and rec["b.name"] == "Bob's") ids = [r["id"] for r in s.run("MATCH (n) RETURN id(n) AS id")] rels = list(s.run("MATCH (a)-[r]->(b) WHERE id(a) IN $existingNodeIds AND id(b) IN $newNodeIds " "RETURN r;", existingNodeIds=ids, newNodeIds=ids)) check("browser relationship autocomplete query", len(rels) == 1) ann = s.run("MATCH (n {name: 'Ann'}) RETURN id(n) AS id").single()["id"] expand = list(s.run(f"""MATCH (a) WHERE id(a) = {ann} WITH a, size([(a)--() | 1]) AS allNeighboursCount MATCH path = (a)--(o) WHERE NOT id(o) IN [] RETURN path, allNeighboursCount ORDER BY id(o) LIMIT 100""")) check("browser expand-node query", len(expand) == 1 and expand[0]["allNeighboursCount"] == 1) meta = list(s.run(""" CALL db.labels() YIELD label RETURN {name:'labels', data:COLLECT(label)[..1000]} AS result UNION ALL CALL db.relationshipTypes() YIELD relationshipType RETURN {name:'relationshipTypes', data:COLLECT(relationshipType)[..1000]} AS result UNION ALL CALL db.propertyKeys() YIELD propertyKey RETURN {name:'propertyKeys', data:COLLECT(propertyKey)[..1000]} AS result """)) check("browser metadata query", meta[0]["result"]["data"] == ["City", "Person"] and meta[1]["result"]["data"] == ["KNOWS"] and "name" in meta[2]["result"]["data"]) counts = list(s.run(""" MATCH () RETURN { name:'nodes', data:count(*) } AS result UNION ALL MATCH ()-[]->() RETURN { name:'relationships', data: count(*)} AS result """)) check("browser count query", [c["result"]["data"] for c in counts] == [3, 1]) schema = s.run("CALL db.schema.visualization()").single() check("schema visualization", len(schema["nodes"]) == 2 and len(schema["relationships"]) == 1) rows = list(s.run("UNWIND range(1, 2500) AS i RETURN i")) check("large result over several PULLs", len(rows) == 2500 and rows[-1]["i"] == 2500) rows = [r["i"] for r in s.run("UNWIND range(1, 2500) AS i RETURN i ORDER BY i")] check("rows in order across batches", rows == list(range(1, 2501))) s.run("UNWIND range(1, 5) AS i CREATE (n:Discarded {i: i}) RETURN n.i").consume() check("discarded result of a write still writes", s.run("MATCH (n:Discarded) RETURN count(n) AS c").single()["c"] == 5) s.run("MATCH (n:Discarded) DELETE n RETURN 1").consume() check("write keyword in a string is a read", s.run("RETURN 'SET' AS v").consume().query_type == "r") rows = [] try: for r in s.run("UNWIND range(1, 1500) AS i RETURN 1 / (1500 - i) AS v"): rows.append(r["v"]) check("error after the first batch is reported", False) except neo4j.exceptions.Neo4jError: check("error after the first batch is reported", len(rows) >= 1000) check("recovers after error in a later batch", s.run("RETURN 1 AS x").single()["x"] == 1) check("string escaping roundtrip", s.run("RETURN $v AS v", v="a'b\\c\n\"d\" $x").single()["v"] == "a'b\\c\n\"d\" $x") check("map/list params", s.run("RETURN $m AS m", m={"k": [1, 2.5, None, True]}).single()["m"] == {"k": [1, 2.5, None, True]}) try: s.run("MATCH (n RETURN n").consume() check("syntax error reported", False) except neo4j.exceptions.CypherSyntaxError: check("syntax error reported", True) # The session must still be usable after a failure. check("recovers after error", s.run("RETURN 1 AS x").single()["x"] == 1) with driver.session(database=GRAPH) as s: with s.begin_transaction() as tx: tx.run("CREATE (:Temp {n: 1})").consume() tx.rollback() check("explicit rollback", s.run("MATCH (n:Temp) RETURN count(n) AS c").single()["c"] == 0) def work(tx): tx.run("CREATE (:Temp {n: 2})").consume() return tx.run("MATCH (n:Temp) RETURN count(n) AS c").single()["c"] check("managed write transaction", s.execute_write(work) == 1) def counts(tx): first = tx.run("CREATE (:Temp {n: 4})").consume().counters second = tx.run("MATCH (n:Temp {n: 4}) SET n.m = 1").consume().counters return first, second first, second = s.execute_write(counts) check("each query in a transaction reports its own counts", first.nodes_created == 1 and first.properties_set == 0 and second.nodes_created == 0 and second.properties_set == 1) def read_your_writes(tx): tx.run("CREATE (:Temp {n: 3}) RETURN 1") return tx.run("UNWIND range(1, 1500) AS i MATCH (n:Temp {n: 3}) RETURN i").value() check("query sees an earlier write in its transaction", len(s.execute_write(read_your_writes)) == 1500) with driver.session(database=GRAPH) as s: c = s.run("CREATE (a:Counted {i: 1})-[:LINK {w: 1}]->(b:Counted {i: 2})").consume().counters check("CREATE counts", c.nodes_created == 2 and c.relationships_created == 1 and c.labels_added == 2 and c.contains_updates) c = s.run("CREATE ({marker: 1})").consume().counters check("a node without a label adds no label", c.nodes_created == 1 and c.labels_added == 0) c = s.run("MATCH (n {marker: 1}) DELETE n").consume().counters check("DELETE counts", c.nodes_deleted == 1) summary = s.run("MATCH (n:Counted) SET n.j = 1, n.k = 2 RETURN n.j").consume() check("SET counts changed nodes", summary.counters.properties_set == 2 and summary.query_type == "rw") check("REMOVE counts", s.run("MATCH (n:Counted) REMOVE n.j").consume().counters.properties_set == 2) check("MATCH of nothing with SET changes nothing", not s.run("MATCH (n:Nothing) SET n.j = 1").consume().counters.contains_updates) c = s.run("MERGE (n:Counted {i: 1})").consume().counters check("MERGE that matches changes nothing", not c.contains_updates) c = s.run("MERGE (n:Counted {i: 3})").consume().counters check("MERGE that creates", c.nodes_created == 1) summary = s.run("CREATE (:Counted {i: 4})").consume() check("query without RETURN is a write", summary.query_type == "w") c = s.run("MATCH (n:Counted) DETACH DELETE n").consume().counters check("DETACH DELETE counts", c.nodes_deleted == 4 and c.relationships_deleted == 1) summary = s.run("MATCH (n) RETURN count(n) AS c").consume() check("a read changes nothing", not summary.counters.contains_updates and summary.query_type == "r") with driver.session(database="system") as s: c = s.run(f"DROP DATABASE {GRAPH}").consume().counters check("system updates", c.system_updates == 1 and c.contains_system_updates) sys.exit(1 if check.failed else 0)