import asyncio
import json
import re
import smtplib
import socketserver
import threading
import time
import unittest
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
from email import policy
from email.parser import BytesParser
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
from urllib.parse import parse_qs, urlsplit

from fastapi import FastAPI
from fastapi.testclient import TestClient

from backend.app import db
from backend.app.auth import get_current_user
from backend.app.clients.jellystat import HistoryLimitError, JellystatError
from backend.app.routers import recaps as router
from backend.app.services import email_recaps as recaps, recap_email as mail, recap_store as store
from backend.app.services.jellyfin_identity import link_user, source_key
from backend.app.services.monthly_reports import change, month_periods, shift_month
from backend.tests.test_backend_quality import TempDatabaseMixin


def fixture_report():
    periods = month_periods(None, datetime.now(timezone.utc))
    summary = dict(minutes=1500, movies=8, episodes=24, plays=35, active_days=20, longest_streak=6)
    changes = {key: change(value, round(value / 2)) for key, value in summary.items()}
    changes['requests'] = change(3, 2)
    return {**periods, 'state': 'ready', 'summary': summary, 'changes': changes, 'requests': {'total': 3},
            'top_titles': [{'title': 'Severance', 'type': 'series', 'minutes': 460, 'plays': 10},
                           {'title': 'Arrival', 'type': 'movie', 'minutes': 116, 'plays': 1}],
            'recent': [{'artwork_url': '/insights/artwork/SECRET?token=PRIVATE-TOKEN'}]}


def runtime():
    return SimpleNamespace(jellyfin_base_url='http://jellyfin', jellystat_base_url='http://jellystat',
        jellystat_api_key='PRIVATE-STATS-KEY', magent_notify_enabled=True, magent_notify_email_enabled=True,
        magent_notify_email_smtp_host='127.0.0.1', magent_notify_email_smtp_port=1,
        magent_notify_email_smtp_username='', magent_notify_email_smtp_password='',
        magent_notify_email_from_address='magent@example.test', magent_notify_email_from_name='Magent',
        magent_notify_email_use_tls=False, magent_notify_email_use_ssl=False)


class RecapFixture(TempDatabaseMixin):
    def setUp(self):
        super().setUp()
        db.create_user('viewer', 'Example-Password123!', role='admin', email='viewer@example.test')
        link_user('viewer', 'jf-viewer', 'http://jellyfin')
        self.user = db.get_user_by_username('viewer')
        self.runtime = runtime()
        for target, name, value in [(recaps, 'get_runtime_settings', self.runtime), (mail, 'get_runtime_settings', self.runtime),
                                     (recaps, 'smtp_email_config_ready', (True, 'ok'))]:
            mocked = patch.object(target, name, return_value=value)
            mocked.start(); self.addCleanup(mocked.stop)
        env = patch.dict('os.environ', {'BACKGROUND_TASKS_ENABLED': 'true'})
        env.start(); self.addCleanup(env.stop)
        self.config = dict(enabled=False, day=2, hour=9, public_url='https://beta.example.test')
        store.save_settings(self.config, datetime.now(timezone.utc))
        self.report = fixture_report()

    def subscribe(self, timestamp=None):
        now = time.time() if timestamp is None else timestamp
        token = store.request_confirmation(self.user, source_key('http://jellyfin'), 'jf-viewer', now)
        sub = store.subscription(self.user['id'])
        self.assertTrue(store.confirm(sub, now + 1))
        return store.subscription(self.user['id']), token

    def queue(self, sub=None, request_id='request-1'):
        if sub is None:
            sub, _ = self.subscribe()
        return store.enqueue_test(sub, self.report['month'], request_id, self.config['public_url'], time.time())

    def delivery(self, delivery_id):
        return store.read_one('SELECT * FROM email_recap_deliveries WHERE id=?', (delivery_id,))


