"""Publication is enforced by actual Cypher, before aggregation/traversal.""" import json import os import uuid import pytest from tools.graphrag import graph_search as gs from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings pytestmark = pytest.mark.skipif(os.getenv('RUN_NEO4J_TESTS') != '1', reason='Requires local Neo4j.') @pytest.fixture def graph(monkeypatch): scope = 'publication-' + uuid.uuid4().hex monkeypatch.setattr(gs, 'GRAPH_SCOPE', scope) settings = get_neo4j_settings() with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: tx = session.begin_transaction() try: for key, name, published in [('a','Alena Nová',True),('b','Boris Iný',None),('c','Cyril Malý',False)]: tx.run('''CREATE (p:Person {id:$id, name:$name, graph_scope:$scope, source_document:$path}) CREATE (d:Document {path:$path, published:$published, graph_scope:$scope, source_url:$url}) CREATE (t:Thesis {id:$thesis, title:$title, year:2024, thesis_type:'bachelor_thesis', source_document:$path, graph_scope:$scope}) MERGE (k:Keyword {id:$tag, name:'sharedcheck', graph_scope:$scope}) CREATE (p)-[:HAS_DOCUMENT]->(d)-[:HAS_KEYWORD]->(k) CREATE (p)-[:AUTHORED]->(t)-[:EVIDENCED_BY]->(d)''', id=scope+key, name=name, scope=scope, path=scope+'/'+key+'.md', published=published, url='https://example.test/'+key, thesis=scope+'th'+key, title='Výskum '+key,tag=scope+'shared').consume() tx.run('''MATCH (p:Person {id:$person}), (public:Document {path:$public}) CREATE (d:Document {path:$path, published:false, source_url:'https://example.test/private', graph_scope:$scope}) CREATE (k:Keyword {id:$keyword, name:'hiddentag', graph_scope:$scope}) CREATE (c:Category {name:$category, graph_scope:$scope}) CREATE (a:DocumentAuthor {id:$author, name:'Hidden Author', graph_scope:$scope}) CREATE (topic:Topic {id:$topic, name:'Hidden Topic', source_document:$path, graph_scope:$scope}) CREATE (p)-[:HAS_DOCUMENT]->(d)-[:HAS_KEYWORD]->(k) CREATE (d)-[:IN_CATEGORY]->(c) CREATE (a)-[:AUTHORED_DOCUMENT]->(d) CREATE (d)-[:DESCRIBES]->(topic) CREATE (public)-[:DESCRIBES]->(topic)''', person=scope+'a',public=scope+'/a.md', path=scope+'/private.md',scope=scope, keyword=scope+'hidden',category=scope+'hidden-category',author=scope+'author',topic=scope+'topic').consume() yield tx, scope finally: tx.rollback() def test_catalog_hides_private_people_and_private_only_vocabulary(graph): tx, _ = graph public = gs.run_rows(tx, gs.CATALOG_QUERY, scope=gs.GRAPH_SCOPE, published_only=True) names = {r.get('name') for r in public} assert 'Alena Nová' in names and 'Boris Iný' in names assert not names & {'Cyril Malý','hiddentag','Hidden Author','Hidden Topic'} all_rows = gs.run_rows(tx, gs.CATALOG_QUERY, scope=gs.GRAPH_SCOPE, published_only=False) assert {'Cyril Malý','hiddentag','Hidden Author','Hidden Topic'} <= {r.get('name') for r in all_rows} def test_traversal_and_citations_do_not_expose_hidden_documents(graph): tx, scope = graph result = gs.search_graph_with_session(tx, 'Alena Nová') serialized = json.dumps(result, ensure_ascii=False) assert scope+'/private.md' not in serialized assert 'Hidden Topic' not in serialized and 'hiddentag' not in serialized all_rows = gs.search_graph_with_session(tx, 'Alena Nová', published_only=False) assert scope+'/private.md' in json.dumps(all_rows) def test_global_count_is_filtered_before_counting(graph): tx, _ = graph public = gs.search_graph_with_session(tx, 'Koľko osôb má prácu v roku 2024?') all_rows = gs.search_graph_with_session(tx, 'Koľko osôb má prácu v roku 2024?', published_only=False) assert public['aggregation']['count'] == 2 assert set(public['aggregation']['people']) == {'Alena Nová','Boris Iný'} assert all_rows['aggregation']['count'] == 3 def test_private_person_is_not_resolved_as_a_public_fact(graph): tx, scope = graph result = gs.search_graph_with_session(tx, 'Aký názov práce má osoba Cyril Malý v roku 2024?') assert result['status']=='no_answer' and result['facts']==[] and result['documents']==[] all_rows = gs.search_graph_with_session(tx, 'Aký názov práce má osoba Cyril Malý v roku 2024?', published_only=False) assert all_rows['status']=='ok' and any(f.get('person')=='Cyril Malý' for f in all_rows['facts']) def test_private_child_tag_is_not_used_in_conjunction_or_metadata(graph): tx, _ = graph result = gs.search_graph_with_session(tx, 'Aký názov práce má osoba Alena Nová v roku 2024 s kľúčovým slovom hiddentag?') assert result['status']=='no_answer' and not result['facts'] all_rows = gs.search_graph_with_session(tx, 'Aký názov práce má osoba Alena Nová v roku 2024 s kľúčovým slovom hiddentag?', published_only=False) assert all_rows['status']=='ok' and any(f.get('keyword')=='hiddentag' for f in all_rows['facts']) def test_shared_keyword_does_not_include_private_person(graph): tx, _ = graph question = 'Majú osoby Alena Nová a Cyril Malý spoločné kľúčové slovo sharedcheck?' public = gs.search_graph_with_session(tx, question) assert public['status']=='no_answer' and not public['facts'] all_rows = gs.search_graph_with_session(tx, question, published_only=False) assert any(f['fact_type']=='shared_keyword' for f in all_rows['facts']) def test_graph_contract_distinguishes_exact_count_from_trimmed_evidence(graph): from tools.retrieval.retrieval_contract import RetrievalResponse tx, _ = graph result = gs.search_graph_with_session(tx, 'Koľko osôb má prácu v roku 2024?', limit=1) RetrievalResponse.model_validate(result) assert result['status']=='ok' and result['aggregation']['count']==2 assert result['aggregation']['complete'] is True assert result['complete'] is False and result['truncated'] assert 'fact_limit' in result['truncation_reasons'] assert {r['document_path'] for r in result['sources']} == {r['document_path'] for r in result['documents']} assert all(r['source_id'].startswith('S') for r in result['sources']) assert result['index_version'] is None # Raw fixture session has no published snapshot. @pytest.mark.parametrize('year,limit,expected',[(2024,1,2),(2024,2,2),(2024,3,2),(1999,1,0)]) def test_delivery_limit_uses_actual_cypher_and_keeps_verified_total(graph,monkeypatch,year,limit,expected): from tools.answers import answer_delivery as delivery tx,_=graph;calls=[] def search(question,**kwargs): calls.append(kwargs['limit']) return gs.search_graph_with_session(tx,question,**kwargs) monkeypatch.setattr(delivery,'search_graph',search) def select(messages): context=json.loads(messages[1]['content'])['graph_context'] answer={'status':'ok','answer':'','count':context['aggregation']['count'], 'claims':[{k:f[k] for k in ['person','title','year','work_type'] if f.get(k) is not None} for f in context['facts']], 'source_urls':list({f['source_url'] for f in context['facts'] if f.get('source_url')})} return {'choices':[{'message':{'content':json.dumps(answer)}}]} r=delivery.deliver_answer(f'Koľko osôb má prácu v roku {year}?',engine='graphrag',limit=limit,complete=select) assert calls==[limit] and r['accepted_for_delivery'] assert r['count']==expected and len(r['claims'])==min(limit,expected) if expected: assert set(r['source_urls'])=={'https://example.test/a','https://example.test/b'} assert r['truncated']==(expected>limit) if expected>limit: assert 'Výpis je neúplný.' in r['answer'] and 'Výpis môžeš zúžiť' in r['answer'] else: assert 'Výpis je neúplný.' not in r['answer'] def test_small_limit_keeps_actual_tag_proof_with_the_same_person(graph,monkeypatch): from tools.answers import answer_delivery as delivery tx,_=graph q='Koľko osôb má kľúčové slovo sharedcheck a zároveň prácu v roku 2024?' monkeypatch.setattr(delivery,'search_graph',lambda question,**kw:gs.search_graph_with_session(tx,question,**kw)) def select(messages): c=json.loads(messages[1]['content'])['graph_context'] assert len(c['answer_candidates'])==1 candidate=c['answer_candidates'][0] answer={'status':'ok','answer':'','count':c['aggregation']['count'], 'claims':[{k:candidate[k] for k in ['person','year','keyword']}], 'source_urls':list({f['source_url'] for f in c['facts'] if f.get('person')==candidate['person'] and f.get('source_url')})} return {'choices':[{'message':{'content':json.dumps(answer)}}]} r=delivery.deliver_answer(q,engine='graphrag',limit=1,complete=select) assert r['accepted_for_delivery'] and r['count']==2 and len(r['claims'])==1 assert r['truncated'] and 'Výpis je neúplný.' in r['answer'] assert 'Cyril Malý' not in r['answer'] def test_graph_traversal_reports_database_limit_not_just_python_slicing(graph): tx, _ = graph result = gs.search_graph_with_session(tx, 'Alena Nová', limit=1) assert len(result['paths'])==1 and result['truncated'] and result['complete'] is False assert 'path_limit' in result['truncation_reasons'] full = gs.search_graph_with_session(tx, 'Alena Nová', limit=100) assert full['complete'] is True and not full['truncated'] def test_graph_no_answer_keeps_common_fields_and_reason(graph): tx, _ = graph result = gs.search_graph_with_session(tx, 'Aký názov práce má osoba Cyril Malý v roku 2024?') assert result['status']=='no_answer' and result['reason']=='unknown_person' assert result['sources']==[] and not result['truncated'] and result['index_version'] is None def test_ambiguous_graph_response_cannot_claim_completeness(graph): tx, scope = graph tx.run('''CREATE (p:Person {id:$id, name:'Alena Nová', source_document:$path, graph_scope:$scope}) CREATE (d:Document {path:$path, published:true, graph_scope:$scope}) CREATE (p)-[:HAS_DOCUMENT]->(d)''', id=scope+'duplicate', path=scope+'/duplicate.md', scope=scope).consume() result = gs.search_graph_with_session(tx, 'Aký názov práce má osoba Alena Nová v roku 2024?') assert result['status']=='ambiguous' and result['reason']=='ambiguous_person' assert result['complete'] is False and not result['truncated']