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"


# ── Module 更多状态转换测试 ──────────────────────────────────────────────────


class TestModuleStateTransitions:
    """Module 状态转换补充测试."""

    def test_activate_twice_idempotent(self):
        """多次激活不报错."""
        mod = Module(name="m1")
        mod.activate()
        mod.activate()
        assert mod.status == ModuleStatus.ACTIVE

    def test_disable_twice_idempotent(self):
        """多次禁用不报错."""
        mod = Module(name="m1", status=ModuleStatus.ACTIVE)
        mod.disable()
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED

    def test_disable_from_registered(self):
        """从registered状态禁用."""
        mod = Module(name="m1")
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED

    def test_activate_after_disable(self):
        """禁用后重新激活."""
        mod = Module(name="m1")
        mod.activate()
        mod.disable()
        mod.activate()
        assert mod.status == ModuleStatus.ACTIVE

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

    def test_error_state_can_be_disabled(self):
        """error状态可以被禁用（disable 不检查状态）."""
        mod = Module(name="m1", status=ModuleStatus.ERROR)
        mod.disable()
        assert mod.status == ModuleStatus.DISABLED


# ── ModuleRegistry 注册补充测试 ──────────────────────────────────────────────


class TestModuleRegistryRegisterMore:
    """模块注册补充场景."""

    def test_register_multiple_modules(self):
        """注册多个模块."""
        registry = ModuleRegistry()
        for i in range(5):
            registry.register(Module(name=f"mod_{i}"))
        assert len(registry.list_modules()) == 5

    def test_register_order_independent_deps(self):
        """先注册依赖方，后注册被依赖方，依赖方不会自动激活."""
        registry = ModuleRegistry()
        registry.register(Module(name="plugin", dependencies=["core"]))
        assert registry.get("plugin").status == ModuleStatus.REGISTERED
        # 注册core后，plugin仍然是REGISTERED（不会自动检查）
        registry.register(Module(name="core"))
        assert registry.get("core").status == ModuleStatus.ACTIVE

    def test_register_with_multiple_deps_all_satisfied(self):
        """所有依赖都满足时自动激活."""
        registry = ModuleRegistry()
        registry.register(Module(name="dep_a"))
        registry.register(Module(name="dep_b"))
        registry.register(Module(name="plugin", dependencies=["dep_a", "dep_b"]))
        assert registry.get("plugin").status == ModuleStatus.ACTIVE

    def test_register_with_multiple_deps_partial_missing(self):
        """部分依赖缺失时不激活."""
        registry = ModuleRegistry()
        registry.register(Module(name="dep_a"))
        registry.register(Module(name="plugin", dependencies=["dep_a", "dep_b"]))
        assert registry.get("plugin").status == ModuleStatus.REGISTERED

    def test_register_self_dependency_handled(self):
        """自依赖不会导致死循环（依赖检查时找不到自己）."""
        registry = ModuleRegistry()
        registry.register(Module(name="self_dep", dependencies=["self_dep"]))
        # 注册时自己还没加入 _modules，检查依赖时找不到，保持REGISTERED
        assert registry.get("self_dep").status == ModuleStatus.REGISTERED


# ── ModuleRegistry 能力查询补充 ─────────────────────────────────────────────


class TestModuleRegistryCapabilitiesMore:
    """能力查询补充测试."""

    def test_multiple_modules_same_capability_returns_first(self):
        """多个模块提供同一能力，get_capability返回第一个."""
        registry = ModuleRegistry()
        cap1 = ModuleCapability(name="render", description="渲染器A")
        cap2 = ModuleCapability(name="render", description="渲染器B")
        registry.register(Module(name="mod_a", capabilities=[cap1]))
        registry.register(Module(name="mod_b", capabilities=[cap2]))
        result = registry.get_capability("render")
        assert result is not None
        assert result.name == "render"

    def test_has_capability_case_sensitive(self):
        """能力名大小写敏感."""
        registry = ModuleRegistry()
        registry.register(Module(name="m1", capabilities=[ModuleCapability(name="Render")]))
        assert registry.has_capability("Render") is True
        assert registry.has_capability("render") is False

    def test_get_quota_rules_nonexistent_capability(self):
        """不存在的能力返回空配额列表."""
        registry = ModuleRegistry()
        rules = registry.get_quota_rules("nonexistent")
        assert rules == []

    def test_get_active_capabilities_empty_registry(self):
        """空注册中心返回空字典."""
        registry = ModuleRegistry()
        result = registry.get_active_capabilities()
        assert result == {}

    def test_get_active_capabilities_skips_inactive(self):
        """非激活模块的能力不计入."""
        registry = ModuleRegistry()
        registry.register(Module(name="active_mod", capabilities=[ModuleCapability(name="cap1")]))
        disabled = Module(
            name="disabled_mod",
            status=ModuleStatus.DISABLED,
            capabilities=[ModuleCapability(name="cap2")],
        )
        registry._modules["disabled_mod"] = disabled
        result = registry.get_active_capabilities()
        assert "active_mod" in result
        assert "disabled_mod" not in result

    def test_get_active_capabilities_skips_no_cap_modules(self):
        """无能力的模块不出现在结果中."""
        registry = ModuleRegistry()
        registry.register(Module(name="no_cap_mod"))
        registry.register(Module(name="has_cap_mod", capabilities=[ModuleCapability(name="cap1")]))
        result = registry.get_active_capabilities()
        assert "no_cap_mod" not in result
        assert "has_cap_mod" in result

    def test_module_with_multiple_capabilities(self):
        """单个模块有多个能力."""
        registry = ModuleRegistry()
        caps = [
            ModuleCapability(name="cap_a"),
            ModuleCapability(name="cap_b"),
            ModuleCapability(name="cap_c"),
        ]
        registry.register(Module(name="multi_mod", capabilities=caps))
        assert registry.has_capability("cap_a")
        assert registry.has_capability("cap_b")
        assert registry.has_capability("cap_c")
        assert len(registry.get_active_capabilities()["multi_mod"]) == 3


