Random & Synthetic Sources¶

RandomSource and SyntheticSource are built-in data sources for automated testing. They generate values programmatically inside your test — no terminals, no real-time waiting, and no data file required.

Use these when you want to feed your app a continuous stream of generated values and assert on the results with code.

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

Choose your source¶

RandomSource¶

RandomSource generates random values within bounds you set. Add a seed to make the test fully reproducible — the same seed always produces the same sequence.

from datetime import timedelta
from kelvin.testing.sources import RandomSource

source = (
    RandomSource(
        datastreams=["temperature", "pressure"],
        min_value=60.0,
        max_value=90.0,
        seed=42,     # same seed = same sequence every run
        count=100,   # number of values to generate per stream
    )
    .with_asset("doc_demo_plunger_01")
    .with_timing(timedelta(seconds=1))
)

SyntheticSource¶

SyntheticSource generates values that follow a predictable wave pattern. Use this when you want to test logic at specific points — for example, verifying an alert fires when a sine wave crosses your threshold.

from datetime import timedelta
from kelvin.testing.sources import SyntheticSource
from kelvin.testing.sources.synthetic import SineWave

source = (
    SyntheticSource(
        pattern=SineWave(amplitude=20.0, period=timedelta(seconds=60), offset=70.0),
        datastream="temperature",
        sample_rate=timedelta(seconds=1),
        # oscillates between 50 and 90, crossing 75 twice per period
    )
    .with_asset("doc_demo_plunger_01")
)

Available wave types:

Wave	Parameters	What it produces
`SineWave`	`amplitude`, `period`, `offset=0.0`, `phase=0.0`	Smooth oscillation. Value = offset + amplitude x sin(t). Good for temperature or pressure cycles
`SquareWave`	`amplitude`, `period`, `offset=0.0`, `duty_cycle=0.5`	Switches between two fixed values. Good for simulating binary on/off states
`RampWave`	`start`, `end`, `period`	Rises linearly from start to end then resets. Good for simulating gradual build-up
`NoiseWave`	`amplitude=1.0`, `offset=0.0`, `seed=None`	Random noise around a centre value. Good for adding realistic jitter to another wave
`ConstantWave`	`value`	A fixed value throughout. Good for testing exact threshold edge cases

Imports¶

import pytest

from main import app
from kelvin.testing import KelvinAppTest, ManifestBuilder
from kelvin.testing.sources import RandomSource, SyntheticSource
from kelvin.testing.sources.synthetic import SineWave, SquareWave, ConstantWave
from kelvin.message import RecommendationMsg

ManifestBuilder¶

from pathlib import Path

manifest = (
    ManifestBuilder()
    .add_input("temperature")                                     # must match an input name in app.yaml
    .add_output("alert")                                          # any output your app produces
    .add_asset("doc_demo_plunger_01", parameters={"limit": 75})   # asset with its parameters
    .add_asset("doc_demo_plunger_02", parameters={"limit": 75})
    .build()
)

Write the test¶

Wire everything together. The harness runs your app, delivers all generated messages, and collects everything your app publishes into harness.outputs.

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

    async with harness:
        await harness.run_until_idle(timeout=130.0)  # processes all messages using virtual time
        outputs = harness.outputs                    # list of every message your app published

    # filter by the message type you expect and assert on the result
    recommendations = [o for o in outputs if isinstance(o, RecommendationMsg)]
    assert len(recommendations) >= 1

harness.outputs is a plain Python list. Filter it by message type and assert on content however you need.

Testing timer-based logic¶

Many Applications use timers — logic that runs on a schedule rather than in response to an incoming message. In real life you would have to wait for the timer to fire. With KelvinAppTest you do not.

The virtual clock advances when you call run_until_idle. Set the timeout value to just past your timer interval and the harness advances time to that point and fires the timer instantly.

@pytest.mark.asyncio
async def test_scheduled_summary_fires():
    # App uses @app.timer(interval=300) — fires every 5 minutes
    harness = KelvinAppTest(app, manifest=manifest)

    async with harness:
        await harness.run_until_idle(timeout=301.0)  # jumps past the 300s timer instantly
        outputs = harness.outputs

    assert len(outputs) >= 1

You can also advance virtual time explicitly if you need precise control over when messages and timers fire relative to each other:

async with harness:
    await harness.advance_time(seconds=300)   # jump forward 300 virtual seconds
    await harness.run_until_idle()            # process anything that fired as a result
    outputs = harness.outputs

Run the tests¶

From your app folder:

pytest

pytest exits with code 0 if all assertions pass, or non-zero if any fail. CI pipelines use this exit code to gate deployments — if the test fails, the deployment does not proceed.

For the manual CLI equivalent, see Manual Testing with Simulator.