"""Local PostgreSQL and simulated Firebase tests. Never contacts Firebase."""
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
import hashlib
import os
from pathlib import Path
import re
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_language_dropdown_offers_both_languages_and_marks_current(self):
        for language in ('de', 'en'):
            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)), 2)
            for target in ('de', 'en'):
                self.assertIn('/language/' + target, html)
            self.assertRegex(html, rf'<a[^>]*lang="{language}"[^>]*aria-current="true"')
            self.assertEqual(html.count('aria-current="true"'), 1)

    def test_all_push_kinds_share_generic_android_payload_and_unique_tags(self):
        from firebase_admin import messaging
        from notifications import TEXT
        sender = FirebaseSender()
        sender.app = object()
        tags = set()
        for kind, (title, body) in TEXT.items():
            with self.subTest(kind=kind), patch.object(messaging, 'send') as send:
                identifier = str(uuid4())
                sender.send('synthetic-token', title, body,
                            {'notification_id': identifier, 'session_tag': 'a' * 64}, identifier)
                payload = send.call_args.args[0]
                self.assertEqual(payload.notification.title, title)
                self.assertEqual(payload.notification.body, body)
                self.assertEqual(set(payload.data), {'notification_id', 'session_tag'})
                self.assertEqual(payload.android.notification.tag, identifier)
                self.assertIsNone(payload.android.notification.channel_id)
                self.assertIsNone(payload.android.notification.click_action)
                self.assertEqual(payload.android.priority, 'high')
                self.assertEqual(payload.android.notification.sound, 'default')
                self.assertEqual(payload.android.notification.icon, 'ic_notification')
                tags.add(identifier)
        self.assertEqual(len(tags), 3)

    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):
        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('/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_opening_chat_before_delivery_drops_push_and_read_tap_returns_home(self):
        self.message()
        self.assertEqual(self.clients[1].get('/messages/sender').status_code, 200)
        self.worker.deliver_one()
        self.sender.send.assert_not_called()
        self.assertEqual(self.scalar('SELECT state FROM push_notifications'), 'dropped')
        self.message()
        self.worker.deliver_one()
        self.sender.send.assert_called_once()
        identifier = self.sender.send.call_args.args[3]['notification_id']
        self.clients[1].get('/messages/sender')
        response = self.clients[1].get('/notifications/' + identifier, follow_redirects=False)
        self.assertEqual(response.headers['location'], '/')

    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 event(self, owner=1, past=False, kind='other', visibility='public'):
        start = datetime.now() + timedelta(days=-10 if past else 10)
        with main.get_db_connection() as db:
            return db.execute("""INSERT INTO concerts
                (artist,start_datetime,end_datetime,event_type,visibility,created_by)
                VALUES (%s,%s,%s,%s,%s,%s) RETURNING id""",
                ('Event access test', start, start + timedelta(days=1) if kind == 'festival' else None,
                 kind, visibility, owner)).fetchone()[0]

    def test_creator_can_delete_own_events_including_past_in_both_languages(self):
        for language in ('de', 'en'):
            self.clients[0].get('/language/' + language, follow_redirects=False)
            for past in (False, True):
                for kind in ('concert', 'festival', 'other'):
                    with self.subTest(language=language, past=past, kind=kind):
                        event = self.event(past=past, kind=kind)
                        page = self.clients[0].get(f'/concerts/{event}')
                        self.assertEqual(page.status_code, 200)
                        self.assertIn(f'action="/concerts/{event}/delete"', page.text)
                        self.assertIn('Delete event' if language == 'en' else 'Veranstaltung löschen', page.text)
                        if past:
                            self.assertNotIn(f'href="/concerts/{event}/edit"', page.text)
                        result = self.clients[0].post(f'/concerts/{event}/delete', follow_redirects=False)
                        self.assertEqual(result.status_code, 303)
                        self.assertEqual(self.scalar('SELECT count(*) FROM concerts WHERE id=%s', (event,)), 0)

    def test_delete_rejects_foreign_and_anonymous_but_preserves_admin_access(self):
        for past in (False, True):
            event = self.event(owner=2, past=past)
            page = self.clients[0].get(f'/concerts/{event}')
            self.assertNotIn(f'action="/concerts/{event}/delete"', page.text)
            self.assertEqual(self.clients[0].post(f'/concerts/{event}/delete', follow_redirects=False).status_code, 403)
            self.assertEqual(self.scalar('SELECT count(*) FROM concerts WHERE id=%s', (event,)), 1)
        with TestClient(main.app) as anonymous:
            self.assertEqual(anonymous.post(f'/concerts/{event}/delete', follow_redirects=False).status_code, 303)
        self.assertEqual(self.clients[1].post(f'/concerts/{event}/delete',
            headers={'Origin':'https://evil.invalid'}, follow_redirects=False).status_code, 403)
        with main.get_db_connection() as db:
            db.execute('UPDATE users SET is_admin=TRUE WHERE id=3')
        self.assertIn(f'action="/concerts/{event}/delete"', self.clients[2].get(f'/concerts/{event}').text)
        self.assertEqual(self.clients[2].post(f'/concerts/{event}/delete', follow_redirects=False).status_code, 303)
        # Deleting a parent does not delete another user's linked event.
        parent = self.event(kind='concert')
        child = self.event(owner=2)
        with main.get_db_connection() as db:
            db.execute('UPDATE concerts SET parent_event_id=%s WHERE id=%s', (parent, child))
        self.assertEqual(self.clients[0].post(f'/concerts/{parent}/delete', follow_redirects=False).status_code, 303)
        self.assertIsNone(self.scalar('SELECT parent_event_id FROM concerts WHERE id=%s', (child,)))

    def test_private_picker_shows_only_confirmed_unblocked_owner_friends(self):
        event = self.event(visibility='private')
        with main.get_db_connection() as db:
            db.execute("INSERT INTO friendships(requester_id,addressee_id,status) VALUES (2,1,'accepted'),(1,3,'pending')")
        for language, heading in (('de','Freunde einladen'), ('en','Invite friends')):
            self.clients[0].get('/language/' + language, follow_redirects=False)
            for route in ('/concerts/new', f'/concerts/{event}/edit'):
                page = self.clients[0].get(route)
                self.assertEqual(page.status_code, 200)
                field = re.search(r'<fieldset id="event-invitations-field".*?</fieldset>', page.text, re.S).group()
                self.assertIn(heading, field)
                self.assertEqual(re.findall(r'name="invited_user_ids" value="(\d+)"', field), ['2'])
        with main.get_db_connection() as db:
            db.execute('UPDATE users SET is_admin=TRUE WHERE id=3')
        admin_page = self.clients[2].get(f'/concerts/{event}/edit')
        self.assertEqual(re.findall(r'name="invited_user_ids" value="(\d+)"', admin_page.text), ['2'])
        # Verify the other direction of the friendship and a block overriding it.
        self.assertEqual([u['id'] for u in main.get_invitable_users(2)], [1])
        with main.get_db_connection() as db:
            db.execute('INSERT INTO user_blocks(blocker_id,blocked_id) VALUES (2,1)')
        self.assertEqual(main.get_invitable_users(1), [])
        for language, empty in (('de','Noch keine bestätigten Freunde vorhanden.'), ('en','No confirmed friends yet.')):
            self.clients[0].get('/language/' + language, follow_redirects=False)
            self.assertIn(empty, self.clients[0].get('/concerts/new').text)

    def test_private_invites_reject_forged_ids_and_preserve_existing_hidden_invites(self):
        form = {'artist':'Invitation access test', 'start_datetime':'2027-04-01T20:00',
                'event_type':'other', 'visibility':'private'}
        for language, error in (('de','Es können nur bestätigte Freunde eingeladen werden.'),
                                ('en','Only confirmed friends can be invited.')):
            self.clients[0].get('/language/' + language, follow_redirects=False)
            for selected in ('1','2','99999'):
                result = self.clients[0].post('/concerts', data=dict(form, invited_user_ids=selected), follow_redirects=False)
                self.assertEqual(result.status_code, 400)
                self.assertIn(error, result.text)
        self.assertEqual(self.scalar('SELECT count(*) FROM concerts'), 0)
        event = self.invitation()
        result = self.clients[0].post(f'/concerts/{event}/edit',
            data=dict(form, invited_user_ids=['2','3']), follow_redirects=False)
        self.assertEqual(result.status_code, 400)
        self.assertEqual(main.get_event_invitee_ids(event), {2})
        self.assertEqual(self.scalar('SELECT artist FROM concerts WHERE id=%s', (event,)), 'Private test event')
        with main.get_db_connection() as db:
            db.execute('DELETE FROM friendships')
        # The now hidden legacy invitation survives an unrelated edit, without a new push.
        result = self.clients[0].post(f'/concerts/{event}/edit', data=form, follow_redirects=False)
        self.assertEqual(result.status_code, 303)
        self.assertEqual(main.get_event_invitee_ids(event), {2})
        self.assertEqual(self.scalar("SELECT count(*) FROM push_notifications WHERE kind='event_invitation'"), 1)
        # An invited non-owner cannot use the edit route to grant access to someone else.
        result = self.clients[1].post(f'/concerts/{event}/edit',
            data=dict(form, invited_user_ids='3'), follow_redirects=False)
        self.assertEqual(result.status_code, 303)
        self.assertEqual(main.get_event_invitee_ids(event), {2})

    def test_linkable_events_are_chronological_in_create_and_edit_forms(self):
        with main.get_db_connection() as db:
            rows = db.execute("""
                INSERT INTO concerts(artist, start_datetime, end_datetime, event_type, visibility, created_by)
                VALUES
                    ('Active festival', date_trunc('day', CURRENT_TIMESTAMP) - INTERVAL '1 day' + INTERVAL '18 hours', date_trunc('day', CURRENT_TIMESTAMP) + INTERVAL '23 hours 59 minutes', 'festival', 'public', 1),
                    ('Near concert', date_trunc('day', CURRENT_TIMESTAMP) + INTERVAL '2 days 20 hours', NULL, 'concert', 'public', 1),
                    ('Same-time concert', date_trunc('day', CURRENT_TIMESTAMP) + INTERVAL '2 days 20 hours', NULL, 'concert', 'public', 1),
                    ('Upcoming festival', date_trunc('day', CURRENT_TIMESTAMP) + INTERVAL '3 days 10 hours', date_trunc('day', CURRENT_TIMESTAMP) + INTERVAL '4 days 23 hours', 'festival', 'public', 1),
                    ('Later concert', date_trunc('day', CURRENT_TIMESTAMP) + INTERVAL '6 days 20 hours', NULL, 'concert', 'public', 1),
                    ('Past concert', date_trunc('day', CURRENT_TIMESTAMP) - INTERVAL '1 day', NULL, 'concert', 'public', 1),
                    ('Finished festival', date_trunc('day', CURRENT_TIMESTAMP) - INTERVAL '3 days', date_trunc('day', CURRENT_TIMESTAMP) - INTERVAL '1 day', 'festival', 'public', 1)
                RETURNING id
            """).fetchall()
            active_festival, near_concert, same_time_concert, upcoming_festival, later_concert, _, _ = [row[0] for row in rows]
            edit_event_id = db.execute("""
                INSERT INTO concerts(artist, start_datetime, event_type, visibility, created_by)
                VALUES ('Test pre-show', date_trunc('day', CURRENT_TIMESTAMP) + INTERVAL '5 days', 'other', 'public', 1)
                RETURNING id
            """).fetchone()[0]

        expected_ids = [active_festival, near_concert, same_time_concert, upcoming_festival, later_concert]
        for language in ('de', 'en'):
            self.clients[0].get('/language/' + language, follow_redirects=False)
            create_page = self.clients[0].get('/concerts/new')
            self.assertEqual(create_page.status_code, 200)
            create_select = re.search(r'<select name="parent_event_id".*?</select>', create_page.text, re.S).group()
            create_ids = [int(value) for value in re.findall(r'<option value="(\d+)"', create_select)]
            self.assertEqual(create_ids, expected_ids)

            edit_page = self.clients[0].get(f'/concerts/{edit_event_id}/edit')
            self.assertEqual(edit_page.status_code, 200)
            edit_select = re.search(r'<select name="parent_event_id".*?</select>', edit_page.text, re.S).group()
            edit_ids = [int(value) for value in re.findall(r'<option value="(\d+)"', edit_select)]
            self.assertEqual(edit_ids, expected_ids)

        submitted = self.clients[0].post('/concerts', data={
            'artist': 'Linked pre-show test',
            'event_type': 'other',
            'parent_event_id': str(upcoming_festival),
            'visibility': 'public',
            'start_datetime': (datetime.now() + timedelta(days=2)).strftime('%Y-%m-%dT%H:%M'),
        }, follow_redirects=False)
        self.assertEqual(submitted.status_code, 303)
        linked_id = int(submitted.headers['location'].rsplit('/', 1)[1])
        self.assertEqual(self.scalar('SELECT parent_event_id FROM concerts WHERE id=%s', (linked_id,)), upcoming_festival)

        self.clients[0].post(f'/concerts/{edit_event_id}/edit', data={
            'artist': 'Test pre-show',
            'event_type': 'other',
            'parent_event_id': str(near_concert),
            'visibility': 'public',
            'start_datetime': (datetime.now() + timedelta(days=5)).strftime('%Y-%m-%dT%H:%M'),
        }, follow_redirects=False)
        self.assertEqual(self.scalar('SELECT parent_event_id FROM concerts WHERE id=%s', (edit_event_id,)), near_concert)

    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()