GarminHealthLab/backend/tests/test_ai.py

"""
Unit tests for the multi-provider LLM layer.

Every HTTP call is mocked — the suite never touches the network and never
needs a real API key.
"""
import json

import pytest
import requests

from services import ai as ai_svc
from services import analysis as analysis_svc


VALID_REPLY = json.dumps(
    [
        {"category": "睡眠", "recommendation": "固定就寝时间，目标 7-8 小时。",
         "priority": "high", "basedOn": ["sleep_duration"]},
        {"category": "运动", "recommendation": "每天增加 20 分钟快走。",
         "priority": "medium", "basedOn": ["steps"]},
    ],
    ensure_ascii=False,
)

SUMMARY = [
    {"date": "2026-08-20", "steps": 6500, "heartRate": 70,
     "heartRateVariability": 45, "stress": 55, "caloriesBurned": 260,
     "sleep": {"duration": 6, "quality": 80}},
    {"date": "2026-08-21", "steps": 9000, "heartRate": 62,
     "heartRateVariability": 46, "stress": 40, "caloriesBurned": 360,
     "sleep": {"duration": 8, "quality": 79}},
]


class FakeResponse:
    def __init__(self, status_code=200, payload=None, text=""):
        self.status_code = status_code
        self._payload = payload
        self.text = text or json.dumps(payload or {})

    def json(self):
        if self._payload is None:
            raise ValueError("no json")
        return self._payload


def gemini_payload(text):
    return {"candidates": [{"content": {"parts": [{"text": text}]}}]}


def openai_payload(text):
    return {"choices": [{"message": {"content": text}}]}


@pytest.fixture
def keys(monkeypatch):
    """Direct vendor keys configured; the gateway stays out of the chain."""
    monkeypatch.setenv("GEMINI_API_KEY", "test-gemini-key")
    monkeypatch.setenv("NVIDIA_API_KEY", "test-nvidia-key")
    monkeypatch.delenv("AI_GATEWAY_TOKEN", raising=False)
    monkeypatch.delenv("AI_GATEWAY_BASE_URL", raising=False)
    return True


@pytest.fixture
def no_keys(monkeypatch):
    for var in (
        "GEMINI_API_KEY", "NVIDIA_API_KEY",
        "AI_GATEWAY_TOKEN", "AI_GATEWAY_BASE_URL",
    ):
        monkeypatch.delenv(var, raising=False)
    return True


@pytest.fixture
def gateway(monkeypatch):
    """Only the self-hosted gateway is configured."""
    monkeypatch.setenv("AI_GATEWAY_TOKEN", "test-gateway-token")
    monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gw.test:5100/v1")
    monkeypatch.delenv("GEMINI_API_KEY", raising=False)
    monkeypatch.delenv("NVIDIA_API_KEY", raising=False)
    return True


# --- prompt construction ----------------------------------------------------
class TestBuildPrompt:
    def test_includes_every_day_as_a_csv_row(self):
        prompt = ai_svc.build_prompt(SUMMARY)
        assert "2026-08-20" in prompt and "2026-08-21" in prompt
        assert "共 2 天" in prompt

    def test_uses_csv_not_json(self):
        """CSV keeps a year of history affordable; JSON would not."""
        prompt = ai_svc.build_prompt(SUMMARY)
        assert "6500,45" in prompt.replace(" ", "") or "6500" in prompt
        assert '"steps":' not in prompt

    def test_missing_metrics_become_empty_cells_not_the_word_none(self):
        prompt = ai_svc.build_prompt([{"date": "2026-08-20", "steps": None}])
        assert "None" not in prompt

    def test_sleep_is_flattened_into_columns(self):
        prompt = ai_svc.build_prompt(SUMMARY)
        assert "sleep_h,sleep_q" in prompt

    def test_day_budget_trims_to_the_most_recent_days(self):
        many = [{"date": f"2026-01-{d:02d}", "steps": d} for d in range(1, 32)]
        prompt = ai_svc.build_prompt(many, day_budget=5)
        assert "共 5 天" in prompt
        assert "2026-01-31" in prompt, "must keep the newest days"
        assert "2026-01-01" not in prompt, "must drop the oldest days"

    def test_activities_included_when_supplied(self):
        prompt = ai_svc.build_prompt(
            SUMMARY, [{"activity_type": "running", "distance": 5.0}]
        )
        assert "running" in prompt

    def test_activities_capped(self):
        acts = [{"activity_type": f"run{i}"} for i in range(500)]
        prompt = ai_svc.build_prompt(SUMMARY, acts)
        assert "共 200 条" in prompt

    def test_prompt_forbids_fabricating_numbers(self):
        assert "不要编造" in ai_svc.build_prompt(SUMMARY)

    def test_prompt_disclaims_medical_advice(self):
        assert "不是医生" in ai_svc.build_prompt(SUMMARY)

    def test_a_year_of_data_stays_compact(self):
        year = [
            {"date": f"2026-{m:02d}-{d:02d}", "steps": 8000, "heartRate": 60,
             "sleep": {"duration": 7, "quality": 80}}
            for m in range(1, 13) for d in range(1, 29)
        ]
        prompt = ai_svc.build_prompt(year)
        # ~4 chars/token: a year must stay far under even the smallest window.
        assert len(prompt) / 4 < 50_000


