feat: Web 管理 API #1390
Merged
feat: Web 管理 API #1390
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
- Implemented web server with Quart and Hypercorn - Added authentication middleware with JWT and bcrypt - Created RESTful APIs for managing: * IM adapters * LLM backends * Dispatch rules * Workflows * Blocks - Introduced first-time setup and password management - Enhanced configuration loading with backup mechanism - Added support for rule and workflow priority, enabling/disabling - Integrated web configuration into global config
|
Here's the translation of the review guide to Chinese: 审阅者指南 by Sourcery此拉取请求引入了一个用于管理系统各个组件的 Web 管理 API，包括 IM 适配器、LLM 后端、分发规则、块和工作流。它涉及跨多个文件的重大更改，以实现 API 端点并将其与现有系统组件集成。 认证流程序列图sequenceDiagram
actor Client
participant Auth as Auth API
participant Utils as Auth Utils
participant Storage as Password Storage
Client->>Auth: POST /api/auth/login
Auth->>Utils: is_first_time()
Utils->>Storage: check password file
Storage-->>Utils: file exists/not exists
alt First time login
Utils-->>Auth: true
Auth->>Utils: save_password()
Utils->>Storage: write password hash
Auth->>Utils: create_access_token()
Utils-->>Auth: JWT token
else Normal login
Utils-->>Auth: false
Auth->>Utils: verify_password()
Utils->>Storage: read password hash
Storage-->>Utils: password hash
Utils-->>Auth: verification result
alt Password valid
Auth->>Utils: create_access_token()
Utils-->>Auth: JWT token
else Password invalid
Auth-->>Client: 401 Unauthorized
end
end
Auth-->>Client: token response
Web API 模型类图classDiagram
class IMAdapterConfig {
+str name
+str adapter
+Dict config
+bool enable
}
class LLMBackendConfig {
+str name
+str adapter
+Dict config
+bool enable
+List[str] models
}
class DispatchRuleConfig {
+str rule_id
+str name
+str description
+str pattern
+int priority
+str workflow_id
+bool enabled
+Dict metadata
}
class WorkflowDefinition {
+str group_id
+str workflow_id
+str name
+str description
+List[BlockInstance] blocks
+List[Wire] wires
+Dict metadata
}
class BlockType {
+str type_name
+str name
+str description
+List[BlockInput] inputs
+List[BlockOutput] outputs
+List[BlockConfig] configs
}
class PluginInfo {
+str name
+str package_name
+str description
+str version
+str author
+bool is_internal
+bool is_enabled
+Dict metadata
}
文件级更改
提示和命令与 Sourcery 交互
自定义您的体验访问您的仪表板以：
获取帮助Original review guide in EnglishReviewer's Guide by SourceryThis pull request introduces a web management API for managing various components of the system, including IM adapters, LLM backends, dispatch rules, blocks, and workflows. It involves significant changes across multiple files to implement the API endpoints and integrate them with the existing system components. Sequence diagram for authentication flowsequenceDiagram
actor Client
participant Auth as Auth API
participant Utils as Auth Utils
participant Storage as Password Storage
Client->>Auth: POST /api/auth/login
Auth->>Utils: is_first_time()
Utils->>Storage: check password file
Storage-->>Utils: file exists/not exists
alt First time login
Utils-->>Auth: true
Auth->>Utils: save_password()
Utils->>Storage: write password hash
Auth->>Utils: create_access_token()
Utils-->>Auth: JWT token
else Normal login
Utils-->>Auth: false
Auth->>Utils: verify_password()
Utils->>Storage: read password hash
Storage-->>Utils: password hash
Utils-->>Auth: verification result
alt Password valid
Auth->>Utils: create_access_token()
Utils-->>Auth: JWT token
else Password invalid
Auth-->>Client: 401 Unauthorized
end
end
Auth-->>Client: token response
Class diagram for Web API modelsclassDiagram
class IMAdapterConfig {
+str name
+str adapter
+Dict config
+bool enable
}
class LLMBackendConfig {
+str name
+str adapter
+Dict config
+bool enable
+List[str] models
}
class DispatchRuleConfig {
+str rule_id
+str name
+str description
+str pattern
+int priority
+str workflow_id
+bool enabled
+Dict metadata
}
class WorkflowDefinition {
+str group_id
+str workflow_id
+str name
+str description
+List[BlockInstance] blocks
+List[Wire] wires
+Dict metadata
}
class BlockType {
+str type_name
+str name
+str description
+List[BlockInput] inputs
+List[BlockOutput] outputs
+List[BlockConfig] configs
}
class PluginInfo {
+str name
+str package_name
+str description
+str version
+str author
+bool is_internal
+bool is_enabled
+Dict metadata
}
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
嗨 @lss233 - 我已经审查了你的更改 - 这里有一些反馈：
总体评论：
- 在合并之前，请为新的 Web API 端点添加测试覆盖率。考虑使用 pytest-asyncio 来测试异步路由。
- 考虑将常见的错误处理模式提取到共享的实用函数中，以减少跨路由处理程序的代码重复。
这是我在审查期间查看的内容
- 🟡 一般问题：发现 1 个问题
- 🟡 安全性：发现 1 个问题
- 🟢 测试：一切看起来都很好
- 🟢 复杂性：一切看起来都很好
- 🟢 文档：一切看起来都很好
帮助我变得更有用！请点击每条评论上的 👍 或 👎，我将使用反馈来改进你的评论。
Original comment in English
Hey @lss233 - I've reviewed your changes - here's some feedback:
Overall Comments:
- Please add test coverage for the new web API endpoints before merging. Consider using pytest-asyncio for testing the async routes.
- Consider extracting common error handling patterns into shared utility functions to reduce code duplication across route handlers.
Here's what I looked at during the review
- 🟡 General issues: 1 issue found
- 🟡 Security: 1 issue found
- 🟢 Testing: all looks good
- 🟢 Complexity: all looks good
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| active_rules = [rule for rule in self.rules.values() if rule.enabled] | ||
| return sorted(active_rules, key=lambda x: x.priority, reverse=True) | ||
|
|
||
| def create_rule(self, rule_id: str, name: str, description: str, rule_type: str, |
There was a problem hiding this comment.
建议： 添加规则模式冲突的验证
考虑添加验证以检测规则模式何时可能与现有规则冲突，特别是可能重叠的正则表达式模式。这将有助于防止模糊的调度场景。
建议的实现：
这个实现假设 DispatchRule 类（及其子类）实现了：
- 一个
has_pattern_overlap方法，用于检查两个正则表达式模式是否可能匹配相同的输入 - 一个
matches_pattern方法，用于检查一个模式是否匹配另一个模式 - 一个
pattern属性，返回规则的模式 - 一个
from_config类方法，从配置创建一个规则实例
这些方法需要在 DispatchRule 类及其子类中实现，如果它们尚不存在的话。
正则表达式模式重叠的示例实现：
Original comment in English
suggestion: Add validation for rule pattern conflicts
Consider adding validation to detect when rule patterns could potentially conflict with existing rules, especially for regex patterns that might overlap. This would help prevent ambiguous dispatch scenarios.
Suggested implementation:
def _check_pattern_conflicts(self, rule_type: str, rule_config: Dict[str, Any]) -> List[str]:
"""检查规则模式是否与现有规则冲突
Returns:
List[str]: 发生冲突的规则ID列表
"""
conflicts = []
new_rule = DispatchRule.get_rule_type(rule_type).from_config(rule_config)
for existing_rule in self.rules.values():
if existing_rule.rule_type != rule_type:
continue
# 对于正则表达式规则，检查模式是否重叠
if rule_type == "regex":
if new_rule.has_pattern_overlap(existing_rule):
conflicts.append(existing_rule.rule_id)
# 对于其他类型的规则，检查是否完全匹配
elif new_rule.matches_pattern(existing_rule.pattern):
conflicts.append(existing_rule.rule_id)
return conflicts
def create_rule(self, rule_id: str, name: str, description: str, rule_type: str,
workflow_id: str, rule_config: Dict[str, Any], priority: int = 5,
enabled: bool = True, metadata: Optional[Dict[str, Any]] = None) -> DispatchRule:
"""创建并注册一个新的规则"""
# 获取工作流构建器
workflow_builder = self.workflow_registry.get(workflow_id)
if not workflow_builder:
raise ValueError(f"Workflow {workflow_id} not found")
# 检查规则冲突
conflicts = self._check_pattern_conflicts(rule_type, rule_config)
if conflicts:
raise ValueError(f"Rule pattern conflicts with existing rules: {', '.join(conflicts)}")
# 获取规则类型
rule_class = DispatchRule.get_rule_type(rule_type)
# 创建规则配置This implementation assumes that the DispatchRule class (and its subclasses) implement:
- A
has_pattern_overlapmethod that checks if two regex patterns could match the same input - A
matches_patternmethod that checks if a pattern matches another pattern - A
patternproperty that returns the rule's pattern - A
from_configclass method that creates a rule instance from config
These methods need to be implemented in the DispatchRule class and its subclasses if they don't already exist.
Example implementation for regex pattern overlap:
def has_pattern_overlap(self, other_rule):
# This is a simplified check - you might want to use a proper regex intersection algorithm
try:
# Try to find common strings that match both patterns
test_str = exrex.getone(self.pattern)
if re.match(other_rule.pattern, test_str):
return True
except:
# If pattern generation fails, err on the side of caution
return True
return False|
|
||
| if not rule_type or not workflow_name: | ||
| raise ValueError("Rule must specify 'type' and 'workflow'") | ||
| def save_rules(self, rules_dir: Optional[str] = None): |
There was a problem hiding this comment.
🚨 建议（安全性）： 实现规则持久化的原子文件写入
使用原子文件操作（写入临时文件并重命名）以确保在写入操作期间系统崩溃时规则文件不会损坏。
建议的实现：
这个实现假设：
- json 模块已经导入
- self.rules 是一个包含要保存的规则的字典
- 你可能需要在文件操作周围添加错误处理
你需要确保在实际代码库中满足这些依赖关系。
Original comment in English
🚨 suggestion (security): Implement atomic file writing for rule persistence
Use atomic file operations (write to temporary file and rename) to ensure rule files don't get corrupted if the system crashes during a write operation.
Suggested implementation:
def save_rules(self, rules_dir: Optional[str] = None):
"""保存所有规则到文件"""
rules_dir = rules_dir or self.rules_dir
import tempfile
if not os.path.exists(rules_dir):
os.makedirs(rules_dir)
for rule_type, rules in self.rules.items():
file_path = os.path.join(rules_dir, f"{rule_type}.json")
# Create temporary file in the same directory to ensure atomic move works across devices
with tempfile.NamedTemporaryFile(mode='w', dir=rules_dir, delete=False) as tmp_file:
json.dump(rules, tmp_file, indent=2, ensure_ascii=False)
tmp_file.flush()
os.fsync(tmp_file.fileno()) # Ensure all data is written to disk
# Atomic replace of the target file
os.replace(tmp_file.name, file_path)This implementation assumes:
- The json module is already imported
- self.rules is a dictionary containing the rules to be saved
- You may want to add error handling around the file operations
You'll need to ensure these dependencies are met in the actual codebase.
| """获取所有调度规则""" | ||
| registry: DispatchRuleRegistry = g.container.resolve(DispatchRuleRegistry) | ||
|
|
||
| rules = [] |
There was a problem hiding this comment.
问题（代码质量）： 将 for 循环转换为列表推导式
Original comment in English
issue (code-quality): Convert for loop into list comprehension (list-comprehension)
Comment on lines
+44
to
+46
| rule = registry.get_rule(rule_id) | ||
| if not rule: | ||
| return jsonify({"error": "Rule not found"}), 404 |
There was a problem hiding this comment.
问题（代码质量）： 我们发现了这些问题：
- 使用命名表达式简化赋值和条件
- 在控制流跳转后将代码提升到 else 中
- 交换 if/else 分支
Original comment in English
issue (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Lift code into else after jump in control flow (
reintroduce-else) - Swap if/else branches (
swap-if-else-branches)
framework/web/api/im/routes.py
Outdated
| manager = g.container.resolve(IMManager) | ||
|
|
||
| # 查找适配器类型 | ||
| adapter_type = None |
There was a problem hiding this comment.
问题（代码质量）： 使用内置函数 next 而不是 for 循环
Original comment in English
issue (code-quality): Use the built-in function next instead of a for-loop (use-next)
| secret_key = get_config().web.secret_key | ||
| jwt.decode(token, secret_key, algorithms=["HS256"]) | ||
| return True | ||
| except: |
There was a problem hiding this comment.
建议（代码质量）： 使用 except Exception: 而不是裸 except:
Original comment in English
suggestion (code-quality): Use except Exception: rather than bare except: (do-not-use-bare-except)
Suggested change
| except: | |
| except Exception: |
Comment on lines
+80
to
+90
| rule = self.create_rule( | ||
| rule_id=rule_id, | ||
| name=name, | ||
| description=description, | ||
| rule_type=rule_type, | ||
| workflow_id=workflow_id, | ||
| rule_config=rule_config, | ||
| priority=priority, | ||
| enabled=enabled, | ||
| metadata=metadata | ||
| ) |
There was a problem hiding this comment.
问题（代码质量）： 内联立即返回的变量
Original comment in English
issue (code-quality): Inline variable that is immediately returned (inline-immediately-returned-variable)
Comment on lines
+101
to
+103
| rule = self.get_rule(rule_id) | ||
| if not rule: | ||
| raise ValueError(f"Rule {rule_id} not found") |
There was a problem hiding this comment.
问题（代码质量）： 我们发现了这些问题：
- 使用命名表达式简化赋值和条件
- 在控制流跳转后将代码提升到 else 中
- 交换 if/else 分支
Original comment in English
issue (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Lift code into else after jump in control flow (
reintroduce-else) - Swap if/else branches (
swap-if-else-branches)
Comment on lines
+108
to
+110
| rule = self.get_rule(rule_id) | ||
| if not rule: | ||
| raise ValueError(f"Rule {rule_id} not found") |
There was a problem hiding this comment.
问题（代码质量）： 我们发现了这些问题：
- 使用命名表达式简化赋值和条件
- 在控制流跳转后将代码提升到 else 中
- 交换 if/else 分支
Original comment in English
issue (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Lift code into else after jump in control flow (
reintroduce-else) - Swap if/else branches (
swap-if-else-branches)
| return self.rules.copy() | ||
| # 添加规则配置 | ||
| config = rule.get_config() | ||
| rule_data.update(config.dict()) |
There was a problem hiding this comment.
问题（代码质量）： 通过联合运算符合并字典更新
Original comment in English
issue (code-quality): Merge dictionary updates via the union operator (dict-assign-update-to-union)
…us APIs - Added new web API modules for system and plugin management - Implemented system status API with resource usage tracking - Created plugin management API with support for: * Plugin listing * Plugin details retrieval * Plugin update (with restrictions) - Enhanced web framework with modular API design - Added setup files for internal plugins - Introduced new web API tests for system, plugin, and workflow management - Improved authentication and authorization mechanisms - Updated configuration models to support new plugin and system features
There was a problem hiding this comment.
嗨 @lss233 - 我已经审查了你的更改 - 这里有一些反馈：
总体评论：
- 这个 PR 涉及许多模块并引入了许多新功能。将来，考虑将如此大的更改分解为更小、更专注的 PR，以便于审查并在集成重大新功能时降低风险。
- 在新的 API 端点中有几个相似的模式（例如，配置保存、依赖解析、令牌验证）。将这些常见逻辑整合到共享的实用函数或中间件中将有助于减少代码重复并提高可维护性。
这是我在审查期间查看的内容
- 🟡 一般问题：发现 4 个问题
- 🟡 安全性：发现 1 个问题
- 🟡 测试：发现 5 个问题
- 🟢 复杂性：一切看起来都很好
- 🟢 文档：一切看起来都很好
帮助我变得更有用！请点击每条评论上的 👍 或 👎，我将使用反馈来改进你的审查。
Original comment in English
Hey @lss233 - I've reviewed your changes - here's some feedback:
Overall Comments:
- This PR covers many modules and introduces a lot of new functionality. In the future, consider breaking such large changes into smaller, more focused PRs to ease review and reduce risk while integrating major new features.
- There are several similar patterns across the new API endpoints (e.g., config saving, dependency resolution, token verification). Consolidating these common logics into shared utility functions or middleware will help reduce code duplication and improve maintainability.
Here's what I looked at during the review
- 🟡 General issues: 4 issues found
- 🟡 Security: 1 issue found
- 🟡 Testing: 5 issues found
- 🟢 Complexity: all looks good
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| raise ValueError(f"Rule {rule_id} not found") | ||
| del self.rules[rule_id] | ||
|
|
||
| def enable_rule(self, rule_id: str): |
There was a problem hiding this comment.
建议 (bug_risk): 为规则状态更改添加原子操作以防止竞争条件
考虑在启用/禁用规则时使用锁或其他同步机制，以防止在并发场景中出现竞争条件。
建议的实现：
你还需要：
- 在文件顶部添加
import threading - 考虑其他修改规则的方法（如 add_rule 如果存在）是否也需要同步
- 考虑 get_rule 是否需要同步，如果它在这些方法之外使用
Original comment in English
suggestion (bug_risk): Add atomic operations for rule state changes to prevent race conditions
Consider using a lock or other synchronization mechanism when enabling/disabling rules to prevent race conditions in concurrent scenarios.
Suggested implementation:
metadata=metadata
)
return rule
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._rules_lock = threading.Lock()
def delete_rule(self, rule_id: str): """删除规则"""
with self._rules_lock:
if rule_id not in self.rules:
raise ValueError(f"Rule {rule_id} not found")
del self.rules[rule_id] def enable_rule(self, rule_id: str):
"""启用规则"""
with self._rules_lock:
rule = self.get_rule(rule_id)
if not rule:
raise ValueError(f"Rule {rule_id} not found")
rule.enabled = True def disable_rule(self, rule_id: str):
"""禁用规则"""
with self._rules_lock:
rule = self.get_rule(rule_id)
if not rule:
raise ValueError(f"Rule {rule_id} not found")
rule.enabled = FalseYou'll also need to:
- Add
import threadingat the top of the file - Consider whether other methods that modify rules (like add_rule if it exists) also need synchronization
- Consider whether get_rule needs synchronization if it's used outside of these methods
|
|
||
| if not rule_type or not workflow_name: | ||
| raise ValueError("Rule must specify 'type' and 'workflow'") | ||
| def save_rules(self, rules_dir: Optional[str] = None): |
There was a problem hiding this comment.
🚨 建议 (security): 使用原子操作实现更安全的文件写入
考虑先写入临时文件，然后原子性地将其移动到最终位置，以防止在写入期间崩溃时出现损坏。
建议的实现：
由于我只能看到部分代码，你需要：
- 在文件顶部导入所需的模块：
import tempfile
- 在注释中指示的地方插入实际的规则写入逻辑
- 调整最终文件名（"rules.json"）以匹配你的实际文件命名方案
- 如果你正在写入多个规则文件，你需要为每个被写入的文件调整此模式
Original comment in English
🚨 suggestion (security): Implement safer file writing with atomic operations
Consider writing to a temporary file first and then atomically moving it to the final location to prevent corruption in case of crashes during writing.
Suggested implementation:
def save_rules(self, rules_dir: Optional[str] = None):
"""保存所有规则到文件"""
rules_dir = rules_dir or self.rules_dir
if not os.path.exists(rules_dir):
os.makedirs(rules_dir)
# Create a temporary file in the same directory
temp_file = tempfile.NamedTemporaryFile(
mode='w',
dir=rules_dir,
prefix='.rules_',
suffix='.tmp',
delete=False
)
try:
# Write rules to temporary file
# Your existing rule writing logic goes here
# Ensure all data is written to disk
temp_file.flush()
os.fsync(temp_file.fileno())
temp_file.close()
# Atomically rename temp file to final destination
final_path = os.path.join(rules_dir, "rules.json") # Adjust filename as needed
os.replace(temp_file.name, final_path)
except Exception as e:
# Clean up the temp file in case of any errors
temp_file.close()
try:
os.unlink(temp_file.name)
except OSError:
pass
raise eSince I can only see a portion of the code, you'll need to:
- Import the required modules at the top of the file:
import tempfile
- Insert your actual rule writing logic where indicated in the comment
- Adjust the final filename ("rules.json") to match your actual file naming scheme
- If you're writing multiple rule files, you'll need to adapt this pattern for each file being written
tests/web/api/im/test_im.py
Outdated
| from framework.im.adapter import IMAdapter | ||
| from framework.im.message import IMMessage, TextMessage | ||
| from framework.im.sender import ChatSender | ||
| from unittest.mock import patch |
There was a problem hiding this comment.
建议 (testing): 考虑使用 AsyncMock 进行异步操作
由于你正在测试异步代码，使用 AsyncMock 而不是 MagicMock 或常规 Mock 将提供更好的兼容性和更准确的异步行为模拟。
建议的实现：
在进行此导入更改后，你需要：
- 替换任何模拟异步方法的
patch()装饰器或上下文管理器，以使用AsyncMock作为规范，例如：@patch('some.async.method', new_callable=AsyncMock)- 或在测试方法中：
with patch('some.async.method', new_callable=AsyncMock):
- 更新任何直接为异步方法实例化的 Mock/MagicMock 以使用 AsyncMock
由于我无法看到文件中的实际测试方法，你需要将这些更改应用于测试中所有相关的异步方法模拟。
Original comment in English
suggestion (testing): Consider using AsyncMock for asynchronous operations
Since you are testing asynchronous code, using AsyncMock instead of MagicMock or regular Mock would provide better compatibility and more accurate mocking of asynchronous behavior.
Suggested implementation:
from unittest.mock import patch, AsyncMockAfter making this import change, you'll need to:
- Replace any
patch()decorators or context managers that mock async methods to useAsyncMockas the spec, like:@patch('some.async.method', new_callable=AsyncMock)- or within test methods:
with patch('some.async.method', new_callable=AsyncMock):
- Update any direct Mock/MagicMock instantiations for async methods to use AsyncMock instead
Since I can't see the actual test methods in the file, you'll need to apply these changes to all relevant async method mocks in your tests.
| name=TEST_ADAPTER_ID, | ||
| enable=True, | ||
| adapter=TEST_ADAPTER_TYPE, | ||
| config=TEST_ADAPTER_CONFIG |
There was a problem hiding this comment.
建议 (testing): 为适配器创建添加边缘情况测试
考虑添加无效适配器类型、缺少必需配置字段和无效配置值的测试，以确保健壮的错误处理。
建议的实现：
请注意，这些测试假设：
- API 对验证错误返回 400 状态码
- 错误响应包含带有描述性消息的 'error' 字段
- TEST_ADAPTER_CONFIG 包含一个应为字符串的 'api_key' 字段
你可能需要调整：
- 预期的错误消息以匹配你的实际 API 响应
- 无效配置测试以使用适配器中实际存在的配置字段
- 如果你的 API 使用不同的错误代码，则状态代码
Original comment in English
suggestion (testing): Add edge case tests for adapter creation
Consider adding tests for invalid adapter types, missing required config fields, and invalid config values to ensure robust error handling.
Suggested implementation:
@pytest.mark.asyncio
async def test_create_adapter(self, test_client, auth_headers):
"""测试创建适配器"""
adapter_data = IMAdapterConfig(
name='new-adapter',
adapter=TEST_ADAPTER_TYPE,
config=TEST_ADAPTER_CONFIG
)
@pytest.mark.asyncio
async def test_create_adapter_invalid_type(self, test_client, auth_headers):
"""测试创建不存在的适配器类型"""
adapter_data = IMAdapterConfig(
name='invalid-adapter',
adapter='non_existent_type',
config=TEST_ADAPTER_CONFIG
)
response = await test_client.post(
'/api/im/adapters',
json=adapter_data.dict(),
headers=auth_headers
)
assert response.status_code == 400
data = await response.get_json()
assert 'error' in data
assert 'Invalid adapter type' in data['error']
@pytest.mark.asyncio
async def test_create_adapter_missing_config(self, test_client, auth_headers):
"""测试创建适配器时缺少必要配置"""
adapter_data = IMAdapterConfig(
name='missing-config',
adapter=TEST_ADAPTER_TYPE,
config={} # Empty config
)
response = await test_client.post(
'/api/im/adapters',
json=adapter_data.dict(),
headers=auth_headers
)
assert response.status_code == 400
data = await response.get_json()
assert 'error' in data
assert 'Missing required config' in data['error']
@pytest.mark.asyncio
async def test_create_adapter_invalid_config(self, test_client, auth_headers):
"""测试创建适配器时配置值无效"""
invalid_config = TEST_ADAPTER_CONFIG.copy()
invalid_config['api_key'] = 123 # Assuming api_key should be string
adapter_data = IMAdapterConfig(
name='invalid-config',
adapter=TEST_ADAPTER_TYPE,
config=invalid_config
)
response = await test_client.post(
'/api/im/adapters',
json=adapter_data.dict(),
headers=auth_headers
)
assert response.status_code == 400
data = await response.get_json()
assert 'error' in data
assert 'Invalid config value' in data['error']Note that these tests assume:
- The API returns 400 status code for validation errors
- Error responses include an 'error' field with a descriptive message
- TEST_ADAPTER_CONFIG contains an 'api_key' field that should be a string
You may need to adjust:
- The expected error messages to match your actual API responses
- The invalid config test to use a config field that actually exists in your adapter
- The status codes if your API uses different error codes
tests/web/api/im/test_im.py
Outdated
Comment on lines
218
to
220
| assert adapter.get('config').get('name') == 'Updated Bot' | ||
|
|
||
|
|
There was a problem hiding this comment.
建议 (testing): 为更新不存在的适配器添加测试
Original comment in English
suggestion (testing): Add test for updating a non-existent adapter
Suggested change
| assert adapter.get('config').get('name') == 'Updated Bot' | |
| assert adapter.get('config').get('name') == 'Updated Bot' | |
| @pytest.mark.asyncio | |
| async def test_update_nonexistent_adapter(self, test_client, auth_headers): | |
| """测试更新不存在的适配器""" | |
| adapter_data = IMAdapterConfig( | |
| name="nonexistent-adapter", | |
| adapter=TEST_ADAPTER_TYPE, | |
| config={'token': 'test-token', 'name': 'Test Bot'} | |
| ) | |
| response = await test_client.put( | |
| '/api/im/adapters/nonexistent-adapter', | |
| headers=auth_headers, | |
| json=adapter_data.model_dump() | |
| ) | |
| assert response.status_code == 404 | |
| data = await response.get_json() | |
| assert 'error' in data | |
| assert 'message' in data | |
| assert 'not found' in data['message'].lower() | |
Comment on lines
+101
to
+103
| rule = self.get_rule(rule_id) | ||
| if not rule: | ||
| raise ValueError(f"Rule {rule_id} not found") |
There was a problem hiding this comment.
问题 (code-quality): 我们发现了这些问题：
- 使用命名表达式简化赋值和条件 (
use-named-expression) - 在控制流跳转后将代码提升到 else 中 (
reintroduce-else) - 交换 if/else 分支 (
swap-if-else-branches)
Original comment in English
issue (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Lift code into else after jump in control flow (
reintroduce-else) - Swap if/else branches (
swap-if-else-branches)
Comment on lines
+108
to
+110
| rule = self.get_rule(rule_id) | ||
| if not rule: | ||
| raise ValueError(f"Rule {rule_id} not found") |
There was a problem hiding this comment.
问题 (code-quality): 我们发现了这些问题：
- 使用命名表达式简化赋值和条件 (
use-named-expression) - 在控制流跳转后将代码提升到 else 中 (
reintroduce-else) - 交换 if/else 分支 (
swap-if-else-branches)
Original comment in English
issue (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Lift code into else after jump in control flow (
reintroduce-else) - Swap if/else branches (
swap-if-else-branches)
| return self.rules.copy() | ||
| # 添加规则配置 | ||
| config = rule.get_config() | ||
| rule_data.update(config.dict()) |
There was a problem hiding this comment.
问题 (code-quality): 通过并集运算符合并字典更新 (dict-assign-update-to-union)
Original comment in English
issue (code-quality): Merge dictionary updates via the union operator (dict-assign-update-to-union)
| workflow_registry._workflows = {'workflow1': MagicMock(), 'workflow2': MagicMock()} | ||
| container.register(WorkflowRegistry, workflow_registry) | ||
|
|
||
| app = create_app(container) |
There was a problem hiding this comment.
问题 (code-quality): 内联立即返回的变量 (inline-immediately-returned-variable)
Original comment in English
issue (code-quality): Inline variable that is immediately returned (inline-immediately-returned-variable)
| ) | ||
| container.register(GlobalConfig, config) | ||
|
|
||
| app = create_app(container) |
There was a problem hiding this comment.
问题 (code-quality): 内联立即返回的变量 (inline-immediately-returned-variable)
Original comment in English
issue (code-quality): Inline variable that is immediately returned (inline-immediately-returned-variable)
- Added new plugin management routes with full lifecycle support - Implemented plugin installation, uninstallation, enabling, and disabling - Updated PluginLoader with async methods for plugin operations - Enhanced plugin info model with more metadata - Added plugin API documentation with detailed usage examples - Created unit tests for plugin management functionality - Integrated plugin configuration updates with config backup mechanism
- Simplify concurrency group definition to use only github.ref - Maintain workflow dispatch and push to browser-version-dev branch
- Implemented `save_config_with_backup` method in ConfigLoader - Added tests for config saving with and without existing backup - Updated LLM and IM API routes to use config backup when modifying configurations - Enhanced config management with automatic file backup before modifications
lss233
force-pushed
the
feature/web_config_panel
branch
from
d9fa1e6 to
d9989b4
Compare
- Added ability tracking to LLMBackendRegistry - Introduced logging for backend adapter registration - Updated register method to include LLMAbility parameter - Improved plugin loader with internal plugin tracking
lss233
force-pushed
the
feature/web_config_panel
branch
from
d9989b4 to
e2351fa
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
实现即时通讯适配器、大语言模型后端、分发规则、块和工作流的Web管理API。
新功能：
测试：
Original summary in English
Summary by Sourcery
Implement web management APIs for IM adapters, LLM backends, dispatch rules, blocks, and workflows.
New Features:
Tests:
Original summary in English
Summary by Sourcery
实现即时通讯适配器、大语言模型后端、分发规则、块和工作流的Web管理API。
新功能：
测试：
Original summary in English
Summary by Sourcery
Implement web management APIs for IM adapters, LLM backends, dispatch rules, blocks, and workflows.
New Features:
Tests: