import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
from types import SimpleNamespace

from tests.test_pr13_multi_provider import CoordinatorAcceptanceTest, frame, _Copied
from app.single_decode_pipeline import capture_size
from app.medium_preprocess import medium_preprocess_config
from app.pipeline_diagnostics import PipelineDiagnostics


class FreshnessTest(unittest.TestCase):
    def test_callback_gate_owns_completion_release(self):
        import ast, inspect, threading
        from app.single_decode_pipeline import run_single_decode_pipeline
        tree = ast.parse(inspect.getsource(run_single_decode_pipeline))
        gate = next(n for n in ast.walk(tree) if isinstance(n, ast.ClassDef) and n.name=='ProviderGate')
        namespace = {'BufferOperator':object, 'threading':threading}
        exec(compile(ast.Module(body=[gate], type_ignores=[]), '<gate-fixture>', 'exec'), namespace)
        calls=[]
        g=namespace['ProviderGate']('pose'); g._seq=7
        g.relay=SimpleNamespace(set=calls.append)
        g.complete(6); self.assertEqual(g._seq,7)
        g.complete(7); self.assertIsNone(g._seq)
        self.assertEqual(calls,[{'complete':True}])

    def make(self, enabled=False):
        self.now = [0.]
        with patch.dict(os.environ, CONTEXTRA_PROVIDER_FRESHNESS_ENABLED=str(enabled).lower(),
                        CONTEXTRA_PROVIDER_STARTUP_SECONDS='10'):
            return CoordinatorAcceptanceTest().make(lambda: self.now[0])

    def test_off_retains_inflight_but_bounded_pending_and_counts_old_response(self):
        c = self.make()
        c.admit(frame(0))
        self.assertEqual(c.provider_gate_time('medium', 0, 0), 0)
        for i in range(1, 30):
            self.now[0] = i/10
            c.admit(frame(i, i*100_000_000))
        self.assertEqual(len(c._frames), 3)
        self.assertIn(0, c._frames)
        self.assertTrue(c.is_open(0))
        self.assertIsNone(c._ready_locked(self.now[0]))
        self.assertTrue(c.submit('medium', frame(0), _Copied()))
        self.assertEqual(c._ready_locked(self.now[0]), 0)
        self.assertEqual(c._metrics['over_freshness_medium'], 1)
        self.assertFalse(c.submit('medium', frame(0), _Copied()))
        c.stop_admission()
        self.assertFalse(c.submit('medium', frame(29), _Copied()))

    def test_on_retains_deadline_and_atomic_gate_recheck(self):
        c = self.make(True)
        c.admit(frame(0))
        self.now[0] = .2
        self.assertIsNone(c.provider_gate_time('medium', 0, .2))
        self.assertFalse(c.is_open(0))
        self.assertFalse(c.submit('medium', frame(0), _Copied()))

    def test_startup_grace_and_steady_stall_are_distinct(self):
        c = self.make()
        c.admit(frame(0))
        for i in range(1, 10):
            self.now[0] = i
            c.admit(frame(i))
            self.assertEqual(c._check_stalls_locked(i), [])
        self.now[0] = 10
        c.admit(frame(10))
        self.assertEqual(c._check_stalls_locked(10), ['medium'])
        c._providers['medium']['last_response_at'] = 10.
        self.now[0] = 12
        c.admit(frame(12))
        self.assertEqual(c._check_stalls_locked(12), ['medium'])

    def test_capture_dimensions_are_model_owned_not_source_owned(self):
        with tempfile.TemporaryDirectory() as d:
            path = Path(d)/'model'
            for size in (512, 576):
                Path(str(path)+'.preprocess').write_text(medium_preprocess_config(1, resolution=size))
                self.assertEqual(capture_size([SimpleNamespace(key='pose')], {'pose': path}), (size,size))
            self.assertIsNone(capture_size([1,2], {}))

    def test_stage_metrics_do_not_double_count_old_total(self):
        d = PipelineDiagnostics(); d.configure(True, 30)
        for stage, ms in [('selected',0), ('preprocess_start',20), ('preprocess',30),
                          ('infer_output',60), ('callback',70)]:
            d.mark(stage, (1,1), ms*1_000_000)
        self.assertEqual(d.durations['selected_to_preprocess']['sum_ns'], 30_000_000)
        self.assertEqual(d.durations['preprocess_start_to_preprocess']['sum_ns'], 10_000_000)
        self.assertEqual(d.durations['infer_output_to_callback']['sum_ns'], 10_000_000)

    def test_off_finalizing_stop_still_completes_handoff(self):
        from tests.test_pr13_boundaries import Boundaries
        with patch.dict(os.environ, CONTEXTRA_PROVIDER_FRESHNESS_ENABLED='false'):
            Boundaries().test_stop_drains_claimed_event_before_close()

    def test_ipc_timing_is_removed_before_public_publisher(self):
        import json, queue, threading, time
        from app.provider_process_group import ProviderProcessGroup
        sent=[]; measured=[]; q=queue.Queue()
        q.put(json.dumps(dict(cameraId=3,runId='run',runGeneration=1,providerKey='pose',
                             _ipc_ready_ns=time.monotonic_ns())))
        g=SimpleNamespace(_snapshots={'pose':q},spec={'source_id':'3'},_split_stop=threading.Event(),
            _split_publisher=SimpleNamespace(channel=lambda *a:True,submit=sent.append,
                                            note_relay=lambda *a:measured.append(a)))
        ProviderProcessGroup._drain(g,'run',1)
        self.assertNotIn('_ipc_ready_ns',sent[0])
        self.assertEqual(measured[0][0],'pose')
        self.assertGreaterEqual(measured[0][1],0)


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