Publish Custom Actions¶

A Custom Action is a named command one Applications sends to another. The receiving app handles the command and responds with a result. For example, one app might decide that a pump should start and send a start-pump Custom Action to the app that controls it.

Testing custom actions locally lets you confirm that your app's handler fires correctly, produces the right result, and handles both success and failure cases.

To learn more about how Custom Actions work conceptually, see Custom Action.

Read Test Harness & Setup first if you have not already.

Setup: ManifestBuilder¶

Declare the custom action input and output channels. The input name must match the action type your handler listens for:

from kelvin.testing import ManifestBuilder

manifest = (
    ManifestBuilder()
    .add_custom_action_input("start-pump")         # the action type your handler listens for
    .add_custom_action_output("start-pump-result") # the result channel your handler publishes to
    .add_asset("doc_demo_plunger_01")
    .add_asset("doc_demo_plunger_02")
    .build()
)

The CustomAction message¶

A CustomAction carries the asset target, action type, a display title, and an expiration window:

from datetime import timedelta
from kelvin.message import CustomAction
from kelvin.krn import KRNAsset

CustomAction(
    resource=KRNAsset("doc_demo_plunger_01"),   # which asset this action targets
    type="start-pump",                          # must match the input name in your manifest
    title="Start Pump",                         # display title shown in the Kelvin UI
    expiration_date=timedelta(seconds=30),      # how long before this action expires
)

Full test example¶

import pytest
from datetime import timedelta
from main import app
from kelvin.testing import KelvinAppTest, ManifestBuilder
from kelvin.krn import KRNAsset
from kelvin.message import CustomAction, CustomActionResultMsg

manifest = (
    ManifestBuilder()
    .add_custom_action_input("start-pump")
    .add_custom_action_output("start-pump-result")
    .add_asset("doc_demo_plunger_01")
    .build()
)

@pytest.mark.asyncio
async def test_start_pump_succeeds():
    harness = KelvinAppTest(app, manifest=manifest)

    async with harness:
        await harness.publish(
            CustomAction(
                resource=KRNAsset("doc_demo_plunger_01"),
                type="start-pump",
                title="Start Pump",
                expiration_date=timedelta(seconds=30),
            )
        )
        await harness.run_until_idle()
        outputs = harness.outputs

    results = [o for o in outputs if isinstance(o, CustomActionResultMsg)]
    assert len(results) == 1
    assert results[0].payload.success is True

harness.outputs is a plain Python list. Filter it by CustomActionResultMsg and assert on the success field, payload content, or asset name.

Testing failure paths¶

Test what your handler does when it cannot complete the action. It should still produce a result with success set to False:

@pytest.mark.asyncio
async def test_start_pump_fails_when_not_ready():
    harness = KelvinAppTest(app, manifest=manifest)

    async with harness:
        await harness.publish(
            CustomAction(
                resource=KRNAsset("doc_demo_plunger_02"),
                type="start-pump",
                title="Start Pump",
                expiration_date=timedelta(seconds=30),
            )
        )
        await harness.run_until_idle()
        outputs = harness.outputs

    results = [o for o in outputs if isinstance(o, CustomActionResultMsg)]
    assert len(results) == 1
    assert results[0].payload.success is False

Testing both outcomes gives you confidence that your handler behaves correctly in all conditions.

Run the tests¶

pytest

pytest exits with code 0 if all assertions pass, or non-zero if any fail.

For the manual CLI equivalent using a generator script, see Manual Testing with Custom Generator.