import tempfile
import time
import unittest
from pathlib import Path
from unittest.mock import patch

from app.fire_config import FIRE_MODEL, FireConfig
from app.pose_config import execution_plan
from app.stream_settings import patch_settings, settings_view, revision


def spec():
    return dict(server_id='s',source_id='c',store=False,active_learning=False,scenarios='off')


class FirePathTest(unittest.TestCase):
    def test_plan_noop_and_atomic_validation(self):
        old=spec(); value=patch_settings(old,{'fire':{'enabled':True}})
        self.assertNotIn('fire',old)
        self.assertEqual(execution_plan(value)['model'],FIRE_MODEL)
        self.assertEqual(revision(value),revision(patch_settings(value,{'fire':{'enabled':True}})))
        self.assertEqual([x['key'] for x in settings_view(value)['categories']],['smoke_visible','flame_visible'])
        for bad in ({'pose':{'enabled':True}}, {'intrusion':{'enabled':True}}, {'display_categories':['person']}):
            with self.assertRaises(ValueError):patch_settings(value,bad)
        with self.assertRaises(ValueError):FireConfig(enabled=1)
        self.assertEqual(execution_plan(patch_settings(value,{'fire':{'enabled':False}}))['model'],'RFDETRMedium')

    def test_no_fallback_and_configuration(self):
        from app.inference_profiles import resolve_from_spec, render_cfg, ProfileError
        value=patch_settings(spec(),{'fire':{'enabled':True}})
        with self.assertRaises(ProfileError):resolve_from_spec('s',value,is_ready=lambda m:m=='RFDETRMedium')
        prof=resolve_from_spec('s',value,is_ready=lambda m:m==FIRE_MODEL)
        self.assertEqual(prof.resolved_models,{'det':FIRE_MODEL})
        with tempfile.TemporaryDirectory() as root, patch('app.inference_profiles._CFG_OUTPUT_DIR',root):
            cfg=Path(render_cfg('s','det',FIRE_MODEL)).read_text()
            self.assertIn('CreateContextraFireProcessor',cfg)
            self.assertIn('input_tensor_from_meta',cfg)
        for changes in ({'store':True},{'active_learning':True},{'server_id':None}):
            with self.assertRaises(ValueError):execution_plan(dict(value,**changes))
        with self.assertRaises(ValueError):execution_plan(dict(spec(),inference_profile={'det_model':FIRE_MODEL}))
        with self.assertRaises(ProfileError):resolve_from_spec('s',dict(spec(),inference_profile={'roles':['det']}),is_ready=lambda m:m==FIRE_MODEL)

    def test_worker_metadata_empty_stale_cleanup(self):
        from app.det_result_path import DetResultPath
        from app.frame_ingress import FrameIngress
        from app.contracts.frames import FrameEnvelope
        from app.native_metadata import NativeFrame,NativeDetection
        from app.byte_tracking import ByteTracker
        class Collector:
            def add(self,*a,**kw):pass
        errors=[];ingress=FrameIngress('s','c','c','r',1)
        path=DetResultPath(ingress,ByteTracker(),FIRE_MODEL,.5,Collector(),lambda:True,
                           lambda:errors.append(True),display_categories=['smoke_visible','flame_visible'])
        path.start()
        try:
            for seq,ds in [(0,(NativeDetection(0,.9,(.1,.1,.5,.5)),NativeDetection(1,.8,(.5,.5,.8,.8)))),(1,())]:
                frame=FrameEnvelope('s','c','c','r',1,seq,seq*100000000,1,1280,720)
                copied=NativeFrame(seq,frame.pts_ns,1280,720,ds,1280,720)
                self.assertTrue(path.submit(frame,copied))
                deadline=time.monotonic()+3
                while path.status().get('processed',0)<seq+1 and not errors and time.monotonic()<deadline:time.sleep(.01)
                self.assertFalse(errors)
                snapshot=path.snapshot()
                self.assertEqual(len(snapshot['objects']),len(ds))
                if ds:
                    self.assertEqual([c.label_text for c in path.latest_result().commands],['smoke_visible','flame_visible'])
                    self.assertNotIn('nms',snapshot)
                self.assertFalse(path.submit(frame,copied))
            self.assertEqual(path.latest_result().provider_model,FIRE_MODEL)
        finally:path.close(3)
        self.assertIsNone(path.latest_result())
        self.assertFalse(path._thread.is_alive())

    def test_manifest_exactness(self):
        from model_tooling.manifests import make_input,ASSETS,ContractError
        key='fire-smoke-medium-v110-strict-v1'; asset=ASSETS[key]
        with patch('model_tooling.manifests.fingerprints',return_value={k:asset[k] for k in ('size_bytes','sha256','md5')}):
            value=make_input(key,Path('/tmp')/asset['filename'])
            self.assertEqual(value['contract']['class_mapping'],{0:'smoke_visible',1:'flame_visible'})
            with self.assertRaises(ContractError):make_input('medium',Path('/tmp')/asset['filename'])


import test_metadata_connection as connection


class FireSettingsAPI(unittest.TestCase):
    def setUp(self):
        self.root=tempfile.TemporaryDirectory()
        self.addCleanup(self.root.cleanup)
        self.manager=connection.StreamManager(self.root.name+'/base',self.root.name+'/overlay')
        self.g=self.manager.add(dict(spec(),stream_id='stream',uri='fixture-only',
                                    inference_profile={'roles':['det']}))

    def test_fire_save_failure_and_cas(self):
        from app.stream_config_store import ConfigPersistenceError
        rev=self.manager.settings('stream')['configured']['revision']
        changes={'fire':{'enabled':True}}
        with patch.object(self.manager.config_store,'update',side_effect=ConfigPersistenceError()):
            with self.assertRaises(ConfigPersistenceError):self.manager.patch_settings('stream',rev,changes)
        self.assertFalse(self.manager.settings('stream')['configured']['fire']['enabled'])
        value=self.manager.patch_settings('stream',rev,changes)
        self.assertTrue(value['configured']['fire']['enabled'])
        with self.assertRaises(ValueError):self.manager.patch_settings('stream',rev,changes)

    def test_http_request_projection(self):
        from app.api.streams import SettingsPatch, patch_settings as route
        from pydantic import ValidationError
        from app.stream_config_store import ConfigPersistenceError
        from fastapi import HTTPException
        rev=self.manager.settings('stream')['configured']['revision']
        body=SettingsPatch.model_validate({'expected_revision':rev,'fire':{'enabled':True}})
        with patch('app.api.streams.stream_service.patch_settings',side_effect=self.manager.patch_settings):
            with patch.object(self.manager.config_store,'update',side_effect=ConfigPersistenceError()):
                with self.assertRaises(HTTPException) as failure:route('stream',body)
                self.assertEqual(failure.exception.status_code,503)
            self.assertFalse(self.manager.settings('stream')['configured']['fire']['enabled'])
            self.assertTrue(route('stream',body)['configured']['fire']['enabled'])
            with self.assertRaises(HTTPException) as conflict:route('stream',body)
            self.assertEqual(conflict.exception.status_code,409)
        with self.assertRaises(ValidationError):
            SettingsPatch.model_validate({'expected_revision':rev,'unknown':True})