xiaoxia-saas/tests/unit/test_module_registry.py

"""Module Registry 单元测试."""

from __future__ import annotations

import pytest

from packages.infrastructure.module_registry import (
    Module,
    ModuleCapability,
    ModuleRegistry,
    ModuleStatus,
    QuotaRule,
    module_registry,
)


@pytest.fixture(autouse=True)
def clean_registry():
    """每个测试前后清空全局单例，避免测试间干扰."""
    module_registry.clear()
    yield
    module_registry.clear()


# ── Module 数据类测试 ──────────────────────────────────────────────────


class TestModuleDataclass:
    """Module 数据类基本行为测试."""

    def test_create_module_defaults(self):
        """创建模块，默认值正确."""
        mod = Module(name="test_module")
        assert mod.name == "test_module"
        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_create_module_full(self):
        """创建模块，完整参数."""
        mod = Module(
            name="ai_voice",
            version="2.0.0",
            description="AI配音模块",
            capabilities=[ModuleCapability(name="gen_voice")],
            dependencies=["core"],
            status=ModuleStatus.ACTIVE,
            config={"key": "value"},
        )
        assert mod.name == "ai_voice"
        assert mod.version == "2.0.0"
        assert mod.description == "AI配音模块"
        assert len(mod.capabilities) == 1
        assert mod.dependencies == ["core"]
        assert mod.status == ModuleStatus.ACTIVE
        assert mod.config == {"key": "value"}

    def test_module_activate(self):
        """激活模块."""
        mod = Module(name="m1")
        assert mod.status == ModuleStatus.REGISTERED
        mod.activate()
        assert mod.status == ModuleStatus.ACTIVE

    def test_module_activate_error_state_ignored(self):
        """error状态的模块不能激活."""
        mod = Module(name="m1", status=ModuleStatus.ERROR)
        mod.activate()
        assert mod.status == ModuleStatus.ERROR

    def test_module_disable(self):
        """禁用模块."""
        mod = Module(name="m1", status=ModuleStatus.ACTIVE)
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED


class TestQuotaRule:
    """QuotaRule 测试."""

    def test_quota_rule_basic(self):
        """基本配额规则."""
        rule = QuotaRule(dimension="credits", per_operation=1.0, description="每次消耗1积分")
        assert rule.dimension == "credits"
        assert rule.per_operation == 1.0
        assert rule.description == "每次消耗1积分"

    def test_quota_rule_default_description(self):
        """默认描述为空."""
        rule = QuotaRule(dimension="storage_gb", per_operation=0.5)
        assert rule.description == ""


class TestModuleCapability:
    """ModuleCapability 测试."""

    def test_capability_basic(self):
        """基本能力定义."""
        cap = ModuleCapability(name="generate_voice", description="文本转配音")
        assert cap.name == "generate_voice"
        assert cap.description == "文本转配音"
        assert cap.quota_rules == []
        assert cap.metadata == {}

    def test_capability_with_quota_rules(self):
        """带配额规则的能力."""
        rules = [
            QuotaRule("ai_credits", 1.0, "配音积分"),
            QuotaRule("storage_gb", 0.1, "存储占用"),
        ]
        cap = ModuleCapability(
            name="generate_voice",
            quota_rules=rules,
            metadata={"speed": "fast"},
        )
        assert len(cap.quota_rules) == 2
        assert cap.metadata["speed"] == "fast"


# ── ModuleRegistry 核心测试 ────────────────────────────────────────


class TestModuleRegistryRegister:
    """模块注册测试."""

    def test_register_single_module(self):
        """注册单个模块."""
        registry = ModuleRegistry()
        mod = Module(name="test_mod")
        registry.register(mod)
        assert registry.get("test_mod") is mod

    def test_register_duplicate_raises(self):
        """重复注册抛异常."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        with pytest.raises(ValueError, match="already registered"):
            registry.register(Module(name="m1"))

    def test_register_auto_activate_no_deps(self):
        """无依赖的模块注册后自动激活."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        assert registry.get("m1").status == ModuleStatus.ACTIVE

    def test_register_with_missing_dependency(self):
        """有未满足依赖的模块保持REGISTERED."""
        registry = ModuleRegistry()
        registry.register(Module(name="m2", dependencies=["m1"]))
        assert registry.get("m2").status == ModuleStatus.REGISTERED

    def test_register_with_satisfied_dependency(self):
        """依赖已满足的模块注册后自动激活."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        registry.register(Module(name="m2", dependencies=["m1"]))
        assert registry.get("m2").status == ModuleStatus.ACTIVE


class TestModuleRegistryUnregister:
    """模块注销测试."""

    def test_unregister_existing(self):
        """注销已存在的模块."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        registry.unregister("m1")
        assert registry.get("m1") is None

    def test_unregister_nonexistent_raises(self):
        """注销不存在的模块抛异常."""
        registry = ModuleRegistry()
        with pytest.raises(KeyError, match="not found"):
            registry.unregister("nonexistent")

    def test_unregister_with_dependents_raises(self):
        """被其他模块依赖时不能注销."""
        registry = ModuleRegistry()
        registry.register(Module(name="core"))
        registry.register(Module(name="plugin", dependencies=["core"]))
        with pytest.raises(ValueError, match="depended on by"):
            registry.unregister("core")


