Link-Log/backend/tests/test_api.py

## Copyright © 2026 Olaf Kolkman
## SPDX-License-Identifier: GPL-3.0-or-later

import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs
from uuid import uuid4
from unittest.mock import MagicMock, patch

from fastapi.testclient import TestClient

from backend.app.main import app
from backend.app.database import get_connection
from backend.app.services.email_service import get_smtp_settings
from backend.app.services.login_throttle import clear_login_failures
from backend.app.services.otp_service import current_code
from backend.app.services.password_reset import create_reset_token
from backend.app.services.token_service import issue_token


client = TestClient(app)


def login_headers(username='alice'):
    token = client.post('/api/auth/login', json={
        'email': 'alice@example.com' if username == 'alice' else f'{username}@example.com',
        'password': 'secret123',
    }).json()['access_token']
    return {'Authorization': f'Bearer {token}'}


def test_login_returns_token():
    assert app.version == '0.1.1'
    response = client.post('/api/auth/login', json={
           'email': 'alice@example.com',
        'password': 'secret123',
    })
    assert response.status_code == 200
    payload = response.json()
    assert 'access_token' in payload
    assert payload['token_type'] == 'bearer'

    admin_session = client.get('/api/auth/me', headers={'Authorization': f"Bearer {payload['access_token']}"})
    assert admin_session.status_code == 200
    assert admin_session.json()['is_admin'] is True

    user_token = client.post('/api/auth/login', json={
           'email': 'bob@example.com',
        'password': 'secret123',
    }).json()['access_token']
    user_session = client.get('/api/auth/me', headers={'Authorization': f"Bearer {user_token}"})
    assert user_session.status_code == 200
    assert user_session.json()['is_admin'] is False
    assert client.get('/api/auth/me', params={'token': payload['access_token']}).status_code == 401


def test_logout_requires_bearer_header_and_revokes_token_family():
    login = client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'}).json()
    token = login['access_token']
    headers = {'Authorization': f'Bearer {token}'}
    assert client.post('/api/auth/logout', json={'token': token}).status_code == 401
    assert client.get('/api/auth/me', headers=headers).status_code == 200
    assert client.post('/api/auth/logout', headers=headers).status_code == 200
    assert client.get('/api/auth/me', headers=headers).status_code == 401
    assert client.post('/api/auth/refresh', json={'refresh_token': login['refresh_token'], 'device_id': login['device_id']}).status_code == 401


def test_login_rate_limit_locks_out_after_five_failures_and_resets_on_success():
    email = f'unknown-{uuid4().hex}@example.com'
    for attempt in range(5):
        response = client.post('/api/auth/login', json={'email': email, 'password': 'wrong-password'})
        assert response.status_code == 401, attempt
    locked = client.post('/api/auth/login', json={'email': email, 'password': 'wrong-password'})
    assert locked.status_code == 429
    assert int(locked.headers['Retry-After']) > 0

    clear_login_failures('testclient', email)
    valid = client.post('/api/auth/login', json={'email': 'alice@example.com', 'password': 'secret123'})
    assert valid.status_code == 200


def test_refresh_token_rotates_and_reuse_revokes_family():
    device_id = f'device-{uuid4().hex}'
    login = client.post('/api/auth/login', json={
        'email': 'alice@example.com',
        'password': 'secret123',
        'device_id': device_id,
    })
    assert login.status_code == 200
    first = login.json()

    rotated = client.post('/api/auth/refresh', json={
        'refresh_token': first['refresh_token'],
        'device_id': device_id,
    })
    assert rotated.status_code == 200
    second = rotated.json()
    assert second['refresh_token'] != first['refresh_token']
    assert client.get('/api/auth/me', headers={'Authorization': f"Bearer {second['access_token']}"}).status_code == 200

    reused = client.post('/api/auth/refresh', json={
        'refresh_token': first['refresh_token'],
        'device_id': device_id,
    })
    assert reused.status_code == 401
    family_revoked = client.post('/api/auth/refresh', json={
        'refresh_token': second['refresh_token'],
        'device_id': device_id,
    })
    assert family_revoked.status_code == 401


