95 lines
5.2 KiB
Python
95 lines
5.2 KiB
Python
import copy
|
|
import random
|
|
from viano_bug_analysis.analysis.json_consolidation import consolidate, restore
|
|
from viano_bug_analysis.analysis.model_context import model_json
|
|
|
|
|
|
def check(value):
|
|
before=copy.deepcopy(value)
|
|
encoded=consolidate(value)
|
|
assert model_json(restore(encoded)) == model_json(value)
|
|
assert value == before and consolidate(value) == encoded
|
|
assert len(model_json(encoded)) <= len(model_json(value))
|
|
return encoded
|
|
|
|
|
|
def test_repeated_metadata_code_and_database_rows_roundtrip():
|
|
origin={'repository':'https://example.invalid/repo','revision':'a'*40,'snapshot':'s'*80}
|
|
code=[{'path':'src/long/path/module.cpp','origin':origin,'sha256':'b'*64,'line':i,'text':'line '+str(i),'text_truncated':False} for i in [1,2,3,8,8,9]]
|
|
value={'first':code,'second':copy.deepcopy(code), 'rows':[{'row':{'id':i,'value':None,'enabled':False},'origin':origin} for i in range(50)]}
|
|
encoded=check(value)
|
|
assert encoded['encoding']=='shared_json_v1'
|
|
assert len(model_json(encoded)) < len(model_json(value))*.65
|
|
|
|
|
|
def test_structural_records_keep_ids_values_and_missing_distinctions():
|
|
value={f'F{i:04d}':{'source_identifier':'original/path','technical_observation':{'row':{'value':v}},'observation_scope':'local'}
|
|
for i,v in enumerate([False,0,None,True,1,'0',{},[],{'absent':None}]*10)}
|
|
result=check(value)
|
|
assert '$record' in model_json(result)
|
|
assert list(restore(result)) == list(value)
|
|
|
|
|
|
def test_flags_gaps_duplicate_lines_unicode_and_heterogeneous_records():
|
|
check([{'line':n,'text':'ä\t\n\\','text_truncated':flag} for n,flag in [(3,False),(3,0),(7,None),(8,True)]])
|
|
check([{'line':1,'text':'a'},{'line':2,'text':'b','missing':None},{'line':3,'text':'c'}])
|
|
check({'already':{'$ref':'literal user data'}, 'other':[1,2,3]})
|
|
for value in (None,False,0,1.0,[],{},'plain'):check(value)
|
|
|
|
|
|
def test_random_json_roundtrip_and_no_expansion():
|
|
rng=random.Random(12)
|
|
def node(depth):
|
|
if not depth:return rng.choice([None,True,False,0,1,1.0,'ß','long text '*30])
|
|
return rng.choice([lambda:node(0),lambda:[node(depth-1) for _ in range(rng.randrange(6))],
|
|
lambda:{str(i):node(depth-1) for i in range(rng.randrange(6))}])()
|
|
for _ in range(100):check(node(4))
|
|
|
|
|
|
def test_input_that_resembles_transport_is_escaped_without_reinterpretation():
|
|
value={'encoding':'shared_json_v1','data':{'$ref':'not-a-reference'},'shared':{},'shapes':{}}
|
|
assert restore(consolidate(value)) == value
|
|
|
|
|
|
def test_new_phase_prompts_decode_to_the_previous_views_with_same_output_contract():
|
|
import json
|
|
from viano_bug_analysis.analysis.prompt import InterpretationPromptBuilder
|
|
from viano_bug_analysis.analysis.hypothesis_prompt import HypothesisPromptBuilder
|
|
from viano_bug_analysis.analysis.diagnosis_builder import build_prompt as diagnosis
|
|
from viano_bug_analysis.analysis.change_request_contracts import build_prompt as handoff
|
|
from test_interpretation_quality import fixture
|
|
from test_handoff_outcomes import minimal
|
|
handoff_context=minimal()[1]
|
|
handoff_context.update(case_id='case',package_id='package',analysis_revision=1,diagnosis_ref='diagnosis.json',diagnosis_sha256='0'*64,interpretation={},policy={},case={})
|
|
handoff_context['evidence']['E000001']['record']['content_sha256']='1'*64
|
|
c=fixture()['context']
|
|
c['sources']['D0099']={'kind':'data_quality','case_pointer':'/interpretation_quality/model_context',
|
|
'data':{'model_context_policy':'compact_storage_v1'}}
|
|
old=InterpretationPromptBuilder().build(c)
|
|
new_context={**c,'model_encoding':'shared_json_v1'}
|
|
new=InterpretationPromptBuilder().build(new_context)
|
|
decoded=restore(json.loads(new['messages'][1]['content'])['input'])
|
|
decoded.pop('model_encoding')
|
|
assert decoded == json.loads(old['messages'][1]['content'])['input']
|
|
assert old['output_schema'] == new['output_schema']
|
|
assert new['prompt_version']=='case_interpretation_resolved_v7'
|
|
for old_version,new_version,builder,context,kwargs in [
|
|
('hypotheses_v2','hypotheses_v3',HypothesisPromptBuilder().build,
|
|
{'policy':{'min_hypotheses':3,'max_hypotheses':5},'sources':c['sources']},{}),
|
|
('diagnosis_v3','diagnosis_v4',diagnosis,
|
|
{'evaluation_protocol':2,'case':{},'budget':{},'investigation':{},'sources':c['sources']}, {'modern':True,'compact':True}),
|
|
('change_request_v4','change_request_v5',handoff,handoff_context,{'compact':True})]:
|
|
if old_version.startswith('hypotheses'):
|
|
old=builder({**context,'prompt_version':old_version})
|
|
new=builder({**context,'prompt_version':new_version})
|
|
elif old_version.startswith('diagnosis'):
|
|
old=builder(context,**kwargs);new=builder(context,**kwargs,consolidated=True)
|
|
else:
|
|
old=builder(context,**kwargs,version=old_version);new=builder(context,**kwargs,version=new_version)
|
|
before=json.loads(old['messages'][1]['content'])['input']
|
|
after=restore(json.loads(new['messages'][1]['content'])['input'])
|
|
if 'prompt_version' in before:before['prompt_version']=new_version
|
|
assert after == before
|
|
assert old['output_schema'] == new['output_schema']
|
|
assert new['prompt_version']==new_version
|