class TestModuleRegistryQuery:
    """模块查询测试."""

    def test_get_nonexistent_returns_none(self):
        """获取不存在的模块返回None."""
        registry = ModuleRegistry()
        assert registry.get("nonexistent") is None

    def test_list_modules_all(self):
        """列出所有模块."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        registry.register(Module(name="m2"))
        assert len(registry.list_modules()) == 2

    def test_list_modules_by_status(self):
        """按状态过滤模块."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))  # ACTIVE
        m2 = Module(name="m2", status=ModuleStatus.DISABLED)
        registry.register(m2)
        m2.disable()
        active = registry.list_modules(status=ModuleStatus.ACTIVE)
        assert len(active) == 1
        assert active[0].name == "m1"

    def test_list_modules_disabled(self):
        """列出已禁用模块."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        m2 = Module(name="m2")
        registry.register(m2)
        m2.disable()
        disabled = registry.list_modules(status=ModuleStatus.DISABLED)
        assert len(disabled) == 1
        assert disabled[0].name == "m2"


class TestModuleRegistryCapabilities:
    """能力查询测试."""

    def test_has_capability_true(self):
        """检查已存在的能力."""
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="ai_mod",
                capabilities=[ModuleCapability(name="generate_voice")],
            )
        )
        assert registry.has_capability("generate_voice") is True

    def test_has_capability_false(self):
        """检查不存在的能力."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        assert registry.has_capability("nonexistent") is False

    def test_has_capability_inactive_module(self):
        """非激活模块的能力不计入."""
        registry = ModuleRegistry()
        m = Module(
            name="ai_mod",
            status=ModuleStatus.DISABLED,
            capabilities=[ModuleCapability(name="generate_voice")],
        )
        registry._modules["ai_mod"] = m
        assert registry.has_capability("generate_voice") is False

    def test_get_capability_returns_definition(self):
        """获取能力定义."""
        registry = ModuleRegistry()
        cap = ModuleCapability(name="gen_voice", description="配音")
        registry.register(Module(name="ai_mod", capabilities=[cap]))
        result = registry.get_capability("gen_voice")
        assert result is not None
        assert result.name == "gen_voice"
        assert result.description == "配音"

    def test_get_capability_nonexistent(self):
        """获取不存在的能力返回None."""
        registry = ModuleRegistry()
        assert registry.get_capability("nonexistent") is None

    def test_get_quota_rules_empty(self):
        """没有配额规则时返回空列表."""
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="m1",
                capabilities=[ModuleCapability(name="do_something")],
            )
        )
        rules = registry.get_quota_rules("do_something")
        assert rules == []

    def test_get_quota_rules_with_rules(self):
        """获取配额规则."""
        registry = ModuleRegistry()
        rules = [QuotaRule("credits", 2.0)]
        registry.register(
            Module(
                name="m1",
                capabilities=[ModuleCapability(name="do_something", quota_rules=rules)],
            )
        )
        result = registry.get_quota_rules("do_something")
        assert len(result) == 1
        assert result[0].dimension == "credits"
        assert result[0].per_operation == 2.0

    def test_get_active_capabilities(self):
        """获取所有已激活模块的能力."""
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="mod_a",
                capabilities=[
                    ModuleCapability(name="cap_a1"),
                    ModuleCapability(name="cap_a2"),
                ],
            )
        )
        registry.register(
            Module(
                name="mod_b",
                capabilities=[ModuleCapability(name="cap_b1")],
            )
        )
        result = registry.get_active_capabilities()
        assert "mod_a" in result
        assert "mod_b" in result
        assert set(result["mod_a"]) == {"cap_a1", "cap_a2"}
        assert result["mod_b"] == ["cap_b1"]


class TestModuleRegistryDependencies:
    """依赖检查测试."""

    def test_check_dependencies_satisfied(self):
        """依赖满足."""
        registry = ModuleRegistry()
        registry.register(Module(name="core"))
        registry.register(Module(name="plugin", dependencies=["core"]))
        assert registry.check_dependencies("plugin") is True

    def test_check_dependencies_missing(self):
        """依赖缺失."""
        registry = ModuleRegistry()
        registry.register(Module(name="plugin", dependencies=["core"]))
        assert registry.check_dependencies("plugin") is False

    def test_check_dependencies_module_not_found(self):
        """模块不存在返回False."""
        registry = ModuleRegistry()
        assert registry.check_dependencies("nonexistent") is False

    def test_check_dependencies_inactive_dep(self):
        """依赖模块未激活."""
        registry = ModuleRegistry()
        core = Module(name="core", status=ModuleStatus.DISABLED)
        registry._modules["core"] = core
        registry.register(Module(name="plugin", dependencies=["core"]))
        # 注册plugin时core不是ACTIVE，所以plugin不会自动激活
        assert registry.check_dependencies("plugin") is False


class TestModuleRegistryClear:
    """清空注册测试."""

    def test_clear_removes_all(self):
        """清空所有模块."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1"))
        registry.register(Module(name="m2"))
        assert len(registry.list_modules()) == 2
        registry.clear()
        assert len(registry.list_modules()) == 0

    def test_global_singleton_clear(self):
        """全局单例清空有效."""
        module_registry.register(Module(name="global_test"))
        assert module_registry.get("global_test") is not None
        # fixture 会在每个测试前后清空，这里手动验证
        module_registry.clear()
        assert module_registry.get("global_test") is None


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"