317 lines
13 KiB
Python
317 lines
13 KiB
Python
import json
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.app.main import app
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def login_headers(username='alice'):
|
|
token = client.post('/api/auth/login', json={
|
|
'username': username,
|
|
'password': 'secret123',
|
|
}).json()['access_token']
|
|
return {'Authorization': f'Bearer {token}'}
|
|
|
|
|
|
def test_login_returns_token():
|
|
response = client.post('/api/auth/login', json={
|
|
'username': 'alice',
|
|
'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', params={'token': 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={
|
|
'username': 'bob',
|
|
'password': 'secret123',
|
|
}).json()['access_token']
|
|
user_session = client.get('/api/auth/me', params={'token': user_token})
|
|
assert user_session.status_code == 200
|
|
assert user_session.json()['is_admin'] is False
|
|
|
|
|
|
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
|
|
|
|
|
|
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
|
|
|
|
|
|
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_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_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
|
|
|
|
|
|
def test_logout_revokes_token_and_admin_can_list_plugins():
|
|
headers = login_headers()
|
|
token = headers['Authorization'].removeprefix('Bearer ')
|
|
assert client.post('/api/auth/logout', json={'token': token}).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 '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 '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 'id="auth-session" class="auth-session hidden"' in login_page
|
|
assert 'logout.js?v=3' in login_page
|
|
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=3' in admin_page
|
|
feed_script = client.get('/static/feed.js?v=7').text
|
|
assert 'if (item.is_owner)' 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 '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['body'] = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
|
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}'
|
|
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['body'] == {
|
|
'status': 'A useful page\nWorth sharing\nFrom my #LinkLog: "https://example.com/useful #python #web',
|
|
}
|
|
finally:
|
|
server.shutdown()
|
|
thread.join()
|
|
server.server_close()
|
|
|
|
|
|
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
|