# --- response parsing -------------------------------------------------------
class TestParseRecommendations:
    def test_plain_json_array(self):
        recs = ai_svc.parse_recommendations(VALID_REPLY)
        assert len(recs) == 2
        assert recs[0]["category"] == "睡眠"

    def test_markdown_fenced_json(self):
        recs = ai_svc.parse_recommendations(f"```json\n{VALID_REPLY}\n```")
        assert len(recs) == 2

    def test_json_with_a_preamble_sentence(self):
        recs = ai_svc.parse_recommendations(f"好的，分析结果如下：\n{VALID_REPLY}")
        assert len(recs) == 2

    def test_single_object_is_wrapped(self):
        recs = ai_svc.parse_recommendations(
            '{"category":"睡眠","recommendation":"早点睡","priority":"high"}'
        )
        assert len(recs) == 1

    def test_results_are_sorted_by_priority(self):
        reply = json.dumps([
            {"category": "a", "recommendation": "low one", "priority": "low"},
            {"category": "b", "recommendation": "high one", "priority": "high"},
            {"category": "c", "recommendation": "medium one", "priority": "medium"},
        ])
        assert [r["priority"] for r in ai_svc.parse_recommendations(reply)] == [
            "high", "medium", "low"
        ]

    def test_invalid_priority_defaults_to_medium(self):
        reply = json.dumps([
            {"category": "a", "recommendation": "x", "priority": "URGENT!!"}
        ])
        assert ai_svc.parse_recommendations(reply)[0]["priority"] == "medium"

    def test_entries_without_recommendation_text_are_dropped(self):
        reply = json.dumps([
            {"category": "a", "recommendation": ""},
            {"category": "b", "recommendation": "keep me"},
        ])
        recs = ai_svc.parse_recommendations(reply)
        assert len(recs) == 1 and recs[0]["recommendation"] == "keep me"

    def test_non_list_based_on_is_normalised(self):
        reply = json.dumps([
            {"category": "a", "recommendation": "x", "basedOn": "steps"}
        ])
        assert ai_svc.parse_recommendations(reply)[0]["basedOn"] == []

    def test_results_are_tagged_as_ai_generated(self):
        assert all(r["source"] == "ai" for r in ai_svc.parse_recommendations(VALID_REPLY))

    @pytest.mark.parametrize(
        "reply", ["", "   ", "抱歉，我无法回答。", "[", "null", "[]", "[1,2,3]"]
    )
    def test_unusable_replies_raise_aierror(self, reply):
        with pytest.raises(ai_svc.AIError):
            ai_svc.parse_recommendations(reply)


