GraphGuru | Neo4j in CI/CD: Versioned Schema Migrations and Testcontainers

Every Neo4j team we work with eventually hits the same wall: the graph in production does not match the graph on the developer's laptop. Someone added a constraint by hand in Browser six months ago. A MERGE in the new service assumes a property that only exists in staging. Nobody can rebuild the schema from scratch, so nobody dares change it.

Relational teams solved this two decades ago with migration tools and disposable test databases. Neo4j has both — they are just less talked about. This tutorial wires up the pair: neo4j-migrations for versioned, repeatable schema-as-code, and Testcontainers for integration tests that run Cypher against a real, throwaway Neo4j in CI. By the end you will have a pipeline where a pull request that breaks the graph fails before it merges.

What we are building

repo/
  neo4j/migrations/
    V001__constraints_and_indexes.cypher
    V002__add_supplier_country.cypher
    R001__reference_data.cypher
  src/...
  tests/test_orders.py
  .github/workflows/ci.yml

Three rules govern the whole design:

No schema change ever happens by hand. Constraints, indexes, and data backfills live in versioned files in the repo.
Tests never touch a shared database. Each test session gets its own container, seeded from the same migrations production runs.
CI runs both, so "the migration applies cleanly and the queries still work" is a merge gate rather than a deploy-day surprise.

Step 1 — the migration files

neo4j-migrations follows Flyway's conventions. Versioned migrations are named V<version>__<description>.cypher and run exactly once, in order, recorded in a __Neo4jMigration chain in the database. Repeatable migrations are named R<version>__<description>.cypher and re-run whenever their checksum changes — perfect for reference data.

neo4j/migrations/V001__constraints_and_indexes.cypher:

CREATE CONSTRAINT customer_id IF NOT EXISTS
FOR (c:Customer) REQUIRE c.id IS UNIQUE;

CREATE CONSTRAINT order_id IF NOT EXISTS
FOR (o:Order) REQUIRE o.id IS UNIQUE;

CREATE CONSTRAINT product_sku IF NOT EXISTS
FOR (p:Product) REQUIRE p.sku IS UNIQUE;

// Property existence constraints are Enterprise-only. Guard them in a
// separate migration if your dev tier is Community.
CREATE INDEX order_placed_at IF NOT EXISTS
FOR (o:Order) ON (o.placedAt);

neo4j/migrations/V002__add_supplier_country.cypher:

// Backfill in batches so a large graph does not blow the heap.
// CALL {} IN TRANSACTIONS requires an implicit (auto-commit) transaction,
// which neo4j-migrations gives you when the file contains a single statement.
MATCH (s:Supplier) WHERE s.country IS NULL
CALL (s) {
  SET s.country = coalesce(s.legacyCountryCode, 'UNKNOWN')
} IN TRANSACTIONS OF 10000 ROWS;

That CALL (s) { ... } form is the Cypher 25 variable-scope syntax; on older versions write CALL { WITH s ... }. If you are still deciding which language version your code targets, see our note on CalVer, Cypher 25, and GQL.

neo4j/migrations/R001__reference_data.cypher:

UNWIND [
  {code: 'STD', label: 'Standard shipping', days: 5},
  {code: 'EXP', label: 'Express shipping',  days: 2}
] AS row
MERGE (m:ShippingMethod {code: row.code})
SET   m.label = row.label, m.days = row.days;

Rules that keep migrations safe

Always IF NOT EXISTS on constraints and indexes, so a partially applied environment converges instead of erroring.
Never edit an applied V file. The tool checksums them; changing one after it ran makes the chain invalid. Fix forward with a new version.
Split schema from data. Constraint creation and a multi-million-node backfill in the same file makes rollback reasoning impossible.
Creating a constraint on dirty data fails. Put the cleanup migration before the constraint migration, not after.

Step 2 — run migrations from the CLI

