import threading
import time
import unittest
from types import SimpleNamespace

from app.stream_lifecycle import RoleRunHandle
from tests.test_provider_freshness import FreshnessTest


def terminal_then_finalize(spec, run, generation, settings, token, commands, output, stop, ack, parent):
    from app.channel_child import parent_death_guard, send_latest
    parent_death_guard(parent);output.cancel_join_thread()
    state=dict(run_id=run,run_generation=generation,state='stopped',observed_state='stopped',
               desired_state='stopped',roles={},last_error=None,cleanup_error=None)
    send_latest(output,dict(kind='terminal',run_id=run,run_generation=generation,value=state,failed=False))
    ack.wait(1);time.sleep(3.5)


class PoseScaleLifecycleTest(unittest.TestCase):
    def test_terminal_finalization_is_not_prematurely_terminated(self):
        import tempfile
        from app.channel_process import ProcessSlots, ProcessStreamGroup
        with tempfile.TemporaryDirectory() as d:
            slots=ProcessSlots(lock_path=d+'/lock')
            g=ProcessStreamGroup(dict(stream_id='fixture',source_id='1',server_id='server',uri='unused',
                                     store=False),slots=slots,child_target=terminal_then_finalize)
            try:
                g.start();self.assertTrue(g._cleanup_done.wait(8))
                self.assertFalse(g._last_cleanup['forced'])
                self.assertEqual(g._last_cleanup['diagnostic']['exit_code'],0)
            finally:g.stop();g._cleanup_done.wait(3);slots.close()

    def test_stop_does_not_touch_a_pipeline_still_being_constructed(self):
        h=RoleRunHandle('shared',1)
        h.bind_pipeline(SimpleNamespace(stop=lambda:self.fail('unstarted SDK stop'),
                                        start=lambda:self.fail('late SDK start')))
        h.stop()
        self.assertIsNone(h.start_pipeline())

    def test_stop_closes_admission_now_but_sends_eos_after_source_ready(self):
        h=RoleRunHandle('shared',1); h.await_source_on_stop(.2)
        calls=[];h.bind_pipeline(SimpleNamespace(start=lambda:None,stop=lambda:calls.append(1)))
        h.start_pipeline();t=threading.Thread(target=h.stop);t.start()
        time.sleep(.02);self.assertTrue(h.stop_requested);self.assertEqual(calls,[])
        h.note_source_ready();t.join(1)
        self.assertFalse(t.is_alive());self.assertEqual(calls,[1])
        h.stop();self.assertEqual(calls,[1])

    def test_source_wait_is_bounded_not_infinite(self):
        h=RoleRunHandle('shared',1);h.await_source_on_stop(.02)
        calls=[];h.bind_pipeline(SimpleNamespace(start=lambda:None,stop=lambda:calls.append(1)))
        h.start_pipeline();started=time.monotonic();failure=h.stop()
        self.assertLess(time.monotonic()-started,.5);self.assertEqual(calls,[])
        self.assertEqual(failure.failure_stage,'source_stop_wait')

    def test_existing_cleanup_deadline_allows_source_readiness_without_early_eos(self):
        h=RoleRunHandle('shared',1);h.await_source_on_stop(.01)
        calls=[];h.bind_pipeline(SimpleNamespace(start=lambda:None,stop=lambda:calls.append(1)))
        h.start_pipeline()
        t=threading.Thread(target=lambda:h.stop(deadline=time.monotonic()+.4));t.start()
        time.sleep(.05);self.assertEqual(calls,[])
        h.note_source_ready();t.join(1);self.assertEqual(calls,[1])

    def test_provider_progress_is_not_a_success_or_unbounded_stall_reset(self):
        fixture=FreshnessTest();c=fixture.make()
        c.admit(__import__('tests.test_pr13_multi_provider',fromlist=['frame']).frame(0))
        c._providers['medium']['last_response_at']=0.
        fixture.now[0]=1.9;c.note_provider_progress('medium','infer_output')
        self.assertEqual(c._check_stalls_locked(2.1),[])
        self.assertEqual(c._providers['medium']['responses'],0)
        c._providers['medium']['last_input_at']=3.9
        self.assertEqual(c._check_stalls_locked(3.91),['medium'])
        with self.assertRaises(ValueError):c.note_provider_progress('medium','selected')

    def test_gst_pipeline_eos_before_streaming_is_not_completion(self):
        import gi
        gi.require_version('Gst','1.0')
        from gi.repository import Gst
        Gst.init(None)
        p=Gst.parse_launch('appsrc name=src is-live=true format=time ! identity name=seen ! appsink sync=false')
        ready=threading.Event()
        def seen(*args):
            ready.set();return Gst.PadProbeReturn.DROP
        p.get_by_name('seen').get_static_pad('src').add_probe(Gst.PadProbeType.BUFFER,seen)
        try:
            p.set_state(Gst.State.PLAYING)
            self.assertTrue(p.send_event(Gst.Event.new_eos()))
            self.assertIsNone(p.get_bus().timed_pop_filtered(100*Gst.MSECOND,Gst.MessageType.EOS))
        finally:p.set_state(Gst.State.NULL)
        # A separate source proves that ready-then-EOS works without inference.
        p=Gst.parse_launch('appsrc name=src is-live=true format=time ! identity name=seen ! appsink sync=false')
        p.get_by_name('seen').get_static_pad('src').add_probe(Gst.PadProbeType.BUFFER,seen)
        try:
            p.set_state(Gst.State.PLAYING)
            p.get_by_name('src').emit('push-buffer',Gst.Buffer.new_allocate(None,1,None))
            self.assertTrue(ready.wait(1))
            self.assertTrue(p.send_event(Gst.Event.new_eos()))
            self.assertIsNotNone(p.get_bus().timed_pop_filtered(Gst.SECOND,Gst.MessageType.EOS))
        finally:p.set_state(Gst.State.NULL)