GarminHealthLab/backend/tests/test_garmin_sync.py

"""
Unit tests for the Garmin sync service.

A stub client stands in for `garminconnect`, so the suite runs without the
library, without credentials and without touching Garmin.

The stub mirrors the real 0.2.8 API shapes on purpose — the original code
called `get_activities(date)` when that method actually takes `(start, limit)`
pagination arguments, a mistake that only surfaces once something exercises it.
"""
import datetime

import pytest

from services import garmin as garmin_svc
from services import health as health_svc


def day(offset=0):
    return (datetime.date.today() - datetime.timedelta(days=offset)).isoformat()


def summary(steps=8000, rhr=60, stress=40, kcal=2200):
    return {
        "totalSteps": steps,
        "restingHeartRate": rhr,
        "averageStressLevel": stress,
        "totalKilocalories": kcal,
    }


def sleep(hours=7.5, score=82):
    return {
        "dailySleepDTO": {
            "sleepTimeSeconds": int(hours * 3600),
            "sleepScores": {"overall": {"value": score}},
        }
    }


def hrv(value=48):
    return {"hrvSummary": {"lastNightAvg": value}}


def activity(activity_id=1001, type_key="running", duration=1800):
    return {
        "activityId": activity_id,
        "activityType": {"typeKey": type_key},
        "startTimeLocal": f"{day()}T07:00:00",
        "duration": duration,
        "distance": 5000.0,
        "calories": 320.0,
        "averageHR": 145,
        "maxHR": 168,
    }


class StubClient:
    """Stands in for garminconnect.Garmin, recording how it was called."""

    def __init__(self, summaries=None, sleeps=None, hrvs=None, activities=None,
                 fail_days=(), fail_activities=False):
        self._summaries = summaries if summaries is not None else {}
        self._sleeps = sleeps if sleeps is not None else {}
        self._hrvs = hrvs if hrvs is not None else {}
        self._activities = activities if activities is not None else []
        self._fail_days = set(fail_days)
        self._fail_activities = fail_activities
        self.calls = []

    def get_user_summary(self, cdate):
        self.calls.append(("summary", cdate))
        if cdate in self._fail_days:
            raise RuntimeError(f"upstream error for {cdate}")
        return self._summaries.get(cdate, summary())

    def get_sleep_data(self, cdate):
        self.calls.append(("sleep", cdate))
        return self._sleeps.get(cdate, sleep())

    def get_hrv_data(self, cdate):
        self.calls.append(("hrv", cdate))
        return self._hrvs.get(cdate, hrv())

    def get_activities_by_date(self, startdate, enddate, activitytype=None):
        self.calls.append(("activities", startdate, enddate))
        if self._fail_activities:
            raise RuntimeError("activities endpoint down")
        return self._activities


CREDS = {"garminEmail": "g@example.com", "garminPassword": "pw"}


class TestHappyPath:
    def test_reports_success(self, db, user):
        out = garmin_svc.sync_data(user["id"], CREDS, days=3, client=StubClient())
        assert out["status"] == "success"
        assert out["recordsSynced"] == 3

    def test_stores_the_days(self, db, user):
        garmin_svc.sync_data(user["id"], CREDS, days=3, client=StubClient())
        rows = health_svc.get_summary(user["id"])
        assert len(rows) == 3

    def test_maps_every_metric(self, db, user):
        client = StubClient(
            summaries={day(): summary(steps=9500, rhr=57, stress=33, kcal=2450)},
            sleeps={day(): sleep(hours=8.0, score=91)},
            hrvs={day(): hrv(52)},
        )
        garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
        row = health_svc.get_summary(user["id"])[0]

        assert row["steps"] == 9500
        assert row["heartRate"] == 57
        assert row["stress"] == 33
        assert row["caloriesBurned"] == 2450
        assert row["heartRateVariability"] == 52
        assert row["sleep"]["duration"] == 8.0
        assert row["sleep"]["quality"] == 91

    def test_sleep_and_hrv_come_from_their_own_endpoints(self, db, user):
        """Regression: both live outside get_user_summary. Reading only the
        summary recorded every night as having no sleep data."""
        client = StubClient()
        garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)

        kinds = {c[0] for c in client.calls}
        assert "sleep" in kinds
        assert "hrv" in kinds

    def test_seconds_are_converted_to_hours(self, db, user):
        client = StubClient(sleeps={day(): sleep(hours=6.5)})
        garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
        assert health_svc.get_summary(user["id"])[0]["sleep"]["duration"] == 6.5