# Download the CLI (a JVM binary; Homebrew and Docker images also exist)
neo4j-migrations \
  --address bolt://localhost:7687 \
  --username neo4j --password "$NEO4J_PASSWORD" \
  --location file:neo4j/migrations \
  migrate

Useful companions:

neo4j-migrations ... info      # what has been applied, and the checksums
neo4j-migrations ... validate  # exit non-zero if the DB is behind or drifted

validate is what you run in a deploy pipeline before starting the new application version. If you deploy on Kubernetes, run migrate as an init container or a Job that must complete before the rollout continues — one pod, never in parallel, because concurrent migration runners are the fastest way to corrupt the chain.

There are also embedded APIs: the JVM library (eu.michael-simons.neo4j:neo4j-migrations), a Spring Boot starter that migrates on startup, and a Quarkus extension. Python and Node shops usually shell out to the CLI in CI and keep the files language-neutral, which is what we assume below.

Step 3 — a disposable Neo4j for tests

Testcontainers starts a real Neo4j in Docker for the duration of your test session and throws it away afterwards. No shared staging database, no test pollution, no "works on my machine" schema.

pip install pytest testcontainers[neo4j] "neo4j>=6"

tests/conftest.py:

import subprocess
import pytest
from neo4j import GraphDatabase
from testcontainers.neo4j import Neo4jContainer

IMAGE = "neo4j:2026.01-enterprise"  # pin the exact version production runs

@pytest.fixture(scope="session")
def neo4j():
    container = (
        Neo4jContainer(IMAGE)
        .with_env("NEO4J_ACCEPT_LICENSE_AGREEMENT", "yes")
        .with_env("NEO4J_server_memory_heap_max__size", "1G")
    )
    with container as c:
        uri = c.get_connection_url()
        password = c.password

        # Apply the real migrations — the same files production uses.
        subprocess.run(
            ["neo4j-migrations", "--address", uri,
             "--username", "neo4j", "--password", password,
             "--location", "file:neo4j/migrations", "migrate"],
            check=True,
        )

        driver = GraphDatabase.driver(uri, auth=("neo4j", password))
        driver.verify_connectivity()
        yield driver
        driver.close()

Pin the image tag. neo4j:latest in CI means a Neo4j upgrade silently becomes part of an unrelated pull request, and you find out from a failing test at 5pm on a Friday.

Clean state between tests, cheaply

Restarting the container per test is far too slow. Delete the data instead and let the (much cheaper) schema survive:

@pytest.fixture(autouse=True)
def clean_graph(neo4j):
    with neo4j.session() as s:
        s.run("MATCH (n) CALL (n) { DETACH DELETE n } IN TRANSACTIONS OF 10000 ROWS")
    yield

On Enterprise you have a faster option: create a fresh database per test class with CREATE DATABASE test_$name and point the session at it. Constraints and indexes then have to be re-applied per database, so either re-run the migrations against it or keep the delete-based approach for simplicity.

A test that actually asserts something

def test_order_total_rolls_up_line_items(neo4j):
    with neo4j.session() as s:
        s.run("""
            CREATE (c:Customer {id: 'c1'})-[:PLACED]->(o:Order {id: 'o1', placedAt: datetime()})
            CREATE (p1:Product {sku: 'A', price: 10.0})
            CREATE (p2:Product {sku: 'B', price: 2.5})
            CREATE (o)-[:CONTAINS {qty: 2}]->(p1)
            CREATE (o)-[:CONTAINS {qty: 4}]->(p2)
        """)
        total = s.run("""
            MATCH (:Customer {id: $cid})-[:PLACED]->(o:Order)-[r:CONTAINS]->(p:Product)
            RETURN o.id AS order, sum(r.qty * p.price) AS total
        """, cid="c1").single()

    assert total["total"] == 30.0

def test_duplicate_sku_is_rejected(neo4j):
    """Proves V001 is actually applied, not just present in the repo."""
    from neo4j.exceptions import ConstraintError
    with neo4j.session() as s:
        s.run("CREATE (:Product {sku: 'A'})")
        with pytest.raises(ConstraintError):
            s.run("CREATE (:Product {sku: 'A'})")