def test_password_hashes_are_salted_and_legacy_hashes_upgrade_on_login():
    from hashlib import sha256
    from backend.app.database import hash_password

    first = hash_password('same-password')
    second = hash_password('same-password')
    assert first.startswith('scrypt$16384$8$1$')
    assert first != second

    legacy_username = f'legacy-{uuid4().hex}'
    legacy_hash = sha256('legacy-password'.encode('utf-8')).hexdigest()
    with get_connection() as conn:
        conn.execute(
            '''INSERT INTO users
               (id, username, email, password_hash, is_admin, email_verified)
               VALUES (?, ?, ?, ?, 0, 1)''',
            (str(uuid4()), legacy_username, f'{legacy_username}@example.com', legacy_hash),
        )
        conn.commit()
        response = client.post('/api/auth/login', json={'email': legacy_username + '@example.com', 'password': 'legacy-password'})
    assert response.status_code == 200
    with get_connection() as conn:
        upgraded = conn.execute('SELECT password_hash FROM users WHERE username = ?', (legacy_username,)).fetchone()['password_hash']
    assert upgraded.startswith('scrypt$16384$8$1$')


def test_configuration_requires_authentication_and_admin_role():
    assert client.get('/api/user/me').status_code == 401
    assert client.get('/api/admin/plugins').status_code == 401
    assert client.get('/api/admin/users').status_code == 401
    assert client.get('/api/admin/plugins', headers=login_headers('bob')).status_code == 403
    assert client.get('/api/admin/users', headers=login_headers('bob')).status_code == 403
    assert client.get('/api/admin/smtp').status_code == 401
    assert client.get('/api/admin/smtp', headers=login_headers('bob')).status_code == 403


def test_admin_can_select_multiple_themes():
    headers = login_headers()
    response = client.put('/api/admin/themes', headers=headers, json={
        'themes': ['plain-day', 'plain-night', 'latte', 'frappe', 'macchiato', 'mocha', 'dracula', 'nord', 'solarized'],
    })
    assert response.status_code == 200
    assert response.json()['enabled'] == ['plain-day', 'plain-night', 'latte', 'frappe', 'macchiato', 'mocha', 'dracula', 'nord', 'solarized']
    public_response = client.get('/api/public/themes')
    assert public_response.status_code == 200
    assert [theme['id'] for theme in public_response.json()] == response.json()['enabled']

    assert client.put('/api/admin/themes', headers=headers, json={'themes': []}).status_code == 422


def test_admin_can_save_and_validate_smtp_settings():
    headers = login_headers()
    original = get_smtp_settings()
    with get_connection() as conn:
        original_row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('smtp',)).fetchone()
    response = client.put('/api/admin/smtp', headers=headers, json={
        'smtp_host': 'smtp.example.com',
        'smtp_port': 587,
        'smtp_username': 'mailer',
        'smtp_password': 'secret',
        'smtp_from': 'LinkLog <no-reply@example.com>',
        'smtp_use_tls': True,
    })
    assert response.status_code == 200
    assert response.json()['smtp_host'] == 'smtp.example.com'
    assert response.json()['password_configured'] is True
    assert 'smtp_password' not in response.json()

    with patch('backend.app.api.admin.send_test_email') as send_test_email:
        validation = client.post('/api/admin/smtp/test', headers=headers, json={
            'smtp_host': 'smtp.unsaved.example.com',
            'smtp_port': 2525,
            'smtp_username': 'temporary-user',
            'smtp_password': 'temporary-secret',
            'smtp_from': 'Temporary <temporary@example.com>',
            'smtp_use_tls': False,
        })
        assert validation.status_code == 200
        send_test_email.assert_called_once_with('alice@example.com', {
            'smtp_host': 'smtp.unsaved.example.com',
            'smtp_port': 2525,
            'smtp_username': 'temporary-user',
            'smtp_password': 'temporary-secret',
            'smtp_from': 'Temporary <temporary@example.com>',
            'smtp_use_tls': False,
        })

    with get_connection() as conn:
        if original_row is None:
            conn.execute('DELETE FROM app_settings WHERE name = ?', ('smtp',))
        else:
            conn.execute(
                'UPDATE app_settings SET value = ?, updated_at = CURRENT_TIMESTAMP WHERE name = ?',
                (original_row['value'], 'smtp'),
            )
        conn.commit()
    with get_connection() as conn:
        conn.execute('DELETE FROM app_settings WHERE name = ?', ('admin_smtp_mail_rate',))
        conn.commit()


def test_admin_reports_smtp_validation_errors():
    headers = login_headers()
    with get_connection() as conn:
        conn.execute('DELETE FROM app_settings WHERE name = ?', ('admin_smtp_mail_rate',))
        conn.commit()
    with patch('backend.app.api.admin.send_test_email', side_effect=RuntimeError('connection refused')):
        failed_validation = client.post('/api/admin/smtp/test', headers=headers, json={
            'smtp_host': 'smtp.unsaved.example.com',
            'smtp_port': 2525,
            'smtp_username': 'temporary-user',
            'smtp_password': 'temporary-secret',
            'smtp_from': 'Temporary <temporary@example.com>',
            'smtp_use_tls': False,
        })
    assert failed_validation.status_code == 503
    assert failed_validation.json()['detail'].startswith('SMTP validation failed. Reference: ')
    assert 'connection refused' not in failed_validation.json()['detail']
    assert failed_validation.headers['X-Request-ID']


