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

Remove switch to enable old graph

Browse source
This commit is contained in:
Cyber MacGeddon 2026-02-16 12:56:27 +00:00
parent 2d357889a2
commit 4b3cb5a852
6 changed files with 92 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

44
tests/unit/test_direct/test_entity_centric_kg.py
View file

@ -463,50 +463,6 @@ class TestQuadResult:
assert result.lang == 'en'
class TestFactoryFunction:
"""Test cases for get_knowledge_graph_class factory function"""
def test_factory_returns_legacy_by_default(self):
"""Test factory returns KnowledgeGraph when env var not set"""
from trustgraph.direct.cassandra_kg import get_knowledge_graph_class, KnowledgeGraph
with patch.dict(os.environ, {}, clear=True):
KGClass = get_knowledge_graph_class()
assert KGClass == KnowledgeGraph
def test_factory_returns_legacy_when_false(self):
"""Test factory returns KnowledgeGraph when env var is false"""
from trustgraph.direct.cassandra_kg import get_knowledge_graph_class, KnowledgeGraph
with patch.dict(os.environ, {'CASSANDRA_ENTITY_CENTRIC': 'false'}):
KGClass = get_knowledge_graph_class()
assert KGClass == KnowledgeGraph
def test_factory_returns_entity_centric_when_true(self):
"""Test factory returns EntityCentricKnowledgeGraph when env var is true"""
from trustgraph.direct.cassandra_kg import (
get_knowledge_graph_class, EntityCentricKnowledgeGraph
)
with patch.dict(os.environ, {'CASSANDRA_ENTITY_CENTRIC': 'true'}):
KGClass = get_knowledge_graph_class()
assert KGClass == EntityCentricKnowledgeGraph
def test_factory_case_insensitive(self):
"""Test factory handles case insensitive values"""
from trustgraph.direct.cassandra_kg import (
get_knowledge_graph_class, EntityCentricKnowledgeGraph
)
with patch.dict(os.environ, {'CASSANDRA_ENTITY_CENTRIC': 'TRUE'}):
KGClass = get_knowledge_graph_class()
assert KGClass == EntityCentricKnowledgeGraph
with patch.dict(os.environ, {'CASSANDRA_ENTITY_CENTRIC': 'True'}):
KGClass = get_knowledge_graph_class()
assert KGClass == EntityCentricKnowledgeGraph
class TestWriteHelperFunctions:
"""Test cases for helper functions in write.py"""

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

