# How to Write Unit Tests for Custom Flowsint Enrichers

> Learn to write unit tests for custom Flowsint enrichers. Mock dependencies, initialize enrichers, and test async scan and sync postprocess methods for robust code.

- Repository: [reconurge/flowsint](https://github.com/reconurge/flowsint)
- Tags: how-to-guide
- Published: 2026-06-05

---

**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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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.

```python
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`.

```python
@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`](https://github.com/reconurge/flowsint/blob/main/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.

```python
@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.

```python
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`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/tests/test_domain_to_ip.py), and execute it with the following command:

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

```

The full test file looks like this:

```python

# 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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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.