# --- providers --------------------------------------------------------------
class TestGeminiProvider:
    def test_successful_call(self, keys, monkeypatch):
        captured = {}

        def fake_post(self, url, **kwargs):
            captured["url"] = url
            captured["headers"] = kwargs.get("headers", {})
            captured["json"] = kwargs.get("json")
            return FakeResponse(200, gemini_payload("hello"))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        out = ai_svc.CATALOG["gemini-flash"].generate("prompt text")

        assert out.text == "hello"
        assert "gemini-flash-latest:generateContent" in captured["url"]
        assert captured["headers"]["X-goog-api-key"] == "test-gemini-key"
        assert captured["json"]["contents"][0]["parts"][0]["text"] == "prompt text"

    def test_http_error_becomes_aierror(self, keys, monkeypatch):
        monkeypatch.setattr(
            requests.Session, "post", lambda *a, **k: FakeResponse(429, text="rate limited")
        )
        with pytest.raises(ai_svc.AIError, match="429"):
            ai_svc.CATALOG["gemini-flash"].generate("p")

    def test_timeout_becomes_aierror(self, keys, monkeypatch):
        def boom(self, *a, **k):
            raise requests.Timeout("timed out")

        monkeypatch.setattr(requests.Session, "post", boom)
        with pytest.raises(ai_svc.AIError, match="请求失败"):
            ai_svc.CATALOG["gemini-flash"].generate("p")

    def test_unexpected_shape_becomes_aierror(self, keys, monkeypatch):
        monkeypatch.setattr(
            requests.Session, "post", lambda *a, **k: FakeResponse(200, {"unexpected": True})
        )
        with pytest.raises(ai_svc.AIError, match="响应格式异常"):
            ai_svc.CATALOG["gemini-flash"].generate("p")

    def test_missing_key_raises_before_any_request(self, no_keys, monkeypatch):
        def boom(self, *a, **k):
            raise AssertionError("must not issue a request without a key")

        monkeypatch.setattr(requests.Session, "post", boom)
        with pytest.raises(ai_svc.AIError, match="GEMINI_API_KEY"):
            ai_svc.CATALOG["gemini-flash"].generate("p")


class TestOpenAICompatProvider:
    def test_successful_call(self, keys, monkeypatch):
        captured = {}

        def fake_post(self, url, **kwargs):
            captured["url"] = url
            captured["headers"] = kwargs.get("headers", {})
            captured["json"] = kwargs.get("json")
            return FakeResponse(200, openai_payload("hi"))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        out = ai_svc.CATALOG["llama-70b"].generate("prompt text")

        assert out.text == "hi"
        assert out.upstream is None, "stock OpenAI replies carry no provider field"
        assert captured["url"].endswith("/chat/completions")
        assert captured["headers"]["Authorization"] == "Bearer test-nvidia-key"
        assert captured["json"]["model"] == "meta/llama-3.3-70b-instruct"

    def test_http_error_becomes_aierror(self, keys, monkeypatch):
        monkeypatch.setattr(
            requests.Session, "post", lambda *a, **k: FakeResponse(500, text="boom")
        )
        with pytest.raises(ai_svc.AIError, match="500"):
            ai_svc.CATALOG["llama-70b"].generate("p")


class TestGatewayProvider:
    """The self-hosted gateway: OpenAI-compatible, plus a `provider` field
    naming whichever upstream actually served the request."""

    def test_reports_the_upstream_that_answered(self, gateway, monkeypatch):
        payload = {**openai_payload("hi"), "provider": "nvidia"}
        monkeypatch.setattr(requests.Session, "post", lambda *a, **k: FakeResponse(200, payload))
        out = ai_svc.CATALOG["gateway"].generate("p")
        assert out.text == "hi"
        assert out.upstream == "nvidia"

    def test_targets_the_configured_base_url(self, gateway, monkeypatch):
        captured = {}

        def fake_post(self, url, **kwargs):
            captured["url"] = url
            captured["headers"] = kwargs.get("headers", {})
            captured["model"] = (kwargs.get("json") or {}).get("model")
            return FakeResponse(200, openai_payload("hi"))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        ai_svc.CATALOG["gateway"].generate("p")

        assert captured["url"] == "http://gw.test:5100/v1/chat/completions"
        assert captured["headers"]["Authorization"] == "Bearer test-gateway-token"
        assert captured["model"] == "ai-gateway-auto"

    def test_upstream_surfaces_in_generate_meta(self, gateway, monkeypatch):
        payload = {**openai_payload(VALID_REPLY), "provider": "gemini"}
        monkeypatch.setattr(requests.Session, "post", lambda *a, **k: FakeResponse(200, payload))
        _, meta = ai_svc.generate(SUMMARY)
        assert meta["model"] == "gateway"
        assert meta["upstream"] == "gemini"

    def test_gateway_401_falls_through(self, monkeypatch):
        """A stale gateway token must not strand the request."""
        monkeypatch.setenv("AI_GATEWAY_TOKEN", "expired")
        monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gw.test:5100/v1")
        monkeypatch.setenv("GEMINI_API_KEY", "k")

        def fake_post(self, url, **kwargs):
            if "gw.test" in url:
                return FakeResponse(401, text="unauthorized")
            return FakeResponse(200, gemini_payload(VALID_REPLY))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        _, meta = ai_svc.generate(SUMMARY)
        assert meta["model"] == "gemini-flash"
        assert meta["fallbackFrom"] == ["gateway"]


