import threading
import unittest
from dataclasses import replace
from unittest.mock import Mock

from app.contracts.frames import DetectionObservation
from app.det_result_path import DetResultPath
from app.frame_ingress import FrameIngress
from app.intrusion import Intrusion, IntrusionConfig, _inside_edges, inside
from app.native_metadata import NativeFrame, NativeDetection
from test_intrusion import frame
from test_metadata_connection import wait_for


def result_path(engine, **kwargs):
    return DetResultPath(FrameIngress('server', 'source', 'stream', 'run', 1),
        Mock(update=lambda obs: [1] * len(obs)), 'RFDETRMedium', .5, Mock(),
        lambda: True, Mock(), intrusion=engine, **kwargs)


def submit(path, seq=0):
    f = frame(seq, seq * .1)
    return path.submit(f, NativeFrame(seq, f.pts_ns, 640, 360,
        (NativeDetection(0, .9, (.1, .1, .3, .8)),), 640, 360))


class IntrusionPreparationTest(unittest.TestCase):
    def test_expiry_exact_boundary_and_live_state_identity(self):
        now = [0.]
        engine = Intrusion(IntrusionConfig(enabled=True), clock=lambda: now[0])
        f = frame(0, 0)
        obs = DetectionObservation(f, 'RFDETRMedium', 0, .9, (.1, .1, .3, .8))
        engine.evaluate(f, (obs,), (1,))
        retained = engine.states[1]
        now[0] = 29.999
        engine.expire()
        self.assertIs(engine.states[1], retained)
        now[0] = 30.
        engine.expire()
        self.assertFalse(engine.states)
        f = frame(1, .1)
        decisions, events = engine.evaluate(f, (replace(obs, frame=f),), (1,))
        self.assertEqual(decisions[0].state, 'CONFIRMING')
        self.assertFalse(events)

    def test_prepared_roi_equivalence_and_independent_new_config(self):
        cfg = IntrusionConfig(enabled=True, roi=((0., 0.), (.8, 0.), (.6, .6), (.2, .9)))
        engine = Intrusion(cfg)
        before = cfg.payload()
        edges = engine._roi_edges
        points = [(x/10, y/10) for x in range(-1, 12) for y in range(-1, 12)] + list(cfg.roi)
        for point in points:
            self.assertEqual(_inside_edges(point, edges), inside(point, cfg.roi))
        engine.clear()
        self.assertIs(engine._roi_edges, edges)
        self.assertEqual(cfg.payload(), before)
        other = Intrusion(replace(cfg, roi=((.8, .8), (1., .8), (1., 1.), (.8, 1.))))
        self.assertNotEqual(engine._roi_edges, other._roi_edges)
        self.assertNotEqual(_inside_edges((.1, .1), engine._roi_edges),
                            _inside_edges((.1, .1), other._roi_edges))

    def test_normal_frame_expires_once(self):
        engine = Intrusion(IntrusionConfig(enabled=True))
        engine.expire = Mock(wraps=engine.expire)
        path = result_path(engine)
        self.addCleanup(lambda: path.close(1))
        submit(path)
        path.start()
        wait_for(lambda: path.status().get('processed') == 1)
        self.assertEqual(engine.expire.call_count, 1)

    def test_stale_frame_keeps_expiry(self):
        now = [0.]
        engine = Intrusion(IntrusionConfig(enabled=True))
        engine.expire = Mock(wraps=engine.expire)
        path = result_path(engine, clock=lambda: now[0])
        self.addCleanup(lambda: path.close(1))
        submit(path)
        now[0] = 10.
        path.start()
        wait_for(lambda: path.status().get('stale') == 1)
        self.assertEqual(engine.expire.call_count, 1)

    def test_idle_expiry_releases_queue_lock_and_close_waits_before_clear(self):
        entered, release, submitted, cleared, closed = (threading.Event() for _ in range(5))
        errors = []
        class PausedIntrusion(Intrusion):
            def expire(self):
                entered.set()
                if not release.wait(3):
                    raise RuntimeError('Test release deadline')
                super().expire()
            def clear(self):
                cleared.set()
                super().clear()
        engine = PausedIntrusion(IntrusionConfig(enabled=True))
        path = result_path(engine)
        def produce():
            try: submit(path)
            except Exception as error: errors.append(error)
            finally: submitted.set()
        def stop():
            try: path.close(2)
            except Exception as error: errors.append(error)
            finally: closed.set()
        producer = threading.Thread(target=produce)
        stopper = threading.Thread(target=stop)
        path.start()
        try:
            self.assertTrue(entered.wait(1))
            producer.start()
            self.assertTrue(submitted.wait(.5), 'Expiration blocked the shared queue')
            stopper.start()
            self.assertFalse(closed.wait(.05))
            self.assertFalse(cleared.is_set())
        finally:
            release.set()
            if producer.ident is not None: producer.join(1)
            if stopper.ident is not None: stopper.join(3)
            path.close(1)
        self.assertFalse(errors)
        self.assertTrue(cleared.is_set())
        self.assertFalse(path._thread.is_alive())
        self.assertIsNone(path.latest_result())
        self.assertEqual(path.status()['queued'], 0)