GarminHealthLab/backend/tests/test_ai_cache.py

"""
Unit tests for the AI recommendation cache.

A generation costs minutes against a large reasoning model, so the result is
stored and reused. These tests pin when it is reused and — more importantly —
when it must not be.
"""
import datetime
import json

import pytest
import requests

from services import analysis as analysis_svc
from services import health as health_svc


VALID_REPLY = json.dumps(
    [{"category": "睡眠", "recommendation": "早点睡。", "priority": "high",
      "basedOn": ["sleep_duration"]}],
    ensure_ascii=False,
)


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


@pytest.fixture
def keys(monkeypatch):
    monkeypatch.setenv("GEMINI_API_KEY", "test-key")
    return True


@pytest.fixture
def counting_llm(monkeypatch):
    """Mock the LLM and count how many times it is actually called."""
    calls = []

    def fake_post(self, url, **kwargs):
        calls.append(url)
        return FakeResponse(
            200, {"candidates": [{"content": {"parts": [{"text": VALID_REPLY}]}}]}
        )

    monkeypatch.setattr(requests.Session, "post", fake_post)
    return calls


@pytest.fixture
def seeded(seed_health, user):
    seed_health([{"date": "2026-08-20", "steps": 5000, "sleep_duration": 6}])
    return user


class TestCacheHit:
    def test_first_call_reaches_the_model(self, seeded, keys, counting_llm):
        out = analysis_svc.get_ai_recommendations(seeded["id"])
        assert out["meta"]["source"] == "ai"
        assert out["meta"]["cached"] is False
        assert len(counting_llm) == 1

    def test_second_call_is_served_from_cache(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        out = analysis_svc.get_ai_recommendations(seeded["id"])

        assert len(counting_llm) == 1, "the model must not be called twice"
        assert out["meta"]["cached"] is True
        assert out["meta"]["source"] == "ai"

    def test_cached_result_matches_the_generated_one(self, seeded, keys, counting_llm):
        first = analysis_svc.get_ai_recommendations(seeded["id"])
        second = analysis_svc.get_ai_recommendations(seeded["id"])
        assert first["recommendations"] == second["recommendations"]

    def test_cache_records_which_model_answered(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        out = analysis_svc.get_ai_recommendations(seeded["id"])
        assert out["meta"]["model"] == "gemini-flash"

    def test_cache_reports_when_it_was_generated(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        out = analysis_svc.get_ai_recommendations(seeded["id"])
        assert out["meta"]["generatedAt"]


class TestCacheInvalidation:
    def test_new_health_data_invalidates(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        health_svc.upsert_health_daily(
            seeded["id"], {"date": "2026-08-21", "steps": 9000}
        )
        out = analysis_svc.get_ai_recommendations(seeded["id"])

        assert len(counting_llm) == 2, "a new day of data must trigger a regeneration"
        assert out["meta"]["cached"] is False

    def test_corrected_value_invalidates(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        # Same date, different step count — a re-sync correcting a value.
        health_svc.upsert_health_daily(
            seeded["id"], {"date": "2026-08-20", "steps": 12345, "sleepDuration": 6}
        )
        analysis_svc.get_ai_recommendations(seeded["id"])
        assert len(counting_llm) == 2

    def test_new_activity_invalidates(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        health_svc.insert_activity(
            seeded["id"],
            {"activityType": "running", "startTime": "2026-08-20T07:00:00",
             "endTime": "2026-08-20T07:30:00"},
        )
        analysis_svc.get_ai_recommendations(seeded["id"])
        assert len(counting_llm) == 2

    def test_refresh_bypasses_the_cache(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        out = analysis_svc.get_ai_recommendations(seeded["id"], refresh=True)
        assert len(counting_llm) == 2
        assert out["meta"]["cached"] is False

    def test_explicit_model_bypasses_the_cache(self, seeded, keys, counting_llm):
        """Asking for a named model means wanting that model's answer."""
        analysis_svc.get_ai_recommendations(seeded["id"])
        analysis_svc.get_ai_recommendations(seeded["id"], model="gemini-flash")
        assert len(counting_llm) == 2

    def test_expired_entry_is_regenerated(self, seeded, keys, counting_llm, db):
        analysis_svc.get_ai_recommendations(seeded["id"])
        stale = (
            datetime.datetime.utcnow()
            - datetime.timedelta(hours=analysis_svc.CACHE_TTL_HOURS + 1)
        ).isoformat(timespec="seconds")
        db.execute(
            "UPDATE ai_recommendations SET created_at = ? WHERE user_id = ?",
            [stale, seeded["id"]],
        )
        analysis_svc.get_ai_recommendations(seeded["id"])
        assert len(counting_llm) == 2

    def test_entry_just_inside_the_ttl_is_kept(self, seeded, keys, counting_llm, db):
        analysis_svc.get_ai_recommendations(seeded["id"])
        fresh = (
            datetime.datetime.utcnow()
            - datetime.timedelta(hours=analysis_svc.CACHE_TTL_HOURS - 1)
        ).isoformat(timespec="seconds")
        db.execute(
            "UPDATE ai_recommendations SET created_at = ? WHERE user_id = ?",
            [fresh, seeded["id"]],
        )
        analysis_svc.get_ai_recommendations(seeded["id"])
        assert len(counting_llm) == 1

    def test_clear_cache_forces_regeneration(self, seeded, keys, counting_llm):
        analysis_svc.get_ai_recommendations(seeded["id"])
        analysis_svc.clear_ai_cache(seeded["id"])
        analysis_svc.get_ai_recommendations(seeded["id"])
        assert len(counting_llm) == 2


class TestIsolationAndRobustness:
    def test_cache_is_per_user(self, seeded, keys, counting_llm, db, make_user):
        analysis_svc.get_ai_recommendations(seeded["id"])

        other = make_user("other@example.com")
        health_svc.upsert_health_daily(
            other["id"], {"date": "2026-08-20", "steps": 5000, "sleepDuration": 6}
        )

        analysis_svc.get_ai_recommendations(other["id"])
        assert len(counting_llm) == 2, "one user's cache must not answer another's"

    def test_only_one_row_per_user(self, seeded, keys, counting_llm, db):
        for _ in range(3):
            analysis_svc.get_ai_recommendations(seeded["id"], refresh=True)
        rows = db.query_all(
            "SELECT * FROM ai_recommendations WHERE user_id = ?", [seeded["id"]]
        )
        assert len(rows) == 1, "regeneration must replace, not accumulate"

    def test_corrupt_payload_regenerates_instead_of_raising(
        self, seeded, keys, counting_llm, db
    ):
        analysis_svc.get_ai_recommendations(seeded["id"])
        db.execute(
            "UPDATE ai_recommendations SET payload = ? WHERE user_id = ?",
            ["not json", seeded["id"]],
        )
        out = analysis_svc.get_ai_recommendations(seeded["id"])
        assert len(counting_llm) == 2
        assert out["recommendations"]

    def test_rule_fallback_is_not_cached(self, seeded, monkeypatch, db):
        """A degraded answer must not be stored as if it were the AI's."""
        def boom(self, *a, **k):
            raise requests.Timeout("down")

        monkeypatch.setattr(requests.Session, "post", boom)
        out = analysis_svc.get_ai_recommendations(seeded["id"])

        assert out["meta"]["source"] == "rules"
        assert db.query_one(
            "SELECT * FROM ai_recommendations WHERE user_id = ?", [seeded["id"]]
        ) is None

    def test_no_data_user_is_not_cached(self, user, keys, counting_llm, db):
        analysis_svc.get_ai_recommendations(user["id"])
        assert len(counting_llm) == 0
        assert db.query_one(
            "SELECT * FROM ai_recommendations WHERE user_id = ?", [user["id"]]
        ) is None


class TestEndpoint:
    def test_second_request_is_cached(self, client, auth, seeded, keys, counting_llm):
        client.get("/api/analysis/ai-recommendations", headers=auth)
        r = client.get("/api/analysis/ai-recommendations", headers=auth)
        assert r.get_json()["meta"]["cached"] is True
        assert len(counting_llm) == 1

    def test_refresh_param_forces_regeneration(
        self, client, auth, seeded, keys, counting_llm
    ):
        client.get("/api/analysis/ai-recommendations", headers=auth)
        r = client.get("/api/analysis/ai-recommendations?refresh=1", headers=auth)
        assert r.get_json()["meta"]["cached"] is False
        assert len(counting_llm) == 2