## Copyright © 2026 Olaf Kolkman
## SPDX-License-Identifier: GPL-3.0-or-later
from io import BytesIO
from fastapi.testclient import TestClient
from PIL import Image
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 '' in page_response.text
assert 'name="username"' not in page_response.text
assert '' 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' in page_response.text
assert 'id="auth-profile-link" class="hidden"' in page_response.text
assert '' 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
assert 'If you have not downloaded the plugin yet' in page_response.text
assert 'https://git.kolkman.org/olaf/Link-Log/raw/branch/main/XPI/signed/LinkLog-0.1.0.xpi' 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
image_buffer = BytesIO()
Image.new('RGB', (2, 2), 'red').save(image_buffer, format='JPEG')
upload_response = client.post(
'/api/user/avatar',
headers=headers,
files={'avatar': ('avatar.jpg', image_buffer.getvalue(), 'image/jpeg')},
)
assert upload_response.status_code == 200
avatar_url = upload_response.json()['avatar_url']
assert avatar_url.startswith('/media/user-1.png')
stored_avatar = client.get(avatar_url)
assert stored_avatar.status_code == 200
assert stored_avatar.headers['content-type'] == 'image/png'
with Image.open(BytesIO(stored_avatar.content)) as image:
assert image.format == 'PNG'
assert image.size == (2, 2)
rejected_upload = client.post(
'/api/user/avatar',
headers=headers,
files={'avatar': ('avatar.png', b'fake-png-data', 'image/png')},
)
assert rejected_upload.status_code == 415
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()