mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-06-02 03:55:55 +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>
142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
"""Unit tests for persistence manager methods.
|
|
|
|
Tests cover:
|
|
- execute_async() with mock database
|
|
- get_db_engine() with mock database manager
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, MagicMock
|
|
from importlib import import_module
|
|
import sqlalchemy
|
|
|
|
|
|
def get_persistence_module():
|
|
"""Lazy import to avoid circular import issues."""
|
|
return import_module('langbot.pkg.persistence.mgr')
|
|
|
|
|
|
class TestExecuteAsync:
|
|
"""Tests for execute_async method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_async_calls_engine_execute(self):
|
|
"""Test that execute_async calls engine execute."""
|
|
persistence = get_persistence_module()
|
|
|
|
mock_app = Mock()
|
|
mock_app.persistence_mgr = None
|
|
|
|
mgr = persistence.PersistenceManager(mock_app)
|
|
|
|
# Mock database manager with async engine
|
|
mock_engine = MagicMock()
|
|
mock_conn = AsyncMock()
|
|
mock_conn.execute = AsyncMock(return_value=Mock())
|
|
mock_conn.commit = AsyncMock()
|
|
|
|
# Setup the async context manager
|
|
async_cm = AsyncMock()
|
|
async_cm.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
async_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_engine.connect = Mock(return_value=async_cm)
|
|
|
|
mock_db = Mock()
|
|
mock_db.get_engine = Mock(return_value=mock_engine)
|
|
mgr.db = mock_db
|
|
|
|
# Execute a simple select
|
|
result = await mgr.execute_async(sqlalchemy.select(1))
|
|
|
|
mock_conn.execute.assert_called_once()
|
|
mock_conn.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_async_returns_result(self):
|
|
"""Test that execute_async returns the result."""
|
|
persistence = get_persistence_module()
|
|
|
|
mock_app = Mock()
|
|
mgr = persistence.PersistenceManager(mock_app)
|
|
|
|
mock_result = Mock(name='query_result')
|
|
|
|
mock_engine = MagicMock()
|
|
mock_conn = AsyncMock()
|
|
mock_conn.execute = AsyncMock(return_value=mock_result)
|
|
mock_conn.commit = AsyncMock()
|
|
|
|
async_cm = AsyncMock()
|
|
async_cm.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
async_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_engine.connect = Mock(return_value=async_cm)
|
|
|
|
mock_db = Mock()
|
|
mock_db.get_engine = Mock(return_value=mock_engine)
|
|
mgr.db = mock_db
|
|
|
|
result = await mgr.execute_async(sqlalchemy.text("SELECT 1"))
|
|
|
|
assert result == mock_result
|
|
|
|
|
|
class TestGetDbEngine:
|
|
"""Tests for get_db_engine method."""
|
|
|
|
def test_get_db_engine_returns_engine_from_db_manager(self):
|
|
"""Test that get_db_engine returns engine from db manager."""
|
|
persistence = get_persistence_module()
|
|
|
|
mock_app = Mock()
|
|
mgr = persistence.PersistenceManager(mock_app)
|
|
|
|
mock_engine = Mock(name='engine')
|
|
mock_db = Mock()
|
|
mock_db.get_engine = Mock(return_value=mock_engine)
|
|
mgr.db = mock_db
|
|
|
|
engine = mgr.get_db_engine()
|
|
|
|
assert engine == mock_engine
|
|
mock_db.get_engine.assert_called_once()
|
|
|
|
def test_get_db_engine_without_db_set_raises(self):
|
|
"""Test that get_db_engine raises when db is not set."""
|
|
persistence = get_persistence_module()
|
|
|
|
mock_app = Mock()
|
|
mgr = persistence.PersistenceManager(mock_app)
|
|
|
|
# db is not initialized
|
|
mgr.db = None
|
|
|
|
with pytest.raises(AttributeError):
|
|
mgr.get_db_engine()
|
|
|
|
|
|
class TestSerializeModelEdgeCases:
|
|
"""Tests for serialize_model edge cases."""
|
|
|
|
def test_serialize_model_with_all_columns_masked(self):
|
|
"""Test serialize_model when all columns are masked."""
|
|
persistence = get_persistence_module()
|
|
|
|
from sqlalchemy import Column, Integer, String
|
|
from sqlalchemy.orm import declarative_base
|
|
|
|
Base = declarative_base()
|
|
|
|
class SimpleModel(Base):
|
|
__tablename__ = 'simple'
|
|
id = Column(Integer, primary_key=True)
|
|
name = Column(String(50))
|
|
|
|
mock_app = Mock()
|
|
mgr = persistence.PersistenceManager(mock_app)
|
|
|
|
instance = SimpleModel(id=1, name='test')
|
|
result = mgr.serialize_model(SimpleModel, instance, masked_columns=['id', 'name'])
|
|
|
|
# Result should be empty dict when all columns masked
|
|
assert result == {} |