class RecapConsentTests(RecapFixture, unittest.IsolatedAsyncioTestCase):
    async def test_opt_in_only_emails_confirmation_and_check_link_does_not_confirm(self):
        with patch.object(mail, 'send_email') as sender, patch.object(recaps, 'get_monthly_report') as report:
            result = await recaps.subscribe(self.user)
        self.assertEqual(result['state'], 'pending')
        report.assert_not_called()
        recipient, rendered, _ = sender.call_args.args
        self.assertEqual(recipient, 'viewer@example.test')
        self.assertNotIn('Severance', rendered['body_html'])
        url = re.search(r'https://[^\s]+', rendered['body_text']).group(0)
        token = parse_qs(urlsplit(url).fragment)['token'][0]
        self.assertNotIn(token, store.subscription(self.user['id'])['confirmation_hash'])
        self.assertEqual(recaps.token_action(token, 'confirm')['state'], 'ready')
        self.assertEqual(store.subscription(self.user['id'])['state'], 'pending')
        self.assertEqual(recaps.token_action(token, 'confirm', apply=True)['state'], 'enabled')
        with self.assertRaises(recaps.RecapError):
            recaps.token_action(token, 'confirm', apply=True)
        with self.assertRaises(recaps.RecapError):
            recaps.token_action(token, 'unsubscribe', apply=True)

    async def test_confirmation_failure_is_pending_and_resend_is_rate_limited(self):
        with patch.object(mail, 'send_email', side_effect=mail.DeliveryError('unknown', 'unknown')):
            with self.assertRaises(recaps.RecapError) as exc:
                await recaps.subscribe(self.user)
        self.assertEqual(exc.exception.status, 502)
        self.assertEqual(recaps.preferences(self.user)['state'], 'pending')
        with patch.object(mail, 'send_email') as sender:
            with self.assertRaises(recaps.RecapError) as exc:
                await recaps.subscribe(self.user)
        self.assertEqual(exc.exception.status, 429)
        sender.assert_not_called()

    def test_unsubscribe_is_public_idempotent_and_cancels_queued_email(self):
        sub, _ = self.subscribe()
        delivery_id = self.queue(sub)
        token = sub['unsubscribe_token']
        self.assertEqual(recaps.token_action(token, 'unsubscribe')['state'], 'ready')
        self.assertEqual(self.delivery(delivery_id)['state'], 'queued')
        recaps.token_action(token, 'unsubscribe', apply=True)
        self.assertEqual(recaps.token_action(token, 'unsubscribe', apply=True)['state'], 'off')
        self.assertEqual(self.delivery(delivery_id)['state'], 'cancelled')

    def test_expired_confirmation_does_not_subscribe(self):
        token = store.request_confirmation(self.user, source_key('http://jellyfin'), 'jf-viewer', time.time() - 90000)
        self.assertEqual(recaps.preferences(self.user)['state'], 'expired')
        with self.assertRaises(recaps.RecapError):
            recaps.token_action(token, 'confirm', apply=True)

    def test_email_change_back_does_not_restore_consent(self):
        self.subscribe()
        db.set_user_email('viewer', 'changed@example.test')
        db.set_user_email('viewer', 'viewer@example.test')
        self.assertEqual(recaps.preferences(self.user)['state'], 'off')

    def test_changed_link_or_source_requires_new_consent(self):
        self.subscribe()
        with store.transaction() as conn:
            conn.execute("UPDATE jellyfin_user_links SET jellyfin_user_id='new-identity' WHERE local_user_id=?", (self.user['id'],))
        self.assertEqual(recaps.preferences(self.user)['state'], 'off')
        with store.transaction() as conn:
            conn.execute("UPDATE email_recap_subscriptions SET state='enabled'")
        self.runtime.jellyfin_base_url = 'http://other-jellyfin'
        self.assertEqual(recaps.preferences(self.user)['state'], 'off')

    def test_missing_email_or_stored_identity_cannot_subscribe(self):
        db.set_user_email('viewer', None)
        self.assertFalse(recaps.preferences(self.user)['can_subscribe'])
        db.set_user_email('viewer', 'viewer@example.test')
        with store.transaction() as conn:
            conn.execute('DELETE FROM jellyfin_user_links')
        self.assertFalse(recaps.preferences(self.user)['can_subscribe'])

    def test_confirmation_rechecks_email_atomically(self):
        store.request_confirmation(self.user, source_key('http://jellyfin'), 'jf-viewer', time.time())
        old = store.subscription(self.user['id'])
        db.set_user_email('viewer', 'different@example.test')
        self.assertFalse(store.confirm(old, time.time()))


