from dataclasses import replace
import json
import os
import threading
import unittest
from uuid import uuid4

from app.metadata_publisher import MetadataSettings, camera_key
from app.process_metadata import FencedSink, StaleWriter, ChildPublisher, ProcessMetadataControl, CHANNEL_BYTES


@unittest.skipUnless(os.environ.get('PROCESS_TEST_REDIS_SOCKET'),'Isolated Redis fixture socket required.')
class FenceTest(unittest.TestCase):
    def setUp(self):
        import redis
        self.settings=MetadataSettings('unused',namespace='fixture:'+uuid4().hex,
            max_bytes=256*1024**2,max_message_bytes=10*1024**2,stream_length=100,stream_ttl=10)
        self.clients=[]
        self.addCleanup(lambda:[c.close() for c in self.clients])

    def sink(self, settings, token):
        import redis
        client=redis.Redis(unix_socket_path=os.environ['PROCESS_TEST_REDIS_SOCKET'],decode_responses=True)
        self.clients.append(client)
        return FencedSink(settings,token,client)

    def test_stale_frame_and_stopped_cannot_delete_new_run(self):
        key=camera_key('camera');a=self.sink(self.settings,'a');b=self.sink(self.settings,'b')
        a.claim(key,None,'a',bootstrap=True)
        sa=dict(cameraId='camera',runId='a',runGeneration=1,state='RUNNING',lastSeq=1)
        sb=dict(sa,runId='b',runGeneration=2)
        a.write(key,sa,'old');b.claim(key,'a','b');b.write(key,sb,'new')
        with self.assertRaises(StaleWriter):a.write(key,dict(sa,state='STOPPED'),None)
        with self.assertRaises(StaleWriter):a.write(key,sa,'late')
        stream=f'{self.settings.namespace}:camera:{key}:frames'
        self.assertEqual(['new'],[f['json'] for _,f in b.client.xrange(stream)])
        for i in range(105):b.write(key,sb,str(i))
        self.assertEqual(100,b.client.xlen(stream))
        self.assertGreater(b.client.ttl(stream),0)

    def test_watch_race_rejects_already_prepared_old_transaction(self):
        import redis
        key=camera_key('camera');a=self.sink(self.settings,'a');b=self.sink(self.settings,'b')
        a.claim(key,None,'a',bootstrap=True)
        with a.client.pipeline() as pipe:
            pipe.watch(a.owner_key(key));self.assertEqual('a',pipe.get(a.owner_key(key)))
            b.claim(key,'a','b')
            pipe.multi();pipe.set(self.settings.namespace+':probe','old')
            with self.assertRaises(redis.WatchError):pipe.execute()
        self.assertIsNone(b.client.get(self.settings.namespace+':probe'))

    def test_control_handoff_child_terminal_and_global_bounds(self):
        control=ProcessMetadataControl(self.settings,sink_factory=self.sink)
        self.assertLessEqual(control.pending_bound,self.settings.max_bytes)
        self.assertTrue(control.channel('camera',None,0,'STOPPED'))
        token=control.claim('camera','run',1)
        with self.assertRaises(RuntimeError):control.claim('camera','other',2)
        sink=self.sink(self.settings,token)
        child=ChildPublisher(self.settings,'camera','run',1,token,sink=sink)
        self.assertEqual(CHANNEL_BYTES,child.settings.max_bytes)
        child.start()
        try:
            self.assertFalse(child.channel('foreign','run',1,'STARTING'))
            self.assertTrue(child.channel('camera','run',1,'STARTING'))
            self.assertTrue(child.channel('camera','run',1,'RUNNING'))
            self.assertTrue(child.submit(dict(cameraId='camera',runId='run',runGeneration=1,frameSeq=0)))
            self.assertFalse(child.submit(dict(cameraId='camera',runId='old',runGeneration=1,frameSeq=1)))
            self.assertTrue(child.channel('camera','run',1,'STOPPED'))
            self.assertTrue(child.wait_channel('camera','run',1,2))
        finally:
            self.assertTrue(child.close(2))
        control.retire('camera',token,'run',1)
        control.update_stats('camera',control._entries[camera_key('camera')]['token'],dict(sent_frames=7,sent_status=1))
        self.assertEqual(7,control.metrics()['sent_frames'])
        next_token=control.claim('camera','new',2)
        self.assertEqual(7,control.metrics()['sent_frames'])
        self.assertNotEqual(token,next_token)
        with self.assertRaises(StaleWriter):
            sink.write(camera_key('camera'),dict(runId='run',state='STOPPED'),None)
        with self.assertRaises(ValueError):ProcessMetadataControl(replace(self.settings,max_bytes=1024))


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