feat(tests): add OpenCypher DDL tests for index and constraint commands

Author: tae898

bindings/python/docs/api/database.md

@@ -213,6 +213,13 @@ db.command("sql", "CREATE DOCUMENT TYPE Person")
 db.command("sql", "CREATE PROPERTY Person.name STRING")
 db.command("sql", "CREATE PROPERTY Person.age INTEGER")
 
+# OpenCypher DDL also passes straight through to the engine
+db.command("opencypher", "CREATE INDEX FOR (p:Person) ON (p.email)")
+db.command(
+    "opencypher",
+    "CREATE CONSTRAINT FOR (p:Person) REQUIRE p.email IS TYPED STRING",
+)
+
 # Data operations must be in a transaction
 with db.transaction():
     db.command("sql", "INSERT INTO Person SET name = ?, age = ?", "Alice", 30)

bindings/python/tests/test_core.py

@@ -340,6 +340,66 @@ def test_truncate_bucket(temp_db_path):
         assert db.count_type("BucketDoc") == 0
 
 
+def test_update_batch_clause(temp_db_path):
+    """UPDATE ... BATCH executes successfully through the Python SQL pass-through."""
+    with arcadedb.create_database(temp_db_path) as db:
+        db.command("sql", "CREATE DOCUMENT TYPE BatchUpdateDoc")
+
+        with db.transaction():
+            for idx in range(25):
+                db.command(
+                    "sql",
+                    "INSERT INTO BatchUpdateDoc SET idx = ?, processed = false",
+                    idx,
+                )
+
+        with db.transaction():
+            db.command(
+                "sql",
+                "UPDATE BatchUpdateDoc SET processed = true WHERE idx >= 10 BATCH 5",
+            )
+
+        processed = db.query(
+            "sql",
+            "SELECT count(*) AS c FROM BatchUpdateDoc WHERE processed = true",
+        ).one()
+        untouched = db.query(
+            "sql",
+            "SELECT count(*) AS c FROM BatchUpdateDoc WHERE processed = false",
+        ).one()
+
+        assert processed.get("c") == 15
+        assert untouched.get("c") == 10
+
+
+def test_delete_batch_clause(temp_db_path):
+    """DELETE ... BATCH executes successfully through the Python SQL pass-through."""
+    with arcadedb.create_database(temp_db_path) as db:
+        db.command("sql", "CREATE DOCUMENT TYPE BatchDeleteDoc")
+
+        with db.transaction():
+            for idx in range(30):
+                db.command("sql", "INSERT INTO BatchDeleteDoc SET idx = ?", idx)
+
+        with db.transaction():
+            db.command(
+                "sql",
+                "DELETE FROM BatchDeleteDoc WHERE idx % 2 = 0 BATCH 4",
+            )
+
+        remaining = db.query(
+            "sql",
+            "SELECT count(*) AS c FROM BatchDeleteDoc",
+        ).one()
+        odd_count = db.query(
+            "sql",
+            "SELECT count(*) AS c FROM BatchDeleteDoc WHERE idx % 2 = 1",
+        ).one()
+
+        assert remaining.get("c") == 15
+        assert odd_count.get("c") == 15
+
+
 def test_transactions(temp_db_path):
     """Test transaction support."""
     with arcadedb.create_database(temp_db_path) as db:

bindings/python/tests/test_cypher.py

@@ -389,6 +389,155 @@ def test_opencypher_count_non_existing_label_returns_zero(temp_db_path):
         assert row.get("c") == 0
 
 
