fix: consolidate remaining assembly validation into GatewaySettings

Round two of the from_env-only validation problem. Fifteen checks still lived
in the env parsing functions: six enum domains, the redis_url requirement for
redis-backed limiter/breaker/cache, cache namespace and TTL, telemetry path and
DSN, non-negative structured retries and non-blank scope. from_settings and
direct construction bypassed all of them.

The five asserts in client.py that claimed config had already validated
redis_url and the telemetry targets now hold on every path, so they revert to
what CLAUDE.md permits: internal invariant declarations that also narrow the
Optional for type checkers. Their comments now name the method that guarantees
them, since the previous wording is exactly what went stale.

Postgres DSNs built by hand now get the SQLAlchemy +driver suffix stripped the
way from_env has always stripped it, with a warning so the rewrite is not
silent. The env path strips earlier, so it stays quiet.
This commit is contained in:
2026-07-30 00:58:33 -04:00
parent 8c9e1179bc
commit a65b504a3d
7 changed files with 264 additions and 28 deletions
+141
View File
@@ -1,8 +1,10 @@
"""config.py 配置聚合测试(设计 §8): 多源命名、键优先级、缺失报错。"""
import contextlib
import dataclasses
import pytest
from loguru import logger
from polygateway.client import GatewayClient
from polygateway.config import GatewaySettings, OcrSettings
@@ -22,6 +24,17 @@ _BASE_ENV = {
}
@contextlib.contextmanager
def _captured_warnings():
"""捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。"""
messages: list[str] = []
sink_id = logger.add(messages.append, level="WARNING")
try:
yield messages
finally:
logger.remove(sink_id)
def _env(**overrides):
env = dict(_BASE_ENV)
env.update({k: v for k, v in overrides.items() if v is not None})
@@ -413,3 +426,131 @@ class TestCrossFieldInvariants:
base = self._base()
with pytest.raises(ValueError, match="lease_ttl_s"):
OcrSettings(gateway=dataclasses.replace(base, lease_ttl_s=1.0))
# —— 第二轮(设计 2026-07-30): 后端枚举合法域 ——
@pytest.mark.parametrize(
("field", "bad_value"),
[
("limiter_backend", "rediss"),
("breaker_backend", "sqlite"),
("cache_backend", "postgres"),
("telemetry_backend", "redis"),
("selector", "random"),
("quota_full", "block"),
],
)
def test_enum_field_rejects_value_outside_domain(self, field, bad_value):
"""域外取值此前只有 from_env 拦得住,直接构造会落进 _build_* 的 else 分支。"""
base = self._base()
with pytest.raises(ValueError, match=field):
dataclasses.replace(base, **{field: bad_value})
# —— 条件必填: 取 redis 的后端必须有 redis_url ——
@pytest.mark.parametrize("field", ["limiter_backend", "breaker_backend"])
def test_redis_backend_requires_redis_url(self, field):
"""client.py 的 assert settings.redis_url is not None 依赖的正是这条。"""
base = self._base() # redis_url=None
with pytest.raises(ValueError, match="redis_url"):
dataclasses.replace(base, **{field: "redis"})
def test_redis_cache_requires_redis_url(self):
base = self._base()
with pytest.raises(ValueError, match="redis_url"):
dataclasses.replace(base, cache_backend="redis", cache_namespace="ns", cache_ttl_s=60)
# —— 条件必填: 启用缓存必须有命名空间与正 TTL ——
def test_cache_requires_namespace(self):
"""缺命名空间即失去租户隔离,踩"无缓存毒化"铁律。"""
base = self._base()
with pytest.raises(ValueError, match="cache_namespace"):
dataclasses.replace(base, cache_backend="memory", cache_ttl_s=60)
def test_cache_ttl_must_be_positive(self):
"""from_env 明令禁止的"永不过期"不能从另一条路进来。"""
base = self._base()
with pytest.raises(ValueError, match="cache_ttl_s"):
dataclasses.replace(base, cache_backend="memory", cache_namespace="ns", cache_ttl_s=0)
# —— 条件必填: 遥测后端各自的落点 ——
def test_sqlite_telemetry_requires_path(self):
base = self._base()
with pytest.raises(ValueError, match="telemetry_sqlite_path"):
dataclasses.replace(base, telemetry_backend="sqlite")
def test_postgres_telemetry_requires_dsn(self):
base = self._base()
with pytest.raises(ValueError, match="telemetry_pg_dsn"):
dataclasses.replace(base, telemetry_backend="postgres")
# —— 标量域 ——
def test_negative_structured_retries_rejected(self):
base = self._base()
with pytest.raises(ValueError, match="structured_max_retries"):
dataclasses.replace(base, structured_max_retries=-1)
def test_blank_scope_rejected(self):
"""空 scope 会污染遥测与缓存命名空间。"""
base = self._base()
with pytest.raises(ValueError, match="scope"):
dataclasses.replace(base, scope=" ")
# —— 合法组合仍可构造(收紧的是错的那些)——
def test_full_redis_stack_constructible(self):
base = self._base()
settings = dataclasses.replace(
base,
limiter_backend="redis",
breaker_backend="redis",
cache_backend="redis",
cache_namespace="ns",
cache_ttl_s=60,
redis_url="redis://127.0.0.1:6379/3",
)
assert settings.cache_ttl_s == 60 and settings.redis_url is not None
# —— Postgres DSN: 剥 SQLAlchemy 驱动后缀并出声(设计 §5 方案 C)——
def test_sqlalchemy_dsn_suffix_stripped_with_warning(self):
"""asyncpg 不认 `+driver`;库替调用方剥掉,但不静默——日志里看得见。"""
base = self._base()
with _captured_warnings() as warnings:
settings = dataclasses.replace(
base,
telemetry_backend="postgres",
telemetry_pg_dsn="postgresql+asyncpg://u@h/db",
)
assert settings.telemetry_pg_dsn == "postgresql://u@h/db"
assert any("asyncpg" in m for m in warnings)
def test_env_path_strips_dsn_without_warning(self):
"""env 路已在 _load_pg_dsn 剥过,不该给三项目的历史 DSN 写法刷噪音。"""
with _captured_warnings() as warnings:
settings = GatewaySettings.from_env(
"LLM",
env=_env(
PGW_TELEMETRY_BACKEND="postgres",
PGW_TELEMETRY_PG_DSN="postgresql+asyncpg://u@h/db",
),
)
assert settings.telemetry_pg_dsn == "postgresql://u@h/db"
assert not warnings
# —— 回归护栏: client.py 的 assert 前提确实被保证了 ——
def test_factory_accepts_valid_redis_stack(self):
"""补齐校验后,client.py:262/282/302 的 assert 退回成纯内部不变量声明。"""
base = self._base()
settings = dataclasses.replace(
base,
limiter_backend="redis",
breaker_backend="redis",
redis_url="redis://127.0.0.1:6379/3",
)
client = GatewayClient.from_settings(settings)
assert client is not None