From 45a704a46b54fb93b2a510c902e0fa840d726067 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 24 Jul 2026 13:28:24 +0800 Subject: [PATCH 1/2] =?UTF-8?q?test:=20P3-1=20=E7=AC=AC45=E6=B3=A2?= =?UTF-8?q?=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95=EF=BC=88feature=5Fflag=5Fs?= =?UTF-8?q?tore/schema=5Fguard/sms=5Fservice=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_feature_flag_store.py | 584 ++++++++++---------------- tests/unit/test_schema_guard.py | 87 +++- tests/unit/test_sms_service.py | 178 ++++++++ 3 files changed, 469 insertions(+), 380 deletions(-) mode change 100644 => 100755 tests/unit/test_schema_guard.py create mode 100755 tests/unit/test_sms_service.py diff --git a/tests/unit/test_feature_flag_store.py b/tests/unit/test_feature_flag_store.py index c5f6dd33c..4c8133a88 100755 --- a/tests/unit/test_feature_flag_store.py +++ b/tests/unit/test_feature_flag_store.py @@ -1,425 +1,271 @@ -""" -FeatureFlagStore 单元测试 +"""Feature Flag Store 单元测试""" -覆盖： -- FeatureFlagConfig: to_dict / from_dict 序列化 -- FeatureFlagConfig.is_active: 全局开关/白名单/百分比哈希 -- InMemoryFeatureFlagStore: CRUD / is_active -""" +from __future__ import annotations + +from unittest.mock import MagicMock, patch import pytest +import redis as redis_module from packages.adapters.redis.feature_flag_store import ( FEATURE_FLAG_REDIS_PREFIX, FeatureFlagConfig, + FeatureFlagStore, InMemoryFeatureFlagStore, + RedisFeatureFlagStore, ) -# ============================================================ -# 常量 -# ============================================================ +class TestFeatureFlagConfig: + """FeatureFlagConfig 测试""" -class TestConstants: - """常量验证""" + def test_default_values(self): + cfg = FeatureFlagConfig(name="test_flag") + assert cfg.name == "test_flag" + assert cfg.enabled is False + assert cfg.percentage == 0 + assert cfg.whitelist == set() - def test_redis_prefix(self): - assert FEATURE_FLAG_REDIS_PREFIX == "feature_flag:" - - -# ============================================================ -# FeatureFlagConfig - 默认值 & 基础 -# ============================================================ - - -class TestFeatureFlagConfigDefaults: - """FeatureFlagConfig 默认值""" - - def test_required_name(self): - config = FeatureFlagConfig(name="test_flag") - assert config.name == "test_flag" - - def test_default_disabled(self): - config = FeatureFlagConfig(name="test_flag") - assert config.enabled is False - - def test_default_percentage_zero(self): - config = FeatureFlagConfig(name="test_flag") - assert config.percentage == 0 - - def test_default_whitelist_empty(self): - config = FeatureFlagConfig(name="test_flag") - assert config.whitelist == set() - - def test_full_config(self): - config = FeatureFlagConfig( - name="full_flag", - enabled=True, - percentage=50, + def test_to_dict(self): + cfg = FeatureFlagConfig( + name="test", enabled=True, percentage=50, whitelist={"user1", "user2"}, ) - assert config.name == "full_flag" - assert config.enabled is True - assert config.percentage == 50 - assert config.whitelist == {"user1", "user2"} - - -# ============================================================ -# FeatureFlagConfig - 序列化 -# ============================================================ - - -class TestFeatureFlagConfigSerialization: - """to_dict / from_dict 序列化""" - - def test_to_dict_defaults(self): - config = FeatureFlagConfig(name="test") - d = config.to_dict() - assert d["name"] == "test" - assert d["enabled"] is False - assert d["percentage"] == 0 - assert d["whitelist"] == [] - - def test_to_dict_with_values(self): - config = FeatureFlagConfig( - name="test", - enabled=True, - percentage=75, - whitelist={"a", "b", "c"}, - ) - d = config.to_dict() + d = cfg.to_dict() assert d["name"] == "test" assert d["enabled"] is True - assert d["percentage"] == 75 - # whitelist 排序后输出 - assert sorted(d["whitelist"]) == ["a", "b", "c"] + assert d["percentage"] == 50 + assert set(d["whitelist"]) == {"user1", "user2"} - def test_from_dict_minimal(self): + def test_from_dict(self): + d = {"name": "test", "enabled": True, "percentage": 30, "whitelist": ["u1", "u2"]} + cfg = FeatureFlagConfig.from_dict(d) + assert cfg.name == "test" + assert cfg.enabled is True + assert cfg.percentage == 30 + assert cfg.whitelist == {"u1", "u2"} + + def test_from_dict_defaults(self): d = {"name": "test"} - config = FeatureFlagConfig.from_dict(d) - assert config.name == "test" - assert config.enabled is False - assert config.percentage == 0 - assert config.whitelist == set() + cfg = FeatureFlagConfig.from_dict(d) + assert cfg.enabled is False + assert cfg.percentage == 0 + assert cfg.whitelist == set() - def test_from_dict_full(self): - d = { - "name": "full", - "enabled": True, - "percentage": 30, - "whitelist": ["u1", "u2"], - } - config = FeatureFlagConfig.from_dict(d) - assert config.name == "full" - assert config.enabled is True - assert config.percentage == 30 - assert config.whitelist == {"u1", "u2"} - - def test_round_trip(self): + def test_roundtrip(self): original = FeatureFlagConfig( - name="round_trip", - enabled=True, - percentage=42, - whitelist={"alice", "bob", "charlie"}, + name="test", enabled=True, percentage=75, + whitelist={"a", "b", "c"}, ) - d = original.to_dict() - restored = FeatureFlagConfig.from_dict(d) + restored = FeatureFlagConfig.from_dict(original.to_dict()) assert restored.name == original.name assert restored.enabled == original.enabled assert restored.percentage == original.percentage assert restored.whitelist == original.whitelist - def test_from_dict_coerces_types(self): - """from_dict 应该做类型转换""" - d = { - "name": "coerce", - "enabled": 1, # int → bool - "percentage": "50", # str → int - "whitelist": ("a", "b"), # tuple → set - } - config = FeatureFlagConfig.from_dict(d) - assert config.enabled is True - assert config.percentage == 50 - assert config.whitelist == {"a", "b"} - -# ============================================================ -# FeatureFlagConfig.is_active - 全局开关 -# ============================================================ - - -class TestIsActiveGlobalSwitch: - """is_active - 全局开关基础""" +class TestFeatureFlagConfigIsActive: + """FeatureFlagConfig.is_active 测试""" def test_disabled_returns_false(self): - config = FeatureFlagConfig(name="test", enabled=False) - assert config.is_active() is False + cfg = FeatureFlagConfig(name="t", enabled=False) + assert cfg.is_active() is False + assert cfg.is_active("user1") is False - def test_disabled_with_identifier_returns_false(self): - config = FeatureFlagConfig(name="test", enabled=False) - assert config.is_active(identifier="user1") is False + def test_enabled_no_whitelist_no_percentage(self): + cfg = FeatureFlagConfig(name="t", enabled=True, percentage=0) + assert cfg.is_active() is False - def test_enabled_no_percentage_no_whitelist_returns_true(self): - config = FeatureFlagConfig(name="test", enabled=True) - # percentage=0, whitelist=空，但 enabled=True - # 按逻辑：全局开了但百分比0且无白名单 → 其实应该是 False？ - # 让我看代码... - # 代码里 percentage <= 0 时返回 False（没有白名单且百分比为0） - assert config.is_active() is False + def test_enabled_100_percentage(self): + cfg = FeatureFlagConfig(name="t", enabled=True, percentage=100) + assert cfg.is_active() is True + assert cfg.is_active("any_user") is True - def test_enabled_100_percent_returns_true(self): - config = FeatureFlagConfig(name="test", enabled=True, percentage=100) - assert config.is_active() is True - - -# ============================================================ -# FeatureFlagConfig.is_active - 白名单 -# ============================================================ - - -class TestIsActiveWhitelist: - """is_active - 白名单优先级""" - - def test_whitelist_match_returns_true(self): - config = FeatureFlagConfig( - name="test", - enabled=True, - whitelist={"user1", "user2"}, + def test_whitelist_priority_over_percentage(self): + cfg = FeatureFlagConfig( + name="t", enabled=True, percentage=0, whitelist={"user_vip"}, ) - assert config.is_active(identifier="user1") is True + assert cfg.is_active("user_vip") is True + assert cfg.is_active("other_user") is False - def test_whitelist_no_match_falls_through(self): - config = FeatureFlagConfig( - name="test", - enabled=True, - percentage=0, - whitelist={"user1"}, - ) - # 不在白名单，且百分比为0 → False - assert config.is_active(identifier="user3") is False + def test_whitelist_priority_over_disabled(self): + cfg = FeatureFlagConfig(name="t", enabled=False, whitelist={"user1"}) + assert cfg.is_active("user1") is False - def test_whitelist_overrides_percentage_zero(self): - """白名单优先级最高，即使百分比为0也能启用""" - config = FeatureFlagConfig( - name="test", - enabled=True, - percentage=0, - whitelist={"vip_user"}, - ) - assert config.is_active(identifier="vip_user") is True + def test_percentage_consistent_for_same_user(self): + cfg = FeatureFlagConfig(name="t", enabled=True, percentage=50) + assert cfg.is_active("user_test_123") == cfg.is_active("user_test_123") - def test_whitelist_overrides_partial_percentage(self): - """白名单用户即使在百分比外也能启用""" - config = FeatureFlagConfig( - name="test", - enabled=True, - percentage=1, # 只有1%的用户 - whitelist={"important_user"}, - ) - # 白名单用户直接通过 - assert config.is_active(identifier="important_user") is True + def test_percentage_zero_with_identifier(self): + cfg = FeatureFlagConfig(name="t", enabled=True, percentage=0) + for i in range(10): + assert cfg.is_active(f"user_{i}") is False - def test_no_identifier_no_whitelist_check(self): - """不传 identifier 时不做白名单检查""" - config = FeatureFlagConfig( - name="test", - enabled=True, - percentage=100, - whitelist={"user1"}, - ) - # 无 identifier，直接看百分比（100%） - assert config.is_active() is True - - -# ============================================================ -# FeatureFlagConfig.is_active - 百分比边界值 -# ============================================================ - - -class TestIsActivePercentageBoundaries: - """is_active - 百分比边界值""" - - def test_percentage_0_returns_false(self): - config = FeatureFlagConfig(name="test", enabled=True, percentage=0) - assert config.is_active(identifier="any_user") is False - - def test_percentage_100_returns_true(self): - config = FeatureFlagConfig(name="test", enabled=True, percentage=100) - assert config.is_active(identifier="any_user") is True - - def test_percentage_negative_treated_as_0(self): - """percentage < 0 应该按 0 处理""" - config = FeatureFlagConfig(name="test", enabled=True, percentage=-5) - assert config.is_active(identifier="any_user") is False - - def test_percentage_over_100_treated_as_100(self): - """percentage > 100 应该按 100 处理""" - config = FeatureFlagConfig(name="test", enabled=True, percentage=150) - assert config.is_active(identifier="any_user") is True - - -# ============================================================ -# FeatureFlagConfig.is_active - 哈希一致性 -# ============================================================ - - -class TestIsActiveHashConsistency: - """is_active - 哈希取模一致性验证""" - - def test_same_user_same_result_every_time(self): - """同一用户多次调用结果一致（确定性哈希）""" - config = FeatureFlagConfig(name="test", enabled=True, percentage=50) - results = {config.is_active(identifier="user_xyz") for _ in range(100)} - assert len(results) == 1 # 全部相同 - - def test_different_flags_same_user_can_differ(self): - """不同 flag 对同一用户可以有不同结果（因为 flag name 参与哈希）""" - config_a = FeatureFlagConfig(name="flag_a", enabled=True, percentage=50) - config_b = FeatureFlagConfig(name="flag_b", enabled=True, percentage=50) - # 不保证一定不同，但大部分情况下应该不同 - # 这里只验证哈希输入包含了 flag name（通过机制保证） - # 具体是否不同取决于哈希值 - - def test_percentage_coverage_roughly_correct(self): - """大量用户中，命中比例大致接近百分比""" - config = FeatureFlagConfig(name="coverage_test", enabled=True, percentage=30) - users = [f"user_{i}" for i in range(1000)] - active_count = sum(1 for u in users if config.is_active(identifier=u)) - # 30% ± 10% 的容差 - assert 200 <= active_count <= 400 - - def test_50_percent_roughly_half(self): - config = FeatureFlagConfig(name="half_test", enabled=True, percentage=50) - users = [f"user_{i}" for i in range(1000)] - active_count = sum(1 for u in users if config.is_active(identifier=u)) - # 50% ± 10% - assert 400 <= active_count <= 600 - - def test_10_percent_roughly_tenth(self): - config = FeatureFlagConfig(name="ten_pct", enabled=True, percentage=10) - users = [f"user_{i}" for i in range(1000)] - active_count = sum(1 for u in users if config.is_active(identifier=u)) - assert 50 <= active_count <= 150 - - def test_empty_identifier_treated_as_no_identifier(self): - """空字符串 identifier 应该如何处理？""" - config = FeatureFlagConfig(name="test", enabled=True, percentage=50) - # 空字符串是 falsy，走无 identifier 分支（随机） - # 但白名单检查也会跳过 - # 验证不会崩溃 - result = config.is_active(identifier="") - assert isinstance(result, bool) - - -# ============================================================ -# InMemoryFeatureFlagStore - CRUD -# ============================================================ + def test_percentage_100_with_identifier(self): + cfg = FeatureFlagConfig(name="t", enabled=True, percentage=100) + for i in range(10): + assert cfg.is_active(f"user_{i}") is True class TestInMemoryFeatureFlagStore: - """InMemoryFeatureFlagStore 内存实现""" + """内存实现测试""" - def test_get_nonexistent_returns_default_disabled(self): - store = InMemoryFeatureFlagStore() - config = store.get("nonexistent") - assert config.name == "nonexistent" - assert config.enabled is False - assert config.percentage == 0 + @pytest.fixture + def store(self): + return InMemoryFeatureFlagStore() - def test_set_and_get(self): - store = InMemoryFeatureFlagStore() - original = FeatureFlagConfig( - name="my_flag", - enabled=True, - percentage=50, - whitelist={"admin"}, - ) - store.set(original) - retrieved = store.get("my_flag") - assert retrieved.name == "my_flag" - assert retrieved.enabled is True - assert retrieved.percentage == 50 - assert retrieved.whitelist == {"admin"} + def test_get_nonexistent_returns_default(self, store): + cfg = store.get("nonexistent") + assert cfg.name == "nonexistent" + assert cfg.enabled is False - def test_set_overwrites_existing(self): - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="flag", enabled=True, percentage=30)) - store.set(FeatureFlagConfig(name="flag", enabled=False, percentage=70)) - config = store.get("flag") - assert config.enabled is False - assert config.percentage == 70 + def test_set_and_get(self, store): + store.set(FeatureFlagConfig(name="my_flag", enabled=True, percentage=80)) + cfg = store.get("my_flag") + assert cfg.enabled is True + assert cfg.percentage == 80 - def test_delete_existing_returns_true(self): - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="delete_me")) - result = store.delete("delete_me") - assert result is True - # 删除后获取返回默认配置 - assert store.get("delete_me").enabled is False + def test_delete_existing(self, store): + store.set(FeatureFlagConfig(name="f1")) + assert store.delete("f1") is True + assert store.get("f1").enabled is False - def test_delete_nonexistent_returns_false(self): - store = InMemoryFeatureFlagStore() - result = store.delete("no_such_flag") - assert result is False + def test_delete_nonexistent(self, store): + assert store.delete("nonexistent") is False - def test_list_all_empty(self): - store = InMemoryFeatureFlagStore() + def test_list_all(self, store): + store.set(FeatureFlagConfig(name="a", enabled=True)) + store.set(FeatureFlagConfig(name="b", enabled=False)) + all_flags = store.list_all() + assert len(all_flags) == 2 + assert "a" in all_flags + assert "b" in all_flags + + def test_list_empty(self, store): assert store.list_all() == {} - def test_list_all_multiple(self): - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="flag1", enabled=True)) - store.set(FeatureFlagConfig(name="flag2", percentage=50)) - store.set(FeatureFlagConfig(name="flag3")) + def test_is_active_convenience(self, store): + store.set(FeatureFlagConfig(name="f1", enabled=True, percentage=100)) + assert store.is_active("f1") is True + assert store.is_active("nonexistent") is False - all_flags = store.list_all() - assert len(all_flags) == 3 - assert "flag1" in all_flags - assert "flag2" in all_flags - assert "flag3" in all_flags - assert all_flags["flag1"].enabled is True - assert all_flags["flag2"].percentage == 50 - - def test_list_all_returns_copy(self): - """返回的是副本，修改不影响内部状态""" - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="flag1")) - flags = store.list_all() - flags["fake"] = FeatureFlagConfig(name="fake") - assert "fake" not in store.list_all() + def test_update_overwrites(self, store): + store.set(FeatureFlagConfig(name="f1", enabled=False)) + store.set(FeatureFlagConfig(name="f1", enabled=True, percentage=50)) + cfg = store.get("f1") + assert cfg.enabled is True + assert cfg.percentage == 50 -# ============================================================ -# InMemoryFeatureFlagStore - is_active -# ============================================================ +class TestRedisFeatureFlagStore: + """Redis 实现测试""" + @pytest.fixture + def store(self): + mock_client = MagicMock() + with patch.object(redis_module, "from_url", return_value=mock_client): + s = RedisFeatureFlagStore("redis://localhost:6379/0") + yield s, mock_client -class TestInMemoryStoreIsActive: - """store.is_active 便捷方法""" - - def test_is_active_enabled_flag(self): - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="on", enabled=True, percentage=100)) - assert store.is_active("on") is True - - def test_is_active_disabled_flag(self): - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="off", enabled=False)) - assert store.is_active("off") is False - - def test_is_active_nonexistent_flag(self): - store = InMemoryFeatureFlagStore() - assert store.is_active("unknown") is False - - def test_is_active_with_identifier_whitelist(self): - store = InMemoryFeatureFlagStore() - store.set( - FeatureFlagConfig( - name="beta", - enabled=True, - percentage=0, - whitelist={"tester1"}, + def test_init_creates_redis_client(self): + with patch.object(redis_module, "from_url") as mock_from_url: + mock_from_url.return_value = MagicMock() + RedisFeatureFlagStore("redis://localhost:6379/0") + mock_from_url.assert_called_once_with( + "redis://localhost:6379/0", decode_responses=True ) + + def test_redis_key_prefix(self, store): + s, _ = store + assert s._redis_key("my_flag") == f"{FEATURE_FLAG_REDIS_PREFIX}my_flag" + + def test_get_from_redis(self, store): + s, mock_client = store + mock_client.hgetall.return_value = { + "enabled": "1", + "percentage": "75", + 'whitelist': '["user1", "user2"]', + } + cfg = s.get("test_flag") + assert cfg.enabled is True + assert cfg.percentage == 75 + assert cfg.whitelist == {"user1", "user2"} + mock_client.hgetall.assert_called_once() + + def test_get_missing_returns_default(self, store): + s, mock_client = store + mock_client.hgetall.return_value = {} + cfg = s.get("missing") + assert cfg.enabled is False + assert cfg.percentage == 0 + + def test_get_redis_error_returns_default(self, store): + s, mock_client = store + mock_client.hgetall.side_effect = Exception("Connection refused") + cfg = s.get("test") + assert cfg.enabled is False + assert cfg.name == "test" + + def test_set_writes_to_redis(self, store): + s, mock_client = store + cfg = FeatureFlagConfig( + name="my_flag", enabled=True, percentage=50, + whitelist={"a", "b"}, ) - assert store.is_active("beta", identifier="tester1") is True - assert store.is_active("beta", identifier="other_user") is False + s.set(cfg) + mock_client.hset.assert_called_once() + mapping = mock_client.hset.call_args[1]["mapping"] + assert mapping["enabled"] == "1" + assert mapping["percentage"] == "50" + assert "a" in mapping["whitelist"] + + def test_delete_success(self, store): + s, mock_client = store + mock_client.delete.return_value = 1 + assert s.delete("my_flag") is True + mock_client.delete.assert_called_once() + + def test_delete_not_found(self, store): + s, mock_client = store + mock_client.delete.return_value = 0 + assert s.delete("nonexistent") is False + + def test_invalidate_cache_single(self, store): + s, mock_client = store + mock_client.hgetall.return_value = {"enabled": "1", "percentage": "0"} + s.get("f1") + s.invalidate_cache("f1") + mock_client.hgetall.reset_mock() + s.get("f1") + mock_client.hgetall.assert_called_once() + + def test_invalidate_cache_all(self, store): + s, mock_client = store + mock_client.hgetall.return_value = {"enabled": "1", "percentage": "0"} + s.get("f1") + s.get("f2") + s.invalidate_cache() + mock_client.hgetall.reset_mock() + s.get("f1") + s.get("f2") + assert mock_client.hgetall.call_count == 2 + + def test_parse_whitelist_valid_json(self, store): + s, _ = store + assert s._parse_whitelist('["a", "b", "c"]') == {"a", "b", "c"} + + def test_parse_whitelist_empty(self, store): + s, _ = store + assert s._parse_whitelist("") == set() + assert s._parse_whitelist(None) == set() + + def test_parse_whitelist_invalid_json(self, store): + s, _ = store + assert s._parse_whitelist("not json") == set() + + def test_local_cache_used(self, store): + s, mock_client = store + mock_client.hgetall.return_value = {"enabled": "1", "percentage": "0"} + s.get("f1") + s.get("f1") + mock_client.hgetall.assert_called_once() diff --git a/tests/unit/test_schema_guard.py b/tests/unit/test_schema_guard.py old mode 100644 new mode 100755 index 12545ac14..5bed36fac --- a/tests/unit/test_schema_guard.py +++ b/tests/unit/test_schema_guard.py @@ -1,19 +1,84 @@ +"""Schema Guard 单元测试""" + +from __future__ import annotations + import pytest -from packages.adapters.sqlalchemy_impl.schema_guard import assert_auto_create_schema_allowed +from packages.adapters.sqlalchemy_impl.schema_guard import ( + BLOCKED_AUTO_CREATE_ENVIRONMENTS, + assert_auto_create_schema_allowed, + normalize_environment, +) -@pytest.mark.parametrize("environment", ["staging", "production", " STAGING ", "Production"]) -def test_auto_create_schema_is_forbidden_in_deployed_environments(environment): - with pytest.raises(RuntimeError, match="AUTO_CREATE_SCHEMA is forbidden"): - assert_auto_create_schema_allowed(environment, enabled=True) +class TestNormalizeEnvironment: + """normalize_environment 测试""" + + def test_development(self): + assert normalize_environment("development") == "development" + + def test_staging(self): + assert normalize_environment("staging") == "staging" + + def test_production(self): + assert normalize_environment("production") == "production" + + def test_none_returns_development(self): + assert normalize_environment(None) == "development" + + def test_empty_string_returns_development(self): + assert normalize_environment("") == "development" + + def test_case_insensitive(self): + assert normalize_environment("PRODUCTION") == "production" + assert normalize_environment("Staging") == "staging" + + def test_strips_whitespace(self): + assert normalize_environment(" production ") == "production" -@pytest.mark.parametrize("environment", ["development", "test", "local", ""]) -def test_auto_create_schema_is_allowed_only_for_local_environments(environment): - assert_auto_create_schema_allowed(environment, enabled=True) +class TestAssertAutoCreateSchemaAllowed: + """assert_auto_create_schema_allowed 测试""" + def test_development_enabled_ok(self): + # development 环境允许 auto_create + assert_auto_create_schema_allowed("development", True) -@pytest.mark.parametrize("environment", ["staging", "production"]) -def test_disabled_auto_create_schema_is_allowed_everywhere(environment): - assert_auto_create_schema_allowed(environment, enabled=False) + def test_development_disabled_ok(self): + assert_auto_create_schema_allowed("development", False) + + def test_staging_disabled_ok(self): + # staging 禁用时没问题 + assert_auto_create_schema_allowed("staging", False) + + def test_production_disabled_ok(self): + assert_auto_create_schema_allowed("production", False) + + def test_staging_enabled_raises(self): + with pytest.raises(RuntimeError, match="AUTO_CREATE_SCHEMA"): + assert_auto_create_schema_allowed("staging", True) + + def test_production_enabled_raises(self): + with pytest.raises(RuntimeError, match="AUTO_CREATE_SCHEMA"): + assert_auto_create_schema_allowed("production", True) + + def test_case_insensitive_blocked(self): + with pytest.raises(RuntimeError): + assert_auto_create_schema_allowed("PRODUCTION", True) + with pytest.raises(RuntimeError): + assert_auto_create_schema_allowed("Staging", True) + + def test_none_environment_enabled_ok(self): + # None 视为 development，允许 + assert_auto_create_schema_allowed(None, True) + + def test_custom_env_enabled_ok(self): + # 其他环境不受限制 + assert_auto_create_schema_allowed("test", True) + assert_auto_create_schema_allowed("qa", True) + + def test_blocked_environments_count(self): + # 确认只有 staging 和 production 被阻止 + assert "staging" in BLOCKED_AUTO_CREATE_ENVIRONMENTS + assert "production" in BLOCKED_AUTO_CREATE_ENVIRONMENTS + assert len(BLOCKED_AUTO_CREATE_ENVIRONMENTS) == 2 diff --git a/tests/unit/test_sms_service.py b/tests/unit/test_sms_service.py new file mode 100755 index 000000000..9bc2db0c0 --- /dev/null +++ b/tests/unit/test_sms_service.py @@ -0,0 +1,178 @@ +"""SMS Service 单元测试""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from packages.adapters.sms.sms_service import ( + AliyunSmsService, + NoopSmsService, + get_sms_service, +) + + +class TestNoopSmsService: + """NoopSmsService 测试""" + + def test_send_verification_code_returns_true(self): + svc = NoopSmsService() + assert svc.send_verification_code("13800138000", "123456") is True + + def test_send_template_sms_returns_true(self): + svc = NoopSmsService() + assert svc.send_template_sms( + "13800138000", "SMS_123", {"code": "123456"} + ) is True + + def test_send_verification_code_empty_code(self): + svc = NoopSmsService() + assert svc.send_verification_code("13800138000", "") is True + + +class TestAliyunSmsServiceInit: + """AliyunSmsService 初始化测试""" + + def test_default_values_from_env(self, monkeypatch): + monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_ID", "env_key") + monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_SECRET", "env_secret") + monkeypatch.setenv("ALIYUN_SMS_SIGN_NAME", "env_sign") + monkeypatch.setenv("ALIYUN_SMS_VERIFY_TEMPLATE_ID", "env_tpl") + + svc = AliyunSmsService() + assert svc.access_key_id == "env_key" + assert svc.access_key_secret == "env_secret" + assert svc.sign_name == "env_sign" + assert svc.verify_template_id == "env_tpl" + + def test_explicit_params_override_env(self, monkeypatch): + monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_ID", "env_key") + + svc = AliyunSmsService(access_key_id="explicit_key") + assert svc.access_key_id == "explicit_key" + + def test_default_sign_name(self, monkeypatch): + monkeypatch.delenv("ALIYUN_SMS_SIGN_NAME", raising=False) + svc = AliyunSmsService() + assert svc.sign_name == "小应剪辑" + + def test_default_template_id(self, monkeypatch): + monkeypatch.delenv("ALIYUN_SMS_VERIFY_TEMPLATE_ID", raising=False) + svc = AliyunSmsService() + assert svc.verify_template_id == "SMS_123456789" + + +class TestAliyunSmsServiceSend: + """发送短信测试（mock SDK）""" + + @pytest.fixture + def svc(self): + return AliyunSmsService( + access_key_id="key", + access_key_secret="secret", + sign_name="测试签名", + verify_template_id="SMS_VERIFY", + ) + + def test_send_verification_code_delegates_to_template(self, svc): + """验证码调用 send_template_sms""" + with patch.object(svc, "send_template_sms", return_value=True) as mock_send: + result = svc.send_verification_code("13800138000", "654321") + assert result is True + mock_send.assert_called_once_with( + "13800138000", "SMS_VERIFY", {"code": "654321"} + ) + + def test_send_template_sms_success(self, svc): + """发送成功返回 True""" + mock_body = MagicMock() + mock_body.code = "OK" + mock_body.message = "OK" + mock_response = MagicMock() + mock_response.body = mock_body + + with patch.dict("sys.modules"): + # mock 整个 alibabacloud 模块 + mock_client_cls = MagicMock() + mock_client_cls.return_value.send_sms.return_value = mock_response + + mock_dysms_models = MagicMock() + mock_dysms_models.SendSmsRequest = MagicMock(return_value=MagicMock()) + + mock_openapi_models = MagicMock() + mock_openapi_models.Config = MagicMock() + + with patch.object(svc, "_AliyunSmsService__import_sdk", create=True): + pass + + # 直接 patch 模块名来模拟 SDK 存在 + import sys + sys.modules["alibabacloud_dysmsapi20170525"] = MagicMock() + sys.modules["alibabacloud_dysmsapi20170525.models"] = mock_dysms_models + sys.modules["alibabacloud_dysmsapi20170525.client"] = MagicMock( + Client=mock_client_cls + ) + sys.modules["alibabacloud_tea_openapi"] = MagicMock() + sys.modules["alibabacloud_tea_openapi.models"] = mock_openapi_models + + try: + result = svc.send_template_sms( + "13800138000", "SMS_TPL", {"code": "123"} + ) + assert result is True + finally: + for key in [ + "alibabacloud_dysmsapi20170525", + "alibabacloud_dysmsapi20170525.models", + "alibabacloud_dysmsapi20170525.client", + "alibabacloud_tea_openapi", + "alibabacloud_tea_openapi.models", + ]: + sys.modules.pop(key, None) + + def test_send_template_sms_sdk_not_installed(self, svc): + """SDK 未安装返回 False""" + with patch.object(svc, "send_template_sms"): + pass + # 确保没有 SDK 时返回 False + import sys + saved_modules = {} + for key in list(sys.modules.keys()): + if "alibabacloud" in key: + saved_modules[key] = sys.modules.pop(key) + + try: + result = svc.send_template_sms("13800138000", "tpl", {}) + assert result is False + finally: + sys.modules.update(saved_modules) + + +class TestGetSmsService: + """工厂函数测试""" + + def test_default_noop(self, monkeypatch): + monkeypatch.delenv("SMS_PROVIDER", raising=False) + svc = get_sms_service() + assert isinstance(svc, NoopSmsService) + + def test_noop_provider(self, monkeypatch): + monkeypatch.setenv("SMS_PROVIDER", "noop") + svc = get_sms_service() + assert isinstance(svc, NoopSmsService) + + def test_aliyun_provider(self, monkeypatch): + monkeypatch.setenv("SMS_PROVIDER", "aliyun") + svc = get_sms_service() + assert isinstance(svc, AliyunSmsService) + + def test_case_insensitive_provider(self, monkeypatch): + monkeypatch.setenv("SMS_PROVIDER", "AliYun") + svc = get_sms_service() + assert isinstance(svc, AliyunSmsService) + + def test_unknown_provider_falls_back_to_noop(self, monkeypatch): + monkeypatch.setenv("SMS_PROVIDER", "unknown") + svc = get_sms_service() + assert isinstance(svc, NoopSmsService) -- 2.54.0 From 9d9ed9f37827596390ec5ff202a51a46abc4925c Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 24 Jul 2026 13:34:19 +0800 Subject: [PATCH 2/2] =?UTF-8?q?test:=20P3-1=20=E7=AC=AC45=E6=B3=A2?= =?UTF-8?q?=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95=EF=BC=88schema=5Fguard/sms?= =?UTF-8?q?=5Fservice/job=5Fuse=5Fcases=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_feature_flag_store.py | 590 ++++++++++++++++---------- tests/unit/test_job_use_cases.py | 318 ++++++++++++++ 2 files changed, 690 insertions(+), 218 deletions(-) create mode 100755 tests/unit/test_job_use_cases.py diff --git a/tests/unit/test_feature_flag_store.py b/tests/unit/test_feature_flag_store.py index 4c8133a88..c5f6dd33c 100755 --- a/tests/unit/test_feature_flag_store.py +++ b/tests/unit/test_feature_flag_store.py @@ -1,271 +1,425 @@ -"""Feature Flag Store 单元测试""" +""" +FeatureFlagStore 单元测试 -from __future__ import annotations - -from unittest.mock import MagicMock, patch +覆盖： +- FeatureFlagConfig: to_dict / from_dict 序列化 +- FeatureFlagConfig.is_active: 全局开关/白名单/百分比哈希 +- InMemoryFeatureFlagStore: CRUD / is_active +""" import pytest -import redis as redis_module from packages.adapters.redis.feature_flag_store import ( FEATURE_FLAG_REDIS_PREFIX, FeatureFlagConfig, - FeatureFlagStore, InMemoryFeatureFlagStore, - RedisFeatureFlagStore, ) +# ============================================================ +# 常量 +# ============================================================ -class TestFeatureFlagConfig: - """FeatureFlagConfig 测试""" - def test_default_values(self): - cfg = FeatureFlagConfig(name="test_flag") - assert cfg.name == "test_flag" - assert cfg.enabled is False - assert cfg.percentage == 0 - assert cfg.whitelist == set() +class TestConstants: + """常量验证""" - def test_to_dict(self): - cfg = FeatureFlagConfig( - name="test", enabled=True, percentage=50, + def test_redis_prefix(self): + assert FEATURE_FLAG_REDIS_PREFIX == "feature_flag:" + + +# ============================================================ +# FeatureFlagConfig - 默认值 & 基础 +# ============================================================ + + +class TestFeatureFlagConfigDefaults: + """FeatureFlagConfig 默认值""" + + def test_required_name(self): + config = FeatureFlagConfig(name="test_flag") + assert config.name == "test_flag" + + def test_default_disabled(self): + config = FeatureFlagConfig(name="test_flag") + assert config.enabled is False + + def test_default_percentage_zero(self): + config = FeatureFlagConfig(name="test_flag") + assert config.percentage == 0 + + def test_default_whitelist_empty(self): + config = FeatureFlagConfig(name="test_flag") + assert config.whitelist == set() + + def test_full_config(self): + config = FeatureFlagConfig( + name="full_flag", + enabled=True, + percentage=50, whitelist={"user1", "user2"}, ) - d = cfg.to_dict() + assert config.name == "full_flag" + assert config.enabled is True + assert config.percentage == 50 + assert config.whitelist == {"user1", "user2"} + + +# ============================================================ +# FeatureFlagConfig - 序列化 +# ============================================================ + + +class TestFeatureFlagConfigSerialization: + """to_dict / from_dict 序列化""" + + def test_to_dict_defaults(self): + config = FeatureFlagConfig(name="test") + d = config.to_dict() assert d["name"] == "test" - assert d["enabled"] is True - assert d["percentage"] == 50 - assert set(d["whitelist"]) == {"user1", "user2"} + assert d["enabled"] is False + assert d["percentage"] == 0 + assert d["whitelist"] == [] - def test_from_dict(self): - d = {"name": "test", "enabled": True, "percentage": 30, "whitelist": ["u1", "u2"]} - cfg = FeatureFlagConfig.from_dict(d) - assert cfg.name == "test" - assert cfg.enabled is True - assert cfg.percentage == 30 - assert cfg.whitelist == {"u1", "u2"} - - def test_from_dict_defaults(self): - d = {"name": "test"} - cfg = FeatureFlagConfig.from_dict(d) - assert cfg.enabled is False - assert cfg.percentage == 0 - assert cfg.whitelist == set() - - def test_roundtrip(self): - original = FeatureFlagConfig( - name="test", enabled=True, percentage=75, + def test_to_dict_with_values(self): + config = FeatureFlagConfig( + name="test", + enabled=True, + percentage=75, whitelist={"a", "b", "c"}, ) - restored = FeatureFlagConfig.from_dict(original.to_dict()) + d = config.to_dict() + assert d["name"] == "test" + assert d["enabled"] is True + assert d["percentage"] == 75 + # whitelist 排序后输出 + assert sorted(d["whitelist"]) == ["a", "b", "c"] + + def test_from_dict_minimal(self): + d = {"name": "test"} + config = FeatureFlagConfig.from_dict(d) + assert config.name == "test" + assert config.enabled is False + assert config.percentage == 0 + assert config.whitelist == set() + + def test_from_dict_full(self): + d = { + "name": "full", + "enabled": True, + "percentage": 30, + "whitelist": ["u1", "u2"], + } + config = FeatureFlagConfig.from_dict(d) + assert config.name == "full" + assert config.enabled is True + assert config.percentage == 30 + assert config.whitelist == {"u1", "u2"} + + def test_round_trip(self): + original = FeatureFlagConfig( + name="round_trip", + enabled=True, + percentage=42, + whitelist={"alice", "bob", "charlie"}, + ) + d = original.to_dict() + restored = FeatureFlagConfig.from_dict(d) assert restored.name == original.name assert restored.enabled == original.enabled assert restored.percentage == original.percentage assert restored.whitelist == original.whitelist + def test_from_dict_coerces_types(self): + """from_dict 应该做类型转换""" + d = { + "name": "coerce", + "enabled": 1, # int → bool + "percentage": "50", # str → int + "whitelist": ("a", "b"), # tuple → set + } + config = FeatureFlagConfig.from_dict(d) + assert config.enabled is True + assert config.percentage == 50 + assert config.whitelist == {"a", "b"} -class TestFeatureFlagConfigIsActive: - """FeatureFlagConfig.is_active 测试""" + +# ============================================================ +# FeatureFlagConfig.is_active - 全局开关 +# ============================================================ + + +class TestIsActiveGlobalSwitch: + """is_active - 全局开关基础""" def test_disabled_returns_false(self): - cfg = FeatureFlagConfig(name="t", enabled=False) - assert cfg.is_active() is False - assert cfg.is_active("user1") is False + config = FeatureFlagConfig(name="test", enabled=False) + assert config.is_active() is False - def test_enabled_no_whitelist_no_percentage(self): - cfg = FeatureFlagConfig(name="t", enabled=True, percentage=0) - assert cfg.is_active() is False + def test_disabled_with_identifier_returns_false(self): + config = FeatureFlagConfig(name="test", enabled=False) + assert config.is_active(identifier="user1") is False - def test_enabled_100_percentage(self): - cfg = FeatureFlagConfig(name="t", enabled=True, percentage=100) - assert cfg.is_active() is True - assert cfg.is_active("any_user") is True + def test_enabled_no_percentage_no_whitelist_returns_true(self): + config = FeatureFlagConfig(name="test", enabled=True) + # percentage=0, whitelist=空，但 enabled=True + # 按逻辑：全局开了但百分比0且无白名单 → 其实应该是 False？ + # 让我看代码... + # 代码里 percentage <= 0 时返回 False（没有白名单且百分比为0） + assert config.is_active() is False - def test_whitelist_priority_over_percentage(self): - cfg = FeatureFlagConfig( - name="t", enabled=True, percentage=0, whitelist={"user_vip"}, + def test_enabled_100_percent_returns_true(self): + config = FeatureFlagConfig(name="test", enabled=True, percentage=100) + assert config.is_active() is True + + +# ============================================================ +# FeatureFlagConfig.is_active - 白名单 +# ============================================================ + + +class TestIsActiveWhitelist: + """is_active - 白名单优先级""" + + def test_whitelist_match_returns_true(self): + config = FeatureFlagConfig( + name="test", + enabled=True, + whitelist={"user1", "user2"}, ) - assert cfg.is_active("user_vip") is True - assert cfg.is_active("other_user") is False + assert config.is_active(identifier="user1") is True - def test_whitelist_priority_over_disabled(self): - cfg = FeatureFlagConfig(name="t", enabled=False, whitelist={"user1"}) - assert cfg.is_active("user1") is False + def test_whitelist_no_match_falls_through(self): + config = FeatureFlagConfig( + name="test", + enabled=True, + percentage=0, + whitelist={"user1"}, + ) + # 不在白名单，且百分比为0 → False + assert config.is_active(identifier="user3") is False - def test_percentage_consistent_for_same_user(self): - cfg = FeatureFlagConfig(name="t", enabled=True, percentage=50) - assert cfg.is_active("user_test_123") == cfg.is_active("user_test_123") + def test_whitelist_overrides_percentage_zero(self): + """白名单优先级最高，即使百分比为0也能启用""" + config = FeatureFlagConfig( + name="test", + enabled=True, + percentage=0, + whitelist={"vip_user"}, + ) + assert config.is_active(identifier="vip_user") is True - def test_percentage_zero_with_identifier(self): - cfg = FeatureFlagConfig(name="t", enabled=True, percentage=0) - for i in range(10): - assert cfg.is_active(f"user_{i}") is False + def test_whitelist_overrides_partial_percentage(self): + """白名单用户即使在百分比外也能启用""" + config = FeatureFlagConfig( + name="test", + enabled=True, + percentage=1, # 只有1%的用户 + whitelist={"important_user"}, + ) + # 白名单用户直接通过 + assert config.is_active(identifier="important_user") is True - def test_percentage_100_with_identifier(self): - cfg = FeatureFlagConfig(name="t", enabled=True, percentage=100) - for i in range(10): - assert cfg.is_active(f"user_{i}") is True + def test_no_identifier_no_whitelist_check(self): + """不传 identifier 时不做白名单检查""" + config = FeatureFlagConfig( + name="test", + enabled=True, + percentage=100, + whitelist={"user1"}, + ) + # 无 identifier，直接看百分比（100%） + assert config.is_active() is True + + +# ============================================================ +# FeatureFlagConfig.is_active - 百分比边界值 +# ============================================================ + + +class TestIsActivePercentageBoundaries: + """is_active - 百分比边界值""" + + def test_percentage_0_returns_false(self): + config = FeatureFlagConfig(name="test", enabled=True, percentage=0) + assert config.is_active(identifier="any_user") is False + + def test_percentage_100_returns_true(self): + config = FeatureFlagConfig(name="test", enabled=True, percentage=100) + assert config.is_active(identifier="any_user") is True + + def test_percentage_negative_treated_as_0(self): + """percentage < 0 应该按 0 处理""" + config = FeatureFlagConfig(name="test", enabled=True, percentage=-5) + assert config.is_active(identifier="any_user") is False + + def test_percentage_over_100_treated_as_100(self): + """percentage > 100 应该按 100 处理""" + config = FeatureFlagConfig(name="test", enabled=True, percentage=150) + assert config.is_active(identifier="any_user") is True + + +# ============================================================ +# FeatureFlagConfig.is_active - 哈希一致性 +# ============================================================ + + +class TestIsActiveHashConsistency: + """is_active - 哈希取模一致性验证""" + + def test_same_user_same_result_every_time(self): + """同一用户多次调用结果一致（确定性哈希）""" + config = FeatureFlagConfig(name="test", enabled=True, percentage=50) + results = {config.is_active(identifier="user_xyz") for _ in range(100)} + assert len(results) == 1 # 全部相同 + + def test_different_flags_same_user_can_differ(self): + """不同 flag 对同一用户可以有不同结果（因为 flag name 参与哈希）""" + config_a = FeatureFlagConfig(name="flag_a", enabled=True, percentage=50) + config_b = FeatureFlagConfig(name="flag_b", enabled=True, percentage=50) + # 不保证一定不同，但大部分情况下应该不同 + # 这里只验证哈希输入包含了 flag name（通过机制保证） + # 具体是否不同取决于哈希值 + + def test_percentage_coverage_roughly_correct(self): + """大量用户中，命中比例大致接近百分比""" + config = FeatureFlagConfig(name="coverage_test", enabled=True, percentage=30) + users = [f"user_{i}" for i in range(1000)] + active_count = sum(1 for u in users if config.is_active(identifier=u)) + # 30% ± 10% 的容差 + assert 200 <= active_count <= 400 + + def test_50_percent_roughly_half(self): + config = FeatureFlagConfig(name="half_test", enabled=True, percentage=50) + users = [f"user_{i}" for i in range(1000)] + active_count = sum(1 for u in users if config.is_active(identifier=u)) + # 50% ± 10% + assert 400 <= active_count <= 600 + + def test_10_percent_roughly_tenth(self): + config = FeatureFlagConfig(name="ten_pct", enabled=True, percentage=10) + users = [f"user_{i}" for i in range(1000)] + active_count = sum(1 for u in users if config.is_active(identifier=u)) + assert 50 <= active_count <= 150 + + def test_empty_identifier_treated_as_no_identifier(self): + """空字符串 identifier 应该如何处理？""" + config = FeatureFlagConfig(name="test", enabled=True, percentage=50) + # 空字符串是 falsy，走无 identifier 分支（随机） + # 但白名单检查也会跳过 + # 验证不会崩溃 + result = config.is_active(identifier="") + assert isinstance(result, bool) + + +# ============================================================ +# InMemoryFeatureFlagStore - CRUD +# ============================================================ class TestInMemoryFeatureFlagStore: - """内存实现测试""" + """InMemoryFeatureFlagStore 内存实现""" - @pytest.fixture - def store(self): - return InMemoryFeatureFlagStore() + def test_get_nonexistent_returns_default_disabled(self): + store = InMemoryFeatureFlagStore() + config = store.get("nonexistent") + assert config.name == "nonexistent" + assert config.enabled is False + assert config.percentage == 0 - def test_get_nonexistent_returns_default(self, store): - cfg = store.get("nonexistent") - assert cfg.name == "nonexistent" - assert cfg.enabled is False + def test_set_and_get(self): + store = InMemoryFeatureFlagStore() + original = FeatureFlagConfig( + name="my_flag", + enabled=True, + percentage=50, + whitelist={"admin"}, + ) + store.set(original) + retrieved = store.get("my_flag") + assert retrieved.name == "my_flag" + assert retrieved.enabled is True + assert retrieved.percentage == 50 + assert retrieved.whitelist == {"admin"} - def test_set_and_get(self, store): - store.set(FeatureFlagConfig(name="my_flag", enabled=True, percentage=80)) - cfg = store.get("my_flag") - assert cfg.enabled is True - assert cfg.percentage == 80 + def test_set_overwrites_existing(self): + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="flag", enabled=True, percentage=30)) + store.set(FeatureFlagConfig(name="flag", enabled=False, percentage=70)) + config = store.get("flag") + assert config.enabled is False + assert config.percentage == 70 - def test_delete_existing(self, store): - store.set(FeatureFlagConfig(name="f1")) - assert store.delete("f1") is True - assert store.get("f1").enabled is False + def test_delete_existing_returns_true(self): + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="delete_me")) + result = store.delete("delete_me") + assert result is True + # 删除后获取返回默认配置 + assert store.get("delete_me").enabled is False - def test_delete_nonexistent(self, store): - assert store.delete("nonexistent") is False + def test_delete_nonexistent_returns_false(self): + store = InMemoryFeatureFlagStore() + result = store.delete("no_such_flag") + assert result is False - def test_list_all(self, store): - store.set(FeatureFlagConfig(name="a", enabled=True)) - store.set(FeatureFlagConfig(name="b", enabled=False)) - all_flags = store.list_all() - assert len(all_flags) == 2 - assert "a" in all_flags - assert "b" in all_flags - - def test_list_empty(self, store): + def test_list_all_empty(self): + store = InMemoryFeatureFlagStore() assert store.list_all() == {} - def test_is_active_convenience(self, store): - store.set(FeatureFlagConfig(name="f1", enabled=True, percentage=100)) - assert store.is_active("f1") is True - assert store.is_active("nonexistent") is False + def test_list_all_multiple(self): + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="flag1", enabled=True)) + store.set(FeatureFlagConfig(name="flag2", percentage=50)) + store.set(FeatureFlagConfig(name="flag3")) - def test_update_overwrites(self, store): - store.set(FeatureFlagConfig(name="f1", enabled=False)) - store.set(FeatureFlagConfig(name="f1", enabled=True, percentage=50)) - cfg = store.get("f1") - assert cfg.enabled is True - assert cfg.percentage == 50 + all_flags = store.list_all() + assert len(all_flags) == 3 + assert "flag1" in all_flags + assert "flag2" in all_flags + assert "flag3" in all_flags + assert all_flags["flag1"].enabled is True + assert all_flags["flag2"].percentage == 50 + + def test_list_all_returns_copy(self): + """返回的是副本，修改不影响内部状态""" + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="flag1")) + flags = store.list_all() + flags["fake"] = FeatureFlagConfig(name="fake") + assert "fake" not in store.list_all() -class TestRedisFeatureFlagStore: - """Redis 实现测试""" +# ============================================================ +# InMemoryFeatureFlagStore - is_active +# ============================================================ - @pytest.fixture - def store(self): - mock_client = MagicMock() - with patch.object(redis_module, "from_url", return_value=mock_client): - s = RedisFeatureFlagStore("redis://localhost:6379/0") - yield s, mock_client - def test_init_creates_redis_client(self): - with patch.object(redis_module, "from_url") as mock_from_url: - mock_from_url.return_value = MagicMock() - RedisFeatureFlagStore("redis://localhost:6379/0") - mock_from_url.assert_called_once_with( - "redis://localhost:6379/0", decode_responses=True +class TestInMemoryStoreIsActive: + """store.is_active 便捷方法""" + + def test_is_active_enabled_flag(self): + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="on", enabled=True, percentage=100)) + assert store.is_active("on") is True + + def test_is_active_disabled_flag(self): + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="off", enabled=False)) + assert store.is_active("off") is False + + def test_is_active_nonexistent_flag(self): + store = InMemoryFeatureFlagStore() + assert store.is_active("unknown") is False + + def test_is_active_with_identifier_whitelist(self): + store = InMemoryFeatureFlagStore() + store.set( + FeatureFlagConfig( + name="beta", + enabled=True, + percentage=0, + whitelist={"tester1"}, ) - - def test_redis_key_prefix(self, store): - s, _ = store - assert s._redis_key("my_flag") == f"{FEATURE_FLAG_REDIS_PREFIX}my_flag" - - def test_get_from_redis(self, store): - s, mock_client = store - mock_client.hgetall.return_value = { - "enabled": "1", - "percentage": "75", - 'whitelist': '["user1", "user2"]', - } - cfg = s.get("test_flag") - assert cfg.enabled is True - assert cfg.percentage == 75 - assert cfg.whitelist == {"user1", "user2"} - mock_client.hgetall.assert_called_once() - - def test_get_missing_returns_default(self, store): - s, mock_client = store - mock_client.hgetall.return_value = {} - cfg = s.get("missing") - assert cfg.enabled is False - assert cfg.percentage == 0 - - def test_get_redis_error_returns_default(self, store): - s, mock_client = store - mock_client.hgetall.side_effect = Exception("Connection refused") - cfg = s.get("test") - assert cfg.enabled is False - assert cfg.name == "test" - - def test_set_writes_to_redis(self, store): - s, mock_client = store - cfg = FeatureFlagConfig( - name="my_flag", enabled=True, percentage=50, - whitelist={"a", "b"}, ) - s.set(cfg) - mock_client.hset.assert_called_once() - mapping = mock_client.hset.call_args[1]["mapping"] - assert mapping["enabled"] == "1" - assert mapping["percentage"] == "50" - assert "a" in mapping["whitelist"] - - def test_delete_success(self, store): - s, mock_client = store - mock_client.delete.return_value = 1 - assert s.delete("my_flag") is True - mock_client.delete.assert_called_once() - - def test_delete_not_found(self, store): - s, mock_client = store - mock_client.delete.return_value = 0 - assert s.delete("nonexistent") is False - - def test_invalidate_cache_single(self, store): - s, mock_client = store - mock_client.hgetall.return_value = {"enabled": "1", "percentage": "0"} - s.get("f1") - s.invalidate_cache("f1") - mock_client.hgetall.reset_mock() - s.get("f1") - mock_client.hgetall.assert_called_once() - - def test_invalidate_cache_all(self, store): - s, mock_client = store - mock_client.hgetall.return_value = {"enabled": "1", "percentage": "0"} - s.get("f1") - s.get("f2") - s.invalidate_cache() - mock_client.hgetall.reset_mock() - s.get("f1") - s.get("f2") - assert mock_client.hgetall.call_count == 2 - - def test_parse_whitelist_valid_json(self, store): - s, _ = store - assert s._parse_whitelist('["a", "b", "c"]') == {"a", "b", "c"} - - def test_parse_whitelist_empty(self, store): - s, _ = store - assert s._parse_whitelist("") == set() - assert s._parse_whitelist(None) == set() - - def test_parse_whitelist_invalid_json(self, store): - s, _ = store - assert s._parse_whitelist("not json") == set() - - def test_local_cache_used(self, store): - s, mock_client = store - mock_client.hgetall.return_value = {"enabled": "1", "percentage": "0"} - s.get("f1") - s.get("f1") - mock_client.hgetall.assert_called_once() + assert store.is_active("beta", identifier="tester1") is True + assert store.is_active("beta", identifier="other_user") is False diff --git a/tests/unit/test_job_use_cases.py b/tests/unit/test_job_use_cases.py new file mode 100755 index 000000000..bbe51e7f4 --- /dev/null +++ b/tests/unit/test_job_use_cases.py @@ -0,0 +1,318 @@ +"""Job Use Cases 单元测试""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from packages.application.jobs import ( + CancelJobUseCase, + CompleteJobCommand, + CompleteJobUseCase, + CreateJobCommand, + CreateJobUseCase, + FailJobCommand, + FailJobUseCase, + GetJobStatisticsUseCase, + GetJobUseCase, + ListJobsUseCase, + RetryJobUseCase, + SubmitJobUseCase, + UpdateJobProgressCommand, + UpdateJobProgressUseCase, +) +from packages.domain.job import Job, JobStatus, JobType + + +@pytest.fixture +def mock_repo(): + return MagicMock() + + +@pytest.fixture +def sample_job(): + return Job.create( + project_id="proj_001", + job_type=JobType.VIDEO_COMPOSE, + payload={"template_id": "tpl_001"}, + source_id="src_001", + created_by_user_id="user_001", + max_retries=3, + ) + + +class TestCreateJobCommand: + """CreateJobCommand 测试""" + + def test_default_values(self): + cmd = CreateJobCommand(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + assert cmd.project_id == "p1" + assert cmd.payload == {} + assert cmd.source_id == "" + assert cmd.created_by_user_id == "" + assert cmd.max_retries == 3 + + +class TestCreateJobUseCase: + """CreateJobUseCase 测试""" + + def test_create_success(self, mock_repo, sample_job): + mock_repo.create.return_value = sample_job + use_case = CreateJobUseCase(mock_repo) + cmd = CreateJobCommand( + project_id="proj_001", + job_type=JobType.VIDEO_COMPOSE, + payload={"template_id": "tpl_001"}, + source_id="src_001", + created_by_user_id="user_001", + max_retries=3, + ) + result = use_case.execute(cmd) + assert result.status == JobStatus.PENDING + assert result.project_id == "proj_001" + mock_repo.create.assert_called_once() + + def test_create_with_string_job_type(self, mock_repo): + mock_repo.create.side_effect = lambda x: x + use_case = CreateJobUseCase(mock_repo) + cmd = CreateJobCommand(project_id="p1", job_type="video_compose") + result = use_case.execute(cmd) + assert result.job_type == JobType.VIDEO_COMPOSE + + +class TestSubmitJobUseCase: + """SubmitJobUseCase 测试""" + + def test_submit_success(self, mock_repo, sample_job): + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = SubmitJobUseCase(mock_repo) + result = use_case.execute(sample_job.id, celery_task_id="celery_123") + assert result.status == JobStatus.RUNNING + assert result.celery_task_id == "celery_123" + + def test_submit_not_found(self, mock_repo): + mock_repo.get.return_value = None + use_case = SubmitJobUseCase(mock_repo) + with pytest.raises(ValueError, match="任务不存在"): + use_case.execute("nonexistent") + + def test_submit_wrong_status(self, mock_repo, sample_job): + sample_job.status = JobStatus.RUNNING + mock_repo.get.return_value = sample_job + use_case = SubmitJobUseCase(mock_repo) + with pytest.raises(ValueError, match="只有 pending"): + use_case.execute(sample_job.id) + + +class TestUpdateJobProgressUseCase: + """UpdateJobProgressUseCase 测试""" + + def test_update_progress_success(self, mock_repo, sample_job): + sample_job.status = JobStatus.RUNNING + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = UpdateJobProgressUseCase(mock_repo) + cmd = UpdateJobProgressCommand( + job_id=sample_job.id, progress=50.0, current_stage="渲染中" + ) + result = use_case.execute(cmd) + assert result.progress == 50.0 + assert result.current_stage == "渲染中" + + def test_update_progress_not_found(self, mock_repo): + mock_repo.get.return_value = None + use_case = UpdateJobProgressUseCase(mock_repo) + with pytest.raises(ValueError, match="任务不存在"): + use_case.execute(UpdateJobProgressCommand(job_id="x", progress=10)) + + def test_update_progress_wrong_status(self, mock_repo, sample_job): + sample_job.status = JobStatus.PENDING + mock_repo.get.return_value = sample_job + use_case = UpdateJobProgressUseCase(mock_repo) + with pytest.raises(ValueError, match="只有 running"): + use_case.execute(UpdateJobProgressCommand(job_id=sample_job.id, progress=10)) + + +class TestCompleteJobUseCase: + """CompleteJobUseCase 测试""" + + def test_complete_from_running(self, mock_repo, sample_job): + sample_job.status = JobStatus.RUNNING + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = CompleteJobUseCase(mock_repo) + cmd = CompleteJobCommand(job_id=sample_job.id, result={"url": "http://..."}) + result = use_case.execute(cmd) + assert result.status == JobStatus.SUCCESS + assert result.result["url"] == "http://..." + + def test_complete_from_pending(self, mock_repo, sample_job): + sample_job.status = JobStatus.PENDING + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = CompleteJobUseCase(mock_repo) + result = use_case.execute(CompleteJobCommand(job_id=sample_job.id)) + assert result.status == JobStatus.SUCCESS + + def test_complete_not_found(self, mock_repo): + mock_repo.get.return_value = None + use_case = CompleteJobUseCase(mock_repo) + with pytest.raises(ValueError, match="任务不存在"): + use_case.execute(CompleteJobCommand(job_id="x")) + + def test_complete_failed_status_raises(self, mock_repo, sample_job): + sample_job.status = JobStatus.FAILED + mock_repo.get.return_value = sample_job + use_case = CompleteJobUseCase(mock_repo) + with pytest.raises(ValueError, match="只有 running/pending"): + use_case.execute(CompleteJobCommand(job_id=sample_job.id)) + + +class TestFailJobUseCase: + """FailJobUseCase 测试""" + + def test_fail_success(self, mock_repo, sample_job): + sample_job.status = JobStatus.RUNNING + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = FailJobUseCase(mock_repo) + cmd = FailJobCommand(job_id=sample_job.id, error_message="渲染失败") + result = use_case.execute(cmd) + assert result.status == JobStatus.FAILED + assert "渲染失败" in result.error_message + + def test_fail_not_found(self, mock_repo): + mock_repo.get.return_value = None + use_case = FailJobUseCase(mock_repo) + with pytest.raises(ValueError, match="任务不存在"): + use_case.execute(FailJobCommand(job_id="x", error_message="err")) + + def test_fail_updates_error_message(self, mock_repo, sample_job): + sample_job.status = JobStatus.RUNNING + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = FailJobUseCase(mock_repo) + result = use_case.execute(FailJobCommand(job_id=sample_job.id, error_message="连接超时")) + assert result.error_message == "连接超时" + + +class TestRetryJobUseCase: + """RetryJobUseCase 测试""" + + def test_retry_success(self, mock_repo, sample_job): + sample_job.status = JobStatus.FAILED + sample_job.retry_count = 1 + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = RetryJobUseCase(mock_repo) + result = use_case.execute(sample_job.id) + assert result.status == JobStatus.PENDING + assert result.retry_count == 2 + + def test_retry_not_found(self, mock_repo): + mock_repo.get.return_value = None + use_case = RetryJobUseCase(mock_repo) + with pytest.raises(ValueError, match="任务不存在"): + use_case.execute("nonexistent") + + +class TestCancelJobUseCase: + """CancelJobUseCase 测试""" + + def test_cancel_pending(self, mock_repo, sample_job): + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = CancelJobUseCase(mock_repo) + result = use_case.execute(sample_job.id) + assert result.status == JobStatus.CANCELLED + + def test_cancel_running(self, mock_repo, sample_job): + sample_job.status = JobStatus.RUNNING + mock_repo.get.return_value = sample_job + mock_repo.update.side_effect = lambda x: x + use_case = CancelJobUseCase(mock_repo) + result = use_case.execute(sample_job.id) + assert result.status == JobStatus.CANCELLED + + def test_cancel_terminal_raises(self, mock_repo, sample_job): + sample_job.status = JobStatus.SUCCESS + mock_repo.get.return_value = sample_job + use_case = CancelJobUseCase(mock_repo) + with pytest.raises(ValueError, match="终态"): + use_case.execute(sample_job.id) + + def test_cancel_not_found(self, mock_repo): + mock_repo.get.return_value = None + use_case = CancelJobUseCase(mock_repo) + with pytest.raises(ValueError, match="任务不存在"): + use_case.execute("nonexistent") + + +class TestGetJobUseCase: + """GetJobUseCase 测试""" + + def test_get_exists(self, mock_repo, sample_job): + mock_repo.get.return_value = sample_job + use_case = GetJobUseCase(mock_repo) + result = use_case.execute(sample_job.id) + assert result.id == sample_job.id + + def test_get_not_found(self, mock_repo): + mock_repo.get.return_value = None + use_case = GetJobUseCase(mock_repo) + result = use_case.execute("nonexistent") + assert result is None + + +class TestListJobsUseCase: + """ListJobsUseCase 测试""" + + def test_list_by_project(self, mock_repo, sample_job): + mock_repo.list_by_project.return_value = [sample_job] + use_case = ListJobsUseCase(mock_repo) + result = use_case.execute(project_id="proj_001") + assert len(result) == 1 + mock_repo.list_by_project.assert_called_once() + + def test_list_by_user(self, mock_repo, sample_job): + mock_repo.list_by_user.return_value = [sample_job] + use_case = ListJobsUseCase(mock_repo) + result = use_case.execute(user_id="user_001") + assert len(result) == 1 + mock_repo.list_by_user.assert_called_once() + + def test_list_no_filter_raises(self, mock_repo): + use_case = ListJobsUseCase(mock_repo) + with pytest.raises(ValueError, match="必须指定"): + use_case.execute() + + def test_list_with_filters(self, mock_repo, sample_job): + mock_repo.list_by_project.return_value = [sample_job] + use_case = ListJobsUseCase(mock_repo) + use_case.execute( + project_id="p1", + job_type=JobType.VIDEO_COMPOSE, + status=JobStatus.RUNNING, + limit=20, + offset=10, + ) + mock_repo.list_by_project.assert_called_once_with( + "p1", job_type=JobType.VIDEO_COMPOSE, status=JobStatus.RUNNING, limit=20, offset=10 + ) + + +class TestGetJobStatisticsUseCase: + """GetJobStatisticsUseCase 测试""" + + def test_statistics(self, mock_repo): + mock_repo.count_by_project.side_effect = [10, 2, 3, 4, 1] + use_case = GetJobStatisticsUseCase(mock_repo) + stats = use_case.execute("proj_001") + assert stats["project_id"] == "proj_001" + assert stats["total"] == 10 + assert stats["pending"] == 2 + assert stats["running"] == 3 + assert stats["success"] == 4 + assert stats["failed"] == 1 -- 2.54.0