"""U17 — LiteLLM 语义缓存管理器单元测试。

覆盖场景（plan 要求的 4 个 + 安全要求 e + 向后兼容 + fallback）：
1. 语义相似查询命中缓存（mock litellm.acompletion 模拟缓存行为）
2. 不同 system prompt 不命中缓存
3. 缓存命中率可统计
4. 阈值调优生效（similarity_threshold 传入 RedisSemanticCache）
5. User A 不返回 User B 缓存（安全要求 e）
6. kb_acl_hash 隔离 — 不同 ACL hash 产生不同 key
7. kb_caching_disabled 禁用缓存（安全要求 c）
8. cache_params_for_hit / no_cache — 返回正确 dict
9. record_cache_result — 记录命中/未命中到 stats 计数器
10. LitellmCacheConfig.from_cache_config — 转换正确，similarity_threshold=0.87
11. LitellmCacheManager.enable/disable — litellm.cache 正确设置/清除
12. generate_cache_key 向后兼容 — user_id=None, kb_acl_hash=None 时与旧版相同
13. RedisSemanticCache fallback — redisvl 缺失时 auto backend 回退

测试用 ``unittest.mock.patch`` mock ``litellm.cache`` 全局变量，避免测试间污染。
每个测试后清理 ``litellm.cache = None``。
"""

from __future__ import annotations

from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock, patch

import pytest

from agentkit.llm.cache import LitellmCacheConfig, LitellmCacheManager
from agentkit.llm.cache_key import generate_cache_key
from agentkit.llm.config import CacheConfig


# ---------------------------------------------------------------------------
# 测试辅助
# ---------------------------------------------------------------------------


def _make_messages(user_content: str = "Hello") -> list[dict[str, str]]:
    return [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": user_content},
    ]


def _make_litellm_response(
    content: str = "Hello!",
    model: str = "gpt-4o-mini",
    prompt_tokens: int = 10,
    completion_tokens: int = 5,
    cache_key: str | None = None,
) -> SimpleNamespace:
    """构造 LiteLLM 响应（OpenAI ChatCompletion 格式），可选 cache_key 标记。"""
    hidden_params: dict[str, Any] = {}
    if cache_key is not None:
        hidden_params["cache_key"] = cache_key
    message = SimpleNamespace(content=content, tool_calls=None)
    return SimpleNamespace(
        choices=[SimpleNamespace(message=message)],
        usage=SimpleNamespace(prompt_tokens=prompt_tokens, completion_tokens=completion_tokens),
        model=model,
        _hidden_params=hidden_params,
    )


@pytest.fixture(autouse=True)
def _cleanup_litellm_cache():
    """每个测试后清理 litellm.cache 全局变量，避免测试间污染。"""
    yield
    import litellm

    litellm.cache = None


# ---------------------------------------------------------------------------
# 1. 语义相似查询命中缓存
# ---------------------------------------------------------------------------


class TestCacheHit:
    """场景 1 — 相同请求第二次命中缓存。"""

    async def test_second_request_is_cache_hit(self):
        """相同 cache_key 的第二次请求返回缓存响应（cache_hit=True）。"""
        config = LitellmCacheConfig(enabled=True, backend="memory")
        manager = LitellmCacheManager(config)

        msgs = _make_messages()
        cache_key = manager.build_cache_key("gpt-4o", msgs, 0.0)
        cache_params = LitellmCacheManager.cache_params_for_hit(cache_key)

        # 模拟 LiteLLM 缓存行为：第一次 miss，第二次 hit
        call_count = 0
        cache_store: dict[str, SimpleNamespace] = {}

        async def fake_acompletion(**kwargs):
            nonlocal call_count
            call_count += 1
            ck = kwargs.get("cache", {}).get("cache_key")
            if ck and ck in cache_store:
                # 缓存命中 — 返回带 cache_key 标记的响应
                return _make_litellm_response(content="Cached!", cache_key=ck)
            # 缓存未命中 — 调用"真实" API 并存储
            resp = _make_litellm_response(content="Fresh response")
            if ck:
                cache_store[ck] = resp
            return resp

        with patch("litellm.acompletion", side_effect=fake_acompletion):
            from agentkit.llm.providers.litellm_provider import LitellmProvider

            provider = LitellmProvider(model_prefix="openai/", api_key="test")
            from agentkit.llm.protocol import LLMRequest

            # 第一次请求 — miss
            req1 = LLMRequest(messages=msgs, model="gpt-4o", temperature=0.0, cache=cache_params)
            resp1 = await provider.chat(req1)
            assert resp1.cache_hit is False
            assert resp1.content == "Fresh response"

            # 第二次相同请求 — hit
            req2 = LLMRequest(messages=msgs, model="gpt-4o", temperature=0.0, cache=cache_params)
            resp2 = await provider.chat(req2)
            assert resp2.cache_hit is True
            assert resp2.content == "Cached!"

        assert call_count == 2  # litellm.acompletion 被调用两次（LiteLLM 内部处理缓存）


