import io
import tempfile
import unittest
from pathlib import Path

from app.pipeline_diagnostics import PipelineDiagnostics
from app.artifact_diagnostics import StderrTail, RenderDiagnostic, save_diagnostic


class ComparisonDiagnosticsTest(unittest.TestCase):
    def test_bounded_frames_and_stage_intervals(self):
        p = PipelineDiagnostics(True)
        for i in range(300):
            p.mark('selected', (i, i), 0)
        self.assertEqual(len(p.frames), 256)
        p.mark('preprocess', (299, 299), 10_000_000)
        p.mark('callback', (299, 299), 40_000_000)
        p.mark('copied', (299, 299), 60_000_000)
        p.copy_duration(20_000_000, 12_000_000)
        d = p.snapshot()['durations']
        self.assertEqual(d['preprocess_to_callback']['sum_ns'], 30_000_000)
        self.assertEqual(d['callback_to_copied']['sum_ns'], 20_000_000)
        self.assertEqual(d['copy_thread_cpu']['sum_ns'], 12_000_000)

    def test_disabled_has_no_frame_retention(self):
        p = PipelineDiagnostics()
        p.mark('selected', (1, 1))
        self.assertEqual(p.snapshot(), {'enabled': False})
        self.assertFalse(p.frames)

    def test_provider_frames_do_not_overwrite_each_other(self):
        p = PipelineDiagnostics(True)
        pose, fire = p.for_provider('pose'), p.for_provider('fire')
        for metric in (pose, fire):
            metric.mark('selected', (1, 10), 0)
        pose.mark('preprocess', (1, 10), 10)
        fire.mark('preprocess', (1, 10), 20)
        pose.mark('callback', (1, 10), 30)
        fire.mark('callback', (1, 10), 80)
        values = p.snapshot()['providers']
        self.assertEqual(values['pose']['durations']['preprocess_to_callback']['sum_ns'], 20)
        self.assertEqual(values['fire']['durations']['preprocess_to_callback']['sum_ns'], 60)
        with self.assertRaises(ValueError):
            p.for_provider('unbounded-label')

    def test_copy_scope_is_thread_local_and_restored_on_error(self):
        from app import pipeline_diagnostics as module
        import threading
        p = PipelineDiagnostics(True)
        seen = []
        with p.copy_scope():
            thread = threading.Thread(target=lambda: seen.append(getattr(module._copy_context, 'value', None)))
            thread.start(); thread.join()
            self.assertIs(module._copy_context.value, p)
            with self.assertRaises(RuntimeError):
                with p.for_provider('fire').copy_scope():
                    raise RuntimeError('fixture')
            self.assertIs(module._copy_context.value, p)
        self.assertEqual(seen, [None])
        self.assertIsNone(module._copy_context.value)

    def test_timed_copy_uses_provider_scope(self):
        from app import pipeline_diagnostics as module
        from types import SimpleNamespace
        from unittest.mock import patch
        p = PipelineDiagnostics(True)
        with patch.object(module, 'diagnostics', p):
            copied = module.timed_copy(lambda: SimpleNamespace(frame_number=1, pts=10))
            with p.for_provider('crowd').copy_scope():
                copied()
        self.assertNotIn('copied', p.snapshot()['counts'])
        crowd = p.snapshot()['providers']['crowd']
        self.assertEqual(crowd['counts']['copied'], 1)
        self.assertEqual(crowd['durations']['copy_thread_cpu']['count'], 1)

    def test_stderr_is_bounded_and_redacted(self):
        reader = StderrTail(io.BytesIO(b'x'*10000+b'\nrtsp://user:pw@host/source\nAuthorization: secret\nERROR\n'), 512)
        text = reader.finish()
        self.assertLessEqual(len(text), 512)
        self.assertNotIn('user:pw', text)
        self.assertNotIn('Authorization:', text)
        self.assertIn('ERROR', text)

    def test_child_and_parent_diagnostic_merge_in_ready(self):
        import json
        with tempfile.TemporaryDirectory() as root:
            stage = Path(root)
            (stage/'ready').mkdir()
            d = RenderDiagnostic()
            d.enter('encode_eos')
            save_diagnostic(stage, dict(**d.snapshot(), result='failed'))
            save_diagnostic(stage, {'returncode': 1})
            data = json.loads((stage/'ready'/'render-diagnostic.json').read_text())
            self.assertEqual(data['phase'], 'encode_eos')
            self.assertEqual(data['returncode'], 1)
            self.assertFalse((stage/'render-diagnostic.json').exists())


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