def test_request_id_is_preserved_and_sensitive_error_text_is_not_returned():
    headers = login_headers()
    with get_connection() as conn:
        conn.execute('DELETE FROM app_settings WHERE name = ?', ('admin_smtp_mail_rate',))
        conn.commit()
    with patch('backend.app.api.admin.send_test_email', side_effect=RuntimeError('password=super-secret token=abc123')):
        response = client.post(
            '/api/admin/smtp/test',
            headers={**headers, 'X-Request-ID': 'audit-test-123'},
            json={
                'smtp_host': 'smtp.example.com',
                'smtp_port': 2525,
                'smtp_from': 'admin@example.com',
            },
        )
    assert response.status_code == 503
    assert response.headers['X-Request-ID'] == 'audit-test-123'
    assert response.json()['detail'] == 'SMTP validation failed. Reference: audit-test-123'
    assert 'super-secret' not in response.text
    assert 'abc123' not in response.text


def test_admin_can_add_list_and_remove_users():
    headers = login_headers()
    create_response = client.post('/api/admin/users', headers=headers, json={
        'username': 'charlie',
        'email': 'charlie@example.com',
        'password': 'charlie-secret',
    })

    assert create_response.status_code == 201
    user = create_response.json()
    assert user['username'] == 'charlie'
    assert 'password_hash' not in user

    users = client.get('/api/admin/users', headers=headers).json()
    assert any(item['id'] == user['id'] for item in users)
    assert client.delete(f"/api/admin/users/{user['id']}", headers=headers).status_code == 200
    assert all(item['id'] != user['id'] for item in client.get('/api/admin/users', headers=headers).json())
    assert client.delete('/api/admin/users/user-1', headers=headers).status_code == 400
    assert client.put('/api/admin/users/user-1', headers=headers, json={'is_admin': False}).status_code == 400


def test_admin_can_reset_another_users_otp():
    admin_headers = login_headers()
    user_login = client.post('/api/auth/login', json={
        'email': 'bob@example.com',
        'password': 'secret123',
    }).json()
    user_headers = {'Authorization': f"Bearer {user_login['access_token']}"}
    setup = client.post('/api/user/otp/setup', headers=user_headers)
    assert setup.status_code == 200
    secret = setup.json()['secret']
    recovery_code = setup.json()['recovery_codes'][0]
    assert client.post('/api/user/otp', headers=user_headers, json={
        'action': 'enable', 'code': current_code(secret),
    }).status_code == 200

    assert client.post('/api/admin/users/user-2/otp/reset', headers=admin_headers).json() == {
        'status': 'otp_reset', 'enabled': False, 'user_id': 'user-2',
    }
    assert client.get('/api/user/otp', headers=user_headers).json() == {'enabled': False}
    assert client.post('/api/user/otp/recover', headers=user_headers, json={
        'current_password': 'secret123', 'recovery_code': recovery_code,
    }).status_code == 400
    assert client.post('/api/admin/users/user-2/otp/reset', headers=login_headers('bob')).status_code == 403
    assert client.post('/api/admin/users/missing-user/otp/reset', headers=admin_headers).status_code == 404


def test_security_audit_events_are_append_only_and_do_not_store_secrets():
    admin_headers = login_headers()
    response = client.put('/api/admin/themes', headers=admin_headers, json={'themes': ['plain-day']})
    assert response.status_code == 200
    with get_connection() as conn:
        event = conn.execute(
            '''SELECT actor_id, action, target_type, outcome, details
               FROM security_audit_events
               WHERE action = 'themes_updated'
               ORDER BY created_at DESC, rowid DESC LIMIT 1''',
        ).fetchone()
        assert event is not None
        assert event['actor_id'] == 'user-1'
        assert event['target_type'] == 'application'
        assert event['outcome'] == 'success'
        assert 'password' not in event['details'].lower()
        assert 'token' not in event['details'].lower()
        assert 'secret' not in event['details'].lower()