@ -70,15 +70,14 @@ class TestCassandraQueryProcessor:
assert result.type == LITERAL
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_spo_query(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_spo_query(self, mock_kg_class):
"""Test querying triples with subject, predicate, and object specified"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
# Setup mock TrustGraph via factory function
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
# SPO query returns a list of results (with mock graph attribute)
mock_result = MagicMock()
mock_result.g = ''
@ -151,15 +150,14 @@ class TestCassandraQueryProcessor:
assert processor.table is None
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_sp_pattern(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_sp_pattern(self, mock_kg_class):
"""Test SP query pattern (subject and predicate, no object)"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
# Setup mock TrustGraph via factory function
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_result = MagicMock()
mock_result.o = 'result_object'
@ -189,14 +187,13 @@ class TestCassandraQueryProcessor:
assert result[0].o.value == 'result_object'
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_s_pattern(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_s_pattern(self, mock_kg_class):
"""Test S query pattern (subject only)"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_result = MagicMock()
mock_result.p = 'result_predicate'
@ -227,14 +224,13 @@ class TestCassandraQueryProcessor:
assert result[0].o.value == 'result_object'
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_p_pattern(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_p_pattern(self, mock_kg_class):
"""Test P query pattern (predicate only)"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_result = MagicMock()
mock_result.s = 'result_subject'
@ -265,14 +261,13 @@ class TestCassandraQueryProcessor:
assert result[0].o.value == 'result_object'
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_o_pattern(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_o_pattern(self, mock_kg_class):
"""Test O query pattern (object only)"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_result = MagicMock()
mock_result.s = 'result_subject'
@ -303,14 +298,13 @@ class TestCassandraQueryProcessor:
assert result[0].o.value == 'test_object'
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_get_all_pattern(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_get_all_pattern(self, mock_kg_class):
"""Test query pattern with no constraints (get all)"""
from trustgraph.schema import TriplesQueryRequest
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_result = MagicMock()
mock_result.s = 'all_subject'
@ -408,14 +402,13 @@ class TestCassandraQueryProcessor:
mock_launch.assert_called_once_with(default_ident, '\nTriples query service. Input is a (s, p, o, g) quad pattern, some values may be\nnull. Output is a list of quads.\n')
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_with_authentication(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_with_authentication(self, mock_kg_class):
"""Test querying with username and password authentication"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
# SPO query returns a list of results
mock_result = MagicMock()
mock_result.g = ''
@ -451,14 +444,13 @@ class TestCassandraQueryProcessor:
)
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_table_reuse(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_table_reuse(self, mock_kg_class):
"""Test that TrustGraph is reused for same table"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
# SPO query returns a list of results
mock_result = MagicMock()
mock_result.g = ''
@ -488,15 +480,14 @@ class TestCassandraQueryProcessor:
assert mock_kg_class.call_count == 1 # Should not increase
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_table_switching(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_table_switching(self, mock_kg_class):
"""Test table switching creates new TrustGraph"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance1 = MagicMock()
mock_tg_instance2 = MagicMock()
mock_kg_class = MagicMock(side_effect=[mock_tg_instance1, mock_tg_instance2])
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.side_effect = [mock_tg_instance1, mock_tg_instance2]
# Setup mock results for both instances
mock_result = MagicMock()
@ -541,14 +532,13 @@ class TestCassandraQueryProcessor:
assert mock_kg_class.call_count == 2
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_exception_handling(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_exception_handling(self, mock_kg_class):
"""Test exception handling during query execution"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_tg_instance.get_spo.side_effect = Exception("Query failed")
processor = Processor(taskgroup=MagicMock())
@ -566,14 +556,13 @@ class TestCassandraQueryProcessor:
await processor.query_triples(query)
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_query_triples_multiple_results(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_query_triples_multiple_results(self, mock_kg_class):
"""Test query returning multiple results"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
# Mock multiple results
mock_result1 = MagicMock()
@ -612,14 +601,13 @@ class TestCassandraQueryPerformanceOptimizations:
"""Test cases for multi-table performance optimizations in query service"""
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_get_po_query_optimization(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_get_po_query_optimization(self, mock_kg_class):
"""Test that get_po queries use optimized table (no ALLOW FILTERING)"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_result = MagicMock()
mock_result.s = 'result_subject'
@ -654,14 +642,13 @@ class TestCassandraQueryPerformanceOptimizations:
assert result[0].o.value == 'test_object'
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_get_os_query_optimization(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_get_os_query_optimization(self, mock_kg_class):
"""Test that get_os queries use optimized table (no ALLOW FILTERING)"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
mock_result = MagicMock()
mock_result.p = 'result_predicate'
@ -696,14 +683,13 @@ class TestCassandraQueryPerformanceOptimizations:
assert result[0].o.value == 'test_object'
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_all_query_patterns_use_correct_tables(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_all_query_patterns_use_correct_tables(self, mock_kg_class):
"""Test that all query patterns route to their optimal tables"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
# Mock empty results for all queries
mock_tg_instance.get_all.return_value = []
@ -769,14 +755,13 @@ class TestCassandraQueryPerformanceOptimizations:
# Mode is determined in KnowledgeGraph initialization
@pytest.mark.asyncio
@patch('trustgraph.query.triples.cassandra.service.get_knowledge_graph_class')
async def test_performance_critical_po_query_no_filtering(self, mock_get_kg_class):
@patch('trustgraph.query.triples.cassandra.service.EntityCentricKnowledgeGraph')
async def test_performance_critical_po_query_no_filtering(self, mock_kg_class):
"""Test the performance-critical PO query that eliminates ALLOW FILTERING"""
from trustgraph.schema import TriplesQueryRequest, Term, IRI, LITERAL
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
# Mock multiple subjects for the same predicate-object pair
mock_results = []

84
tests/unit/test_storage/test_triples_cassandra_storage.py
View file

@ -87,13 +87,12 @@ class TestCassandraStorageProcessor:
assert processor.cassandra_username == 'new-user' # Only cassandra_* params work
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_table_switching_with_auth(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_table_switching_with_auth(self, mock_kg_class):
"""Test table switching logic when authentication is provided"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
processor = Processor(
taskgroup=taskgroup_mock,
@ -119,13 +118,12 @@ class TestCassandraStorageProcessor:
assert processor.table == 'user1'
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_table_switching_without_auth(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_table_switching_without_auth(self, mock_kg_class):
"""Test table switching logic when no authentication is provided"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
processor = Processor(taskgroup=taskgroup_mock)
@ -145,13 +143,12 @@ class TestCassandraStorageProcessor:
assert processor.table == 'user2'
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_table_reuse_when_same(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_table_reuse_when_same(self, mock_kg_class):
"""Test that TrustGraph is not recreated when table hasn't changed"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
processor = Processor(taskgroup=taskgroup_mock)
@ -170,13 +167,12 @@ class TestCassandraStorageProcessor:
assert mock_kg_class.call_count == 1 # Should not increase
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_triple_insertion(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_triple_insertion(self, mock_kg_class):
"""Test that triples are properly inserted into Cassandra"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
processor = Processor(taskgroup=taskgroup_mock)
@ -227,13 +223,12 @@ class TestCassandraStorageProcessor:
)
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_triple_insertion_with_empty_list(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_triple_insertion_with_empty_list(self, mock_kg_class):
"""Test behavior when message has no triples"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
processor = Processor(taskgroup=taskgroup_mock)
@ -249,13 +244,12 @@ class TestCassandraStorageProcessor:
mock_tg_instance.insert.assert_not_called()
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
@patch('trustgraph.storage.triples.cassandra.write.time.sleep')
async def test_exception_handling_with_retry(self, mock_sleep, mock_get_kg_class):
async def test_exception_handling_with_retry(self, mock_sleep, mock_kg_class):
"""Test exception handling during TrustGraph creation"""
taskgroup_mock = MagicMock()
mock_kg_class = MagicMock(side_effect=Exception("Connection failed"))
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.side_effect = Exception("Connection failed")
processor = Processor(taskgroup=taskgroup_mock)
@ -355,14 +349,13 @@ class TestCassandraStorageProcessor:
mock_launch.assert_called_once_with(default_ident, '\nGraph writer. Input is graph edge. Writes edges to Cassandra graph.\n')
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_store_triples_table_switching_between_different_tables(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_store_triples_table_switching_between_different_tables(self, mock_kg_class):
"""Test table switching when different tables are used in sequence"""
taskgroup_mock = MagicMock()
mock_tg_instance1 = MagicMock()
mock_tg_instance2 = MagicMock()
mock_kg_class = MagicMock(side_effect=[mock_tg_instance1, mock_tg_instance2])
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.side_effect = [mock_tg_instance1, mock_tg_instance2]
processor = Processor(taskgroup=taskgroup_mock)
@ -390,13 +383,12 @@ class TestCassandraStorageProcessor:
assert mock_kg_class.call_count == 2
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_store_triples_with_special_characters_in_values(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_store_triples_with_special_characters_in_values(self, mock_kg_class):
"""Test storing triples with special characters and unicode"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
processor = Processor(taskgroup=taskgroup_mock)
@ -434,8 +426,8 @@ class TestCassandraStorageProcessor:
)
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_store_triples_preserves_old_table_on_exception(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_store_triples_preserves_old_table_on_exception(self, mock_kg_class):
"""Test that table remains unchanged when TrustGraph creation fails"""
taskgroup_mock = MagicMock()
@ -445,8 +437,7 @@ class TestCassandraStorageProcessor:
processor.table = ('old_user', 'old_collection')
# Mock TrustGraph to raise exception
mock_kg_class = MagicMock(side_effect=Exception("Connection failed"))
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.side_effect = Exception("Connection failed")
mock_message = MagicMock()
mock_message.metadata.user = 'new_user'
@ -466,13 +457,12 @@ class TestCassandraPerformanceOptimizations:
"""Test cases for multi-table performance optimizations"""
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_legacy_mode_uses_single_table(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_legacy_mode_uses_single_table(self, mock_kg_class):
"""Test that legacy mode still works with single table"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
with patch.dict('os.environ', {'CASSANDRA_USE_LEGACY': 'true'}):
processor = Processor(taskgroup=taskgroup_mock)
@ -488,13 +478,12 @@ class TestCassandraPerformanceOptimizations:
assert mock_tg_instance is not None
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_optimized_mode_uses_multi_table(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_optimized_mode_uses_multi_table(self, mock_kg_class):
"""Test that optimized mode uses multi-table schema"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
with patch.dict('os.environ', {'CASSANDRA_USE_LEGACY': 'false'}):
processor = Processor(taskgroup=taskgroup_mock)
@ -510,13 +499,12 @@ class TestCassandraPerformanceOptimizations:
assert mock_tg_instance is not None
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.get_knowledge_graph_class')
async def test_batch_write_consistency(self, mock_get_kg_class):
@patch('trustgraph.storage.triples.cassandra.write.EntityCentricKnowledgeGraph')
async def test_batch_write_consistency(self, mock_kg_class):
"""Test that all tables stay consistent during batch writes"""
taskgroup_mock = MagicMock()
mock_tg_instance = MagicMock()
mock_kg_class = MagicMock(return_value=mock_tg_instance)
mock_get_kg_class.return_value = mock_kg_class
mock_kg_class.return_value = mock_tg_instance
processor = Processor(taskgroup=taskgroup_mock)

20
trustgraph-flow/trustgraph/direct/cassandra_kg.py
View file

@ -20,6 +20,11 @@ DEFAULT_GRAPH = ""
class KnowledgeGraph:
"""
REDUNDANT: This 7-table implementation has been superseded by
EntityCentricKnowledgeGraph which uses a more efficient 2-table model.
This class is retained temporarily for reference but should not be used
for new deployments.
Cassandra-backed knowledge graph supporting quads (s, p, o, g).
Uses 7 tables to support all 16 query patterns efficiently:
@ -1088,18 +1093,3 @@ class QuadResult:
self.lang = lang
def get_knowledge_graph_class():
"""
Factory function to select KnowledgeGraph implementation.
Uses CASSANDRA_ENTITY_CENTRIC environment variable to select:
- "true": EntityCentricKnowledgeGraph (new 2-table model)
- Otherwise: KnowledgeGraph (original 7-table model)
"""
use_entity_centric = os.environ.get("CASSANDRA_ENTITY_CENTRIC", "").lower() == "true"
if use_entity_centric:
logger.info("Using EntityCentricKnowledgeGraph (2-table model)")
return EntityCentricKnowledgeGraph
else:
logger.info("Using KnowledgeGraph (7-table model)")
return KnowledgeGraph

4
trustgraph-flow/trustgraph/query/triples/cassandra/service.py
View file

@ -7,7 +7,7 @@ null. Output is a list of quads.
import logging
from .... direct.cassandra_kg import (
KnowledgeGraph, GRAPH_WILDCARD, DEFAULT_GRAPH, get_knowledge_graph_class
EntityCentricKnowledgeGraph, GRAPH_WILDCARD, DEFAULT_GRAPH
)
from .... schema import TriplesQueryRequest, TriplesQueryResponse, Error
from .... schema import Term, Triple, IRI, LITERAL
@ -106,7 +106,7 @@ class Processor(TriplesQueryService):
if user != self.table:
# Use factory function to select implementation
KGClass = get_knowledge_graph_class()
KGClass = EntityCentricKnowledgeGraph
if self.cassandra_username and self.cassandra_password:
self.tg = KGClass(

8
trustgraph-flow/trustgraph/storage/triples/cassandra/write.py
View file

@ -11,7 +11,7 @@ import time
import logging
from .... direct.cassandra_kg import (
KnowledgeGraph, DEFAULT_GRAPH, get_knowledge_graph_class
EntityCentricKnowledgeGraph, DEFAULT_GRAPH
)
from .... base import TriplesStoreService, CollectionConfigHandler
from .... base import AsyncProcessor, Consumer, Producer
@ -121,7 +121,7 @@ class Processor(CollectionConfigHandler, TriplesStoreService):
self.tg = None
# Use factory function to select implementation
KGClass = get_knowledge_graph_class()
KGClass = EntityCentricKnowledgeGraph
try:
if self.cassandra_username and self.cassandra_password:
@ -174,7 +174,7 @@ class Processor(CollectionConfigHandler, TriplesStoreService):
self.tg = None
# Use factory function to select implementation
KGClass = get_knowledge_graph_class()
KGClass = EntityCentricKnowledgeGraph
try:
if self.cassandra_username and self.cassandra_password:
@ -216,7 +216,7 @@ class Processor(CollectionConfigHandler, TriplesStoreService):
self.tg = None
# Use factory function to select implementation
KGClass = get_knowledge_graph_class()
KGClass = EntityCentricKnowledgeGraph
try:
if self.cassandra_username and self.cassandra_password: