Files
Typer_Body 77a0de5ef0 Feat: bot message routing (#2100)
* refactor: pipeline routing rules - add routed_by_rule bypass and diagnostic logging

- Add routing rules editor (RoutingRulesEditor component)
- Add routed_by_rule bypass logic in response rules
- Add diagnostic logging for pipeline routing
- Database migration for bot pipeline routing rules
- Extract RoutingRulesEditor component from BotForm
- Revert log levels to debug

* feat: add message_has_element routing rule type

Support routing by message element type (Image, Voice, File, Forward,
Face, At, AtAll, Quote) with eq/neq operators.

* test: add unit tests for pipeline routing rules

20 tests covering _match_operator (eq/neq/contains/not_contains/
starts_with/regex/invalid) and resolve_pipeline_uuid (launcher_type/
launcher_id/message_content/message_has_element/first-match-wins/
skip-invalid/default-operator).

* fix(web): add missing 'message_has_element' to routing rule type validation

The Zod schema and TypeScript type for PipelineRoutingRule.type were
missing the 'message_has_element' variant, causing silent form validation
failure when saving routing rules with this type.

* feat: add pipeline discard functionality and localization support

* feat(web): improve drag-and-drop with DragOverlay, add discard monitoring and pipeline icons

- Add DragOverlay for smooth cursor-following drag in routing rules editor
- Remove transition to eliminate redundant swap animation on drop
- Record discarded messages in monitoring system via _record_discarded_message
- Display pipeline name (Workflow icon) and runner name (Play icon) on session monitor messages
- Show discard badge on discarded messages in session monitor
- Add i18n translations for discarded/userMessage/botMessage

* fix: ensure discarded messages appear in session monitor and improve icons

- Create/update monitoring session for discarded messages so they show in
  the bot session monitor (was only inserting message rows, not sessions)
- Use human-readable 'Discarded' as pipeline_name instead of '__discard__'
- Change runner icon from Play to Bot for better AI Agent semantics

* fix: merge discarded messages into same session and remove session-level pipeline name

- Use LauncherTypes enum for session_id in discarded messages to match
  the format used by monitoring_helper (fixes duplicate sessions)
- Don't overwrite session pipeline info on discard — a session can have
  messages from multiple pipelines
- Remove pipeline_name from session list and chat header since it's
  now shown per-message and a session is no longer single-pipeline

* fix(web): only show save button on config tab in bot detail page

* fix(web): scroll to bottom after messages render in session monitor

---------

Co-authored-by: RockChinQ <rockchinq@gmail.com>
2026-04-03 23:56:58 +08:00
..

LangBot Test Suite

This directory contains the test suite for LangBot, with a focus on comprehensive unit testing of pipeline stages.

Important Note

Due to circular import dependencies in the pipeline module structure, the test files use lazy imports via importlib.import_module() instead of direct imports. This ensures tests can run without triggering circular import errors.

Structure

tests/
├── pipeline/                      # Pipeline stage tests
│   ├── conftest.py               # Shared fixtures and test infrastructure
│   ├── test_simple.py            # Basic infrastructure tests (always pass)
│   ├── test_bansess.py           # BanSessionCheckStage tests
│   ├── test_ratelimit.py         # RateLimit stage tests
│   ├── test_preproc.py           # PreProcessor stage tests
│   ├── test_respback.py          # SendResponseBackStage tests
│   ├── test_resprule.py          # GroupRespondRuleCheckStage tests
│   ├── test_pipelinemgr.py       # PipelineManager tests
│   └── test_stages_integration.py # Integration tests
└── README.md                      # This file

Test Architecture

Fixtures (conftest.py)

The test suite uses a centralized fixture system that provides:

  • MockApplication: Comprehensive mock of the Application object with all dependencies
  • Mock objects: Pre-configured mocks for Session, Conversation, Model, Adapter
  • Sample data: Ready-to-use Query objects, message chains, and configurations
  • Helper functions: Utilities for creating results and common assertions

Design Principles

  1. Isolation: Each test is independent and doesn't rely on external systems
  2. Mocking: All external dependencies are mocked to ensure fast, reliable tests
  3. Coverage: Tests cover happy paths, edge cases, and error conditions
  4. Extensibility: Easy to add new tests by reusing existing fixtures

Running Tests

bash run_tests.sh

This script automatically:

  • Activates the virtual environment
  • Installs test dependencies if needed
  • Runs tests with coverage
  • Generates HTML coverage report

Manual test execution

Run all tests

pytest tests/pipeline/

Run only simple tests (no imports, always pass)

pytest tests/pipeline/test_simple.py -v

Run specific test file

pytest tests/pipeline/test_bansess.py -v

Run with coverage

pytest tests/pipeline/ --cov=pkg/pipeline --cov-report=html

Run specific test

pytest tests/pipeline/test_bansess.py::test_bansess_whitelist_allow -v

Known Issues

Some tests may encounter circular import errors. This is a known issue with the current module structure. The test infrastructure is designed to work around this using lazy imports, but if you encounter issues:

  1. Make sure you're running from the project root directory
  2. Ensure the virtual environment is activated
  3. Try running test_simple.py first to verify the test infrastructure works

CI/CD Integration

Tests are automatically run on:

  • Pull request opened
  • Pull request marked ready for review
  • Push to PR branch
  • Push to master/develop branches

The workflow runs tests on Python 3.10, 3.11, and 3.12 to ensure compatibility.

Adding New Tests

1. For a new pipeline stage

Create a new test file test_<stage_name>.py:

"""
<StageName> stage unit tests
"""

import pytest
from pkg.pipeline.<module>.<stage> import <StageClass>
from pkg.pipeline import entities as pipeline_entities


@pytest.mark.asyncio
async def test_stage_basic_flow(mock_app, sample_query):
    """Test basic flow"""
    stage = <StageClass>(mock_app)
    await stage.initialize({})

    result = await stage.process(sample_query, '<StageName>')

    assert result.result_type == pipeline_entities.ResultType.CONTINUE

2. For additional fixtures

Add new fixtures to conftest.py:

@pytest.fixture
def my_custom_fixture():
    """Description of fixture"""
    return create_test_data()

3. For test data

Use the helper functions in conftest.py:

from tests.pipeline.conftest import create_stage_result, assert_result_continue

result = create_stage_result(
    result_type=pipeline_entities.ResultType.CONTINUE,
    query=sample_query
)

assert_result_continue(result)

Best Practices

  1. Test naming: Use descriptive names that explain what's being tested
  2. Arrange-Act-Assert: Structure tests clearly with setup, execution, and verification
  3. One assertion per test: Focus each test on a single behavior
  4. Mock appropriately: Mock external dependencies, not the code under test
  5. Use fixtures: Reuse common test data through fixtures
  6. Document tests: Add docstrings explaining what each test validates

Troubleshooting

Import errors

Make sure you've installed the package in development mode:

uv pip install -e .

Async test failures

Ensure you're using @pytest.mark.asyncio decorator for async tests.

Mock not working

Check that you're mocking at the right level and using AsyncMock for async functions.

Future Enhancements

  • Add integration tests for full pipeline execution
  • Add performance benchmarks
  • Add mutation testing for better coverage quality
  • Add property-based testing with Hypothesis