"""Real result-worker threads, fake tracker/capture/publisher; no SDK or Redis I/O."""
from dataclasses import dataclass
from types import SimpleNamespace
import json
import time
import unittest
from unittest.mock import Mock, patch
from app.det_result_path import DetResultPath
from app.metadata_samples import MetadataSamples
from app.loitering import Loitering, LoiteringConfig
from app.intrusion import Intrusion, IntrusionConfig
from loitering_contract_helpers import frame, WALL_MS


@dataclass(frozen=True)
class CopiedDetection:
    class_id: int = 0
    score: float = .9
    bbox_xyxy: tuple = (.1,.2,.3,.8)
    keypoints: object = None
    def downstream(self):
        return dict(class_id=self.class_id,score=self.score,bbox=list(self.bbox_xyxy))


class Sink:
    def __init__(self):self.rows=[]
    def channel(self,*args):return True
    def submit(self,snapshot):self.rows.append(json.loads(json.dumps(snapshot)));return True
    def wait_channel(self,*args):return True


class ResultWorkerTests(unittest.TestCase):
    def make(self, engine=None, source='77', tracker=True):
        self.now=[0.]
        ingress=SimpleNamespace(identity=('server',source,'stream','run',1))
        tracking=Mock(update=lambda observations:list(range(1,len(observations)+1))) if tracker else None
        sink=Sink();errors=Mock();collector=Mock(add=Mock(return_value=1))
        with patch.object(MetadataSamples,'from_environment',return_value=None):
            path=DetResultPath(ingress,tracking,'RFDETRMedium',.5,collector,lambda:True,errors,
                 publisher=sink,intrusion=engine,scenario_revision='a'*64,camera_id=int(source),clock=lambda:self.now[0])
        path.start();self.addCleanup(lambda:path.close(2))
        return path,sink,errors

    def feed(self,path,i,seconds,observations=None):
        f=frame(i,seconds,source=path.ingress.identity[1])
        copied=SimpleNamespace(width=640,height=360,frame_number=i,pts=f.pts_ns,
            detections=tuple([CopiedDetection()] if observations is None else observations))
        previous=path.status().get('processed',0)
        self.assertTrue(path.submit(f,copied))
        end=time.monotonic()+2
        while path.status().get('processed',0)==previous and not path.status()['error'] and time.monotonic()<end:
            time.sleep(.001)
        self.assertIsNone(path.status()['error'])
        self.assertEqual(path.status().get('processed',0),previous+1)
        return path.latest_result()

    def engine(self, **values):
        return Loitering(LoiteringConfig(enabled=True,**values),wall_ns=lambda:WALL_MS*1_000_000)

    def test_five_seconds_to_same_frame_event_and_sink(self):
        engine=self.engine();path,sink,errors=self.make(engine)
        for i in range(51):
            result=self.feed(path,i,i/10)
            self.assertEqual(len(result.events),int(i==50))
        self.assertEqual(result.events[0].frame,result.frame)
        self.assertEqual(result.events[0].frame.frame_seq,50)
        self.assertEqual(result.events[0].ai_recorded_at_wall_ms,WALL_MS)
        self.assertEqual(result.commands[0].color,'#FF3030')
        self.assertEqual(sink.rows[-1],result.snapshot())
        self.assertEqual(sink.rows[-1]['events'][0]['eventId'],sink.rows[-1]['objects'][0]['eventIds'][0])
        self.assertFalse(self.feed(path,51,5.1).events)
        errors.assert_not_called()

    def test_inclusive_score_half_is_visible_and_triggers(self):
        path,_,_=self.make(self.engine())
        for i in range(51):last=self.feed(path,i,i/10,[CopiedDetection(score=.5)])
        self.assertTrue(last.commands[0].visible);self.assertEqual(len(last.events),1)
        self.assertEqual(last.commands[0].color,'#FF3030')

    def test_intrusion_keeps_strict_comparison(self):
        path,_,_=self.make(Intrusion(IntrusionConfig(enabled=True)))
        last=self.feed(path,0,0,[CopiedDetection(score=.5)])
        self.assertFalse(last.commands[0].visible)
        self.assertEqual(last.commands[0].scenario.state,'LOW_CONFIDENCE')

    def test_intrusion_still_triggers_after_half_second(self):
        path,_,_=self.make(Intrusion(IntrusionConfig(enabled=True)))
        for i in range(6):last=self.feed(path,i,i/10)
        self.assertEqual(len(last.events),1)
        self.assertEqual(last.commands[0].scenario.state,'INTRUSION')
        self.assertEqual(last.commands[0].color,'#FF3030')

    def test_explicit_miss_freezes_both_missing_edges(self):
        engine=self.engine();path,_,_=self.make(engine)
        self.feed(path,0,0);self.feed(path,1,.2);self.feed(path,2,.3,[]);self.feed(path,3,.4)
        self.assertEqual(engine.states[1].dwell_ns,200_000_000)

    def test_no_id_has_no_state(self):
        engine=self.engine();path,_,_=self.make(engine)
        path.tracker.update=lambda obs:[None]*len(obs)
        last=self.feed(path,0,0)
        self.assertEqual(last.commands[0].scenario.state,'PENDING_ID');self.assertFalse(engine.states)

    def test_nonperson_does_not_acquire_scenario_and_object_ref_is_preserved(self):
        path,_,_=self.make(self.engine())
        detections=[CopiedDetection(class_id=2),CopiedDetection()]
        for i in range(51):last=self.feed(path,i,i/10,detections)
        self.assertIsNone(last.commands[0].scenario)
        self.assertEqual(last.events[0].object_ref,'1')
        self.assertEqual(last.events[0].track_id,2)

    def test_capture_receives_same_result_without_engine_file_io(self):
        path,_,_=self.make(self.engine())
        capture=Mock(close=Mock(return_value=True));path.artifacts=capture
        last=self.feed(path,0,0)
        self.assertIs(capture.submit.call_args.args[0],last)

    def test_off_has_no_new_optional_metadata(self):
        path,sink,_=self.make()
        self.feed(path,0,0)
        self.assertNotIn('scenarioConfig',sink.rows[-1])
        self.assertNotIn('events',sink.rows[-1])
        self.assertNotIn('scenarioResults',sink.rows[-1]['objects'][0])

    def test_clear_occurs_after_join(self):
        engine=self.engine();path,_,_=self.make(engine)
        self.feed(path,0,0);self.assertTrue(engine.states)
        path.close(2)
        self.assertFalse(path._thread.is_alive());self.assertFalse(engine.states)
        self.assertIsNone(path.latest_result())
        path.tracker.reset.assert_called_once()

    def test_streams_do_not_share_visit_state(self):
        a=self.engine();b=self.engine();pa,_,_=self.make(a,'77');pb,_,_=self.make(b,'78')
        self.feed(pa,0,0);self.feed(pa,1,.2);self.feed(pb,0,0)
        self.assertEqual(a.states[1].dwell_ns,200_000_000)
        self.assertEqual(b.states[1].dwell_ns,0)
        self.assertNotEqual(a.identity,b.identity)

    def test_idle_retention_on_existing_worker(self):
        now=[0.];engine=Loitering(LoiteringConfig(enabled=True),clock=lambda:now[0])
        path,_,_=self.make(engine);self.feed(path,0,0);now[0]=61.
        end=time.monotonic()+2
        while engine.states and time.monotonic()<end:time.sleep(.01)
        self.assertFalse(engine.states)

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