"""Local PostgreSQL and simulated Firebase tests. Never contacts Firebase.""" from concurrent.futures import ThreadPoolExecutor from datetime import datetime import hashlib import os from pathlib import Path import unittest from unittest.mock import Mock, patch from uuid import uuid4 from zoneinfo import ZoneInfo import bcrypt import psycopg from psycopg import sql from psycopg.conninfo import make_conninfo from fastapi.testclient import TestClient import main from community_badges import cohort, reconcile from i18n import current_language from notifications import DeliveryError, FirebaseSender, PushWorker, enqueue, save_language class BadgeAndLanguageTests(unittest.TestCase): def test_registration_boundaries_and_timezone(self): with patch.dict(os.environ, {'ALPHA_TESTER_UNTIL':'2026-10-31', 'BETA_TESTER_UNTIL':'2026-12-31'}): for value, expected in ( (datetime(2026, 10, 31, 23, 59, 59), 'alpha_tester'), (datetime(2026, 11, 1), 'beta_tester'), (datetime(2026, 12, 31, 23, 59, 59), 'beta_tester'), (datetime(2027, 1, 1), 'early_bird'), (datetime(2026, 10, 31, 23, tzinfo=ZoneInfo('UTC')), 'beta_tester')): self.assertEqual(cohort(value), expected) def test_configurable_dates_and_invalid_order(self): with patch.dict(os.environ, {'ALPHA_TESTER_UNTIL':'2026-11-30', 'BETA_TESTER_UNTIL':'2027-01-31'}): self.assertEqual(cohort(datetime(2026, 11, 15)), 'alpha_tester') self.assertEqual(cohort(datetime(2027, 1, 1)), 'beta_tester') with patch.dict(os.environ, {'ALPHA_TESTER_UNTIL':'2027-02-01', 'BETA_TESTER_UNTIL':'2027-01-31'}): with self.assertRaises(ValueError): cohort(datetime(2026, 1, 1)) def test_one_language_link_targets_opposite_language(self): import re for language, target in [('de', 'en'), ('en', 'de')]: token = current_language.set(language) try: html = main.templates.get_template('_language_switch.html').render() finally: current_language.reset(token) self.assertEqual(len(re.findall(r'' + target.upper() + '', html) def test_sender_missing_credentials_is_safe(self): with patch.dict(os.environ, {'GOOGLE_APPLICATION_CREDENTIALS':'', 'FIREBASE_PROJECT_ID':''}): with self.assertRaisesRegex(DeliveryError, '^configuration$'): FirebaseSender().send('secret-token', 'title', 'body', {}, 'tag') def test_sdk_payload_errors_and_no_credentials_in_payload(self): from firebase_admin import messaging, exceptions sender = FirebaseSender() sender.app = object() with patch.object(messaging, 'send') as send: sender.send('synthetic-token', 'New message', 'You have a new message.', {'notification_id':str(uuid4()), 'session_tag':'a'*64}, 'tag') payload = send.call_args.args[0] self.assertEqual(payload.android.notification.visibility, 'private') self.assertEqual(payload.android.ttl.total_seconds(), 300) self.assertEqual(payload.notification.body, 'You have a new message.') for failure, expected in ((messaging.UnregisteredError('sensitive'), 'unregistered'), (exceptions.UnavailableError('sensitive'), 'transient'), (exceptions.PermissionDeniedError('sensitive'), 'configuration')): with patch.object(messaging, 'send', side_effect=failure): with self.assertRaisesRegex(DeliveryError, '^' + expected + '$'): sender.send('secret-token', 'title', 'body', {}, 'tag') @unittest.skipUnless(os.environ.get('METALCIRCLE_TEST_DATABASE') == '1', 'explicit local DB opt-in required') class NotificationDatabaseTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.original_dsn = main.DATABASE_URL cls.schema = 'metalcircle_push_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(Path('/test-init/01_initial.sql').read_text()) with patch.object(main, 'INITIAL_ADMIN_USERNAME', None): main.ensure_schema() cls.password_hash = bcrypt.hashpw(b'Push-local-test-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.flags = patch.dict(os.environ, {'PUSH_ENABLED':'true', 'ALPHA_TESTER_UNTIL':'2026-10-31', 'BETA_TESTER_UNTIL':'2026-12-31'}) self.flags.start() 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,concerts RESTART IDENTITY CASCADE') for name in ('sender', 'recipient', 'outsider'): db.execute('INSERT INTO users(username,email,password_hash,created_at) VALUES (%s,%s,%s,%s)', (name, name+'@example.invalid', self.password_hash, datetime(2026, 9, 1))) self.clients = [] for name in ('sender', 'recipient', 'outsider'): client = TestClient(main.app) client.headers['Origin'] = 'http://testserver' self.assertEqual(client.post('/login', data={'username':name, 'password':'Push-local-test-123'}, follow_redirects=False).status_code, 303) self.clients.append(client) self.device = dict(device_id=str(uuid4()), token='synthetic-fcm-token-'+'x'*120, platform='android', session_tag=self.clients[1].get('/api/push/session').json()['session_tag']) self.assertEqual(self.clients[1].post('/api/push/devices', json=self.device).status_code, 200) self.sender = Mock() self.worker = PushWorker(main.get_db_connection, self.sender) def tearDown(self): for client in self.clients: client.close() self.secure.stop() self.flags.stop() def scalar(self, query, args=()): with main.get_db_connection() as db: return db.execute(query, args).fetchone()[0] def friendship(self): result = self.clients[0].post('/users/recipient/friend-request', follow_redirects=False) self.assertEqual(result.status_code, 303) def message(self): with main.get_db_connection() as db: db.execute("INSERT INTO friendships(requester_id,addressee_id,status) VALUES (1,2,'accepted') ON CONFLICT DO NOTHING") result = self.clients[0].post('/messages/recipient', data={'body':'PRIVATE message content'}, follow_redirects=False) self.assertEqual(result.status_code, 303) def invitation(self): result = self.clients[0].post('/concerts', data={'artist':'Private test event', 'start_datetime':'2027-04-01T20:00', 'event_type':'other','visibility':'private','invited_user_ids':'2'}, follow_redirects=False) self.assertEqual(result.status_code, 303) return int(result.headers['location'].rsplit('/', 1)[1]) def test_friend_event_deduplicated_and_english_recipient(self): self.clients[1].get('/language/en', follow_redirects=False) self.friendship() self.friendship() self.assertEqual(self.scalar('SELECT count(*) FROM push_notifications'), 1) self.assertTrue(self.worker.deliver_one()) args = self.sender.send.call_args.args self.assertEqual(args[1], 'New friend request') self.assertNotIn('PRIVATE', str(args)) target = self.clients[1].get('/notifications/'+args[3]['notification_id'], follow_redirects=False) self.assertEqual(target.headers['location'], '/users/sender') denied = self.clients[2].get('/notifications/'+args[3]['notification_id'], follow_redirects=False) self.assertEqual(denied.headers['location'], '/') def test_message_route_and_private_content(self): self.message() self.worker.deliver_one() args = self.sender.send.call_args.args self.assertEqual(args[1:3], ('Neue Nachricht', 'Du hast eine neue Nachricht.')) self.assertNotIn('PRIVATE message content', str(args)) response = self.clients[1].get('/notifications/'+args[3]['notification_id'], follow_redirects=False) self.assertEqual(response.headers['location'], '/messages/sender#latest') def test_invitation_route_and_removed_invitation(self): concert = self.invitation() self.worker.deliver_one() args = self.sender.send.call_args.args response = self.clients[1].get('/notifications/'+args[3]['notification_id'], follow_redirects=False) self.assertEqual(response.headers['location'], '/concerts/'+str(concert)) with main.get_db_connection() as db: db.execute('DELETE FROM event_invitations') response = self.clients[1].get('/notifications/'+args[3]['notification_id'], follow_redirects=False) self.assertEqual(response.headers['location'], '/') def test_disabled_sender_has_no_backlog(self): with patch.dict(os.environ, {'PUSH_ENABLED':'false'}): self.friendship() self.assertEqual(self.scalar('SELECT count(*) FROM push_notifications'), 0) def test_invitation_edit_only_notifies_new_invitees(self): concert = self.invitation() form = {'artist':'Private test event','start_datetime':'2027-04-01T20:00', 'event_type':'other','visibility':'private','invited_user_ids':'2'} result = self.clients[0].post(f'/concerts/{concert}/edit', data=form, follow_redirects=False) self.assertEqual(result.status_code, 303) self.assertEqual(self.scalar("SELECT count(*) FROM push_notifications WHERE state='pending'"), 1) form.pop('invited_user_ids') self.clients[0].post(f'/concerts/{concert}/edit', data=form, follow_redirects=False) form['invited_user_ids'] = '2' self.clients[0].post(f'/concerts/{concert}/edit', data=form, follow_redirects=False) self.assertEqual(self.scalar("SELECT count(*) FROM push_notifications WHERE state='pending'"), 1) def test_profile_preferences_and_alpha_render_in_both_languages(self): for language, label in [('de', 'Push-Benachrichtigungen'), ('en', 'Push notifications')]: self.clients[1].get('/language/'+language, follow_redirects=False) response = self.clients[1].get('/profile') self.assertEqual(response.status_code, 200) self.assertIn(label, response.text) self.assertIn('patch-alpha-tester.svg', response.text) self.assertNotIn('id="patch-beta_tester"', response.text) exported = self.clients[1].get('/profile/export') self.assertEqual(exported.status_code, 200) self.assertEqual(exported.json()['notification_preferences']['language'], language) def test_enqueue_rolls_back_with_domain_transaction(self): try: with main.get_db_connection() as db: request_id = db.execute('INSERT INTO friendships(requester_id,addressee_id) VALUES (1,2) RETURNING id').fetchone()[0] enqueue(db.cursor(), 'friend_request', 1, 2, request_id) raise RuntimeError('simulated rollback') except RuntimeError: pass self.assertEqual(self.scalar('SELECT count(*) FROM push_notifications'), 0) self.assertEqual(self.scalar('SELECT count(*) FROM friendships'), 0) def test_preference_opt_out_and_csrf_auth(self): unauth = TestClient(main.app) self.assertEqual(unauth.post('/profile/notifications', follow_redirects=False).status_code, 303) self.assertEqual(self.clients[1].post('/profile/notifications', headers={'Origin':'http://evil.invalid'}).status_code, 403) self.friendship() result = self.clients[1].post('/profile/notifications', data={'direct_message':'true'}, follow_redirects=False) self.assertEqual(result.status_code, 303) self.assertEqual(self.scalar("SELECT count(*) FROM push_notifications WHERE state='dropped'"), 1) self.assertFalse(self.worker.deliver_one()) self.sender.send.assert_not_called() def test_disabled_category_never_enqueues(self): self.clients[1].post('/profile/notifications', data={}, follow_redirects=False) self.friendship() self.assertEqual(self.scalar('SELECT count(*) FROM push_notifications'), 0) def test_logout_cancels_queue_and_switch_does_not_receive_old_push(self): self.friendship() self.clients[1].post('/logout', follow_redirects=False) self.assertEqual(self.scalar('SELECT count(*) FROM push_notifications'), 0) self.assertFalse(self.worker.deliver_one()) self.sender.send.assert_not_called() def test_token_rotation_invalidates_old_delivery(self): self.friendship() self.device['token'] = 'synthetic-fcm-token-'+'y'*120 self.clients[1].post('/api/push/devices', json=self.device) self.worker.deliver_one() self.sender.send.assert_not_called() self.assertEqual(self.scalar('SELECT state FROM push_notifications'), 'dropped') def test_multiple_devices_each_receive_once(self): another = dict(self.device, device_id=str(uuid4()), token='synthetic-fcm-token-'+'z'*120) self.clients[1].post('/api/push/devices', json=another) self.friendship() self.worker.deliver_one() self.worker.deliver_one() self.assertEqual(self.sender.send.call_count, 2) self.assertFalse(self.worker.deliver_one()) def test_read_message_block_and_removed_friendship_drop_pending(self): for operation in ('read', 'block', 'unfriend'): with self.subTest(operation=operation): self.message() with main.get_db_connection() as db: if operation == 'read': db.execute('UPDATE direct_messages SET read_at=CURRENT_TIMESTAMP') elif operation == 'block': db.execute('INSERT INTO user_blocks(blocker_id,blocked_id) VALUES (2,1)') else: db.execute('DELETE FROM friendships') self.worker.deliver_one() self.sender.send.assert_not_called() with main.get_db_connection() as db: db.execute('DELETE FROM user_blocks') def test_retry_limit_and_safe_logging(self): self.friendship() self.sender.send.side_effect = DeliveryError('transient') with self.assertLogs('notifications', 'WARNING') as logs: for attempt in range(4): self.worker.deliver_one() with main.get_db_connection() as db: db.execute('UPDATE push_notifications SET available_at=CURRENT_TIMESTAMP') self.assertNotIn(self.device['token'], str(logs.output)) self.assertEqual(self.scalar('SELECT attempts FROM push_notifications'), 4) self.assertEqual(self.scalar('SELECT state FROM push_notifications'), 'failed') self.assertEqual(self.clients[1].get('/impressum').status_code, 200) def test_unregistered_removes_device_but_configuration_error_does_not(self): self.friendship() self.sender.send.side_effect = DeliveryError('configuration') self.worker.deliver_one() self.assertEqual(self.scalar('SELECT count(*) FROM push_devices'), 1) with main.get_db_connection() as db: db.execute('UPDATE push_notifications SET available_at=CURRENT_TIMESTAMP') self.sender.send.side_effect = DeliveryError('unregistered') self.worker.deliver_one() self.assertEqual(self.scalar('SELECT count(*) FROM push_devices'), 0) def test_expired_job_and_expired_session_are_dropped(self): self.friendship() with main.get_db_connection() as db: db.execute("UPDATE push_notifications SET expires_at=CURRENT_TIMESTAMP-INTERVAL '1 second'") self.worker.deliver_one() self.sender.send.assert_not_called() with main.get_db_connection() as db: db.execute("UPDATE push_notifications SET expires_at=CURRENT_TIMESTAMP+INTERVAL '1 hour',state='pending'") db.execute("UPDATE sessions SET expires_at=CURRENT_TIMESTAMP-INTERVAL '1 second' WHERE user_id=2") self.worker.deliver_one() self.sender.send.assert_not_called() def test_two_workers_do_not_send_same_job_twice(self): self.friendship() workers = [PushWorker(main.get_db_connection, self.sender) for _ in range(2)] with ThreadPoolExecutor(max_workers=2) as pool: list(pool.map(lambda worker: worker.deliver_one(), workers)) self.assertEqual(self.sender.send.call_count, 1) def test_backfill_is_exclusive_idempotent_and_reconfigurable(self): with main.get_db_connection() as db: db.execute("INSERT INTO user_badges(user_id,badge_code) VALUES (1,'beta_tester')") db.execute("UPDATE users SET created_at='2026-11-01' WHERE id=2") db.execute("UPDATE users SET created_at='2027-01-01' WHERE id=3") reconcile(db.cursor()) reconcile(db.cursor()) rows = db.execute('SELECT user_id,badge_code FROM user_badges ORDER BY user_id').fetchall() self.assertEqual(rows, [(1,'alpha_tester'),(2,'beta_tester'),(3,'early_bird')]) with patch.dict(os.environ, {'ALPHA_TESTER_UNTIL':'2026-11-30'}): reconcile(db.cursor()) self.assertEqual(db.execute('SELECT badge_code FROM user_badges WHERE user_id=2').fetchone()[0], 'alpha_tester') def test_db_utc_registration_respects_berlin_midnight(self): with main.get_db_connection() as db: db.execute("SET LOCAL TIME ZONE 'UTC'") db.execute("UPDATE users SET created_at='2026-10-31 22:59:59' WHERE id=1") db.execute("UPDATE users SET created_at='2026-10-31 23:00:00' WHERE id=2") db.execute("UPDATE users SET created_at='2026-12-31 23:00:00' WHERE id=3") reconcile(db.cursor()) rows = db.execute('SELECT user_id,badge_code FROM user_badges ORDER BY user_id').fetchall() self.assertEqual(rows, [(1,'alpha_tester'),(2,'beta_tester'),(3,'early_bird')]) if __name__ == '__main__': unittest.main()