How to Write Unit Tests for Custom Flowsint Enrichers

To write unit tests for custom Flowsint enrichers, mock the vault and Neo4j graph service dependencies, initialize the enricher with await enricher.async_init(), and independently verify the asynchronous scan() method outputs and the synchronous postprocess() mock interactions.

Writing reliable unit tests for custom Flowsint enrichers ensures your workflow logic remains correct without requiring live vault secrets or a running Neo4j database. In the reconurge/flowsint repository, every enricher inherits from the abstract Enricher base class defined in flowsint-core, which provides a two-stage execution model that is straightforward to exercise in isolation. Understanding how to mock the external integration points is the key to writing fast, deterministic tests for any enricher you build.

Understand the Enricher Execution Model

Before writing assertions, it is important to understand the contract defined in flowsint-core/src/flowsint_core/core/enricher_base.py. The Enricher base class orchestrates input validation, parameter resolution (including vault secrets), and exposes a two-step execution model that concrete implementations must follow.

The lifecycle consists of two methods you must test:

• scan – An asynchronous method that receives a list of validated InputType objects and returns a list of OutputType objects.
• postprocess – A synchronous method that creates Neo4j graph nodes and relationships from the scan results.

When you write unit tests for custom Flowsint enrichers, verify three behaviors: that the parameter resolver pulls vault secrets correctly in resolve_params, that the scan logic produces expected outputs without calling external services, and that postprocess interacts with the graph service as intended.

Mock External Dependencies

Unit tests should never contact real vaults or databases. The repository’s own vault-integration tests in flowsint-enrichers/tests/test_vault_integration.py demonstrate the standard pattern: replace the vault client and the graph service with deterministic mocks.

Create fixtures that return a mocked vault and a mocked graph service. The vault mock must return predictable secrets so that parameter resolution succeeds during async_init(). The graph service mock must use AsyncMock for async methods like create_node so you can record calls without a live Neo4j instance.

import pytest
from unittest.mock import AsyncMock, Mock

from flowsint_enrichers.domain.to_ip import DomainToIpEnricher
from flowsint_types import Domain, Ip


@pytest.fixture
def mock_vault():
    vault = Mock()
    vault.get_secret = Mock(return_value="dummy-secret")
    return vault


@pytest.fixture
def mock_graph_service():
    service = Mock()
    service.create_node = AsyncMock()
    service.create_relationship = AsyncMock()
    return service

Initialize the Enricher in a Pytest Fixture

Instantiate the concrete enricher—such as the sample DomainToIpEnricher shown in the documentation at docs/developers/managing-enrichers.mdx—by passing the mocked vault and an empty params dictionary. You must call await enricher.async_init() to trigger parameter resolution, which is the same initialization path used in production. Then swap the real graph client for your mock by assigning e._graph_service.

@pytest.fixture
async def enricher(mock_vault, mock_graph_service):
    e = DomainToIpEnricher(
        sketch_id="test-sketch",
        scan_id="test-scan",
        vault=mock_vault,
        params={},
    )
    e._graph_service = mock_graph_service
    await e.async_init()
    return e

This fixture pattern mirrors the approach found in flowsint-enrichers/tests/test_vault_integration.py, where mocked dependencies are injected before the enricher performs any work.

Test the Asynchronous scan Method

The scan method contains the core enrichment logic and expects strongly typed inputs such as Domain, defined in the flowsint-types package. Because scan is async, decorate the test with @pytest.mark.asyncio and supply a list of validated InputType objects. Assert that the returned OutputType list matches your expectations.

@pytest.mark.asyncio
async def test_scan_resolves_domains(enricher):
    inputs = [Domain(domain="example.com"), Domain(domain="python.org")]
    results = await enricher.scan(inputs)

    assert len(results) == len(inputs)

    for ip_obj in results:
        assert isinstance(ip_obj, Ip)
        assert "." in ip_obj.address

This test verifies that the business logic produces the expected Ip objects for a given set of Domain inputs, completely in isolation from DNS servers or external APIs.

Test the Synchronous postprocess Method

After scan returns, postprocess transforms those outputs into graph structures. Because postprocess is synchronous, you do not need an async test decorator. Pass the fake scan results and original inputs to postprocess, then inspect the mock graph service to confirm it received the correct creation calls.

