xiaoxia-saas/tests/unit/test_module_registry.py

"""
Module Registry 模块注册中心单元测试

覆盖：
- ModuleStatus 枚举
- QuotaRule / ModuleCapability / Module 数据类
- Module.activate / disable 状态转换
- ModuleRegistry 注册/注销/查询/能力发现/依赖检查
"""

import pytest

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

# ============================================================
# ModuleStatus
# ============================================================


class TestModuleStatus:
    """ModuleStatus 枚举"""

    def test_enum_values(self):
        assert ModuleStatus.REGISTERED.value == "registered"
        assert ModuleStatus.ACTIVE.value == "active"
        assert ModuleStatus.DISABLED.value == "disabled"
        assert ModuleStatus.ERROR.value == "error"

    def test_is_str_enum(self):
        assert isinstance(ModuleStatus.ACTIVE, str)
        assert ModuleStatus.ACTIVE == "active"

    def test_has_four_states(self):
        assert len(ModuleStatus) == 4


# ============================================================
# QuotaRule
# ============================================================


class TestQuotaRule:
    """QuotaRule 配额规则"""

    def test_required_fields(self):
        rule = QuotaRule(dimension="ai_credits", per_operation=1.0)
        assert rule.dimension == "ai_credits"
        assert rule.per_operation == 1.0

    def test_default_description_empty(self):
        rule = QuotaRule(dimension="storage_gb", per_operation=0.5)
        assert rule.description == ""

    def test_custom_description(self):
        rule = QuotaRule(
            dimension="credits",
            per_operation=2.0,
            description="每次生成消耗2积分",
        )
        assert rule.description == "每次生成消耗2积分"

    def test_float_per_operation(self):
        rule = QuotaRule(dimension="gb", per_operation=0.25)
        assert rule.per_operation == 0.25


# ============================================================
# ModuleCapability
# ============================================================


class TestModuleCapability:
    """ModuleCapability 能力定义"""

    def test_required_name(self):
        cap = ModuleCapability(name="generate_voice")
        assert cap.name == "generate_voice"

    def test_defaults(self):
        cap = ModuleCapability(name="test_cap")
        assert cap.description == ""
        assert cap.quota_rules == []
        assert cap.metadata == {}

    def test_with_quota_rules(self):
        rules = [QuotaRule(dimension="credits", per_operation=1.0)]
        cap = ModuleCapability(
            name="generate",
            description="生成功能",
            quota_rules=rules,
        )
        assert cap.description == "生成功能"
        assert len(cap.quota_rules) == 1
        assert cap.quota_rules[0].dimension == "credits"

    def test_with_metadata(self):
        cap = ModuleCapability(
            name="export",
            metadata={"format": "mp4", "max_resolution": "1080p"},
        )
        assert cap.metadata["format"] == "mp4"
        assert cap.metadata["max_resolution"] == "1080p"


# ============================================================
# Module
# ============================================================


class TestModuleDefaults:
    """Module 数据类默认值"""

    def test_required_name(self):
        mod = Module(name="ai_voice")
        assert mod.name == "ai_voice"

    def test_default_version(self):
        mod = Module(name="test")
        assert mod.version == "1.0.0"

    def test_default_description(self):
        mod = Module(name="test")
        assert mod.description == ""

    def test_default_capabilities_empty(self):
        mod = Module(name="test")
        assert mod.capabilities == []

    def test_default_dependencies_empty(self):
        mod = Module(name="test")
        assert mod.dependencies == []

    def test_default_status_registered(self):
        mod = Module(name="test")
        assert mod.status == ModuleStatus.REGISTERED

    def test_default_config_empty(self):
        mod = Module(name="test")
        assert mod.config == {}

    def test_full_module(self):
        cap = ModuleCapability(name="do_something")
        mod = Module(
            name="full_module",
            version="2.0.0",
            description="完整模块",
            capabilities=[cap],
            dependencies=["dep1", "dep2"],
            status=ModuleStatus.ACTIVE,
            config={"key": "value"},
        )
        assert mod.version == "2.0.0"
        assert mod.description == "完整模块"
        assert len(mod.capabilities) == 1
        assert mod.dependencies == ["dep1", "dep2"]
        assert mod.status == ModuleStatus.ACTIVE
        assert mod.config["key"] == "value"


class TestModuleActivate:
    """Module.activate 状态转换"""

    def test_activate_from_registered(self):
        mod = Module(name="test")
        mod.activate()
        assert mod.status == ModuleStatus.ACTIVE

    def test_activate_from_disabled(self):
        mod = Module(name="test", status=ModuleStatus.DISABLED)
        mod.activate()
        assert mod.status == ModuleStatus.ACTIVE

    def test_activate_from_error_stays_error(self):
        mod = Module(name="test", status=ModuleStatus.ERROR)
        mod.activate()
        # error 状态不可激活
        assert mod.status == ModuleStatus.ERROR

    def test_activate_already_active(self):
        mod = Module(name="test", status=ModuleStatus.ACTIVE)
        mod.activate()
        assert mod.status == ModuleStatus.ACTIVE


class TestModuleDisable:
    """Module.disable 状态转换"""

    def test_disable_from_registered(self):
        mod = Module(name="test")
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED

    def test_disable_from_active(self):
        mod = Module(name="test", status=ModuleStatus.ACTIVE)
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED

    def test_disable_from_error(self):
        mod = Module(name="test", status=ModuleStatus.ERROR)
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED

    def test_disable_already_disabled(self):
        mod = Module(name="test", status=ModuleStatus.DISABLED)
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED


# ============================================================
# ModuleRegistry - 基础操作
# ============================================================


class TestModuleRegistryBasic:
    """ModuleRegistry 基础操作"""

    def test_empty_registry(self):
        registry = ModuleRegistry()
        assert registry.list_modules() == []
        assert registry.get_active_capabilities() == {}

    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="test_mod"))
        with pytest.raises(ValueError, match="already registered"):
            registry.register(Module(name="test_mod"))

    def test_get_nonexistent_returns_none(self):
        registry = ModuleRegistry()
        assert registry.get("no_such_module") is None

    def test_unregister_success(self):
        registry = ModuleRegistry()
        registry.register(Module(name="test_mod"))
        registry.unregister("test_mod")
        assert registry.get("test_mod") is None

    def test_unregister_nonexistent_raises(self):
        registry = ModuleRegistry()
        with pytest.raises(KeyError, match="not found"):
            registry.unregister("no_such_module")

    def test_unregister_with_dependents_raises(self):
        registry = ModuleRegistry()
        registry.register(Module(name="base_module"))
        registry.register(Module(name="dependent_module", dependencies=["base_module"]))
        with pytest.raises(ValueError, match="depended on by"):
            registry.unregister("base_module")

    def test_clear(self):
        registry = ModuleRegistry()
        registry.register(Module(name="mod1"))
        registry.register(Module(name="mod2"))
        registry.clear()
        assert registry.list_modules() == []


# ============================================================
# ModuleRegistry - 自动激活 & 依赖
# ============================================================


class TestModuleRegistryAutoActivate:
    """注册时自动激活逻辑"""

    def test_no_deps_auto_activates(self):
        registry = ModuleRegistry()
        mod = Module(name="standalone")
        registry.register(mod)
        assert mod.status == ModuleStatus.ACTIVE

    def test_with_deps_all_satisfied_auto_activates(self):
        registry = ModuleRegistry()
        registry.register(Module(name="base"))  # 无依赖，自动激活
        dep_mod = Module(name="dependent", dependencies=["base"])
        registry.register(dep_mod)
        assert dep_mod.status == ModuleStatus.ACTIVE

    def test_with_deps_not_satisfied_stays_registered(self):
        registry = ModuleRegistry()
        mod = Module(name="dependent", dependencies=["missing_dep"])
        registry.register(mod)
        # 依赖不满足，保持 REGISTERED
        assert mod.status == ModuleStatus.REGISTERED

    def test_later_dep_registered_manual_activate(self):
        """先注册依赖模块，再注册被依赖模块时不自动激活前者
        （需要手动或在注册完所有模块后调用 check_dependencies + activate）"""
        registry = ModuleRegistry()
        # 先注册依赖方（依赖未满足，不激活）
        dependent = Module(name="dependent", dependencies=["base"])
        registry.register(dependent)
        assert dependent.status == ModuleStatus.REGISTERED

        # 再注册被依赖方
        base = Module(name="base")
        registry.register(base)
        assert base.status == ModuleStatus.ACTIVE

        # 依赖方仍然是 REGISTERED（不会自动激活）
        assert dependent.status == ModuleStatus.REGISTERED


class TestModuleRegistryCheckDependencies:
    """check_dependencies 依赖检查"""

    def test_module_not_found_returns_false(self):
        registry = ModuleRegistry()
        assert registry.check_dependencies("nonexistent") is False

    def test_no_deps_returns_true(self):
        registry = ModuleRegistry()
        registry.register(Module(name="standalone"))
        assert registry.check_dependencies("standalone") is True

    def test_all_deps_active_returns_true(self):
        registry = ModuleRegistry()
        registry.register(Module(name="dep1"))
        registry.register(Module(name="dep2"))
        registry.register(Module(name="main", dependencies=["dep1", "dep2"]))
        # main 在注册时因依赖满足已自动激活
        assert registry.check_dependencies("main") is True

    def test_dep_not_registered_returns_false(self):
        registry = ModuleRegistry()
        mod = Module(name="main", dependencies=["missing"])
        registry.register(mod)
        assert registry.check_dependencies("main") is False

    def test_dep_registered_but_not_active_returns_false(self):
        registry = ModuleRegistry()
        dep = Module(name="dep", status=ModuleStatus.DISABLED)
        registry.register(dep)
        # 手动设为 disabled（因为 register 时无依赖会自动激活）
        dep.disable()
        main = Module(name="main", dependencies=["dep"])
        registry.register(main)
        # 依赖未激活
        assert registry.check_dependencies("main") is False


# ============================================================
# ModuleRegistry - list_modules & 状态过滤
# ============================================================


class TestModuleRegistryList:
    """list_modules 列表与过滤"""

    def test_list_all(self):
        registry = ModuleRegistry()
        registry.register(Module(name="mod1"))
        registry.register(Module(name="mod2"))
        modules = registry.list_modules()
        assert len(modules) == 2
        names = {m.name for m in modules}
        assert names == {"mod1", "mod2"}

    def test_filter_by_active(self):
        registry = ModuleRegistry()
        registry.register(Module(name="active_mod"))  # 自动激活
        disabled = Module(name="disabled_mod")
        registry.register(disabled)
        disabled.disable()

        active = registry.list_modules(status=ModuleStatus.ACTIVE)
        assert len(active) == 1
        assert active[0].name == "active_mod"

    def test_filter_by_disabled(self):
        registry = ModuleRegistry()
        registry.register(Module(name="active_mod"))
        disabled = Module(name="disabled_mod")
        registry.register(disabled)
        disabled.disable()

        disabled_list = registry.list_modules(status=ModuleStatus.DISABLED)
        assert len(disabled_list) == 1
        assert disabled_list[0].name == "disabled_mod"

    def test_filter_registered(self):
        registry = ModuleRegistry()
        # 有依赖未满足的模块保持 REGISTERED
        mod = Module(name="waiting_mod", dependencies=["missing"])
        registry.register(mod)

        registered = registry.list_modules(status=ModuleStatus.REGISTERED)
        assert len(registered) == 1
        assert registered[0].name == "waiting_mod"


# ============================================================
# ModuleRegistry - 能力发现
# ============================================================


class TestModuleRegistryCapabilities:
    """能力发现：has_capability / get_capability / get_quota_rules"""

    def test_has_capability_true(self):
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="voice_module",
                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="voice_module",
                capabilities=[ModuleCapability(name="generate_voice")],
            )
        )
        assert registry.has_capability("generate_video") is False

    def test_has_capability_inactive_module_not_counted(self):
        registry = ModuleRegistry()
        mod = Module(
            name="inactive_mod",
            capabilities=[ModuleCapability(name="secret_cap")],
        )
        registry.register(mod)
        mod.disable()
        assert registry.has_capability("secret_cap") is False

    def test_get_capability_returns_first_match(self):
        registry = ModuleRegistry()
        cap1 = ModuleCapability(name="export", description="导出1")
        cap2 = ModuleCapability(name="export", description="导出2")
        registry.register(Module(name="mod1", capabilities=[cap1]))
        registry.register(Module(name="mod2", capabilities=[cap2]))

        result = registry.get_capability("export")
        assert result is not None
        assert result.name == "export"
        # 返回第一个匹配的（mod1）
        assert result.description == "导出1"

    def test_get_capability_nonexistent_returns_none(self):
        registry = ModuleRegistry()
        assert registry.get_capability("no_such_cap") is None

    def test_get_quota_rules(self):
        rules = [
            QuotaRule(dimension="credits", per_operation=1.0),
            QuotaRule(dimension="storage", per_operation=0.5),
        ]
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="voice_mod",
                capabilities=[ModuleCapability(name="gen", quota_rules=rules)],
            )
        )
        result = registry.get_quota_rules("gen")
        assert len(result) == 2
        assert result[0].dimension == "credits"
        assert result[1].dimension == "storage"

    def test_get_quota_rules_nonexistent_returns_empty(self):
        registry = ModuleRegistry()
        assert registry.get_quota_rules("no_cap") == []


# ============================================================
# ModuleRegistry - get_active_capabilities
# ============================================================


class TestModuleRegistryActiveCapabilities:
    """get_active_capabilities 已激活能力汇总"""

    def test_empty_registry(self):
        registry = ModuleRegistry()
        assert registry.get_active_capabilities() == {}

    def test_single_module_with_caps(self):
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="voice_mod",
                capabilities=[
                    ModuleCapability(name="generate_voice"),
                    ModuleCapability(name="clone_voice"),
                ],
            )
        )
        result = registry.get_active_capabilities()
        assert "voice_mod" in result
        assert set(result["voice_mod"]) == {"generate_voice", "clone_voice"}

    def test_skips_inactive_modules(self):
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="active_mod",
                capabilities=[ModuleCapability(name="active_cap")],
            )
        )
        inactive = Module(
            name="inactive_mod",
            capabilities=[ModuleCapability(name="inactive_cap")],
        )
        registry.register(inactive)
        inactive.disable()

        result = registry.get_active_capabilities()
        assert "active_mod" in result
        assert "inactive_mod" not in result

    def test_skips_modules_without_caps(self):
        registry = ModuleRegistry()
        registry.register(Module(name="no_cap_mod"))
        result = registry.get_active_capabilities()
        assert "no_cap_mod" not in result

    def test_multiple_modules(self):
        registry = ModuleRegistry()
        registry.register(
            Module(
                name="mod1",
                capabilities=[ModuleCapability(name="cap_a")],
            )
        )
        registry.register(
            Module(
                name="mod2",
                capabilities=[ModuleCapability(name="cap_b"), ModuleCapability(name="cap_c")],
            )
        )
        result = registry.get_active_capabilities()
        assert len(result) == 2
        assert result["mod1"] == ["cap_a"]
        assert set(result["mod2"]) == {"cap_b", "cap_c"}


# ============================================================
# 全局单例
# ============================================================


class TestGlobalSingleton:
    """全局 module_registry 单例"""

    def test_singleton_exists(self):
        assert module_registry is not None
        assert isinstance(module_registry, ModuleRegistry)

    def test_singleton_is_same_instance(self):
        from packages.infrastructure.module_registry import module_registry as mr2

        assert module_registry is mr2