import httpx
import pytest
from config.schema import Provider
from vrag.providers import Adapter, ProviderFailure

@pytest.fixture
def remote(monkeypatch):
    monkeypatch.setenv('TEST_URL', 'https://inference.example/v1')
    monkeypatch.setenv('VM_ALLOWED_HOSTS', 'inference.example')
    monkeypatch.setattr('socket.getaddrinfo', lambda *a, **k: [(2, 1, 6, '', ('8.8.8.8', 443))])

def provider(adapter='openai-chat-vision', caps=None):
    return Provider(adapter=adapter, base_url_env='TEST_URL', model='test', capabilities=caps or ['image_understanding'], trust='approved_sensitive')

@pytest.mark.asyncio
async def test_chat_never_invents_boxes(remote):
    async def mock(req):
        assert req.url.path == '/v1/chat/completions'
        return httpx.Response(200, json={'choices': [{'message': {'content': 'a hand at [0,0,1,1]'}}]})
    out = await Adapter(httpx.MockTransport(mock)).call('vlm', provider(), {'image': 'data:image/jpeg;base64,eA=='})
    assert out.detections == []
    assert out.decisions[0].source == 'vlm'

@pytest.mark.asyncio
async def test_embeddings(remote):
    out = await Adapter(httpx.MockTransport(lambda r: httpx.Response(200, json={'data': [{'embedding': [1,2,3]}]}))).call('emb', provider('openai-embeddings', ['text_embeddings']), {'text': 'hello'})
    assert out.embedding == [1,2,3]

@pytest.mark.asyncio
async def test_destination_denied(remote, monkeypatch):
    monkeypatch.setenv('TEST_URL', 'http://169.254.169.254')
    with pytest.raises(ProviderFailure, match='destination_denied'):
        await Adapter().call('x', provider(), {})

@pytest.mark.asyncio
async def test_no_secret_in_errors(remote):
    with pytest.raises(ProviderFailure, match='provider_error'):
        await Adapter(httpx.MockTransport(lambda r: httpx.Response(401, text='secret'))).call('x', provider(), {})

@pytest.mark.asyncio
async def test_remote_whisper_multipart(remote):
    def mock(req):
        assert req.url.path == '/v1/audio/transcriptions'
        assert b'name="file"' in req.content and b'speech.webm' in req.content
        return httpx.Response(200,json={'text':'hello'})
    out=await Adapter(httpx.MockTransport(mock)).call('stt',provider('openai-transcription',['transcription']),{'audio':b'webm'})
    assert out.text=='hello'

@pytest.mark.asyncio
async def test_detector_provenance_and_frame_validation(remote):
    payload={'image':'x','frame_id':1,'camera_id':'camera2','width':640,'height':480}
    data={'frame_id':1,'camera_id':'camera2','width':640,'height':480,'detections':[{'label':'hand','bbox_xyxy':[.1,.2,.4,.6],'confidence':.9}]}
    a=Adapter(httpx.MockTransport(lambda r:httpx.Response(200,json=data)))
    p=provider('detection-http',['object_detection','bounding_boxes'])
    out=await a.call('detector',p,payload)
    assert out.detections[0].source=='detector' and out.detections[0].object_id is None
    data['frame_id']=2
    with pytest.raises(ProviderFailure,match='invalid_output'):
        await a.call('detector',p,payload)

@pytest.mark.asyncio
async def test_private_ip_and_redirect_are_denied(remote,monkeypatch):
    monkeypatch.setattr('socket.getaddrinfo',lambda *a,**k:[(2,1,6,'',('127.0.0.1',443))])
    with pytest.raises(ProviderFailure,match='destination_denied'):
        await Adapter().call('x',provider(),{})
    monkeypatch.setattr('socket.getaddrinfo',lambda *a,**k:[(2,1,6,'',('8.8.8.8',443))])
    calls=[]
    def mock(req):
        calls.append(req.url)
        return httpx.Response(302,headers={'location':'https://evil.example'})
    with pytest.raises(ProviderFailure):
        await Adapter(httpx.MockTransport(mock)).call('x',provider(),{})
    assert len(calls)==1

@pytest.mark.asyncio
async def test_timeout_circuit_breaker_and_concurrency(remote):
    import asyncio
    p=provider().model_copy(update={'timeout_ms':100,'concurrency':1})
    async def slow(req):
        await asyncio.sleep(1)
        return httpx.Response(200,json={})
    a=Adapter(httpx.MockTransport(slow))
    first=asyncio.create_task(a.call('slow',p,{}))
    await asyncio.sleep(.01)
    with pytest.raises(ProviderFailure,match='busy'):
        await a.call('slow',p,{})
    with pytest.raises(ProviderFailure,match='timeout'):
        await first
    for _ in range(2):
        with pytest.raises(ProviderFailure,match='timeout'):
            await a.call('slow',p,{})
    with pytest.raises(ProviderFailure,match='circuit_open'):
        await a.call('slow',p,{})

@pytest.mark.asyncio
async def test_sensitive_destination_checked_on_each_call(remote):
    p=provider().model_copy(update={'trust':'public_only'})
    with pytest.raises(ProviderFailure,match='destination_denied'):
        await Adapter().call('public',p,{'image':'sensitive'})