♻️ fix monitoring unitest

Author: Phinease

sdk/pyproject.toml

@@ -7,20 +7,11 @@ name = "nexent"
 version = "0.1.2"
 description = "Nexent Agent Framework"
 authors = [
-    { name = "Shuangrui Chen" },
-    { name = "Simeng Bian" },
-    { name = "Jingyuan Li" },
-    { name = "Mingchen Wan" },
-    { name = "Yichen Xia" },
-    { name = "Tao Liu" },
-    { name = "Peiling Jiang" },
-    { name = "Yu Lin" }
+    { name = "Nexent Dev Team" }
 ]
-readme = "README.md"
 requires-python = ">=3.10"
 keywords = ["agent", "ai", "framework"]
 classifiers = [
-    "Development Status :: 4 - Beta",
     "Intended Audience :: Developers",
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3.10",
@@ -91,7 +82,7 @@ performance = [
     "prometheus-client"
 ]
 dev = [
-    "nexent[dev, data_process]"
+    "nexent[quality, data_process, performance]"
 ]
 
 [tool.setuptools.packages.find]

test/backend/app/test_agent_app.py

@@ -1,4 +1,5 @@
-from apps.agent_app import router
+import atexit
+from unittest.mock import patch, Mock
 import os
 import sys
 import types
@@ -8,10 +9,28 @@
 from fastapi.responses import StreamingResponse
 from fastapi.testclient import TestClient
 
-# Dynamically determine the backend path
+# Dynamically determine the backend path - MUST BE FIRST
 current_dir = os.path.dirname(os.path.abspath(__file__))
 backend_dir = os.path.abspath(os.path.join(current_dir, "../../../backend"))
-sys.path.append(backend_dir)
+sys.path.insert(0, backend_dir)
+
+from apps.agent_app import router
+
+# Apply patches before importing any app modules (similar to test_base_app.py)
+
+patches = [
+    # Mock boto3 client
+    patch('boto3.client', return_value=Mock()),
+    # Mock boto3 resource
+    patch('boto3.resource', return_value=Mock()),
+    # Mock database sessions
+    patch('backend.database.client.get_db_session', return_value=Mock()),
+    # Mock Elasticsearch to prevent connection errors
+    patch('elasticsearch.Elasticsearch', return_value=Mock())
+]
+
+for p in patches:
+    p.start()
 
 # Mock external dependencies before importing the modules that use them
 # Stub nexent.core.agents.agent_model.ToolConfig to satisfy type imports in consts.model
@@ -92,6 +111,17 @@ def decorator(func):
 sys.modules['services.memory_config_service'] = pytest.importorskip(
     "unittest.mock").MagicMock()
 
+# Now safe to import app modules after all mocks are set up
+
+# Stop all patches at the end of the module
+
+
+def stop_patches():
+    for p in patches:
+        p.stop()
+
+
+atexit.register(stop_patches)
 
 # Create FastAPI app for testing
 app = FastAPI()

test/backend/utils/test_langchain_utils.py

@@ -1,12 +1,11 @@
 import unittest
-import logging
 import sys
 import os
-from unittest.mock import MagicMock, patch, Mock, call
-import pytest
+from unittest.mock import MagicMock, patch
 
 # 添加项目根目录到Python路径
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))
+sys.path.insert(0, os.path.abspath(os.path.join(
+    os.path.dirname(__file__), '..', '..', '..')))
 
 # 模拟主要依赖
 sys.modules['langchain_core.tools'] = MagicMock()
@@ -31,12 +30,12 @@ def test_is_langchain_tool(self):
         """测试_is_langchain_tool函数"""
         # 创建一个BaseTool实例的模拟
         mock_tool = MagicMock()
-        
+
         # 模拟isinstance返回值
         with patch('backend.utils.langchain_utils.isinstance', return_value=True):
             result = self._is_langchain_tool(mock_tool)
             self.assertTrue(result)
-        
+
         # 测试非BaseTool对象
         with patch('backend.utils.langchain_utils.isinstance', return_value=False):
             result = self._is_langchain_tool("not a tool")
@@ -46,44 +45,45 @@ def test_discover_langchain_modules_success(self):
         """测试成功发现LangChain工具的情况"""
         # 创建一个临时目录结构
         with patch('os.path.isdir', return_value=True), \
-             patch('os.listdir', return_value=['tool1.py', 'tool2.py', '__init__.py', 'not_a_py_file.txt']), \
-             patch('importlib.util.spec_from_file_location') as mock_spec, \
-             patch('importlib.util.module_from_spec') as mock_module_from_spec:
-            
+                patch('os.listdir', return_value=['tool1.py', 'tool2.py', '__init__.py', 'not_a_py_file.txt']), \
+                patch('importlib.util.spec_from_file_location') as mock_spec, \
+                patch('importlib.util.module_from_spec') as mock_module_from_spec:
+
             # 创建模拟工具对象
             mock_tool1 = MagicMock(name="tool1")
             mock_tool2 = MagicMock(name="tool2")
-            
+
             # 设置模拟module
             mock_module_obj1 = MagicMock()
             mock_module_obj1.tool_obj1 = mock_tool1
 
             mock_module_obj2 = MagicMock()
             mock_module_obj2.tool_obj2 = mock_tool2
