107 lines
6.4 KiB
Python
107 lines
6.4 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from test_llm_gateway import http_endpoint, active_case, request, provider, Adapter
|
|
from viano_bug_analysis.llm import LLMError, LLMGateway, GatewayConfig, ProviderConfig, ProviderResponse, CaseLLMRunner
|
|
from viano_bug_analysis.llm.diagnostics import sanitize
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.telemetry.safety import detail
|
|
|
|
|
|
@pytest.mark.parametrize('reason,refusal,tools', [('length',None,None),('content_filter','PRIVATE_REFUSAL',None),
|
|
('tool_calls',None,[{'PRIVATE_TOOL':'PRIVATE_ARGUMENT'}])])
|
|
def test_incomplete_api_responses_keep_usage_and_safe_finish_details(http_endpoint,reason,refusal,tools):
|
|
url,seen,reply=http_endpoint
|
|
reply['body']={'choices':[{'finish_reason':reason,'message':{'content':'PRIVATE_OUTPUT','refusal':refusal,'tool_calls':tools}}],
|
|
'usage':{'prompt_tokens':123,'completion_tokens':8192,'completion_tokens_details':{'reasoning_tokens':8000}}}
|
|
config=GatewayConfig((ProviderConfig(id='test',model='configured-model',base_url=url,max_output_tokens=8192),))
|
|
with pytest.raises(LLMError,match='LLM_INCOMPLETE_OUTPUT') as caught:
|
|
asyncio.run(LLMGateway(config).complete_round(request()))
|
|
error=caught.value
|
|
assert error.diagnostics['finish_reason']==reason
|
|
assert error.diagnostics['has_refusal']==bool(refusal) and error.diagnostics['has_tool_calls']==bool(tools)
|
|
assert error.diagnostics['reasoning_tokens']==8000
|
|
assert error.metadata['input_tokens']==123 and error.metadata['output_tokens']==8192
|
|
assert error.metadata['elapsed_seconds']>=0
|
|
assert error.diagnostics['max_output_tokens']==8192
|
|
assert 'PRIVATE' not in json.dumps(error.diagnostics) and 'PRIVATE' not in str(error)
|
|
assert len(seen)==1
|
|
|
|
|
|
def test_unknown_finish_reasons_and_bad_usage_are_not_copied(http_endpoint):
|
|
url,_,reply=http_endpoint
|
|
reply['body']={'choices':[{'finish_reason':'PRIVATE_REASON','message':{'content':'PRIVATE_BODY'}}],
|
|
'usage':{'prompt_tokens':True,'completion_tokens':'PRIVATE_TOKEN_VALUE'}}
|
|
config=GatewayConfig((ProviderConfig(id='test',model='configured',base_url=url),))
|
|
with pytest.raises(LLMError) as caught:asyncio.run(LLMGateway(config).complete_round(request()))
|
|
error=caught.value
|
|
assert error.diagnostics['finish_reason']=='unknown'
|
|
assert error.metadata['input_tokens'] is None and error.metadata['output_tokens'] is None
|
|
assert 'PRIVATE' not in json.dumps({'metadata':error.metadata,'diagnostics':error.diagnostics})
|
|
|
|
|
|
def test_failed_json_parsing_still_keeps_api_usage(http_endpoint):
|
|
url,_,reply=http_endpoint
|
|
reply['body']['choices'][0]['message']['content']='PRIVATE_NON_JSON'
|
|
config=GatewayConfig((ProviderConfig(id='test',model='configured',base_url=url),))
|
|
with pytest.raises(LLMError,match='LLM_RESPONSE') as caught:asyncio.run(LLMGateway(config).complete_round(request()))
|
|
assert caught.value.metadata['input_tokens']==3 and caught.value.diagnostics['finish_reason']=='stop'
|
|
|
|
|
|
def test_http_schema_rejection_retains_completion_details(http_endpoint):
|
|
url,_,reply=http_endpoint
|
|
reply['body']['choices'][0]['message']['content']='{"wrong":1}'
|
|
config=GatewayConfig((ProviderConfig(id='test',model='configured',base_url=url),))
|
|
with pytest.raises(LLMError,match='LLM_OUTPUT_SCHEMA') as caught:asyncio.run(LLMGateway(config).complete_round(request()))
|
|
assert caught.value.diagnostics['finish_reason']=='stop'
|
|
assert caught.value.diagnostics['response_bytes']>0
|
|
assert caught.value.metadata['output_tokens']==5
|
|
|
|
|
|
def test_schema_rejection_persists_usage_and_replays_without_network(active_case):
|
|
store,_,_=active_case
|
|
adapter=Adapter({'first':ProviderResponse({'PRIVATE_UNKNOWN_FIELD':'PRIVATE_VALUE'},'returned-model',100,20)})
|
|
config=GatewayConfig((provider(input_cost_per_million=2,output_cost_per_million=4),))
|
|
runner=CaseLLMRunner(store,LLMGateway(config,{'fixture':adapter}))
|
|
result=asyncio.run(runner.run('case-A','stable-package',operation_id='diagnostics',request=request()))
|
|
assert result.status=='ERROR' and PackageValidator().validate(result.package).valid
|
|
with PackageReader(result.package) as reader:
|
|
record=json.loads(reader.read_bytes(result.result_ref))
|
|
assert record['outcome']['metadata']['input_tokens']==100
|
|
assert record['outcome']['metadata']['estimated_cost']==pytest.approx(0.00028)
|
|
assert record['outcome']['diagnostics']['validation_keyword'] in {'required','additionalProperties'}
|
|
assert 'PRIVATE' not in json.dumps(record['outcome']['diagnostics'])
|
|
assert record['outcome']['rejected_document']=={'PRIVATE_UNKNOWN_FIELD':'PRIVATE_VALUE'}
|
|
assert asyncio.run(runner.run('case-A','stable-package',operation_id='diagnostics',request=request()))==result
|
|
assert adapter.calls==['first']
|
|
|
|
|
|
def test_safe_diagnostics_survive_log_sanitization():
|
|
dirty={'finish_reason':'length','input_tokens':3,'output_tokens':8,'has_refusal':True,
|
|
'provider_body':'PRIVATE_BODY','validation_keyword':{},'output_characters':4}
|
|
cleaned=sanitize(dirty)
|
|
assert 'provider_body' not in cleaned and 'validation_keyword' not in cleaned
|
|
event=detail('llm_round',{'diagnostics':dirty,'elapsed_seconds':1.25})
|
|
assert event['diagnostics']==cleaned and event['elapsed_seconds']==1.25
|
|
assert 'PRIVATE' not in json.dumps(event)
|
|
|
|
|
|
def test_failed_usage_is_counted_and_finish_reason_visible_in_service_log(active_case):
|
|
from viano_bug_analysis.telemetry import Telemetry
|
|
from viano_bug_analysis.telemetry.__main__ import read_events
|
|
store,_,_=active_case
|
|
error=LLMError('LLM_INCOMPLETE_OUTPUT',diagnostics={'finish_reason':'length','input_tokens':10,'output_tokens':7})
|
|
adapter=Adapter({'first':error})
|
|
with Telemetry(store.workspace) as observer:
|
|
store.telemetry=observer
|
|
runner=CaseLLMRunner(store,LLMGateway(GatewayConfig((provider(),)),{'fixture':adapter}))
|
|
result=asyncio.run(runner.run('case-A','stable-package',operation_id='incomplete',request=request()))
|
|
assert result.status=='ERROR'
|
|
event=next(e for e in read_events(store.workspace.root) if e['event']=='llm_round')
|
|
assert event['detail']['diagnostics']['finish_reason']=='length'
|
|
metrics=(observer.directory/'metrics.prom').read_text()
|
|
assert 'vba_llm_tokens_total{direction="input_tokens"} 10' in metrics
|
|
assert 'vba_llm_tokens_total{direction="output_tokens"} 7' in metrics
|