apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-23 12:11:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Tests

Browse source
This commit is contained in:
Cyber MacGeddon 2026-01-27 00:29:23 +00:00
parent 6b7b710fe5
commit 2f5d639938
3 changed files with 121 additions and 121 deletions
Download patch file Download diff file Expand all files Collapse all files

6
tests/unit/test_extract/test_ontology/test_ontology_triples.py
View file

@ -227,7 +227,7 @@ class TestOntologyTripleGeneration:
assert len(serves_range_triples) == 1, \ assert len(serves_range_triples) == 1, \
"Should generate range triple for datatype property" "Should generate range triple for datatype property"
assert serves_range_triples[0].o.value == "http://www.w3.org/2001/XMLSchema#string", \ assert serves_range_triples[0].o.iri == "http://www.w3.org/2001/XMLSchema#string", \
"Range should be XSD type URI (xsd:string expanded)" "Range should be XSD type URI (xsd:string expanded)"
def test_generates_triples_for_all_classes(self, extractor, sample_ontology_subset): def test_generates_triples_for_all_classes(self, extractor, sample_ontology_subset):
@ -238,7 +238,7 @@ class TestOntologyTripleGeneration:
class_subjects = set( class_subjects = set(
t.s.iri for t in triples t.s.iri for t in triples
if t.p.iri == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" if t.p.iri == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
and t.o.value == "http://www.w3.org/2002/07/owl#Class" and t.o.iri == "http://www.w3.org/2002/07/owl#Class"
) )
assert len(class_subjects) == 3, \ assert len(class_subjects) == 3, \
@ -252,7 +252,7 @@ class TestOntologyTripleGeneration:
property_subjects = set( property_subjects = set(
t.s.iri for t in triples t.s.iri for t in triples
if t.p.iri == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" if t.p.iri == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
and ("ObjectProperty" in t.o.value or "DatatypeProperty" in t.o.value) and ("ObjectProperty" in t.o.iri or "DatatypeProperty" in t.o.iri)
) )
assert len(property_subjects) == 3, \ assert len(property_subjects) == 3, \

36
tests/unit/test_extract/test_ontology/test_prompt_and_extraction.py
View file