The second test is the one people skip and the one that pays. It fails the moment somebody forgets to add a constraint to the migration set, which is exactly the drift you are trying to prevent.

Step 4 — guard the query plans too

Correctness tests do not catch a query that silently started scanning all nodes because an index was renamed. Assert on the plan:

def test_customer_lookup_uses_an_index(neo4j):
    with neo4j.session() as s:
        plan = s.run(
            "EXPLAIN MATCH (c:Customer {id: $id}) RETURN c", id="c1"
        ).consume().plan
    operators = str(plan)
    assert "NodeIndexSeek" in operators or "NodeUniqueIndexSeek" in operators
    assert "AllNodesScan" not in operators

EXPLAIN does not execute the query, so this is fast enough to run on every hot query you care about. For reading the plan itself — eager operators, row estimates, db hits — see Tuning Slow Cypher. Note that with an empty or tiny test graph the planner may legitimately choose a scan; seed a few hundred nodes in the fixture if the assertion proves flaky.

Step 5 — the CI workflow

name: ci
on: [pull_request]

jobs:
  test:
    runs-on: ubuntu-latest   # Docker is available, which Testcontainers needs
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - uses: actions/setup-java@v4
        with: { distribution: temurin, java-version: "21" }

      - name: Install neo4j-migrations CLI
        run: |
          curl -sSL -o migrations.zip \
            https://github.com/michael-simons/neo4j-migrations/releases/latest/download/neo4j-migrations.zip
          unzip -q migrations.zip -d "$HOME/tools"
          echo "$HOME/tools/neo4j-migrations/bin" >> "$GITHUB_PATH"

      - run: pip install -r requirements-dev.txt
      - run: pytest -q

Two things make this pipeline fast enough to keep: a session-scoped container (start Neo4j once, not once per test) and a warm Docker layer cache for the pinned image. A suite of a few hundred graph tests typically lands in two to four minutes.

Step 6 — the same migrations on the way to production

PR  ──> pytest + Testcontainers (migrations applied to a fresh DB)
    ──> merge
    ──> deploy job: neo4j-migrations validate  (fails if drift)
    ──> deploy job: neo4j-migrations migrate   (single runner)
    ──> app rollout

Before the first migrate against a live database, take a backup and know how to restore it — see Backups You Can Actually Restore. And for an existing graph that has never been under migration control, adopt it rather than rewrite it: write V001 to describe the schema as it exists today, run migrate against production once so the baseline is recorded, then make every subsequent change a new file.

Common failure modes

Symptom	Cause	Fix
`Checksum mismatch` on deploy	An applied `V` file was edited	Restore the original content; fix forward in a new version
Migration hangs on a big graph	Backfill in one transaction	`CALL { } IN TRANSACTIONS OF n ROWS`, one statement per file
Constraint creation fails	Duplicates already in the data	Cleanup migration first, constraint second
Testcontainers times out in CI	Enterprise image, no license env var, or a slow pull	Set `NEO4J_ACCEPT_LICENSE_AGREEMENT`, pin and cache the tag
Two pods both run `migrate`	Migration ran as part of app startup	Move it to a single-run `Job` or init container
Tests pass, production breaks	Test image ≠ production version	Pin the test image to the exact production version

Where to start tomorrow

You do not need the whole pipeline on day one. Get V001__constraints_and_indexes.cypher into the repo, run neo4j-migrations info against each environment, and look at what disagrees. That single command has surfaced undocumented indexes, missing constraints, and half-finished backfills in nearly every audit we have run — it is one of the standing items on our Neo4j health check checklist.

If you would like help retrofitting migrations onto a graph that has been hand-managed for years, or standing up a CI suite that your team will actually keep green, our senior Neo4j consultants do this regularly — get in touch.