ReferencesPython SDKGuides

Testing

Unit testing

If you'd like to unit test without an Inngest server, the mocked (requires v0.4.14+) library can simulate much of the Inngest server's behavior.

The mocked library is experimental. It may have interface and behavioral changes that don't follow semantic versioning.

Let's say you've defined this function somewhere in your app:

import inngest

def create_message(name: object) -> str:
    return f"Hello, {name}!"

client = inngest.Inngest(app_id="my-app")

@client.create_function(
    fn_id="greet",
    trigger=inngest.TriggerEvent(event="user.login"),
)
async def greet(
    ctx: inngest.Context,
    step: inngest.Step,
) -> str:
    message = await step.run(
        "create-message",
        create_message,
        ctx.event.data["name"],
    )

    return message

You can unit test it like this:

import unittest
import inngest
from inngest.experimental import mocked
from .functions import greet

# Mocked Inngest client. The app_id can be any string (it's currently unused)
client_mock = mocked.Inngest(app_id="test")

# A normal Python test class
class TestGreet(unittest.TestCase):
    def test_greet(self) -> None:
        # Trigger the function with an in-memory, simulated Inngest server
        res = mocked.trigger(
            greet,
            inngest.Event(name="user.login", data={"name": "Alice"}),
            client_mock,
        )

        # Assert that it ran as expected
        assert res.status is mocked.Status.COMPLETED
        assert res.output == "Hello, Alice!"

Limitations

The mocked library has some notable limitations:

  • step.invoke and step.wait_for_event must be stubbed using the step_stubs parameter of mocked.trigger.
  • step.send_event does not send events. It returns a stubbed value.
  • step.sleep and step.sleep_until always sleep for 0 seconds.

Stubbing

Stubbing is required for step.invoke and step.wait_for_event. Here's an example of how to stub these functions:

# Real production function
@client.create_function(
    fn_id="signup",
    trigger=inngest.TriggerEvent(event="user.signup"),
)
def signup(
    ctx: inngest.Context,
    step: inngest.StepSync,
) -> bool:
    email_id = step.invoke(
        "send-email",
        function=send_email,
    )

    event = step.wait_for_event(
        "wait-for-reply",
        event="email.reply",
        if_exp=f"async.data.email_id == '{email_id}'",
        timeout=datetime.timedelta(days=1),
    )
    user_replied = event is not None
    return user_replied

# Mocked Inngest client
client_mock = mocked.Inngest(app_id="test")

class TestSignup(unittest.TestCase):
    def test_signup(self) -> None:
        res = mocked.trigger(
            fn,
            inngest.Event(name="test"),
            client_mock,

            # Stub the invoke and wait_for_event steps. The keys are the step
            # IDs
            step_stubs={
                "send-email": "email-id-abc123",
                "wait-for-reply": inngest.Event(
                    data={"text": "Sounds good!"}, name="email.reply"
                ),
            },
        )
        assert res.status is mocked.Status.COMPLETED
        assert res.output is True

To simulate a step.wait_for_event timeout, stub the step with mocked.Timeout.

Integration testing

If you'd like to start and stop a real Dev Server with your integration tests, the dev_server (requires v0.4.15+) library can help. It requires npm to be installed on your machine.

The dev_server library is experimental. It may have interface and behavioral changes that don't follow semantic versioning.

You can use the library in your conftest.py:

import pytest
from inngest.experimental import dev_server

def pytest_configure(config: pytest.Config) -> None:
    dev_server.server.start()

def pytest_unconfigure(config: pytest.Config) -> None:
    dev_server.server.stop()

This Dev Server will not automatically discover your app. You'll need to manually sync by sending a PUT request to your app's Inngest endpoint (/api/inngest by default).

Since Pytest automatically discovers and runs conftest.py files, simply running your pytest command will start the Dev Server before running tests and stop the Dev Server after running tests.