Перейти к основному содержимому

Тестирование плагинов

Каждый плагин в production проходит три уровня проверок:

  1. Unit-тесты — изолированная логика плагина, mock-ресурсы, быстрый фидбэк.
  2. Контрактные тесты — автоматическая проверка соблюдения восьми runtime-инвариантов через run_contract_suite (или эквивалент в TS/Go реализации).
  3. Интеграционные тесты — плагин запускается в реальном host-окружении с реальными ресурсами (настоящий HTTP-клиент, настоящий clock, настоящая БД).

Эта страница — как эти три уровня запускать и встраивать в CI. Что именно инъектировать в контекст (frozen-clock, in-memory blob-store и т.д.) — на странице Ресурсы.

Структура директорий​

Рекомендуемый layout для локальных тестов плагина:

plugins/
└── my-plugin/
    ├── dagstack.toml
    ├── plugin.py                 # реализация
    ├── tests/
    │   ├── test_plugin.py        # unit-тесты
    │   ├── test_contract.py      # контрактные тесты
    │   └── test_integration.py   # интеграционные (опционально)
    └── README.md

tests/ внутри плагина позволяет запускать проверки изолированно:

cd plugins/my-plugin && pytest tests/

Unit-тесты​

Пишутся как обычные unit-тесты языка, плагин получает тестовый PluginContext с подменёнными ресурсами.

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(now=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

Контрактные тесты​

Контрактная рамка автоматически проверяет все восемь runtime-инвариантов через run_contract_suite. Каждая проверка — assert_*-функция, агрегированный результат даёт понятный отчёт.

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

Отладка проваленных контрактных проверок​

Контракт-suite возвращает структурированный отчёт с указанием нарушенного инварианта и конкретного места. Типичные проблемы:

ПроверкаСообщениеЧто проверить
assert_no_ambient_stateAmbientStateViolation: module-level mutable stateУбрать global-переменные, module-level dict/list; перенести в экземпляр плагина или в Resources.
assert_json_serializable_boundariesSerializationError: <class 'httpx.AsyncClient'> is not JSON-serializableЖивой клиент попал в output хука; возвращать данные, не ресурсы.
assert_json_serializable_boundariesSerializationError: datetime not JSON-serializableКонвертируйте datetime в ISO 8601 строку перед возвратом.
assert_lifecycle_cleanLeakDetected: tmpdir contains N files after teardownВ teardown() добавить очистку временных файлов.
assert_deterministicDeterminismError: different output on second run with same seedИспользовать ctx.clock.now() и ctx.rng вместо time.time() / random().
assert_manifest_validManifestInvalid: required field missingПроверить поля манифеста против страницы «Манифест плагина» (полная схема в spec-репозитории).

Интеграционные тесты​

Запускают плагин в настоящем host-окружении: реальный реестр, реальные ресурсы (или их staging-версии), реальный конфиг.

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-интеграция​

Каждый PR запускает три группы проверок:

.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"

Контрактные тесты — обязательные. Unit и интеграционные — recommended, приоритеты зависят от важности плагина.

Параметризованные контрактные тесты​

Если у вас много плагинов одного вида, параметризуйте один общий контрактный тест:

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()

Новый плагин, добавленный в plugins/, автоматически попадает в параметризованный запуск — отдельный тестовый код не нужен.

См. также​