eb97374578
Mirrors the dashboard's row update: "Sync now" on the account page refreshes the status block in place and shows a toast, instead of navigating to a separate result page. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
336 lines
12 KiB
Python
336 lines
12 KiB
Python
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_as_toast(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 == 200
|
|
assert "already running" in response.text.lower()
|
|
|
|
|
|
def test_account_sync_updates_status_block_and_shows_toast(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
|
|
|
|
page = client.get("/account")
|
|
csrf = extract_csrf(page.text)
|
|
response = client.post("/account/sync", data={"csrf_token": csrf})
|
|
|
|
assert response.status_code == 200
|
|
assert 'id="account-status"' in response.text
|
|
assert 'hx-swap-oob="true"' in response.text
|
|
assert "0 imported, 0 failed" in response.text
|
|
|
|
|
|
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 == []
|