# ---------------------------------------------------------------------------
# 2. 不同 system prompt 不命中缓存
# ---------------------------------------------------------------------------


class TestSystemPromptIsolation:
    """场景 2 — 不同 system prompt 产生不同 cache_key，不误命中。"""

    def test_different_system_prompt_different_key(self):
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        msgs1 = [
            {"role": "system", "content": "Be concise"},
            {"role": "user", "content": "Hello"},
        ]
        msgs2 = [
            {"role": "system", "content": "Be verbose"},
            {"role": "user", "content": "Hello"},
        ]
        key1 = manager.build_cache_key("gpt-4o", msgs1, 0.0)
        key2 = manager.build_cache_key("gpt-4o", msgs2, 0.0)
        assert key1 != key2


# ---------------------------------------------------------------------------
# 3. 缓存命中率可统计
# ---------------------------------------------------------------------------


class TestCacheStats:
    """场景 3 — LitellmCacheManager.stats() 返回正确 hits/misses。"""

    def test_stats_after_hits_and_misses(self):
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))

        # 2 hits
        manager.record_cache_result(True)
        manager.record_cache_result(True)
        # 1 miss
        manager.record_cache_result(False)

        stats = manager.stats()
        assert stats["total_hits"] == 2
        assert stats["total_misses"] == 1


# ---------------------------------------------------------------------------
# 4. 阈值调优生效
# ---------------------------------------------------------------------------


class TestSimilarityThreshold:
    """场景 4 — similarity_threshold 传入 RedisSemanticCache 构造函数。"""

    def test_threshold_passed_to_redis_semantic_cache(self):
        """RedisSemanticCache 构造时接收 similarity_threshold=0.87。"""
        config = LitellmCacheConfig(
            enabled=True,
            backend="redis_semantic",
            similarity_threshold=0.83,
            redis_url="redis://localhost:6379",
        )
        manager = LitellmCacheManager(config)

        mock_cache_instance = MagicMock()
        with patch(
            "litellm.caching.RedisSemanticCache", return_value=mock_cache_instance
        ) as mock_cls:
            instance = manager._create_cache_instance()
            mock_cls.assert_called_once_with(
                redis_url="redis://localhost:6379",
                similarity_threshold=0.83,
                embedding_model="text-embedding-ada-002",
            )
            assert instance is mock_cache_instance

    def test_default_threshold_is_087(self):
        """from_cache_config 固定 similarity_threshold=0.87。"""
        old_config = CacheConfig(similarity_threshold=0.92)
        litellm_config = LitellmCacheConfig.from_cache_config(old_config)
        assert litellm_config.similarity_threshold == 0.87


# ---------------------------------------------------------------------------
# 5. User A 不返回 User B 缓存（安全要求 e）
# ---------------------------------------------------------------------------


class TestUserIsolation:
    """安全要求 e — User A 的查询不返回 User B 的缓存响应。"""

    def test_different_users_different_keys(self):
        """不同 user_id 产生不同 cache_key。"""
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        msgs = _make_messages("What is my salary?")
        key_a = manager.build_cache_key("gpt-4o", msgs, 0.0, user_id="user_a")
        key_b = manager.build_cache_key("gpt-4o", msgs, 0.0, user_id="user_b")
        assert key_a != key_b

    def test_same_user_same_key(self):
        """相同 user_id 产生相同 cache_key。"""
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        msgs = _make_messages("What is my salary?")
        key1 = manager.build_cache_key("gpt-4o", msgs, 0.0, user_id="user_a")
        key2 = manager.build_cache_key("gpt-4o", msgs, 0.0, user_id="user_a")
        assert key1 == key2

    def test_no_user_id_same_as_no_user_id(self):
        """user_id=None 时两次调用产生相同 key（向后兼容）。"""
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        msgs = _make_messages()
        key1 = manager.build_cache_key("gpt-4o", msgs, 0.0, user_id=None)
        key2 = manager.build_cache_key("gpt-4o", msgs, 0.0, user_id=None)
        assert key1 == key2


