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

from app.stream_lifecycle import RoleRunHandle, make_stream_failure


class StopObservationTest(unittest.TestCase):
    def setUp(self):
        self.env = patch.dict(os.environ, {"CONTEXTRA_START_DIAGNOSTICS_DIR": "/unused"})
        self.env.start()
        self.addCleanup(self.env.stop)

    def test_correct_target_and_changed_target(self):
        h = RoleRunHandle("det", 4)
        calls = []
        p = SimpleNamespace(start=lambda: None, stop=lambda: calls.append(1))
        h.bind_pipeline(p)
        h.start_pipeline()
        h.stop()
        h.stop()
        s = h.diagnostic_snapshot()
        self.assertTrue(s["stop_matches_started"])
        self.assertEqual([1], calls)
        self.assertIn("stop_already_finished", s["events"])
        other = RoleRunHandle("det", 4)
        other.bind_pipeline(p)
        other.start_pipeline()
        other.bind_pipeline(SimpleNamespace(stop=lambda: None))
        other.stop()
        self.assertFalse(other.diagnostic_snapshot()["stop_matches_started"])
        self.assertNotEqual(h.diagnostic.token, other.diagnostic.token)

    def test_skip_exception_and_no_secret(self):
        h = RoleRunHandle("det", 1)
        h.stop()
        self.assertIn("stop_no_pipeline", h.diagnostic_snapshot()["events"])
        def fail():
            raise ValueError("private credential")
        h = RoleRunHandle("det", 1)
        h.bind_pipeline(SimpleNamespace(stop=fail))
        failure = h.stop()
        self.assertIs(failure, h.stop())
        self.assertIn("stop_raised", h.diagnostic_snapshot()["events"])
        self.assertNotIn("private", json.dumps(h.diagnostic_snapshot()))

    def test_status_nonblocking_and_late_exit_with_sticky_error(self):
        from app.stream_manager import StreamGroup
        group = StreamGroup({"stream_id": "test-only", "store": False})
        gen = group.lifecycle.begin()
        group.lifecycle.configure_roles(gen, ["det"])
        h = RoleRunHandle("det", gen)
        group.role_handles["det"] = h
        entered, release, finish = threading.Event(), threading.Event(), threading.Event()
        def stop():
            entered.set()
            release.wait(2)
        h.bind_pipeline(SimpleNamespace(start=lambda: None, stop=stop))
        h.start_pipeline()
        worker = threading.Thread(target=lambda: finish.wait(2))
        h.bind_thread(worker)
        worker.start()
        stopper = threading.Thread(target=h.stop)
        stopper.start()
        try:
            self.assertTrue(entered.wait(1))
            read_done = threading.Event()
            status = []
            reader = threading.Thread(target=lambda: (status.append(group.status()), read_done.set()))
            reader.start()
            self.assertTrue(read_done.wait(.5))
            reader.join(1)
            self.assertTrue(status[0]["runtime_observation"][0]["worker_alive"])
            self.assertIn("stop_enter", status[0]["runtime_observation"][0]["events"])
            group.lifecycle.finish_stop(gen, make_stream_failure("STOP_CLEANUP_FAILED", "worker_join", "det"))
        finally:
            release.set()
            finish.set()
            stopper.join(2)
            worker.join(2)
        s = group.status()
        self.assertFalse(s["runtime_observation"][0]["worker_alive"])
        self.assertEqual("STOP_CLEANUP_FAILED", s["cleanup_error"]["error_code"])

    def test_diagnostic_counter_after_stop_does_not_emit_stale_result(self):
        from app.multi_pipeline import _DecodedFrameObserver, _make_retriever
        events = []
        h = RoleRunHandle("det", 1, lambda *args: events.append(args))
        h.stop()
        _DecodedFrameObserver(h).handle_buffer(None)
        R = _make_retriever("det", "", "", None, 0, {"n": 0}, role_handle=h)
        self.assertEqual(0, R().consume(None))
        s = h.diagnostic_snapshot()
        self.assertEqual(1, s["counters"]["consume_entered"]["count"])
        self.assertEqual(1, s["counters"]["decoded_frame_received"]["count"])
        self.assertFalse(events)

    def test_disabled_status_and_observation_failure(self):
        from app.stream_manager import StreamGroup
        with patch.dict(os.environ, {"CONTEXTRA_START_DIAGNOSTICS_DIR": ""}):
            group = StreamGroup({"stream_id": "test-only", "store": False})
            group.role_handles["det"] = RoleRunHandle("det", 1)
            self.assertNotIn("runtime_observation", group.status())
        h = RoleRunHandle("det", 1)
        calls = []
        h.bind_pipeline(SimpleNamespace(stop=lambda: calls.append(1)))
        with patch("app.start_diagnostics.time.monotonic_ns", side_effect=OSError("private")):
            self.assertIsNone(h.stop())
        self.assertEqual([1], calls)
        self.assertEqual("OBSERVATION_FAILED", h.diagnostic_snapshot()["diagnostic_error"])


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