Testing

It is highly recommended to write tests for your contribution. This ensures that your code works as expected and helps maintain the quality of the project.

Running Tests

To run the tests, you can use the following command:

uv sync --extra dev
uv run pytest

Using Shared Fixtures

Shared test fixtures are defined in tests/juice/conftest.py. Declare them as function parameters — pytest injects them automatically.

Mocking the Database

Two fixtures cover InfluxDB interactions:

• mock_write_api — a spec’d WriteApi mock with .write returning None

• mock_query_api — a spec’d QueryApi mock with .query returning None

Writing to the database:

from influxdb_client import WriteApi

def test_my_write(mock_write_api: WriteApi) -> None:
    my_function(write_api=mock_write_api)

    mock_write_api.write.assert_called_once()
    _, kwargs = mock_write_api.write.call_args
    assert kwargs["bucket"] == "expected_bucket"

Querying the database:

By default mock_query_api.query_data_frame returns None. Set a return value before calling your function when you need the query to return data:

import pandas as pd
from influxdb_client.client.query_api import QueryApi

def test_my_query(mock_query_api: QueryApi) -> None:
    mock_query_api.query_data_frame.return_value = pd.DataFrame({"value": [1.0, 2.0]})  # type: ignore

    result = my_function(query_api=mock_query_api)

    mock_query_api.query_data_frame.assert_called_once()  # type: ignore
    assert result is not None

Mocking the Publisher

mock_publisher_async provides a Mock with an AsyncMock .publish method, suitable for testing any code that publishes messages.

from unittest.mock import Mock

async def test_my_publisher(mock_publisher_async: Mock) -> None:
    await my_function(publisher=mock_publisher_async)

    mock_publisher_async.publish.assert_called_once()

To verify the published payload:

async def test_publish_payload(mock_publisher_async: Mock) -> None:
    await my_function(publisher=mock_publisher_async)

    _, kwargs = mock_publisher_async.publish.call_args
    assert kwargs["topic"] == "expected_topic"

Patching Module-Level Dependencies

When your code under test calls module-level factory functions (e.g. influxdb2_write_api()) rather than accepting the API as a parameter, use monkeypatch to replace them with the mock fixtures:

import pytest
from unittest.mock import Mock
from influxdb_client import WriteApi
from influxdb_client.client.query_api import QueryApi
import my_module

@pytest.fixture(autouse=True)
async def setup_mocks(
    mock_publisher_async: Mock,
    mock_write_api: WriteApi,
    mock_query_api: QueryApi,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(my_module, "influxdb2_query_api", lambda: mock_query_api)
    monkeypatch.setattr(my_module, "influxdb2_write_api", lambda: mock_write_api)
    monkeypatch.setattr(my_module, "publisher_async", lambda: mock_publisher_async)

With this autouse fixture in place, every test in the module automatically gets the mocks injected without repeating the patching.