"""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, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP); 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); CREATE TABLE user_badges(user_id INTEGER REFERENCES users(id) ON DELETE CASCADE, badge_code TEXT, awarded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY(user_id,badge_code)); ''') 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', '21_push_notifications.sql', '22_registration_badges.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()