dp-zp-agent/test/test_delivery_contract.py

134 lines
7.6 KiB
Python

"""Adversarial checks grade request fulfilment, not transport acceptance."""
import copy
import json
import pytest
from tools.answers import delivery_contract as c
from tools.answers.question_plan import QuestionPlan,PersonRequest,WorkRequest,Period,MeetingRequest
from tools.knowledge.sources import profile_units
def fixture():
units=profile_units({'title':'Eva Nová','path':'pages/students/2022/eva_nova/README.md',
'content':'# Diplomová práca 2027\nStretnutie 5.11.2026\nStav:\n- Funguje model BERT.\n'
'Stretnutie 1.10.2026\nÚlohy:\n- Vyhodnoťte model.'},include_records=True)
row=next(r for r in units if 'Funguje' in r['text'])
plan=QuestionPlan(question='Aké výsledky má Eva Nová pri diplomovej práci 2027?',intent='facets',
people=[PersonRequest(name='Eva Nová')],works=[WorkRequest(person='Eva Nová',work_type='diploma_thesis',years=[2027])],facets=['results'])
result={'status':'ok','accepted_for_delivery':True,'index_version':{'version':'v'},
'answer':'Eva Nová: funguje model BERT.\nZdroj: '+row['source_url'],
'evidence':[row],'claims':[],'source_urls':[row['source_url']],'presentation_notes':[]}
return plan,result,{'index_version':{'version':'v'},'units':units}
def audit(**changes):
def model(messages):
p=json.loads(messages[-1]['content'])
value={'plan_matches_request':True,'all_requirements_identified':True,'facts_supported':True,'clear':True,
'requirements':[{'description':p['resolved_question'],'question_segments':['Q1'],'status':'answered',
'answer_segments':[p['answer_segments'][0]['id']],'evidence_ids':['E1']}]}
value.update(changes)
return {'choices':[{'message':{'content':json.dumps(value)}}]}
return model
@pytest.mark.parametrize('engine',['rag','graphrag'])
def test_matching_request_plan_original_evidence_and_final_answer_pass(engine):
plan,result,index=fixture()
out=c.enforce(plan.question,engine,plan,result,index=index,complete=audit())
assert out['accepted_for_delivery'] and out['delivery_check']['code']=='complete'
assert out['delivery_check']['request_fingerprint']==c.fingerprint(plan.question,engine)
@pytest.mark.parametrize('change,code',[
({'person':'Dana Iná'},'wrong_person'),({'work_year':2024},'wrong_work_or_meeting'),
({'phase':'planned'},'altered_evidence'),({'text':'Dokončila celú prácu.'},'altered_evidence')])
def test_true_but_wrong_or_relabelled_evidence_cannot_be_waived_by_friendly_model(change,code):
plan,result,index=fixture();result=copy.deepcopy(result);result['evidence'][0].update(change)
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=audit())
assert not out['accepted_for_delivery'] and out['delivery_check']['code']==code
assert 'Dokončila' not in out['answer'] and 'evidence' not in out
@pytest.mark.parametrize('field',['record_date','event_date','target_date'])
def test_temporal_range_is_checked_even_outside_meeting_route(field):
plan,result,index=fixture();plan.period=Period(start='2026-10-01',end='2026-10-01',field=field)
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=audit())
assert out['delivery_check']['code']=='wrong_period'
def test_latest_is_verified_against_published_records_not_only_the_selected_old_record():
plan,result,index=fixture();plan.meeting=MeetingRequest(selection='latest')
result['evidence']=[next(r for r in index['units'] if 'Vyhodnoťte' in r['text'])]
assert c.enforce(plan.question,'graphrag',plan,result,index=index,complete=audit())['delivery_check']['code']=='wrong_work_or_meeting'
def test_missing_person_is_not_silently_removed_from_comparison():
plan,result,index=fixture();plan.people.append(PersonRequest(name='Dana Iná'))
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=audit())
assert out['delivery_check']['code']=='missing_person'
@pytest.mark.parametrize('change',[
{'all_requirements_identified':False},{'facts_supported':False},{'plan_matches_request':False},{'clear':False},
{'requirements':[{'description':'results','question_segments':['Q1'],'status':'answered','answer_segments':['invented'],'evidence_ids':['fake']}]}])
def test_delivery_is_not_success_if_request_or_final_text_fails_review(change):
plan,result,index=fixture()
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=audit(**change))
assert not out['accepted_for_delivery'] and out['reason']=='answer_rejected'
assert 'evidence' not in out and 'synthesis' not in out
def test_explicit_missing_person_is_partial_only_when_missing_notice_is_really_in_answer():
plan,result,index=fixture();plan.question+=' A Dana Iná?';plan.people.append(PersonRequest(name='Dana Iná'))
notice='Dana Iná: požadované podklady sa nenašli.'
result['presentation_notes']=[notice];result['answer']+='\n\n'+notice
model=audit(requirements=[{'description':'Dana Iná','question_segments':['Q1'],'status':'missing_disclosed','answer_segments':['A2'],'evidence_ids':[]},
{'description':'Eva Nová','question_segments':['Q1'],'status':'answered','answer_segments':['A1'],'evidence_ids':['E1']}])
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=model)
assert out['accepted_for_delivery'] and out['delivery_check']['code']=='partial'
def test_distributed_requirement_uses_question_references_without_manufacturing_a_quote():
plan,result,index=fixture()
model=audit(requirements=[{'description':'Results of the specified person and work',
'question_segments':['Q1'],'status':'answered','answer_segments':['A1'],'evidence_ids':['E1']}])
assert c.enforce(plan.question,'rag',plan,result,index=index,complete=model)['accepted_for_delivery']
def test_unknown_question_segment_cannot_pass_review():
plan,result,index=fixture()
model=audit(requirements=[{'description':'Results','question_segments':['Q999'],
'status':'answered','answer_segments':['A1'],'evidence_ids':['E1']}])
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=model)
assert out['delivery_check']['code']=='invalid_review'
def test_negative_semantic_verdict_is_never_retried_to_get_acceptance():
plan,result,index=fixture();calls=[]
def model(messages):
calls.append(1)
return audit(facts_supported=False)(messages)
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=model)
assert out['delivery_check']['code']=='request_not_fulfilled' and len(calls)==1
def test_verification_outage_is_distinct_and_diagnostic_does_not_leak_exception():
plan,result,index=fixture()
def down(_):raise RuntimeError('SECRET_API_KEY_OR_PROVIDER_RESPONSE')
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=down)
assert out['reason']=='verification_unavailable' and out['delivery_check']['layer']=='verification'
assert 'SECRET' not in json.dumps(out)
def test_self_declared_foreign_citation_does_not_authorize_itself():
plan,result,index=fixture();result['source_urls'].append('https://invented.test')
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=audit())
assert out['delivery_check']['code']=='invalid_citation'
def test_metadata_work_filters_are_checked_against_actual_execution_plan():
plan,result,index=fixture();plan.intent='metadata';result['evidence']=[]
result['claims']=[{'person':'Eva Nová','work_type':'bachelor_thesis','year':2024}]
out=c.enforce(plan.question,'rag',plan,result,index=index,complete=lambda _:pytest.fail('Exact metadata does not need prose review'))
assert out['delivery_check']['code']=='wrong_work_or_meeting'