class TestActivities:
    def test_fetched_by_date_range_in_one_call(self, db, user):
        """Regression: the old code called get_activities(date), but that
        method takes (start, limit) pagination arguments, not a date."""
        client = StubClient(activities=[activity()])
        garmin_svc.sync_data(user["id"], CREDS, days=7, client=client)

        activity_calls = [c for c in client.calls if c[0] == "activities"]
        assert len(activity_calls) == 1, "one range call, not one call per day"
        _, start, end = activity_calls[0]
        assert start == day(6) and end == day(0)

    def test_stored_with_fields_mapped(self, db, user):
        garmin_svc.sync_data(
            user["id"], CREDS, days=1, client=StubClient(activities=[activity()])
        )
        rows = health_svc.get_activities(user["id"])
        assert len(rows) == 1
        assert rows[0]["activity_type"] == "running"
        assert rows[0]["heart_rate_average"] == 145

    def test_count_is_reported(self, db, user):
        client = StubClient(activities=[activity(1), activity(2)])
        out = garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
        assert out["activitiesSynced"] == 2

    def test_resync_does_not_duplicate(self, db, user):
        """Garmin activity ids are stable, so a re-synced window must skip
        what is already stored."""
        client = StubClient(activities=[activity(1001)])
        garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
        out = garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)

        assert len(health_svc.get_activities(user["id"])) == 1
        assert out["activitiesSynced"] == 0

    def test_end_time_derived_from_duration(self, db, user):
        client = StubClient(activities=[activity(duration=1800)])
        garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
        row = health_svc.get_activities(user["id"])[0]
        assert row["start_time"] != row["end_time"]

    def test_failure_does_not_lose_the_daily_data(self, db, user):
        out = garmin_svc.sync_data(
            user["id"], CREDS, days=2, client=StubClient(fail_activities=True)
        )
        assert out["status"] == "success"
        assert len(health_svc.get_summary(user["id"])) == 2


class TestResync:
    def test_same_day_is_updated_not_duplicated(self, db, user):
        garmin_svc.sync_data(
            user["id"], CREDS, days=1,
            client=StubClient(summaries={day(): summary(steps=5000)}),
        )
        garmin_svc.sync_data(
            user["id"], CREDS, days=1,
            client=StubClient(summaries={day(): summary(steps=9000)}),
        )
        rows = health_svc.get_summary(user["id"])
        assert len(rows) == 1
        assert rows[0]["steps"] == 9000


class TestPartialAndTotalFailure:
    def test_one_bad_day_is_skipped_not_fatal(self, db, user):
        client = StubClient(fail_days=[day(1)])
        out = garmin_svc.sync_data(user["id"], CREDS, days=3, client=client)

        assert out["status"] == "success"
        assert out["recordsSynced"] == 2
        assert "跳过" in out["message"]

    def test_every_day_failing_is_reported_as_an_error(self, db, user):
        """A systemic failure reported as a clean success would hide it."""
        client = StubClient(fail_days=[day(0), day(1), day(2)])
        out = garmin_svc.sync_data(user["id"], CREDS, days=3, client=client)

        assert out["status"] == "error"
        assert out["recordsSynced"] == 0

    def test_login_failure_is_reported(self, db, user, monkeypatch):
        def boom(_creds, _uid=None):
            raise RuntimeError("401 Unauthorized")

        monkeypatch.setattr(garmin_svc, "_connect", boom)
        out = garmin_svc.sync_data(user["id"], CREDS, days=1)

        assert out["status"] == "error"
        assert "401" in out["message"]

    def test_days_without_data_are_not_stored(self, db, user):
        """Garmin returns all-None for a day it has nothing for; an empty row
        would just have to be filtered back out by every read endpoint."""
        client = StubClient(
            summaries={day(): {}}, sleeps={day(): {}}, hrvs={day(): {}}
        )
        out = garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
        assert out["recordsSynced"] == 0
        assert health_svc.get_summary(user["id"]) == []


