Publish Data Streams¶

Data stream messages are the most common input your Application receives — numeric or string values tied to a specific asset and stream name. In a test, you can inject them directly using harness.publish() with a Number message.

Use this when you need to inject specific values at specific moments, test exact threshold crossings, or drive a custom sequence of inputs that a data source cannot produce.

What you need¶

Read Test Harness & Setup first if you have not already. You need pytest and pytest-asyncio installed and your app = KelvinApp() instance exposed in main.py.

Setup: ManifestBuilder¶

Declare the input and output streams your test will use:

from kelvin.testing import ManifestBuilder
from pathlib import Path

manifest = (
    ManifestBuilder()
    .add_input("temperature")                                     # input stream your app reads
    .add_output("alert")                                          # output your app produces
    .add_asset("doc_demo_plunger_01", parameters={"limit": 75})
    .add_asset("doc_demo_plunger_02", parameters={"limit": 75})
    .build()
)

Publishing a single message¶

Use harness.publish() with a Number message and a KRNAssetDataStream resource:

import pytest
from main import app
from kelvin.testing import KelvinAppTest
from kelvin.message import Number
from kelvin.krn import KRNAssetDataStream

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

    async with harness:
        await harness.publish(
            Number(
                resource=KRNAssetDataStream("doc_demo_plunger_01", "temperature"),
                payload=85.0,
            )
        )
        await harness.run_until_idle()
        outputs = harness.outputs

KRNAssetDataStream(asset_name, stream_name) is the resource identifier. It tells your app which asset and which data stream the value belongs to — the same way real platform data is addressed.

Publishing a sequence¶

Publish multiple messages in order to test how your app responds to a series of values:

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

    async with harness:
        for value in [60.0, 65.0, 70.0, 78.0, 85.0, 92.0]:
            await harness.publish(
                Number(
                    resource=KRNAssetDataStream("doc_demo_plunger_01", "temperature"),
                    payload=value,
                )
            )
        await harness.run_until_idle()
        outputs = harness.outputs

Each publish() call delivers one message. Your app processes them in the order you publish them.

Publishing multiple assets at once¶

Use publish_batch() to inject a list of messages simultaneously — useful when your app expects correlated values from multiple assets:

from kelvin.message import Number
from kelvin.krn import KRNAssetDataStream

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

    async with harness:
        await harness.publish_batch([
            Number(resource=KRNAssetDataStream("doc_demo_plunger_01", "temperature"), payload=85.0),
            Number(resource=KRNAssetDataStream("doc_demo_plunger_02", "temperature"), payload=88.0),
        ])
        await harness.run_until_idle()
        outputs = harness.outputs

Full test example¶

import pytest
from main import app
from kelvin.testing import KelvinAppTest, ManifestBuilder
from kelvin.message import Number, RecommendationMsg
from kelvin.krn import KRNAssetDataStream

manifest = (
    ManifestBuilder()
    .add_input("temperature")
    .add_asset("doc_demo_plunger_01", parameters={"limit": 75})
    .build()
)

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

    async with harness:
        await harness.publish(
            Number(
                resource=KRNAssetDataStream("doc_demo_plunger_01", "temperature"),
                payload=85.0,                      # above the limit of 75
            )
        )
        await harness.run_until_idle()
        outputs = harness.outputs

    recommendations = [o for o in outputs if isinstance(o, RecommendationMsg)]
    assert len(recommendations) == 1
    assert recommendations[0].resource.asset == "doc_demo_plunger_01"


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

    async with harness:
        await harness.publish(
            Number(
                resource=KRNAssetDataStream("doc_demo_plunger_01", "temperature"),
                payload=60.0,                      # below the limit of 75
            )
        )
        await harness.run_until_idle()
        outputs = harness.outputs

    recommendations = [o for o in outputs if isinstance(o, RecommendationMsg)]
    assert len(recommendations) == 0

Run the tests¶

pytest

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

For feeding a continuous stream of data rather than individual messages, see CSV Source or Random & Synthetic Sources.