CI Bot
358 lines
11 KiB
Python
Executable File
358 lines
11 KiB
Python
Executable File
"""Module Registry 单元测试"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from packages.infrastructure.module_registry import (
|
|
Module,
|
|
ModuleCapability,
|
|
ModuleRegistry,
|
|
ModuleStatus,
|
|
QuotaRule,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def registry():
|
|
"""空的注册中心"""
|
|
reg = ModuleRegistry()
|
|
yield reg
|
|
reg.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_module():
|
|
"""示例模块"""
|
|
return Module(
|
|
name="ai_voice",
|
|
version="1.0.0",
|
|
description="AI 配音模块",
|
|
capabilities=[
|
|
ModuleCapability(
|
|
name="generate_voice",
|
|
description="文本转配音",
|
|
quota_rules=[QuotaRule("ai_voice_credits", 1.0, "每次消耗 1 积分")],
|
|
metadata={"category": "audio"},
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
class TestModuleStatus:
|
|
"""ModuleStatus 枚举测试"""
|
|
|
|
def test_status_values(self):
|
|
"""所有状态值存在"""
|
|
assert ModuleStatus.REGISTERED.value == "registered"
|
|
assert ModuleStatus.ACTIVE.value == "active"
|
|
assert ModuleStatus.DISABLED.value == "disabled"
|
|
assert ModuleStatus.ERROR.value == "error"
|
|
|
|
def test_default_status(self):
|
|
"""模块默认状态为 REGISTERED"""
|
|
mod = Module(name="test")
|
|
assert mod.status == ModuleStatus.REGISTERED
|
|
|
|
|
|
class TestModule:
|
|
"""Module 数据类测试"""
|
|
|
|
def test_default_values(self):
|
|
"""默认值正确"""
|
|
mod = Module(name="test_mod")
|
|
assert mod.name == "test_mod"
|
|
assert mod.version == "1.0.0"
|
|
assert mod.description == ""
|
|
assert mod.capabilities == []
|
|
assert mod.dependencies == []
|
|
assert mod.status == ModuleStatus.REGISTERED
|
|
assert mod.config == {}
|
|
|
|
def test_activate_success(self):
|
|
"""激活成功"""
|
|
mod = Module(name="test")
|
|
mod.activate()
|
|
assert mod.status == ModuleStatus.ACTIVE
|
|
|
|
def test_activate_error_state_remains_error(self):
|
|
"""ERROR 状态的模块无法激活"""
|
|
mod = Module(name="test", status=ModuleStatus.ERROR)
|
|
mod.activate()
|
|
assert mod.status == ModuleStatus.ERROR
|
|
|
|
def test_disable(self):
|
|
"""禁用模块"""
|
|
mod = Module(name="test", status=ModuleStatus.ACTIVE)
|
|
mod.disable()
|
|
assert mod.status == ModuleStatus.DISABLED
|
|
|
|
def test_disable_from_registered(self):
|
|
"""从 registered 状态禁用"""
|
|
mod = Module(name="test")
|
|
mod.disable()
|
|
assert mod.status == ModuleStatus.DISABLED
|
|
|
|
|
|
class TestModuleCapability:
|
|
"""ModuleCapability 测试"""
|
|
|
|
def test_default_values(self):
|
|
"""默认值"""
|
|
cap = ModuleCapability(name="do_something")
|
|
assert cap.name == "do_something"
|
|
assert cap.description == ""
|
|
assert cap.quota_rules == []
|
|
assert cap.metadata == {}
|
|
|
|
|
|
class TestQuotaRule:
|
|
"""QuotaRule 测试"""
|
|
|
|
def test_default_values(self):
|
|
"""默认值"""
|
|
rule = QuotaRule(dimension="credits", per_operation=1.0)
|
|
assert rule.dimension == "credits"
|
|
assert rule.per_operation == 1.0
|
|
assert rule.description == ""
|
|
|
|
|
|
class TestModuleRegistryRegister:
|
|
"""注册模块测试"""
|
|
|
|
def test_register_success(self, registry, sample_module):
|
|
"""注册成功"""
|
|
registry.register(sample_module)
|
|
mod = registry.get("ai_voice")
|
|
assert mod is not None
|
|
assert mod.name == "ai_voice"
|
|
assert mod.status == ModuleStatus.ACTIVE # 无依赖自动激活
|
|
|
|
def test_register_duplicate_raises(self, registry, sample_module):
|
|
"""重复注册抛 ValueError"""
|
|
registry.register(sample_module)
|
|
with pytest.raises(ValueError, match="already registered"):
|
|
registry.register(sample_module)
|
|
|
|
def test_register_with_dependency_not_met(self, registry):
|
|
"""依赖未满足时注册，状态为 REGISTERED"""
|
|
mod = Module(name="dep_mod", dependencies=["base_module"])
|
|
registry.register(mod)
|
|
# 依赖不存在，不激活
|
|
result = registry.get("dep_mod")
|
|
assert result.status == ModuleStatus.REGISTERED
|
|
|
|
def test_register_with_dependency_met(self, registry):
|
|
"""依赖已满足时注册，自动激活"""
|
|
base = Module(name="base_module")
|
|
dependent = Module(name="dependent_mod", dependencies=["base_module"])
|
|
|
|
registry.register(base)
|
|
registry.register(dependent)
|
|
|
|
result = registry.get("dependent_mod")
|
|
assert result.status == ModuleStatus.ACTIVE
|
|
|
|
|
|
class TestModuleRegistryUnregister:
|
|
"""注销模块测试"""
|
|
|
|
def test_unregister_success(self, registry, sample_module):
|
|
"""注销成功"""
|
|
registry.register(sample_module)
|
|
registry.unregister("ai_voice")
|
|
assert registry.get("ai_voice") is None
|
|
|
|
def test_unregister_not_found_raises(self, registry):
|
|
"""注销不存在的模块抛 KeyError"""
|
|
with pytest.raises(KeyError, match="not found"):
|
|
registry.unregister("nonexistent")
|
|
|
|
def test_unregister_with_dependents_raises(self, registry):
|
|
"""有其他模块依赖时不能注销"""
|
|
base = Module(name="base_mod")
|
|
dependent = Module(name="dependent_mod", dependencies=["base_mod"])
|
|
|
|
registry.register(base)
|
|
registry.register(dependent)
|
|
|
|
with pytest.raises(ValueError, match="depended on by"):
|
|
registry.unregister("base_mod")
|
|
|
|
|
|
class TestModuleRegistryGet:
|
|
"""获取模块测试"""
|
|
|
|
def test_get_exists(self, registry, sample_module):
|
|
"""获取存在的模块"""
|
|
registry.register(sample_module)
|
|
mod = registry.get("ai_voice")
|
|
assert mod.name == "ai_voice"
|
|
|
|
def test_get_not_exists_returns_none(self, registry):
|
|
"""获取不存在的模块返回 None"""
|
|
assert registry.get("nonexistent") is None
|
|
|
|
|
|
class TestModuleRegistryList:
|
|
"""列出模块测试"""
|
|
|
|
def test_list_all(self, registry, sample_module):
|
|
"""列出所有模块"""
|
|
registry.register(sample_module)
|
|
registry.register(Module(name="second_mod"))
|
|
|
|
all_mods = registry.list_modules()
|
|
assert len(all_mods) == 2
|
|
|
|
def test_list_filter_by_status(self, registry, sample_module):
|
|
"""按状态过滤"""
|
|
sample_module.status = ModuleStatus.DISABLED
|
|
registry._modules["ai_voice"] = sample_module
|
|
registry.register(Module(name="active_mod"))
|
|
|
|
active_mods = registry.list_modules(status=ModuleStatus.ACTIVE)
|
|
assert len(active_mods) == 1
|
|
assert active_mods[0].name == "active_mod"
|
|
|
|
disabled_mods = registry.list_modules(status=ModuleStatus.DISABLED)
|
|
assert len(disabled_mods) == 1
|
|
assert disabled_mods[0].name == "ai_voice"
|
|
|
|
def test_list_empty(self, registry):
|
|
"""空注册中心"""
|
|
assert registry.list_modules() == []
|
|
|
|
|
|
class TestModuleRegistryCapability:
|
|
"""能力查询测试"""
|
|
|
|
def test_has_capability_true(self, registry, sample_module):
|
|
"""有指定能力"""
|
|
registry.register(sample_module)
|
|
assert registry.has_capability("generate_voice") is True
|
|
|
|
def test_has_capability_false(self, registry):
|
|
"""没有指定能力"""
|
|
assert registry.has_capability("nonexistent_cap") is False
|
|
|
|
def test_has_capability_inactive_module(self, registry, sample_module):
|
|
"""未激活模块的能力不算"""
|
|
sample_module.status = ModuleStatus.DISABLED
|
|
registry._modules["ai_voice"] = sample_module
|
|
assert registry.has_capability("generate_voice") is False
|
|
|
|
def test_get_capability_success(self, registry, sample_module):
|
|
"""获取能力定义"""
|
|
registry.register(sample_module)
|
|
cap = registry.get_capability("generate_voice")
|
|
assert cap is not None
|
|
assert cap.name == "generate_voice"
|
|
assert len(cap.quota_rules) == 1
|
|
|
|
def test_get_capability_not_found(self, registry):
|
|
"""获取不存在的能力返回 None"""
|
|
assert registry.get_capability("nonexistent") is None
|
|
|
|
def test_get_quota_rules(self, registry, sample_module):
|
|
"""获取配额规则"""
|
|
registry.register(sample_module)
|
|
rules = registry.get_quota_rules("generate_voice")
|
|
assert len(rules) == 1
|
|
assert rules[0].dimension == "ai_voice_credits"
|
|
assert rules[0].per_operation == 1.0
|
|
|
|
def test_get_quota_rules_empty(self, registry):
|
|
"""不存在的能力返回空列表"""
|
|
rules = registry.get_quota_rules("nonexistent")
|
|
assert rules == []
|
|
|
|
|
|
class TestModuleRegistryDependencies:
|
|
"""依赖检查测试"""
|
|
|
|
def test_check_dependencies_met(self, registry):
|
|
"""依赖满足返回 True"""
|
|
base = Module(name="base")
|
|
dep = Module(name="dep", dependencies=["base"])
|
|
|
|
registry.register(base)
|
|
registry.register(dep)
|
|
|
|
assert registry.check_dependencies("dep") is True
|
|
|
|
def test_check_dependencies_not_met(self, registry):
|
|
"""依赖不满足返回 False"""
|
|
dep = Module(name="dep", dependencies=["missing"])
|
|
registry._modules["dep"] = dep
|
|
|
|
assert registry.check_dependencies("dep") is False
|
|
|
|
def test_check_dependencies_module_not_found(self, registry):
|
|
"""模块不存在返回 False"""
|
|
assert registry.check_dependencies("nonexistent") is False
|
|
|
|
def test_check_dependencies_inactive_dep(self, registry):
|
|
"""依赖模块未激活返回 False"""
|
|
base = Module(name="base", status=ModuleStatus.DISABLED)
|
|
dep = Module(name="dep", dependencies=["base"])
|
|
|
|
registry._modules["base"] = base
|
|
registry._modules["dep"] = dep
|
|
|
|
assert registry.check_dependencies("dep") is False
|
|
|
|
|
|
class TestModuleRegistryActiveCapabilities:
|
|
"""获取已激活能力测试"""
|
|
|
|
def test_get_active_capabilities(self, registry, sample_module):
|
|
"""获取所有已激活模块的能力"""
|
|
registry.register(sample_module)
|
|
registry.register(
|
|
Module(
|
|
name="video_edit",
|
|
capabilities=[
|
|
ModuleCapability(name="cut_video"),
|
|
ModuleCapability(name="add_effect"),
|
|
],
|
|
)
|
|
)
|
|
|
|
caps = registry.get_active_capabilities()
|
|
assert "ai_voice" in caps
|
|
assert "generate_voice" in caps["ai_voice"]
|
|
assert "video_edit" in caps
|
|
assert "cut_video" in caps["video_edit"]
|
|
assert "add_effect" in caps["video_edit"]
|
|
|
|
def test_get_active_capabilities_empty(self, registry):
|
|
"""空注册中心返回空字典"""
|
|
assert registry.get_active_capabilities() == {}
|
|
|
|
def test_get_active_capabilities_excludes_inactive(self, registry):
|
|
"""排除未激活模块"""
|
|
mod = Module(
|
|
name="test_mod",
|
|
status=ModuleStatus.DISABLED,
|
|
capabilities=[ModuleCapability(name="do_nothing")],
|
|
)
|
|
registry._modules["test_mod"] = mod
|
|
|
|
caps = registry.get_active_capabilities()
|
|
assert "test_mod" not in caps
|
|
|
|
|
|
class TestModuleRegistryClear:
|
|
"""清空注册中心测试"""
|
|
|
|
def test_clear(self, registry, sample_module):
|
|
"""清空所有注册"""
|
|
registry.register(sample_module)
|
|
registry.register(Module(name="second"))
|
|
|
|
assert len(registry.list_modules()) == 2
|
|
|
|
registry.clear()
|
|
assert len(registry.list_modules()) == 0
|