test: P3-1 第43波单元测试（ffmpeg_utils/ai_client/config_base）

tests/unit/test_config_base.py

@@ -0,0 +1,159 @@
+"""Config Base 单元测试"""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.config.base import (
+    SharedSettings,
+    get_cached_settings,
+    get_shared_settings,
+    reload_settings_cache,
+)
+
+
+class TestSharedSettingsDefaults:
+    """SharedSettings 默认值测试"""
+
+    @pytest.fixture(autouse=True)
+    def clean_env(self, monkeypatch):
+        """清除所有可能影响的环境变量，确保测的是代码默认值"""
+        env_vars = [
+            "ENVIRONMENT", "DEBUG", "AUTO_CREATE_SCHEMA",
+            "DATABASE_URL", "DATABASE_POOL_SIZE", "DATABASE_MAX_OVERFLOW",
+            "DATABASE_POOL_TIMEOUT", "DATABASE_POOL_RECYCLE",
+            "REDIS_URL", "CELERY_BROKER_URL", "CELERY_RESULT_BACKEND",
+            "OSS_ENDPOINT", "OSS_ACCESS_KEY_ID", "OSS_ACCESS_KEY_SECRET",
+            "OSS_BUCKET_NAME", "OSS_DIRECT_UPLOAD_MAX_MB", "OSS_DIRECT_UPLOAD_EXPIRE_SECONDS",
+            "COSYVOICE_API_KEY", "COSYVOICE_BASE_URL", "COSYVOICE_MODEL",
+            "COSYVOICE_VOICE", "COSYVOICE_SAMPLE_RATE", "COSYVOICE_FORMAT",
+            "COSYVOICE_CLONE_MODEL",
+            "DOUBAO_API_KEY", "DOUBAO_MODEL", "DOUBAO_BASE_URL",
+            "DOUBAO_TIMEOUT", "DOUBAO_MAX_RETRIES",
+        ]
+        for var in env_vars:
+            monkeypatch.delenv(var, raising=False)
+        reload_settings_cache()
+        yield
+        reload_settings_cache()
+
+    def _make_settings(self):
+        """构造不读 env 文件的纯净 settings"""
+        return SharedSettings(_env_file='/dev/null')
+
+    def test_default_environment(self):
+        """默认环境为 development"""
+        s = self._make_settings()
+        assert s.environment == "development"
+
+    def test_default_debug(self):
+        """默认开启 debug"""
+        s = self._make_settings()
+        assert s.debug is True
+
+    def test_default_database_config(self):
+        """数据库默认配置"""
+        s = self._make_settings()
+        assert "postgresql" in s.database_url
+        assert s.database_pool_size == 20
+        assert s.database_max_overflow == 10
+        assert s.database_pool_timeout == 30
+        assert s.database_pool_recycle == 3600
+
+    def test_default_redis_config(self):
+        """Redis 默认配置"""
+        s = self._make_settings()
+        assert s.redis_url.startswith("redis://")
+
+    def test_default_celery_config(self):
+        """Celery 默认配置"""
+        s = self._make_settings()
+        assert s.celery_broker_url.startswith("redis://")
+        assert s.celery_result_backend.startswith("redis://")
+
+    def test_default_oss_config(self):
+        """OSS 默认配置"""
+        s = self._make_settings()
+        assert s.oss_endpoint.endswith("aliyuncs.com")
+        assert s.oss_bucket_name == "xiaoxia-autocut"
+        assert s.oss_direct_upload_max_mb == 2000
+        assert s.oss_direct_upload_expire_seconds == 900
+
+    def test_default_cosyvoice_config(self):
+        """CosyVoice 默认配置"""
+        s = self._make_settings()
+        assert s.cosyvoice_model == "cosyvoice-v3-flash"
+        assert s.cosyvoice_sample_rate == 22050
+        assert s.cosyvoice_format == "mp3"
+        assert s.cosyvoice_clone_model == "voice-enrollment"
+
+    def test_default_doubao_config(self):
+        """豆包默认配置"""
+        s = self._make_settings()
+        assert s.doubao_timeout == 30
+        assert s.doubao_max_retries == 2
+        assert "volces.com" in s.doubao_base_url
+
+    def test_default_empty_api_keys(self):
+        """API Key 默认空字符串"""
+        s = self._make_settings()
+        assert s.oss_access_key_id == ""
+        assert s.oss_access_key_secret == ""
+        assert s.cosyvoice_api_key == ""
+        assert s.doubao_api_key == ""
+
+    def test_auto_create_schema_default(self):
+        """auto_create_schema 默认 False"""
+        s = self._make_settings()
+        assert s.auto_create_schema is False
+
+
+class TestSettingsSingleton:
+    """单例管理测试"""
+
+    def setup_method(self):
+        """每个测试前清空缓存"""
+        reload_settings_cache()
+
+    def teardown_method(self):
+        """每个测试后清空缓存"""
+        reload_settings_cache()
+
+    def test_get_cached_settings_same_instance(self):
+        """同一类两次调用返回同一实例"""
+        s1 = get_cached_settings(SharedSettings)
+        s2 = get_cached_settings(SharedSettings)
+        assert s1 is s2
+
+    def test_get_shared_settings_returns_shared_settings(self):
+        """get_shared_settings 返回 SharedSettings 实例"""
+        s = get_shared_settings()
+        assert isinstance(s, SharedSettings)
+
+    def test_get_shared_settings_singleton(self):
+        """get_shared_settings 是单例"""
+        s1 = get_shared_settings()
+        s2 = get_shared_settings()
+        assert s1 is s2
+
+    def test_reload_settings_cache_clears(self):
+        """reload 后获取新实例"""
+        s1 = get_cached_settings(SharedSettings)
+        reload_settings_cache()
+        s2 = get_cached_settings(SharedSettings)
+        assert s1 is not s2
+
+    def test_custom_cache_key(self):
+        """自定义 cache_key 分开缓存"""
+        s1 = get_cached_settings(SharedSettings, cache_key="key_a")
+        s2 = get_cached_settings(SharedSettings, cache_key="key_b")
+        assert s1 is not s2
+        # 但值相同
+        assert s1.database_url == s2.database_url
+
+    def test_different_classes_separate_cache(self):
+        """不同类使用不同缓存"""
+        from packages.config.api_settings import APISettings
+        shared = get_shared_settings()
+        api = get_cached_settings(APISettings)
+        assert shared is not api

