mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-06-02 20:14:36 +00:00
3872e3e1ac
- Add test_handler_helpers.py for plugin handler helpers (7 tests) - Add test_mgr_methods.py for persistence manager (5 tests) - Add test_app_config_validation.py for core app config (12 tests) - Add test_knowledge_service.py for API knowledge service (22 tests) - Add test_kbmgr.py for RAG knowledge base manager (39 tests) - Add test_survey_manager.py for survey manager (22 tests) - Add test_connector_methods.py for plugin connector (24 tests) - Add test_funcschema.py for utils function schema (9 tests) - Add test_platform.py for utils platform detection (7 tests) - Add test_extract_deps.py for plugin deps extraction (7 tests) - Add test_database_decorator.py for persistence decorator (7 tests) - Add test_load_config.py for core config loading (19 tests) - Add COVERAGE_EXCLUSIONS.md documenting external adapter exclusions - Fix test_chat_session_limit.py path for portability Coverage: core 28% → 30%, persistence 24% → 24.4%, plugin 27% → 28% Total: 1082 tests passed, core module coverage 45.5% Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
"""Unit tests for plugin handler helper functions and methods.
|
|
|
|
Tests cover:
|
|
- _make_rag_error_response() helper function
|
|
- RuntimeConnectionHandler cleanup_plugin_data method
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock
|
|
from importlib import import_module
|
|
|
|
|
|
def get_handler_module():
|
|
"""Lazy import to avoid circular import issues."""
|
|
return import_module('langbot.pkg.plugin.handler')
|
|
|
|
|
|
class TestMakeRagErrorResponse:
|
|
"""Tests for _make_rag_error_response helper function."""
|
|
|
|
def test_creates_error_response_with_exception(self):
|
|
"""Test basic error response creation."""
|
|
handler = get_handler_module()
|
|
|
|
error = ValueError("test error message")
|
|
result = handler._make_rag_error_response(error, 'TestError')
|
|
|
|
# ActionResponse.error() returns code=1 (error status)
|
|
assert result.code == 1
|
|
assert 'TestError' in result.message
|
|
assert 'ValueError' in result.message
|
|
assert 'test error message' in result.message
|
|
|
|
def test_includes_error_type_in_message(self):
|
|
"""Test that error type is included in message."""
|
|
handler = get_handler_module()
|
|
|
|
error = RuntimeError("something went wrong")
|
|
result = handler._make_rag_error_response(error, 'VectorStoreError')
|
|
|
|
assert '[VectorStoreError/RuntimeError]' in result.message
|
|
|
|
def test_includes_extra_context_in_message(self):
|
|
"""Test that extra context fields are included."""
|
|
handler = get_handler_module()
|
|
|
|
error = Exception("embedding failed")
|
|
result = handler._make_rag_error_response(
|
|
error,
|
|
'EmbeddingError',
|
|
embedding_model_uuid='test-uuid-123',
|
|
collection_id='collection-456',
|
|
)
|
|
|
|
assert 'embedding_model_uuid=test-uuid-123' in result.message
|
|
assert 'collection_id=collection-456' in result.message
|
|
|
|
def test_handles_exception_with_no_message(self):
|
|
"""Test handling exception with empty message."""
|
|
handler = get_handler_module()
|
|
|
|
error = Exception()
|
|
result = handler._make_rag_error_response(error, 'GenericError')
|
|
|
|
# ActionResponse.error() returns code=1 (error status)
|
|
assert result.code == 1
|
|
assert '[GenericError/Exception]' in result.message
|
|
|
|
def test_formats_context_with_multiple_fields(self):
|
|
"""Test multiple context fields are comma separated."""
|
|
handler = get_handler_module()
|
|
|
|
error = IOError("file not found")
|
|
result = handler._make_rag_error_response(
|
|
error,
|
|
'FileServiceError',
|
|
storage_path='/data/file.pdf',
|
|
kb_id='kb-001',
|
|
)
|
|
|
|
assert '[storage_path=/data/file.pdf, kb_id=kb-001]' in result.message
|
|
|
|
|
|
class TestCleanupPluginData:
|
|
"""Tests for cleanup_plugin_data method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deletes_plugin_settings(self):
|
|
"""Test that plugin settings are deleted."""
|
|
handler_module = get_handler_module()
|
|
|
|
mock_app = Mock()
|
|
mock_app.persistence_mgr = AsyncMock()
|
|
mock_app.persistence_mgr.execute_async = AsyncMock()
|
|
|
|
# Mock the handler without connection (we only need ap)
|
|
handler_instance = Mock(spec=handler_module.RuntimeConnectionHandler)
|
|
handler_instance.ap = mock_app
|
|
|
|
# Call cleanup_plugin_data
|
|
await handler_module.RuntimeConnectionHandler.cleanup_plugin_data(
|
|
handler_instance, 'test-author', 'test-plugin'
|
|
)
|
|
|
|
# Verify plugin settings delete was called
|
|
calls = mock_app.persistence_mgr.execute_async.call_args_list
|
|
assert len(calls) >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deletes_binary_storage(self):
|
|
"""Test that binary storage is deleted."""
|
|
handler_module = get_handler_module()
|
|
|
|
mock_app = Mock()
|
|
mock_app.persistence_mgr = AsyncMock()
|
|
mock_app.persistence_mgr.execute_async = AsyncMock()
|
|
|
|
handler_instance = Mock(spec=handler_module.RuntimeConnectionHandler)
|
|
handler_instance.ap = mock_app
|
|
|
|
await handler_module.RuntimeConnectionHandler.cleanup_plugin_data(
|
|
handler_instance, 'author', 'plugin-name'
|
|
)
|
|
|
|
# Should have at least 2 calls: one for settings, one for binary storage
|
|
assert mock_app.persistence_mgr.execute_async.call_count >= 2 |