from fastapi.testclient import TestClient from app.db.repositories import UserRepository from app.security.credentials import CredentialCipher def extract_csrf(html: str) -> str: marker = 'name="csrf_token" value="' start = html.index(marker) + len(marker) end = html.index('"', start) return html[start:end] def admin_login(client: TestClient) -> None: page = client.get("/login") csrf = extract_csrf(page.text) response = client.post( "/login", data={"password": "admin-secret", "csrf_token": csrf}, follow_redirects=False, ) assert response.status_code == 303 def create_user_via_admin(client: TestClient, **overrides) -> int: admin_login(client) page = client.get("/users/new") csrf = extract_csrf(page.text) payload = { "csrf_token": csrf, "name": "Max", "mywhoosh_email": "max@mywhoosh.example", "mywhoosh_password": "mw-secret", "garmin_email": "max@garmin.example", "garmin_password": "garmin-secret", "enabled": "on", } payload.update(overrides) response = client.post("/users", data=payload, follow_redirects=False) assert response.status_code == 303 user_id = int(response.headers["location"].rsplit("/", 1)[-1]) client.cookies.clear() return user_id def account_login(client: TestClient, *, email: str, password: str): page = client.get("/account-login") csrf = extract_csrf(page.text) return client.post( "/account-login", data={"csrf_token": csrf, "email": email, "password": password}, follow_redirects=False, ) def test_login_with_mywhoosh_credentials_succeeds(client: TestClient) -> None: create_user_via_admin(client) response = account_login(client, email="max@mywhoosh.example", password="mw-secret") assert response.status_code == 303 assert response.headers["location"] == "/account" def test_login_with_garmin_credentials_succeeds(client: TestClient) -> None: create_user_via_admin(client) response = account_login(client, email="max@garmin.example", password="garmin-secret") assert response.status_code == 303 assert response.headers["location"] == "/account" def test_login_with_wrong_password_is_rejected(client: TestClient) -> None: create_user_via_admin(client) response = account_login(client, email="max@mywhoosh.example", password="wrong") assert response.status_code == 401 assert "Invalid email or password" in response.text def test_login_never_makes_the_stored_password_appear_in_response(client: TestClient) -> None: create_user_via_admin(client) response = account_login(client, email="max@mywhoosh.example", password="wrong") assert "mw-secret" not in response.text def test_account_detail_requires_login(client: TestClient) -> None: response = client.get("/account", follow_redirects=False) assert response.status_code == 303 assert response.headers["location"] == "/account-login" def test_account_detail_shows_own_status_only(client: TestClient) -> None: create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") response = client.get("/account") assert response.status_code == 200 assert "Max" in response.text assert "mw-secret" not in response.text assert "garmin-secret" not in response.text def test_account_edit_page_prefills_emails_not_passwords(client: TestClient) -> None: create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") response = client.get("/account/edit") assert response.status_code == 200 assert "max@mywhoosh.example" in response.text assert "max@garmin.example" in response.text assert "mw-secret" not in response.text assert "garmin-secret" not in response.text def test_account_update_blank_password_preserves_existing_password(client: TestClient) -> None: user_id = create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") edit_page = client.get("/account/edit") csrf = extract_csrf(edit_page.text) response = client.post( "/account/edit", data={ "csrf_token": csrf, "mywhoosh_email": "max@mywhoosh.example", "mywhoosh_password": "", "garmin_email": "max@garmin.example", "garmin_password": "", }, follow_redirects=False, ) assert response.status_code == 303 with client.app.state.session_factory() as session: user = UserRepository(session).get(user_id) cipher = CredentialCipher(client.app.state.settings.credential_encryption_key) assert cipher.decrypt(user.mywhoosh_password_enc) == "mw-secret" assert cipher.decrypt(user.garmin_password_enc) == "garmin-secret" def test_account_update_can_set_new_password(client: TestClient) -> None: user_id = create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") edit_page = client.get("/account/edit") csrf = extract_csrf(edit_page.text) response = client.post( "/account/edit", data={ "csrf_token": csrf, "mywhoosh_email": "max@mywhoosh.example", "mywhoosh_password": "new-mw-secret", "garmin_email": "max@garmin.example", "garmin_password": "", }, follow_redirects=False, ) assert response.status_code == 303 with client.app.state.session_factory() as session: user = UserRepository(session).get(user_id) cipher = CredentialCipher(client.app.state.settings.credential_encryption_key) assert cipher.decrypt(user.mywhoosh_password_enc) == "new-mw-secret" def test_account_update_cannot_change_name_or_enabled(client: TestClient) -> None: """Self-service editing must not expose name/enabled -- those stay administrative decisions, not something the account owner can flip.""" user_id = create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") edit_page = client.get("/account/edit") csrf = extract_csrf(edit_page.text) client.post( "/account/edit", data={ "csrf_token": csrf, "mywhoosh_email": "max@mywhoosh.example", "mywhoosh_password": "", "garmin_email": "max@garmin.example", "garmin_password": "", }, follow_redirects=False, ) with client.app.state.session_factory() as session: user = UserRepository(session).get(user_id) assert user.name == "Max" assert user.enabled is True def test_account_update_persists_notification_preferences(client: TestClient) -> None: user_id = create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") edit_page = client.get("/account/edit") csrf = extract_csrf(edit_page.text) response = client.post( "/account/edit", data={ "csrf_token": csrf, "mywhoosh_email": "max@mywhoosh.example", "mywhoosh_password": "", "garmin_email": "max@garmin.example", "garmin_password": "", "notify_email_enabled": "on", "notification_email": "alerts@example.com", }, follow_redirects=False, ) assert response.status_code == 303 with client.app.state.session_factory() as session: user = UserRepository(session).get(user_id) assert user.notify_email_enabled is True assert user.notification_email == "alerts@example.com" def test_account_edit_rejects_invalid_csrf(client: TestClient) -> None: create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") response = client.post( "/account/edit", data={ "csrf_token": "invalid-token", "mywhoosh_email": "max@mywhoosh.example", "mywhoosh_password": "", "garmin_email": "max@garmin.example", "garmin_password": "", }, ) assert response.status_code == 403 def test_logout_clears_session(client: TestClient) -> None: create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") page = client.get("/account") csrf = extract_csrf(page.text) response = client.post("/account-logout", data={"csrf_token": csrf}, follow_redirects=False) assert response.status_code == 303 response = client.get("/account", follow_redirects=False) assert response.status_code == 303 assert response.headers["location"] == "/account-login" def test_cannot_view_other_users_account(client: TestClient) -> None: """Each self-service session is bound to the user_id captured at login; another user created afterwards must not be reachable from it.""" create_user_via_admin(client, name="Max") account_login(client, email="max@mywhoosh.example", password="mw-secret") with client.app.state.session_factory() as session: UserRepository(session).create( name="Other", enabled=True, mywhoosh_email_enc="unused", mywhoosh_password_enc="unused", garmin_email_enc="unused", garmin_password_enc="unused", ) response = client.get("/account") assert response.status_code == 200 assert "Max" in response.text assert "Other" not in response.text def test_account_sync_triggers_own_user_only(app, client: TestClient, fake_sync_manager) -> None: user_id = create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") app.state.sync_manager = fake_sync_manager page = client.get("/account") csrf = extract_csrf(page.text) response = client.post("/account/sync", data={"csrf_token": csrf}) assert response.status_code == 200 assert fake_sync_manager.user_calls == [user_id] def test_account_sync_reports_already_running(app, client: TestClient, fake_sync_manager) -> None: create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") app.state.sync_manager = fake_sync_manager fake_sync_manager.raise_already_running = True page = client.get("/account") csrf = extract_csrf(page.text) response = client.post("/account/sync", data={"csrf_token": csrf}) assert response.status_code == 409 assert "already running" in response.text.lower() def test_account_sync_requires_login(client: TestClient) -> None: response = client.post("/account/sync", data={"csrf_token": "whatever"}, follow_redirects=False) assert response.status_code == 303 assert response.headers["location"] == "/account-login" def test_account_sync_rejects_invalid_csrf(app, client: TestClient, fake_sync_manager) -> None: create_user_via_admin(client) account_login(client, email="max@mywhoosh.example", password="mw-secret") app.state.sync_manager = fake_sync_manager response = client.post("/account/sync", data={"csrf_token": "invalid-token"}) assert response.status_code == 403 assert fake_sync_manager.user_calls == []