124 lines
6.6 KiB
Python
124 lines
6.6 KiB
Python
import copy
|
|
import pytest
|
|
from viano_bug_analysis.prompt_probe import prepare
|
|
from viano_bug_analysis.minimal_prompt_test import measure
|
|
from viano_bug_analysis.llm import LLMError
|
|
from test_minimal_prompt_test import provider
|
|
|
|
|
|
def request():return {'prompt_version':'fixture','messages':[{'role':'system','content':'Test instructions'},
|
|
{'role':'user','content':'Test context'}], 'output_schema':{'type':'object','required':['query'],
|
|
'properties':{'query':{'const':'database.schema'}},'additionalProperties':False}}
|
|
|
|
|
|
def test_native_schema_sent_once_without_wrapper_or_tools():
|
|
r=request();before=copy.deepcopy(r);case=prepare(provider(),r)
|
|
payload=case['payload']
|
|
assert set(payload)=={'systemInstruction','contents','generationConfig'}
|
|
assert payload['contents'][0]['parts'][0]['text']=='Test context'
|
|
assert payload['generationConfig']['responseJsonSchema']['properties']['query']=={'type':'string','enum':['database.schema']}
|
|
assert case['validation_schema']==r['output_schema']
|
|
assert payload['generationConfig']['thinkingConfig']=={'thinkingLevel':'low'}
|
|
assert r==before
|
|
|
|
|
|
def test_validates_the_prepared_schema_not_the_minimal_test_schema():
|
|
case=prepare(provider(),request())
|
|
def post(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"database.schema"}'}]}}],
|
|
'usageMetadata':{'promptTokenCount':200,'candidatesTokenCount':8,'thoughtsTokenCount':30}}
|
|
result=measure(provider(),case,10000,post)
|
|
assert result['status']=='SUCCESS' and result['schema_valid']
|
|
assert result['usage']['input_tokens']==200
|
|
|
|
|
|
@pytest.mark.parametrize('field,value',[('messages',[]),('messages',[{'role':'user','content':'x'}])])
|
|
def test_rejects_invalid_envelopes(field,value):
|
|
r=request();r[field]=value
|
|
with pytest.raises(LLMError):prepare(provider(),r)
|
|
|
|
|
|
def test_gemini_projection_types_const_and_keeps_strict_local_validation():
|
|
from viano_bug_analysis.prompt_probe import gemini_schema
|
|
from jsonschema import Draft202012Validator
|
|
r=request()
|
|
r['output_schema']['properties']['query']={'const':'database.schema','minLength':5,'pattern':'^database'}
|
|
original=copy.deepcopy(r)
|
|
c=prepare(provider(),r)
|
|
wire=c['payload']['generationConfig']['responseJsonSchema']
|
|
Draft202012Validator.check_schema(wire)
|
|
assert wire['properties']['query']=={'type':'string','enum':['database.schema']}
|
|
assert r==original and c['validation_schema']==r['output_schema']
|
|
assert gemini_schema({'type':['string','number','boolean','null']})=={
|
|
'anyOf':[{'type':v} for v in ['string','number','boolean','null']]}
|
|
|
|
|
|
def test_local_constraints_not_silently_dropped_from_response_check():
|
|
r=request();r['output_schema']['properties']['query']={'type':'string','pattern':'^database'}
|
|
c=prepare(provider(),r)
|
|
def post(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"invented"}'}]}}],
|
|
'usageMetadata':{'promptTokenCount':200}}
|
|
result=measure(provider(),c,10000,post)
|
|
assert result['status']=='INVALID_JSON' and result['schema_valid'] is False
|
|
|
|
|
|
def test_provider_error_is_bounded_and_redacts_credentials_and_literals(monkeypatch):
|
|
from dataclasses import replace
|
|
from viano_bug_analysis.prompt_probe import provider_diagnostic
|
|
monkeypatch.setenv('PROBE_TEST_SECRET','key-secret-value')
|
|
p=replace(provider(),api_key_env='PROBE_TEST_SECRET')
|
|
detail=provider_diagnostic({'error':{'status':'INVALID_ARGUMENT',
|
|
'message':'generation_config schema failed: key-secret-value "private text" https://example.invalid/private',
|
|
'details':[{'fieldViolations':[{'field':'generation_config.response_json_schema','description':'Too many states'}]}],
|
|
'extra':'do not retain'}},p)
|
|
import json
|
|
text=json.dumps(detail)
|
|
assert 'key-secret-value' not in text and 'private text' not in text
|
|
assert 'https://' not in text and 'do not retain' not in text
|
|
assert detail['status']=='INVALID_ARGUMENT'
|
|
assert detail['field_violations'][0]['description']=='Too many states'
|
|
|
|
|
|
def test_opt_in_http_capture_preserves_controlled_error(monkeypatch):
|
|
import io,json,time,urllib.error
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
body=json.dumps({'error':{'status':'INVALID_ARGUMENT','message':'Schema too complex'}}).encode()
|
|
class Opener:
|
|
def open(self,*args,**kwargs):
|
|
raise urllib.error.HTTPError('https://example.invalid',400,'Bad Request',{},io.BytesIO(body))
|
|
monkeypatch.setattr('urllib.request.build_opener',lambda *args:Opener())
|
|
seen=[]
|
|
with pytest.raises(LLMError) as caught:
|
|
GeminiProvider._post('https://example.invalid',{}, {},10000,time.monotonic()+10,error_capture=seen.append)
|
|
assert caught.value.code=='LLM_HTTP_CONFIGURATION' and caught.value.http_status==400
|
|
assert seen[0]['error']['message']=='Schema too complex'
|
|
|
|
|
|
def test_existing_output_is_named_error_without_network(tmp_path,monkeypatch,capsys):
|
|
import json
|
|
from viano_bug_analysis import prompt_probe
|
|
from viano_bug_analysis.llm import GatewayConfig
|
|
path=tmp_path/'request.json';path.write_text(json.dumps(request()))
|
|
output=tmp_path/'result.json';output.write_text('keep original')
|
|
monkeypatch.setattr(prompt_probe.GatewayConfig,'load',lambda _:GatewayConfig(providers=(provider(),),privacy_profile='cloud_allowed'))
|
|
def forbidden(*args):raise AssertionError('Unexpected network call')
|
|
monkeypatch.setattr(prompt_probe,'measure_diagnostic',forbidden)
|
|
result=prompt_probe.main(['--run','--request',str(path),'--output',str(output)])
|
|
assert result==2 and 'PROMPT_PROBE_OUTPUT_EXISTS' in capsys.readouterr().out
|
|
assert output.read_text()=='keep original'
|
|
|
|
|
|
def test_prompt_schema_mode_avoids_native_union_and_keeps_contract_once():
|
|
import json
|
|
r=request();case=prepare(provider(),r,schema_mode='prompt')
|
|
payload=case['payload']
|
|
assert 'responseJsonSchema' not in payload['generationConfig']
|
|
assert payload['generationConfig']['responseMimeType']=='application/json'
|
|
contract=json.dumps(r['output_schema'],ensure_ascii=False,separators=(',',':'))
|
|
assert payload['systemInstruction']['parts'][0]['text'].count(contract)==1
|
|
assert payload['contents'][0]['parts'][0]['text']==r['messages'][1]['content']
|
|
assert case['validation_schema']==r['output_schema']
|
|
def post(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"database.schema"}'}]}}]}
|
|
assert measure(provider(),case,10000,post)['status']=='SUCCESS'
|
|
def bad(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"invented"}'}]}}]}
|
|
assert measure(provider(),case,10000,bad)['status']=='INVALID_JSON'
|