"""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, avatar_path TEXT, instagram_url TEXT, profile_visibility TEXT DEFAULT 'public', 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(id SERIAL PRIMARY KEY, requester_id INTEGER, addressee_id INTEGER, status TEXT); CREATE TABLE direct_messages(recipient_id INTEGER, read_at TIMESTAMP); CREATE TABLE event_invitations(concert_id INTEGER, user_id INTEGER, viewed_at TIMESTAMP); CREATE TABLE user_blocks(blocker_id INTEGER REFERENCES users(id) ON DELETE CASCADE, blocked_id INTEGER REFERENCES users(id) ON DELETE CASCADE, PRIMARY KEY(blocker_id,blocked_id)); CREATE TABLE venues(id SERIAL PRIMARY KEY, name TEXT, city TEXT); CREATE TABLE concerts(id SERIAL PRIMARY KEY, artist TEXT, start_datetime TIMESTAMP, end_datetime TIMESTAMP, venue_id INTEGER REFERENCES venues(id), visibility TEXT, created_by INTEGER, event_type TEXT DEFAULT 'concert'); CREATE TABLE concert_bands(concert_id INTEGER, band_key TEXT, display_name TEXT, position SMALLINT DEFAULT 0); CREATE TABLE concert_attendance(concert_id INTEGER, user_id INTEGER, status TEXT); CREATE TABLE followed_bands(user_id INTEGER, band_key TEXT, display_name TEXT); CREATE TABLE followed_venues(user_id INTEGER, venue_id INTEGER); CREATE TABLE user_badges(user_id INTEGER REFERENCES users(id) ON DELETE CASCADE, badge_code TEXT, awarded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, trigger_concert_id INTEGER, 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_follow_user_and_show_their_attending_events(self): with main.get_db_connection() as db: db.execute("INSERT INTO venues(name, city) VALUES ('Followed Test Venue', 'Hagen')") db.execute("INSERT INTO followed_venues(user_id, venue_id) VALUES (1, 1)") db.execute("INSERT INTO concerts(artist,start_datetime,visibility,created_by,event_type) " "VALUES ('Followed Friend Band',CURRENT_TIMESTAMP + INTERVAL '3 days','public',2,'concert')") db.execute("INSERT INTO concert_attendance(concert_id,user_id,status) VALUES (1,2,'attending')") result = self.client.post('/users/tester_b/follow', follow_redirects=False) self.assertEqual(result.status_code, 303) self.client.post('/users/tester_b/follow', follow_redirects=False) with main.get_db_connection() as db: self.assertEqual(db.execute('SELECT count(*) FROM followed_users').fetchone()[0], 1) page = self.client.get('/following') self.assertEqual(page.status_code, 200) self.assertIn('tester_b', page.text) self.assertIn('Followed Test Venue', page.text) self.assertIn('/following/venues/remove', page.text) self.assertIn('Followed Friend Band', page.text) self.assertIn('Gehen hin:', page.text) self.client.get('/language/en', follow_redirects=False) english_page = self.client.get('/following') self.assertIn('People', english_page.text) self.assertIn('Going:', english_page.text) with main.get_db_connection() as db: db.execute("UPDATE users SET profile_visibility='friends' WHERE id=2") hidden_attendance = self.client.get('/following') self.assertNotIn('Going: tester_b', hidden_attendance.text) with main.get_db_connection() as db: db.execute("UPDATE users SET profile_visibility='public' WHERE id=2") db.execute("UPDATE concerts SET visibility='private' WHERE id=1") hidden_event = self.client.get('/following') self.assertNotIn('Followed Friend Band', hidden_event.text) def test_follow_privacy_self_unfollow_and_block_cleanup(self): with main.get_db_connection() as db: db.execute("UPDATE users SET profile_visibility='nobody' WHERE username='tester_b'") denied = self.client.post('/users/tester_b/follow', follow_redirects=False) self.assertEqual(denied.status_code, 403) self.assertEqual(self.client.post('/users/tester_a/follow', follow_redirects=False).status_code, 400) with main.get_db_connection() as db: db.execute("UPDATE users SET profile_visibility='public' WHERE username='tester_b'") self.client.post('/users/tester_b/follow', follow_redirects=False) removed = self.client.post('/following/users/remove', data={'followed_id':2}, follow_redirects=False) self.assertEqual(removed.status_code, 303) self.client.post('/users/tester_b/follow', follow_redirects=False) self.client.post('/users/tester_b/block', follow_redirects=False) with main.get_db_connection() as db: self.assertEqual(db.execute('SELECT count(*) FROM followed_users').fetchone()[0], 0) def test_following_page_requires_login(self): self.client.post('/logout', follow_redirects=False) self.assertEqual(self.client.get('/following', follow_redirects=False).status_code, 303) 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', '23_followed_users.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()