def test_new_user_must_verify_email_before_login():
    headers = login_headers()
    username = f'unverified-{uuid4().hex}'
    created = client.post('/api/admin/users', headers=headers, json={
        'username': username,
        'email': f'{username}@example.com',
        'password': 'secret123',
    })
    assert created.status_code == 201
    assert created.json()['email_verified'] is False

    login = client.post('/api/auth/login', json={'email': f'{username}@example.com', 'password': 'secret123'})
    assert login.status_code == 403
    assert login.json()['detail'] == 'Email address is not verified'


def test_email_verification_link_enables_login():
    headers = login_headers()
    username = f'verifiable-{uuid4().hex}'
    created = client.post('/api/admin/users', headers=headers, json={
        'username': username,
        'email': f'{username}@example.com',
        'password': 'secret123',
    })
    assert created.status_code == 201
    user_id = created.json()['id']
    from backend.app.services.email_verification import create_verification_token
    verification_token = create_verification_token(user_id)

    verified = client.get('/api/auth/verify-email', params={'token': verification_token})
    assert verified.status_code == 200
    assert client.post('/api/auth/login', json={'email': f'{username}@example.com', 'password': 'secret123'}).status_code == 200
    assert client.get('/api/auth/verify-email', params={'token': verification_token}).status_code == 400


def test_mistyped_password_sends_reset_link_without_changing_login_error():
    username = f'mistyped-{uuid4().hex}'
    admin_headers = {'Authorization': f"Bearer {issue_token('user-1', 'alice')['access_token']}"}
    created = client.post('/api/admin/users', headers=admin_headers, json={
        'username': username,
        'email': f'{username}@example.com',
        'password': 'secret123',
    })
    user_id = created.json()['id']
    with get_connection() as conn:
        conn.execute('UPDATE users SET email_verified = 1 WHERE id = ?', (user_id,))
        conn.commit()
    with patch('backend.app.api.auth.smtp_configured', return_value=True), \
            patch('backend.app.api.auth.create_reset_token', return_value='reset-token') as create_token, \
            patch('backend.app.api.auth.send_password_reset_email') as send_email:
            response = client.post('/api/auth/login', json={'email': f'{username}@example.com', 'password': 'wrong-password'})

    assert response.status_code == 401
    assert response.json()['detail'] == 'Invalid username or password'
    create_token.assert_called_once_with(user_id)
    send_email.assert_called_once()
    assert send_email.call_args.args[2].endswith('/reset-password?token=reset-token')


def test_password_reset_is_single_use_and_revokes_sessions():
    username = f'reset-owner-{uuid4().hex}'
    admin_headers = {'Authorization': f"Bearer {issue_token('user-1', 'alice')['access_token']}"}
    created = client.post('/api/admin/users', headers=admin_headers, json={
        'username': username,
        'email': f'{username}@example.com',
        'password': 'secret123',
    })
    user_id = created.json()['id']
    with get_connection() as conn:
        conn.execute('UPDATE users SET email_verified = 1 WHERE id = ?', (user_id,))
        conn.commit()
    token = create_reset_token(user_id)
    reset = client.post('/api/auth/reset-password', json={'token': token, 'password': 'new-secret123'})
    assert reset.status_code == 200
    assert client.post('/api/auth/reset-password', json={'token': token, 'password': 'another-secret'}).status_code == 400
    assert client.post('/api/auth/login', json={'email': f'{username}@example.com', 'password': 'new-secret123'}).status_code == 200


def test_admin_can_remove_user_with_owned_data():
    headers = login_headers()
    username = f'data-owner-{uuid4().hex}'
    create_response = client.post('/api/admin/users', headers=headers, json={
        'username': username,
        'email': f'{username}@example.com',
        'password': 'secret123',
    })
    assert create_response.status_code == 201
    user_id = create_response.json()['id']
    user_token = issue_token(user_id, username)['access_token']
    user_headers = {'Authorization': f'Bearer {user_token}'}

    link_response = client.post('/api/links', headers=user_headers, json={
        'title': 'Owned link',
        'url': 'https://example.com/owned',
        'comment': 'Owned data',
        'tags': ['owned'],
    })
    assert link_response.status_code == 201
    label_response = client.post('/api/user/labels', headers=user_headers, json={'name': f'{username}-label'})
    assert label_response.status_code == 201
    plugin_response = client.put('/api/user/plugins/mastodon', headers=user_headers, json={
        'instance': 'mastodon.social',
    })
    assert plugin_response.status_code == 200

    removed = client.delete(f'/api/admin/users/{user_id}', headers=headers)
    assert removed.status_code == 200
    assert client.get('/api/auth/me', params={'token': user_token}).status_code == 401


