"""Offline Celery tests only: no worker, Redis, container or Vercel is started."""

from contextlib import redirect_stdout
import io
import json
import os
import unittest
from unittest.mock import patch

os.environ['CELERY_BROKER_URL'] = 'rediss://user:synthetic@broker.example.invalid:6379/0'
from app import checked_broker_url, probe


JOB_ID = '0' * 31 + '1'


class ProbeTests(unittest.TestCase):
    def test_eager_execution_correlates_start_and_completion_without_a_broker(self):
        stream = io.StringIO()
        with redirect_stdout(stream), patch('app.time.sleep') as sleep:
            with patch('socket.socket', side_effect=AssertionError('network forbidden')):
                result = probe.apply(args=[JOB_ID, 2], task_id='local-probe-1', throw=True)
                self.assertEqual(result.get(), {'job_id': JOB_ID, 'completed': True})
        sleep.assert_called_once_with(2)
        rows = [json.loads(line) for line in stream.getvalue().splitlines()]
        self.assertEqual([row['event'] for row in rows], ['started', 'completed'])
        self.assertEqual({row['task_id'] for row in rows}, {'local-probe-1'})
        self.assertEqual({row['job_id'] for row in rows}, {JOB_ID})

    def test_invalid_work_does_not_sleep_or_emit_completion(self):
        cases = [('', 2), ('sample@example.invalid', 2), (JOB_ID, -1),
                 (JOB_ID, 91), (JOB_ID, True), (JOB_ID, '2'), (JOB_ID, 2.0)]
        for args in cases:
            stream = io.StringIO()
            with self.subTest(args=args), redirect_stdout(stream), patch('app.time.sleep') as sleep:
                with self.assertRaises(ValueError):
                    probe.apply(args=args, throw=True)
                sleep.assert_not_called()
                self.assertEqual(stream.getvalue(), '')

    def test_same_business_id_does_not_silently_deduplicate_celery_tasks(self):
        stream = io.StringIO()
        with redirect_stdout(stream), patch('app.time.sleep'):
            for task_id in ('local-attempt-a', 'local-attempt-b'):
                probe.apply(args=[JOB_ID, 0], task_id=task_id, throw=True).get()
        rows = [json.loads(line) for line in stream.getvalue().splitlines()]
        completed = [row for row in rows if row['event'] == 'completed']
        self.assertEqual(len(completed), 2)
        self.assertEqual({row['job_id'] for row in completed}, {JOB_ID})
        self.assertEqual(len({row['task_id'] for row in completed}), 2)

    def test_insecure_or_malformed_broker_settings_do_not_echo_credentials(self):
        urls = [None, 'redis://user:private-value@host/0',
                'rediss://user:private-value@host/0?ssl_cert_reqs=none',
                'rediss://user:private-value@host/0?ssl_cert_reqs=',
                'rediss://user:private-value@host:notaport/0', 'rediss:///0']
        for value in urls:
            with self.subTest(value=value), self.assertRaises(ValueError) as raised:
                checked_broker_url(value)
            self.assertNotIn('private-value', str(raised.exception))


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