class TestSyncStatus:
    def test_idle_before_any_sync(self, db, user):
        assert garmin_svc.get_sync_status(user["id"])["status"] == "idle"

    def test_success_leaves_status_idle(self, db, user):
        garmin_svc.sync_data(user["id"], CREDS, days=1, client=StubClient())
        status = garmin_svc.get_sync_status(user["id"])
        assert status["status"] == "idle"
        assert status["recordsSynced"] == 1
        assert status["lastSyncTime"]

    def test_total_failure_leaves_status_error(self, db, user):
        garmin_svc.sync_data(
            user["id"], CREDS, days=2, client=StubClient(fail_days=[day(0), day(1)])
        )
        status = garmin_svc.get_sync_status(user["id"])
        assert status["status"] == "error"
        assert status["lastError"]

    def test_a_later_success_clears_the_error(self, db, user):
        garmin_svc.sync_data(
            user["id"], CREDS, days=1, client=StubClient(fail_days=[day(0)])
        )
        garmin_svc.sync_data(user["id"], CREDS, days=1, client=StubClient())
        status = garmin_svc.get_sync_status(user["id"])
        assert status["status"] == "idle"
        assert not status["lastError"]


class TestEndpoint:
    def test_requires_auth(self, client):
        assert client.post("/api/garmin/sync", json={}).status_code == 401

    def test_missing_password_is_refused_with_a_reason(self, client, auth):
        r = client.post("/api/garmin/sync", headers=auth, json={})
        assert r.status_code == 400
        assert "garminPassword" in r.get_json()["message"]

    def test_status_endpoint(self, client, auth):
        r = client.get("/api/garmin/status", headers=auth)
        assert r.status_code == 200
        assert r.get_json()["status"] == "idle"


class TestTokenStore:
    """Tokens are what make unattended sync possible on an MFA-protected
    account: the web worker has no stdin to type a code into."""

    def test_absent_before_any_login(self, db, user):
        assert garmin_svc.has_token(user["id"]) is False

    def test_saved_token_round_trips(self, db, user):
        garmin_svc.save_token(user["id"], "token-blob", "g@example.com")
        assert garmin_svc.has_token(user["id"]) is True
        assert garmin_svc.load_token(user["id"]) == "token-blob"

    def test_re_login_replaces_rather_than_accumulates(self, db, user):
        garmin_svc.save_token(user["id"], "first", "g@example.com")
        garmin_svc.save_token(user["id"], "second", "g@example.com")

        rows = db.query_all(
            "SELECT * FROM garmin_tokens WHERE user_id = ?", [user["id"]]
        )
        assert len(rows) == 1
        assert garmin_svc.load_token(user["id"]) == "second"

    def test_tokens_are_per_user(self, db, user, client):
        garmin_svc.save_token(user["id"], "mine", "g@example.com")
        other = client.post(
            "/api/auth/register",
            json={"email": "o@example.com", "garminEmail": "og@example.com",
                  "garminPassword": "pw123456"},
        ).get_json()
        assert garmin_svc.has_token(other["id"]) is False