# --- proxy handling ---------------------------------------------------------
class TestProxyPolicy:
    """Overseas vendors may only be reachable through a local proxy, while a
    self-hosted box on a public IP breaks when forced through one — so the two
    must not share a policy."""

    def test_hosted_vendors_honour_environment_proxies(self):
        assert ai_svc.CATALOG["gemini-flash"].use_proxy is True
        assert ai_svc.CATALOG["llama-70b"].use_proxy is True

    def test_self_hosted_gateway_bypasses_proxies(self):
        assert ai_svc.CATALOG["gateway"].use_proxy is False

    def test_session_trust_env_follows_the_flag(self):
        """Regression: requests picked up ALL_PROXY and routed the gateway
        call through a local proxy, which timed out after 120s."""
        assert ai_svc.CATALOG["gateway"]._session().trust_env is False
        assert ai_svc.CATALOG["gemini-flash"]._session().trust_env is True


# --- output budget ----------------------------------------------------------
class TestMaxTokens:
    def test_default_applies_to_ordinary_models(self):
        assert ai_svc.CATALOG["gemini-flash"].max_tokens == ai_svc.FALLBACK_MAX_TOKENS

    def test_reasoning_endpoint_declares_a_larger_budget(self):
        """Regression: the gateway's primary upstream thinks out loud before
        answering; at the default cap the trace consumed the whole budget and
        the reply was truncated before any JSON appeared."""
        assert ai_svc.CATALOG["gateway"].max_tokens > ai_svc.FALLBACK_MAX_TOKENS

    def test_env_overrides_the_default_but_not_an_explicit_budget(self, monkeypatch):
        monkeypatch.setenv("AI_MAX_TOKENS", "77")
        assert ai_svc.CATALOG["gemini-flash"].max_tokens == 77
        assert ai_svc.CATALOG["gateway"].max_tokens == 3000

    def test_budget_reaches_the_openai_payload(self, gateway, monkeypatch):
        seen = {}

        def fake_post(self, url, **kwargs):
            seen["max_tokens"] = kwargs["json"]["max_tokens"]
            return FakeResponse(200, openai_payload("hi"))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        ai_svc.CATALOG["gateway"].generate("p")
        assert seen["max_tokens"] == 3000

    def test_budget_reaches_the_gemini_payload(self, keys, monkeypatch):
        seen = {}

        def fake_post(self, url, **kwargs):
            seen["cap"] = kwargs["json"]["generationConfig"]["maxOutputTokens"]
            return FakeResponse(200, gemini_payload("hi"))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        ai_svc.CATALOG["gemini-flash"].generate("p")
        assert seen["cap"] == ai_svc.FALLBACK_MAX_TOKENS