def test_deleting_link_removes_all_mastodon_posts_first():
    owner_headers = login_headers('alice')
    created = client.post('/api/links', headers=owner_headers, json={
        'title': 'Remote cleanup',
        'url': 'https://example.com/remote-cleanup',
    })
    assert created.status_code == 201
    link_id = created.json()['id']
    with get_connection() as conn:
        conn.execute(
            'UPDATE links SET mastodon_posted = 1, mastodon_post_id = ?, mastodon_post_ids = ? WHERE id = ?',
            ('post-2', json.dumps(['post-1', 'post-2']), link_id),
        )
        conn.commit()

    with patch('backend.app.api.links.plugin_manager.delete_mastodon_posts', return_value={'status': 'deleted', 'count': 2}) as delete_posts:
        deleted = client.delete(f'/api/links/{link_id}', headers=owner_headers)
    assert deleted.status_code == 200
    delete_posts.assert_called_once()
    assert delete_posts.call_args.args[0]['mastodon_post_ids'] == ['post-1', 'post-2']
    assert client.delete(f'/api/links/{link_id}', headers=owner_headers).status_code == 404


def test_link_is_kept_when_mastodon_cleanup_fails():
    owner_headers = login_headers('alice')
    created = client.post('/api/links', headers=owner_headers, json={
        'title': 'Failed remote cleanup',
        'url': 'https://example.com/failed-remote-cleanup',
    })
    link_id = created.json()['id']
    with get_connection() as conn:
        conn.execute(
            'UPDATE links SET mastodon_posted = 1, mastodon_post_id = ?, mastodon_post_ids = ? WHERE id = ?',
            ('post-failed', json.dumps(['post-failed']), link_id),
        )
        conn.commit()
    with patch('backend.app.api.links.plugin_manager.delete_mastodon_posts', return_value={'status': 'failed', 'reason': 'remote refused'}):
        deleted = client.delete(f'/api/links/{link_id}', headers=owner_headers)
    assert deleted.status_code == 502
    assert 'remote refused' in deleted.json()['detail']
    assert client.get('/api/links').json()


def test_admin_can_toggle_privileges_without_removing_last_admin():
    headers = login_headers()
    bob = next(user for user in client.get('/api/admin/users', headers=headers).json() if user['username'] == 'bob')

    promote = client.put(f"/api/admin/users/{bob['id']}", headers=headers, json={'is_admin': True})
    assert promote.status_code == 200
    assert promote.json()['is_admin'] is True

    demote = client.put(f"/api/admin/users/{bob['id']}", headers=headers, json={'is_admin': False})
    assert demote.status_code == 200
    assert demote.json()['is_admin'] is False

    last_admin = client.put('/api/admin/users/user-1', headers=headers, json={'is_admin': False})
    assert last_admin.status_code == 400


def test_users_manage_owned_labels_and_admin_can_delete_any_label():
    alice_headers = login_headers('alice')
    created = client.post('/api/user/labels', headers=alice_headers, json={'name': 'My Label'})
    assert created.status_code == 201
    label = created.json()
    assert label['name'] == '#My Label'

    edited = client.put(f"/api/user/labels/{label['id']}", headers=alice_headers, json={'name': '#Renamed'})
    assert edited.status_code == 200
    assert edited.json()['name'] == '#Renamed'

    denied = client.put(f"/api/user/labels/{label['id']}", headers=login_headers('bob'), json={'name': '#Nope'})
    assert denied.status_code == 404
    assert client.delete(f"/api/user/labels/{label['id']}", headers=login_headers('bob')).status_code == 404

    admin_delete = client.delete(f"/api/admin/labels/{label['id']}", headers=alice_headers)
    assert admin_delete.status_code == 200


def test_labels_page_renders_authenticated_management_shell():
    page = client.get('/labels')
    assert page.status_code == 200
    assert 'Labels' in page.text
    assert 'labels.js?v=1' in page.text


def test_submit_link_stores_cleaned_url_and_public_feed():
    response = client.post('/api/links', headers=login_headers(), json={
        'title': 'Example page',
        'url': 'https://example.com/path?utm_source=ad&utm_medium=email&keep=yes',
        'comment': 'Interesting read',
        'timestamp': '2026-08-24T12:00:00Z',
    })

    assert response.status_code == 201
    data = response.json()
    assert data['url'] == 'https://example.com/path?keep=yes'

    feed = client.get('/api/public/feed').json()
    matching = next(item for item in feed if item['id'] == data['id'])
    assert matching['comment'] == 'Interesting read'
    assert matching['user']['username'] == 'alice'
    assert 'avatar_url' in matching['user']

    filtered_response = client.get('/api/public/feed/alice')
    assert filtered_response.status_code == 200
    assert all(item['user']['username'] == 'alice' for item in filtered_response.json())
    assert client.get('/api/public/feed/does-not-exist').json() == []
    users_response = client.get('/api/public/users')
    assert users_response.status_code == 200
    assert 'alice' in users_response.json()