class TestMfaHandling:
    def test_eof_from_the_mfa_prompt_becomes_an_actionable_error(
        self, db, user, monkeypatch
    ):
        """garth's default MFA prompt calls input(); with no stdin that raises
        a bare EOFError, which says nothing about what to do about it."""
        class StubGarth:
            def loads(self, s): pass
            def refresh_oauth2(self): pass

        class StubGarmin:
            def __init__(self, *a, **k):
                self.garth = StubGarth()
                self.username = None
                self.password = None

            def login(self, *a, **k):
                raise EOFError("EOF when reading a line")

        monkeypatch.setattr(garmin_svc, "_import_garmin", lambda: StubGarmin)
        with pytest.raises(garmin_svc.MFARequired, match="两步验证"):
            garmin_svc._connect(CREDS, user["id"])

    def test_sync_flags_mfa_so_the_ui_can_explain(self, db, user, monkeypatch):
        def boom(_creds, _uid=None):
            raise garmin_svc.MFARequired("需要两步验证")

        monkeypatch.setattr(garmin_svc, "_connect", boom)
        out = garmin_svc.sync_data(user["id"], CREDS, days=1)

        assert out["status"] == "error"
        assert out["mfaRequired"] is True

    def test_ordinary_failures_are_not_flagged_as_mfa(self, db, user, monkeypatch):
        def boom(_creds, _uid=None):
            raise RuntimeError("401 Unauthorized")

        monkeypatch.setattr(garmin_svc, "_connect", boom)
        assert garmin_svc.sync_data(user["id"], CREDS, days=1)["mfaRequired"] is False

    def test_stored_token_is_used_instead_of_logging_in(self, db, user, monkeypatch):
        garmin_svc.save_token(user["id"], "saved-blob", "g@example.com")
        loaded = {}

        class StubGarth:
            profile = {"displayName": "Tester"}

            def loads(self, s):
                loaded["blob"] = s

            def refresh_oauth2(self):
                loaded["refreshed"] = True

        class StubGarmin:
            def __init__(self, *a, **k):
                self.garth = StubGarth()

            def login(self, *a, **k):
                raise AssertionError("must not log in when a token exists")

        monkeypatch.setattr(garmin_svc, "_import_garmin", lambda: StubGarmin)
        garmin_svc._connect({}, user["id"])

        assert loaded["blob"] == "saved-blob"
        assert loaded["refreshed"] is True

    def test_no_token_and_no_password_is_refused_clearly(self, db, user, monkeypatch):
        class StubGarmin:
            def __init__(self, *a, **k):
                self.garth = None

        monkeypatch.setattr(garmin_svc, "_import_garmin", lambda: StubGarmin)
        with pytest.raises(RuntimeError, match="缺少 Garmin 密码"):
            garmin_svc._connect({}, user["id"])


class TestAuthStatusEndpoint:
    def test_requires_auth(self, client):
        assert client.get("/api/garmin/auth-status").status_code == 401

    def test_reports_false_then_true(self, client, auth, user, db):
        assert client.get("/api/garmin/auth-status", headers=auth).get_json()[
            "hasToken"] is False
        garmin_svc.save_token(user["id"], "blob", "g@example.com")
        assert client.get("/api/garmin/auth-status", headers=auth).get_json()[
            "hasToken"] is True

    def test_sync_without_password_allowed_once_a_token_exists(
        self, client, auth, user, db
    ):
        """The password field exists only because no token is stored yet."""
        garmin_svc.save_token(user["id"], "blob", "g@example.com")
        r = client.post("/api/garmin/sync", headers=auth, json={})
        assert r.status_code != 400


