335 lines
18 KiB
Python
335 lines
18 KiB
Python
"""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'<a\s', html)), 1)
|
|
self.assertIn('/language/' + target, html)
|
|
self.assertIn('>' + target.upper() + '</a>', 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()
|