208 lines
13 KiB
Python
208 lines
13 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.analysis.contracts import validate_interpretation, entity_binding_sources
|
|
from viano_bug_analysis.analysis.quality import consolidate_entities
|
|
from viano_bug_analysis.analysis.errors import InterpretationError
|
|
from viano_bug_analysis.analysis.prompt import InterpretationPromptBuilder
|
|
from test_entity_resolution import built_case, case_setup, Model, service_for, run, context_of, gateway, SETTINGS
|
|
|
|
|
|
def entity(identity, name, capture, locale=None, kind='creature'):
|
|
data={'type':kind,'id':identity,'name':name,'roles':['affected_subject'],
|
|
'citation_policy':'grouped_capture_v3','provenance':{'source_ref':capture,'json_pointer':'/result/items/0/row'},
|
|
'reason':'Contextual match','mention_source_ids':['F0016'],'assessment_ref':'assessment.json'}
|
|
if locale:data['locale']=locale
|
|
return {'kind':'entity','case_pointer':'/entity_resolution/selected/0','data':data}
|
|
|
|
|
|
def test_group_preserves_names_locales_roles_and_each_capture():
|
|
context={'sources':{'N0001':entity('23487','Wild Wolpertinger','a.json'),
|
|
'N0002':entity('23487','Wild Wolpertinger','b.json'),
|
|
'N0003':entity('23487','Wilder Wolpertinger','c.json','deDE'),
|
|
'N0004':entity('22943','Wolpertinger','d.json'),
|
|
'N0005':entity('23487','Different domain','e.json',kind='item')}}
|
|
consolidate_entities(context)
|
|
assert set(context['sources'])=={'N0001','N0004','N0005'}
|
|
associations=context['sources']['N0001']['data']['lookup_associations']
|
|
assert len(associations)==3
|
|
assert associations[-1]['name']=='Wilder Wolpertinger' and associations[-1]['locale']=='deDE'
|
|
assert {a['provenance']['source_ref'] for a in associations}=={'a.json','b.json','c.json'}
|
|
for i,ref in enumerate(['a.json','b.json','c.json'],1):
|
|
context['sources'][f'F{i:04d}']={'kind':'fact','data':{'name':'entity_name_lookup','provenance':{'source_ref':ref}}}
|
|
assert entity_binding_sources('N0001',context['sources']['N0001'],context['sources'])=={'N0001','F0001','F0002','F0003'}
|
|
|
|
|
|
def test_existing_canonical_entity_keeps_its_alias_and_original_provenance():
|
|
original={'type':'creature','id':'23487','roles':['observed'],'provenance':[{'source_ref':'original.json'}]}
|
|
context={'sources':{'N0001':{'kind':'entity','case_pointer':'/entities/0','data':copy.deepcopy(original)},
|
|
'N0002':entity('23487','Wild Wolpertinger','a.json')}}
|
|
consolidate_entities(context)
|
|
assert set(context['sources'])=={'N0001'}
|
|
data=context['sources']['N0001']['data']
|
|
assert data['provenance']==original['provenance']
|
|
assert data['roles']==['observed','affected_subject']
|
|
assert data['lookup_associations'][0]['provenance']['source_ref']=='a.json'
|
|
|
|
|
|
def fixture():
|
|
d=json.loads((Path(__file__).parent/'fixtures/interpretation/real-capture-citations.json').read_text(encoding='utf-8'))
|
|
d['context']['sources']['N0005']['data']['citation_policy']='grouped_capture_v3'
|
|
consolidate_entities(d['context'])
|
|
d['response']['described_behavior']=[{'text':'The quest describes catching creatures using a net while intoxicated or wearing goggles.',
|
|
'source_ids':['F0016'],'basis':'local_description','verification':'not_independently_verified'}]
|
|
d['response']['provisional_behavior_gap']={'text':'The tester did not see the described quest creatures while wearing goggles.',
|
|
'source_ids':['O0002','F0016'],'basis':'local_description','status':'provisional'}
|
|
return d
|
|
|
|
|
|
def test_provisional_gap_is_sourced_and_does_not_promote_formal_completeness():
|
|
d=fixture()
|
|
validate_interpretation(d['response'],d['context'])
|
|
assert d['context']['expected_behavior_completeness']=='missing'
|
|
assert d['response']['expected_behavior']==[] and d['response']['behavior_gap'] is None
|
|
request=InterpretationPromptBuilder().build(d['context'])
|
|
assert request['prompt_version']=='case_interpretation_resolved_v3'
|
|
assert 'provisional_behavior_gap' in request['output_schema']['required']
|
|
|
|
|
|
@pytest.mark.parametrize('field,refs,code',[
|
|
('described_behavior',['F0001'],'INTERPRETATION_DESCRIPTION_SOURCES'),
|
|
('described_behavior',['O0002'],'INTERPRETATION_SOURCE_KIND'),
|
|
('provisional_behavior_gap',['F0016'],'INTERPRETATION_PROVISIONAL_GAP_SOURCES'),
|
|
('provisional_behavior_gap',['O0002'],'INTERPRETATION_PROVISIONAL_GAP_SOURCES'),
|
|
])
|
|
def test_unrelated_facts_and_unilateral_gaps_are_rejected(field,refs,code):
|
|
d=fixture();statement=d['response'][field]
|
|
(statement[0] if isinstance(statement,list) else statement)['source_ids']=refs
|
|
with pytest.raises(InterpretationError,match=code):validate_interpretation(d['response'],d['context'])
|
|
|
|
|
|
def test_local_description_cannot_be_promoted_to_formal_expectation_or_verified():
|
|
d=fixture();d['response']['expected_behavior']=[{'text':'Expected','source_ids':['F0016']}]
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_SOURCE_KIND'):validate_interpretation(d['response'],d['context'])
|
|
d=fixture();d['response']['behavior_gap']={'text':'Formal gap','source_ids':['F0016','O0002']}
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_GAP_SOURCES'):validate_interpretation(d['response'],d['context'])
|
|
d=fixture();d['response']['described_behavior'][0]['verification']='verified'
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_SCHEMA'):validate_interpretation(d['response'],d['context'])
|
|
|
|
|
|
def test_protocol_v2_complete_context_still_replays_without_new_fields(built_case):
|
|
import asyncio
|
|
from viano_bug_analysis.analysis import CaseInterpreter
|
|
from viano_bug_analysis.analysis.entity_resolution import EntityResolver,policy
|
|
store=built_case[0];service=service_for(store);model=Model();owner=CaseInterpreter(store,gateway(model),project_research=service)
|
|
context=context_of(store)
|
|
EntityResolver(owner).save(context,'input',{'context':context,'policy':policy(SETTINGS),'binding':service.database.binding(),
|
|
'protocol_version':2,'interpretation_protocol_version':2})
|
|
result=asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status=='COMPLETED'
|
|
assert model.requests[-1]['prompt_version']=='case_interpretation_resolved_v2'
|
|
assert run(store,model).reused
|
|
|
|
|
|
def test_new_fields_are_carried_into_hypothesis_context_with_their_provenance():
|
|
from viano_bug_analysis.analysis.hypothesis_contracts import build_context
|
|
from viano_bug_analysis.analysis.hypothesis_config import HypothesisConfig,HypothesisLimits
|
|
d=fixture()
|
|
context=build_context(d['context'],{'interpretation':d['response']},'interpretation.json',
|
|
HypothesisConfig().policy(),[],HypothesisLimits())
|
|
descriptions=[s for s in context['interpretation_sources'].values() if s['kind']=='described_behavior']
|
|
gaps=[s for s in context['interpretation_sources'].values() if s['kind']=='provisional_behavior_gap']
|
|
assert descriptions[0]['data']['verification']=='not_independently_verified'
|
|
assert gaps[0]['data']['status']=='provisional'
|
|
|
|
|
|
def test_repeated_resolution_of_one_id_produces_one_entity_in_complete_pipeline(built_case):
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.package import PackageReader
|
|
plan={'searches':[{'catalog':'components','term':term,'role':'affected_subject','source_ids':['O0001']}
|
|
for term in ['switch','The switch']]}
|
|
class Repeated(Model):
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
if request['prompt_version']=='entity_resolution_assessment_v4':
|
|
self.requests.append(copy.deepcopy(request))
|
|
data=json.loads(request['messages'][1]['content'])
|
|
resolutions=[]
|
|
for index in range(2):
|
|
candidate=next(c for c in data['candidates'] if c['search_index']==index and c['id']=='17')
|
|
resolutions.append({'search_index':index,'candidate_id':candidate['candidate_id'],'status':'selected',
|
|
'source_ids':['O0001',f'V{index+1:04d}'],'reason':'Matching affected switch'})
|
|
return ProviderResponse({'resolutions':resolutions},'fixture-model',100,20)
|
|
return await super().generate(config,request,max_response_bytes=max_response_bytes)
|
|
store=built_case[0];model=Repeated(plan=plan);service=service_for(store)
|
|
result=run(store,model,service)
|
|
assert result.status=='COMPLETED'
|
|
with PackageReader(result.package) as reader:
|
|
doc=json.loads(reader.read_bytes(result.interpretation_ref));context=json.loads(reader.read_bytes(doc['context_ref']))
|
|
subjects=[s for s in context['sources'].values() if s['kind']=='entity' and s['data']['id']=='17']
|
|
assert len(subjects)==1 and len(subjects[0]['data']['lookup_associations'])==2
|
|
assert run(store,model).reused and len(model.requests)==3
|
|
|
|
|
|
def test_unbound_item_name_is_rejected_and_neutral_identity_is_accepted():
|
|
from jsonschema import Draft202012Validator
|
|
from viano_bug_analysis.analysis.quality import unnamed_entity_reasons
|
|
from viano_bug_analysis.analysis.contracts import output_schema
|
|
d=fixture()
|
|
d['context']['sources']['D0099']={'kind':'data_quality','case_pointer':'/interpretation_quality/identity',
|
|
'data':{'identity_policy':'explicit_names_v1'}}
|
|
reasons=unnamed_entity_reasons(d['context'])
|
|
item_sid=next(sid for sid in reasons if d['context']['sources'][sid]['data']['type']=='item')
|
|
for entity in d['response']['relevant_entities']:
|
|
if entity['entity_id'] in reasons:
|
|
entity['reason']=reasons[entity['entity_id']]
|
|
schema=output_schema(d['context'])
|
|
Draft202012Validator.check_schema(schema)
|
|
Draft202012Validator(schema).validate(d['response'])
|
|
validate_interpretation(d['response'],d['context'])
|
|
item=next(e for e in d['response']['relevant_entities'] if e['entity_id']==item_sid)
|
|
item['reason']='This item is the capture net, inferred from the nearby quest text.'
|
|
assert list(Draft202012Validator(schema).iter_errors(d['response']))
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_UNGROUNDED_ENTITY_LABEL'):
|
|
validate_interpretation(d['response'],d['context'])
|
|
# Historical v3 context remains governed by its original contract.
|
|
del d['context']['sources']['D0099']
|
|
validate_interpretation(d['response'],d['context'])
|
|
|
|
|
|
def test_explicit_entity_names_do_not_require_a_neutral_fallback():
|
|
from viano_bug_analysis.analysis.quality import unnamed_entity_reasons
|
|
context={'sources':{
|
|
'N0001':{'kind':'entity','data':{'type':'component','id':'1','name':'Verified switch'}},
|
|
'N0002':{'kind':'entity','data':{'type':'component','id':'2','lookup_associations':[{'name':'Localized switch'}]}},
|
|
'N0003':{'kind':'entity','data':{'type':'component','id':'3','roles':['required_tool']}},
|
|
'F0001':{'kind':'fact','data':{'value':'Use the named tool, no ID/name binding'}}}}
|
|
assert set(unnamed_entity_reasons(context))=={'N0003'}
|
|
|
|
|
|
def test_bound_name_requires_its_typed_fact_citation():
|
|
from jsonschema import Draft202012Validator
|
|
from viano_bug_analysis.analysis.quality import bind_entity_names, unnamed_entity_reasons
|
|
from viano_bug_analysis.analysis.contracts import output_schema
|
|
d=fixture()
|
|
ctx=d['context']
|
|
ctx['sources']['D0099']={'kind':'data_quality','case_pointer':'/interpretation_quality/identity',
|
|
'data':{'identity_policy':'explicit_names_v1','name_binding_policy':'typed_name_bindings_v1'}}
|
|
entity=next(e for e in d['response']['relevant_entities'] if e['entity_id']=='N0001')
|
|
data=ctx['sources']['N0001']['data']
|
|
ctx['sources']['F0099']={'kind':'fact','case_pointer':'/test/name','data':{'name':'project_entity_name',
|
|
'value':{'type':data['type'],'id':data['id'],'name':'Captured display name',
|
|
'provenance':{'source_ref':'capture.json','json_pointer':'/result/items/0/row/name','id_pointer':'/result/items/0/row/id'}}}}
|
|
bind_entity_names(ctx)
|
|
reasons=unnamed_entity_reasons(ctx)
|
|
assert 'N0001' not in reasons
|
|
for e in d['response']['relevant_entities']:
|
|
if e['entity_id'] in reasons:e['reason']=reasons[e['entity_id']]
|
|
entity['reason']='Captured display name is the interaction participant, not the missing subject.'
|
|
entity['source_ids']=['N0001']
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_ENTITY_NAME_SOURCE'):
|
|
validate_interpretation(d['response'],ctx)
|
|
entity['source_ids'].append('F0099')
|
|
validate_interpretation(d['response'],ctx)
|
|
Draft202012Validator(output_schema(ctx)).validate(d['response'])
|
|
assert InterpretationPromptBuilder.version(ctx)=='case_interpretation_resolved_v5'
|