import ast
from pathlib import Path
import threading
import time
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch

from app.frame_ingress import FrameIngress
from app import frame_ingress
from app.stream_lifecycle import StreamLifecycle, make_stream_failure


class PoseBatchPathTest(unittest.TestCase):
    def test_sampling_preserves_decoded_sequence_clock_and_reset(self):
        ingress = FrameIngress('server', 'camera', 'stream', 'run', 1)
        ingress.rtp_clock = Mock()
        ingress.rtp_clock.resolve.return_value = None
        captured = []
        selected = []
        with patch('app.frame_ingress.identity_meta_type', return_value=1), \
             patch('app.frame_ingress.rtp_meta_type', return_value=2), \
             patch('app.frame_ingress.append_identity', side_effect=lambda b,f,v: captured.append(v)):
            for number, pts in enumerate((0, 50_000_000, 100_000_000, 50_000_000)):
                frame = SimpleNamespace(buffer_pts=pts, frame_number=number,
                    source_width=1280, source_height=720, user_meta_items=lambda kind: iter(()))
                selected.append(ingress.stamp(SimpleNamespace(frame_items=[frame])))
        self.assertEqual([True, False, True, True], selected)
        self.assertEqual([0, 2, 3], [x['frame']['frame_seq'] for x in captured])
        self.assertEqual(4, ingress.rtp_clock.resolve.call_count)
        self.assertEqual(1, ingress.metrics.snapshot()['counts']['pts_reset'])
        self.assertEqual(1280, captured[0]['frame']['frame_width'])

    def test_active_fence_matches_snapshot_without_copy(self):
        lifecycle = StreamLifecycle()
        generation = lifecycle.begin()
        self.assertTrue(lifecycle.is_active(generation))
        self.assertFalse(lifecycle.is_active(generation - 1))
        lifecycle.signal_error(generation, make_stream_failure('ROLE_RUNTIME_FAILED','pipeline_runtime','pose'))
        self.assertFalse(lifecycle.is_active(generation))
        lifecycle.finish_stop(generation)
        self.assertFalse(lifecycle.is_active(generation))

    def test_collector_pruning_matches_stable_sort_with_reordering(self):
        source = Path(frame_ingress.__file__).resolve().parent / 'multi_pipeline.py'
        node = next(n for n in ast.parse(source.read_text()).body if isinstance(n, ast.ClassDef) and n.name=='ResultCollector')
        scope = {'threading': threading, 'time': time, 'PTS_TOL_NS': 1}
        exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), 'exec'), scope)
        collector = scope['ResultCollector'](['pose'])
        reference = {}
        for fn in (5, 6, 2, 8, 7, 9, 1, 10):
            reference.setdefault(fn, {'_fn':fn})
            if len(reference)>3:
                for key,_ in sorted(reference.items(), key=lambda x:x[1]['_fn'])[:len(reference)-3]:
                    reference.pop(key)
            collector.add('pose', fn, fn*100, {}, max_frames=3)
            self.assertEqual(list(reference), list(collector.frames))
        collector.add('pose', 11, 1100, {}, max_frames=1)
        self.assertEqual([11], list(collector.frames))


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