143 lines
7.9 KiB
Python
143 lines
7.9 KiB
Python
"""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_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.
|
|
|
|
|
|
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']
|