boltage/tests/smoke_test.py

# SPDX-FileCopyrightText: 2026 Bart Broere
# SPDX-License-Identifier: GPL-3.0-or-later
"""End-to-end smoke test for a running peo bridge, using the official Neo4j driver.

Usage: python tests/smoke_test.py [bolt-uri] [user] [password]
Requires: pip install neo4j
Set PEO_SMOKE_PG_DSN (e.g. "host=localhost dbname=postgres") to also check the
generated property columns over SQL; that needs psycopg.
"""

import os
import sys
import time

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 = "peo_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)

        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)

    dsn = os.environ.get("PEO_SMOKE_PG_DSN")
    if dsn:
        import psycopg

        with psycopg.connect(dsn, user=user, password=password, autocommit=True) as pg:
            query = f'SELECT name, age FROM "{GRAPH}"."Person" ORDER BY name'
            for _ in range(30):
                try:
                    rows = pg.execute(query).fetchall()
                    break
                except psycopg.errors.UndefinedColumn:
                    time.sleep(0.5)
            else:
                rows = None
            check("generated property columns", rows == [("Ann", "30"), ("Bob's", None)])

    with driver.session(database="system") as s:
        s.run(f"DROP DATABASE {GRAPH}").consume()

sys.exit(1 if check.failed else 0)