171 lines
7.8 KiB
Python
171 lines
7.8 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import patch
|
|
|
|
from backend.app.main import app
|
|
from backend.app.database import get_connection
|
|
from backend.app.services.otp_service import current_code
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_user_config_api_and_profile_page():
|
|
login = client.post('/api/auth/login', json={
|
|
'email': 'alice@example.com',
|
|
'password': 'secret123',
|
|
}).json()
|
|
headers = {'Authorization': f"Bearer {login['access_token']}"}
|
|
|
|
profile_response = client.get('/api/user/me', headers=headers)
|
|
assert profile_response.status_code == 200
|
|
payload = profile_response.json()
|
|
assert payload['username'] == 'alice'
|
|
|
|
update_response = client.put('/api/user/me', json={
|
|
'email': 'alice@example.com',
|
|
'bio': 'Updated bio',
|
|
}, headers=headers)
|
|
assert update_response.status_code == 200
|
|
assert client.get('/api/user/me', headers=headers).json()['username'] == 'alice'
|
|
|
|
page_response = client.get('/profile')
|
|
assert page_response.status_code == 200
|
|
assert 'Profile' in page_response.text
|
|
assert '<output id="username" class="readonly-value">Loading...</output>' in page_response.text
|
|
assert 'name="username"' not in page_response.text
|
|
assert '<textarea id="bio" name="bio" rows="4"></textarea>' in page_response.text
|
|
assert 'name="avatar" type="file"' in page_response.text
|
|
assert 'name="avatar_url"' not in page_response.text
|
|
assert 'value="mastodon.social"' in page_response.text
|
|
assert 'value="From my #LinkLog: "' in page_response.text
|
|
assert 'id="auth-menu" class="auth-menu hidden"' in page_response.text
|
|
assert 'id="auth-home-link" href="/">Home</a>' in page_response.text
|
|
assert 'id="auth-profile-link" class="hidden"' in page_response.text
|
|
assert '<a id="auth-username" class="user-name" href="/">' in page_response.text
|
|
assert 'id="auth-avatar"' not in page_response.text
|
|
assert 'name="new_password_confirmation"' in page_response.text
|
|
assert 'id="additional-email-form"' in page_response.text
|
|
|
|
bob_login = client.post('/api/auth/login', json={
|
|
'email': 'bob@example.com',
|
|
'password': 'secret123',
|
|
}).json()
|
|
bob_headers = {'Authorization': f"Bearer {bob_login['access_token']}"}
|
|
password_response = client.put('/api/user/password', json={
|
|
'current_password': 'secret123',
|
|
'new_password': 'new-secret-123',
|
|
}, headers=bob_headers)
|
|
assert password_response.status_code == 200
|
|
assert client.post('/api/auth/login', json={
|
|
'email': 'bob@example.com',
|
|
'password': 'new-secret-123',
|
|
}).status_code == 200
|
|
assert client.put('/api/user/password', json={
|
|
'current_password': 'new-secret-123',
|
|
'new_password': 'secret123',
|
|
}, headers=bob_headers).status_code == 200
|
|
|
|
upload_response = client.post(
|
|
'/api/user/avatar',
|
|
headers=headers,
|
|
files={'avatar': ('avatar.png', b'fake-png-data', 'image/png')},
|
|
)
|
|
assert upload_response.status_code == 200
|
|
avatar_url = upload_response.json()['avatar_url']
|
|
assert avatar_url.startswith('/media/user-1.png')
|
|
assert client.get(avatar_url).content == b'fake-png-data'
|
|
|
|
updated_profile = client.get('/api/user/me', headers=headers).json()
|
|
assert updated_profile['avatar_url'] == avatar_url
|
|
|
|
|
|
def test_user_can_enable_and_use_otp():
|
|
login = client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'}).json()
|
|
headers = {'Authorization': f"Bearer {login['access_token']}"}
|
|
setup = client.post('/api/user/otp/setup', headers=headers)
|
|
assert setup.status_code == 200
|
|
secret = setup.json()['secret']
|
|
assert setup.json()['otpauth_url'].startswith('otpauth://totp/')
|
|
recovery_codes = setup.json()['recovery_codes']
|
|
assert len(recovery_codes) == 10
|
|
|
|
enabled = client.post('/api/user/otp', headers=headers, json={
|
|
'action': 'enable', 'code': current_code(secret),
|
|
})
|
|
assert enabled.status_code == 200
|
|
assert enabled.json()['enabled'] is True
|
|
assert client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'}).status_code == 401
|
|
|
|
otp_login = client.post('/api/auth/login', json={
|
|
'email': 'alice@example.com', 'password': 'secret123', 'otp': current_code(secret),
|
|
})
|
|
assert otp_login.status_code == 200
|
|
|
|
disabled = client.post('/api/user/otp', headers=headers, json={
|
|
'action': 'disable', 'code': current_code(secret), 'current_password': 'secret123',
|
|
})
|
|
assert disabled.status_code == 200
|
|
assert disabled.json()['enabled'] is False
|
|
|
|
|
|
def test_otp_recovery_code_requires_password_and_is_single_use():
|
|
login = client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'}).json()
|
|
headers = {'Authorization': f"Bearer {login['access_token']}"}
|
|
setup = client.post('/api/user/otp/setup', headers=headers)
|
|
secret = setup.json()['secret']
|
|
recovery_code = setup.json()['recovery_codes'][0]
|
|
assert client.post('/api/user/otp', headers=headers, json={
|
|
'action': 'enable', 'code': current_code(secret),
|
|
}).status_code == 200
|
|
|
|
rejected = client.post('/api/user/otp/recover', headers=headers, json={
|
|
'current_password': 'wrong-password', 'recovery_code': recovery_code,
|
|
})
|
|
assert rejected.status_code == 400
|
|
recovered = client.post('/api/user/otp/recover', headers=headers, json={
|
|
'current_password': 'secret123', 'recovery_code': recovery_code,
|
|
})
|
|
assert recovered.status_code == 200
|
|
reused = client.post('/api/user/otp/recover', headers=headers, json={
|
|
'current_password': 'secret123', 'recovery_code': recovery_code,
|
|
})
|
|
assert reused.status_code == 400
|
|
|
|
|
|
def test_verified_alternative_can_become_primary():
|
|
login = client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'}).json()
|
|
headers = {'Authorization': f"Bearer {login['access_token']}"}
|
|
with get_connection() as conn:
|
|
address = conn.execute(
|
|
'INSERT INTO user_email_addresses (id, user_id, email, verified) VALUES (?, ?, ?, 1) RETURNING id',
|
|
('alternative-test', 'user-1', 'alice-alternative@example.com'),
|
|
).fetchone()
|
|
conn.commit()
|
|
promoted = client.post(f"/api/user/emails/{address['id']}/make-primary", headers=headers)
|
|
assert promoted.status_code == 200
|
|
assert client.post('/api/auth/login', json={'email': 'alice-alternative@example.com', 'password': 'secret123'}).status_code == 200
|
|
assert client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'}).status_code == 200
|
|
with get_connection() as conn:
|
|
conn.execute('DELETE FROM user_email_addresses WHERE email IN (?, ?)', ('alice@example.com', 'alice-alternative@example.com'))
|
|
conn.execute('UPDATE users SET email = ?, email_verified = 1 WHERE id = ?', ('alice@example.com', 'user-1'))
|
|
conn.commit()
|
|
|
|
|
|
def test_unverified_alternative_cannot_become_primary():
|
|
login = client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'}).json()
|
|
headers = {'Authorization': f"Bearer {login['access_token']}"}
|
|
with get_connection() as conn:
|
|
address = conn.execute(
|
|
'INSERT INTO user_email_addresses (id, user_id, email, verified) VALUES (?, ?, ?, 0) RETURNING id',
|
|
('unverified-alternative-test', 'user-1', 'alice-unverified@example.com'),
|
|
).fetchone()
|
|
conn.commit()
|
|
rejected = client.post(f"/api/user/emails/{address['id']}/make-primary", headers=headers)
|
|
assert rejected.status_code == 400
|
|
with get_connection() as conn:
|
|
conn.execute('DELETE FROM user_email_addresses WHERE id = ?', (address['id'],))
|
|
conn.commit()
|