# --- lazily-read configuration ----------------------------------------------
class TestLazyConfig:
    """Regression: these were module-level constants, so they froze whatever
    the environment held at import — hiding config changes and letting a
    developer's .env leak into the test run."""

    def test_chain_reflects_the_current_environment(self, monkeypatch):
        monkeypatch.setenv("AI_MODEL_CHAIN", "llama-70b,gemini-flash")
        assert ai_svc.default_chain() == ["llama-70b", "gemini-flash"]
        monkeypatch.setenv("AI_MODEL_CHAIN", "gateway")
        assert ai_svc.default_chain() == ["gateway"]

    def test_timeout_reflects_the_current_environment(self, monkeypatch):
        monkeypatch.setenv("AI_TIMEOUT_SECONDS", "7")
        assert ai_svc.default_timeout() == 7.0

    def test_day_budget_reflects_the_current_environment(self, monkeypatch):
        monkeypatch.setenv("AI_DAY_BUDGET", "42")
        assert ai_svc.default_day_budget() == 42

    def test_defaults_apply_when_unset(self):
        assert ai_svc.default_timeout() == ai_svc.FALLBACK_TIMEOUT
        assert ai_svc.default_day_budget() == ai_svc.FALLBACK_DAY_BUDGET
        assert ai_svc.default_chain()[0] == "gateway"

    def test_default_flag_tracks_the_chain_head(self, monkeypatch, keys):
        monkeypatch.setenv("AI_MODEL_CHAIN", "llama-70b,gemini-flash")
        by_id = {m["id"]: m for m in ai_svc.list_models()}
        assert by_id["llama-70b"]["default"] is True
        assert by_id["gemini-flash"]["default"] is False


# --- context sizing ---------------------------------------------------------
class TestPerModelSizing:
    """Chain members' windows differ by >30x, so the payload is sized per
    model rather than once for the whole chain."""

    def test_small_window_gets_fewer_days_than_a_large_one(self):
        # A budget above what 128k can hold, so the window is what binds.
        budget = 100_000
        small = ai_svc.max_days_for(ai_svc.CATALOG["llama-70b"], budget)      # 128k
        large = ai_svc.max_days_for(ai_svc.CATALOG["gemini-flash"], budget)   # 1M
        assert small < large

    def test_budget_binds_when_it_is_the_tighter_limit(self):
        """At the default 365-day budget every model gets the same 365 days —
        no window in the catalog is small enough to bind first."""
        budget = ai_svc.default_day_budget()
        days = {
            mid: ai_svc.max_days_for(p, budget) for mid, p in ai_svc.CATALOG.items()
        }
        assert set(days.values()) == {budget}

    def test_never_exceeds_the_configured_budget(self):
        assert ai_svc.max_days_for(ai_svc.CATALOG["gemini-flash"], day_budget=30) == 30

    def test_always_allows_at_least_one_day(self):
        tiny = ai_svc.OpenAICompatProvider(
            model_id="tiny", context_window=10,
            base_url_env="X", default_base_url="http://x", requires_key=False,
        )
        assert ai_svc.max_days_for(tiny) >= 1

    def test_each_model_gets_a_prompt_sized_for_itself(self, keys, monkeypatch):
        """Regression: one prompt was built for the whole chain, so a payload
        sized for Gemini's 1M window was also sent to 128k models.

        Needs more days than the 128k window holds (~6.4k) for the trimming to
        bite, hence the deliberately oversized history.
        """
        history = [{"date": "2026-01-01", "steps": 8000} for _ in range(8000)]
        sizes = {}

        def fake_post(self, url, **kwargs):
            if "generativelanguage" in url:
                sizes["gemini"] = len(kwargs["json"]["contents"][0]["parts"][0]["text"])
                raise requests.Timeout("force fallback")
            sizes["nvidia"] = len(kwargs["json"]["messages"][0]["content"])
            return FakeResponse(200, openai_payload(VALID_REPLY))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        ai_svc.generate(history, day_budget=100_000)

        assert sizes["nvidia"] < sizes["gemini"], (
            "the 128k model must receive a smaller prompt than the 1M model"
        )


