56 lines
2.8 KiB
Python
56 lines
2.8 KiB
Python
import copy
|
|
import pytest
|
|
from viano_bug_analysis.llm.config import ProviderConfig
|
|
from viano_bug_analysis.llm.errors import ProviderUnavailable
|
|
from viano_bug_analysis.minimal_prompt_test import make_plan,measure
|
|
|
|
|
|
def provider():
|
|
return ProviderConfig(id='fixture',model='fixture-model',base_url='https://example.invalid/v1beta',
|
|
adapter='gemini',privacy='cloud',reasoning_effort='low',max_output_tokens=8192)
|
|
|
|
|
|
def test_four_requests_differ_only_in_language_and_native_format():
|
|
p=provider();plan=make_plan(p)
|
|
assert [c['id'] for c in plan['calls']]==['de_plain','de_json','en_plain','en_json']
|
|
for plain,structured in (plan['calls'][:2],plan['calls'][2:]):
|
|
assert plain['payload']['contents']==structured['payload']['contents']
|
|
assert plain['prompt_characters']<500
|
|
assert set(plain['payload'])=={'contents','generationConfig'}
|
|
generation=copy.deepcopy(structured['payload']['generationConfig'])
|
|
generation.pop('responseMimeType');generation.pop('responseJsonSchema')
|
|
assert generation==plain['payload']['generationConfig']
|
|
assert generation=={'maxOutputTokens':8192,'thinkingConfig':{'thinkingLevel':'low'}}
|
|
|
|
|
|
@pytest.mark.parametrize('index',[0,1,2,3])
|
|
def test_usage_reports_candidates_and_thoughts_separately(index):
|
|
case=make_plan(provider())['calls'][index]
|
|
def post(url,payload,headers,limit,deadline):
|
|
assert payload==case['payload']
|
|
return {'modelVersion':'fixture','candidates':[{'finishReason':'STOP','content':{'parts':[
|
|
{'text':'hidden','thought':True},{'text':'{"step":"lookup","purpose":"test"}' if index%2 else 'Look up spawns.'}]}}],
|
|
'usageMetadata':{'promptTokenCount':123,'candidatesTokenCount':12,'thoughtsTokenCount':30,'totalTokenCount':165}}
|
|
result=measure(provider(),case,10000,post)
|
|
assert result['status']=='SUCCESS'
|
|
assert result['usage']['input_tokens']==123
|
|
assert result['usage']['candidate_output_tokens']==12
|
|
assert result['usage']['reasoning_tokens']==30
|
|
assert result['usage']['cached_input_tokens'] is None
|
|
assert 'hidden' not in result['answer']
|
|
|
|
|
|
def test_no_retries_and_no_provider_body_leak():
|
|
calls=[]
|
|
def post(*args):
|
|
calls.append(1);raise ProviderUnavailable('LLM_HTTP_TEMPORARY',http_status=503)
|
|
result=measure(provider(),make_plan(provider())['calls'][0],1000,post)
|
|
assert len(calls)==1 and result['status']=='ERROR' and result['http_status']==503
|
|
|
|
|
|
def test_incomplete_retains_usage():
|
|
def post(*args):return {'candidates':[{'finishReason':'MAX_TOKENS','content':{'parts':[]}}],
|
|
'usageMetadata':{'promptTokenCount':99,'thoughtsTokenCount':100}}
|
|
result=measure(provider(),make_plan(provider())['calls'][0],1000,post)
|
|
assert result['status']=='INCOMPLETE' and result['usage']['input_tokens']==99
|