# ---------------------------------------------------------------------------
# 6. kb_acl_hash 隔离
# ---------------------------------------------------------------------------


class TestACLIsolation:
    """安全要求 b — 不同 ACL scope 产生不同 cache_key。"""

    def test_different_acl_hash_different_keys(self):
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        msgs = _make_messages("Summarize the document")
        key1 = manager.build_cache_key("gpt-4o", msgs, 0.0, kb_acl_hash="acl_v1")
        key2 = manager.build_cache_key("gpt-4o", msgs, 0.0, kb_acl_hash="acl_v2")
        assert key1 != key2

    def test_same_acl_hash_same_key(self):
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        msgs = _make_messages()
        key1 = manager.build_cache_key("gpt-4o", msgs, 0.0, kb_acl_hash="acl_v1")
        key2 = manager.build_cache_key("gpt-4o", msgs, 0.0, kb_acl_hash="acl_v1")
        assert key1 == key2


# ---------------------------------------------------------------------------
# 7. kb_caching_disabled 禁用缓存
# ---------------------------------------------------------------------------


class TestKBCachingDisabled:
    """安全要求 c — KB 设置 caching_disabled=True 时禁用缓存。"""

    def test_should_cache_returns_false_when_disabled(self):
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        assert manager.should_cache(kb_caching_disabled=True, user_id="u1") is False

    def test_should_cache_returns_true_when_enabled(self):
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        assert manager.should_cache(kb_caching_disabled=False, user_id="u1") is True

    def test_should_cache_default_with_user(self):
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        assert manager.should_cache(user_id="u1") is True

    def test_should_cache_rejects_anonymous(self):
        """安全要求 (a)(e) — per_user_namespace 开启且 user_id=None 时拒绝缓存。"""
        manager = LitellmCacheManager(LitellmCacheConfig(backend="memory"))
        assert manager.should_cache(user_id=None) is False

    def test_should_cache_allows_anonymous_when_namespace_off(self):
        """per_user_namespace=False 时匿名请求可缓存（向后兼容场景）。"""
        cfg = LitellmCacheConfig(backend="memory")
        cfg.per_user_namespace = False
        manager = LitellmCacheManager(cfg)
        assert manager.should_cache(user_id=None) is True


# ---------------------------------------------------------------------------
# 8. cache_params_for_hit / no_cache
# ---------------------------------------------------------------------------


class TestCacheParams:
    def test_cache_params_for_hit(self):
        params = LitellmCacheManager.cache_params_for_hit("my_cache_key")
        assert params == {"cache_key": "my_cache_key"}

    def test_cache_params_for_no_cache(self):
        params = LitellmCacheManager.cache_params_for_no_cache()
        assert params == {"no-cache": True}


# ---------------------------------------------------------------------------
# 10. LitellmCacheConfig.from_cache_config
# ---------------------------------------------------------------------------


class TestFromCacheConfig:
    def test_basic_conversion(self):
        old = CacheConfig(
            enabled=True,
            backend="memory",
            redis_url="redis://myhost:6379",
            semantic_ttl=3600,
            embedding_model="bge-m3",
        )
        litellm_config = LitellmCacheConfig.from_cache_config(old)
        assert litellm_config.enabled is True
        assert litellm_config.backend == "memory"
        assert litellm_config.redis_url == "redis://myhost:6379"
        assert litellm_config.ttl == 3600
        assert litellm_config.similarity_threshold == 0.87  # 固定，忽略旧 0.92
        assert litellm_config.per_user_namespace is True  # 强制开启

    def test_unknown_backend_falls_to_auto(self):
        old = CacheConfig(backend="unknown_backend")
        litellm_config = LitellmCacheConfig.from_cache_config(old)
        assert litellm_config.backend == "auto"

    def test_redis_backend_preserved(self):
        old = CacheConfig(backend="redis")
        litellm_config = LitellmCacheConfig.from_cache_config(old)
        assert litellm_config.backend == "redis"

    def test_empty_embedding_model_falls_to_default(self):
        old = CacheConfig(embedding_model="")
        litellm_config = LitellmCacheConfig.from_cache_config(old)
        assert litellm_config.embedding_model == "text-embedding-ada-002"


