import threading
import unittest
from types import SimpleNamespace
from unittest.mock import patch

from app.stream_lifecycle import RoleRunHandle, StreamLifecycle, make_stream_failure, stop_role_handles


class CleanupBoundaryTest(unittest.TestCase):
    def test_join_timeout_is_failure_not_stopped(self):
        events = []
        handle = RoleRunHandle('det', 1, lambda *args: events.append(args))
        handle.bind_thread(SimpleNamespace(join=lambda timeout: None, is_alive=lambda: True))
        failure = stop_role_handles({'det': handle})
        self.assertEqual('worker_join', failure.failure_stage)
        self.assertFalse(any(e[2] == 'stopped' for e in events))

    def test_stop_failure_is_sticky(self):
        events = []
        handle = RoleRunHandle('det', 1, lambda *args: events.append(args))
        calls = []
        def stop():
            calls.append(1)
            raise RuntimeError('unsafe detail')
        handle.bind_pipeline(SimpleNamespace(stop=stop))
        first = handle.stop()
        self.assertIs(first, handle.stop())
        self.assertEqual([1], calls)
        self.assertNotIn('unsafe', first.message)
        self.assertIs(first, stop_role_handles({'det': handle}))
        self.assertFalse(any(e[2] == 'stopped' for e in events))

    def test_stop_before_late_start_prevents_start(self):
        handle = RoleRunHandle('det', 1)
        handle.stop()
        handle.bind_pipeline(SimpleNamespace(start=lambda: self.fail('late start')))
        self.assertIsNone(handle.start_pipeline())

    def test_start_stop_serialized_without_callback_lock(self):
        entered, release = threading.Event(), threading.Event()
        handle = RoleRunHandle('det', 1)
        calls = []
        def start():
            entered.set()
            release.wait(1)
            handle.mark_initialized()
            calls.append('start')
        handle.bind_pipeline(SimpleNamespace(start=start, stop=lambda: calls.append('stop')))
        a = threading.Thread(target=handle.start_pipeline)
        b = threading.Thread(target=handle.stop)
        a.start()
        self.assertTrue(entered.wait(1))
        b.start()
        release.set()
        a.join(1)
        b.join(1)
        self.assertFalse(a.is_alive() or b.is_alive())
        self.assertEqual(['start', 'stop'], calls)

    def test_initial_and_cleanup_error_preserved(self):
        life = StreamLifecycle()
        gen = life.begin()
        life.configure_roles(gen, ['det'])
        life.signal_error(gen, make_stream_failure('FIRST_FRAME_TIMEOUT', 'first_frame_wait'))
        life.finish_stop(gen, make_stream_failure('STOP_CLEANUP_FAILED', 'worker_join'))
        result = life.snapshot()
        self.assertEqual('FIRST_FRAME_TIMEOUT', result['last_error']['error_code'])
        self.assertEqual('STOP_CLEANUP_FAILED', result['cleanup_error']['error_code'])
        self.assertNotEqual('stopped', result['roles']['det']['state'])

    def test_stage_counters_first_time_and_generation(self):
        life = StreamLifecycle()
        gen = life.begin()
        life.configure_roles(gen, ['det'])
        life.signal_source_stage(gen, 'det', 'consume_entered', 1)
        life.signal_source_stage(gen, 'det', 'consume_entered', 2)
        life.signal_source_stage(gen-1, 'det', 'consume_entered', 3)
        role = life.snapshot()['roles']['det']
        self.assertEqual(2, role['stage_counts']['consume_entered'])
        self.assertEqual(1, role['stage_first_at']['consume_entered'])
        self.assertEqual('starting', life.snapshot()['state'])
        life.request_stop()
        self.assertFalse(life.signal_frame(gen, 'det'))
        self.assertFalse(life.signal_source_stage(gen, 'det', 'native_completed'))
        self.assertFalse(life.signal_initialized(gen, 'det'))


class GroupCleanupBoundaryTest(unittest.TestCase):
    def test_concurrent_stop_has_one_owner_and_retains_timeout(self):
        from app.stream_manager import StreamGroup
        group = StreamGroup({'stream_id': 'test-only', 'store': False})
        group.lifecycle.begin()
        entered, release = threading.Event(), threading.Event()
        handle = RoleRunHandle('det', group.run_generation)
        calls = []
        def stop():
            calls.append(1)
            entered.set()
            release.wait(2)
        handle.bind_pipeline(SimpleNamespace(stop=stop))
        group.role_handles['det'] = handle
        group.CLEANUP_TIMEOUT_S = .02
        a, b = threading.Thread(target=group.stop), threading.Thread(target=group.stop)
        try:
            a.start()
            self.assertTrue(entered.wait(1))
            b.start()
            a.join(1)
            b.join(1)
            self.assertFalse(a.is_alive() or b.is_alive())
            self.assertEqual([1], calls)
            self.assertIs(handle, group.role_handles['det'])
            threading.Event().wait(.04)
            self.assertEqual('error', group.state)
            with self.assertRaises(RuntimeError):
                group.start()
        finally:
            release.set()
            self.assertTrue(group._cleanup_done.wait(2))
        self.assertIsNotNone(group._cleanup_failure)

    def test_bootstrap_requests_cleanup_without_self_join(self):
        from app.stream_manager import StreamGroup
        group = StreamGroup({'stream_id': 'test-only', 'store': False})
        gen = group.lifecycle.begin()
        ready = threading.Event()
        def bootstrap():
            ready.wait(1)
            group._teardown_pipelines(gen, wait=False)
        t = threading.Thread(target=bootstrap)
        group._bootstrap_thread = t
        t.start()
        ready.set()
        t.join(1)
        self.assertFalse(t.is_alive())
        self.assertTrue(group._cleanup_done.wait(1))
        self.assertIsNone(group._cleanup_failure)


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