File size: 11,780 Bytes
33962ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 | """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')
|