Files
insistence 2a055ba648 feat: 新增插件系统 (#112)
* feat: 初始化插件系统

* refactor: 收口插件系统运行时重构

* perf: 优化插件系统类型提示

* fix&perf: 修复和优化插件系统

* fix: 修复gitignore规则误忽略插件文件的问题

* fix: 修复运行时插件根路径算错的问题

* fix: 加强插件发现和路由注册的防护措施

* revert: 回滚定时任务白名单

* fix: 移除未使用的应用路由注册探测

* revert: 恢复部分代码

* perf: 优化插件系统

* docs: 新增插件开发文档

* perf: 优化插件管理模块

* perf: 提升插件系统核心能力

* refactor: 重构生命周期 step runner

* fix: 修复lint错误

* test: 清理测试用例

* test: 调整测试目录名称

* fix: 修复前后端目录硬编码的问题

* fix: 修复插件系统安全性缺口

* refactor: 重新设计插件生命周期 Migration 事务与回滚

* perf: 优化插件系统边界问题

* refactor: 重构当前插件系统的依赖体系设计

* perf: 优化代码

* perf: 优化代码

* fix: 修复代码合并问题

* fix: 修复bug

* perf: 优化代码

* perf&fix: 优化代码和修复bug

* docs: 优化文档格式

* feat: 适配Vue2版本

* docs: 更新README文档

* fix: 修复ruff lint错误

* chore: 更新后端依赖文件
2026-07-28 20:35:18 +08:00

92 lines
3.7 KiB
Python

from types import SimpleNamespace
from plugins.core.state import PluginStateResolver, PluginStateSnapshot, PluginStateTransitionTable
def test_state_resolver_covers_lifecycle_states() -> None:
"""校验插件状态解析器覆盖全部生命周期状态。"""
cases = [
(PluginStateSnapshot(source_version='1.0.0', installed_version=None, enabled=True), 'discovered'),
(PluginStateSnapshot(source_version='1.0.0', installed_version='1.0.0', enabled=False), 'installed'),
(PluginStateSnapshot(source_version='1.1.0', installed_version='1.0.0', enabled=True), 'pending_upgrade'),
(
PluginStateSnapshot(
source_version='1.1.0',
installed_version='1.0.0',
enabled=True,
current_status='error',
),
'error',
),
]
assert [PluginStateResolver.resolve(snapshot) for snapshot, _expected in cases] == [
expected for _snapshot, expected in cases
]
def test_state_resolver_enables_only_installed_healthy_database_plugins() -> None:
"""校验状态解析器只启用已安装且健康的数据库插件。"""
cases = [
(SimpleNamespace(enabled='0', status='installed', installed_version='1.0.0'), True),
(SimpleNamespace(enabled='1', status='installed', installed_version='1.0.0'), False),
(SimpleNamespace(enabled='0', status='error', installed_version='1.0.0'), False),
(SimpleNamespace(enabled='0', status='discovered', installed_version=None), False),
]
assert [PluginStateResolver.is_enabled(plugin) for plugin, _expected in cases] == [
expected for _plugin, expected in cases
]
def test_database_enabled_state_respects_error_status() -> None:
"""校验数据库启用状态不会覆盖插件错误状态。"""
cases = [
(SimpleNamespace(enabled='0', status='installed'), True),
(SimpleNamespace(enabled='0', status='error'), False),
]
assert [PluginStateResolver.is_database_plugin_enabled(plugin) for plugin, _expected in cases] == [
expected for _plugin, expected in cases
]
def test_state_transition_table_covers_supported_and_rejected_operations() -> None:
"""校验状态转换表覆盖支持与拒绝的操作。"""
supported_cases = [
('discovered', 'install', 'installed'),
('discovered', 'disable', 'discovered'),
('installed', 'mark_error', 'error'),
('pending_upgrade', 'mark_error', 'error'),
]
assert [
PluginStateTransitionTable.resolve_target(source, operation) for source, operation, _expected in supported_cases
] == [expected for _source, _operation, expected in supported_cases]
assert PluginStateTransitionTable.can_transition('discovered', 'upgrade') is False
assert PluginStateTransitionTable.can_transition('error', 'enable') is False
def test_state_resolver_and_transition_table_remain_consistent() -> None:
"""校验状态解析器与转换表保持一致。"""
cases = [
(None, 'discover', PluginStateSnapshot(source_version='1.0.0', installed_version=None, enabled=True)),
(
'discovered',
'install',
PluginStateSnapshot(source_version='1.0.0', installed_version='1.0.0', enabled=True),
),
(
'installed',
'upgrade_available',
PluginStateSnapshot(source_version='1.1.0', installed_version='1.0.0', enabled=True),
),
]
for source, operation, snapshot in cases:
assert PluginStateResolver.resolve(snapshot) == PluginStateTransitionTable.resolve_target(source, operation)
transitions = PluginStateTransitionTable.list_transitions()
assert transitions
assert all(transition.description for transition in transitions)