diff --git a/tests/unit/test_query/test_triples_cassandra_query.py b/tests/unit/test_query/test_triples_cassandra_query.py index eb81c8c8..0b27458f 100644 --- a/tests/unit/test_query/test_triples_cassandra_query.py +++ b/tests/unit/test_query/test_triples_cassandra_query.py @@ -3,7 +3,7 @@ Tests for Cassandra triples query service """ import pytest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from trustgraph.query.triples.cassandra.service import Processor from trustgraph.schema import Value @@ -67,4 +67,51 @@ class TestCassandraQueryProcessor: assert isinstance(result, Value) assert result.value == "ftp://example.com/file" - assert result.is_uri is False \ No newline at end of file + assert result.is_uri is False + + @pytest.mark.asyncio + @patch('trustgraph.query.triples.cassandra.service.TrustGraph') + async def test_query_triples_spo_query(self, mock_trustgraph): + """Test querying triples with subject, predicate, and object specified""" + from trustgraph.schema import TriplesQueryRequest, Value + + # Setup mock TrustGraph + mock_tg_instance = MagicMock() + mock_trustgraph.return_value = mock_tg_instance + mock_tg_instance.get_spo.return_value = None # SPO query returns None if found + + processor = Processor( + taskgroup=MagicMock(), + id='test-cassandra-query', + graph_host='localhost' + ) + + # Create query request with all SPO values + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value='test_subject', is_uri=False), + p=Value(value='test_predicate', is_uri=False), + o=Value(value='test_object', is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify TrustGraph was created with correct parameters + mock_trustgraph.assert_called_once_with( + hosts=['localhost'], + keyspace='test_user', + table='test_collection' + ) + + # Verify get_spo was called with correct parameters + mock_tg_instance.get_spo.assert_called_once_with( + 'test_subject', 'test_predicate', 'test_object', limit=100 + ) + + # Verify result contains the queried triple + assert len(result) == 1 + assert result[0].s.value == 'test_subject' + assert result[0].p.value == 'test_predicate' + assert result[0].o.value == 'test_object' \ No newline at end of file