from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import tempfile
import unittest
import threading
from unittest.mock import patch

from app.contracts.cameras import CameraStart
from app.stream_config_store import ConfigPersistenceError, StreamConfigStore
from app.stream_manager import StreamManager
from app.stream_lifecycle import StreamLifecycle


class FakeGroup:
    def __init__(self, spec, *, metadata_publisher=None):
        self.spec = dict(spec)
        self.metadata_publisher = metadata_publisher
        self.stream_id = spec["stream_id"]
        self.lifecycle = StreamLifecycle()
        self.starts = 0
        self.stops = 0
        self._teardown_lock = threading.RLock()

    def check_restart(self, candidate):
        pass

    def restart(self, candidate):
        self.stop()
        if self.observed_state == "stopped":
            self.spec = candidate
            self.start()

    @property
    def state(self):
        return self.lifecycle.snapshot()["state"]

    observed_state = state

    def start(self):
        if self.state in ("starting", "running"):
            return
        self.starts += 1
        self.lifecycle.begin()

    def stop(self):
        self.stops += 1
        self.lifecycle.finish_stop(self.lifecycle.request_stop())


class StreamApplyTest(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.manager = StreamManager(base_path=self.directory.name + "/base.yaml",
                                     overlay_path=self.directory.name + "/overlay.yaml")
        self.patch = patch("app.stream_manager.StreamGroup", FakeGroup)
        self.patch.start()
        self.addCleanup(self.patch.stop)

    def spec(self, camera="camera", suffix="live"):
        return CameraStart(camera, "rtsp://camera.invalid/" + suffix).runtime_spec("server")

    def test_failed_new_save_does_not_register_or_start(self):
        with patch("app.stream_config_store.write_yaml_atomic", side_effect=OSError("sensitive")):
            with self.assertRaises(ConfigPersistenceError) as error:
                self.manager.apply_and_start(self.spec())
        self.assertNotIn("sensitive", str(error.exception))
        self.assertEqual(self.manager._groups, {})

    def test_failed_change_preserves_existing_run(self):
        group = self.manager.apply_and_start(self.spec())
        run = group.lifecycle.snapshot()["run_id"]
        with patch("app.stream_config_store.write_yaml_atomic", side_effect=OSError()):
            with self.assertRaises(ConfigPersistenceError):
                self.manager.apply_and_start(self.spec(suffix="changed"))
        self.assertEqual(group.stops, 0)
        self.assertEqual(group.lifecycle.snapshot()["run_id"], run)
        self.assertEqual(group.spec, self.spec())

    def test_duplicate_start_and_changed_config(self):
        group = self.manager.apply_and_start(self.spec())
        run = group.lifecycle.snapshot()["run_id"]
        self.manager.apply_and_start(self.spec())
        self.assertEqual(group.starts, 1)
        self.manager.apply_and_start(self.spec(suffix="changed"))
        self.assertEqual(group.starts, 2)
        self.assertEqual(group.stops, 1)
        self.assertNotEqual(group.lifecycle.snapshot()["run_id"], run)
        self.assertEqual(self.manager.effective_specs()[0], self.spec(suffix="changed"))

    def test_parallel_camera_updates_do_not_lose_overlay_entries(self):
        with ThreadPoolExecutor(max_workers=4) as pool:
            list(pool.map(lambda n: self.manager.apply_and_start(self.spec(str(n))), range(12)))
        self.assertEqual(len(self.manager.effective_specs()), 12)

    def test_same_camera_concurrent_start_is_once(self):
        with ThreadPoolExecutor(max_workers=4) as pool:
            groups = list(pool.map(lambda _: self.manager.apply_and_start(self.spec()), range(12)))
        self.assertTrue(all(g is groups[0] for g in groups))
        self.assertEqual(groups[0].starts, 1)

    def test_corrupt_overlay_is_not_silently_overwritten(self):
        path = Path(self.manager.overlay_path)
        path.write_text("[bad")
        with self.assertRaises(ConfigPersistenceError):
            self.manager.apply_and_start(self.spec())
        self.assertEqual(path.read_text(), "[bad")

    def test_delete_save_failure_preserves_run(self):
        group = self.manager.apply_and_start(self.spec())
        with patch("app.stream_config_store.write_yaml_atomic", side_effect=OSError()):
            with self.assertRaises(ConfigPersistenceError):
                self.manager.remove(group.stream_id)
        self.assertIs(self.manager.get(group.stream_id), group)
        self.assertEqual(group.stops, 0)

    def test_cleanup_failure_does_not_allocate_new_run(self):
        from app.stream_lifecycle import make_stream_failure
        group = self.manager.apply_and_start(self.spec())
        run = group.lifecycle.snapshot()["run_id"]
        def fail_stop():
            generation = group.lifecycle.request_stop()
            group.lifecycle.finish_stop(generation, make_stream_failure(
                "STOP_CLEANUP_FAILED", "pipeline_stop"))
        group.stop = fail_stop
        result = self.manager.apply_and_start(self.spec(suffix="changed"))
        self.assertEqual(result.observed_state, "error")
        self.assertEqual(result.lifecycle.snapshot()["run_id"], run)
        self.assertEqual(result.starts, 1)

    def test_restore_has_config_but_no_autostart_or_run_uuid(self):
        self.manager.apply_and_start(self.spec())
        restored = StreamManager(base_path=self.manager.base_path,
                                 overlay_path=self.manager.overlay_path)
        restored.load_and_autostart()
        group = restored.get(self.spec()["stream_id"])
        self.assertEqual(group.state, "stopped")
        self.assertIsNone(group.lifecycle.snapshot()["run_id"])