class RecapScheduleTests(RecapFixture, unittest.TestCase):
    def test_defaults_are_paused_and_no_users_are_opted_in(self):
        self.assertFalse(store.settings()['enabled'])
        self.assertEqual(store.history()['subscribers'], 0)
        self.assertEqual(store.enqueue_due(datetime.now(timezone.utc)), 0)

    def test_utc_next_send_month_end_leap_year_and_new_year(self):
        for now, expected in [
            (datetime(2026, 12, 31, tzinfo=timezone.utc), '2027-01-02T09:00:00+00:00'),
            (datetime(2024, 2, 29, tzinfo=timezone.utc), '2024-03-02T09:00:00+00:00'),
            (datetime(2026, 9, 2, 8, tzinfo=timezone.utc), '2026-09-02T09:00:00+00:00'),
            (datetime(2026, 9, 2, 9, tzinfo=timezone.utc), '2026-10-02T09:00:00+00:00')]:
            self.assertEqual(store.next_due(now, 2, 9).isoformat(), expected)

    def test_schedule_catches_up_once_and_excludes_late_subscribers(self):
        before = datetime(2026, 8, 30, tzinfo=timezone.utc)
        self.subscribe(before.timestamp())
        config = store.save_settings({**self.config, 'enabled': True}, before)
        self.assertEqual(config['next_send_at'], datetime(2026, 9, 2, 9, tzinfo=timezone.utc).timestamp())
        db.create_user('late', 'Example-Password123!', email='late@example.test')
        late = db.get_user_by_username('late')
        store.request_confirmation(late, 'source', 'late-id', datetime(2026, 9, 2, 10, tzinfo=timezone.utc).timestamp())
        store.confirm(store.subscription(late['id']), datetime(2026, 9, 2, 11, tzinfo=timezone.utc).timestamp())
        now = datetime(2026, 9, 5, tzinfo=timezone.utc)
        with ThreadPoolExecutor(max_workers=4) as pool:
            counts = list(pool.map(store.enqueue_due, [now] * 4))
        self.assertEqual(sum(counts), 1)
        rows = store.history()['deliveries']
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]['month'], '2026-08')
        self.assertEqual(rows[0]['email'], 'viewer@example.test')
        # Revisit the same due date after a restart: the durable unique key still wins.
        with store.transaction() as conn:
            conn.execute('UPDATE email_recap_settings SET next_send_at=?', (config['next_send_at'],))
        self.assertEqual(store.enqueue_due(now), 0)

    def test_long_downtime_does_not_backfill_multiple_months(self):
        before = datetime(2026, 5, 1, tzinfo=timezone.utc)
        self.subscribe(before.timestamp())
        store.save_settings({**self.config, 'enabled': True}, before)
        self.assertEqual(store.enqueue_due(datetime(2026, 9, 9, tzinfo=timezone.utc)), 1)
        self.assertEqual(store.history()['deliveries'][0]['month'], '2026-08')

    def test_enable_after_due_date_waits_and_pause_cancels_pending_monthlies(self):
        now = datetime(2026, 9, 9, tzinfo=timezone.utc)
        self.subscribe(now.timestamp())
        result = store.save_settings({**self.config, 'enabled': True}, now)
        self.assertEqual(result['next_send_at'], datetime(2026, 10, 2, 9, tzinfo=timezone.utc).timestamp())
        self.assertEqual(store.enqueue_due(now), 0)
        store.enqueue_due(datetime(2026, 10, 3, tzinfo=timezone.utc))
        store.save_settings(self.config, now)
        self.assertEqual(store.history()['deliveries'][0]['state'], 'cancelled')
        self.assertIsNone(store.settings()['next_send_at'])