# --- catalog & chain --------------------------------------------------------
class TestCatalog:
    def test_all_models_listed(self, keys):
        assert {m["id"] for m in ai_svc.list_models()} == {
            "gateway", "gemini-flash", "llama-70b", "nemotron-49b", "mistral-large"
        }

    def test_configured_flag_tracks_the_environment(self, no_keys, monkeypatch):
        assert all(not m["configured"] for m in ai_svc.list_models())
        monkeypatch.setenv("GEMINI_API_KEY", "k")
        by_id = {m["id"]: m for m in ai_svc.list_models()}
        assert by_id["gemini-flash"]["configured"] is True
        assert by_id["llama-70b"]["configured"] is False

    def test_every_model_declares_a_large_window(self):
        assert all(m["contextWindow"] >= 128_000 for m in ai_svc.list_models())

    def test_no_vision_models_registered(self):
        assert not any("vision" in m["model"] for m in ai_svc.list_models())


class TestResolveChain:
    def test_preferred_model_goes_first(self, keys):
        assert ai_svc.resolve_chain("nemotron-49b")[0] == "nemotron-49b"

    def test_chain_has_no_duplicates(self, keys):
        chain = ai_svc.resolve_chain("gemini-flash")
        assert len(chain) == len(set(chain))

    def test_unconfigured_models_are_skipped(self, no_keys, monkeypatch):
        monkeypatch.setenv("GEMINI_API_KEY", "k")
        assert ai_svc.resolve_chain() == ["gemini-flash"]

    def test_unknown_model_raises(self, keys):
        with pytest.raises(ai_svc.AIError, match="未知模型"):
            ai_svc.resolve_chain("gpt-nonexistent")

    def test_no_credentials_raises_with_actionable_message(self, no_keys):
        with pytest.raises(ai_svc.AIError, match="GEMINI_API_KEY"):
            ai_svc.resolve_chain()

    def test_gateway_needs_both_token_and_base_url(self, no_keys, monkeypatch):
        monkeypatch.setenv("AI_GATEWAY_TOKEN", "t")
        assert ai_svc.CATALOG["gateway"].is_configured() is False
        monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gw.test:5100/v1")
        assert ai_svc.CATALOG["gateway"].is_configured() is True


# --- generate + fallback ----------------------------------------------------
class TestGenerate:
    def test_returns_recommendations_and_meta(self, keys, monkeypatch):
        monkeypatch.setattr(
            requests.Session, "post", lambda *a, **k: FakeResponse(200, gemini_payload(VALID_REPLY))
        )
        recs, meta = ai_svc.generate(SUMMARY)
        assert len(recs) == 2
        assert meta["model"] == "gemini-flash"
        assert meta["days"] == 2
        assert meta["fallbackFrom"] == []

    def test_falls_back_to_the_next_model(self, keys, monkeypatch):
        calls = []

        def fake_post(self, url, **kwargs):
            calls.append(url)
            if "generativelanguage" in url:
                raise requests.Timeout("gemini down")
            return FakeResponse(200, openai_payload(VALID_REPLY))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        recs, meta = ai_svc.generate(SUMMARY)

        assert len(recs) == 2
        assert meta["model"] == "llama-70b"
        assert meta["fallbackFrom"] == ["gemini-flash"]
        assert len(calls) == 2

    def test_falls_back_when_a_model_returns_unparseable_text(self, keys, monkeypatch):
        def fake_post(self, url, **kwargs):
            if "generativelanguage" in url:
                return FakeResponse(200, gemini_payload("抱歉，我帮不了你。"))
            return FakeResponse(200, openai_payload(VALID_REPLY))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        _, meta = ai_svc.generate(SUMMARY)
        assert meta["model"] == "llama-70b"

    def test_raises_when_every_model_fails(self, keys, monkeypatch):
        def boom(self, *a, **k):
            raise requests.Timeout("all down")

        monkeypatch.setattr(requests.Session, "post", boom)
        with pytest.raises(ai_svc.AIError, match="所有模型均失败"):
            ai_svc.generate(SUMMARY)

    def test_preferred_model_is_honoured(self, keys, monkeypatch):
        seen = {}

        def fake_post(self, url, **kwargs):
            seen["model"] = (kwargs.get("json") or {}).get("model")
            return FakeResponse(200, openai_payload(VALID_REPLY))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        _, meta = ai_svc.generate(SUMMARY, preferred_model="nemotron-49b")
        assert meta["model"] == "nemotron-49b"
        assert seen["model"] == "nvidia/llama-3.3-nemotron-super-49b-v1.5"

    def test_no_second_call_after_the_first_succeeds(self, keys, monkeypatch):
        calls = []

        def fake_post(self, url, **kwargs):
            calls.append(url)
            return FakeResponse(200, gemini_payload(VALID_REPLY))

        monkeypatch.setattr(requests.Session, "post", fake_post)
        ai_svc.generate(SUMMARY)
        assert len(calls) == 1


