kai/pingu-concerts
1
0
Fork 0
Code Issues 10 Pull Requests Actions Packages Projects Releases Wiki Activity

Add FCM device registration and Gitea bug reporter

Browse Source
This commit is contained in:
MetalCircle Codex Bot
2026-09-15 01:01:08 +02:00
parent b2af866cf0
commit b96b1890d2
33 changed files with 1232 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tests/test_feature_api.py
+210
View File
@@ -0,0 +1,210 @@
"""Opt-in PostgreSQL integration tests, isolated in a disposable schema.
METALCIRCLE_TEST_DATABASE=1 python -m unittest discover -s tests
Never enable this flag against a non-local database.
"""
from concurrent.futures import ThreadPoolExecutor
import os
from pathlib import Path
import re
import unittest
from unittest.mock import patch
from uuid import uuid4
import bcrypt
import psycopg
from psycopg import sql
from psycopg.conninfo import make_conninfo
from fastapi.testclient import TestClient
import main
from feature_schema import FEATURE_SCHEMA
from gitea_service import GiteaError, GiteaService
@unittest.skipUnless(os.environ.get('METALCIRCLE_TEST_DATABASE') == '1', 'requires explicit local test DB opt-in')
class FeatureApiTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.original_dsn = main.DATABASE_URL
cls.schema = 'metalcircle_test_' + uuid4().hex
with psycopg.connect(cls.original_dsn) as db:
db.execute(sql.SQL('CREATE SCHEMA {}').format(sql.Identifier(cls.schema)))
main.DATABASE_URL = make_conninfo(cls.original_dsn, options='-csearch_path='+cls.schema)
with main.get_db_connection() as db:
db.execute('''
CREATE TABLE users(id SERIAL PRIMARY KEY, username TEXT UNIQUE, email TEXT UNIQUE,
display_name TEXT, password_hash TEXT, is_admin BOOLEAN DEFAULT FALSE);
CREATE TABLE sessions(id SERIAL PRIMARY KEY, user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
token_hash TEXT UNIQUE, expires_at TIMESTAMP);
CREATE TABLE friendships(addressee_id INTEGER, status TEXT);
CREATE TABLE direct_messages(recipient_id INTEGER, read_at TIMESTAMP);
CREATE TABLE event_invitations(user_id INTEGER, viewed_at TIMESTAMP);
''')
for statement in FEATURE_SCHEMA: db.execute(statement)
cls.password_hash = bcrypt.hashpw(b'Test-only-password-123', bcrypt.gensalt()).decode()
@classmethod
def tearDownClass(cls):
main.DATABASE_URL = cls.original_dsn
with psycopg.connect(cls.original_dsn) as db:
db.execute(sql.SQL('DROP SCHEMA {} CASCADE').format(sql.Identifier(cls.schema)))
def setUp(self):
self.secure = patch.object(main, 'COOKIE_SECURE', False)
self.secure.start()
main.rate_limit_buckets.clear()
with main.get_db_connection() as db:
db.execute('TRUNCATE users RESTART IDENTITY CASCADE')
for username in ('tester_a', 'tester_b'):
db.execute('INSERT INTO users(username,email,display_name,password_hash) VALUES (%s,%s,%s,%s)',
(username, username+'@example.invalid', username, self.password_hash))
self.client = TestClient(main.app)
self.client.headers['Origin'] = 'http://testserver'
self.login('tester_a')
def tearDown(self):
self.secure.stop()
def login(self, username):
result = self.client.post('/login', data={'username': username, 'password': 'Test-only-password-123'}, follow_redirects=False)
self.assertEqual(result.status_code, 303)
def device(self, **updates):
data = dict(device_id=str(uuid4()), token='synthetic-fcm-token-'+'a'*120, platform='android',
app_version='1.1.0-debug', session_tag=self.client.get('/api/push/session').json()['session_tag'])
data.update(updates)
return data
def count_devices(self):
with main.get_db_connection() as db: return db.execute('SELECT count(*) FROM push_devices').fetchone()[0]
def test_migrations_repeat_and_match_startup_schema(self):
with main.get_db_connection() as db:
for name, runtime in zip(('19_push_devices.sql', '20_bug_report_submissions.sql'), FEATURE_SCHEMA):
source = Path('/test-migrations', name).read_text()
normalize = lambda s: re.sub(r'\s+', '', re.sub(r'--[^\n]*', '', s))
self.assertEqual(normalize(source), normalize(runtime))
db.execute(source)
db.execute(source)
def test_repeated_registration_rotation_and_multiple_devices(self):
data = self.device()
for _ in range(2): self.assertEqual(self.client.post('/api/push/devices', json=data).status_code, 200)
self.assertEqual(self.count_devices(), 1)
data['token'] = 'synthetic-fcm-token-'+'b'*120
self.assertEqual(self.client.post('/api/push/devices', json=data).status_code, 200)
with main.get_db_connection() as db:
self.assertEqual(db.execute('SELECT token FROM push_devices').fetchone()[0], data['token'])
self.assertEqual(self.client.post('/api/push/devices', json=self.device()).status_code, 200)
self.assertEqual(self.count_devices(), 2)
def test_logout_login_user_switch_and_stale_registration(self):
data = self.device()
self.client.post('/api/push/devices', json=data)
self.client.post('/logout', follow_redirects=False)
self.assertEqual(self.count_devices(), 0)
self.assertFalse(self.client.get('/api/push/session').json()['authenticated'])
self.assertEqual(self.client.post('/api/push/devices', json=data).status_code, 401)
self.login('tester_b')
self.assertEqual(self.client.post('/api/push/devices', json=data).status_code, 409)
data['session_tag'] = self.client.get('/api/push/session').json()['session_tag']
self.assertEqual(self.client.post('/api/push/devices', json=data).status_code, 200)
with main.get_db_connection() as db:
self.assertEqual(db.execute('SELECT user_id FROM push_devices').fetchone()[0], 2)
self.login('tester_a') # successful replacement login also revokes B's binding
self.assertEqual(self.count_devices(), 0)
def test_token_transfer_and_owner_only_unregister(self):
data = self.device()
self.client.post('/api/push/devices', json=data)
other = TestClient(main.app, headers={'Origin': 'http://testserver'})
other.post('/login', data={'username':'tester_b','password':'Test-only-password-123'}, follow_redirects=False)
self.assertEqual(other.delete('/api/push/devices/'+data['device_id']).status_code, 200)
self.assertEqual(self.count_devices(), 1)
transferred = dict(data, session_tag=other.get('/api/push/session').json()['session_tag'])
other.post('/api/push/devices', json=transferred)
self.client.post('/logout', follow_redirects=False)
self.assertEqual(self.count_devices(), 1)
other.post('/logout', follow_redirects=False)
self.assertEqual(self.count_devices(), 0)
def test_push_validation_csrf_and_no_echoed_token(self):
data = self.device(token='secret invalid token!')
response = self.client.post('/api/push/devices', json=data)
self.assertEqual(response.status_code, 422)
self.assertNotIn(data['token'], response.text)
self.assertEqual(self.client.post('/api/push/devices', json=self.device(), headers={'Origin':'http://evil.invalid'}).status_code, 403)
def new_report(self):
page = self.client.get('/bug-report?route=/concerts/481?token=secret-query')
self.assertEqual(page.status_code, 200)
return re.search(r'name="submission_id" value="([^"]+)"', page.text).group(1)
def report(self, **updates):
data = dict(submission_id=self.new_report(), title='Local test report', description='A sufficiently detailed description.',
expected='The expected result.', steps='One, two, three.', category='android', severity='normal')
data.update(updates)
return data
def test_report_success_session_identity_and_no_internal_link(self):
data = self.report(username='forged', user_id='9999', technical='true', route='/concerts/481?token=private')
with patch.object(GiteaService, 'create_issue', return_value=123) as create:
response = self.client.post('/bug-report', data=data)
self.assertEqual(response.status_code, 200)
self.assertIn('#123', response.text)
body = create.call_args.args[1]
self.assertIn('tester_a', body)
self.assertNotIn('forged', body)
self.assertNotIn('9999', body)
self.assertNotIn('token=private', body)
self.assertNotIn('192.168.', response.text)
self.assertNotIn('GITEA_TOKEN', response.text)
self.client.post('/logout', follow_redirects=False)
self.assertEqual(self.client.get('/bug-report', follow_redirects=False).status_code, 303)
self.assertEqual(self.client.post('/bug-report', data=data, follow_redirects=False).status_code, 303)
def test_report_validation(self):
for changes in ({'title':''}, {'description':''}, {'expected':''}, {'title':'x'*161},
{'description':'x'*5001}, {'expected':'x'*3001}, {'steps':'x'*3001},
{'category':'bad'}, {'severity':'bad'}):
main.rate_limit_buckets.clear()
with self.subTest(changes=list(changes)), patch.object(GiteaService, 'create_issue') as create:
self.assertEqual(self.client.post('/bug-report', data=self.report(**changes)).status_code, 400)
create.assert_not_called()
def test_duplicate_submit_and_cooldown(self):
data = self.report()
with patch.object(GiteaService, 'create_issue', return_value=123) as create:
self.client.post('/bug-report', data=data)
self.client.post('/bug-report', data=data)
self.assertEqual(create.call_count, 1)
self.assertEqual(self.client.post('/bug-report', data=self.report()).status_code, 429)
self.assertEqual(create.call_count, 1)
def test_parallel_duplicate(self):
data = self.report()
with patch.object(GiteaService, 'create_issue', return_value=123) as create:
with ThreadPoolExecutor(max_workers=2) as pool:
results = list(pool.map(lambda _: self.client.post('/bug-report', data=data, follow_redirects=False), range(2)))
self.assertEqual(create.call_count, 1)
self.assertTrue(all(response.status_code in (303,409) for response in results))
def test_gitea_failure_keeps_form_and_core_app_available(self):
data = self.report()
with patch.object(GiteaService, 'create_issue', side_effect=GiteaError('http_403')):
response = self.client.post('/bug-report', data=data)
self.assertEqual(response.status_code, 503)
self.assertIn(data['title'], response.text)
self.assertEqual(self.client.get('/impressum').status_code, 200)
def test_unknown_delivery_is_not_retried(self):
data = self.report()
with patch.object(GiteaService, 'create_issue', side_effect=GiteaError('timeout', uncertain=True)) as create:
self.assertEqual(self.client.post('/bug-report', data=data).status_code, 503)
self.assertEqual(self.client.post('/bug-report', data=data).status_code, 409)
self.assertEqual(create.call_count, 1)
if __name__ == '__main__':
unittest.main()
app/tests/test_gitea.py
+104
View File
@@ -0,0 +1,104 @@
import json
import unittest
from unittest.mock import patch
import httpx
from starlette.requests import Request
from bug_reporter import browser_summary, issue_body, safe_route
from gitea_service import GiteaConfig, GiteaError, GiteaService, redact
class GiteaTests(unittest.TestCase):
def service(self, handler):
return GiteaService(GiteaConfig('http://gitea.invalid', 'test-secret-never-print', 'kai', 'pingu-concerts'), httpx.MockTransport(handler))
def test_existing_labels_bot_and_payload(self):
requests = []
def handler(request):
requests.append(request)
if request.url.path.endswith('/user'):
return httpx.Response(200, json={'login': 'metalcircle-bot'})
if request.url.path.endswith('/labels'):
return httpx.Response(200, json=[{'id': 7, 'name': 'reported-from-metalcircle'}, {'id': 9, 'name': 'android'}, {'id': 99, 'name': 'unrelated'}])
return httpx.Response(201, json={'number': 123, 'user': {'login': 'metalcircle-bot'}})
self.assertEqual(self.service(handler).create_issue('A test title', 'A test body', 'android'), 123)
payload = json.loads(requests[-1].content)
self.assertEqual(payload['labels'], [7, 9])
self.assertEqual(payload['title'], '[MetalCircle Bug] A test title')
self.assertEqual(requests[-1].url.path, '/api/v1/repos/kai/pingu-concerts/issues')
self.assertNotIn('test-secret', str(requests[-1].url))
def test_missing_or_inaccessible_labels_do_not_block(self):
for response in (httpx.Response(200, json=[]), httpx.Response(403), httpx.Response(200, json={'invalid': True})):
def handler(request):
if request.url.path.endswith('/user'): return httpx.Response(200, json={'login': 'metalcircle-bot'})
if request.url.path.endswith('/labels'): return response
self.assertEqual(json.loads(request.content)['labels'], [])
return httpx.Response(201, json={'number': 4, 'user': {'login': 'metalcircle-bot'}})
self.assertEqual(self.service(handler).create_issue('Test title', 'body', 'general'), 4)
def test_personal_account_is_rejected_before_post(self):
def handler(request):
self.assertEqual(request.method, 'GET')
return httpx.Response(200, json={'login': 'kai'})
with self.assertRaisesRegex(GiteaError, 'wrong_account'):
self.service(handler).create_issue('title', 'body', 'general')
def test_http_errors_and_secret_free_logs(self):
for code in (401, 403, 404, 500, 502):
def handler(request):
if request.url.path.endswith('/user'): return httpx.Response(200, json={'login': 'metalcircle-bot'})
if request.url.path.endswith('/labels'): return httpx.Response(200, json=[])
return httpx.Response(code, text='test-secret-never-print')
with self.subTest(code=code), self.assertLogs('gitea_service', 'WARNING') as logs:
with self.assertRaises(GiteaError) as caught:
self.service(handler).create_issue('title', 'body', 'general')
self.assertEqual(caught.exception.uncertain, code >= 500)
self.assertNotIn('test-secret-never-print', '\n'.join(logs.output))
def test_timeout_unreachable_and_invalid_response(self):
for failure in ('timeout', 'unreachable', 'invalid_json', 'invalid_issue'):
def handler(request):
if failure == 'unreachable': raise httpx.ConnectError('test-secret-never-print')
if request.url.path.endswith('/user'): return httpx.Response(200, json={'login': 'metalcircle-bot'})
if request.url.path.endswith('/labels'): return httpx.Response(200, json=[])
if failure == 'timeout': raise httpx.ReadTimeout('test-secret-never-print')
if failure == 'invalid_json': return httpx.Response(201, text='not json')
return httpx.Response(201, json={'number': 'x'})
with self.subTest(failure=failure), self.assertLogs('gitea_service', 'WARNING') as logs:
with self.assertRaises(GiteaError) as caught:
self.service(handler).create_issue('title', 'body', 'general')
self.assertEqual(caught.exception.uncertain, failure != 'unreachable')
self.assertNotIn('test-secret', '\n'.join(logs.output))
def test_context_allowlist_and_redaction(self):
secret = 'session-secret-never-send'
request = Request({'type': 'http', 'headers': [
(b'cookie', ('pingu_session='+secret).encode()),
(b'authorization', b'Bearer authorization-secret'),
(b'user-agent', b'Chrome/128.0 secret-header'),
]})
data = dict(description='This is '+secret, expected='Expected result', steps='', category='android',
severity='normal', technical=True, platform='Android', app_version='1.1.0-debug',
route='/concerts/1?token=secret-query')
with patch.dict('os.environ', {'GITEA_TOKEN': 'bot-secret-never-send'}):
body = issue_body(data, {'id': 12, 'username': 'real-user'}, request)
for excluded in (secret, 'authorization-secret', 'secret-header', 'secret-query', 'bot-secret-never-send'):
self.assertNotIn(excluded, body)
self.assertIn('real-user', body)
self.assertIn('12', body)
self.assertIn('/concerts/1', body)
self.assertIn('Chrome 128.0', body)
self.assertEqual(safe_route('/password-reset/sensitive'), '/')
self.assertEqual(safe_route('/register/sensitive'), '/')
self.assertEqual(browser_summary('Cookie: evil'), 'Unknown')
data['technical'] = False
self.assertNotIn('Chrome', issue_body(data, {'id': 12, 'username': 'real-user'}, request))
self.assertNotIn('/concerts/1', issue_body(data, {'id': 12, 'username': 'real-user'}, request))
self.assertNotIn('super-secret', redact('password=super-secret'))
self.assertNotIn('private bytes', redact('-----BEGIN PRIVATE KEY-----\nprivate bytes\n-----END PRIVATE KEY-----'))
if __name__ == '__main__':
unittest.main()
app/tests/test_i18n.py
+2 -1
View File
@@ -121,7 +121,8 @@ class TranslationTests(unittest.TestCase):
current_language.set(language)
for path in Path(main.BASE_DIR, 'templates').glob('*.html'):
with self.subTest(language=language, template=path.name):
page = env.get_template(path.name).render(**context)
from bug_reporter import CATEGORIES, SEVERITIES
page = env.get_template(path.name).render(**context, categories=CATEGORIES, severities=SEVERITIES, form={})
if not path.name.startswith('_'):
self.assertIn(f'<html lang="{language}">', page)
self.assertIn('class="language-switch"', page)