+def test_opencypher_create_index_command(temp_db_path):
+    """Test CREATE INDEX DDL passes through to the OpenCypher engine."""
+    with arcadedb.create_database(temp_db_path) as db:
+        _ensure_opencypher(db)
+
+        db.command("opencypher", "CREATE INDEX FOR (n:Event) ON (n.code)")
+
+        with db.transaction():
+            db.command("opencypher", "CREATE (:Event {code: 'evt-1', name: 'Launch'})")
+
+        row = db.query(
+            "opencypher",
+            "MATCH (n:Event {code: 'evt-1'}) RETURN n.name AS name",
+        ).one()
+
+        assert row.get("name") == "Launch"
+
+
+def test_opencypher_typed_constraint_command(temp_db_path):
+    """Test typed Cypher constraints are accepted through the Python bindings."""
+    with arcadedb.create_database(temp_db_path) as db:
+        _ensure_opencypher(db)
+
+        db.command(
+            "opencypher",
+            "CREATE CONSTRAINT FOR (p:Person) REQUIRE p.email IS TYPED STRING",
+        )
+
+        with db.transaction():
+            db.command(
+                "opencypher",
+                "CREATE (:Person {email: 'alice@example.com', name: 'Alice'})",
+            )
+
+        row = db.query(
+            "opencypher",
+            "MATCH (p:Person {email: 'alice@example.com'}) RETURN p.name AS name",
+        ).one()
+
+        assert row.get("name") == "Alice"
+
+
+def test_opencypher_create_index_if_not_exists_command(temp_db_path):
+    """Test CREATE INDEX IF NOT EXISTS is idempotent through the Python bindings."""
+    with arcadedb.create_database(temp_db_path) as db:
+        _ensure_opencypher(db)
+
+        db.command(
+            "opencypher",
+            "CREATE INDEX IF NOT EXISTS FOR (n:Metric) ON (n.code)",
+        )
+        db.command(
+            "opencypher",
+            "CREATE INDEX IF NOT EXISTS FOR (n:Metric) ON (n.code)",
+        )
+
+        with db.transaction():
+            db.command("opencypher", "CREATE (:Metric {code: 'm-1', value: 42})")
+
+        row = db.query(
+            "opencypher",
+            "MATCH (n:Metric {code: 'm-1'}) RETURN n.value AS value",
+        ).one()
+
+        assert row.get("value") == 42
+
+
+def test_opencypher_ddl_auto_creates_vertex_and_edge_types(temp_db_path):
+    """Test Cypher DDL auto-creates missing vertex and edge types via Python."""
+    with arcadedb.create_database(temp_db_path) as db:
+        _ensure_opencypher(db)
+
+        assert db.schema.exists_type("AutoEvent") is False
+        assert db.schema.exists_type("RELATES_TO") is False
+
+        db.command("opencypher", "CREATE INDEX FOR (n:AutoEvent) ON (n.code)")
+        db.command(
+            "opencypher",
+            "CREATE CONSTRAINT FOR ()-[r:RELATES_TO]-() REQUIRE r.since IS NOT NULL",
+        )
+
+        assert db.schema.exists_type("AutoEvent") is True
+        assert db.schema.exists_type("RELATES_TO") is True
+
+        with db.transaction():
+            db.command("opencypher", "CREATE (:AutoEvent {code: 'a'})")
+            db.command("opencypher", "CREATE (:AutoEvent {code: 'b'})")
+
+        source = db.query(
+            "sql",
+            "SELECT @rid AS rid FROM AutoEvent WHERE code = 'a'",
+        ).one()
+        target = db.query(
+            "sql",
+            "SELECT @rid AS rid FROM AutoEvent WHERE code = 'b'",
+        ).one()
+
+        with db.transaction():
+            db.command(
+                "sql",
+                f"CREATE EDGE RELATES_TO FROM {source.get('rid')} TO {target.get('rid')} "
+                "SET since = '2026-04-07'",
+            )
+
+        row = db.query(
+            "opencypher",
+            "MATCH (:AutoEvent {code: 'a'})-[r:RELATES_TO]->(:AutoEvent {code: 'b'}) "
+            "RETURN r.since AS since",
+        ).one()
+
+        assert row.get("since") == "2026-04-07"
+
+
+def test_opencypher_edge_typed_constraint_command(temp_db_path):
+    """Test typed edge constraints are accepted through the Python bindings."""
+    with arcadedb.create_database(temp_db_path) as db:
+        _ensure_opencypher(db)
+
+        db.command("sql", "CREATE VERTEX TYPE Person")
+        db.command(
+            "opencypher",
+            "CREATE CONSTRAINT FOR ()-[r:KNOWS]-() REQUIRE r.since IS TYPED DATE",
+        )
+
+        with db.transaction():
+            db.command("sql", "INSERT INTO Person SET name = 'Alice'")
+            db.command("sql", "INSERT INTO Person SET name = 'Bob'")
+
+        alice = db.query(
+            "sql",
+            "SELECT @rid AS rid FROM Person WHERE name = 'Alice'",
+        ).one()
+        bob = db.query(
+            "sql",
+            "SELECT @rid AS rid FROM Person WHERE name = 'Bob'",
+        ).one()
+
+        with db.transaction():
+            db.command(
+                "sql",
+                f"CREATE EDGE KNOWS FROM {alice.get('rid')} TO {bob.get('rid')} "
+                "SET since = date('2026-04-07')",
+            )
+
+        row = db.query("sql", "SELECT since FROM KNOWS").one()
+
+        assert row.get("since") is not None
+
+
 def test_opencypher_projection_property_order_is_preserved(temp_db_path):
     """Test projected property order is stable and matches RETURN clause order."""
     with arcadedb.create_database(temp_db_path) as db: