# Testing Strategies for DB-GPT Custom Plugins and Operators: A Complete Guide

> Learn DB-GPT testing strategies for custom plugins and operators. Discover unit, integration, and end-to-end tests to ensure your extensions function perfectly. A complete guide for eosphoros-ai/DB-GPT.

- Repository: [eosphoros/DB-GPT](https://github.com/eosphoros-ai/db-gpt)
- Tags: how-to-guide
- Published: 2026-02-23

---

**DB-GPT recommends a three-layer testing pyramid—unit tests for plugin discovery and operator logic, integration tests for DAG wiring, and end-to-end smoke tests—to ensure custom extensions work correctly in isolation and within live workflows.**

DB-GPT (Database-GPT) from `eosphoros-ai/DB-GPT` provides a layered architecture that separates **plugins** (external tools that extend the platform) from **AWEL operators** (building-block tasks used in workflow DAGs). Because both extensions introduce new code paths, the project ships a comprehensive test suite covering everything from individual functions to full-stack integration. This guide details the recommended testing strategies for DB-GPT custom plugins and operators based on the actual source code implementation.

## Unit Tests for Plugin Discovery and Validation

DB-GPT plugins are discovered via the `scan_plugins` function in [`dbgpt/plugins/__init__.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/dbgpt/plugins/__init__.py). The framework validates each plugin against deny-lists and allow-lists before loading, making it essential to test discovery logic, zip inspection, and configuration filtering.

### Testing Plugin Scanning and Loading

The `scan_plugins` function accepts a configuration object containing `plugins_dir`, `plugins_denylist`, `plugins_allowlist`, and `plugins_openai` parameters. Unit tests should verify that a plugin can be discovered, validated, and loaded without side effects.

Key functions to test include:
- `scan_plugins` – Main entry point for plugin discovery
- `inspect_zip_for_modules` – Inspects zipped packages for Python modules
- `denylist_allowlist_check` – Filters plugins based on configuration lists

The official unit tests reside in [`tests/unit_tests/test_plugins.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/tests/unit_tests/test_plugins.py). For example, `test_scan_plugins_generic` (lines 124–131) validates generic plugin scanning, while `test_denylist_allowlist_check_denylist` (lines 41–45) ensures rejected plugins are properly filtered.

```python
import pytest
from dbgpt.plugins import scan_plugins, PluginInfo

@pytest.fixture
def dummy_cfg(tmp_path):
    class Cfg:
        plugins_dir = str(tmp_path / "plugins")
        plugins_denylist = []
        plugins_allowlist = ["MyPlugin"]
        plugins_openai = []
    return Cfg()

def test_my_plugin_is_discovered(dummy_cfg, tmp_path):
    # Create plugin directory structure

    plug_dir = tmp_path / "plugins"
    plug_dir.mkdir()
    
    # Scan and validate

    plugins = scan_plugins(dummy_cfg, debug=True)
    
    # Assert plugin metadata

    assert any(isinstance(p, PluginInfo) and p.name == "MyPlugin" for p in plugins)

```

Run these tests with `pytest tests/unit_tests/test_plugins.py`. They execute in milliseconds and require no external services, making them ideal for CI pipelines.

## Unit Tests for AWEL Operators

AWEL (Agentic Workflow Expression Language) operators form the building blocks of DB-GPT workflows. Each operator implements transformation logic in its `run` or `stream` methods, which must behave deterministically.

### Testing Operator Transformations

Unit tests for operators should instantiate the class, feed it known inputs, and await results for validation. The test suite in [`packages/dbgpt-core/src/dbgpt/core/interface/operators/tests/test_message_operator.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/interface/operators/tests/test_message_operator.py) demonstrates this pattern. For instance, `test_buffered_conversation_mapper` (lines 6–15) validates that a message mapper correctly transforms a list of `Message` objects into string representations.

For synchronous operators inheriting from `MapOperator`, testing is straightforward:

```python
import pytest
from dbgpt.core.awel.operators.common_operator import MapOperator

class AddOneOperator(MapOperator[int, int]):
    def __call__(self, num: int) -> int:
        return num + 1

def test_add_one_operator():
    op = AddOneOperator()
    assert op.run(5) == 6
    assert op.run(-1) == 0

```

For async operators, use **pytest-asyncio** to test asynchronous `run` or `stream` methods without starting a full DB-GPT server.

## Integration Tests for Operator DAGs

Individual operators must work correctly when wired into directed acyclic graphs (DAGs). Integration tests verify data flows between nodes, branching logic, joining operations, and streaming pipelines.

### Wiring Operators with WorkflowRunner

The AWEL testing framework provides `WorkflowRunner` to execute DAGs in-memory. Tests should construct a `DAG` object, connect operators using the `>>` syntax, and run the workflow against expected outputs.

Reference [`packages/dbgpt-core/src/dbgpt/core/awel/tests/test_run_dag.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/tests/test_run_dag.py) (lines 7–34) for examples of end-to-end DAG execution with input, map, branch, join, and reduce nodes.

```python
import pytest
from dbgpt.core.awel import DAG, InputOperator, MapOperator, WorkflowRunner

class SquareOperator(MapOperator[int, int]):
    def __call__(self, x: int) -> int:
        return x * x

@pytest.mark.asyncio
async def test_square_dag():
    # Build the workflow

    dag = DAG("square_test")
    src = InputOperator([1, 2, 3, 4], task_id="src")
    square = SquareOperator()
    src >> square  # Wire nodes

    # Execute and verify

    runner = WorkflowRunner(dag)
    result = await runner.run(src.task_id)
    assert result == [1, 4, 9, 16]

```

### Testing Streaming and HTTP Triggers

Complex scenarios require testing streaming sources and HTTP endpoints. The file [`packages/dbgpt-core/src/dbgpt/core/awel/tests/test_iterator_trigger.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/tests/test_iterator_trigger.py) (lines 15–44) demonstrates testing a `NumberProducerOperator` feeding a downstream `TransformStreamAbsOperator`. For HTTP-triggered workflows, [`test_http_operator.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/test_http_operator.py) (lines 5–18) validates that `HttpTrigger` operators correctly start FastAPI endpoints and process POST data.

Execute integration tests via `pytest -m awel` in the CI matrix. These tests use asyncio fixtures and run against in-memory runners without external dependencies.

## End-to-End Smoke Tests for Plugins

While unit and integration tests provide fast feedback, end-to-end (E2E) tests verify that packaged plugins install and execute within a running DB-GPT server. The CI pipeline spins up a Docker container running `dbgpt_server` with the `--plugins-dir` flag pointing to a temporary directory.

A test script uses the high-level SDK (`dbgpt.plugins.load`) to load the plugin and invoke its `run` method, asserting the output matches expectations. This flow is exercised in `docker/examples/plugin_test` (referenced in documentation) and provides a final safety net before releases. These tests run slower (seconds to minutes) but catch packaging and runtime configuration errors that unit tests miss.

## Summary

- **Unit test plugins** using `scan_plugins` and `denylist_allowlist_check` to verify discovery and validation logic without external services.
- **Unit test operators** by instantiating classes and asserting transformation logic on known inputs, using pytest-asyncio for async operators.
- **Integration test DAGs** with `WorkflowRunner` to ensure operators wire correctly and data flows through branches, joins, and streams.
- **Run E2E smoke tests** in Docker to confirm plugins load and execute in a live DB-GPT server environment.
- **Maintain >80% code coverage** using `pytest --cov=dbgpt` to ensure new extensions are fully exercised.

## Frequently Asked Questions

### How do I test plugin discovery locally without a full DB-GPT server?

Use the `scan_plugins` function from [`dbgpt/plugins/__init__.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/dbgpt/plugins/__init__.py) with a mocked configuration object. Create a temporary directory containing your zipped plugin, set `plugins_dir` to this path, and call `scan_plugins(cfg, debug=True)`. Assert that the returned list contains a `PluginInfo` object with your plugin's metadata. This approach requires only pytest and runs in milliseconds.

### What is the recommended test coverage threshold for DB-GPT extensions?

The DB-GPT CI configuration in [`.github/workflows/ci.yml`](https://github.com/eosphoros-ai/DB-GPT/blob/main/.github/workflows/ci.yml) enforces coverage checks using `pytest --cov=dbgpt`. While the exact threshold varies by module, aim for **greater than 80% coverage** for new plugin and operator code. Focus on covering validation logic, error paths, and the main transformation methods to ensure reliability.

### How can I test async AWEL operators?

Import **pytest-asyncio** and decorate your test functions with `@pytest.mark.asyncio`. Instantiate your operator class and await the `run` or `stream` method directly. The test suite in `packages/dbgpt-core/src/dbgpt/core/interface/operators/tests/` provides examples of testing async operators without requiring a running server, using pure Python data models like `Message`.

### Where can I find examples of integration tests for streaming operators?

Reference [`packages/dbgpt-core/src/dbgpt/core/awel/tests/test_iterator_trigger.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/tests/test_iterator_trigger.py) (lines 15–44), which demonstrates a `NumberProducerOperator` feeding a `TransformStreamAbsOperator`. This file shows how to test streaming data flows using the AWEL testing utilities and asyncio fixtures, verifying that operators handle asynchronous iterators correctly.