def test_duplicate_link_updates_comment_tags_and_retriggers_plugins():
    headers = login_headers()
    payload = {
        'title': 'Duplicate candidate',
        'url': 'https://example.com/duplicate?utm_source=campaign',
        'comment': 'first comment',
        'tags': ['#First'],
    }
    first = client.post('/api/links', headers=headers, json=payload)
    assert first.status_code == 201

    with patch('backend.app.api.links.plugin_manager.dispatch', return_value=[]) as dispatch:
        duplicate = client.post('/api/links', headers=headers, json={
            **payload,
            'comment': 'updated comment',
            'tags': ['#Second'],
        })
    assert duplicate.status_code == 200
    assert duplicate.json()['duplicate'] is True
    assert duplicate.json()['id'] == first.json()['id']
    dispatch.assert_called_once()
    assert dispatch.call_args.args[0]['comment'] == 'updated comment'
    assert dispatch.call_args.args[0]['tags'] == ['#Second']
    feed_item = next(item for item in client.get('/api/public/feed').json() if item['id'] == first.json()['id'])
    assert feed_item['comment'] == 'updated comment'
    assert feed_item['tags'] == ['#Second']


def test_duplicate_link_check_is_authenticated_and_detects_existing_entry():
    headers = login_headers()
    payload = {'title': 'Check candidate', 'url': 'https://example.com/check-candidate'}
    assert client.get('/api/links/check', params=payload).status_code == 401
    created = client.post('/api/links', headers=headers, json=payload)
    assert created.status_code == 201
    check = client.get('/api/links/check', headers=headers, params=payload)
    assert check.status_code == 200
    assert check.json()['exists'] is True


def test_link_save_reports_plugin_posting_errors():
    with patch('backend.app.api.links.plugin_manager.dispatch', return_value=[
        {'status': 'failed', 'plugin': 'mastodon', 'reason': 'HTTP 503: unavailable'},
    ]):
        response = client.post('/api/links', headers=login_headers(), json={
            'title': 'Plugin error report',
            'url': 'https://example.com/plugin-error-report',
        })
    assert response.status_code == 201
    assert response.json()['plugin_errors'] == [{'plugin': 'mastodon', 'reason': 'HTTP 503: unavailable'}]


def test_links_support_tags_and_tag_filtering():
    headers = login_headers()
    response = client.post('/api/links', headers=headers, json={
        'title': 'Tagged page',
        'url': 'https://example.com/tagged',
        'tags': ['#CasePreserved', '#web', 'casepreserved'],
    })
    assert response.status_code == 201
    link_id = response.json()['id']
    assert response.json()['tags'] == ['#CasePreserved', '#web']

    tags = client.get('/api/tags').json()
    assert any(tag.casefold() == '#casepreserved' for tag in tags) and '#web' in tags
    assert {
        '#Internet', '#Cybersecurity', '#Fediverse', '#Food', '#Photography', '#Music', '#AI'
    } <= set(tags)
    tagged_feed = client.get('/api/public/feed').json()
    tagged_item = next(item for item in tagged_feed if item['id'] == link_id)
    assert {tag.casefold() for tag in tagged_item['tags']} == {'#casepreserved', '#web'}

    too_many = client.post('/api/links', headers=headers, json={
        'title': 'Too many tags',
        'url': 'https://example.com/too-many',
        'tags': [f'tag-{index}' for index in range(11)],
    })
    assert too_many.status_code == 422

    edited = client.put(f'/api/links/{link_id}', headers=headers, json={
        'title': 'Tagged page',
        'url': 'https://example.com/tagged',
        'tags': ['edited'],
    })
    assert edited.status_code == 200
    assert edited.json()['tags'] == ['#edited']


