* feat(vector): add Valkey Search vector database backend Add a new opt-in VectorDatabase backend backed by the Valkey Search module (valkey/valkey-bundle), accessed via the official valkey-glide client's native ft command namespace. - Implements the full VectorDatabase ABC: VECTOR, FULL_TEXT and HYBRID search, all 8 metadata filter operators, and pagination with exact totals. - HYBRID uses filter-then-KNN (no app-side weighted fusion); vector_weight is accepted for interface parity but NOT honored (docstring + one-time warning + docs caveat). - Lazy connect so a down Valkey never blocks boot; mandatory client_name=langbot_vector_client; optional auth + TLS (never logged). - Registered via a single elif branch in vector/mgr.py; disabled by default (vdb.use stays chroma) for toC compatibility. - Adds valkey-glide>=2.4.1,<3.0.0; no protobuf/pydantic downgrade; no ORM change so no Alembic migration. - Unit tests (fast lane, no server) + slow-gated integration tests (TEST_VALKEY_URL, valkey/valkey-bundle:9.1.0) + integration doc. * fix(vector): paginate Valkey Search deletes and guard delete_by_filter Address self-review follow-ups for the Valkey Search VDB backend: - _search_keys now paginates through the full result set in batches of _DELETE_SCAN_BATCH instead of capping at a single hard-coded 10000-key page, so delete_by_file_id / delete_by_filter fully remove files and filters that match more than one page of chunks (no orphaned vectors). - Add unit regression tests for the delete_by_filter mass-deletion guard: a filter referencing only non-indexed fields must skip and return 0 (never fall back to match-all), and a supported filter still deletes matching keys. * refactor(vector): harden Valkey Search backend and add adversarial tests Address the self-review NICE-TO-HAVE items for the Valkey Search VDB backend: - Guard the username-without-password credential edge (skip auth + warn instead of building ServerCredentials(password=None, ...), which glide rejects). - Add an async close() teardown that closes the glide client and resets cached state (re-init is safe via the existing None guard). - Hoist 'import json' to module top (was imported inside three methods). - Document the FT TAG literal-brace limitation in _escape_tag (fails closed, never widens). Tests: - Add an adversarial-input integration test proving crafted file_id / query_text cannot break out of or widen a query (fail-closed on braces). - Add unit tests for close() and the credential-build guard. Signed-off-by: Daria Korenieva <daric2612@gmail.com> * fix(vector): make Valkey Search file_id TAG support arbitrary characters Valkey Search's FT TAG query parser cannot handle '{', '}' or '*' even when backslash-escaped, so a file_id containing those characters previously produced an unparseable query (it failed closed / raised). Percent-encode exactly those FT-unsafe characters (plus '%' for reversibility) in the file_id TAG value, applied identically at write time and query time, so an arbitrary file_id round-trips. For normal UUID/hash ids this is a no-op and the stored value is unchanged; the original file_id is always preserved verbatim in metadata_json. Strengthen the adversarial integration test to assert a brace/star-bearing file_id matches and deletes exactly its own row (no widening, no raise), and add unit tests for _encode_file_id and the filter encoding. Signed-off-by: Daria Korenieva <daric2612@gmail.com> * refactor(vector): address Valkey Search review feedback - Add configurable request_timeout (default 5000ms; glide default 250ms is too low for KNN); expose in config.yaml + docs table - Validate embedding dimension consistency in add_embeddings (fail fast on mixed lengths to avoid silent KNN corruption) - Use ft.info (O(1)) instead of ft.list (O(n)) for index existence checks in the query hot path; also closes the check-then-create TOCTOU window - Pipeline HSETs via a non-atomic Batch instead of N sequential awaits - Extract shared _iter_reply_docs to deduplicate reply parsing between _reply_to_chroma and list_by_filter - Parenthesize multi-condition pre-filters before the => KNN clause - Fail closed when a username is configured without a password - Catch only RequestError on ft.dropindex (let connection/auth errors surface) - Bound the delete_collection SCAN loop with a safety cap - Add VectorDatabase.close() (no-op default) + VectorDBManager.shutdown() - Simplify _MATCH_ALL literal; normalize typing to builtin generics * fix(vector/valkey_search): address round-2 review feedback - Serialize lazy client creation with an asyncio.Lock (double-checked) so concurrent first-use callers don't construct and leak duplicate clients. - Make the filter operator chain exhaustive: raise on an unhandled op rather than silently dropping the condition (which could widen delete_by_filter). - Cast numeric range (///) values to float, failing closed on non-numeric input and pre-empting a future NUMERIC-field injection surface. * refactor(vector): remove shutdown/close from base ABC per maintainer feedback Per maintainer request, interface changes to VectorDatabase ABC and VectorDBManager should be in a separate PR with implementation across all backends. The ValkeySearchVectorDatabase.close() method remains but does not override an ABC method. Signed-off-by: Daria Korenieva <daric2612@gmail.com> * docs(test): list valkey_search in vdb coverage exclusions Add valkey_search to the documented vector/vdbs/ coverage-exclusion list, matching the existing chroma/milvus/pgvector/qdrant/seekdb entries. These adapters require a live database instance and are covered by env-gated integration tests instead of unit tests. Signed-off-by: Daria Korenieva <daric2612@gmail.com> --------- Signed-off-by: Daria Korenieva <daric2612@gmail.com>
LangBot Test Suite
This directory contains the LangBot backend test suite, including unit tests, integration tests, startup E2E tests, and container-backed Box runtime tests.
Quality Gate Layers
LangBot uses a layered quality gate system for developers and CI:
| Layer | Command | What it runs | When to use |
|---|---|---|---|
| Quick | make test-quick or bash scripts/test-quick.sh |
Ruff lint + Unit tests + Smoke tests | Before every commit |
| Fast Integration | make test-integration-fast or bash scripts/test-integration-fast.sh |
SQLite/API/Pipeline integration (no external services) | Before PR, weekly |
| Backend E2E | uv run --python 3.12 pytest tests/e2e -q --tb=short |
Starts a real LangBot process with minimal config | Before release, CI |
| Box Integration | uv run --python 3.12 pytest tests/integration_tests -q --tb=short |
Real Box sandbox/runtime integration | Before Box/runtime changes, CI |
| Frontend E2E | cd web && pnpm test:e2e |
Playwright smoke tests with mocked backend and Space APIs | Before web changes, CI |
| Coverage Gate | make test-coverage or bash scripts/test-coverage.sh |
All tests with coverage, threshold: 18% | Before merge, CI |
| Full Local | make test-all-local |
Quick + Integration + Coverage | Before major changes |
Note: PostgreSQL migration tests and slow tests are NOT in local default
gates. They run in separate CI workflows. Frontend Playwright tests live under
web/tests/e2e and are documented in web/README.md.
Developer Workflow
# Daily: Quick self-test
bash scripts/test-quick.sh
# Before PR: Full local gate
make test-all-local
# Or run each layer separately:
bash scripts/test-quick.sh # ~2 min
bash scripts/test-integration-fast.sh # ~3 min
bash scripts/test-coverage.sh # ~8 min
uv run --python 3.12 pytest tests/e2e -q --tb=short
uv run --python 3.12 pytest tests/integration_tests -q --tb=short
cd web && pnpm test:e2e
Coverage Baseline
Current coverage threshold: 18% Actual coverage: 30%
This is a conservative baseline to prevent coverage regression. It does NOT represent the final quality target. Key modules have higher coverage:
pipeline.preproc.preproc: 53%pipeline.process.process: 96%pipeline.respback.respback: 88%telemetry.telemetry: 87%provider.session.sessionmgr: 100%provider.tools.toolmgr: 83%storage.providers.s3storage: 80%
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/
├── __init__.py
├── factories/ # Shared test factories
│ ├── __init__.py # Factory exports
│ ├── app.py # FakeApp factory
│ ├── message.py # Message/query factories
│ ├── provider.py # FakeProvider factory
│ └── platform.py # FakePlatform factory
├── integration/ # Integration tests (real resources)
│ ├── __init__.py
│ ├── api/ # HTTP API tests
│ │ ├── __init__.py
│ │ └── test_smoke.py # API smoke tests
│ ├── pipeline/ # Pipeline stage-chain tests
│ │ ├── __init__.py
│ │ └── test_full_flow.py # Full flow integration
│ └── persistence/ # Database/persistence tests
│ ├── __init__.py
│ └── test_migrations.py # Alembic migration tests
├── e2e/ # Real LangBot startup E2E tests
│ ├── conftest.py
│ ├── test_startup.py
│ └── utils/
├── integration_tests/ # Container-backed integration tests
│ └── box/ # Box runtime and MCP process tests
├── smoke/ # Smoke tests (quick validation)
│ └── test_fake_message_flow.py
├── unit_tests/ # Unit tests
│ ├── box/ # Box module tests
│ ├── config/ # Configuration tests
│ ├── pipeline/ # Pipeline stage tests
│ │ └── conftest.py # Shared fixtures and test infrastructure
│ ├── platform/ # Platform adapter tests
│ ├── plugin/ # Plugin system tests
│ │ └── test_handler_actions.py # Action handler tests
│ ├── provider/ # Provider tests
│ │ ├── test_session_manager.py # SessionManager tests
│ │ └── test_tool_manager.py # ToolManager tests
│ ├── rag/ # RAG tests
│ │ └── test_file_storage.py # File/ZIP storage tests
│ ├── storage/ # Storage tests
│ │ └── test_s3storage.py # S3StorageProvider tests
│ ├── vector/ # Vector tests
│ │ └── test_vdb_filter_conversion.py # VDB filter tests
│ └── telemetry/ # Telemetry tests (rewritten)
├── utils/ # Test utilities
│ ├── __init__.py
│ └── import_isolation.py # sys.modules isolation for circular imports
└── README.md # This file
Test Factories
The tests/factories/ package provides reusable test factories:
from tests.factories import (
FakeApp, # Mock application
FakeProvider, # Fake LLM provider
FakePlatform, # Fake platform adapter
text_query, # Create text query
group_text_query, # Create group query
command_query, # Create command query
)
# Create fake app
app = FakeApp()
# Create query with text
query = text_query("hello world")
# Create fake provider that returns specific response
provider = FakeProvider().returns("test response")
# Create fake platform for outbound capture
platform = FakePlatform()
await platform.reply_message(query.message_event, reply_chain)
outbound = platform.get_outbound_messages()
See tests/factories/__init__.py for all available factories.
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
- Isolation: Each test is independent and doesn't rely on external systems
- Mocking: All external dependencies are mocked to ensure fast, reliable tests
- Coverage: Tests cover happy paths, edge cases, and error conditions
- Extensibility: Easy to add new tests by reusing existing fixtures
Running Tests
Quick self-test for developers
For local branch validation without real provider keys:
make test-quick
or
bash scripts/test-quick.sh
This runs:
- Ruff lint check
- Unit tests
- Smoke tests
Suitable for quick validation before committing.
Using the test runner script (recommended for full coverage)
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 unit tests
uv run pytest tests/unit_tests/ --cov=langbot --cov-report=xml --cov-report=term
Run specific test module
uv run pytest tests/unit_tests/pipeline/ -v
Run specific test file
uv run pytest tests/unit_tests/pipeline/test_bansess.py -v
Run with coverage
uv run pytest tests/unit_tests/pipeline/ --cov=langbot --cov-report=html
Run specific test
uv run pytest tests/unit_tests/pipeline/test_bansess.py::test_bansess_whitelist_allow -v
Using markers
# Run only unit tests
uv run pytest tests/unit_tests/ -m unit
# Run only integration tests
uv run pytest tests/integration/ -m integration
# Run integration tests excluding slow ones
uv run pytest tests/integration/ -m "not slow" -q
# Skip slow tests
uv run pytest tests/unit_tests/ -m "not slow"
Running integration tests
Integration tests validate real system behavior with actual database/network resources.
# Run all integration tests (excluding slow ones)
uv run pytest tests/integration/ -m "not slow" -q
# Run SQLite migration integration tests
uv run pytest tests/integration/persistence/test_migrations.py -q --tb=short
# Run API smoke integration tests
uv run pytest tests/integration/api/test_smoke.py -q
# Run pipeline full-flow integration tests
uv run pytest tests/integration/pipeline/test_full_flow.py -q
# Run with verbose output
uv run pytest tests/integration/ -v
Note: Integration tests use:
- Temporary databases (tmp_path) for persistence tests
- Fake app/services for API tests (no real provider/platform)
- Fake runner/provider for pipeline tests (no real LLM API)
- Do not require external services
Running migration tests locally
SQLite migration tests can be run locally without any external dependencies:
# SQLite migration tests (uses tmp_path, no external DB needed)
uv run pytest tests/integration/persistence/test_migrations.py -q --tb=short
PostgreSQL migration tests require an external PostgreSQL database:
# PostgreSQL migration tests (requires PostgreSQL service)
# Tests are marked as slow and skipped if TEST_POSTGRES_URL is not set
TEST_POSTGRES_URL=postgresql+asyncpg://user:pass@localhost:5432/test_db \
uv run pytest tests/integration/persistence/test_migrations_postgres.py -q --tb=short
# Or skip by default (no PostgreSQL available)
uv run pytest tests/integration/persistence/test_migrations_postgres.py -q --tb=short
# Output: SKIPPED (TEST_POSTGRES_URL not set)
Note: PostgreSQL tests are not included in fast integration gate because they:
- Require external PostgreSQL service
- Are marked with
@pytest.mark.slow - Need
TEST_POSTGRES_URLenvironment variable
CI workflow .github/workflows/test-migrations.yml runs:
- SQLite tests in
test-migrations-sqlitejob (fast, no external services) - PostgreSQL tests in
test-migrations-postgresjob (uses PostgreSQL service container)
Running pipeline integration tests locally
Pipeline full-flow integration tests validate real stage interactions:
# Run pipeline integration tests (uses fake runner, no real LLM API)
uv run pytest tests/integration/pipeline/test_full_flow.py -q --tb=short
# Run with coverage for pipeline modules
uv run pytest tests/integration/pipeline \
--cov=langbot.pkg.pipeline.preproc.preproc \
--cov=langbot.pkg.pipeline.process.process \
--cov=langbot.pkg.pipeline.respback.respback \
--cov-report=term -q
These tests:
- Use
FakeRunnerclass to simulate LLM responses without real API calls - Import real
PreProcessor,MessageProcessor,SendResponseBackStagestages - Validate stage chain: PreProcessor → Processor → SendResponseBackStage
- Test prevent_default, exception handling, and full message flow
- Do not require real LLM provider keys
Running backend E2E startup tests
Backend E2E tests start a real LangBot process with a generated minimal
data/config.yaml, SQLite database, local storage, and embedded Chroma path.
They do not require provider keys or external services.
uv run --python 3.12 pytest tests/e2e -q --tb=short
These tests verify startup orchestration, migrations, API route registration, and the minimal no-LLM startup path. The E2E process manager disables ambient proxy variables for subprocess startup and uses direct localhost HTTP clients, so local proxy settings should not affect the health checks.
Running Box integration tests
Box integration tests exercise the real sandbox runtime path, including command execution, session persistence, managed process WebSocket attachment, and cleanup behavior.
uv run --python 3.12 pytest tests/integration_tests -q --tb=short
These tests require a working Docker or Podman runtime. In CI, the dedicated Box integration job checks Docker availability before running the tests.
Running frontend E2E tests
Frontend E2E tests live in web/tests/e2e and use Playwright. They start Vite
and mock the LangBot backend and Space APIs, so no backend process is required.
cd web
pnpm test:e2e
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:
- Make sure you're running from the project root directory
- Ensure dependencies are installed:
uv sync --dev - Try running a simple test 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.11, 3.12, and 3.13 to ensure compatibility.
Startup E2E and Box integration tests run as separate Python 3.12 jobs because
they exercise process/container behavior instead of pure Python compatibility.
Frontend Playwright smoke tests run in .github/workflows/frontend-tests.yml.
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 langbot.pkg.pipeline.<module>.<stage> import <StageClass>
from langbot.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 the appropriate 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.unit_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
- Test naming: Use descriptive names that explain what's being tested
- Arrange-Act-Assert: Structure tests clearly with setup, execution, and verification
- One assertion per test: Focus each test on a single behavior
- Mock appropriately: Mock external dependencies, not the code under test
- Use fixtures: Reuse common test data through fixtures
- Document tests: Add docstrings explaining what each test validates
Troubleshooting
Import errors
Make sure you've installed the package in development mode:
uv sync --dev
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 database migrations (SQLite)
- Add PostgreSQL migration integration tests (G-003)
- Add integration tests for full pipeline execution
- Add API smoke integration tests
- Add E2E tests
- Add performance benchmarks
- Add mutation testing for better coverage quality
- Add property-based testing with Hypothesis