class RecapDeliveryTests(RecapFixture, unittest.IsolatedAsyncioTestCase):
    async def run_claim(self):
        delivery = store.claim_delivery(time.time())
        self.assertIsNotNone(delivery)
        await recaps.process_delivery(delivery)

    async def test_private_report_is_delivered_once_using_confirmed_account(self):
        delivery_id = self.queue()
        sent = []
        def capture(recipient, rendered, message_id, before_data):
            before_data()
            self.assertEqual(self.delivery(delivery_id)['state'], 'sending')
            sent.append((recipient, rendered, message_id))
        with patch.object(recaps, 'get_monthly_report', new=AsyncMock(return_value=self.report)) as report, patch.object(mail, 'send_email', side_effect=capture):
            await recaps.run_once()
            await recaps.run_once()
        self.assertEqual(len(sent), 1)
        self.assertEqual(sent[0][0], 'viewer@example.test')
        self.assertIn(f'?month={self.report["month"]}', sent[0][1]['body_html'])
        self.assertNotIn('PRIVATE-TOKEN', json.dumps(sent))
        self.assertEqual(report.await_args.args[0]['id'], self.user['id'])
        self.assertEqual(self.delivery(delivery_id)['state'], 'sent')
        self.assertNotIn('unsubscribe_token', json.dumps(store.history()))

    def test_concurrent_claim_and_test_deduplication(self):
        sub, _ = self.subscribe()
        with ThreadPoolExecutor(max_workers=4) as pool:
            ids = list(pool.map(lambda _: self.queue(sub), range(4)))
            rows = list(pool.map(lambda _: store.claim_delivery(time.time()), range(4)))
        self.assertEqual(len(set(ids)), 1)
        self.assertEqual(sum(row is not None for row in rows), 1)
        with self.assertRaises(ValueError):
            self.queue(sub, 'another-click')

    async def test_unsubscribe_or_email_change_during_report_prevents_sending(self):
        delivery_id = self.queue()
        async def report(*args):
            db.set_user_email('viewer', 'other@example.test')
            return self.report
        def transport(recipient, rendered, message_id, before_data):
            before_data()
            self.fail('Private data must not reach SMTP DATA after an address change')
        with patch.object(recaps, 'get_monthly_report', side_effect=report), patch.object(mail, 'send_email', side_effect=transport):
            await self.run_claim()
        self.assertEqual(self.delivery(delivery_id)['state'], 'cancelled')

    async def test_stats_permission_revoked_during_report_cancels_email(self):
        from backend.app.feature_access import update_permissions
        delivery_id = self.queue()
        db.set_user_role('viewer', 'user')
        async def report(*args):
            update_permissions({'stats': False}, 'viewer')
            return self.report
        def transport(recipient, rendered, message_id, before_data):
            before_data()
            self.fail('Report must not be sent after stats permission is revoked')
        with patch.object(recaps, 'get_monthly_report', side_effect=report), patch.object(mail, 'send_email', side_effect=transport):
            await self.run_claim()
        self.assertEqual(self.delivery(delivery_id)['state'], 'cancelled')

    async def test_blocked_expired_and_deleted_accounts_are_not_sent(self):
        for kind in ['blocked', 'expired', 'deleted']:
            with self.subTest(kind=kind):
                # Each subcase starts with a fresh account and confirmed subscription.
                db.create_user(kind, 'Example-Password123!', email=f'{kind}@example.test')
                account = db.get_user_by_username(kind)
                link_user(kind, f'jf-{kind}', 'http://jellyfin')
                store.request_confirmation(account, source_key('http://jellyfin'), f'jf-{kind}', time.time())
                store.confirm(store.subscription(account['id']), time.time())
                delivery_id = self.queue(store.subscription(account['id']), kind)
                with store.transaction() as conn:
                    if kind == 'blocked': conn.execute('UPDATE users SET is_blocked=1 WHERE id=?', (account['id'],))
                    elif kind == 'expired': conn.execute("UPDATE users SET expires_at='2000-01-01T00:00:00+00:00' WHERE id=?", (account['id'],))
                    else: conn.execute('DELETE FROM users WHERE id=?', (account['id'],))
                with patch.object(mail, 'send_email') as sender, patch.object(recaps, 'get_monthly_report') as report:
                    await recaps.run_once()
                sender.assert_not_called(); report.assert_not_called()
                self.assertEqual(self.delivery(delivery_id)['state'], 'cancelled')

    async def test_known_temporary_failure_retries_three_times_with_stable_id(self):
        delivery_id = self.queue()
        with patch.object(recaps, 'get_monthly_report', new=AsyncMock(return_value=self.report)), patch.object(mail, 'send_email', side_effect=mail.DeliveryError('retry', 'SMTP 451')) as sender:
            for attempt in range(1, 4):
                await self.run_claim()
                row = self.delivery(delivery_id)
                self.assertEqual(row['attempts'], attempt)
                self.assertEqual(row['state'], 'failed' if attempt == 3 else 'retry')
                if attempt < 3:
                    self.assertGreater(row['next_attempt_at'], time.time() + 250)
                    with store.transaction() as conn:
                        conn.execute('UPDATE email_recap_deliveries SET next_attempt_at=0 WHERE id=?', (delivery_id,))
        self.assertEqual(len(set(call.args[2] for call in sender.call_args_list)), 1)
        self.assertIsNone(store.claim_delivery(time.time()))

    async def test_ambiguous_smtp_failure_never_automatically_retries(self):
        delivery_id = self.queue()
        with patch.object(recaps, 'get_monthly_report', new=AsyncMock(return_value=self.report)), patch.object(mail, 'send_email', side_effect=mail.DeliveryError('unknown', 'Check mail logs')):
            await self.run_claim()
        self.assertEqual(self.delivery(delivery_id)['state'], 'unknown')
        self.assertIsNone(store.claim_delivery(time.time() + 86400))

    def test_stale_worker_claims_are_recovered_without_resending_uncertain_mail(self):
        delivery_id = self.queue()
        first = store.claim_delivery(time.time())
        second = store.claim_delivery(time.time() + 1801)
        self.assertNotEqual(first['claim'], second['claim'])
        self.assertFalse(store.begin_sending(first, time.time()))
        self.assertTrue(store.begin_sending(second, time.time()))
        store.claim_delivery(time.time() + 1801)
        self.assertEqual(self.delivery(delivery_id)['state'], 'unknown')
        store.finish(first, 'sent', 'Old worker', time.time())
        self.assertEqual(self.delivery(delivery_id)['state'], 'unknown')

    async def test_partial_or_over_limit_report_is_not_emailed(self):
        delivery_id = self.queue()
        with patch.object(recaps, 'get_monthly_report', new=AsyncMock(side_effect=HistoryLimitError('limit'))), patch.object(mail, 'send_email') as sender:
            await self.run_claim()
        sender.assert_not_called()
        self.assertEqual(self.delivery(delivery_id)['state'], 'failed')


