import json
import threading
import time
import unittest
from unittest.mock import Mock

from app.channel_publisher import ChannelMetadataPublisher
from app.metadata_publisher import MetadataSettings, camera_key
from app.metadata_runtime import publisher_from_environment


def wait(predicate):
    deadline = time.monotonic() + 2
    while not predicate() and time.monotonic() < deadline:
        time.sleep(.002)
    assert predicate()


def snapshot(camera='a', seq=0, generation=1):
    return dict(cameraId=camera, runId='r' + str(generation), runGeneration=generation,
                frameSeq=seq, objects=[])


class ChannelPublisherTest(unittest.TestCase):
    def make(self, factory=None, **kw):
        self.sinks = {}
        def default(key):
            sink = Mock()
            self.sinks[key] = sink
            return sink
        p = ChannelMetadataPublisher(MetadataSettings('unused', worker_mode='channel', **kw),
                                     sink_factory=factory or default)
        self.addCleanup(p.close)
        return p

    def test_sixteen_concurrent_admission_seventeenth_rejected(self):
        p = self.make()
        p.start()
        barrier = threading.Barrier(16)
        accepted = []
        def run(i):
            barrier.wait()
            accepted.append(p.channel(str(i), 'r1', 1, 'RUNNING'))
        threads = [threading.Thread(target=run, args=(i,)) for i in range(16)]
        for t in threads:t.start()
        for t in threads:t.join(2)
        self.assertEqual(accepted, [True] * 16)
        self.assertEqual(p.metrics()['workers'], 16)
        self.assertFalse(p.channel('17', 'r1', 1, 'STARTING'))
        self.assertNotIn(camera_key('17'), p._channels)
        self.assertEqual(p.metrics()['workers'], 16)
        for i in range(16):
            self.assertTrue(p.submit(snapshot(str(i))))
        wait(lambda: p.metrics().get('sent_frames') == 16)
        for i in range(16):p.channel(str(i), 'r1', 1, 'STOPPED')
        for i in range(16):self.assertTrue(p.wait_channel(str(i), 'r1', 1, 1))
        self.assertEqual(p.metrics()['workers'], 0)
        self.assertEqual(p.metrics()['workers_reclaimed'], 16)

    def test_registered_stopped_heartbeat_uses_control_only(self):
        p = self.make(stream_ttl=1)
        p.channel('a', None, 0, 'STOPPED')
        self.assertEqual(p.metrics()['workers'], 0)
        p.start()
        wait(lambda: self.sinks['control'].write.call_count >= 2)
        self.assertEqual(p.metrics()['workers'], 0)
        self.assertEqual(set(self.sinks), {'control'})

    def test_latest_noop_order_generation_and_close(self):
        p = self.make()
        p.channel('a', 'r1', 1, 'RUNNING')
        self.assertTrue(p.submit(snapshot(seq=1)))
        self.assertTrue(p.channel('a', 'r1', 1, 'RUNNING'))
        self.assertTrue(p.submit(snapshot(seq=2)))
        self.assertEqual(p.metrics()['replaced'], 1)
        self.assertEqual(p.metrics()['workers_created'], 1)
        self.assertFalse(p.submit(snapshot(seq=1)))
        p.start()
        wait(lambda: p.metrics().get('sent_frames') == 1)
        sink = self.sinks[camera_key('a')]
        self.assertEqual(json.loads(sink.write.call_args.args[2])['frameSeq'], 2)
        self.assertFalse(p.channel('a', 'r2', 2, 'RUNNING'))
        p.channel('a', 'r1', 1, 'STOPPED')
        self.assertTrue(p.wait_channel('a', 'r1', 1, 1))
        sink.close.assert_called_once()
        self.assertTrue(p.channel('a', 'r2', 2, 'RUNNING'))
        self.assertFalse(p.submit(snapshot(seq=3)))
        self.assertTrue(p.submit(snapshot(seq=0, generation=2)))
        wait(lambda: p.metrics().get('sent_frames') == 2)

    def test_slow_channel_isolation_and_stop_wait(self):
        entered, release = threading.Event(), threading.Event()
        writes = []
        def factory(key):
            sink = Mock()
            def write(k, state, payload):
                if key == camera_key('a') and payload:
                    entered.set();release.wait(2)
                writes.append((k, state['state'], payload))
            sink.write.side_effect = write
            return sink
        p = self.make(factory)
        self.addCleanup(release.set)
        p.start();p.channel('a', 'r1', 1, 'RUNNING');p.channel('b', 'r1', 1, 'RUNNING')
        p.submit(snapshot('a'));self.assertTrue(entered.wait(1))
        p.submit(snapshot('b'));wait(lambda: p.metrics().get('sent_frames') == 1)
        self.assertGreater(p.metrics()['inflightBytes'], 0)
        p.channel('a', 'r1', 1, 'STOPPED')
        self.assertFalse(p.wait_channel('a', 'r1', 1, .01))
        self.assertFalse(p.channel('a', 'r2', 2, 'RUNNING'))
        self.assertEqual(p.metrics()['retiringWorkers'], 1)
        release.set();self.assertTrue(p.wait_channel('a', 'r1', 1, 1))
        a = [x for x in writes if x[0] == camera_key('a')]
        self.assertEqual(a[-1][1], 'STOPPED')
        self.assertTrue(p.channel('a', 'r2', 2, 'RUNNING'))

    def test_control_old_stop_cannot_cross_new_run(self):
        entered, release = threading.Event(), threading.Event()
        writes = []
        def factory(key):
            sink = Mock()
            def write(k, state, payload):
                if key == 'control':entered.set();release.wait(2)
                writes.append((state['runId'], state['state']))
            sink.write.side_effect = write
            return sink
        p = self.make(factory)
        self.addCleanup(release.set)
        p.start();p.channel('a', None, 0, 'STOPPED')
        self.assertTrue(entered.wait(1))
        accepted = []
        t = threading.Thread(target=lambda: accepted.append(p.channel('a', 'r1', 1, 'RUNNING')))
        t.start();time.sleep(.02)
        self.assertEqual(p.metrics()['workers'], 0)
        release.set();t.join(1)
        self.assertEqual(accepted, [True])
        wait(lambda: len(writes) == 2)
        self.assertEqual(writes, [(None, 'STOPPED'), ('r1', 'RUNNING')])

    def test_terminal_write_failure_blocks_until_control_confirmation(self):
        def factory(key):
            sink = Mock()
            if key != 'control':
                sink.write.side_effect = lambda k, s, p: (_ for _ in ()).throw(OSError()) if s['state']=='STOPPED' else None
            return sink
        p = self.make(factory)
        p.start();p.channel('a', 'r1', 1, 'RUNNING')
        wait(lambda: p.metrics().get('sent_status') == 1)
        p.channel('a', 'r1', 1, 'STOPPED')
        self.assertFalse(p.wait_channel('a', 'r1', 1, 1))
        self.assertFalse(p.channel('a', 'r2', 2, 'RUNNING'))
        self.assertEqual(p.metrics()['terminalErrors'], 1)

    def test_factory_failure_and_close_failure_visible(self):
        def factory(key):
            sink = Mock()
            if key != 'control':sink.close.side_effect = OSError()
            return sink
        p = self.make(factory)
        p.start();p.channel('a', 'r1', 1, 'RUNNING')
        p.channel('a', 'r1', 1, 'STOPPED')
        self.assertFalse(p.wait_channel('a', 'r1', 1, 1))
        self.assertFalse(p.channel('a', 'r2', 2, 'RUNNING'))
        self.assertFalse(p.close(1))

    def test_whole_close_has_one_deadline(self):
        entered = [threading.Event(), threading.Event()]
        release = threading.Event()
        def factory(key):
            sink = Mock()
            if key != 'control':
                i = 0 if key == camera_key('a') else 1
                def write(*args):entered[i].set();release.wait(2)
                sink.write.side_effect = write
            return sink
        p = self.make(factory)
        self.addCleanup(release.set)
        p.start()
        for key in ('a','b'):p.channel(key, 'r1', 1, 'RUNNING')
        for event in entered:self.assertTrue(event.wait(1))
        started = time.monotonic()
        self.assertFalse(p.close(.04))
        self.assertLess(time.monotonic() - started, .09)
        release.set();self.assertTrue(p.close(1))
        self.assertEqual(p.metrics()['workers'], 0)

    def test_global_bytes_message_limit_and_expiry(self):
        p = self.make(max_bytes=120, max_message_bytes=150)
        for key in ('a','b'):p.channel(key, 'r1', 1, 'RUNNING')
        self.assertTrue(p.submit(snapshot('a')))
        self.assertFalse(p.submit(snapshot('b')))
        self.assertLessEqual(p.metrics()['queuedBytes'], 120)
        big = dict(snapshot('a',seq=1),objects=['x'*200])
        self.assertFalse(p.submit(big))
        self.assertEqual(p.metrics()['message_limit'], 1)
        p.clock = lambda: time.monotonic() + 10
        p.start();wait(lambda:p.metrics().get('expired') == 1)

    def test_environment_and_config_limits(self):
        p = publisher_from_environment(dict(CONTEXTRA_METADATA_ENABLED='true',
            CONTEXTRA_METADATA_REDIS_URL='redis://reader:fixture@localhost:1',
            CONTEXTRA_METADATA_WORKER_MODE='channel',CONTEXTRA_METADATA_WORKERS='4',
            CONTEXTRA_METADATA_ACTIVE_WORKER_LIMIT='16'))
        self.addCleanup(p.close)
        self.assertIsInstance(p, ChannelMetadataPublisher)
        self.assertEqual(len(p._sinks),1)
        self.assertEqual(p.settings.stream_length,100)
        self.assertEqual(p.settings.stream_ttl,10)
        for n in (0,17,True):
            with self.assertRaises(ValueError):MetadataSettings('x',active_worker_limit=n)
        with self.assertRaises(ValueError):MetadataSettings('x',worker_mode='unknown')

    def test_diagnostics_off_expire(self):
        p = self.make()
        self.assertFalse(p.metrics()['profile']['enabled'])
        now=[0.]
        p._profile.clock=lambda:now[0]
        p.configure_metrics(True,1)
        now[0]=2
        self.assertFalse(p.metrics()['profile']['enabled'])

    def test_frame_failure_does_not_stop_other_channel_or_retry_frame(self):
        def factory(key):
            sink = Mock()
            if key == camera_key('a'):
                sink.write.side_effect = lambda k, s, payload: (_ for _ in ()).throw(OSError()) if payload else None
            return sink
        p = self.make(factory)
        p.start()
        for key in ('a','b'):
            p.channel(key,'r1',1,'RUNNING');p.submit(snapshot(key))
        wait(lambda:p.metrics().get('send_error') == 1 and p.metrics().get('sent_frames') == 1)
        self.assertEqual(p.metrics()['queued'],0)
        p.submit(snapshot('b',1));wait(lambda:p.metrics().get('sent_frames') == 2)
        self.assertEqual(p.metrics()['send_error'],1)

    def test_admission_and_cleanup_connected_to_result_path(self):
        from app.det_result_path import DetResultPath
        p = self.make(active_worker_limit=1)
        p.start();p.channel('occupied','r1',1,'RUNNING')
        ingress=Mock(identity=('server','a','stream','r1',1))
        path=DetResultPath(ingress,None,'RFDETRMedium',.5,Mock(),lambda:True,Mock(),publisher=p)
        with self.assertRaisesRegex(RuntimeError,'admission rejected'):path.start()
        self.assertIsNone(path._thread)
        p.channel('occupied','r1',1,'STOPPED')
        self.assertTrue(p.wait_channel('occupied','r1',1,1))
        path.start()
        thread=p._workers[camera_key('a')]
        path.close(1)
        self.assertFalse(thread.is_alive())
        self.assertEqual(p.metrics()['workers'],0)