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