def test_postprocess_creates_graph_nodes(enricher):
    ip_results = [Ip(address="93.184.216.34"), Ip(address="138.197.63.241")]
    input_data = [Domain(domain="example.com"), Domain(domain="python.org")]

    enricher.postprocess(ip_results, input_data)

    assert enricher._graph_service.create_node.call_count == len(ip_results)

    for call_args in enricher._graph_service.create_node.call_args_list:
        payload = call_args[0][0]
        assert "address" in payload
        assert isinstance(payload["address"], str)

These assertions confirm that the enricher interacts with the graph layer exactly as intended, without requiring a running Neo4j database.

Complete Working Example

Combining the snippets above into a single test module gives you a ready-to-run file. Save it next to your enricher implementation at flowsint-enrichers/tests/test_domain_to_ip.py, and execute it with the following command:

uv run pytest flowsint-enrichers/tests/test_domain_to_ip.py

The full test file looks like this:

# File: flowsint-enrichers/tests/test_domain_to_ip.py

import pytest
from unittest.mock import AsyncMock, Mock

from flowsint_enrichers.domain.to_ip import DomainToIpEnricher
from flowsint_types import Domain, Ip


@pytest.fixture
def mock_vault():
    vault = Mock()
    vault.get_secret = Mock(return_value="dummy-secret")
    return vault


@pytest.fixture
def mock_graph_service():
    service = Mock()
    service.create_node = AsyncMock()
    service.create_relationship = AsyncMock()
    return service


@pytest.fixture
async def enricher(mock_vault, mock_graph_service):
    e = DomainToIpEnricher(
        sketch_id="test-sketch",
        scan_id="test-scan",
        vault=mock_vault,
        params={},
    )
    e._graph_service = mock_graph_service
    await e.async_init()
    return e


@pytest.mark.asyncio
async def test_scan_resolves_domains(enricher):
    inputs = [Domain(domain="example.com"), Domain(domain="python.org")]
    results = await enricher.scan(inputs)

    assert len(results) == len(inputs)
    for ip_obj in results:
        assert isinstance(ip_obj, Ip)
        assert "." in ip_obj.address


def test_postprocess_creates_graph_nodes(enricher):
    ip_results = [Ip(address="93.184.216.34"), Ip(address="138.197.63.241")]
    input_data = [Domain(domain="example.com"), Domain(domain="python.org")]

    enricher.postprocess(ip_results, input_data)

    assert enricher._graph_service.create_node.call_count == len(ip_results)
    for call_args in enricher._graph_service.create_node.call_args_list:
        payload = call_args[0][0]
        assert "address" in payload

Summary

• Mock the vault to return deterministic secrets so resolve_params succeeds during initialization without real credentials.
• Mock the graph service with AsyncMock to avoid needing a live Neo4j instance while still verifying node and relationship creation calls.
• Always call await enricher.async_init() in your fixture to trigger parameter resolution before testing scan or postprocess.
• Test scan with @pytest.mark.asyncio to validate that input objects produce the expected output objects.
• Test postprocess synchronously by inspecting call_count and call_args_list on the mocked graph service.
• Follow repository conventions by placing tests in flowsint-enrichers/tests/ and modeling fixtures after test_vault_integration.py as implemented in reconurge/flowsint.

Frequently Asked Questions

Do I need a running Neo4j database to unit test a custom enricher?

No. You should replace the real graph client with a mock that records calls. According to the reconurge/flowsint source code, the base class stores the graph service in EnricherBase._graph_service, so assigning a Mock with AsyncMock methods lets you verify postprocess behavior without any database running.

How do I handle vault secrets when writing unit tests for custom Flowsint enrichers?

Pass a unittest.mock.Mock object as the vault argument when constructing the enricher, and set vault.get_secret to return a fixed string. This mirrors the pattern in flowsint-enrichers/tests/test_vault_integration.py and ensures async_init() resolves parameters successfully without fetching real secrets.

Why must I call async_init() before testing scan or postprocess?

The async_init() method triggers parameter resolution, including evaluating vault references and applying defaults, which prepares the enricher for execution. Skipping this step leaves the enricher in an uninitialized state, so your tests may fail before reaching the actual logic you want to verify.

Can I test postprocess inside an async test function?

No, postprocess is a synchronous method, so it should be tested inside a standard def test function. While the enclosing fixture that provides the enricher is async, the test itself only needs to call enricher.postprocess() and inspect the mock graph service without any await expressions.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}