# ---------------------------------------------------------------------------
# 11. LitellmCacheManager.enable/disable
# ---------------------------------------------------------------------------


class TestEnableDisable:
    def test_enable_sets_litellm_cache(self):
        import litellm

        config = LitellmCacheConfig(backend="memory")
        manager = LitellmCacheManager(config)
        manager.enable()
        assert litellm.cache is not None
        assert manager._cache_instance is not None

    def test_disable_clears_litellm_cache(self):
        import litellm

        config = LitellmCacheConfig(backend="memory")
        manager = LitellmCacheManager(config)
        manager.enable()
        assert litellm.cache is not None
        manager.disable()
        assert litellm.cache is None
        assert manager._cache_instance is None


# ---------------------------------------------------------------------------
# 12. generate_cache_key 向后兼容
# ---------------------------------------------------------------------------


class TestBackwardCompatibility:
    """user_id=None, kb_acl_hash=None 时与旧版 generate_cache_key 完全一致。"""

    def test_none_user_id_same_as_not_passing(self):
        msgs = _make_messages()
        key1 = generate_cache_key("gpt-4o", msgs, 0.0, user_id=None, kb_acl_hash=None)
        key2 = generate_cache_key("gpt-4o", msgs, 0.0)
        assert key1 == key2

    def test_backward_compat_deterministic(self):
        msgs = _make_messages()
        key1 = generate_cache_key("gpt-4o", msgs, 0.0, user_id=None, kb_acl_hash=None)
        key2 = generate_cache_key("gpt-4o", msgs, 0.0, user_id=None, kb_acl_hash=None)
        assert key1 == key2

    def test_user_id_changes_key(self):
        msgs = _make_messages()
        key_none = generate_cache_key("gpt-4o", msgs, 0.0)
        key_user = generate_cache_key("gpt-4o", msgs, 0.0, user_id="user_a")
        assert key_none != key_user

    def test_acl_hash_changes_key(self):
        msgs = _make_messages()
        key_none = generate_cache_key("gpt-4o", msgs, 0.0)
        key_acl = generate_cache_key("gpt-4o", msgs, 0.0, kb_acl_hash="acl_v1")
        assert key_none != key_acl


# ---------------------------------------------------------------------------
# 13. RedisSemanticCache fallback
# ---------------------------------------------------------------------------


class TestRedisSemanticCacheFallback:
    """redisvl 缺失时 auto backend 回退到 RedisCache。"""

    def test_auto_fallback_to_redis_cache_when_redisvl_missing(self):
        """auto 模式下 RedisSemanticCache 构造 ImportError → 回退到 RedisCache。"""
        import litellm.caching

        config = LitellmCacheConfig(backend="auto", redis_url="redis://localhost:6379")
        manager = LitellmCacheManager(config)

        mock_redis_cache = MagicMock(name="RedisCacheInstance")

        # 模拟 RedisSemanticCache 构造时 ImportError（redisvl 未安装）
        def raise_import_error(*args, **kwargs):
            raise ImportError("No module named 'redisvl'")

        with patch.object(litellm.caching, "RedisSemanticCache", side_effect=raise_import_error):
            with patch.object(
                litellm.caching, "RedisCache", return_value=mock_redis_cache
            ) as mock_rc:
                instance = manager._create_cache_instance()
                mock_rc.assert_called_once_with(redis_url="redis://localhost:6379")
                assert instance is mock_redis_cache

    def test_redis_semantic_backend_raises_when_redisvl_missing(self):
        """显式 redis_semantic backend + redisvl 缺失 → raise ImportError。"""
        import litellm.caching

        config = LitellmCacheConfig(backend="redis_semantic")
        manager = LitellmCacheManager(config)

        def raise_import_error(*args, **kwargs):
            raise ImportError("No module named 'redisvl'")

        with patch.object(litellm.caching, "RedisSemanticCache", side_effect=raise_import_error):
            with pytest.raises(ImportError):
                manager._create_cache_instance()

    def test_memory_backend_uses_in_memory_cache(self):
        """memory backend 直接使用 InMemoryCache，不尝试 Redis。"""
        import litellm.caching

        config = LitellmCacheConfig(backend="memory")
        manager = LitellmCacheManager(config)
        instance = manager._create_cache_instance()
        assert isinstance(instance, litellm.caching.InMemoryCache)