@ -8,7 +8,7 @@ and extracts/validates triples from LLM responses.
import pytest import pytest
from trustgraph.extract.kg.ontology.extract import Processor from trustgraph.extract.kg.ontology.extract import Processor
from trustgraph.extract.kg.ontology.ontology_selector import OntologySubset from trustgraph.extract.kg.ontology.ontology_selector import OntologySubset
from trustgraph.schema.core.primitives import Triple, Value from trustgraph.schema.core.primitives import Triple, Term, IRI, LITERAL
@pytest.fixture @pytest.fixture
@ -248,9 +248,9 @@ class TestTripleParsing:
validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset) validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset)
assert len(validated) == 1, "Should parse one valid triple" assert len(validated) == 1, "Should parse one valid triple"
assert validated[0].s.value == "https://trustgraph.ai/food/cornish-pasty" assert validated[0].s.iri == "https://trustgraph.ai/food/cornish-pasty"
assert validated[0].p.value == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" assert validated[0].p.iri == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
assert validated[0].o.value == "http://purl.org/ontology/fo/Recipe" assert validated[0].o.iri == "http://purl.org/ontology/fo/Recipe"
def test_parse_multiple_triples(self, extractor, sample_ontology_subset): def test_parse_multiple_triples(self, extractor, sample_ontology_subset):
"""Test parsing multiple triples.""" """Test parsing multiple triples."""
@ -307,11 +307,11 @@ class TestTripleParsing:
assert len(validated) == 1 assert len(validated) == 1
# Subject should be expanded to entity URI # Subject should be expanded to entity URI
assert validated[0].s.value.startswith("https://trustgraph.ai/food/") assert validated[0].s.iri.startswith("https://trustgraph.ai/food/")
# Predicate should be expanded to ontology URI # Predicate should be expanded to ontology URI
assert validated[0].p.value == "http://purl.org/ontology/fo/produces" assert validated[0].p.iri == "http://purl.org/ontology/fo/produces"
# Object should be expanded to class URI # Object should be expanded to class URI
assert validated[0].o.value == "http://purl.org/ontology/fo/Food" assert validated[0].o.iri == "http://purl.org/ontology/fo/Food"
def test_creates_proper_triple_objects(self, extractor, sample_ontology_subset): def test_creates_proper_triple_objects(self, extractor, sample_ontology_subset):
"""Test that Triple objects are properly created.""" """Test that Triple objects are properly created."""
@ -324,12 +324,12 @@ class TestTripleParsing:
assert len(validated) == 1 assert len(validated) == 1
triple = validated[0] triple = validated[0]
assert isinstance(triple, Triple), "Should create Triple objects" assert isinstance(triple, Triple), "Should create Triple objects"
assert isinstance(triple.s, Value), "Subject should be Value object" assert isinstance(triple.s, Term), "Subject should be Term object"
assert isinstance(triple.p, Value), "Predicate should be Value object" assert isinstance(triple.p, Term), "Predicate should be Term object"
assert isinstance(triple.o, Value), "Object should be Value object" assert isinstance(triple.o, Term), "Object should be Term object"
assert triple.s.is_uri, "Subject should be marked as URI" assert triple.s.type == IRI, "Subject should be IRI type"
assert triple.p.is_uri, "Predicate should be marked as URI" assert triple.p.type == IRI, "Predicate should be IRI type"
assert not triple.o.is_uri, "Object literal should not be marked as URI" assert triple.o.type == LITERAL, "Object literal should be LITERAL type"
class TestURIExpansionInExtraction: class TestURIExpansionInExtraction:
@ -343,8 +343,8 @@ class TestURIExpansionInExtraction:
validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset) validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset)
assert validated[0].o.value == "http://purl.org/ontology/fo/Recipe" assert validated[0].o.iri == "http://purl.org/ontology/fo/Recipe"
assert validated[0].o.is_uri, "Class reference should be URI" assert validated[0].o.type == IRI, "Class reference should be URI"
def test_expands_property_names(self, extractor, sample_ontology_subset): def test_expands_property_names(self, extractor, sample_ontology_subset):
"""Test that property names are expanded to full URIs.""" """Test that property names are expanded to full URIs."""
@ -354,7 +354,7 @@ class TestURIExpansionInExtraction:
validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset) validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset)
assert validated[0].p.value == "http://purl.org/ontology/fo/produces" assert validated[0].p.iri == "http://purl.org/ontology/fo/produces"
def test_expands_entity_instances(self, extractor, sample_ontology_subset): def test_expands_entity_instances(self, extractor, sample_ontology_subset):
"""Test that entity instances get constructed URIs.""" """Test that entity instances get constructed URIs."""
@ -364,8 +364,8 @@ class TestURIExpansionInExtraction:
validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset) validated = extractor.parse_and_validate_triples(triples_response, sample_ontology_subset)
assert validated[0].s.value.startswith("https://trustgraph.ai/food/") assert validated[0].s.iri.startswith("https://trustgraph.ai/food/")
assert "my-special-recipe" in validated[0].s.value assert "my-special-recipe" in validated[0].s.iri
class TestEdgeCases: class TestEdgeCases:

200
tests/unit/test_query/test_triples_cassandra_query.py
View file

