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

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

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 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:

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 (lines 7–34) for examples of end-to-end DAG execution with input, map, branch, join, and reduce nodes.

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 (lines 15–44) demonstrates testing a NumberProducerOperator feeding a downstream TransformStreamAbsOperator. For HTTP-triggered workflows, 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 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 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 (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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}