def test_only_link_owner_can_edit_link():
    owner_headers = login_headers('alice')
    response = client.post('/api/links', headers=owner_headers, json={
        'title': 'Editable link',
        'url': 'https://example.com/editable?utm_source=test',
        'comment': 'Before edit',
    })
    assert response.status_code == 201
    link_id = response.json()['id']

    unauthenticated = client.put(f'/api/links/{link_id}', json={
        'title': 'Not allowed',
        'url': 'https://example.com/not-allowed',
    })
    assert unauthenticated.status_code == 401

    anonymous_feed = client.get('/api/public/feed').json()
    anonymous_link = next(item for item in anonymous_feed if item['id'] == link_id)
    assert anonymous_link['is_owner'] is False
    assert anonymous_link['can_edit'] is False

    owner_feed = client.get('/api/public/feed', headers=owner_headers).json()
    owner_link = next(item for item in owner_feed if item['id'] == link_id)
    assert owner_link['is_owner'] is True
    assert owner_link['can_edit'] is True

    edited = client.put(f'/api/links/{link_id}', headers=owner_headers, json={
        'title': 'Edited link',
        'url': 'https://example.com/edited',
        'comment': 'After edit',
    })
    assert edited.status_code == 200
    assert edited.json()['title'] == 'Edited link'

    denied = client.put(f'/api/links/{link_id}', headers=login_headers('bob'), json={
        'title': 'Not allowed',
        'url': 'https://example.com/not-allowed',
    })
    assert denied.status_code == 404

    deleted = client.delete(f'/api/links/{link_id}', headers=owner_headers)
    assert deleted.status_code == 200
    assert client.delete(f'/api/links/{link_id}', headers=owner_headers).status_code == 404
    assert client.delete(f'/api/links/{link_id}', headers=login_headers('bob')).status_code == 404


def test_logout_revokes_token_and_admin_can_list_plugins():
    headers = login_headers()
    assert client.post('/api/auth/logout', headers=headers).status_code == 200

    revoked_response = client.post('/api/links', headers=headers, json={
        'title': 'Should fail',
        'url': 'https://example.com/path',
        'comment': 'nope',
    })
    assert revoked_response.status_code == 401

    plugins_response = client.get('/api/admin/plugins', headers=login_headers())
    assert plugins_response.status_code == 200
    assert any(plugin['name'] == 'default_frontend' for plugin in plugins_response.json())


def test_public_and_admin_pages_render_html():
    root_page = client.get('/')
    assert 'LinkLog' in root_page.text
    assert 'src="/static/logo.svg"' in root_page.text
    assert 'class="site-footer"' in root_page.text
    assert 'https://git.kolkman.org/' in root_page.text
    assert 'class="menu-toggle"' in root_page.text
    assert 'id="auth-menu" class="auth-menu hidden"' in root_page.text
    assert 'id="auth-home-link" href="/">Home</a>' in root_page.text
    assert '<a id="auth-username" class="user-name" href="/">' in root_page.text
    assert 'id="auth-avatar"' not in root_page.text
    assert 'id="auth-login-button" href="/login"' in root_page.text
    assert 'id="auth-profile-link" class="hidden"' in root_page.text
    assert 'id="auth-admin-link" class="hidden"' in root_page.text
    assert '<select id="tag-filter">' in root_page.text
    assert root_page.text.index('class="site-logo"') < root_page.text.index('<section class="toolbar">')
    assert 'logout.js?v=3' in root_page.text
    assert client.get('/alice').status_code == 200
    assert client.get('/alice/').status_code == 200
    user_page = client.get('/alice').text
    assert 'alice' in user_page
    assert 'data-user-filter="alice"' in user_page
    assert 'profile-summary' in user_page
    feed_script = TestClient(app).get('/static/feed.js?v=4').text
    assert 'window.location.assign(selectedUser ? `/${encodeURIComponent(selectedUser)}/` : \'/\')' in feed_script
    assert client.get('/login').status_code == 200
    login_page = client.get('/login').text
    assert 'Sign in' in login_page
    assert 'name="otp"' in login_page
    assert 'src="/static/logo.svg"' in login_page
    assert 'id="auth-session" class="auth-session hidden"' in login_page
    assert 'logout.js?v=3' in login_page
    about_page = client.get('/about')
    assert about_page.status_code == 200
    assert 'Save the good stuff' in about_page.text
    assert '<h2>Plugin</h2>' in about_page.text
    assert 'https://git.kolkman.org/olaf/Link-Log/raw/branch/main/XPI/signed/LinkLog-0.1.0.xpi' in about_page.text
    assert 'id="auth-about-link" href="/about"' in about_page.text
    assert client.get('/admin').status_code == 200
    admin_page = client.get('/admin').text
    assert 'Admin' in admin_page
    assert 'id="admin-controls" class="hidden"' in admin_page
    assert 'id="admin-auth-notice" class="auth-notice hidden"' in admin_page
    assert 'id="auth-menu" class="auth-menu hidden"' in admin_page
    assert 'id="auth-home-link" href="/">Home</a>' in admin_page
    assert '<a id="auth-username" class="user-name" href="/">' in admin_page
    assert 'admin.js?v=5' in admin_page
    feed_script = client.get('/static/feed.js?v=7').text
    assert 'if (item.is_owner && !showIdentity)' in feed_script
    assert 'deleteEntry(item, deleteButton)' in feed_script
    assert 'postToMastodon(item, mastodonButton)' in feed_script
    assert 'tag.toLowerCase() === pref.tag.toLowerCase()' in feed_script
    assert 'return `${date.getFullYear()} ${months[date.getMonth()]} ${date.getDate()} - ${hours}:${minutes}`' in feed_script
    assert "entryMeta.className = 'entry-meta'" in feed_script
    assert "meta.className = 'meta'" in feed_script
    assert "comment.textContent = item.comment || ''" in feed_script
    assert 'profileLink.href = `/${encodeURIComponent(username)}/`' in feed_script
    assert 'edit-tag-options' in feed_script
    assert 'new_tags' in feed_script
    assert 'A link can have at most 10 tags.' in feed_script


