Skip to main content

Testing plugins

Every production plugin goes through three levels of checks:

  1. Unit tests — isolated plugin logic, mocked resources, fast feedback.
  2. Contract tests — automatic verification of the eight runtime invariants via run_contract_suite (or its equivalent in the TS/Go binding).
  3. Integration tests — the plugin is exercised in a real host environment with real resources (a real HTTP client, a real clock, a real database).

This page covers how to run these three levels and how to wire them into CI. What exactly to inject into the context (frozen clock, in-memory blob store, and so on) is documented on the Resources page.

Directory layout​

The recommended layout for a plugin's local tests:

plugins/
└── my-plugin/
    ├── dagstack.toml
    ├── plugin.py                 # implementation
    ├── tests/
    │   ├── test_plugin.py        # unit tests
    │   ├── test_contract.py      # contract tests
    │   └── test_integration.py   # integration tests (optional)
    └── README.md

A tests/ folder inside the plugin lets you run checks in isolation:

cd plugins/my-plugin && pytest tests/

Unit tests​

Written as ordinary unit tests for the language. The plugin receives a test PluginContext with substituted resources.

plugins/fixed-chunker/tests/test_plugin.py
import asyncio
import logging
from datetime import datetime, UTC

import pytest
from dagstack.plugin_system import PluginContext, PluginRegistry
from dagstack.plugin_system import FrozenClock, ResourceRegistry
from plugins.fixed_chunker.plugin import FixedChunker


@pytest.fixture
def plugin():
    resources = ResourceRegistry()
    resources.register("clock", FrozenClock(at=datetime(2026, 1, 1, tzinfo=UTC)))
    registry = PluginRegistry(resource_registry=resources)
    ctx = PluginContext(
        config={},
        logger=logging.getLogger("test"),
        registry=registry,
    )
    plugin = FixedChunker()
    asyncio.run(plugin.setup(ctx))
    return plugin


def test_chunks_empty_text(plugin):
    assert plugin.chunk("", max_size=10) == []


def test_chunks_exact_size(plugin):
    result = plugin.chunk("abcde", max_size=5)
    assert len(result) == 1
    assert result[0].text == "abcde"
    assert result[0].offset == 0


def test_chunks_splits_by_size(plugin):
    result = plugin.chunk("abcdefgh", max_size=3)
    assert [c.text for c in result] == ["abc", "def", "gh"]
    assert [c.offset for c in result] == [0, 3, 6]


def test_rejects_invalid_size(plugin):
    with pytest.raises(ValueError, match="positive"):
        plugin.chunk("any", max_size=0)
cd plugins/fixed-chunker && pytest tests/test_plugin.py -v

Contract tests​

The contract harness automatically checks all eight runtime invariants through run_contract_suite. Each check is an assert_* function, and the aggregated result produces a readable report.

plugins/fixed-chunker/tests/test_contract.py
from dagstack.plugin_system import (
    load_manifest,
    run_contract_suite,
    ALL_CHECKS,
)
from plugins.fixed_chunker.plugin import FixedChunker


def test_contract_passes():
    manifest = load_manifest("plugins/fixed-chunker/dagstack.toml")
    result = run_contract_suite(
        plugin_class=FixedChunker,
        manifest=manifest,
        checks=ALL_CHECKS,
    )
    assert result.ok, result.format_failures()

Run:

pytest plugins/fixed-chunker/tests/test_contract.py -v

Debugging failed contract checks​

The contract suite returns a structured report identifying the violated invariant and the exact location. Typical issues:

CheckMessageWhat to verify
assert_no_ambient_stateAmbientStateViolation: module-level mutable stateRemove globals and module-level dict/list; move them onto the plugin instance or into Resources.
assert_json_serializable_boundariesSerializationError: <class 'httpx.AsyncClient'> is not JSON-serializableA live client leaked into a hook output; return data, not resources.
assert_json_serializable_boundariesSerializationError: datetime not JSON-serializableConvert datetime to an ISO 8601 string before returning.
assert_lifecycle_cleanLeakDetected: tmpdir contains N files after teardownAdd cleanup of temporary files to teardown().
assert_deterministicDeterminismError: different output on second run with same seedUse ctx.clock.now() and ctx.rng instead of time.time() / random().
assert_manifest_validManifestInvalid: required field missingCheck the manifest fields against the Plugin manifest page (the full schema lives in the spec repository).

Integration tests​

Integration tests run the plugin in a real host environment: real registry, real resources (or their staging variants), real configuration.

tests/test_integration.py
import asyncio
import logging

import pytest
from dagstack.plugin_system import PluginContext, PluginRegistry


@pytest.fixture
def real_registry():
    registry = PluginRegistry()
    registry.discover("plugins/")
    ctx = PluginContext(
        config={},
        logger=logging.getLogger("test"),
        registry=registry,
    )
    asyncio.run(registry.setup_all(ctx))
    yield registry
    asyncio.run(registry.teardown_all())


def test_pipeline_end_to_end(real_registry):
    chunker = real_registry.get_plugin("chunker", name="fixed")
    result = chunker.chunk("Hello, dagstack!", max_size=5)
    # The plugin interacts with the real Clock/Rng — assert the structure,
    # not the exact values.
    assert len(result) == 4
    assert all(c.offset % 5 == 0 for c in result[:-1])

CI integration​

Every PR runs three groups of checks:

.gitea/workflows/plugin-tests.yml
name: Plugin tests

on:
  push:
  pull_request:

jobs:
  test:
    runs-on: dagstack-runner
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install deps
        run: pip install -e '.[test]'

      - name: Unit tests
        run: pytest plugins/*/tests/test_plugin.py -v

      - name: Contract tests
        run: pytest plugins/*/tests/test_contract.py -v

      - name: Integration tests
        run: pytest tests/test_integration.py -v
        env:
          DAGSTACK_TEST_MODE: "1"

Contract tests are mandatory. Unit and integration tests are recommended; their priority depends on how critical the plugin is.

Parameterised contract tests​

If you have many plugins of the same kind, parameterise a single shared contract test:

import pytest
from pathlib import Path
from dagstack.plugin_system import load_manifest, run_contract_suite, ALL_CHECKS


ALL_PLUGINS = list(Path("plugins").glob("*/dagstack.toml"))


@pytest.mark.parametrize("manifest_path", ALL_PLUGINS, ids=lambda p: p.parent.name)
def test_every_plugin_passes_contract(manifest_path):
    manifest = load_manifest(manifest_path)
    plugin_class = _load_plugin_class(manifest_path.parent, manifest.entry_point)
    result = run_contract_suite(
        plugin_class=plugin_class,
        manifest=manifest,
        checks=ALL_CHECKS,
    )
    assert result.ok, result.format_failures()

A new plugin added to plugins/ is picked up automatically by the parameterised run — no extra test code required.

See also​