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

from app.input_timing import read
from app.pipeline_diagnostics import PipelineDiagnostics
from app.channel_child import send_latest


class InputTimingTest(unittest.TestCase):
    def test_private_metadata_validation_and_missing(self):
        row = dict(rtp=17, ssrc=1, rate=90000, depay_first_ns=10,
                   depay_last_ns=12, source_exit_ns=20, source_wall_ns=100)
        frame = SimpleNamespace(user_meta_items=lambda _: [SimpleNamespace(get_user_data_json=lambda: row)])
        with patch('app.input_timing.meta_type', return_value=1):
            self.assertEqual(read(frame), row)
            row['source_exit_ns'] = 11
            self.assertIsNone(read(frame))
            frame.user_meta_items = lambda _: []
            self.assertIsNone(read(frame))

    def test_timeline_is_bounded_and_off_is_empty(self):
        metric = PipelineDiagnostics(True)
        for seq in range(400):
            metric.mark('selected', (seq, seq), 30)
            metric.input_timing((seq, seq), dict(depay_first_ns=10, source_exit_ns=20), 30)
        snap = metric.snapshot()
        self.assertEqual(len(snap['input_timeline']), 8)
        self.assertEqual(snap['durations']['source_exit_to_ingress']['count'], 400)
        metric.queue_sample({'current-level-buffers': 3})
        self.assertEqual(metric.snapshot()['queue_levels']['current-level-buffers']['maximum'], 3)
        metric.configure(False)
        self.assertEqual(metric.snapshot(), {'enabled': False})

    def test_extended_timing_cannot_exhaust_status_ipc_budget(self):
        metric = PipelineDiagnostics(True)
        for seq in range(64):
            key = (seq, seq * 100_000_000)
            metric.mark('selected', key, 10**18)
            metric.input_timing(key, dict(depay_first_ns=10**18, source_exit_ns=10**18,
                                         **{f'counter_{i}': 10**18 for i in range(40)}), 10**18)
        diagnostic = metric.snapshot()
        self.assertLessEqual(len(json.dumps(diagnostic['input_timeline']).encode()), 8192)
        self.assertTrue(diagnostic['input_timeline'])
        # Leave ample room for existing lifecycle, provider and publisher status.
        message = dict(kind='status', value=dict(existing='x' * 48_000, diagnostics=diagnostic))
        output = queue.Queue(1)
        self.assertTrue(send_latest(output, message))
        self.assertEqual(output.get_nowait(), message)

    def test_small_export_keeps_aggregate_counts_and_bounds(self):
        with patch.dict(os.environ, CONTEXTRA_INPUT_TIMELINE_LIMIT='2'):
            metric = PipelineDiagnostics(True)
        for seq in range(100):
            metric.mark('selected', (seq, seq), 30)
            metric.input_timing((seq, seq), dict(depay_first_ns=10, source_exit_ns=20), 30)
        snap = metric.snapshot()
        self.assertEqual(len(snap['input_timeline']), 2)
        self.assertEqual(snap['input_timeline_limit'], 2)
        self.assertEqual(snap['durations']['source_exit_to_ingress']['count'], 100)
        self.assertEqual(snap['counts']['selected'], 100)
        for value in ('0', '9', 'true'):
            with patch.dict(os.environ, CONTEXTRA_INPUT_TIMELINE_LIMIT=value):
                with self.assertRaises(ValueError):
                    PipelineDiagnostics(True)


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