iomgaa
b8f738f8cb
The lease, stall and probe-TTL guards only ran inside GatewaySettings.from_env, so from_settings() and direct construction could produce settings that violate the class's own invariants: the permit lease could expire mid-request (silently exceeding the concurrency quota), a normal slow first token could be killed as a stall, and a half-open probe could be taken over while still in flight. Guards move into __post_init__ as _validate_* methods, matching every frozen dataclass in types.py, so all six factories plus dataclasses.replace are covered by one check. Adds a non-empty sources invariant that previously only from_env enforced. Messages now name fields instead of env keys, since callers who build settings by hand never set those keys.
416 lines
19 KiB
Python
416 lines
19 KiB
Python
"""config.py 配置聚合测试(设计 §8): 多源命名、键优先级、缺失报错。"""
|
||
|
||
import dataclasses
|
||
|
||
import pytest
|
||
|
||
from polygateway.client import GatewayClient
|
||
from polygateway.config import GatewaySettings, OcrSettings
|
||
|
||
_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 TestCrossFieldInvariants:
|
||
"""四条跨字段不变量必须在**任何**构造路径上生效(设计 2026-07-29)。
|
||
|
||
这些约束单看一个字段都合法,组合起来才非法,因此 types.py 各子配置的
|
||
__post_init__ 看不见——只能由聚合层 GatewaySettings 把关。守卫若只挂在
|
||
from_env 上,from_settings 这条同等官方的装配路(CLAUDE.md §4.5)就能
|
||
装出违反不变量的配置,类会存在于自己 docstring 声称不可能的状态。
|
||
|
||
每条不变量测两侧: 越界必拒、边界值(恰好相等)必过——收紧的是错的组合,
|
||
不是所有直接构造。
|
||
"""
|
||
|
||
def _base(self, **overrides) -> GatewaySettings:
|
||
return GatewaySettings.from_env("LLM", env=_env(**overrides))
|
||
|
||
def _with_watchdog(self) -> GatewaySettings:
|
||
"""带看门狗的基准: TTFT/inter-token 成对配置才满足 SourceConfig 不变式。"""
|
||
return self._base(
|
||
**{
|
||
"LLM__QWEN__1__TTFT_TIMEOUT_S": "30",
|
||
"LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15",
|
||
"LLM__BACKPRESSURE__STALL_WINDOW_S": "300",
|
||
}
|
||
)
|
||
|
||
# —— 源超时 ≤ permit 租约 TTL(ARCH §7.3: 防租约先于请求过期,并发悄悄超配额)——
|
||
|
||
def test_lease_rejects_timeout_above_ttl_on_direct_construction(self):
|
||
base = self._base() # 源 timeout_s=120
|
||
with pytest.raises(ValueError, match="lease_ttl_s"):
|
||
dataclasses.replace(base, lease_ttl_s=1.0)
|
||
|
||
def test_lease_accepts_timeout_equal_to_ttl(self):
|
||
base = self._base()
|
||
assert dataclasses.replace(base, lease_ttl_s=120.0).lease_ttl_s == 120.0
|
||
|
||
# —— stall 窗口 ≥ 最大源 TTFT(ARCH §7.3: 防正常慢首包被误判卡死掐断)——
|
||
|
||
def test_stall_rejects_window_below_max_ttft_on_direct_construction(self):
|
||
base = self._with_watchdog() # 源 ttft_timeout_s=30
|
||
narrowed = dataclasses.replace(base.backpressure, stall_window_s=20.0)
|
||
with pytest.raises(ValueError, match="stall_window_s"):
|
||
dataclasses.replace(base, backpressure=narrowed)
|
||
|
||
def test_stall_accepts_window_equal_to_max_ttft(self):
|
||
base = self._with_watchdog()
|
||
exact = dataclasses.replace(base.backpressure, stall_window_s=30.0)
|
||
assert dataclasses.replace(base, backpressure=exact).backpressure.stall_window_s == 30.0
|
||
|
||
# —— 探针租约 ≥ 最慢源超时 + 5(M2 设计 §3: 防半开探针在途即被接管)——
|
||
|
||
def test_probe_rejects_ttl_below_floor_on_direct_construction(self):
|
||
base = self._base() # 最慢 timeout_s=120,故下限 125
|
||
shortened = dataclasses.replace(base.breaker, probe_ttl_s=100.0)
|
||
with pytest.raises(ValueError, match="probe_ttl_s"):
|
||
dataclasses.replace(base, breaker=shortened)
|
||
|
||
def test_probe_accepts_ttl_at_floor(self):
|
||
base = self._base()
|
||
at_floor = dataclasses.replace(base.breaker, probe_ttl_s=125.0)
|
||
assert dataclasses.replace(base, breaker=at_floor).breaker.probe_ttl_s == 125.0
|
||
|
||
# —— sources 非空 ——
|
||
|
||
def test_empty_sources_rejected_with_actionable_message(self):
|
||
"""零源装出来的 client 选源必然失败;消息须点明原因,不能泄漏 max() 的内置异常。"""
|
||
base = self._base()
|
||
with pytest.raises(ValueError) as exc:
|
||
dataclasses.replace(base, sources=())
|
||
assert "至少一个源" in str(exc.value)
|
||
assert "empty sequence" not in str(exc.value)
|
||
|
||
# —— 装配路径覆盖 ——
|
||
|
||
def test_factory_cannot_receive_invalid_settings(self):
|
||
"""from_settings 这条路吃不到非法配置。
|
||
|
||
异常实际抛在实参求值(构造 settings)那一刻,而不是工厂内部——这正是
|
||
把守卫放构造期换来的性质: 非法实例根本不存在,无需每个工厂各自设防。
|
||
"""
|
||
base = self._base()
|
||
with pytest.raises(ValueError, match="lease_ttl_s"):
|
||
GatewayClient.from_settings(dataclasses.replace(base, lease_ttl_s=1.0))
|
||
|
||
def test_ocr_settings_cannot_wrap_invalid_gateway(self):
|
||
"""OcrSettings/EmbeddingSettings 只是包一层 GatewaySettings,自动继承同一把关。"""
|
||
base = self._base()
|
||
with pytest.raises(ValueError, match="lease_ttl_s"):
|
||
OcrSettings(gateway=dataclasses.replace(base, lease_ttl_s=1.0))
|