import threading
import tempfile
import unittest
from unittest.mock import patch

from app.analysis_metrics import AnalysisMetrics, PtsSampler, analysis_fps
from app.stream_settings import patch_settings, settings_view


class SamplingTest(unittest.TestCase):
    def test_15fps_selects_10_without_renumbering(self):
        sampler = PtsSampler(10)
        chosen = [i for i in range(150) if sampler.select(round(i * 1e9 / 15))[0]]
        self.assertEqual(len(chosen), 100)
        self.assertEqual(chosen[:5], [0, 2, 3, 5, 6])

    def test_irregular_duplicate_null_and_reset(self):
        sampler = PtsSampler()
        self.assertEqual(sampler.select(None), (False, 'invalid_pts'))
        self.assertTrue(sampler.select(0)[0])
        self.assertEqual(sampler.select(0), (False, 'duplicate_pts'))
        self.assertFalse(sampler.select(70_000_000)[0])
        self.assertTrue(sampler.select(320_000_000)[0])
        self.assertFalse(sampler.select(350_000_000)[0])
        self.assertEqual(sampler.select(5), (True, 'pts_reset'))
        self.assertTrue(PtsSampler().select(0)[0])

    def test_validation_and_noop(self):
        spec = dict(server_id='server', source_id='source')
        self.assertEqual(patch_settings(spec, {'analysisFpsTarget': 10}), spec)
        changed = patch_settings(spec, {'analysisFpsTarget': 15})
        self.assertEqual(settings_view(changed)['analysisFpsTarget'], 15)
        self.assertNotIn('analysis_fps_target', spec)
        for bad in (True, None, 0, 61, float('nan'), float('inf'), '10'):
            with self.assertRaises(ValueError): analysis_fps(bad)

    def test_settings_save_failure_and_generation(self):
        from app.stream_manager import StreamManager
        from app.stream_config_store import ConfigPersistenceError
        with tempfile.TemporaryDirectory() as root:
            m = StreamManager(root+'/base', root+'/overlay')
            g = m.add(dict(stream_id='s', server_id='server', source_id='camera', uri='private', store=False))
            rev = m.settings('s')['configured']['revision']
            with patch.object(m.config_store, 'update', side_effect=ConfigPersistenceError()):
                with self.assertRaises(ConfigPersistenceError):
                    m.patch_settings('s', rev, {'analysisFpsTarget': 15})
            self.assertEqual(m.settings('s')['effective']['analysisFpsTarget'], 10)
            m.patch_settings('s', rev, {'analysisFpsTarget': 15})
            self.assertEqual(m.settings('s')['effective']['analysisFpsTarget'], 15)
            with self.assertRaises(ValueError): m.patch_settings('s', rev, {'analysisFpsTarget': 20})

    def test_metrics_endpoint_fences_generation(self):
        from types import SimpleNamespace
        from fastapi import HTTPException
        from app.api.streams import MetricsWindow, metrics_window, stream_service
        metrics = AnalysisMetrics()
        group = SimpleNamespace(_teardown_lock=threading.RLock(), state='running',
            result_path=SimpleNamespace(ingress=SimpleNamespace(identity=('s','c','t','run',2),metrics=metrics)))
        with patch.object(stream_service, 'get', return_value=group):
            with self.assertRaises(HTTPException):
                metrics_window('test', MetricsWindow(run_generation=1, enabled=True))
            self.assertFalse(metrics.snapshot()['detail'])
            self.assertTrue(metrics_window('test', MetricsWindow(run_generation=2, enabled=True))['detail'])
            group.state = 'stopping'
            with self.assertRaises(HTTPException):
                metrics_window('test', MetricsWindow(run_generation=2, enabled=True))


class MetricsTest(unittest.TestCase):
    def setUp(self):
        self.now = 0
        self.m = AnalysisMetrics(clock=lambda: self.now)

    def test_exact_frame_and_timeout(self):
        self.m.configure(True, 1)
        self.m.mark('selected', (0, 123))
        self.now = 2_000_000
        self.m.mark('preprocess', (1, 123))
        self.m.mark('preprocess', (0, 123))
        self.assertEqual(self.m.snapshot()['durations']['selected_to_preprocess']['mean_ms'], 2)
        self.now = 1_000_000_001
        self.m.mark('infer', (0, 123))
        s = self.m.snapshot()
        self.assertFalse(s['detail'])
        self.assertEqual(s['pending_samples'], 0)
        self.assertNotIn('preprocess_to_infer', s['durations'])

    def test_no_wait_and_bounded_memory_close(self):
        self.m.configure(True)
        self.m._lock.acquire()
        t = threading.Thread(target=lambda: self.m.mark('decoded'))
        t.start(); t.join(.2)
        self.assertFalse(t.is_alive())
        self.m._lock.release()
        self.assertEqual(self.m.snapshot()['observation_missed'], 1)
        for i in range(1000):
            self.m.mark('selected', (i, i))
            self.m.duration('tracking', i)
        self.assertEqual(self.m.snapshot()['pending_samples'], 256)
        self.assertEqual(self.m.snapshot()['durations']['tracking']['samples'], 256)
        self.m.close()
        self.assertEqual(self.m.snapshot()['pending_samples'], 0)
        with self.assertRaises(RuntimeError): self.m.configure(True)

    def test_off_still_counts_and_on_does_not_reset_counts(self):
        self.m.mark('decoded')
        self.m.duration('tracking', 100)
        self.assertFalse(self.m.snapshot()['durations'])
        self.m.configure(True)
        self.m.mark('decoded')
        self.assertEqual(self.m.snapshot()['counts']['decoded'], 2)
        self.m.configure(False)
        self.m.duration('tracking', 100)
        self.assertFalse(self.m.snapshot()['durations'])