"""Release regression checks; every network transport here is mocked."""
import asyncio
import json

import httpx
import pytest

from config.schema import Configuration, Provider, Route, parse_yaml
from vrag.egress import PublicAddressTransport
from vrag.providers import Adapter, ProviderFailure
from vrag.routing import route

@pytest.fixture
def network(monkeypatch):
    monkeypatch.setenv('URL', 'https://models.example/v1')
    monkeypatch.setenv('VM_ALLOWED_HOSTS', 'models.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='URL', model='test', capabilities=['image_understanding'] if caps is None else caps, trust='approved_sensitive')

@pytest.mark.asyncio
async def test_direct_transport_pins_validated_ip_and_preserves_tls_hostname(network):
    seen=[]
    def reply(request):
        seen.append(request)
        assert request.url.host=='8.8.8.8'
        assert request.headers['host']=='models.example'
        assert request.extensions['sni_hostname']=='models.example'
        return httpx.Response(200,json={'ok':True})
    async with httpx.AsyncClient(transport=PublicAddressTransport(httpx.MockTransport(reply))) as client:
        assert (await client.get('https://models.example/v1/info')).status_code==200
    assert len(seen)==1

@pytest.mark.asyncio
async def test_rebound_private_address_is_never_connected(network, monkeypatch):
    monkeypatch.setattr('socket.getaddrinfo', lambda *a,**k:[(2,1,6,'',('127.0.0.1',443))])
    def should_not_connect(request):
        pytest.fail('private destination reached the transport')
    async with httpx.AsyncClient(transport=PublicAddressTransport(httpx.MockTransport(should_not_connect))) as client:
        with pytest.raises(httpx.ConnectError,match='denied'):
            await client.get('https://models.example/v1/info')

@pytest.mark.asyncio
@pytest.mark.parametrize('url',['https://[broken','https://models.example:bad/v1','https://models.example:8443','https://user:password@models.example','http://models.example'])
async def test_malformed_or_disallowed_url_is_safe_failure(network,monkeypatch,url):
    monkeypatch.setenv('URL',url)
    with pytest.raises(ProviderFailure,match='destination_denied'):
        await Adapter().call('vlm',provider(),{})

@pytest.mark.asyncio
@pytest.mark.parametrize('detections',[[None],None,{},[{'label':'hand','bbox_xyxy':[0,0,2,1],'confidence':1}]])
async def test_parallel_malformed_detector_is_independent(network,monkeypatch,detections):
    def reply(request):
        if request.url.path.endswith('/detect'):
            return httpx.Response(200,json={'camera_id':'camera2','frame_id':7,'width':640,'height':480,'detections':detections})
        return httpx.Response(200,json={'choices':[{'message':{'content':'A person is present.'}}]})
    a=Adapter(httpx.MockTransport(reply));monkeypatch.setattr('vrag.routing.adapter',a)
    config=Configuration(providers={'vision':provider(),'detector':provider('detection-http',['object_detection','bounding_boxes'])},features={
        'live.camera2.detection':Route(strategy='parallel',providers=['vision','detector'],fusion='spatial_semantic')})
    out,status=await route(config,'live.camera2.detection',{'image':'synthetic','camera_id':'camera2','frame_id':7,'width':640,'height':480})
    assert out.decisions[0].source=='vision' and out.detections==[]
    assert {s['provider']:s['status'] for s in status}=={'vision':'ok','detector':'invalid_output'}

@pytest.mark.asyncio
async def test_parallel_spatial_semantic_fusion_preserves_provenance(network,monkeypatch):
    entered=set();both=asyncio.Event()
    async def reply(request):
        entered.add(request.url.path)
        if len(entered)==2:both.set()
        await asyncio.wait_for(both.wait(),1)
        if request.url.path.endswith('/detect'):
            return httpx.Response(200,json={'camera_id':'camera2','frame_id':7,'width':640,'height':480,'detections':[{'label':'hand','bbox_xyxy':[.1,.2,.3,.4],'confidence':.9}]})
        return httpx.Response(200,json={'choices':[{'message':{'content':'Hand visible; inspect alignment.'}}]})
    a=Adapter(httpx.MockTransport(reply));monkeypatch.setattr('vrag.routing.adapter',a)
    config=Configuration(providers={'semantic':provider(),'nvidia_gateway':provider('detection-http',['object_detection','bounding_boxes'])},features={
        'live.camera2.detection':Route(strategy='parallel',providers=['semantic','nvidia_gateway'],fusion='spatial_semantic')})
    out,status=await route(config,'live.camera2.detection',{'image':'synthetic','camera_id':'camera2','frame_id':7,'width':640,'height':480})
    assert len(status)==2 and all(s['status']=='ok' for s in status)
    assert out.detections[0].source=='nvidia_gateway' and out.detections[0].bbox_xyxy==(.1,.2,.3,.4)
    assert out.detections[0].object_id is None and out.decisions[0].source=='semantic'

@pytest.mark.asyncio
@pytest.mark.parametrize('placeholder',[False,True])
async def test_laya_plus_detector_parallel_real_adapter_contract(network,monkeypatch,placeholder):
    from pathlib import Path
    monkeypatch.setenv('LAYA_URL','https://models.example')
    laya=Provider(adapter='laya-vision',base_url_env='LAYA_URL',model='thaitea/laya-vision',capabilities=['image_understanding','visual_decision'],trust='approved_sensitive')
    detector=provider('detection-http',['object_detection','bounding_boxes'])
    fixture=json.loads((Path(__file__).parent/'fixtures/laya-live-synthetic.json').read_text())
    if placeholder:fixture['model']='laya-vision-placeholder'
    called=set();both=asyncio.Event()
    async def reply(request):
        called.add(request.url.path)
        if len(called)==2:both.set()
        await asyncio.wait_for(both.wait(),1)
        if request.url.path=='/v1/system-one':
            body=json.loads(request.content)
            assert body['imageBase64']=='eA==' and 'image' not in body
            assert body['questions']['visual_signal']['type']=='noul'
            return httpx.Response(200,json=fixture)
        assert request.url.path=='/v1/detect'
        return httpx.Response(200,json={'frame_id':7,'camera_id':'camera2','width':640,'height':480,'detections':[{'label':'hand','bbox_xyxy':[.1,.2,.3,.4],'confidence':.9}]})
    cfg=Configuration(providers={'laya':laya,'nvidia_gateway':detector},features={
        'live.camera2.detection':Route(strategy='parallel',providers=['laya','nvidia_gateway'],fusion='spatial_semantic')})
    monkeypatch.setattr('vrag.routing.adapter',Adapter(httpx.MockTransport(reply)))
    out,status=await route(cfg,'live.camera2.detection',{'image':'data:image/jpeg;base64,eA==','camera_id':'camera2','frame_id':7,'width':640,'height':480})
    assert out.detections[0].source=='nvidia_gateway' and out.detections[0].object_id is None
    assert {s['provider']:s['status'] for s in status}=={'laya':'invalid_output' if placeholder else 'ok','nvidia_gateway':'ok'}
    assert len(out.decisions)==(0 if placeholder else 1)
    if not placeholder:assert out.decisions[0].source=='laya' and '0.3126' in out.decisions[0].reason

@pytest.mark.asyncio
async def test_fallback_reauthorizes_sensitive_destination(network,monkeypatch):
    p=provider();public=p.model_copy(update={'trust':'public_only'})
    calls=[]
    def reply(request):
        calls.append(request)
        return httpx.Response(503)
    monkeypatch.setattr('vrag.routing.adapter',Adapter(httpx.MockTransport(reply)))
    cfg=Configuration(providers={'first':p,'unapproved':public},features={'live.camera1.scene':Route(strategy='fallback',providers=['first','unapproved'])})
    out,status=await route(cfg,'live.camera1.scene',{'image':'sensitive'})
    assert len(calls)==1 and status[-1]['status']=='destination_denied' and not out.decisions

@pytest.mark.asyncio
async def test_changed_provider_does_not_inherit_old_circuit(network):
    a=Adapter(httpx.MockTransport(lambda r:httpx.Response(500)))
    for _ in range(3):
        with pytest.raises(ProviderFailure):await a.call('vision',provider(),{})
    a.transport=httpx.MockTransport(lambda r:httpx.Response(200,json={'choices':[{'message':{'content':'ok'}}]}))
    out=await a.call('vision',provider().model_copy(update={'model':'replacement'}),{})
    assert out.text=='ok'

@pytest.mark.asyncio
@pytest.mark.parametrize('vector',[[[1,2]],[1e100,2],[],[0,0]])
async def test_invalid_embeddings_rejected(network,vector):
    a=Adapter(httpx.MockTransport(lambda r:httpx.Response(200,json={'data':[{'embedding':vector}]})))
    with pytest.raises(ProviderFailure,match='invalid_output'):
        await a.call('emb',provider('openai-embeddings',['text_embeddings']),{'text':'test'})

def test_role_and_predicate_configuration_validation():
    with pytest.raises(ValueError):
        parse_yaml('''providers:
  emb: {adapter: openai-embeddings, base_url_env: URL, model: e, capabilities: [text_embeddings]}
  vlm: {adapter: openai-chat-vision, base_url_env: URL, model: v, capabilities: [reasoning]}
features:
  rag.text_embeddings: {strategy: fallback, providers: [emb, vlm]}
''')
    with pytest.raises(ValueError):
        Route(strategy='conditional',providers=['x'],conditions=[{'field':'has_detections','equals':'false','provider':'x'}])

@pytest.mark.asyncio
async def test_proxy_only_dns_preserves_trusted_path_and_direct_fails_closed(network,monkeypatch):
    import socket
    def unavailable(*args,**kwargs):raise socket.gaierror(socket.EAI_AGAIN,'local DNS unavailable')
    monkeypatch.setattr('socket.getaddrinfo',unavailable)
    monkeypatch.setattr('vrag.providers.proxy_for',lambda base:'http://trusted-proxy:8080')
    a=Adapter(httpx.MockTransport(lambda r:httpx.Response(200,json={'choices':[{'message':{'content':'ok'}}]})))
    assert (await a.call('vision',provider(),{})).text=='ok'
    monkeypatch.setattr('vrag.providers.proxy_for',lambda base:None)
    with pytest.raises(ProviderFailure,match='destination_denied'):await a.call('vision',provider(),{})

@pytest.mark.asyncio
@pytest.mark.parametrize('failed',['laya','nvidia_gateway'])
async def test_laya_nvidia_timeout_preserves_other_typed_output(network,monkeypatch,failed):
    from pathlib import Path
    monkeypatch.setenv('LAYA_URL','https://models.example')
    laya=Provider(adapter='laya-vision',base_url_env='LAYA_URL',model='thaitea/laya-vision',capabilities=['image_understanding','visual_decision'],trust='approved_sensitive')
    detector=provider('detection-http',['object_detection','bounding_boxes'])
    response=json.loads((Path(__file__).parent/'fixtures/laya-live-synthetic.json').read_text())
    async def reply(request):
        is_laya=request.url.path.endswith('/system-one')
        if (is_laya and failed=='laya') or (not is_laya and failed=='nvidia_gateway'):
            raise httpx.ReadTimeout('synthetic timeout')
        if is_laya:return httpx.Response(200,json=response)
        return httpx.Response(200,json={'frame_id':7,'camera_id':'camera2','width':640,'height':480,'detections':[{'label':'hand','bbox_xyxy':[.1,.2,.3,.4],'confidence':.9}]})
    cfg=Configuration(providers={'laya':laya,'nvidia_gateway':detector},features={'live.camera2.detection':Route(strategy='parallel',providers=['laya','nvidia_gateway'],fusion='spatial_semantic')})
    monkeypatch.setattr('vrag.routing.adapter',Adapter(httpx.MockTransport(reply)))
    out,status=await route(cfg,'live.camera2.detection',{'image':'data:image/jpeg;base64,eA==','camera_id':'camera2','frame_id':7,'width':640,'height':480})
    assert next(s for s in status if s['provider']==failed)['status']=='timeout'
    assert bool(out.detections)==(failed=='laya')
    assert bool(out.decisions)==(failed=='nvidia_gateway')