# --- service + endpoint integration -----------------------------------------
class TestAiRecommendationsService:
    def test_uses_the_rule_engine_when_there_is_no_data(self, db, user, keys):
        out = analysis_svc.get_ai_recommendations(user["id"])
        assert out["meta"]["source"] == "rules"
        assert out["recommendations"][0]["id"] == "no-data"

    def test_returns_ai_results_when_a_model_answers(
        self, seed_health, user, keys, monkeypatch
    ):
        seed_health([{"date": "2026-08-20", "steps": 5000}])
        monkeypatch.setattr(
            requests.Session, "post", lambda *a, **k: FakeResponse(200, gemini_payload(VALID_REPLY))
        )
        out = analysis_svc.get_ai_recommendations(user["id"])
        assert out["meta"]["source"] == "ai"
        assert len(out["recommendations"]) == 2

    def test_degrades_to_rules_when_all_models_fail(
        self, seed_health, user, keys, monkeypatch
    ):
        seed_health([{"date": "2026-08-20", "steps": 5000}])

        def boom(self, *a, **k):
            raise requests.Timeout("down")

        monkeypatch.setattr(requests.Session, "post", boom)
        out = analysis_svc.get_ai_recommendations(user["id"])
        assert out["meta"]["source"] == "rules"
        assert "所有模型均失败" in out["meta"]["reason"]
        assert out["recommendations"], "must still return rule-based advice"

    def test_degrades_to_rules_when_no_key_is_configured(
        self, seed_health, user, no_keys
    ):
        seed_health([{"date": "2026-08-20", "steps": 5000}])
        out = analysis_svc.get_ai_recommendations(user["id"])
        assert out["meta"]["source"] == "rules"
        assert "GEMINI_API_KEY" in out["meta"]["reason"]


class TestEndpoints:
    def test_models_requires_auth(self, client):
        assert client.get("/api/analysis/models").status_code == 401

    def test_ai_recommendations_requires_auth(self, client):
        assert client.get("/api/analysis/ai-recommendations").status_code == 401

    def test_models_endpoint_lists_catalog(self, client, auth, keys):
        r = client.get("/api/analysis/models", headers=auth)
        assert r.status_code == 200
        assert {m["id"] for m in r.get_json()} >= {"gemini-flash", "llama-70b"}

    def test_models_endpoint_never_leaks_api_keys(self, client, auth, keys):
        body = client.get("/api/analysis/models", headers=auth).get_data(as_text=True)
        assert "test-gemini-key" not in body
        assert "test-nvidia-key" not in body

    def test_ai_endpoint_returns_200_even_with_no_models(self, client, auth, no_keys):
        r = client.get("/api/analysis/ai-recommendations", headers=auth)
        assert r.status_code == 200
        assert r.get_json()["meta"]["source"] == "rules"

    def test_ai_endpoint_passes_model_param_through(
        self, client, auth, seed_health, keys, monkeypatch
    ):
        seed_health([{"date": "2026-08-20", "steps": 5000}])
        monkeypatch.setattr(
            requests.Session, "post", lambda *a, **k: FakeResponse(200, openai_payload(VALID_REPLY))
        )
        r = client.get(
            "/api/analysis/ai-recommendations?model=nemotron-49b", headers=auth
        )
        assert r.get_json()["meta"]["model"] == "nemotron-49b"

    def test_unknown_model_param_degrades_to_rules(
        self, client, auth, seed_health, keys
    ):
        seed_health([{"date": "2026-08-20", "steps": 5000}])
        r = client.get("/api/analysis/ai-recommendations?model=bogus", headers=auth)
        assert r.status_code == 200
        assert r.get_json()["meta"]["source"] == "rules"