dp-zp-agent/test/test_source_audit.py
2026-11-07 12:11:41 +01:00

174 lines
8.2 KiB
Python

"""Source-link availability, lifecycle qualifications and supervisor semantics."""
import io
import json
import urllib.error
import pytest
from tools.answers import reference_links as links
from tools.answer_evaluation.content_selection import reference_only,evidence_phase,facets_for
from tools.knowledge.sources import profile_units
from tools.knowledge.query import answer_from_index
from tools.retrieval.person_names import resolve_person_names
REPO='https://git.kemt.fei.tuke.sk/sh662er/Rasa'
SOURCE='https://zp.kemt.fei.tuke.sk/students/2019/samuel_horani'
def result(text):
return {'accepted_for_delivery':True,'status':'ok','answer':text,'source_urls':[SOURCE],'claims':[]}
@pytest.mark.parametrize('text',[
'- [Repozitár s chatbotom]('+REPO+')',
'Repozitár: '+REPO,
'- Výstup je tu '+REPO,
])
def test_a_bare_reference_is_not_a_completed_result(text):
assert reference_only(text)
assert evidence_phase(text,'progress','result')=='unknown'
assert facets_for(text,'progress','Aké výsledky dosiahol?','result')==[]
def test_real_implementation_with_a_reference_remains_an_implementation():
text='Implementovaný chatbot používa RASA. Kód: '+REPO
assert not reference_only(text)
assert evidence_phase(text,'progress')=='result'
@pytest.mark.parametrize('text',[
'[Repozitár]('+REPO+')',REPO,
'[Repozitár][r]\n\n[r]: '+REPO,
'<a href="'+REPO+'">Repozitár</a>',
])
def test_unverified_repository_is_not_delivered_as_a_clickable_link(text):
value=links.checked_repository_links(result(text),check=lambda _: 'not_public')
assert REPO not in value['answer'] and value['source_urls']==[SOURCE]
assert value['accepted_for_delivery'] and 'súkromný' in value['answer']
assert value['suppressed_repository_links']==1
def test_verified_repository_preserves_the_exact_answer_and_sources():
value=result('[Repozitár]('+REPO+')')
assert links.checked_repository_links(value,check=lambda _: 'verified')==value
def test_html_escaped_markdown_reference_cannot_bypass_link_suppression():
url='https://github.com/example/missing?x=1&amp;y=2'
value=links.checked_repository_links(result('[Repozitár]('+url+')'),check=lambda _: 'unverified')
assert 'github.com' not in value['answer'] and '](' not in value['answer']
def test_unverifiable_repository_cannot_be_the_only_factual_citation():
value={**result('Práca je hotová.'),'source_urls':[REPO]}
checked=links.checked_repository_links(value,check=lambda _: 'unverified')
assert not checked['accepted_for_delivery'] and checked['reason']=='reference_unavailable'
assert not checked['source_urls']
@pytest.mark.parametrize('url',[
'http://git.kemt.fei.tuke.sk/a/b','https://git.kemt.fei.tuke.sk:8443/a/b',
'https://user:password@git.kemt.fei.tuke.sk/a/b','https://git.internal/a/b',
'https://github.com/a/b?access_token=private','https://git.kemt.fei.tuke.sk/../admin',
])
def test_repository_probe_never_contacts_unapproved_destinations(monkeypatch,url):
monkeypatch.setattr(links.urllib.request,'build_opener',lambda *a:pytest.fail('Unsafe network request'))
assert links._probe.__wrapped__(url,0)=='unverified'
@pytest.mark.parametrize('status,expected',[(404,'not_public'),(401,'not_public'),(403,'not_public'),(302,'unverified'),(503,'unverified')])
def test_repository_api_failures_are_not_treated_as_success(monkeypatch,status,expected):
class Opener:
def open(self,request,**kw):
assert request.full_url.endswith('/api/v1/repos/sh662er/Rasa')
assert 'Authorization' not in request.headers
raise urllib.error.HTTPError(request.full_url,status,'Unavailable',{},None)
monkeypatch.setattr(links.urllib.request,'build_opener',lambda *a:Opener())
assert links._probe.__wrapped__(REPO,0)==expected
def test_public_repo_api_metadata_is_required(monkeypatch):
class Opener:
def open(self,request,**kw):return io.BytesIO(json.dumps({'private':False,'full_name':'sh662er/Rasa'}).encode())
monkeypatch.setattr(links.urllib.request,'build_opener',lambda *a:Opener())
assert links._probe.__wrapped__(REPO,0)=='verified'
def index():
doc={'title':'Alena Nová','path':'pages/students/2022/alena/README.md',
'content':'# Diplomová práca 2024\nNázov: Overené grafy\nVedúci: Daniel Hládek\n'
'Stav:\n- Implementovaný systém RAG.\nUkončil predčasne\n'}
rows=profile_units(doc)
return {'units':rows,'version':'fixture','index_version':{'version':'fixture'}}
def test_lifecycle_qualification_is_preserved_even_when_it_has_only_two_words():
rows=index()['units']
note=next(r for r in rows if r['kind']=='lifecycle')
assert note['text']=='Ukončil predčasne' and note['work_year']==2024
r=answer_from_index('Ako pokročila Alena Nová?',index(),engine='rag')
assert 'Ukončil predčasne' in r['answer'] and any(e['evidence_role']=='lifecycle' for e in r['evidence'])
def test_supervisor_year_is_never_satisfied_by_a_historical_work():
r=answer_from_index('Ktoré práce vedie Daniel Hladek v roku 2026?',index(),engine='rag')
assert r['status']=='no_answer' and r['reason']=='supervision_not_documented'
assert '2026' in r['answer'] and 'autor dokumentu' in r['answer'].lower()
assert not r['evidence'] and not r['source_urls']
def test_current_supervision_is_not_inferred_from_a_formal_supervisor_line():
r=answer_from_index('Ktoré práce aktuálne vedie Daniel Hladek?',index(),engine='graphrag')
assert r['status']=='no_answer' and r['reason']=='active_supervision_unknown'
assert 'historickom' in r['answer'] and not r['evidence']
def test_missing_supervisor_in_followup_requires_a_name_and_keeps_the_year():
r=answer_from_index('Ktoré práce sú teraz vedené v roku 2026?',index(),engine='rag')
assert r['status']=='ambiguous' and r['reason']=='missing_supervisor'
assert '2026' in r['answer']
def test_title_typo_is_not_misclassified_as_a_person():
catalog=[{'name':'Daniel Hládek','kind':'Person'}]
r=resolve_person_names('Ktoré práce aktuálne vedie Daniel Hladek a nazv prác?',catalog)
assert r['status']=='ok' and [m['name'] for m in r['matches']]==['Daniel Hládek']
def test_dated_status_overrides_previous_meeting_instead_of_inheriting_its_year():
from tools.knowledge.sources import meeting_date
assert meeting_date('Stav 8.11.2024:')['date']=='2024-11-08'
rows=profile_units({'title':'Alena Nová','path':'pages/students/2022/alena/README.md',
'content':'# Bakalárska práca 2025\nStretnutie 4.2.2025\nStav:\n- Implementovaný systém RAG.\n\n'
'Stav 8.11.2024:\n- Urobené poznámky o RAG.\n'})
assert next(r for r in rows if 'poznámky' in r['text'])['date']=='2024-11-08'
def test_new_lifecycle_note_does_not_reextract_unchanged_later_profiles(tmp_path):
import hashlib
from tools.knowledge.build import build
from tools.sync.sync_indexes import source_files
from tools.sync.index_state import durable_json
wiki=tmp_path/'wiki';data=tmp_path/'data';data.mkdir()
paths=[]
for slug,name in [('alena','Alena Nová'),('boris','Boris Starý')]:
p=wiki/f'pages/students/2022/{slug}/README.md';p.parent.mkdir(parents=True)
p.write_text(f'---\ntitle: {name}\n---\n# Diplomová práca 2026\nStav:\n- Implementovaný systém RAG.\n')
paths.append(p)
def version(number):
return {'version':str(number),'source_sha256':hashlib.sha256(json.dumps(source_files(wiki),sort_keys=True).encode()).hexdigest()}
def complete(messages):
return {'choices':[{'message':{'content':json.dumps({'items':[]})}}]}
durable_json(data/'index_version.json',version(1))
build(wiki,data,complete,{'model':'fixture'},graph=False,vectors=False)
old=json.loads((data/'knowledge/index.json').read_text())
old_ids={u['person']:u['id'] for u in old['units']}
paths[0].write_text(paths[0].read_text()+'\nUkončil predčasne\n')
durable_json(data/'index_version.json',version(2))
report=build(wiki,data,complete,{'model':'fixture'},graph=False,vectors=False)
new=json.loads((data/'knowledge/index.json').read_text())
assert report['llm_batches']==0
assert all(u['id']==old_ids[u['person']] for u in new['units'] if u['kind']!='lifecycle')
assert any(u['kind']=='lifecycle' for u in new['units'])