208 lines
8.1 KiB
Python
208 lines
8.1 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'
|
|
|
|
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() == []
|
|
|
|
|
|
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="login-button" href="/login"' in root_page.text
|
|
assert client.get('/alice').status_code == 200
|
|
assert 'alice' in client.get('/alice').text
|
|
assert client.get('/login').status_code == 200
|
|
assert 'Sign in' in client.get('/login').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="admin-login-button" class="login-button" href="/login"' in admin_page
|
|
assert 'id="logout-button" class="logout-button hidden"' in admin_page
|
|
|
|
|
|
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',
|
|
})
|
|
|
|
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',
|
|
}
|
|
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
|