# ── ModuleRegistry 依赖检查补充 ────────────────────────────────────────────


class TestModuleRegistryDependenciesMore:
    """依赖检查补充测试."""

    def test_multiple_dependencies_all_active(self):
        """多个依赖都激活."""
        registry = ModuleRegistry()
        registry.register(Module(name="dep1"))
        registry.register(Module(name="dep2"))
        registry.register(Module(name="dep3"))
        registry.register(Module(name="plugin", dependencies=["dep1", "dep2", "dep3"]))
        assert registry.check_dependencies("plugin") is True

    def test_multiple_dependencies_one_inactive(self):
        """多个依赖中有一个未激活."""
        registry = ModuleRegistry()
        registry.register(Module(name="dep1"))
        dep2 = Module(name="dep2", status=ModuleStatus.DISABLED)
        registry._modules["dep2"] = dep2
        registry.register(Module(name="plugin", dependencies=["dep1", "dep2"]))
        # 注册时dep2不是ACTIVE，plugin不会自动激活
        assert registry.check_dependencies("plugin") is False

    def test_chain_dependencies(self):
        """链式依赖 A→B→C."""
        registry = ModuleRegistry()
        registry.register(Module(name="c"))
        registry.register(Module(name="b", dependencies=["c"]))
        registry.register(Module(name="a", dependencies=["b"]))
        # a 依赖 b（ACTIVE），b 依赖 c（ACTIVE）
        # check_dependencies 只检查直接依赖，b 是 ACTIVE 的
        assert registry.check_dependencies("a") is True
        assert registry.get("a").status == ModuleStatus.ACTIVE

    def test_no_dependencies_always_satisfied(self):
        """无依赖的模块总是满足依赖检查."""
        registry = ModuleRegistry()
        registry.register(Module(name="standalone"))
        assert registry.check_dependencies("standalone") is True


# ── ModuleRegistry list_modules 补充 ──────────────────────────────────────


class TestModuleRegistryListMore:
    """list_modules 补充测试."""

    def test_list_modules_empty(self):
        """空注册中心."""
        registry = ModuleRegistry()
        assert registry.list_modules() == []

    def test_list_modules_registered_status(self):
        """按registered状态过滤."""
        registry = ModuleRegistry()
        registry.register(Module(name="active_mod"))  # auto ACTIVE
        pending = Module(name="pending_mod")
        registry._modules["pending_mod"] = pending  # REGISTERED
        registered = registry.list_modules(status=ModuleStatus.REGISTERED)
        assert len(registered) == 1
        assert registered[0].name == "pending_mod"

    def test_list_modules_error_status(self):
        """按error状态过滤."""
        registry = ModuleRegistry()
        error_mod = Module(name="err", status=ModuleStatus.ERROR)
        registry._modules["err"] = error_mod
        errors = registry.list_modules(status=ModuleStatus.ERROR)
        assert len(errors) == 1
        assert errors[0].name == "err"


# ── QuotaRule 补充测试 ──────────────────────────────────────────────────────


class TestQuotaRuleMore:
    """QuotaRule 补充测试."""

    def test_zero_per_operation(self):
        """零消耗配额规则."""
        rule = QuotaRule(dimension="free_ops", per_operation=0.0)
        assert rule.per_operation == 0.0

    def test_fractional_per_operation(self):
        """小数消耗配额规则."""
        rule = QuotaRule(dimension="storage", per_operation=0.001)
        assert rule.per_operation == 0.001

    def test_large_per_operation(self):
        """大数值消耗."""
        rule = QuotaRule(dimension="tokens", per_operation=10000.0)
        assert rule.per_operation == 10000.0


# ── ModuleCapability 补充测试 ──────────────────────────────────────────────


class TestModuleCapabilityMore:
    """ModuleCapability 补充测试."""

    def test_empty_metadata(self):
        """默认metadata为空字典."""
        cap = ModuleCapability(name="test")
        assert cap.metadata == {}

    def test_metadata_preserved(self):
        """元数据完整保存."""
        meta = {"model": "v1", "speed": 1.5, "enabled": True}
        cap = ModuleCapability(name="test", metadata=meta)
        assert cap.metadata["model"] == "v1"
        assert cap.metadata["speed"] == 1.5
        assert cap.metadata["enabled"] is True

    def test_multiple_quota_rules(self):
        """多个配额规则."""
        rules = [
            QuotaRule("dim1", 1.0),
            QuotaRule("dim2", 2.0),
            QuotaRule("dim3", 3.0),
        ]
        cap = ModuleCapability(name="test", quota_rules=rules)
        assert len(cap.quota_rules) == 3
        assert cap.quota_rules[0].dimension == "dim1"
        assert cap.quota_rules[2].per_operation == 3.0