class TestApiUserAgent:
    """Regression: garth keeps its browser User-Agent after login, and the
    Garmin data API answers that UA with HTTP 200 and an empty array — every
    endpoint silently returns nothing."""

    def test_a_browser_user_agent_is_not_used_for_the_api(self):
        assert "Mozilla" not in garmin_svc.API_USER_AGENT
        assert "iPhone" not in garmin_svc.API_USER_AGENT

    def test_header_is_swapped_after_loading_a_token(self, db, user):
        garmin_svc.save_token(user["id"], "blob", "g@example.com")
        headers = {"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_5)"}

        class StubSess:
            def __init__(self): self.headers = headers

        class StubGarth:
            profile = {"displayName": "Tester"}
            def __init__(self): self.sess = StubSess()
            def loads(self, s): pass
            def refresh_oauth2(self): pass

        class StubGarmin:
            def __init__(self, *a, **k): self.garth = StubGarth()

        monkey = pytest.MonkeyPatch()
        monkey.setattr(garmin_svc, "_import_garmin", lambda: StubGarmin)
        try:
            garmin_svc._connect({}, user["id"])
        finally:
            monkey.undo()

        assert headers["User-Agent"] == garmin_svc.API_USER_AGENT

    def test_swap_is_harmless_on_a_client_without_a_session(self):
        class Bare:
            garth = object()

        garmin_svc._use_api_user_agent(Bare())  # must not raise

    def test_display_name_is_populated(self, db, user):
        """garminconnect builds URLs from display_name; unset sends every
        request to '.../None'."""
        class StubGarth:
            profile = {"displayName": "Tester"}
            sess = type("S", (), {"headers": {}})()
            def loads(self, s): pass
            def refresh_oauth2(self): pass

        class StubGarmin:
            def __init__(self, *a, **k): self.garth = StubGarth()

        garmin_svc.save_token(user["id"], "blob", "g@example.com")
        monkey = pytest.MonkeyPatch()
        monkey.setattr(garmin_svc, "_import_garmin", lambda: StubGarmin)
        try:
            client = garmin_svc._connect({}, user["id"])
        finally:
            monkey.undo()

        assert client.display_name == "Tester"


class TestErrorMessagesAreNeverEmpty:
    """Regression: a bare `assert` inside garth raised AssertionError with an
    empty str(), which was stored as the sync's reason — a failed sync with a
    blank explanation cannot be diagnosed."""

    def test_exception_without_text_still_describes_itself(self):
        assert garmin_svc.describe(AssertionError()) == "AssertionError"

    def test_exception_with_text_keeps_it(self):
        assert "boom" in garmin_svc.describe(RuntimeError("boom"))
        assert "RuntimeError" in garmin_svc.describe(RuntimeError("boom"))

    def test_whitespace_only_text_is_treated_as_empty(self):
        assert garmin_svc.describe(ValueError("   ")) == "ValueError"

    def test_connect_failure_records_a_non_empty_reason(self, db, user, monkeypatch):
        def boom(_creds, _uid=None):
            raise AssertionError()  # no message at all

        monkeypatch.setattr(garmin_svc, "_connect", boom)
        out = garmin_svc.sync_data(user["id"], CREDS, days=1)

        assert out["status"] == "error"
        assert out["message"].strip()
        assert garmin_svc.get_sync_status(user["id"])["lastError"].strip()


class TestTimestampNormalisation:
    """Regression: Garmin mixes ISO strings and epoch milliseconds in one
    payload. Writing the numeric form to a DATETIME column is rejected, which
    failed the entire personal-records batch."""

    def test_iso_string_passes_through(self):
        assert garmin_svc._to_datetime("2019-10-13T10:10:12.0").startswith(
            "2019-10-13T10:10:12"
        )

    def test_epoch_milliseconds_are_converted(self):
        assert garmin_svc._to_datetime(1570961412000).startswith("2019-10-13")

    def test_epoch_seconds_are_converted(self):
        assert garmin_svc._to_datetime(1570961412).startswith("2019-10-13")

    def test_first_usable_value_wins(self):
        assert garmin_svc._to_datetime(None, "", "2020-01-01T00:00:00") == (
            "2020-01-01T00:00:00"
        )

    def test_all_empty_gives_none(self):
        assert garmin_svc._to_datetime(None, "") is None

    def test_nonsense_value_does_not_raise(self):
        assert garmin_svc._to_datetime(float("inf")) is None