ericwyuan/GarminHealthLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(auth): 接入 auth-hub 统一登录，网页登录与 Garmin 同步彻底分离

Browse Source 
网页身份改由 auth-hub 做 OAuth2 + PKCE 单点登录，本地邮箱/密码登录与注册整条链路删除
（routes/auth.py、auth.py 的密码哈希、config.py 的 ALLOW_REGISTRATION）。Garmin 账号绑定/
同步保持完全独立、可选：routes/garmin.py 不再直接查 users 表，Garmin 邮箱回退统一走新增
的 services/garmin.py::get_remembered_email()（优先读 garmin_tokens 当前绑定，兼容早期账号
落在 users.garmin_email 的历史值），彻底把「你是谁」和「你绑没绑 Garmin」两件事拆开。

- db.py: users 表新增 auth_hub_sub/auth_hub_username，MIGRATIONS 补上这两列（此前遗漏导致
  已存在的生产 MariaDB 表永远不会自动加列）；同时把历史遗留的 garmin_email/
  garmin_password_hash NOT NULL 约束在线迁移为可空，因为新账号不再在注册时收集这些字段。
- routes/auth.py: 修掉 /callback 路由重复拼接 /api/auth 前缀导致 404 的 bug。
- client: LoginPage 去掉本地登录/注册标签页，只保留 auth-hub 统一登录；登录成功/失败后都
  用 history.replaceState 清理地址栏，修掉 Framework7 browserHistory 读取
  /auth/callback?code=... 导致「找不到页面」的问题。
- 新增 test_auth_hub_client.py 锁定 find_or_create_user 按 auth_hub_sub 幂等——生产上曾经因为
  这个函数在没有该测试保护时被测试触发，误建过一个空账号，靠手工核对 health_data 计数才发现。
- 生产 auth-hub 侧另行为该项目注册了正式 client（未随本次提交变更，凭证只存在服务器 .env）。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
2026-08-31 23:12:17 +08:00
parent 7e51223eb9
commit 9503fca370
24 changed files with 592 additions and 1007 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
backend/tests/conftest.py
View File