@ -5,8 +5,8 @@ Tests for Cassandra triples query service
import pytest import pytest
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from trustgraph.query.triples.cassandra.service import Processor from trustgraph.query.triples.cassandra.service import Processor, create_term
from trustgraph.schema import Value from trustgraph.schema import Term, IRI, LITERAL
class TestCassandraQueryProcessor: class TestCassandraQueryProcessor:
@ -21,59 +21,59 @@ class TestCassandraQueryProcessor:
graph_host='localhost' graph_host='localhost'
) )
def test_create_value_with_http_uri(self, processor): def test_create_term_with_http_uri(self, processor):
"""Test create_value with HTTP URI""" """Test create_term with HTTP URI"""
result = processor.create_value("http://example.com/resource") result = create_term("http://example.com/resource")
assert isinstance(result, Value)
assert result.value == "http://example.com/resource"
assert result.is_uri is True
def test_create_value_with_https_uri(self, processor): assert isinstance(result, Term)
"""Test create_value with HTTPS URI""" assert result.iri == "http://example.com/resource"
result = processor.create_value("https://example.com/resource") assert result.type == IRI
assert isinstance(result, Value)
assert result.value == "https://example.com/resource"
assert result.is_uri is True
def test_create_value_with_literal(self, processor): def test_create_term_with_https_uri(self, processor):
"""Test create_value with literal value""" """Test create_term with HTTPS URI"""
result = processor.create_value("just a literal string") result = create_term("https://example.com/resource")
assert isinstance(result, Value) assert isinstance(result, Term)
assert result.iri == "https://example.com/resource"
assert result.type == IRI
def test_create_term_with_literal(self, processor):
"""Test create_term with literal value"""
result = create_term("just a literal string")
assert isinstance(result, Term)
assert result.value == "just a literal string" assert result.value == "just a literal string"
assert result.is_uri is False assert result.type == LITERAL
def test_create_value_with_empty_string(self, processor): def test_create_term_with_empty_string(self, processor):
"""Test create_value with empty string""" """Test create_term with empty string"""
result = processor.create_value("") result = create_term("")
assert isinstance(result, Value) assert isinstance(result, Term)
assert result.value == "" assert result.value == ""
assert result.is_uri is False assert result.type == LITERAL
def test_create_value_with_partial_uri(self, processor): def test_create_term_with_partial_uri(self, processor):
"""Test create_value with string that looks like URI but isn't complete""" """Test create_term with string that looks like URI but isn't complete"""
result = processor.create_value("http") result = create_term("http")
assert isinstance(result, Value) assert isinstance(result, Term)
assert result.value == "http" assert result.value == "http"
assert result.is_uri is False assert result.type == LITERAL
def test_create_value_with_ftp_uri(self, processor): def test_create_term_with_ftp_uri(self, processor):
"""Test create_value with FTP URI (should not be detected as URI)""" """Test create_term with FTP URI (should not be detected as URI)"""
result = processor.create_value("ftp://example.com/file") result = create_term("ftp://example.com/file")
assert isinstance(result, Value) assert isinstance(result, Term)
assert result.value == "ftp://example.com/file" assert result.value == "ftp://example.com/file"
assert result.is_uri is False assert result.type == LITERAL
@pytest.mark.asyncio @pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_spo_query(self, mock_trustgraph): async def test_query_triples_spo_query(self, mock_trustgraph):
"""Test querying triples with subject, predicate, and object specified""" """Test querying triples with subject, predicate, and object specified"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
# Setup mock TrustGraph # Setup mock TrustGraph
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
@ -90,9 +90,9 @@ class TestCassandraQueryProcessor:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=Value(value='test_object', is_uri=False), o=Term(type=LITERAL, value='test_object'),
limit=100 limit=100
) )
@ -111,8 +111,8 @@ class TestCassandraQueryProcessor:
# Verify result contains the queried triple # Verify result contains the queried triple
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'test_subject' assert result[0].s.iri == 'test_subject'
assert result[0].p.value == 'test_predicate' assert result[0].p.iri == 'test_predicate'
assert result[0].o.value == 'test_object' assert result[0].o.value == 'test_object'
def test_processor_initialization_with_defaults(self): def test_processor_initialization_with_defaults(self):
@ -146,7 +146,7 @@ class TestCassandraQueryProcessor:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_sp_pattern(self, mock_trustgraph): async def test_query_triples_sp_pattern(self, mock_trustgraph):
"""Test SP query pattern (subject and predicate, no object)""" """Test SP query pattern (subject and predicate, no object)"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
# Setup mock TrustGraph and response # Setup mock TrustGraph and response
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
@ -161,8 +161,8 @@ class TestCassandraQueryProcessor:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=None, o=None,
limit=50 limit=50
) )
@ -171,15 +171,15 @@ class TestCassandraQueryProcessor:
mock_tg_instance.get_sp.assert_called_once_with('test_collection', 'test_subject', 'test_predicate', limit=50) mock_tg_instance.get_sp.assert_called_once_with('test_collection', 'test_subject', 'test_predicate', limit=50)
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'test_subject' assert result[0].s.iri == 'test_subject'
assert result[0].p.value == 'test_predicate' assert result[0].p.iri == 'test_predicate'
assert result[0].o.value == 'result_object' assert result[0].o.value == 'result_object'
@pytest.mark.asyncio @pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_s_pattern(self, mock_trustgraph): async def test_query_triples_s_pattern(self, mock_trustgraph):
"""Test S query pattern (subject only)""" """Test S query pattern (subject only)"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -194,7 +194,7 @@ class TestCassandraQueryProcessor:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=None, p=None,
o=None, o=None,
limit=25 limit=25
@ -204,15 +204,15 @@ class TestCassandraQueryProcessor:
mock_tg_instance.get_s.assert_called_once_with('test_collection', 'test_subject', limit=25) mock_tg_instance.get_s.assert_called_once_with('test_collection', 'test_subject', limit=25)
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'test_subject' assert result[0].s.iri == 'test_subject'
assert result[0].p.value == 'result_predicate' assert result[0].p.iri == 'result_predicate'
assert result[0].o.value == 'result_object' assert result[0].o.value == 'result_object'
@pytest.mark.asyncio @pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_p_pattern(self, mock_trustgraph): async def test_query_triples_p_pattern(self, mock_trustgraph):
"""Test P query pattern (predicate only)""" """Test P query pattern (predicate only)"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -228,7 +228,7 @@ class TestCassandraQueryProcessor:
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=None, s=None,
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=None, o=None,
limit=10 limit=10
) )
@ -237,15 +237,15 @@ class TestCassandraQueryProcessor:
mock_tg_instance.get_p.assert_called_once_with('test_collection', 'test_predicate', limit=10) mock_tg_instance.get_p.assert_called_once_with('test_collection', 'test_predicate', limit=10)
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'result_subject' assert result[0].s.iri == 'result_subject'
assert result[0].p.value == 'test_predicate' assert result[0].p.iri == 'test_predicate'
assert result[0].o.value == 'result_object' assert result[0].o.value == 'result_object'
@pytest.mark.asyncio @pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_o_pattern(self, mock_trustgraph): async def test_query_triples_o_pattern(self, mock_trustgraph):
"""Test O query pattern (object only)""" """Test O query pattern (object only)"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -262,7 +262,7 @@ class TestCassandraQueryProcessor:
collection='test_collection', collection='test_collection',
s=None, s=None,
p=None, p=None,
o=Value(value='test_object', is_uri=False), o=Term(type=LITERAL, value='test_object'),
limit=75 limit=75
) )
@ -270,8 +270,8 @@ class TestCassandraQueryProcessor:
mock_tg_instance.get_o.assert_called_once_with('test_collection', 'test_object', limit=75) mock_tg_instance.get_o.assert_called_once_with('test_collection', 'test_object', limit=75)
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'result_subject' assert result[0].s.iri == 'result_subject'
assert result[0].p.value == 'result_predicate' assert result[0].p.iri == 'result_predicate'
assert result[0].o.value == 'test_object' assert result[0].o.value == 'test_object'
@pytest.mark.asyncio @pytest.mark.asyncio
@ -304,8 +304,8 @@ class TestCassandraQueryProcessor:
mock_tg_instance.get_all.assert_called_once_with('test_collection', limit=1000) mock_tg_instance.get_all.assert_called_once_with('test_collection', limit=1000)
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'all_subject' assert result[0].s.iri == 'all_subject'
assert result[0].p.value == 'all_predicate' assert result[0].p.iri == 'all_predicate'
assert result[0].o.value == 'all_object' assert result[0].o.value == 'all_object'
def test_add_args_method(self): def test_add_args_method(self):
@ -378,7 +378,7 @@ class TestCassandraQueryProcessor:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_with_authentication(self, mock_trustgraph): async def test_query_triples_with_authentication(self, mock_trustgraph):
"""Test querying with username and password authentication""" """Test querying with username and password authentication"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -393,9 +393,9 @@ class TestCassandraQueryProcessor:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=Value(value='test_object', is_uri=False), o=Term(type=LITERAL, value='test_object'),
limit=100 limit=100
) )
@ -413,7 +413,7 @@ class TestCassandraQueryProcessor:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_table_reuse(self, mock_trustgraph): async def test_query_triples_table_reuse(self, mock_trustgraph):
"""Test that TrustGraph is reused for same table""" """Test that TrustGraph is reused for same table"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -424,9 +424,9 @@ class TestCassandraQueryProcessor:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=Value(value='test_object', is_uri=False), o=Term(type=LITERAL, value='test_object'),
limit=100 limit=100
) )
@ -442,7 +442,7 @@ class TestCassandraQueryProcessor:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_table_switching(self, mock_trustgraph): async def test_query_triples_table_switching(self, mock_trustgraph):
"""Test table switching creates new TrustGraph""" """Test table switching creates new TrustGraph"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance1 = MagicMock() mock_tg_instance1 = MagicMock()
mock_tg_instance2 = MagicMock() mock_tg_instance2 = MagicMock()
@ -454,7 +454,7 @@ class TestCassandraQueryProcessor:
query1 = TriplesQueryRequest( query1 = TriplesQueryRequest(
user='user1', user='user1',
collection='collection1', collection='collection1',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=None, p=None,
o=None, o=None,
limit=100 limit=100
@ -467,7 +467,7 @@ class TestCassandraQueryProcessor:
query2 = TriplesQueryRequest( query2 = TriplesQueryRequest(
user='user2', user='user2',
collection='collection2', collection='collection2',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=None, p=None,
o=None, o=None,
limit=100 limit=100
@ -483,7 +483,7 @@ class TestCassandraQueryProcessor:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_exception_handling(self, mock_trustgraph): async def test_query_triples_exception_handling(self, mock_trustgraph):
"""Test exception handling during query execution""" """Test exception handling during query execution"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -494,9 +494,9 @@ class TestCassandraQueryProcessor:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=Value(value='test_object', is_uri=False), o=Term(type=LITERAL, value='test_object'),
limit=100 limit=100
) )
@ -507,7 +507,7 @@ class TestCassandraQueryProcessor:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_query_triples_multiple_results(self, mock_trustgraph): async def test_query_triples_multiple_results(self, mock_trustgraph):
"""Test query returning multiple results""" """Test query returning multiple results"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -524,8 +524,8 @@ class TestCassandraQueryProcessor:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=None, o=None,
limit=100 limit=100
) )
@ -544,7 +544,7 @@ class TestCassandraQueryPerformanceOptimizations:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_get_po_query_optimization(self, mock_trustgraph): async def test_get_po_query_optimization(self, mock_trustgraph):
"""Test that get_po queries use optimized table (no ALLOW FILTERING)""" """Test that get_po queries use optimized table (no ALLOW FILTERING)"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -560,8 +560,8 @@ class TestCassandraQueryPerformanceOptimizations:
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=None, s=None,
p=Value(value='test_predicate', is_uri=False), p=Term(type=LITERAL, value='test_predicate'),
o=Value(value='test_object', is_uri=False), o=Term(type=LITERAL, value='test_object'),
limit=50 limit=50
) )
@ -573,15 +573,15 @@ class TestCassandraQueryPerformanceOptimizations:
) )
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'result_subject' assert result[0].s.iri == 'result_subject'
assert result[0].p.value == 'test_predicate' assert result[0].p.iri == 'test_predicate'
assert result[0].o.value == 'test_object' assert result[0].o.value == 'test_object'
@pytest.mark.asyncio @pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_get_os_query_optimization(self, mock_trustgraph): async def test_get_os_query_optimization(self, mock_trustgraph):
"""Test that get_os queries use optimized table (no ALLOW FILTERING)""" """Test that get_os queries use optimized table (no ALLOW FILTERING)"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -596,9 +596,9 @@ class TestCassandraQueryPerformanceOptimizations:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value='test_subject', is_uri=False), s=Term(type=LITERAL, value='test_subject'),
p=None, p=None,
o=Value(value='test_object', is_uri=False), o=Term(type=LITERAL, value='test_object'),
limit=25 limit=25
) )
@ -610,15 +610,15 @@ class TestCassandraQueryPerformanceOptimizations:
) )
assert len(result) == 1 assert len(result) == 1
assert result[0].s.value == 'test_subject' assert result[0].s.iri == 'test_subject'
assert result[0].p.value == 'result_predicate' assert result[0].p.iri == 'result_predicate'
assert result[0].o.value == 'test_object' assert result[0].o.value == 'test_object'
@pytest.mark.asyncio @pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_all_query_patterns_use_correct_tables(self, mock_trustgraph): async def test_all_query_patterns_use_correct_tables(self, mock_trustgraph):
"""Test that all query patterns route to their optimal tables""" """Test that all query patterns route to their optimal tables"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -655,9 +655,9 @@ class TestCassandraQueryPerformanceOptimizations:
query = TriplesQueryRequest( query = TriplesQueryRequest(
user='test_user', user='test_user',
collection='test_collection', collection='test_collection',
s=Value(value=s, is_uri=False) if s else None, s=Term(type=LITERAL, value=s) if s else None,
p=Value(value=p, is_uri=False) if p else None, p=Term(type=LITERAL, value=p) if p else None,
o=Value(value=o, is_uri=False) if o else None, o=Term(type=LITERAL, value=o) if o else None,
limit=10 limit=10
) )
@ -690,7 +690,7 @@ class TestCassandraQueryPerformanceOptimizations:
@patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph') @patch('trustgraph.query.triples.cassandra.service.KnowledgeGraph')
async def test_performance_critical_po_query_no_filtering(self, mock_trustgraph): async def test_performance_critical_po_query_no_filtering(self, mock_trustgraph):
"""Test the performance-critical PO query that eliminates ALLOW FILTERING""" """Test the performance-critical PO query that eliminates ALLOW FILTERING"""
from trustgraph.schema import TriplesQueryRequest, Value from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock() mock_tg_instance = MagicMock()
mock_trustgraph.return_value = mock_tg_instance mock_trustgraph.return_value = mock_tg_instance
@ -711,8 +711,8 @@ class TestCassandraQueryPerformanceOptimizations:
user='large_dataset_user', user='large_dataset_user',
collection='massive_collection', collection='massive_collection',
s=None, s=None,
p=Value(value='http://www.w3.org/1999/02/22-rdf-syntax-ns#type', is_uri=True), p=Term(type=IRI, iri='http://www.w3.org/1999/02/22-rdf-syntax-ns#type'),
o=Value(value='http://example.com/Person', is_uri=True), o=Term(type=IRI, iri='http://example.com/Person'),
limit=1000 limit=1000
) )