test: pin explicit cache migration and reasoning row semantics
This commit is contained in:
@@ -592,3 +592,229 @@ class TestTelemetryCapDoesNotPoisonTheCacheKey:
|
||||
assert "(略 112 字)" in logged[1]["content"][0]["text"]
|
||||
|
||||
assert build_cache_key("m", messages, "proj", None) == before
|
||||
|
||||
|
||||
class TestExplicitCacheMigration:
|
||||
"""相同模型身份不代表相同推理策略，隔离必须由调用方显式选择。"""
|
||||
|
||||
def _client(self, cache, source, *, capabilities=None, registry=None):
|
||||
import httpx
|
||||
|
||||
from polygateway.transports.openai_compat import OpenAICompatTransport
|
||||
from tests.unit.test_client import _client, _sse
|
||||
|
||||
sent = []
|
||||
|
||||
def handler(request):
|
||||
payload = json.loads(request.content)
|
||||
sent.append(payload)
|
||||
return _sse(json.dumps(payload, sort_keys=True))
|
||||
|
||||
transport = OpenAICompatTransport(
|
||||
client_factory=lambda src: httpx.AsyncClient(transport=httpx.MockTransport(handler)),
|
||||
capabilities=capabilities,
|
||||
registry=registry,
|
||||
)
|
||||
client = _client(
|
||||
sources=[source],
|
||||
transport=transport,
|
||||
cache=cache,
|
||||
cache_namespace="tenant-a",
|
||||
cache_ttl_s=60,
|
||||
)
|
||||
return client, transport, sent
|
||||
|
||||
@pytest.mark.parametrize("isolation", ["namespace", "salt"])
|
||||
@pytest.mark.parametrize("change", ["capability", "fallback"])
|
||||
async def test_capability_change_requires_explicit_identity(self, isolation, change):
|
||||
from polygateway.client import build_model_fingerprint
|
||||
from polygateway.errors import RequestRejectedError
|
||||
from polygateway.thinking import ThinkingCapability
|
||||
from tests.unit.test_client import _source
|
||||
|
||||
cache = InMemoryCache()
|
||||
source = _source(provider="openai", model="migration-model")
|
||||
if change == "capability":
|
||||
old_cap = ThinkingCapability((Effort.AUTO, Effort.HIGH), "本地旧声明")
|
||||
new_cap = ThinkingCapability((Effort.HIGH,), "本地新声明")
|
||||
tier = Effort.AUTO
|
||||
new_source = source
|
||||
else:
|
||||
old_cap = new_cap = ThinkingCapability((Effort.LOW, Effort.HIGH), "本地映射声明")
|
||||
source = dataclasses.replace(source, effort_fallback="nearest")
|
||||
new_source = dataclasses.replace(source, effort_fallback="error")
|
||||
tier = Effort.MEDIUM
|
||||
assert build_model_fingerprint([source]) == build_model_fingerprint([new_source])
|
||||
old, old_transport, old_sent = self._client(
|
||||
cache, source, capabilities={source.model: old_cap}
|
||||
)
|
||||
new, new_transport, new_sent = self._client(
|
||||
cache, new_source, capabilities={source.model: new_cap}
|
||||
)
|
||||
identity = (
|
||||
{"cache_namespace": "tenant-a:migrated"}
|
||||
if isolation == "namespace"
|
||||
else {"cache_salt": "migrated"}
|
||||
)
|
||||
try:
|
||||
original = await old.chat(_MSGS, reasoning_effort=tier)
|
||||
replay = await new.chat(_MSGS, reasoning_effort=tier)
|
||||
assert replay.cache_hit and replay.content == original.content
|
||||
assert len(old_sent) == 1 and not new_sent
|
||||
with pytest.raises(RequestRejectedError):
|
||||
await new.chat(_MSGS, reasoning_effort=tier, **identity)
|
||||
assert not new_sent
|
||||
assert (await old.chat(_MSGS, reasoning_effort=tier)).cache_hit
|
||||
finally:
|
||||
await old_transport.aclose()
|
||||
await new_transport.aclose()
|
||||
|
||||
@pytest.mark.parametrize("isolation", ["namespace", "salt"])
|
||||
async def test_custom_wire_change_requires_explicit_identity(self, isolation):
|
||||
from polygateway.client import build_model_fingerprint
|
||||
from polygateway.providers import ProviderProfile, ThinkingWire
|
||||
from polygateway.thinking import ThinkingCapability
|
||||
from tests.unit.test_client import _source
|
||||
|
||||
source = _source(provider="custom", model="migration-model")
|
||||
caps = {source.model: ThinkingCapability((Effort.HIGH,), "本地声明")}
|
||||
|
||||
def profile(key):
|
||||
return {
|
||||
"custom": ProviderProfile(
|
||||
name="custom",
|
||||
thinking=ThinkingWire(off=None, on_base={}, effort_key=key),
|
||||
strip_think_tags=False,
|
||||
)
|
||||
}
|
||||
|
||||
cache = InMemoryCache()
|
||||
old, t1, sent1 = self._client(cache, source, capabilities=caps, registry=profile("depth_a"))
|
||||
new, t2, sent2 = self._client(cache, source, capabilities=caps, registry=profile("depth_b"))
|
||||
assert build_model_fingerprint(old._terminal._sources) == build_model_fingerprint(
|
||||
new._terminal._sources
|
||||
)
|
||||
identity = (
|
||||
{"cache_namespace": "tenant-a:migrated"}
|
||||
if isolation == "namespace"
|
||||
else {"cache_salt": "migrated"}
|
||||
)
|
||||
try:
|
||||
original = await old.chat(_MSGS, reasoning_effort=Effort.HIGH)
|
||||
assert (await new.chat(_MSGS, reasoning_effort=Effort.HIGH)).cache_hit
|
||||
migrated = await new.chat(_MSGS, reasoning_effort=Effort.HIGH, **identity)
|
||||
assert not migrated.cache_hit and migrated.content != original.content
|
||||
assert len(sent1) == len(sent2) == 1
|
||||
assert sent2[0]["depth_b"] == "high" and "depth_a" not in sent2[0]
|
||||
assert (await old.chat(_MSGS, reasoning_effort=Effort.HIGH)).content == original.content
|
||||
finally:
|
||||
await t1.aclose()
|
||||
await t2.aclose()
|
||||
|
||||
@pytest.mark.parametrize("isolation", ["namespace", "salt"])
|
||||
async def test_legacy_raw_override_requires_explicit_identity(self, isolation):
|
||||
from polygateway.client import build_model_fingerprint
|
||||
from polygateway.errors import RequestRejectedError
|
||||
from tests.unit.test_client import _source
|
||||
|
||||
source = _source(
|
||||
provider="openai", reasoning_effort="high", extra_body={"reasoning_effort": "low"}
|
||||
)
|
||||
cache = InMemoryCache()
|
||||
key = build_cache_key(build_model_fingerprint([source]), _MSGS, "tenant-a", None)
|
||||
legacy = dataclasses.asdict(_resp(content="legacy-raw-low", applied_effort=Effort.HIGH))
|
||||
legacy.pop("structured_data", None)
|
||||
await cache.set(key, json.dumps(legacy), 60)
|
||||
client, transport, sent = self._client(cache, source)
|
||||
identity = (
|
||||
{"cache_namespace": "tenant-a:migrated"}
|
||||
if isolation == "namespace"
|
||||
else {"cache_salt": "migrated"}
|
||||
)
|
||||
try:
|
||||
assert (await client.chat(_MSGS)).content == "legacy-raw-low"
|
||||
with pytest.raises(RequestRejectedError, match="冲突"):
|
||||
await client.chat(_MSGS, **identity)
|
||||
assert sent == []
|
||||
assert await cache.get(key) is not None
|
||||
finally:
|
||||
await transport.aclose()
|
||||
|
||||
async def test_per_call_namespace_survives_a_changed_factory_default(self):
|
||||
from polygateway import GatewayClient, GatewaySettings
|
||||
from tests.unit.test_client import _ENV, _source
|
||||
|
||||
cache = InMemoryCache()
|
||||
source = _source()
|
||||
old, transport, sent = self._client(cache, source)
|
||||
try:
|
||||
await old.chat(_MSGS, cache_namespace="tenant-a")
|
||||
# 工厂路径和全量注入配置同模型身份；只改默认不能改变显式租户覆盖。
|
||||
settings = GatewaySettings.from_env(
|
||||
env={
|
||||
**_ENV,
|
||||
"PGW_CACHE_BACKEND": "memory",
|
||||
"PGW_CACHE_NAMESPACE": "changed-default",
|
||||
"PGW_CACHE_TTL_S": "60",
|
||||
}
|
||||
)
|
||||
new = GatewayClient.from_settings(settings, cache=cache)
|
||||
try:
|
||||
assert (await new.chat(_MSGS, cache_namespace="tenant-a")).cache_hit
|
||||
assert len(sent) == 1
|
||||
finally:
|
||||
await new.aclose()
|
||||
finally:
|
||||
await transport.aclose()
|
||||
|
||||
async def test_shared_source_pool_migration_preserves_tenant_boundaries(self):
|
||||
import httpx
|
||||
|
||||
from polygateway.errors import RequestRejectedError
|
||||
from polygateway.thinking import ThinkingCapability
|
||||
from polygateway.transports.openai_compat import OpenAICompatTransport
|
||||
from tests.unit.test_client import _client, _source, _sse
|
||||
|
||||
sources = [
|
||||
_source(name=name, provider="openai", model="shared-model") for name in ("a", "b")
|
||||
]
|
||||
cache = InMemoryCache()
|
||||
sent = []
|
||||
|
||||
def handler(request):
|
||||
sent.append(request)
|
||||
return _sse()
|
||||
|
||||
transports = [
|
||||
OpenAICompatTransport(
|
||||
client_factory=lambda src: httpx.AsyncClient(
|
||||
transport=httpx.MockTransport(handler)
|
||||
),
|
||||
capabilities={"shared-model": ThinkingCapability(choices, "本地声明")},
|
||||
)
|
||||
for choices in ((Effort.AUTO, Effort.HIGH), (Effort.HIGH,))
|
||||
]
|
||||
clients = [
|
||||
_client(
|
||||
sources=sources, transport=t, cache=cache, cache_namespace="default", cache_ttl_s=60
|
||||
)
|
||||
for t in transports
|
||||
]
|
||||
try:
|
||||
for tenant in ("tenant-a", "tenant-b"):
|
||||
await clients[0].chat(_MSGS, reasoning_effort="auto", cache_namespace=tenant)
|
||||
assert (
|
||||
await clients[1].chat(_MSGS, reasoning_effort="auto", cache_namespace=tenant)
|
||||
).cache_hit
|
||||
with pytest.raises(RequestRejectedError):
|
||||
await clients[1].chat(
|
||||
_MSGS, reasoning_effort="auto", cache_namespace=tenant + ":new"
|
||||
)
|
||||
assert len(sent) == 2
|
||||
for tenant in ("tenant-a", "tenant-b"):
|
||||
assert (
|
||||
await clients[0].chat(_MSGS, reasoning_effort="auto", cache_namespace=tenant)
|
||||
).cache_hit
|
||||
finally:
|
||||
for transport in transports:
|
||||
await transport.aclose()
|
||||
|
||||
Reference in New Issue
Block a user