class RecapApiTests(RecapFixture, unittest.TestCase):
    def setUp(self):
        super().setUp()
        app = FastAPI()
        app.include_router(router.router)
        self.app = app
        self.client = TestClient(app)
        self.addCleanup(self.client.close)

    def login(self, role='admin'):
        self.app.dependency_overrides[get_current_user] = lambda: {**self.user, 'role': role, 'features': {'stats': True}}

    def test_authentication_roles_and_recipient_override(self):
        self.assertEqual(self.client.get('/admin/email-recaps').status_code, 401)
        self.assertEqual(self.client.get('/profile/email-recaps').status_code, 401)
        self.login('user')
        self.assertEqual(self.client.get('/admin/email-recaps').status_code, 403)
        self.assertEqual(self.client.get('/admin/email-recaps/preview').status_code, 403)
        self.assertEqual(self.client.post('/admin/email-recaps/test', json={}).status_code, 403)
        self.login()
        result = self.client.get('/admin/email-recaps')
        self.assertEqual(result.status_code, 200)
        self.assertEqual(result.headers['cache-control'], 'no-store')
        self.assertNotIn('PRIVATE-STATS-KEY', result.text)
        result = self.client.post('/admin/email-recaps/test', json={'request_id': 'c49b0c52-4528-4c1d-8c78-57aafeb24f58', 'recipient_email': 'other@example.test'})
        self.assertEqual(result.status_code, 422)
        result = self.client.put('/profile/email-recaps', json={'enabled': False, 'user_id': 5})
        self.assertEqual(result.status_code, 422)

    def test_url_and_schedule_validation_do_not_write_partial_settings(self):
        self.login()
        for value in ['javascript:alert(1)', 'https://user:secret@example.test', 'https://example.test/path', 'https://example.test?token=secret', 'https://example.test#token', 'https://example.test:0', 'https://example.test\\evil']:
            result = self.client.put('/admin/email-recaps', json={**self.config, 'public_url': value})
            self.assertEqual(result.status_code, 422, value)
        for field, value in [('day', 0), ('day', 29), ('hour', 24)]:
            self.assertEqual(self.client.put('/admin/email-recaps', json={**self.config, field: value}).status_code, 422)
        with patch.object(recaps, 'smtp_email_config_ready', return_value=(False, 'Email is disabled.')):
            self.assertEqual(self.client.put('/admin/email-recaps', json={**self.config, 'enabled': True}).status_code, 409)
        self.assertEqual(store.settings()['public_url'], self.config['public_url'])
        self.assertFalse(store.settings()['enabled'])

    def test_preview_uses_own_report_and_test_requires_confirmed_email(self):
        self.login()
        with patch.object(recaps, 'get_monthly_report', new=AsyncMock(return_value=self.report)) as report, patch.object(mail, 'send_email') as sender:
            result = self.client.get('/admin/email-recaps/preview')
        self.assertEqual(result.status_code, 200)
        self.assertEqual(report.await_args.args[0]['id'], self.user['id'])
        self.assertNotIn('PRIVATE-TOKEN', result.text)
        sender.assert_not_called()
        payload = {'request_id': 'c49b0c52-4528-4c1d-8c78-57aafeb24f58', 'month': self.report['month']}
        self.assertEqual(self.client.post('/admin/email-recaps/test', json=payload).status_code, 409)
        self.subscribe()
        with patch.object(mail, 'send_email') as sender:
            first = self.client.post('/admin/email-recaps/test', json=payload)
            second = self.client.post('/admin/email-recaps/test', json=payload)
        self.assertEqual(first.status_code, 202)
        self.assertEqual(first.json()['id'], second.json()['id'])
        sender.assert_not_called()

    def test_partial_month_test_rejected_and_public_get_does_not_mutate(self):
        self.login(); sub, token = self.subscribe()
        result = self.client.post('/admin/email-recaps/test', json={'request_id': 'c49b0c52-4528-4c1d-8c78-57aafeb24f58', 'month': datetime.now(timezone.utc).strftime('%Y-%m')})
        self.assertEqual(result.status_code, 422)
        self.assertEqual(self.client.get('/email-recaps/confirm').status_code, 405)
        result = self.client.post('/email-recaps/check', json={'action': 'unsubscribe', 'token': sub['unsubscribe_token']})
        self.assertEqual(result.status_code, 200)
        self.assertEqual(store.subscription(self.user['id'])['state'], 'enabled')