@@ -8,6 +8,7 @@ tests never share state.
import os
import sys
import tempfile
import uuid
import pytest
@@ -24,6 +25,7 @@ os.environ.setdefault(
import db as db_module # noqa: E402
from app import create_app # noqa: E402
from auth import sign_token # noqa: E402
# config.py calls load_dotenv() at import, so backend/.env leaks into the test
# process — a developer's real AI_MODEL_CHAIN or API keys would silently change
@@ -48,10 +50,6 @@ _AI_ENV_VARS = (
def _isolate_ai_env(monkeypatch):
for var in _AI_ENV_VARS:
monkeypatch.delenv(var, raising=False)
# Most tests need to create users freely; the production default closes
# registration once one account exists. test_registration_policy.py clears
# this to exercise the real default.
monkeypatch.setenv("ALLOW_REGISTRATION", "true")
@pytest.fixture(autouse=True)
@@ -90,21 +88,32 @@ def client(app):
return app.test_client()
@pytest.fixture
def user(client):
"""A registered user: returns {id, email, token, password}."""
password = "secret123"
resp = client.post(
"/api/auth/register",
json={
"email": "tester@example.com",
"garminEmail": "gm@example.com",
"garminPassword": password,
},
def _insert_user(email):
"""Create a user row directly, bypassing HTTP.
Accounts now come from auth-hub's OAuth dance (see routes/auth.py); the
test suite has no reason to exercise that network round trip just to get
a user id and a valid JWT.
"""
uid = str(uuid.uuid4())
token = sign_token(uid)
db_module.execute(
"INSERT INTO users (id, email, auth_hub_username, jwt_token) VALUES (?, ?, ?, ?)",
[uid, email, email, token],
)
assert resp.status_code == 201, resp.get_data(as_text=True)
body = resp.get_json()
return {**body, "password": password}
return {"id": uid, "email": email, "token": token}
@pytest.fixture
def make_user(db):
"""Factory for creating additional users, e.g. for cross-account isolation tests."""
return _insert_user
@pytest.fixture
def user(db):
"""A user, as if they had signed in through auth-hub: {id, email, token}."""
return _insert_user("tester@example.com")
@pytest.fixture

33
backend/tests/smoke.py
View File

@@ -1,14 +1,15 @@
"""
Smoke test for the Flask backend (SQLite).
Exercises the full request path: register -> login -> authenticated reads for
health summary/steps/heart-rate/sleep/activities, analysis trends +
recommendations, and Garmin sync status. Run: `python tests/smoke.py`.
Exercises the full request path: a user account (as if signed in through
auth-hub) -> authenticated reads for health summary/steps/heart-rate/sleep/
activities, analysis trends + recommendations, and Garmin sync status.
Run: `python tests/smoke.py`.
"""
import os
import sys
import tempfile
import json
import uuid
# Configure the backend BEFORE importing app/config.
_TMP_DB = os.path.join(tempfile.mkdtemp(), "smoke.db")
@@ -21,6 +22,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app import create_app # noqa: E402
from db import execute # noqa: E402
from auth import sign_token # noqa: E402
app = create_app()
client = app.test_client()
@@ -42,28 +44,19 @@ def auth_headers(token):
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
print("\n[1] Auth: register + login")
r = client.post(
"/api/auth/register",
json={"email": "tester@example.com", "garminEmail": "gm@example.com", "garminPassword": "secret123"},
print("\n[1] Auth: user account (as if signed in through auth-hub)")
uid = str(uuid.uuid4())
token = sign_token(uid)
execute(
"INSERT INTO users (id, email, auth_hub_username, jwt_token) VALUES (?, ?, ?, ?)",
[uid, "tester@example.com", "tester@example.com", token],
)
check("register 201", r.status_code == 201, r.get_data(as_text=True))
token = (r.get_json() or {}).get("token")
check("register returns token", bool(token))
r = client.post("/api/auth/login", json={"email": "tester@example.com", "password": "secret123"})
check("login 200", r.status_code == 200, r.get_data(as_text=True))
token = (r.get_json() or {}).get("token")
check("login returns token", bool(token))
r = client.post("/api/auth/login", json={"email": "tester@example.com", "password": "wrong"})
check("login rejects bad password (401)", r.status_code == 401)
check("user created", bool(uid))
r = client.get("/api/health/summary")
check("unauthenticated read 401", r.status_code == 401)
print("\n[2] Seed health data (3 days)")
uid = (client.post("/api/auth/login", json={"email": "tester@example.com", "password": "secret123"}).get_json())["id"]
for i, (steps, hr, sleep, stress) in enumerate([(6500, 70, 6.2, 55), (9000, 62, 7.5, 40), (7500, 68, 6.8, 48)]):
date = f"2026-08-{20 + i}"
execute(

16
backend/tests/test_activity_detail.py
View File

@@ -259,12 +259,8 @@ class TestStoreAndRead:
)
assert svc.read_activity_detail(user["id"], "abc") is None
def test_details_are_per_account(self, db, user, client):
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
def test_details_are_per_account(self, db, user, make_user):
other = make_user("b@example.com")
svc._store_detail(user["id"], "abc", {"summary": {"duration": 600}})
assert svc.read_activity_detail(other["id"], "abc") is None
@@ -334,12 +330,8 @@ class TestSyncActivityDetails:
finally:
g._build_detail = original
def test_only_this_accounts_activities(self, db, user, client):
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
def test_only_this_accounts_activities(self, db, user, make_user):
other = make_user("b@example.com")
self.seed(db, user, ["mine"])
db.execute(
"INSERT INTO activities (id, user_id, activity_type, start_time, end_time) "

8
backend/tests/test_ai_cache.py
View File

@@ -168,14 +168,10 @@ class TestCacheInvalidation:
class TestIsolationAndRobustness:
def test_cache_is_per_user(self, seeded, keys, counting_llm, db, client):
def test_cache_is_per_user(self, seeded, keys, counting_llm, db, make_user):
analysis_svc.get_ai_recommendations(seeded["id"])
other = client.post(
"/api/auth/register",
json={"email": "other@example.com", "garminEmail": "o@example.com",
"garminPassword": "pw123456"},
).get_json()
other = make_user("other@example.com")
health_svc.upsert_health_daily(
other["id"], {"date": "2026-08-20", "steps": 5000, "sleepDuration": 6}
)

151
backend/tests/test_auth.py
View File

@@ -1,7 +1,5 @@
"""Unit tests for password hashing, JWT handling, and the auth endpoints."""
"""Unit tests for JWT handling and the auth endpoints."""
import datetime
import hashlib
import os
import jwt
import pytest
@@ -10,66 +8,6 @@ import auth
from config import JWT_SECRET
# --- password hashing -------------------------------------------------------
class TestPasswordHashing:
def test_hash_is_verifiable(self):
stored = auth.hash_password("correct horse")
assert auth.verify_password("correct horse", stored) is True
def test_wrong_password_rejected(self):
stored = auth.hash_password("correct horse")
assert auth.verify_password("wrong horse", stored) is False
def test_salt_makes_hashes_unique(self):
a = auth.hash_password("same")
b = auth.hash_password("same")
assert a != b, "identical passwords must not produce identical hashes"
def test_hash_format_is_self_describing(self):
stored = auth.hash_password("pw")
prefix, iterations, salt, digest = stored.split("$")
assert prefix == auth.PBKDF2_PREFIX
assert int(iterations) == auth.PBKDF2_ITERATIONS
assert len(bytes.fromhex(salt)) == 16
assert len(bytes.fromhex(digest)) == 64
def test_password_never_appears_in_hash(self):
stored = auth.hash_password("supersecret")
assert "supersecret" not in stored
@pytest.mark.parametrize(
"stored",
["", None, "garbage", "pbkdf2_sha256$notanint$aa$bb", "nothex:nothex"],
)
def test_malformed_hashes_rejected_not_raised(self, stored):
assert auth.verify_password("anything", stored) is False
def test_hashing_does_not_require_scrypt(self, monkeypatch):
"""Regression: macOS system Python (LibreSSL) has no hashlib.scrypt.
Registration used to raise AttributeError -> HTTP 500 on those builds.
"""
monkeypatch.delattr(hashlib, "scrypt", raising=False)
stored = auth.hash_password("pw")
assert auth.verify_password("pw", stored) is True
def test_unicode_password(self):
stored = auth.hash_password("密码🔒")
assert auth.verify_password("密码🔒", stored) is True
assert auth.verify_password("密码", stored) is False
@pytest.mark.skipif(
not hasattr(hashlib, "scrypt"), reason="interpreter built without scrypt"
)
def test_legacy_scrypt_hash_still_verifies(self):
salt = os.urandom(16)
digest = hashlib.scrypt(
b"legacy", salt=salt, n=16384, r=8, p=1, dklen=64
).hex()
assert auth.verify_password("legacy", f"{salt.hex()}:{digest}") is True
assert auth.verify_password("nope", f"{salt.hex()}:{digest}") is False
# --- JWT --------------------------------------------------------------------
class TestTokens:
def test_sign_and_verify_roundtrip(self):
@@ -96,93 +34,6 @@ class TestTokens:
auth.verify_token(f"{head}.{payload}.{sig[:-2]}xx")
# --- register ---------------------------------------------------------------
class TestRegister:
def test_returns_201_and_token(self, client):
r = client.post(
"/api/auth/register",
json={
"email": "a@example.com",
"garminEmail": "g@example.com",
"garminPassword": "pw123456",
},
)
assert r.status_code == 201
body = r.get_json()
assert body["email"] == "a@example.com"
assert auth.verify_token(body["token"])["user_id"] == body["id"]
def test_duplicate_email_conflicts(self, client, user):
r = client.post(
"/api/auth/register",
json={
"email": user["email"],
"garminEmail": "other@example.com",
"garminPassword": "pw123456",
},
)
assert r.status_code == 409
@pytest.mark.parametrize(
"payload",
[
{},
{"email": "a@example.com"},
{"email": "a@example.com", "garminEmail": "g@example.com"},
{"email": "", "garminEmail": "g@example.com", "garminPassword": "x"},
],
)
def test_missing_fields_rejected(self, client, payload):
assert client.post("/api/auth/register", json=payload).status_code == 400
def test_password_stored_only_as_hash(self, client, db):
client.post(
"/api/auth/register",
json={
"email": "h@example.com",
"garminEmail": "g@example.com",
"garminPassword": "plaintext-secret",
},
)
row = db.query_one(
"SELECT garmin_password_hash FROM users WHERE email = ?", ["h@example.com"]
)
assert "plaintext-secret" not in row["garmin_password_hash"]
assert auth.verify_password("plaintext-secret", row["garmin_password_hash"])
# --- login ------------------------------------------------------------------
class TestLogin:
def test_valid_credentials(self, client, user):
r = client.post(
"/api/auth/login",
json={"email": user["email"], "password": user["password"]},
)
assert r.status_code == 200
assert r.get_json()["id"] == user["id"]
def test_wrong_password(self, client, user):
r = client.post(
"/api/auth/login", json={"email": user["email"], "password": "nope"}
)
assert r.status_code == 401
def test_unknown_email(self, client):
r = client.post(
"/api/auth/login", json={"email": "ghost@example.com", "password": "pw"}
)
assert r.status_code == 401
def test_error_does_not_reveal_which_field_was_wrong(self, client, user):
unknown = client.post(
"/api/auth/login", json={"email": "ghost@example.com", "password": "pw"}
).get_json()
bad_pw = client.post(
"/api/auth/login", json={"email": user["email"], "password": "nope"}
).get_json()
assert unknown == bad_pw, "responses must not distinguish the two cases"
# --- require_auth -----------------------------------------------------------
class TestRequireAuth:
def test_missing_header(self, client):

56
backend/tests/test_auth_hub_client.py Normal file
View File

@@ -0,0 +1,56 @@
"""
Tests for the auth-hub account-linking logic.
This is the one place a login bug can silently orphan a real account: if
`find_or_create_user` ever created a fresh row for a `auth_hub_sub` it had
already seen, the same person logging in twice would end up owning two
disconnected accounts — the second with none of the health data synced
under the first. That exact bug shipped once already (a login test created
a real duplicate in production before this file existed), so it is worth
locking down explicitly rather than only trusting `find_or_create_user`'s
docstring.
"""
from services.auth_hub_client import find_or_create_user
class TestFindOrCreateUser:
def test_first_login_creates_a_user(self, db):
uid = find_or_create_user("42", "alice")
row = db.query_one("SELECT * FROM users WHERE id = ?", [uid])
assert row["auth_hub_sub"] == "42"
assert row["auth_hub_username"] == "alice"
def test_repeat_login_returns_the_same_user_not_a_duplicate(self, db):
first = find_or_create_user("42", "alice")
second = find_or_create_user("42", "alice")
assert first == second
assert db.query_one(
"SELECT COUNT(*) AS n FROM users WHERE auth_hub_sub = ?", ["42"]
)["n"] == 1
def test_different_sub_gets_a_different_user(self, db):
alice = find_or_create_user("42", "alice")
bob = find_or_create_user("43", "bob")
assert alice != bob
def test_a_username_change_upstream_does_not_split_the_account(self, db):
"""auth-hub identifies accounts by `sub`; `preferred_username` can be
renamed there without that being treated as a new local account."""
first = find_or_create_user("42", "alice")
second = find_or_create_user("42", "alice_renamed")
assert first == second
def test_links_to_a_pre_existing_account_with_that_sub(self, db):
"""The legacy migration path: an account created before auth-hub
existed gets its auth_hub_sub set once (by an operator, or a future
self-service linking flow), and every login after that must resolve
to that same row rather than minting a new one."""
import uuid
legacy_id = str(uuid.uuid4())
db.execute(
"INSERT INTO users (id, email, auth_hub_sub, auth_hub_username) "
"VALUES (?, ?, ?, ?)",
[legacy_id, "legacy@example.com", "42", "alice"],
)
assert find_or_create_user("42", "alice") == legacy_id

24
backend/tests/test_garmin_mfa.py
View File

@@ -184,28 +184,20 @@ class TestRendezvousIsNotInMemory:
class TestSessionIsolation:
def test_another_users_session_is_not_readable(self, db, user, client):
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 = client.post(
"/api/auth/register",
json={"email": "o@example.com", "garminEmail": "og@example.com",
"garminPassword": "pw123456"},
).get_json()
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, client):
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 = client.post(
"/api/auth/register",
json={"email": "o2@example.com", "garminEmail": "og2@example.com",
"garminPassword": "pw123456"},
).get_json()
other = make_user("o2@example.com")
ok, _ = garmin_auth.submit_code(sid, other["id"], "123456")
assert ok is False
@@ -243,7 +235,9 @@ class TestEndpoints:
garmin_auth, "start_login", lambda *a, **k: "session-123"
)
r = client.post(
"/api/garmin/login", headers=auth, json={"garminPassword": "pw"}
"/api/garmin/login",
headers=auth,
json={"garminEmail": "g@example.com", "garminPassword": "pw"},
)
assert r.status_code == 202
assert r.get_json()["session"] == "session-123"
@@ -263,7 +257,9 @@ class TestEndpoints:
garmin_svc, "_import_garmin", StubGarmin.make()
)
r = client.post(
"/api/garmin/login", headers=auth, json={"garminPassword": "pw"}
"/api/garmin/login",
headers=auth,
json={"garminEmail": "g@example.com", "garminPassword": "pw"},
)
sid = r.get_json()["session"]

47
backend/tests/test_garmin_sync.py
View File

@@ -305,16 +305,51 @@ class TestTokenStore:
assert len(rows) == 1
assert garmin_svc.load_token(user["id"]) == "second"
def test_tokens_are_per_user(self, db, user, client):
def test_tokens_are_per_user(self, db, user, make_user):
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()
other = make_user("o@example.com")
assert garmin_svc.has_token(other["id"]) is False
class TestRememberedEmail:
"""`garmin_tokens` is the live Garmin binding; `users.garmin_email` is a
legacy column kept only for accounts that bound Garmin before that table
existed and have not signed in again since (see services/garmin.py)."""
def test_none_when_never_bound(self, db, user):
assert garmin_svc.get_remembered_email(user["id"]) == ""
def test_reads_from_the_current_binding(self, db, user):
garmin_svc.save_token(user["id"], "tok", "current@example.com")
assert garmin_svc.get_remembered_email(user["id"]) == "current@example.com"
def test_current_binding_wins_over_the_legacy_column(self, db, user):
db.execute(
"UPDATE users SET garmin_email = ? WHERE id = ?",
["legacy@example.com", user["id"]],
)
garmin_svc.save_token(user["id"], "tok", "current@example.com")
assert garmin_svc.get_remembered_email(user["id"]) == "current@example.com"
def test_falls_back_to_the_legacy_column_when_never_bound_since(self, db, user):
db.execute(
"UPDATE users SET garmin_email = ? WHERE id = ?",
["legacy@example.com", user["id"]],
)
assert garmin_svc.get_remembered_email(user["id"]) == "legacy@example.com"
def test_disconnecting_drops_the_current_binding_but_not_the_legacy_value(
self, db, user
):
db.execute(
"UPDATE users SET garmin_email = ? WHERE id = ?",
["legacy@example.com", user["id"]],
)
garmin_svc.save_token(user["id"], "tok", "current@example.com")
garmin_svc.delete_token(user["id"])
assert garmin_svc.get_remembered_email(user["id"]) == "legacy@example.com"
class TestMfaHandling:
def test_eof_from_the_mfa_prompt_becomes_an_actionable_error(
self, db, user, monkeypatch

16
backend/tests/test_health.py
View File

@@ -219,23 +219,15 @@ class TestBadgesAndRecords:
assert len(rows) == 1
assert rows[0]["earned_count"] == 2
def test_badges_are_per_user(self, db, user, client):
def test_badges_are_per_user(self, db, user, make_user):
health_svc.upsert_badge(user["id"], self.BADGE)
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
other = make_user("b@example.com")
assert health_svc.get_badges(other["id"]) == []
def test_two_users_may_hold_the_same_badge_id(self, db, user, client):
def test_two_users_may_hold_the_same_badge_id(self, db, user, make_user):
"""The key is (user, badge), so the same Garmin badge on two accounts
must not collide."""
other = client.post(
"/api/auth/register",
json={"email": "c@example.com", "garminEmail": "cg@example.com",
"garminPassword": "pw123456"},
).get_json()
other = make_user("c@example.com")
health_svc.upsert_badge(user["id"], self.BADGE)
health_svc.upsert_badge(other["id"], self.BADGE)
assert len(health_svc.get_badges(user["id"])) == 1

87
backend/tests/test_registration_policy.py
View File

@@ -1,87 +0,0 @@
"""
Tests for who may create an account.
This matters because the deployment is reachable from the public internet: an
unconditionally open /register would let a stranger sign up and start pulling
health data.
"""
import pytest
def signup(client, email="new@example.com"):
return client.post(
"/api/auth/register",
json={
"email": email,
"garminEmail": "g@example.com",
"garminPassword": "pw123456",
},
)
@pytest.fixture(autouse=True)
def _default_policy(monkeypatch):
monkeypatch.delenv("ALLOW_REGISTRATION", raising=False)
class TestAutoPolicy:
"""Default: open until the first account exists, then closed."""
def test_first_account_is_allowed(self, client, db):
assert signup(client).status_code == 201
def test_second_account_is_refused(self, client, user):
r = signup(client, "stranger@example.com")
assert r.status_code == 403
assert "注册已关闭" in r.get_json()["error"]
def test_refusal_does_not_create_the_account(self, client, user, db):
signup(client, "stranger@example.com")
assert db.query_one(
"SELECT id FROM users WHERE email = ?", ["stranger@example.com"]
) is None
def test_status_reports_open_before_any_signup(self, client, db):
assert client.get("/api/auth/registration-status").get_json()["open"] is True
def test_status_reports_closed_afterwards(self, client, user):
assert client.get("/api/auth/registration-status").get_json()["open"] is False
class TestExplicitPolicies:
def test_true_keeps_it_open_even_with_existing_users(self, client, user, monkeypatch):
monkeypatch.setenv("ALLOW_REGISTRATION", "true")
assert signup(client, "second@example.com").status_code == 201
def test_false_closes_it_even_on_an_empty_instance(self, client, db, monkeypatch):
monkeypatch.setenv("ALLOW_REGISTRATION", "false")
assert signup(client).status_code == 403
def test_policy_is_read_per_request_not_at_import(self, client, db, monkeypatch):
monkeypatch.setenv("ALLOW_REGISTRATION", "false")
assert client.get("/api/auth/registration-status").get_json()["open"] is False
monkeypatch.setenv("ALLOW_REGISTRATION", "true")
assert client.get("/api/auth/registration-status").get_json()["open"] is True
def test_value_is_case_insensitive(self, client, user, monkeypatch):
monkeypatch.setenv("ALLOW_REGISTRATION", "TRUE")
assert signup(client, "second@example.com").status_code == 201
class TestUnaffectedBehaviour:
def test_status_endpoint_needs_no_auth(self, client, db):
"""The login page must be able to ask before anyone is signed in."""
assert client.get("/api/auth/registration-status").status_code == 200
def test_closing_registration_does_not_block_login(self, client, user):
r = client.post(
"/api/auth/login",
json={"email": user["email"], "password": user["password"]},
)
assert r.status_code == 200
def test_duplicate_email_still_reports_409_when_open(
self, client, user, monkeypatch
):
monkeypatch.setenv("ALLOW_REGISTRATION", "true")
assert signup(client, user["email"]).status_code == 409

16
backend/tests/test_scheduler.py
View File

@@ -78,14 +78,10 @@ class TestSyncAllAccounts:
def test_no_accounts_is_a_no_op(self, db, user):
assert scheduler.sync_all_accounts() == []
def test_syncs_every_account_holding_a_token(self, db, user, monkeypatch, client):
def test_syncs_every_account_holding_a_token(self, db, user, monkeypatch, make_user):
from services import garmin as garmin_svc
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
other = make_user("b@example.com")
garmin_svc.save_token(user["id"], "t1", "a@example.com")
garmin_svc.save_token(other["id"], "t2", "b@example.com")
@@ -101,15 +97,11 @@ class TestSyncAllAccounts:
assert all(r["status"] == "success" for r in results)
def test_one_failing_account_does_not_stop_the_others(
self, db, user, monkeypatch, client
self, db, user, monkeypatch, make_user
):
from services import garmin as garmin_svc
other = client.post(
"/api/auth/register",
json={"email": "c@example.com", "garminEmail": "cg@example.com",
"garminPassword": "pw123456"},
).get_json()
other = make_user("c@example.com")
garmin_svc.save_token(user["id"], "t1")
garmin_svc.save_token(other["id"], "t2")

8
backend/tests/test_settings.py
View File

@@ -65,12 +65,8 @@ class TestSaving:
svc.save_settings(user["id"], {})
assert svc.get_settings(user["id"])["heightCm"] == 180
def test_settings_are_per_account(self, db, user, client):
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
def test_settings_are_per_account(self, db, user, make_user):
other = make_user("b@example.com")
svc.save_settings(user["id"], {"heightCm": 178})
svc.save_settings(other["id"], {"heightCm": 160})
assert svc.get_settings(user["id"])["heightCm"] == 178