diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py index c83e423..3f5b4b6 100644 --- a/tests/unit/test_cache.py +++ b/tests/unit/test_cache.py @@ -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()