fix(garmin): 两步验证账号同步报 EOFError,改用令牌登录
现象:网页触发同步报 "EOF when reading a line"。 原因:garth 的默认 MFA 提示是 input(),向 stdin 索取验证码。 gunicorn worker 没有 stdin,于是抛出 EOFError——错误信息本身 完全没提到 MFA,看不出该做什么。 方案:把"输验证码"和"日常同步"拆开。 - 新增 garmin_tokens 表存 garth 令牌(Client.dumps/loads 序列化) - garmin_login.py:在终端里跑一次,可正常输入验证码, 成功后令牌存库 - _connect() 优先加载令牌并 refresh_oauth2(),命中则完全跳过登录, 既不需要密码也不需要验证码(令牌有效期约一年) - 无令牌且密码登录撞上 MFA 时,抛 MFARequired 并给出具体该执行 哪条命令,而不是把 EOFError 原样抛给用户 接口: - GET /api/garmin/auth-status 返回是否已有令牌 - /api/garmin/sync 在已有令牌时不再强制要求密码 前端: - 有令牌时隐藏密码输入框,提示无需密码 - 同步返回 mfaRequired 时,展示需要在 NAS 上执行的具体命令 - 同步请求超时放宽到 180s(一周的天数 + 运动是多次上游调用) - 成功消息补上运动记录条数 tests (test_garmin_sync.py 新增 12 条,共 35): - 令牌存取、覆盖不累积、按用户隔离 - 有令牌时绝不调用 login() - MFA 的 EOFError 转成带操作指引的 MFARequired - 普通 401 不会被误标成 mfaRequired - 无令牌且无密码时给出明确拒绝 NAS 真机: 252 passed Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -218,7 +218,7 @@ class TestPartialAndTotalFailure:
|
||||
assert out["recordsSynced"] == 0
|
||||
|
||||
def test_login_failure_is_reported(self, db, user, monkeypatch):
|
||||
def boom(_creds):
|
||||
def boom(_creds, _uid=None):
|
||||
raise RuntimeError("401 Unauthorized")
|
||||
|
||||
monkeypatch.setattr(garmin_svc, "_connect", boom)
|
||||
@@ -280,3 +280,131 @@ class TestEndpoint:
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user