class RecapEmailTests(unittest.TestCase):
    def setUp(self):
        self.runtime = runtime()
        patched = patch.object(mail, 'get_runtime_settings', return_value=self.runtime)
        patched.start(); self.addCleanup(patched.stop)
        self.rendered = mail.render_recap(fixture_report(), 'Viewer', 'https://beta.example.test', 'https://beta.example.test/email-recaps#action=unsubscribe&token=fixture')

    def fake_smtp(self):
        smtp = MagicMock()
        smtp.mail.return_value = (250, b'OK')
        smtp.rcpt.return_value = (250, b'OK')
        smtp.data.return_value = (250, b'Accepted')
        return smtp

    def test_render_escapes_names_and_titles_and_includes_no_artwork_credentials(self):
        report = fixture_report()
        report['top_titles'][0]['title'] = '<img src=x onerror=alert(1)>'
        rendered = mail.render_recap(report, '<script>alert(1)</script>', 'https://beta.example.test', 'https://beta.example.test/email-recaps#token=example')
        self.assertNotIn('<script>', rendered['body_html'])
        self.assertNotIn('<img src=x', rendered['body_html'])
        self.assertIn('&lt;script&gt;', rendered['body_html'])
        self.assertNotIn('PRIVATE-TOKEN', str(rendered))
        self.assertIn('Unsubscribe', rendered['body_text'])
        self.assertIn('UTC', rendered['body_text'])
        self.assertIn('1,500', rendered['body_html'])

    def test_mailbox_validation_rejects_injection_and_multiple_recipients(self):
        for value in ['a@example.test\r\nBcc:b@example.test', 'a@example.test,b@example.test', 'Name <a@example.test>', 'x@', 'a;b@example.test']:
            self.assertIsNone(mail.valid_email(value))

    def test_smtp_acceptance_survives_quit_error_and_preserves_mime_message_id(self):
        smtp = self.fake_smtp()
        smtp.quit.side_effect = smtplib.SMTPServerDisconnected('after acceptance')
        before = MagicMock()
        with patch.object(mail.smtplib, 'SMTP', return_value=smtp):
            mail.send_email('viewer@example.test', self.rendered, '<stable@example.test>', before)
        before.assert_called_once()
        message = BytesParser(policy=policy.default).parsebytes(smtp.data.call_args.args[0])
        self.assertEqual(message['Message-ID'], '<stable@example.test>')
        self.assertEqual(message['To'], 'viewer@example.test')
        self.assertIsNone(message['Bcc'])
        self.assertIn('1,500', message.get_body(('plain',)).get_content())
        self.assertIn('<!doctype html>', message.get_body(('html',)).get_content())

    def test_temporary_permanent_and_ambiguous_delivery_failures(self):
        for operation, failure, expected in [
            ('mail', (451, b'temporary PRIVATE-KEY'), 'retry'), ('rcpt', (550, b'bad recipient'), 'failed'),
            ('data', (451, b'retry'), 'retry'), ('data', smtplib.SMTPServerDisconnected('lost after DATA'), 'unknown'),
            ('rcpt', smtplib.SMTPServerDisconnected('lost before DATA'), 'retry')]:
            smtp = self.fake_smtp()
            if isinstance(failure, Exception): getattr(smtp, operation).side_effect = failure
            else: getattr(smtp, operation).return_value = failure
            with self.subTest(operation=operation, expected=expected), patch.object(mail.smtplib, 'SMTP', return_value=smtp):
                with self.assertRaises(mail.DeliveryError) as exc:
                    mail.send_email('viewer@example.test', self.rendered, '<stable@example.test>')
            self.assertEqual(exc.exception.state, expected)
            self.assertNotIn('PRIVATE-KEY', exc.exception.detail)

    def test_consent_cancellation_happens_before_smtp_data(self):
        smtp = self.fake_smtp()
        with patch.object(mail.smtplib, 'SMTP', return_value=smtp), self.assertRaises(mail.DeliveryCancelled):
            mail.send_email('viewer@example.test', self.rendered, '<stable@example.test>', MagicMock(side_effect=mail.DeliveryCancelled))
        smtp.data.assert_not_called()

    def test_real_smtp_is_captured_locally_without_external_delivery(self):
        messages = []
        class Capture(socketserver.StreamRequestHandler):
            def handle(self):
                self.wfile.write(b'220 local capture\r\n')
                while line := self.rfile.readline():
                    command = line.split(b' ', 1)[0].strip().upper()
                    if command in (b'EHLO', b'HELO'):
                        self.wfile.write(b'250-localhost\r\n250 SIZE 1000000\r\n')
                    elif command == b'DATA':
                        self.wfile.write(b'354 Send content\r\n')
                        data = []
                        while (part := self.rfile.readline()) != b'.\r\n':
                            if not part: return
                            data.append(part[1:] if part.startswith(b'..') else part)
                        messages.append(b''.join(data))
                        self.wfile.write(b'250 Captured\r\n')
                    elif command == b'QUIT':
                        self.wfile.write(b'221 Bye\r\n'); return
                    else:
                        self.wfile.write(b'250 OK\r\n')
        with socketserver.TCPServer(('127.0.0.1', 0), Capture) as server:
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()
            self.runtime.magent_notify_email_smtp_port = server.server_address[1]
            try:
                mail.send_email('viewer@example.test', self.rendered, '<local-capture@example.test>')
            finally:
                server.shutdown(); thread.join(timeout=5)
        self.assertEqual(len(messages), 1)
        parsed = BytesParser(policy=policy.default).parsebytes(messages[0])
        self.assertEqual(parsed['Message-ID'], '<local-capture@example.test>')
        self.assertIn('Severance', parsed.get_body(('html',)).get_content())