tests/unit/test_module_registry.py

@@ -1,6 +1,12 @@
-"""Module Registry 单元测试"""
+"""
+Module Registry 模块注册中心单元测试
 
-from __future__ import annotations
+覆盖：
+- ModuleStatus 枚举
+- QuotaRule / ModuleCapability / Module 数据类
+- Module.activate / disable 状态转换
+- ModuleRegistry 注册/注销/查询/能力发现/依赖检查
+"""
 
 import pytest
 
@@ -10,348 +16,554 @@ from packages.infrastructure.module_registry import (
     ModuleRegistry,
     ModuleStatus,
     QuotaRule,
+    module_registry,
 )
 
-
-@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"},
-            ),
-        ],
-    )
+# ============================================================
+# ModuleStatus
+# ============================================================
 
 
 class TestModuleStatus:
-    """ModuleStatus 枚举测试"""
+    """ModuleStatus 枚举"""
 
-    def test_status_values(self):
-        """所有状态值存在"""
+    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_default_status(self):
-        """模块默认状态为 REGISTERED"""
-        mod = Module(name="test")
-        assert mod.status == ModuleStatus.REGISTERED
+    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
 
 
-class TestModule:
-    """Module 数据类测试"""
+# ============================================================
+# QuotaRule
+# ============================================================
 
-    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
+class TestQuotaRule:
+    """QuotaRule 配额规则"""
 
-    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_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_disable(self):
-        """禁用模块"""
-        mod = Module(name="test", status=ModuleStatus.ACTIVE)
-        mod.disable()
-        assert mod.status == ModuleStatus.DISABLED
+    def test_default_description_empty(self):
+        rule = QuotaRule(dimension="storage_gb", per_operation=0.5)
+        assert rule.description == ""
 
