GarminHealthLab/backend/tests/test_garmin_mfa.py

"""
Unit tests for the web-driven two-factor Garmin login.

No network and no real library: a stub Garmin/garth pair stands in, and its
`prompt_mfa` callback is invoked exactly the way garth invokes it — blocking,
mid-login — because that blocking behaviour is the whole reason this flow
needs a background thread and a database rendezvous.
"""
import time

import pytest

from services import garmin as garmin_svc
from services import garmin_auth


def wait_for(predicate, timeout=8.0, interval=0.05):
    """Poll until the background thread reaches the expected state."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        value = predicate()
        if value:
            return value
        time.sleep(interval)
    return None


def status_of(session_id):
    row = garmin_auth.get_session(session_id)
    return row["status"] if row else None


def wait_status(session_id, *wanted, timeout=8.0):
    return wait_for(lambda: status_of(session_id) in wanted and status_of(session_id),
                    timeout=timeout)


class StubGarth:
    profile = {"displayName": "Tester", "fullName": "Test User"}

    def __init__(self, needs_mfa=True, accept_code="123456", fail_login=False):
        self.needs_mfa = needs_mfa
        self.accept_code = accept_code
        self.fail_login = fail_login
        self.seen_code = None
        self.logged_in_with = None

    def login(self, email, password, prompt_mfa=None):
        self.logged_in_with = (email, password)
        if self.fail_login:
            raise RuntimeError("401 Unauthorized")
        if self.needs_mfa:
            # garth calls this synchronously, in the middle of the login.
            self.seen_code = prompt_mfa()
            if self.seen_code != self.accept_code:
                raise RuntimeError("验证码错误")

    def dumps(self):
        return "token-blob"


class StubGarmin:
    """Class factory: `make()` returns something usable as `Garmin`."""

    last = None

    @classmethod
    def make(cls, **garth_kwargs):
        def factory(is_cn=False, **_):
            instance = cls()
            instance.garth = StubGarth(**garth_kwargs)
            cls.last = instance
            return instance
        return lambda: factory


@pytest.fixture(autouse=True)
def _fast_polling(monkeypatch):
    """Keep the rendezvous poll short so tests stay quick."""
    monkeypatch.setattr(garmin_auth, "POLL_INTERVAL_SECONDS", 0.05)
    monkeypatch.setattr(garmin_auth, "CODE_WAIT_SECONDS", 5)


class TestMfaFlow:
    def test_login_parks_waiting_for_a_code(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        assert wait_status(sid, "awaiting_code") == "awaiting_code"

    def test_submitting_the_code_completes_the_login(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        wait_status(sid, "awaiting_code")

        ok, _ = garmin_auth.submit_code(sid, user["id"], "123456")
        assert ok is True
        assert wait_status(sid, "done", "failed") == "done"

    def test_token_is_saved_on_success(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        wait_status(sid, "awaiting_code")
        garmin_auth.submit_code(sid, user["id"], "123456")
        wait_status(sid, "done", "failed")

        assert garmin_svc.has_token(user["id"]) is True
        assert garmin_svc.load_token(user["id"]) == "token-blob"

    def test_the_code_reaches_garth(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        wait_status(sid, "awaiting_code")
        garmin_auth.submit_code(sid, user["id"], "123456")
        wait_status(sid, "done", "failed")

        assert StubGarmin.last.garth.seen_code == "123456"

    def test_account_without_mfa_completes_without_a_code(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw",
            import_garmin=StubGarmin.make(needs_mfa=False),
        )
        assert wait_status(sid, "done", "failed") == "done"
        assert garmin_svc.has_token(user["id"]) is True

    def test_wrong_code_fails_with_a_reason(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        wait_status(sid, "awaiting_code")
        garmin_auth.submit_code(sid, user["id"], "000000")
        assert wait_status(sid, "done", "failed") == "failed"

        assert "验证码错误" in garmin_auth.get_session(sid)["error"]
        assert garmin_svc.has_token(user["id"]) is False

    def test_bad_password_fails_before_any_code(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "wrong",
            import_garmin=StubGarmin.make(fail_login=True),
        )
        assert wait_status(sid, "failed") == "failed"
        assert "401" in garmin_auth.get_session(sid)["error"]

    def test_timeout_when_no_code_arrives(self, db, user, monkeypatch):
        monkeypatch.setattr(garmin_auth, "CODE_WAIT_SECONDS", 0.2)
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        assert wait_status(sid, "failed") == "failed"
        assert "超时" in garmin_auth.get_session(sid)["error"]


class TestRendezvousIsNotInMemory:
    """The handoff must survive the code arriving on a different worker, so it
    goes through the database rather than process memory."""

    def test_code_written_directly_to_the_row_is_picked_up(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        wait_status(sid, "awaiting_code")

        # Exactly what another worker's request would do: write the row.
        db.execute(
            "UPDATE garmin_mfa_sessions SET code = ? WHERE id = ?", ["123456", sid]
        )
        assert wait_status(sid, "done", "failed") == "done"

    def test_cancelling_releases_the_parked_thread(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        wait_status(sid, "awaiting_code")
        garmin_auth.cancel(sid, user["id"])

        # The row is gone, so the waiting thread must stop rather than spin
        # until its timeout.
        assert wait_for(lambda: garmin_auth.get_session(sid) is None)


class TestSessionIsolation:
    def test_another_users_session_is_not_readable(self, db, user, make_user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        other = make_user("o@example.com")

        assert garmin_auth.get_session(sid, other["id"]) is None

    def test_another_user_cannot_submit_a_code(self, db, user, make_user):
        sid = garmin_auth.start_login(
            user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
        )
        wait_status(sid, "awaiting_code")
        other = make_user("o2@example.com")

        ok, _ = garmin_auth.submit_code(sid, other["id"], "123456")
        assert ok is False

    def test_unknown_session_is_refused(self, db, user):
        ok, msg = garmin_auth.submit_code("no-such-session", user["id"], "123456")
        assert ok is False
        assert "不存在" in msg


class TestPasswordHandling:
    def test_password_is_never_written_to_the_session_row(self, db, user):
        sid = garmin_auth.start_login(
            user["id"], "hunter2@example.com", "SuperSecret123",
            import_garmin=StubGarmin.make(),
        )
        wait_status(sid, "awaiting_code")

        row = garmin_auth.get_session(sid)
        assert "SuperSecret123" not in str(dict(row))


class TestEndpoints:
    def test_all_require_auth(self, client):
        assert client.post("/api/garmin/login", json={}).status_code == 401
        assert client.get("/api/garmin/login-status?session=x").status_code == 401
        assert client.post("/api/garmin/mfa", json={}).status_code == 401

    def test_login_requires_a_password(self, client, auth):
        r = client.post("/api/garmin/login", headers=auth, json={})
        assert r.status_code == 400

    def test_login_returns_a_session(self, client, auth, monkeypatch):
        monkeypatch.setattr(
            garmin_auth, "start_login", lambda *a, **k: "session-123"
        )
        r = client.post(
            "/api/garmin/login",
            headers=auth,
            json={"garminEmail": "g@example.com", "garminPassword": "pw"},
        )
        assert r.status_code == 202
        assert r.get_json()["session"] == "session-123"

    def test_status_of_unknown_session_is_404(self, client, auth):
        assert client.get(
            "/api/garmin/login-status?session=nope", headers=auth
        ).status_code == 404

    def test_mfa_requires_both_fields(self, client, auth):
        assert client.post(
            "/api/garmin/mfa", headers=auth, json={"session": "x"}
        ).status_code == 400

    def test_full_flow_through_http(self, client, auth, user, db, monkeypatch):
        monkeypatch.setattr(
            garmin_svc, "_import_garmin", StubGarmin.make()
        )
        r = client.post(
            "/api/garmin/login",
            headers=auth,
            json={"garminEmail": "g@example.com", "garminPassword": "pw"},
        )
        sid = r.get_json()["session"]

        assert wait_status(sid, "awaiting_code") == "awaiting_code"
        assert client.get(
            f"/api/garmin/login-status?session={sid}", headers=auth
        ).get_json()["status"] == "awaiting_code"

        r = client.post(
            "/api/garmin/mfa", headers=auth, json={"session": sid, "code": "123456"}
        )
        assert r.status_code == 200

        assert wait_status(sid, "done", "failed") == "done"
        assert client.get("/api/garmin/auth-status", headers=auth).get_json()[
            "hasToken"] is True