226 lines
13 KiB
Python
226 lines
13 KiB
Python
"""Witnessed local graph paths, fixed Cypher boundaries and production routing."""
|
|
import copy
|
|
import os
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from tools.answers.question_plan import build_question_plan
|
|
from tools.knowledge.sources import profile_units
|
|
from tools.knowledge.evidence import STRUCTURAL_KINDS, relation_has_local_support
|
|
from tools.knowledge import local_graph as local, store
|
|
from tools.knowledge.graph_answers import render_similarity, graph_details
|
|
from tools.knowledge.concepts import substantive_concept
|
|
|
|
|
|
@pytest.mark.parametrize('concept',['články','poznámky','prehľad literatúry','reading papers','zdrojové kódy','Python'])
|
|
def test_generic_study_activities_cannot_be_a_research_similarity(concept):
|
|
assert not substantive_concept(concept)
|
|
|
|
|
|
@pytest.mark.parametrize('concept',['znalostný graf','RAG','NER','poznámky o RAG','PEFT','anotácia pomenovaných entít'])
|
|
def test_substantive_topics_are_not_removed_with_generic_words(concept):
|
|
assert substantive_concept(concept)
|
|
|
|
|
|
def graph_index():
|
|
rows=[]
|
|
entries=[('Alena Nová','alena_nova','Systém RAG používa PEFT.',['RAG','PEFT'],'RAG'),
|
|
('Peter Starý','peter_stary','Systém chatbot používa PEFT.',['chatbot','PEFT'],'chatbot'),
|
|
('Dana Mladá','dana_mlada','Systém RAG funguje.',['RAG'],None)]
|
|
for name,slug,text,concepts,subject in entries:
|
|
units=profile_units({'title':name,'path':f'pages/students/2022/{slug}/README.md',
|
|
'content':'# Diplomová práca 2026\nStretnutie 5.5.2026\nStav:\n- '+text},include_records=True)
|
|
for row in units:
|
|
row.setdefault('concepts',[]);row.setdefault('relations',[])
|
|
row.setdefault('annotation_status','structural')
|
|
if row['kind'] not in STRUCTURAL_KINDS:
|
|
row.update(annotation_status='knowledge',concepts=concepts,relations=[])
|
|
if subject:
|
|
relation={'subject':subject,'predicate':'uses','object':'PEFT',
|
|
'quote':subject+' používa PEFT.','verification':'supported'}
|
|
assert relation_has_local_support(row,relation)
|
|
row['relations']=[relation]
|
|
rows+=units
|
|
return {'version':'content-test','index_version':{'version':'source-test'},
|
|
'extraction_schema':2,'academic_schema':1,'units':rows,'profiles':3}
|
|
|
|
|
|
def bound(index,question='Kto pracoval na podobnej téme ako Alena Nová?'):
|
|
catalog=list({r['source_path']:{'kind':'Person','name':r['person'],
|
|
'source_document':r['source_path']} for r in index['units']}.values())
|
|
return build_question_plan(question,catalog,index_version=index['index_version'])
|
|
|
|
|
|
def witnessed_record(row):
|
|
return {'evidence':row,'person':{'name':row['person'],'source_path':row['source_path']},
|
|
'work':{'id':row['work_id'],'source_path':row['source_path'],'label':row['work_label'],
|
|
'type':row['work_type'],'year':row['work_year']},
|
|
'activity':row['id'],'concepts':[local.concept_key(c) for c in row['concepts']]}
|
|
|
|
|
|
@pytest.mark.parametrize('field,value', [('person','Other Person'),('work_id','wrong-work'),
|
|
('text','Invented result'),('phase','planned'),('source_path','other.md')])
|
|
def test_graph_evidence_cannot_change_checkpoint_identity_or_fact(field,value):
|
|
index=graph_index();row=next(r for r in index['units'] if r.get('concepts'))
|
|
record=witnessed_record(copy.deepcopy(row));record['evidence'][field]=value
|
|
with pytest.raises(ValueError):local.validated_row(record,{row['id']:row})
|
|
|
|
|
|
def test_graph_edges_must_bind_the_actual_work_and_concept():
|
|
row=next(r for r in graph_index()['units'] if r.get('concepts'))
|
|
record=witnessed_record(row)
|
|
record['work']['id']='another-work'
|
|
with pytest.raises(ValueError):local.validated_row(record,{row['id']:row})
|
|
record=witnessed_record(row);record['activity']=None
|
|
with pytest.raises(ValueError):local.validated_row(record,{row['id']:row})
|
|
record=witnessed_record(row);record['concepts'].append('invented topic')
|
|
with pytest.raises(ValueError):local.validated_row(record,{row['id']:row})
|
|
|
|
|
|
def test_large_witnesses_require_narrowing_instead_of_false_no_answer():
|
|
index=graph_index();a,b=[next(r for r in index['units'] if r['person']==name and r.get('concepts'))
|
|
for name in ['Alena Nová','Peter Starý']]
|
|
a['text']='Doložený veľmi dlhý výňatok. '*1000
|
|
result=local.LocalGraph(units=[a,b],seeds=[a],paths=[{'kind':'shared_topic','concept':'peft',
|
|
'anchor_id':a['id'],'peer_id':b['id']}])
|
|
response=render_similarity(bound(index).spec.question,index,bound(index),result)
|
|
assert response['status']=='ambiguous' and response['reason']=='graph_answer_budget'
|
|
|
|
|
|
def test_unverified_concept_cooccurrence_cannot_create_a_semantic_predicate():
|
|
index=graph_index();row=next(r for r in index['units'] if r.get('relations'))
|
|
record={'relation':{'predicate':'uses','quote':'RAG používa PEFT.',
|
|
**{k:row.get(k) for k in ('actor','work_id','phase','source_path','source_url','line')}},
|
|
'evidence':row['id'],'subject':'rag','object':'peft'}
|
|
assert local.validated_relation(record,{row['id']:row})
|
|
row['relations']=[]
|
|
with pytest.raises(ValueError):local.validated_relation(record,{row['id']:row})
|
|
|
|
|
|
def test_projection_upgrade_cannot_mix_a_different_sqlite_snapshot(tmp_path,monkeypatch):
|
|
from tools.knowledge import graph_projection as projection
|
|
from contextlib import nullcontext
|
|
index=graph_index();index['_checkpoint_sha256']='checkpoint'
|
|
monkeypatch.setattr(projection,'acquire_reindex_lock',lambda *a:nullcontext())
|
|
monkeypatch.setattr(projection,'publication_lock',lambda *a,**kw:nullcontext())
|
|
monkeypatch.setattr(projection,'read_version',lambda *a:index['index_version'])
|
|
monkeypatch.setattr(store,'load_index',lambda *a:index)
|
|
monkeypatch.setattr(projection,'sqlite_version',lambda *a:{'version':'different'})
|
|
monkeypatch.setattr(projection,'create_neo4j_driver',lambda *a:pytest.fail('No writes on snapshot mismatch'))
|
|
with pytest.raises(RuntimeError,match='different snapshots'):
|
|
projection.refresh_projection(tmp_path,tmp_path/'db.sqlite')
|
|
|
|
|
|
def test_similarity_shows_both_witnesses_and_only_two_verified_predicates_license_shared_use():
|
|
index=graph_index();a,b=[next(r for r in index['units'] if r['person']==name and r.get('concepts'))
|
|
for name in ['Alena Nová','Peter Starý']]
|
|
path={'kind':'shared_topic','concept':'peft','anchor_id':a['id'],'peer_id':b['id'],
|
|
'anchor_work_id':a['work_id'],'peer_work_id':b['work_id']}
|
|
result=local.LocalGraph(units=[a,b],seeds=[a],paths=[path])
|
|
response=render_similarity(bound(index).spec.question,index,bound(index),result)
|
|
assert response['accepted_for_delivery'] and len(response['source_urls'])==2
|
|
assert a['text'] in response['answer'] and b['text'] in response['answer']
|
|
assert 'Spoločný overený vzťah' not in response['answer']
|
|
result.relations=[{'evidence_id':r['id'],**r['relations'][0]} for r in [a,b]]
|
|
response=render_similarity(bound(index).spec.question,index,bound(index),result)
|
|
assert 'Spoločný overený vzťah: používa → PEFT' in response['answer']
|
|
assert len(response['graph_paths'])==1 and len(response['graph_relations'])==2
|
|
result.paths.append(dict(path))
|
|
response=render_similarity(bound(index).spec.question,index,bound(index),result)
|
|
assert response['answer'].count(a['text'])==1 and response['answer'].count(b['text'])==1
|
|
|
|
|
|
def test_planned_use_is_not_rendered_as_actual_shared_use():
|
|
index=graph_index();rows=[r for r in index['units'] if r['person'] in {'Alena Nová','Peter Starý'} and r.get('concepts')]
|
|
a,b=rows;b['phase']='planned'
|
|
result=local.LocalGraph(units=rows,seeds=[a],paths=[{'kind':'shared_topic','concept':'peft',
|
|
'anchor_id':a['id'],'peer_id':b['id']}],relations=[{'evidence_id':r['id'],**r['relations'][0]} for r in rows])
|
|
response=render_similarity(bound(index).spec.question,index,bound(index),result)
|
|
assert 'Spoločný overený vzťah' not in response['answer'] and '(plán' in response['answer']
|
|
|
|
|
|
def test_truncation_is_explicit_and_no_path_does_not_mean_no_similar_thesis_exists():
|
|
index=graph_index();result=local.LocalGraph(truncation_reasons=['graph_concept_limit'])
|
|
response=render_similarity(bound(index).spec.question,index,bound(index),result)
|
|
assert response['status']=='no_answer' and response['truncated'] and response['complete'] is False
|
|
assert 'neexistuje' in response['answer'] and 'Zúž' in response['answer']
|
|
|
|
|
|
def test_production_graph_delivery_uses_local_traversal_not_the_whole_graph(tmp_path,monkeypatch):
|
|
from tools.knowledge import query
|
|
from contextlib import nullcontext
|
|
index=graph_index();plan=bound(index)
|
|
monkeypatch.setattr(query,'index_snapshot',lambda *a,**kw:nullcontext())
|
|
monkeypatch.setattr(query,'read_version',lambda *a:index['index_version'])
|
|
monkeypatch.setattr(query,'load_index',lambda *a:index)
|
|
monkeypatch.setattr(query,'read_graph',lambda *a:pytest.fail('Full corpus graph scan'))
|
|
called=[]
|
|
monkeypatch.setattr(local,'retrieve_local_graph',lambda *a,**kw:called.append(kw) or local.LocalGraph())
|
|
response=query.deliver_content(plan.spec.question,engine='graphrag',limit=2,db_file=tmp_path/'db.sqlite',plan=plan)
|
|
assert called==[{'limit':2}] and response['reason']=='no_local_graph_connection'
|
|
|
|
|
|
@pytest.fixture
|
|
def live_graph(monkeypatch):
|
|
if os.getenv('RUN_NEO4J_TESTS')!='1':pytest.skip('Requires local Neo4j')
|
|
scope='local-graph-test-'+uuid.uuid4().hex
|
|
monkeypatch.setattr(store,'SCOPE',scope)
|
|
index=graph_index()
|
|
store.write_graph(index)
|
|
try:yield index
|
|
finally:
|
|
settings=store.get_neo4j_settings()
|
|
with store.create_neo4j_driver(settings) as driver,driver.session(database=settings.database) as session:
|
|
session.run('MATCH (n {graph_scope:$scope}) DETACH DELETE n',scope=scope).consume()
|
|
session.run('MATCH (s:KnowledgeState {scope:$scope}) DELETE s',scope=scope).consume()
|
|
|
|
|
|
def test_live_bounded_paths_join_people_works_activities_concepts_and_checked_relations(live_graph):
|
|
index=live_graph
|
|
result=local.retrieve_local_graph(bound(index).spec.question,index,bound(index))
|
|
assert result.paths and {r['person'] for r in result.units}=={'Alena Nová','Peter Starý','Dana Mladá'}
|
|
assert any(r['_graph_activity'] for r in result.units if r.get('concepts'))
|
|
assert {r['evidence_id'] for r in result.relations}=={r['id'] for r in index['units'] if r.get('relations')}
|
|
response=render_similarity(bound(index).spec.question,index,bound(index),result)
|
|
assert response['status']=='ok' and 'Spoločný overený vzťah' in response['answer']
|
|
|
|
|
|
def test_live_wrong_year_has_no_neighborhood_fallback(live_graph):
|
|
plan=bound(live_graph,'Aké problémy riešila Alena Nová pri diplomovej práci v roku 1999?')
|
|
result=local.retrieve_local_graph(plan.spec.question,live_graph,plan)
|
|
assert result.units==[] and result.paths==[]
|
|
|
|
|
|
def test_live_shared_articles_do_not_introduce_an_unrelated_student(live_graph):
|
|
index=live_graph
|
|
for row in index['units']:
|
|
if row['person']=='Alena Nová' and row.get('concepts'):
|
|
row['text']+='\nPrečítané články.';row['end_line']+=1;row['concepts'].append('články')
|
|
extra=profile_units({'title':'Cyril Nový','path':'pages/students/2022/cyril_novy/README.md',
|
|
'content':'# Diplomová práca 2026\nStav:\n- Prečítané články o nesúvisiacej téme.'},include_records=True)
|
|
for row in extra:
|
|
row.update(concepts=[] if row['kind'] in STRUCTURAL_KINDS else ['články'],relations=[],
|
|
annotation_status='structural' if row['kind'] in STRUCTURAL_KINDS else 'knowledge')
|
|
index['units']+=extra
|
|
store.write_graph(index)
|
|
result=local.retrieve_local_graph(bound(index).spec.question,index,bound(index))
|
|
assert result.paths and not any(r['person']=='Cyril Nový' for r in result.units)
|
|
assert not any(p['concept']=='clanky' for p in result.paths)
|
|
|
|
|
|
def test_live_projection_version_and_tampered_evidence_fail_closed(live_graph):
|
|
with pytest.raises(RuntimeError):local.retrieve_local_graph(bound(live_graph).spec.question,
|
|
{**live_graph,'version':'different'},bound(live_graph))
|
|
settings=store.get_neo4j_settings()
|
|
with store.create_neo4j_driver(settings) as driver,driver.session(database=settings.database) as session:
|
|
session.run("MATCH (e:KnowledgeEvidence {graph_scope:$scope,person:'Alena Nová'}) SET e.text='Invented result'",scope=store.SCOPE).consume()
|
|
with pytest.raises(ValueError):local.retrieve_local_graph(bound(live_graph).spec.question,live_graph,bound(live_graph))
|
|
|
|
|
|
def test_live_evidence_budget_reports_actual_clipping(live_graph,monkeypatch):
|
|
monkeypatch.setattr(local,'EVIDENCE_LIMIT',1)
|
|
plan=bound(live_graph,'Aké metódy použila Alena Nová?')
|
|
result=local.retrieve_local_graph(plan.spec.question,live_graph,plan)
|
|
assert len(result.units)==1 and 'graph_evidence_limit' in result.truncation_reasons
|