CSV Source¶

CSVSource reads a CSV file and replays each row as a data stream message inside your automated test. Your app receives the data exactly as it would from a live data source, but the whole replay happens through a virtual clock with no real-time waiting.

Use this when you have recorded data from your process and want to write repeatable assertions against it.

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

Your CSV file¶

Structure your CSV with these columns:

timestamp — event time for each row
asset_name — which asset the row belongs to
One column per input data stream defined in your app.yaml

timestamp,asset_name,temperature
2023-10-27T10:00:01,doc_demo_plunger_01,50
2023-10-27T10:00:02,doc_demo_plunger_01,51
2023-10-27T10:00:03,doc_demo_plunger_01,52
2023-10-27T10:00:04,doc_demo_plunger_01,55
2023-10-27T10:00:05,doc_demo_plunger_01,60
2023-10-27T10:00:06,doc_demo_plunger_02,56
2023-10-27T10:00:07,doc_demo_plunger_02,58
2023-10-27T10:00:08,doc_demo_plunger_02,60

Basic usage¶

from kelvin.testing.sources import CSVSource

source = CSVSource("data/test-data.csv")

Pass the source to the harness before running the test:

harness = KelvinAppTest(app, manifest=manifest).add_source(source)

Constructor parameters¶

Parameter	Default	Description
`path`	Required	Path to your CSV file.
`playback`	`False`	When `True`, waits between rows based on the actual timestamp difference in the CSV, replaying at real-time speed.
`ignore_timestamps`	`False`	When `True`, ignores CSV timestamps and uses virtual clock time instead.
`now_offset`	`False`	When `True`, shifts all timestamps so the first row starts at the current time.
`asset_column`	`"asset_name"`	Name of the column that contains asset names.
`timestamp_column`	`"timestamp"`	Name of the column that contains timestamps.

Example using constructor parameters:

source = CSVSource(
    path="data/test-data.csv",
    ignore_timestamps=True,       # use virtual clock time instead of CSV timestamps
    asset_column="device_id",     # if your CSV uses a different column name for assets
)

Fluent methods¶

Chain these after CSVSource(...) to configure the source:

Method	Description
`.with_columns("col1", "col2")`	Select specific columns to publish. All other columns are ignored.
`.with_asset("asset_name")`	Fallback asset name when the CSV has no `asset_name` column.
`.with_timing(timedelta(seconds=1))`	Override the interval between row messages.
`.with_column_mapping({"csv_col": "stream_name"})`	Map CSV column names to the data stream names in your app.

Example:

from datetime import timedelta

source = (
    CSVSource("data/test-data.csv")
    .with_columns("temperature")                          # publish only this column
    .with_column_mapping({"temp_c": "temperature"})       # rename if column name differs
    .with_asset("doc_demo_plunger_01")                    # fallback if no asset_name column
)

Full test example¶

import pytest
from main import app
from kelvin.testing import KelvinAppTest, ManifestBuilder
from kelvin.testing.sources import CSVSource
from kelvin.message import RecommendationMsg

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

source = (
    CSVSource("data/test-data.csv")
    .with_columns("temperature")
)

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

    async with harness:
        await harness.run_until_idle(timeout=15.0)
        outputs = harness.outputs

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

Run the tests¶

pytest

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

For the manual CLI equivalent, see Manual Testing with CSV File.