-    def test_disable_from_registered(self):
-        """从 registered 状态禁用"""
-        mod = Module(name="test")
-        mod.disable()
-        assert mod.status == ModuleStatus.DISABLED
+    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 测试"""
+    """ModuleCapability 能力定义"""
 
-    def test_default_values(self):
-        """默认值"""
-        cap = ModuleCapability(name="do_something")
-        assert cap.name == "do_something"
+    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"
 
-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 == ""
+    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"
 
 
-class TestModuleRegistryRegister:
-    """注册模块测试"""
+# ============================================================
+# Module
+# ============================================================
 
-    def test_register_success(self, registry, sample_module):
-        """注册成功"""
-        registry.register(sample_module)
-        mod = registry.get("ai_voice")
-        assert mod is not None
+
+class TestModuleDefaults:
+    """Module 数据类默认值"""
+
+    def test_required_name(self):
+        mod = Module(name="ai_voice")
         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_default_version(self):
+        mod = Module(name="test")
+        assert mod.version == "1.0.0"
 
-    def test_register_with_dependency_not_met(self, registry):
-        """依赖未满足时注册，状态为 REGISTERED"""
-        mod = Module(name="dep_mod", dependencies=["base_module"])
+    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)
-        # 依赖不存在，不激活
-        result = registry.get("dep_mod")
-        assert result.status == ModuleStatus.REGISTERED
+        assert registry.get("test_mod") is mod
 
-    def test_register_with_dependency_met(self, registry):
-        """依赖已满足时注册，自动激活"""
-        base = Module(name="base_module")
-        dependent = Module(name="dependent_mod", dependencies=["base_module"])
+    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"))
 
-        registry.register(base)
-        registry.register(dependent)
+    def test_get_nonexistent_returns_none(self):
+        registry = ModuleRegistry()
+        assert registry.get("no_such_module") is None
 
-        result = registry.get("dependent_mod")
-        assert result.status == ModuleStatus.ACTIVE
+    def test_unregister_success(self):
+        registry = ModuleRegistry()
+        registry.register(Module(name="test_mod"))
+        registry.unregister("test_mod")
+        assert registry.get("test_mod") is None
 
-
-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"""
+    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):
-        """有其他模块依赖时不能注销"""
-        base = Module(name="base_mod")
-        dependent = Module(name="dependent_mod", dependencies=["base_mod"])
-
-        registry.register(base)
-        registry.register(dependent)
+            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_mod")
+            registry.unregister("base_module")
 
-
-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):
-        """空注册中心"""
+    def test_clear(self):
+        registry = ModuleRegistry()
+        registry.register(Module(name="mod1"))
+        registry.register(Module(name="mod2"))
+        registry.clear()
         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 == []
+# ============================================================
+# ModuleRegistry - 自动激活 & 依赖
+# ============================================================
 
 
-class TestModuleRegistryDependencies:
-    """依赖检查测试"""
+class TestModuleRegistryAutoActivate:
+    """注册时自动激活逻辑"""
 
-    def test_check_dependencies_met(self, registry):
-        """依赖满足返回 True"""
+    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")
-        dep = Module(name="dep", dependencies=["base"])
-
         registry.register(base)
-        registry.register(dep)
+        assert base.status == ModuleStatus.ACTIVE
 
-        assert registry.check_dependencies("dep") is True
+        # 依赖方仍然是 REGISTERED（不会自动激活）
+        assert dependent.status == ModuleStatus.REGISTERED
 
-    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
+class TestModuleRegistryCheckDependencies:
+    """check_dependencies 依赖检查"""
 
-    def test_check_dependencies_module_not_found(self, registry):
-        """模块不存在返回 False"""
+    def test_module_not_found_returns_false(self):
+        registry = ModuleRegistry()
         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"])
+    def test_no_deps_returns_true(self):
+        registry = ModuleRegistry()
+        registry.register(Module(name="standalone"))
+        assert registry.check_dependencies("standalone") is True
 
-        registry._modules["base"] = base
-        registry._modules["dep"] = dep
+    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
 
-        assert registry.check_dependencies("dep") is False
+    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_get_active_capabilities(self, registry, sample_module):
-        """获取所有已激活模块的能力"""
-        registry.register(sample_module)
+    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="video_edit",
+                name="voice_mod",
                 capabilities=[
-                    ModuleCapability(name="cut_video"),
-                    ModuleCapability(name="add_effect"),
+                    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"}
 
-        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")],
+    def test_skips_inactive_modules(self):
+        registry = ModuleRegistry()
+        registry.register(
+            Module(
+                name="active_mod",
+                capabilities=[ModuleCapability(name="active_cap")],
+            )
         )
-        registry._modules["test_mod"] = mod
+        inactive = Module(
+            name="inactive_mod",
+            capabilities=[ModuleCapability(name="inactive_cap")],
+        )
+        registry.register(inactive)
+        inactive.disable()
 
-        caps = registry.get_active_capabilities()
-        assert "test_mod" not in caps
+        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 TestModuleRegistryClear:
-    """清空注册中心测试"""
+# ============================================================
+# 全局单例
+# ============================================================
 
-    def test_clear(self, registry, sample_module):
-        """清空所有注册"""
-        registry.register(sample_module)
-        registry.register(Module(name="second"))
 
-        assert len(registry.list_modules()) == 2
+class TestGlobalSingleton:
+    """全局 module_registry 单例"""
 
-        registry.clear()
-        assert len(registry.list_modules()) == 0
+    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