From 6b75e7633d46c79c8eb7a858668f75d5dfebf154 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 24 Jul 2026 12:49:01 +0800 Subject: [PATCH] =?UTF-8?q?test:=20P3-1=20=E7=AC=AC43=E6=B3=A2=E5=8D=95?= =?UTF-8?q?=E5=85=83=E6=B5=8B=E8=AF=95=EF=BC=88ffmpeg=5Futils/ai=5Fclient/?= =?UTF-8?q?config=5Fbase=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_config_base.py | 159 ++++++ tests/unit/test_module_registry.py | 754 ++++++++++++++++++----------- 2 files changed, 642 insertions(+), 271 deletions(-) create mode 100755 tests/unit/test_config_base.py diff --git a/tests/unit/test_config_base.py b/tests/unit/test_config_base.py new file mode 100755 index 000000000..af9291732 --- /dev/null +++ b/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 diff --git a/tests/unit/test_module_registry.py b/tests/unit/test_module_registry.py index 8fbfa91b6..103069d36 100755 --- a/tests/unit/test_module_registry.py +++ b/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