import tempfile
from pathlib import Path
import threading
import unittest
from unittest.mock import Mock, patch

import numpy as np

from app.capture_pixels import copy_sdk_rgb
from app.contracts.visualization import FrameResult
from app.metadata_render import HEADER, render_rgb
from app.metadata_visual import MetadataVisual
from test_intrusion import frame
from test_metadata_connection import wait_for


class CaptureBoundaryTest(unittest.TestCase):
    def test_sdk_rgb_dlpack_copy_and_renderer_channel_contract(self):
        import torch
        rgb = np.zeros((360,640,3),np.uint8)
        rgb[:,:200] = (255,0,0)
        rgb[:,200:400] = (0,255,0)
        rgb[:,400:] = (0,0,255)
        tensor = torch.from_numpy(rgb)
        class SDKTensorShape:
            def __dlpack__(self, stream):
                return tensor.__dlpack__()
        copied = copy_sdk_rgb(SDKTensorShape())
        self.assertTrue(np.array_equal(copied,rgb))
        image,_ = render_rgb(FrameResult(frame(0,0),()),copied)
        for x,color in ((100,(255,0,0)),(300,(0,255,0)),(500,(0,0,255))):
            self.assertEqual(image.getpixel((x,HEADER+100)),color)

    def test_opt_in_preparation_precedes_stream_boot_and_failure_is_nonfatal(self):
        import os
        from unittest.mock import AsyncMock
        from fastapi.testclient import TestClient
        from app.main import create_app
        order=[]
        async def boot(publisher):
            order.append('streams')
        with patch.dict(os.environ,{'CONTEXTRA_METADATA_VISUAL_DIR':'/unused'}), \
                patch('app.capture_pixels.prepare_capture',side_effect=lambda:(order.append('capture') or {'ready':False})), \
                patch('app.main._boot_streams',new=boot), patch('app.main._boot_retention',return_value=None), \
                patch('app.metadata_runtime.publisher_from_environment',return_value=None):
            app=create_app()
            with TestClient(app) as client:
                self.assertEqual(client.get('/live').status_code,200)
                self.assertFalse(app.state.capture_preparation['ready'])
            self.assertEqual(order,['capture','streams'])

    def test_capture_reserve_avoids_file_io_and_rate_limits(self):
        with tempfile.TemporaryDirectory() as root:
            path=Path(root); path.chmod(0o700); (path/'capture.enabled').touch()
            now=[0.]
            visual=MetadataVisual(path,clock=lambda:now[0])
            try:
                # Only calls from this thread are prohibited; the worker owns file polling.
                caller=threading.get_ident()
                original=Path.is_file
                def check(p):
                    if threading.get_ident()==caller:
                        raise AssertionError('callback file I/O')
                    return original(p)
                with patch.object(Path,'is_file',check):
                    self.assertTrue(visual.reserve_image(frame(0,0)))
                    self.assertFalse(visual.reserve_image(frame(1,.1)))
                    now[0]=.5
                    self.assertTrue(visual.reserve_image(frame(5,.5)))
                self.assertEqual(visual.status()['captures'],2)
            finally: visual.close(1)

    def test_preparation_failure_disables_debug_not_lifecycle(self):
        with tempfile.TemporaryDirectory() as root, patch('app.capture_pixels.preparation_status',
                return_value={'ready':False,'error_type':'RuntimeError'}):
            path=Path(root);path.chmod(0o700);(path/'capture.enabled').touch()
            visual=MetadataVisual(path,require_prepared=True)
            try:
                self.assertFalse(visual.reserve_image(frame(0,0)))
                self.assertFalse(visual.status()['preparation']['ready'])
            finally:self.assertTrue(visual.close(1))

    def test_slow_io_does_not_hold_callback_lock_and_releases(self):
        with tempfile.TemporaryDirectory() as root:
            path=Path(root);path.chmod(0o700);(path/'capture.enabled').touch()
            visual=MetadataVisual(path)
            entered,release=threading.Event(),threading.Event()
            try:
                with patch.object(visual,'_save',side_effect=lambda *args:(entered.set(),release.wait(2))):
                    f=frame(0,0)
                    visual.submit_image(f,np.zeros((360,640,3),np.uint8))
                    visual.submit_result(FrameResult(f,()))
                    self.assertTrue(entered.wait(1))
                    self.assertTrue(visual.reserve_image(frame(1,.1)))
                    release.set()
            finally:
                release.set()
                self.assertTrue(visual.close(1))