-            
-            mock_module_from_spec.side_effect = [mock_module_obj1, mock_module_obj2]
-            
+
+            mock_module_from_spec.side_effect = [
+                mock_module_obj1, mock_module_obj2]
+
             # 设置模拟spec和loader
             mock_spec_obj1 = MagicMock()
             mock_spec_obj2 = MagicMock()
             mock_spec.side_effect = [mock_spec_obj1, mock_spec_obj2]
-            
+
             mock_loader1 = MagicMock()
             mock_loader2 = MagicMock()
             mock_spec_obj1.loader = mock_loader1
             mock_spec_obj2.loader = mock_loader2
-            
+
             # 设置过滤函数始终返回True
             def mock_filter(obj):
                 return obj is mock_tool1 or obj is mock_tool2
-                
+
             # 执行函数
             result = self.discover_langchain_modules(filter_func=mock_filter)
-            
+
             # 验证loader.exec_module被调用
             mock_loader1.exec_module.assert_called_once_with(mock_module_obj1)
             mock_loader2.exec_module.assert_called_once_with(mock_module_obj2)
-            
+
             # 验证结果
             self.assertEqual(len(result), 2)
             discovered_objs = [obj for (obj, _) in result]
@@ -93,39 +93,41 @@ def mock_filter(obj):
     def test_discover_langchain_modules_directory_not_found(self):
         """测试目录不存在的情况"""
         with patch('os.path.isdir', return_value=False):
-            result = self.discover_langchain_modules(directory="non_existent_dir")
+            result = self.discover_langchain_modules(
+                directory="non_existent_dir")
             self.assertEqual(result, [])
 
     def test_discover_langchain_modules_module_exception(self):
         """测试处理模块异常的情况"""
         with patch('os.path.isdir', return_value=True), \
-             patch('os.listdir', return_value=['error_module.py']), \
-             patch('importlib.util.spec_from_file_location') as mock_spec, \
-             patch('backend.utils.langchain_utils.logger', logger_mock):
-            
+                patch('os.listdir', return_value=['error_module.py']), \
+                patch('importlib.util.spec_from_file_location') as mock_spec, \
+                patch('backend.utils.langchain_utils.logger', logger_mock):
+
             # 设置spec_from_file_location抛出异常
             mock_spec.side_effect = Exception("Module error")
-            
+
             # 执行函数 - 应该捕获异常并继续
             result = self.discover_langchain_modules()
-            
+
             # 验证结果为空列表
             self.assertEqual(result, [])
             # 验证错误被记录
             self.assertTrue(logger_mock.error.called)
             # 验证错误消息包含预期内容
-            logger_mock.error.assert_called_with("Error processing module error_module.py: Module error")
+            logger_mock.error.assert_called_with(
+                "Error processing module error_module.py: Module error")
 
     def test_discover_langchain_modules_spec_loader_none(self):
         """测试spec或loader为None的情况"""
         with patch('os.path.isdir', return_value=True), \
-             patch('os.listdir', return_value=['invalid_module.py']), \
-             patch('importlib.util.spec_from_file_location', return_value=None), \
-             patch('backend.utils.langchain_utils.logger', logger_mock):
-            
+                patch('os.listdir', return_value=['invalid_module.py']), \
+                patch('importlib.util.spec_from_file_location', return_value=None), \
+                patch('backend.utils.langchain_utils.logger', logger_mock):
+
             # 执行函数
             result = self.discover_langchain_modules()
-            
+
             # 验证结果为空列表
             self.assertEqual(result, [])
             # 验证警告被记录
@@ -138,36 +140,36 @@ def test_discover_langchain_modules_spec_loader_none(self):
     def test_discover_langchain_modules_custom_filter(self):
         """测试使用自定义过滤函数的情况"""
         with patch('os.path.isdir', return_value=True), \
-             patch('os.listdir', return_value=['tool.py']), \
-             patch('importlib.util.spec_from_file_location') as mock_spec, \
-             patch('importlib.util.module_from_spec') as mock_module_from_spec:
-            
+                patch('os.listdir', return_value=['tool.py']), \
+                patch('importlib.util.spec_from_file_location') as mock_spec, \
+                patch('importlib.util.module_from_spec') as mock_module_from_spec:
+
             # 创建两个对象，一个通过过滤，一个不通过
             obj_pass = MagicMock(name="pass_object")
             obj_fail = MagicMock(name="fail_object")
-            
+
             # 设置模拟module，使其包含我们的两个测试对象
             mock_module_obj = MagicMock()
             mock_module_obj.obj_pass = obj_pass
             mock_module_obj.obj_fail = obj_fail
             mock_module_from_spec.return_value = mock_module_obj
-            
+
             # 设置模拟spec和loader
             mock_spec_obj = MagicMock()
             mock_spec.return_value = mock_spec_obj
             mock_loader = MagicMock()
             mock_spec_obj.loader = mock_loader
-            
+
             # 自定义过滤函数，只接受obj_pass
             def custom_filter(obj):
                 return obj is obj_pass
-            
+
             # 执行函数
             result = self.discover_langchain_modules(filter_func=custom_filter)
-            
+
             # 验证loader.exec_module被调用
             mock_loader.exec_module.assert_called_once_with(mock_module_obj)
-            
+
             # 验证结果 - 应该只有一个对象通过过滤
             self.assertEqual(len(result), 1)
             self.assertEqual(result[0][0], obj_pass)