Testing

Tests for ETL pipelines are, by definition, integration tests. The library provides utilities to simplify setting up a connection to Neo4j.

The utilities integrate with pytest. See the provided examples for possible setup and usage.

There are two ways to use Neo4j with tests: either with the excellent TestContainers or by using an existing Neo4j installation.

The module utils contains mock implementations of core components to set up an ETLContext and its dependencies. These utilities are designed to integrate seamlessly and use environment variables for configuration.

Fixtures

In addition to the above module, the following code is recommended to be placed in the root of the testing package as conftest.py. The pytest fixtures defined within it will be available to all tests.

@pytest.fixture(scope="session")
def neo4j_driver():
    """
    Creates a Neo4j driver instance.
    If the environment variable NEO4J_TEST_CONTAINER is set, it will be used as the image name to start Neo4j in a
    TestContainer.
    If the variable is not set, then the following environment variables will be used to connect to running instance:
    `NEO4J_URI`
    `NEO4J_USERNAME`
    `NEO4J_PASSWORD`
    `NEO4J_TEST_DATABASE`
    The later can be used to direct tests away from the default DB,
    :return:
    """
    neo4j_container = os.getenv("NEO4J_TEST_CONTAINER")
    if neo4j_container is not None:
        print(f"found NEO4J_TEST_CONTAINER with {neo4j_container}, using test containers")
        from testcontainers.neo4j import Neo4jContainer
        with (Neo4jContainer(
                image=neo4j_container,
                username="neo4j",
                password=str(uuid.uuid4()))
                      .with_env("NEO4J_PLUGINS", "[\"apoc\", \"graph-data-science\"],")
                      .with_env("NEO4J_ACCEPT_LICENSE_AGREEMENT", "yes") as neo4j):
            driver = neo4j.get_driver()
            yield driver
            driver.close()
            return
    else:
        print(
            f"found NEO4J_TEST_CONTAINER not set, using instance at  {os.getenv('NEO4J_URI')} and database={os.getenv('NEO4J_TEST_DATABASE')}")
        # do not use test containers, but a running remote db
        with GraphDatabase.driver(os.getenv('NEO4J_URI'),
                                  auth=(os.getenv('NEO4J_USERNAME'),
                                        os.getenv('NEO4J_PASSWORD')),
                                  notifications_min_severity="OFF") as driver:
            yield driver
            driver.close()
            return


@pytest.fixture
def neo4j_driver_with_empty_db(neo4j_driver):
    with neo4j_driver.session(database=get_database_name(), default_access_mode=WRITE_ACCESS) as session:
        session.run("MATCH (n) DETACH DELETE n")
        yield neo4j_driver


@pytest.fixture
def etl_context(neo4j_driver_with_empty_db, tmp_path) -> ETLContext:
    return MockETLContext(neo4j_driver_with_empty_db, tmp_path)

The following fixtures are provided:

neo4j_driver_with_empty_db : Provides a Neo4j driver object connected to an empty database. Typically used in data setup steps.
neo4j_driver : Provides a Neo4j driver object connected to a running Neo4j installation.
etl_context : Provides a ETLContext for testing purposes. The simple reporter is used, and a pytest temporary directory is set up for error files (Validation).

Selecting a Neo4j Installation

Testcontainers

If the NEO4J_TEST_CONTAINER environment variable is set to the name of a Neo4j Docker image, TestContainers will be used to run Neo4j. This requires Docker to be available on the host running the tests.

The provided installation will have apoc core and gds plugins installed.

External Installation

If the NEO4J_TEST_CONTAINER environment variable is not set, the NEO4J_URI, NEO4J_USERNAME, and NEO4J_PASSWORD variables must be set. These will be used to connect to an existing Neo4j instance.

The functions in utils respect the NEO4J_TEST_DATABASE variable when running queries, allowing for separate databases per user.