def test_link_submission_posts_to_enabled_mastodon_plugin():
    received = {}

    class MastodonHandler(BaseHTTPRequestHandler):
        def do_POST(self):
            received['path'] = self.path
            received['authorization'] = self.headers['Authorization']
            received['content_type'] = self.headers['Content-Type']
            received['body'] = parse_qs(self.rfile.read(int(self.headers['Content-Length'])).decode())
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            self.wfile.write(b'{"id":"status-1"}')

        def log_message(self, format, *args):
            return

    server = ThreadingHTTPServer(('127.0.0.1', 0), MastodonHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        headers = login_headers()
        base_url = f'http://127.0.0.1:{server.server_port}'
        with patch('backend.app.services.plugin_manager.validate_public_instance', return_value=base_url):
            assert client.put('/api/user/plugins/mastodon', headers=headers, json={
                'instance': base_url,
                'access_token': 'test-token',
                'post_prefix': 'From my #LinkLog: ',
            }).status_code == 200
            assert client.put('/api/admin/plugins/mastodon', headers=headers, json={'enabled': True}).status_code == 200

            response = client.post('/api/links', headers=headers, json={
                'title': 'A useful page',
                'url': 'https://example.com/useful',
                'comment': 'Worth sharing',
                'tags': ['#python', '#web'],
            })

        assert response.status_code == 201
        assert received['path'] == '/api/v1/statuses'
        assert received['authorization'] == 'Bearer test-token'
        assert received['content_type'] == 'application/x-www-form-urlencoded'
        assert received['body'] == {'status': ['From my #LinkLog:\n\nA useful page\n\nWorth sharing\n\nfrom: https://example.com/useful\n\n#python #web']}
        posted_item = next(item for item in client.get('/api/public/feed/alice', headers=headers).json() if item['id'] == response.json()['id'])
        assert posted_item['mastodon_posted'] is True
    finally:
        server.shutdown()
        thread.join()
        server.server_close()


def test_mastodon_post_without_title_omits_source_line():
    from backend.app.services.plugin_manager import MastodonPlugin

    response = MagicMock()
    response.__enter__.return_value.read.return_value = b'{"id":"status-2"}'
    plugin = MastodonPlugin()
    plugin.initialize({'instance': 'https://mastodon.example', 'access_token': 'test-token'})

    with patch('backend.app.services.plugin_manager.open_no_redirect', return_value=response) as open_url, \
            patch('backend.app.services.plugin_manager.validate_public_instance', return_value='https://mastodon.example'):
        result = plugin.handle_event({
            'url': 'https://example.com/useful',
            'title': '',
            'comment': 'A comment',
            'tags': ['#tag'],
        })

    body = parse_qs(open_url.call_args.args[0].data.decode())['status'][0]
    assert result['status'] == 'posted'
    assert body == 'From my #LinkLog:\n\nA comment\n\n#tag'


def test_plugin_config_can_be_saved_for_mastodon():
    headers = login_headers()
    assert client.put('/api/user/plugins/mastodon', headers=headers, json={
        'instance': 'mastodon.social',
        'access_token': 'demo-token',
        'post_prefix': 'From my #LinkLog: ',
    }).status_code == 200

    payload = client.get('/api/user/plugins/mastodon', headers=headers).json()
    assert payload['instance'] == 'mastodon.social'
    assert payload['post_prefix'] == 'From my #LinkLog: '

    admin_update = client.put('/api/admin/plugins/mastodon', headers=headers, json={
        'enabled': True,
        'config': {'instance': 'mastodon.social'},
    })
    assert admin_update.status_code == 200