"""config.py 配置聚合测试(设计 §8): 多源命名、键优先级、缺失报错。""" import pytest from polygateway.config import GatewaySettings _BASE_ENV = { "LLM__QWEN__1__BASE_URL": "https://gw-a.example/v1", "LLM__QWEN__1__API_KEY": "sk-a", "LLM__QWEN__1__MODEL": "qwen-max", "LLM__QWEN__1__TIMEOUT_S": "120", "LLM_MAX_RETRIES": "3", "LLM_RETRY_BASE_DELAY": "2.0", "LLM_RETRY_MAX_DELAY": "30.0", "LLM_CIRCUIT_BREAKER_THRESHOLD": "5", "LLM_CIRCUIT_BREAKER_COOLDOWN": "60", "PGW_CACHE_BACKEND": "none", "PGW_TELEMETRY_BACKEND": "none", } def _env(**overrides): env = dict(_BASE_ENV) env.update({k: v for k, v in overrides.items() if v is not None}) for k, v in overrides.items(): if v is None: env.pop(k, None) return env class TestSourceAggregation: def test_single_source_parsed(self): s = GatewaySettings.from_env("LLM", env=_env()) assert len(s.sources) == 1 src = s.sources[0] assert src.name == "qwen_1" and src.provider == "qwen" assert src.base_url == "https://gw-a.example/v1" and src.timeout_s == 120.0 def test_multi_source_and_optional_fields(self): env = _env( **{ "LLM__DEEPSEEK__2__BASE_URL": "https://gw-b.example/v1", "LLM__DEEPSEEK__2__API_KEY": "sk-b", "LLM__DEEPSEEK__2__MODEL": "deepseek-chat", "LLM__DEEPSEEK__2__TIMEOUT_S": "90", "LLM__DEEPSEEK__2__RPM": "60", "LLM__DEEPSEEK__2__ENABLE_THINKING": "true", "LLM__DEEPSEEK__2__MISSING_DONE": "salvage", } ) s = GatewaySettings.from_env("LLM", env=env) by_name = {src.name: src for src in s.sources} assert set(by_name) == {"qwen_1", "deepseek_2"} ds = by_name["deepseek_2"] assert ds.rpm == 60 and ds.enable_thinking is True and ds.missing_done == "salvage" assert by_name["qwen_1"].enable_thinking is None # 未配置 = 三态 None def test_other_scope_keys_ignored(self): env = _env( **{ "OCR__MONKEY__1__BASE_URL": "http://lan/parse", "OCR__MONKEY__1__API_KEY": "x", "OCR__MONKEY__1__MODEL": "monkey", "OCR__MONKEY__1__TIMEOUT_S": "60", } ) s = GatewaySettings.from_env("LLM", env=env) assert len(s.sources) == 1 def test_flat_timeout_is_source_default(self): env = _env(LLM_TIMEOUT="300", **{"LLM__QWEN__1__TIMEOUT_S": None}) s = GatewaySettings.from_env("LLM", env=env) assert s.sources[0].timeout_s == 300.0 @pytest.mark.parametrize("missing", ["BASE_URL", "API_KEY", "MODEL"]) def test_missing_required_source_field_fails(self, missing): with pytest.raises(ValueError, match=missing): GatewaySettings.from_env("LLM", env=_env(**{f"LLM__QWEN__1__{missing}": None})) def test_unknown_field_fails_loudly(self): with pytest.raises(ValueError, match="TEMPRATURE"): GatewaySettings.from_env("LLM", env=_env(**{"LLM__QWEN__1__TEMPRATURE": "0.7"})) def test_no_sources_fails(self): env = {k: v for k, v in _BASE_ENV.items() if not k.startswith("LLM__")} with pytest.raises(ValueError, match="源"): GatewaySettings.from_env("LLM", env=env) class TestResilienceKeys: def test_flat_legacy_keys(self): s = GatewaySettings.from_env("LLM", env=_env()) assert s.retry.max_attempts == 3 and s.retry.backoff_base_s == 2.0 assert s.breaker.fail_threshold == 5 and s.breaker.cooldown_s == 60.0 def test_scope_keys_override_flat(self): env = _env(**{"LLM__RETRY__MAX_ATTEMPTS": "7", "LLM__BREAKER__COOLDOWN_S": "15"}) s = GatewaySettings.from_env("LLM", env=env) assert s.retry.max_attempts == 7 assert s.breaker.cooldown_s == 15.0 assert s.breaker.fail_threshold == 5 # 未覆盖的仍取平铺键 def test_missing_retry_config_fails(self): with pytest.raises(ValueError, match="MAX_RETRIES|MAX_ATTEMPTS"): GatewaySettings.from_env("LLM", env=_env(LLM_MAX_RETRIES=None)) def test_probe_ttl_derived_when_absent(self): s = GatewaySettings.from_env("LLM", env=_env()) # 派生规则(M2 补第三项): max(2 × 最大源 timeout, cooldown, 最大源 timeout + 5) assert s.breaker.probe_ttl_s == max(2 * 120.0, 60.0, 120.0 + 5) s2 = GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "300"})) assert s2.breaker.probe_ttl_s == 300.0 def test_probe_ttl_derivation_third_term_wins(self): # timeout=4、cooldown=2 → max(8, 2, 9) = 9;守卫 9 ≥ 4+5 恰好成立不报错 env = _env( LLM_CIRCUIT_BREAKER_COOLDOWN="2", **{"LLM__QWEN__1__TIMEOUT_S": "4"}, ) s = GatewaySettings.from_env("LLM", env=env) assert s.breaker.probe_ttl_s == 9.0 def test_explicit_probe_ttl_below_guard_rejected(self): # 守卫: probe_ttl_s ≥ max(timeout_s) + 5(CHS container 语义,M2 设计 §3) with pytest.raises(ValueError, match="probe"): GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"})) def test_selector_and_quota_full(self): # M2.5: 缺省选源改 health_aware(生产级默认);显式配置者不变 s = GatewaySettings.from_env("LLM", env=_env()) assert s.selector == "health_aware" and s.quota_full == "wait" assert ( GatewaySettings.from_env("LLM", env=_env(**{"LLM__SELECTOR": "round_robin"})).selector == "round_robin" ) s2 = GatewaySettings.from_env( "LLM", env=_env(**{"LLM__SELECTOR": "least_inflight", "LLM__QUOTA_FULL": "fail_fast"}) ) assert s2.selector == "least_inflight" and s2.quota_full == "fail_fast" with pytest.raises(ValueError): GatewaySettings.from_env("LLM", env=_env(**{"LLM__SELECTOR": "random"})) def test_global_limits(self): env = _env(**{"LLM__GLOBAL__MAX_CONCURRENCY": "8", "LLM__GLOBAL__RPM": "120"}) s = GatewaySettings.from_env("LLM", env=env) assert s.global_limits.max_concurrency == 8 and s.global_limits.rpm == 120 assert s.global_limits.tpm == 0 class TestAssemblyGuards: def test_structured_retries_default_two(self): # M2.5 迭代 4: 缺省重问 1→2(instructor 缺省 3 的保守版;P6 阶梯死亡 3.2% 实证) s = GatewaySettings.from_env("LLM", env=_env()) assert s.structured_max_retries == 2 s2 = GatewaySettings.from_env("LLM", env=_env(PGW_STRUCTURED_MAX_RETRIES="0")) assert s2.structured_max_retries == 0 def test_cache_requires_namespace_and_ttl(self): env = _env(PGW_CACHE_BACKEND="memory") with pytest.raises(ValueError, match="NAMESPACE"): GatewaySettings.from_env("LLM", env=env) env2 = _env(PGW_CACHE_BACKEND="memory", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="0") with pytest.raises(ValueError, match="TTL"): GatewaySettings.from_env("LLM", env=env2) env3 = _env(PGW_CACHE_BACKEND="memory", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="3600") s = GatewaySettings.from_env("LLM", env=env3) assert s.cache_namespace == "proj" and s.cache_ttl_s == 3600 def test_redis_cache_requires_url(self): env = _env(PGW_CACHE_BACKEND="redis", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="3600") with pytest.raises(ValueError, match="REDIS_URL"): GatewaySettings.from_env("LLM", env=env) def test_sqlite_telemetry_requires_path(self): env = _env(PGW_TELEMETRY_BACKEND="sqlite") with pytest.raises(ValueError, match="SQLITE_PATH"): GatewaySettings.from_env("LLM", env=env) def test_timeout_must_fit_lease_ttl(self): env = _env(PGW_LEASE_TTL_S="60", **{"LLM__QWEN__1__TIMEOUT_S": "120"}) with pytest.raises(ValueError, match="租约|lease"): GatewaySettings.from_env("LLM", env=env) def test_effective_breaker_threshold_auto_raised(self): env = _env(**{"LLM__QWEN__1__MAX_CONCURRENCY": "8"}) s = GatewaySettings.from_env("LLM", env=env) # 有效阈值 = max(配置值 5, 源级并发 8 × 2) = 16(M2.5: 抬升只看源级) assert s.breaker.fail_threshold == 16 def test_breaker_threshold_not_raised_by_global_concurrency(self): # M2.5 病灶 2 回归: 全局并发不再抬升阈值(M2 曾 max(5, 100×2)=200 使熔断失灵) env = _env(**{"LLM__GLOBAL__MAX_CONCURRENCY": "100", "LLM__GLOBAL__RPM": "600"}) s = GatewaySettings.from_env("LLM", env=env) assert s.breaker.fail_threshold == 5 def test_breaker_rate_channel_defaults_and_overrides(self): # M2.5 失败率通道参数: 库缺省(韧性参数缺省先例)与显式覆盖 s = GatewaySettings.from_env("LLM", env=_env()) assert s.breaker.min_calls == 10 assert s.breaker.fail_rate == pytest.approx(0.6) assert s.breaker.window_s == pytest.approx(60.0) assert s.breaker.max_cooldown_s == pytest.approx(300.0) # max(300, cooldown 60) env = _env( **{ "LLM__BREAKER__MIN_CALLS": "20", "LLM__BREAKER__FAIL_RATE": "0.5", "LLM__BREAKER__WINDOW_S": "30", "LLM__BREAKER__MAX_COOLDOWN_S": "600", "LLM_CIRCUIT_BREAKER_COOLDOWN": "400", } ) s2 = GatewaySettings.from_env("LLM", env=env) assert s2.breaker.min_calls == 20 and s2.breaker.fail_rate == pytest.approx(0.5) assert s2.breaker.window_s == pytest.approx(30.0) assert s2.breaker.max_cooldown_s == pytest.approx(600.0) def test_breaker_max_cooldown_floor_follows_cooldown(self): # 缺省封顶 = max(300, cooldown): 冷却 400s 时封顶随之 400s env = _env(**{"LLM_CIRCUIT_BREAKER_COOLDOWN": "400"}) s = GatewaySettings.from_env("LLM", env=env) assert s.breaker.max_cooldown_s == pytest.approx(400.0) def test_redis_governance_backend_requires_url(self): """M2 解禁 redis 后端: 取 redis 时 REDIS_URL 必在,缺则装配报错。""" env = _env( PGW_LIMITER_BACKEND="redis", PGW_BREAKER_BACKEND="redis", REDIS_URL="redis://:pw@10.0.0.1:6379/3", ) s = GatewaySettings.from_env("LLM", env=env) assert s.limiter_backend == "redis" and s.breaker_backend == "redis" with pytest.raises(ValueError, match="REDIS_URL"): GatewaySettings.from_env("LLM", env=_env(PGW_LIMITER_BACKEND="redis")) def test_postgres_telemetry_requires_dsn_and_strips_driver_suffix(self): """M2: telemetry=postgres 需 PGW_TELEMETRY_PG_DSN;SQLAlchemy 风格驱动后缀剥离。""" with pytest.raises(ValueError, match="PG_DSN"): GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_BACKEND="postgres")) env = _env( PGW_TELEMETRY_BACKEND="postgres", PGW_TELEMETRY_PG_DSN="postgresql+psycopg://u:p@h:5432/polygateway", ) s = GatewaySettings.from_env("LLM", env=env) assert s.telemetry_pg_dsn == "postgresql://u:p@h:5432/polygateway" env2 = _env( PGW_TELEMETRY_BACKEND="postgres", PGW_TELEMETRY_PG_DSN="postgresql+asyncpg://u:p@h:5432/polygateway", ) assert ( GatewaySettings.from_env("LLM", env=env2).telemetry_pg_dsn == "postgresql://u:p@h:5432/polygateway" ) def test_telemetry_backend_whitelist(self): with pytest.raises(ValueError, match="TELEMETRY_BACKEND"): GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_BACKEND="mysql")) def test_pricing_path_optional(self): assert GatewaySettings.from_env("LLM", env=_env()).pricing_path is None s = GatewaySettings.from_env("LLM", env=_env(PGW_PRICING_PATH="conf/prices.json")) assert s.pricing_path == "conf/prices.json" def test_stall_window_must_cover_ttft(self): """守卫: stall_window_s ≥ 最大 ttft_timeout_s(防误判卡死,ARCH §7.3)。""" env = _env( **{ "LLM__QWEN__1__TTFT_TIMEOUT_S": "30", "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15", "LLM__BACKPRESSURE__STALL_WINDOW_S": "20", } ) with pytest.raises(ValueError, match="stall"): GatewaySettings.from_env("LLM", env=env) env_ok = dict(env) env_ok["LLM__BACKPRESSURE__STALL_WINDOW_S"] = "60" assert GatewaySettings.from_env("LLM", env=env_ok).backpressure.stall_window_s == 60.0 class TestOcrSettings: """M3 OcrSettings(设计 §3.4): 复用 GatewaySettings,无 OCR 专用键。""" _OCR_ENV = { "OCR__MONKEY__1__BASE_URL": "http://10.77.0.20:7866", "OCR__MONKEY__1__API_KEY": "none", # 无鉴权占位惯例 "OCR__MONKEY__1__MODEL": "monkey-ocr", "OCR__MONKEY__1__TIMEOUT_S": "120", "LLM_MAX_RETRIES": "3", "LLM_RETRY_BASE_DELAY": "2.0", "LLM_RETRY_MAX_DELAY": "30.0", "LLM_CIRCUIT_BREAKER_THRESHOLD": "5", "LLM_CIRCUIT_BREAKER_COOLDOWN": "60", "PGW_CACHE_BACKEND": "none", "PGW_TELEMETRY_BACKEND": "none", } def test_minimal_ocr_scope(self): from polygateway.config import OcrSettings settings = OcrSettings.from_env("OCR", env=dict(self._OCR_ENV)) gw = settings.gateway assert gw.scope == "ocr" # GatewaySettings 统一小写归一(既有约定) assert gw.sources[0].name == "monkey_1" assert gw.sources[0].provider == "monkey" assert gw.sources[0].trust_env is True def test_scope_resilience_override(self): from polygateway.config import OcrSettings env = dict(self._OCR_ENV) env["OCR__RETRY__MAX_ATTEMPTS"] = "5" env["OCR__MONKEY__1__TRUST_ENV"] = "false" settings = OcrSettings.from_env("OCR", env=env) assert settings.gateway.retry.max_attempts == 5 assert settings.gateway.sources[0].trust_env is False def test_missing_base_url_fails(self): from polygateway.config import OcrSettings env = {k: v for k, v in self._OCR_ENV.items() if k != "OCR__MONKEY__1__BASE_URL"} with pytest.raises(ValueError): OcrSettings.from_env("OCR", env=env) class TestGuardsRunOnEveryConstruction: """守卫必须在**任何**构造路径上生效,不只是 from_env。 背景: `GatewaySettings` 的 docstring 声称"构造经 from_env 聚合并通过全部守卫", 而 CLAUDE.md §4.5 明确装配有两条路(`from_env()`/`from_settings()`)。 守卫原先只写在 `from_env` 里,于是 `from_settings` 这条官方路径能装出 一个违反类不变量的 settings —— 类可以合法地存在于它自己声称不可能的状态。 """ @staticmethod def _replace(settings, **changes): """按既有 settings 派生一个改了几项的新 settings(走构造函数,不走 from_env)。""" import dataclasses return dataclasses.replace(settings, **changes) def test_lease_guard_runs_on_direct_construction(self): """租约守卫: 源 timeout_s 超过 lease_ttl_s 时,直接构造也必须报错。""" base = GatewaySettings.from_env("LLM", env=_env()) with pytest.raises(ValueError, match="租约|lease"): self._replace(base, lease_ttl_s=1.0) def test_stall_guard_runs_on_direct_construction(self): """卡死窗口守卫: stall_window_s 小于最大 ttft 时,直接构造也必须报错。""" import dataclasses base = GatewaySettings.from_env( "LLM", env=_env( **{ "LLM__QWEN__1__TTFT_TIMEOUT_S": "30", "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15", "LLM__BACKPRESSURE__STALL_WINDOW_S": "60", } ), ) narrowed = dataclasses.replace(base.backpressure, stall_window_s=20.0) with pytest.raises(ValueError, match="stall"): self._replace(base, backpressure=narrowed) def test_valid_settings_still_constructible(self): """合法组合不受影响 —— 守卫收紧的是错的那些,不是所有直接构造。""" base = GatewaySettings.from_env("LLM", env=_env()) assert self._replace(base, lease_ttl_s=base.lease_ttl_s).lease_ttl_s > 0 def test_guard_message_names_fields_not_only_env_keys(self): """报错要点得出字段名。 守卫一旦在每次构造时都跑,一个在代码里拼 settings 的调用方 (不走 env)会收到这条消息;只点环境变量名会让他去改几个他从没设过的键。 """ base = GatewaySettings.from_env("LLM", env=_env()) with pytest.raises(ValueError) as exc: self._replace(base, lease_ttl_s=1.0) assert "lease_ttl_s" in str(exc.value)