if __name__ == '__main__':
    unittest.main()


class OnDemandReportTests(RecapFixture, unittest.IsolatedAsyncioTestCase):
    async def test_new_confirmation_defaults_to_manual_without_changing_schedule(self):
        with patch.object(mail, 'send_email'):
            result = await recaps.subscribe(self.user)
        self.assertFalse(result['automatic_monthly'])
        self.assertFalse(store.settings()['enabled'])
        self.assertEqual(result['state'], 'pending')
        with self.assertRaises(recaps.RecapError):
            recaps.queue_personal(self.user, None, 'pending')

    async def test_manual_current_month_delivers_with_monthly_schedule_off(self):
        sub, _ = self.subscribe()
        store.set_automatic(self.user['id'], False)
        month = datetime.now(timezone.utc).strftime('%Y-%m')
        queued = recaps.queue_personal(self.user, month, 'manual-1')
        self.assertEqual(recaps.queue_personal(self.user, month, 'manual-1')['id'], queued['id'])
        report = {**self.report, **month_periods(month, datetime.now(timezone.utc))}
        def send(recipient, rendered, message_id, before_data):
            before_data()
            self.assertEqual(recipient, self.user['email'])
            self.assertIn('so far', rendered['subject'])
            self.assertNotIn('[Test]', rendered['subject'])
        with patch.object(recaps, 'get_monthly_report', new_callable=AsyncMock, return_value=report), patch.object(mail, 'send_email', side_effect=send):
            await recaps.process_delivery(store.claim_delivery(time.time()))
        self.assertEqual(self.delivery(queued['id'])['state'], 'sent')
        self.assertFalse(store.settings()['enabled'])
        self.assertFalse(store.subscription(self.user['id'])['automatic_monthly'])
        with self.assertRaises(recaps.RecapError) as error:
            recaps.queue_personal(self.user, month, 'manual-2')
        self.assertEqual(error.exception.status, 429)

    async def test_automatic_opt_out_cancels_scheduled_but_keeps_manual(self):
        sub, _ = self.subscribe()
        with store.transaction() as conn:
            scheduled = store._enqueue(conn, sub, self.report['month'], 'scheduled', 'scheduled-fixture', self.config['public_url'], time.time())
        manual = recaps.queue_personal(self.user, None, 'manual')
        store.set_automatic(self.user['id'], False)
        self.assertEqual(self.delivery(scheduled)['state'], 'cancelled')
        self.assertEqual(self.delivery(manual['id'])['state'], 'queued')
        self.assertEqual(store.subscription(self.user['id'])['state'], 'enabled')
        now = datetime.now(timezone.utc)
        store.save_settings({**self.config, 'enabled': True}, now)
        self.assertEqual(store.enqueue_due(now + timedelta(days=40)), 0)

    async def test_changed_identity_cancels_manual_delivery(self):
        self.subscribe()
        queued = recaps.queue_personal(self.user, None, 'manual')
        delivery = store.claim_delivery(time.time())
        db.set_user_email('viewer', 'changed@example.test')
        with patch.object(mail, 'send_email') as send:
            await recaps.process_delivery(delivery)
        send.assert_not_called()
        self.assertEqual(self.delivery(queued['id'])['state'], 'cancelled')

    async def test_regular_user_can_only_send_to_self(self):
        self.subscribe()
        app = FastAPI(); app.include_router(router.router)
        app.dependency_overrides[get_current_user] = lambda: {'username': 'viewer', 'role': 'user', 'features': {'stats': True}}
        client = TestClient(app)
        body = {'month': self.report['month'], 'request_id': '11111111-1111-4111-8111-111111111111'}
        for extra in [{'email': 'other@example.test'}, {'user_id': 42}, {'kind': 'scheduled'}]:
            self.assertEqual(client.post('/profile/email-recaps/send', json={**body, **extra}).status_code, 422)
        self.assertEqual(client.post('/profile/email-recaps/send', json=body).status_code, 202)
        response = client.get('/profile/email-recaps')
        self.assertEqual(response.headers['cache-control'], 'no-store')
        self.assertEqual(len(response.json()['deliveries']), 1)