From 0f36d1d16e092ba2cdd81a7037b49628e9bdca0b Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 14 Jul 2025 21:14:59 +0100 Subject: [PATCH] Fixing storage and adding tests --- .../test_query/test_triples_falkordb_query.py | 556 +++++++++++++++++ .../test_query/test_triples_memgraph_query.py | 568 ++++++++++++++++++ .../test_query/test_triples_neo4j_query.py | 338 +++++++++++ .../test_triples_neo4j_storage.py | 548 +++++++++++++++++ .../query/triples/falkordb/service.py | 204 +++---- .../query/triples/memgraph/service.py | 165 ++--- .../trustgraph/query/triples/neo4j/service.py | 131 ++-- .../trustgraph/storage/triples/neo4j/write.py | 34 +- 8 files changed, 2221 insertions(+), 323 deletions(-) create mode 100644 tests/unit/test_query/test_triples_falkordb_query.py create mode 100644 tests/unit/test_query/test_triples_memgraph_query.py create mode 100644 tests/unit/test_query/test_triples_neo4j_query.py create mode 100644 tests/unit/test_storage/test_triples_neo4j_storage.py diff --git a/tests/unit/test_query/test_triples_falkordb_query.py b/tests/unit/test_query/test_triples_falkordb_query.py new file mode 100644 index 00000000..3e7d07db --- /dev/null +++ b/tests/unit/test_query/test_triples_falkordb_query.py @@ -0,0 +1,556 @@ +""" +Tests for FalkorDB triples query service +""" + +import pytest +from unittest.mock import MagicMock, patch + +from trustgraph.query.triples.falkordb.service import Processor +from trustgraph.schema import Value, TriplesQueryRequest + + +class TestFalkorDBQueryProcessor: + """Test cases for FalkorDB query processor""" + + @pytest.fixture + def processor(self): + """Create a processor instance for testing""" + with patch('trustgraph.query.triples.falkordb.service.FalkorDB'): + return Processor( + taskgroup=MagicMock(), + id='test-falkordb-query', + graph_url='falkor://localhost:6379' + ) + + def test_create_value_with_http_uri(self, processor): + """Test create_value with HTTP URI""" + result = processor.create_value("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): + """Test create_value with HTTPS URI""" + result = processor.create_value("https://example.com/resource") + + 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): + """Test create_value with literal value""" + result = processor.create_value("just a literal string") + + assert isinstance(result, Value) + assert result.value == "just a literal string" + assert result.is_uri is False + + def test_create_value_with_empty_string(self, processor): + """Test create_value with empty string""" + result = processor.create_value("") + + assert isinstance(result, Value) + assert result.value == "" + assert result.is_uri is False + + def test_create_value_with_partial_uri(self, processor): + """Test create_value with string that looks like URI but isn't complete""" + result = processor.create_value("http") + + assert isinstance(result, Value) + assert result.value == "http" + assert result.is_uri is False + + def test_create_value_with_ftp_uri(self, processor): + """Test create_value with FTP URI (should not be detected as URI)""" + result = processor.create_value("ftp://example.com/file") + + assert isinstance(result, Value) + assert result.value == "ftp://example.com/file" + assert result.is_uri is False + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + def test_processor_initialization_with_defaults(self, mock_falkordb): + """Test processor initialization with default parameters""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + processor = Processor(taskgroup=taskgroup_mock) + + assert processor.db == 'falkordb' + mock_falkordb.from_url.assert_called_once_with('falkor://falkordb:6379') + mock_client.select_graph.assert_called_once_with('falkordb') + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + def test_processor_initialization_with_custom_params(self, mock_falkordb): + """Test processor initialization with custom parameters""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + processor = Processor( + taskgroup=taskgroup_mock, + graph_url='falkor://custom:6379', + database='customdb' + ) + + assert processor.db == 'customdb' + mock_falkordb.from_url.assert_called_once_with('falkor://custom:6379') + mock_client.select_graph.assert_called_once_with('customdb') + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_spo_query(self, mock_falkordb): + """Test SPO query (all values specified)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results - both queries return one record each + mock_result = MagicMock() + mock_result.result_set = [["record1"]] + mock_graph.query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=Value(value="http://example.com/predicate", is_uri=True), + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify result contains the queried triple (appears twice - once from each query) + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal object" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_sp_query(self, mock_falkordb): + """Test SP query (subject and predicate specified)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results with different objects + mock_result1 = MagicMock() + mock_result1.result_set = [["literal result"]] + mock_result2 = MagicMock() + mock_result2.result_set = [["http://example.com/uri_result"]] + + mock_graph.query.side_effect = [mock_result1, mock_result2] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=Value(value="http://example.com/predicate", is_uri=True), + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify results contain different objects + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal result" + + assert result[1].s.value == "http://example.com/subject" + assert result[1].p.value == "http://example.com/predicate" + assert result[1].o.value == "http://example.com/uri_result" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_so_query(self, mock_falkordb): + """Test SO query (subject and object specified)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results with different predicates + mock_result1 = MagicMock() + mock_result1.result_set = [["http://example.com/pred1"]] + mock_result2 = MagicMock() + mock_result2.result_set = [["http://example.com/pred2"]] + + mock_graph.query.side_effect = [mock_result1, mock_result2] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=None, + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify results contain different predicates + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/pred1" + assert result[0].o.value == "literal object" + + assert result[1].s.value == "http://example.com/subject" + assert result[1].p.value == "http://example.com/pred2" + assert result[1].o.value == "literal object" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_s_query(self, mock_falkordb): + """Test S query (subject only)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results with different predicate-object pairs + mock_result1 = MagicMock() + mock_result1.result_set = [["http://example.com/pred1", "literal1"]] + mock_result2 = MagicMock() + mock_result2.result_set = [["http://example.com/pred2", "http://example.com/uri2"]] + + mock_graph.query.side_effect = [mock_result1, mock_result2] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=None, + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify results contain different predicate-object pairs + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/pred1" + assert result[0].o.value == "literal1" + + assert result[1].s.value == "http://example.com/subject" + assert result[1].p.value == "http://example.com/pred2" + assert result[1].o.value == "http://example.com/uri2" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_po_query(self, mock_falkordb): + """Test PO query (predicate and object specified)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results with different subjects + mock_result1 = MagicMock() + mock_result1.result_set = [["http://example.com/subj1"]] + mock_result2 = MagicMock() + mock_result2.result_set = [["http://example.com/subj2"]] + + mock_graph.query.side_effect = [mock_result1, mock_result2] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=Value(value="http://example.com/predicate", is_uri=True), + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify results contain different subjects + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subj1" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal object" + + assert result[1].s.value == "http://example.com/subj2" + assert result[1].p.value == "http://example.com/predicate" + assert result[1].o.value == "literal object" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_p_query(self, mock_falkordb): + """Test P query (predicate only)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results with different subject-object pairs + mock_result1 = MagicMock() + mock_result1.result_set = [["http://example.com/subj1", "literal1"]] + mock_result2 = MagicMock() + mock_result2.result_set = [["http://example.com/subj2", "http://example.com/uri2"]] + + mock_graph.query.side_effect = [mock_result1, mock_result2] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=Value(value="http://example.com/predicate", is_uri=True), + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify results contain different subject-object pairs + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subj1" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal1" + + assert result[1].s.value == "http://example.com/subj2" + assert result[1].p.value == "http://example.com/predicate" + assert result[1].o.value == "http://example.com/uri2" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_o_query(self, mock_falkordb): + """Test O query (object only)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results with different subject-predicate pairs + mock_result1 = MagicMock() + mock_result1.result_set = [["http://example.com/subj1", "http://example.com/pred1"]] + mock_result2 = MagicMock() + mock_result2.result_set = [["http://example.com/subj2", "http://example.com/pred2"]] + + mock_graph.query.side_effect = [mock_result1, mock_result2] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=None, + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify results contain different subject-predicate pairs + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subj1" + assert result[0].p.value == "http://example.com/pred1" + assert result[0].o.value == "literal object" + + assert result[1].s.value == "http://example.com/subj2" + assert result[1].p.value == "http://example.com/pred2" + assert result[1].o.value == "literal object" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_wildcard_query(self, mock_falkordb): + """Test wildcard query (no constraints)""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query results + mock_result1 = MagicMock() + mock_result1.result_set = [["http://example.com/s1", "http://example.com/p1", "literal1"]] + mock_result2 = MagicMock() + mock_result2.result_set = [["http://example.com/s2", "http://example.com/p2", "http://example.com/o2"]] + + mock_graph.query.side_effect = [mock_result1, mock_result2] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=None, + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_graph.query.call_count == 2 + + # Verify results contain different triples + assert len(result) == 2 + assert result[0].s.value == "http://example.com/s1" + assert result[0].p.value == "http://example.com/p1" + assert result[0].o.value == "literal1" + + assert result[1].s.value == "http://example.com/s2" + assert result[1].p.value == "http://example.com/p2" + assert result[1].o.value == "http://example.com/o2" + + @patch('trustgraph.query.triples.falkordb.service.FalkorDB') + @pytest.mark.asyncio + async def test_query_triples_exception_handling(self, mock_falkordb): + """Test exception handling during query processing""" + taskgroup_mock = MagicMock() + mock_client = MagicMock() + mock_graph = MagicMock() + mock_falkordb.from_url.return_value = mock_client + mock_client.select_graph.return_value = mock_graph + + # Mock query to raise exception + mock_graph.query.side_effect = Exception("Database connection failed") + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=None, + o=None, + limit=100 + ) + + # Should raise the exception + with pytest.raises(Exception, match="Database connection failed"): + await processor.query_triples(query) + + def test_add_args_method(self): + """Test that add_args properly configures argument parser""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + # Mock the parent class add_args method + with patch('trustgraph.query.triples.falkordb.service.TriplesQueryService.add_args') as mock_parent_add_args: + Processor.add_args(parser) + + # Verify parent add_args was called + mock_parent_add_args.assert_called_once() + + # Verify our specific arguments were added + # Parse empty args to check defaults + args = parser.parse_args([]) + + assert hasattr(args, 'graph_url') + assert args.graph_url == 'falkor://falkordb:6379' + assert hasattr(args, 'database') + assert args.database == 'falkordb' + + def test_add_args_with_custom_values(self): + """Test add_args with custom command line values""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.query.triples.falkordb.service.TriplesQueryService.add_args'): + Processor.add_args(parser) + + # Test parsing with custom values + args = parser.parse_args([ + '--graph-url', 'falkor://custom:6379', + '--database', 'querydb' + ]) + + assert args.graph_url == 'falkor://custom:6379' + assert args.database == 'querydb' + + def test_add_args_short_form(self): + """Test add_args with short form arguments""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.query.triples.falkordb.service.TriplesQueryService.add_args'): + Processor.add_args(parser) + + # Test parsing with short form + args = parser.parse_args(['-g', 'falkor://short:6379']) + + assert args.graph_url == 'falkor://short:6379' + + @patch('trustgraph.query.triples.falkordb.service.Processor.launch') + def test_run_function(self, mock_launch): + """Test the run function calls Processor.launch with correct parameters""" + from trustgraph.query.triples.falkordb.service import run, default_ident + + run() + + mock_launch.assert_called_once_with( + default_ident, + "\nTriples query service for FalkorDB.\nInput is a (s, p, o) triple, some values may be null. Output is a list of\ntriples.\n" + ) \ No newline at end of file diff --git a/tests/unit/test_query/test_triples_memgraph_query.py b/tests/unit/test_query/test_triples_memgraph_query.py new file mode 100644 index 00000000..bd394ae4 --- /dev/null +++ b/tests/unit/test_query/test_triples_memgraph_query.py @@ -0,0 +1,568 @@ +""" +Tests for Memgraph triples query service +""" + +import pytest +from unittest.mock import MagicMock, patch + +from trustgraph.query.triples.memgraph.service import Processor +from trustgraph.schema import Value, TriplesQueryRequest + + +class TestMemgraphQueryProcessor: + """Test cases for Memgraph query processor""" + + @pytest.fixture + def processor(self): + """Create a processor instance for testing""" + with patch('trustgraph.query.triples.memgraph.service.GraphDatabase'): + return Processor( + taskgroup=MagicMock(), + id='test-memgraph-query', + graph_host='bolt://localhost:7687' + ) + + def test_create_value_with_http_uri(self, processor): + """Test create_value with HTTP URI""" + result = processor.create_value("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): + """Test create_value with HTTPS URI""" + result = processor.create_value("https://example.com/resource") + + 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): + """Test create_value with literal value""" + result = processor.create_value("just a literal string") + + assert isinstance(result, Value) + assert result.value == "just a literal string" + assert result.is_uri is False + + def test_create_value_with_empty_string(self, processor): + """Test create_value with empty string""" + result = processor.create_value("") + + assert isinstance(result, Value) + assert result.value == "" + assert result.is_uri is False + + def test_create_value_with_partial_uri(self, processor): + """Test create_value with string that looks like URI but isn't complete""" + result = processor.create_value("http") + + assert isinstance(result, Value) + assert result.value == "http" + assert result.is_uri is False + + def test_create_value_with_ftp_uri(self, processor): + """Test create_value with FTP URI (should not be detected as URI)""" + result = processor.create_value("ftp://example.com/file") + + assert isinstance(result, Value) + assert result.value == "ftp://example.com/file" + assert result.is_uri is False + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + def test_processor_initialization_with_defaults(self, mock_graph_db): + """Test processor initialization with default parameters""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + processor = Processor(taskgroup=taskgroup_mock) + + assert processor.db == 'memgraph' + mock_graph_db.driver.assert_called_once_with( + 'bolt://memgraph:7687', + auth=('memgraph', 'password') + ) + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + def test_processor_initialization_with_custom_params(self, mock_graph_db): + """Test processor initialization with custom parameters""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + processor = Processor( + taskgroup=taskgroup_mock, + graph_host='bolt://custom:7687', + username='queryuser', + password='querypass', + database='customdb' + ) + + assert processor.db == 'customdb' + mock_graph_db.driver.assert_called_once_with( + 'bolt://custom:7687', + auth=('queryuser', 'querypass') + ) + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_spo_query(self, mock_graph_db): + """Test SPO query (all values specified)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results - both queries return one record each + mock_records = [MagicMock()] + mock_driver.execute_query.return_value = (mock_records, None, None) + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=Value(value="http://example.com/predicate", is_uri=True), + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify result contains the queried triple (appears twice - once from each query) + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal object" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_sp_query(self, mock_graph_db): + """Test SP query (subject and predicate specified)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results with different objects + mock_record1 = MagicMock() + mock_record1.data.return_value = {"dest": "literal result"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"dest": "http://example.com/uri_result"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=Value(value="http://example.com/predicate", is_uri=True), + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different objects + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal result" + + assert result[1].s.value == "http://example.com/subject" + assert result[1].p.value == "http://example.com/predicate" + assert result[1].o.value == "http://example.com/uri_result" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_so_query(self, mock_graph_db): + """Test SO query (subject and object specified)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results with different predicates + mock_record1 = MagicMock() + mock_record1.data.return_value = {"rel": "http://example.com/pred1"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"rel": "http://example.com/pred2"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=None, + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different predicates + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/pred1" + assert result[0].o.value == "literal object" + + assert result[1].s.value == "http://example.com/subject" + assert result[1].p.value == "http://example.com/pred2" + assert result[1].o.value == "literal object" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_s_query(self, mock_graph_db): + """Test S query (subject only)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results with different predicate-object pairs + mock_record1 = MagicMock() + mock_record1.data.return_value = {"rel": "http://example.com/pred1", "dest": "literal1"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"rel": "http://example.com/pred2", "dest": "http://example.com/uri2"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=None, + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different predicate-object pairs + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/pred1" + assert result[0].o.value == "literal1" + + assert result[1].s.value == "http://example.com/subject" + assert result[1].p.value == "http://example.com/pred2" + assert result[1].o.value == "http://example.com/uri2" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_po_query(self, mock_graph_db): + """Test PO query (predicate and object specified)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results with different subjects + mock_record1 = MagicMock() + mock_record1.data.return_value = {"src": "http://example.com/subj1"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"src": "http://example.com/subj2"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=Value(value="http://example.com/predicate", is_uri=True), + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different subjects + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subj1" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal object" + + assert result[1].s.value == "http://example.com/subj2" + assert result[1].p.value == "http://example.com/predicate" + assert result[1].o.value == "literal object" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_p_query(self, mock_graph_db): + """Test P query (predicate only)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results with different subject-object pairs + mock_record1 = MagicMock() + mock_record1.data.return_value = {"src": "http://example.com/subj1", "dest": "literal1"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"src": "http://example.com/subj2", "dest": "http://example.com/uri2"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=Value(value="http://example.com/predicate", is_uri=True), + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different subject-object pairs + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subj1" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal1" + + assert result[1].s.value == "http://example.com/subj2" + assert result[1].p.value == "http://example.com/predicate" + assert result[1].o.value == "http://example.com/uri2" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_o_query(self, mock_graph_db): + """Test O query (object only)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results with different subject-predicate pairs + mock_record1 = MagicMock() + mock_record1.data.return_value = {"src": "http://example.com/subj1", "rel": "http://example.com/pred1"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"src": "http://example.com/subj2", "rel": "http://example.com/pred2"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=None, + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different subject-predicate pairs + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subj1" + assert result[0].p.value == "http://example.com/pred1" + assert result[0].o.value == "literal object" + + assert result[1].s.value == "http://example.com/subj2" + assert result[1].p.value == "http://example.com/pred2" + assert result[1].o.value == "literal object" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_wildcard_query(self, mock_graph_db): + """Test wildcard query (no constraints)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results + mock_record1 = MagicMock() + mock_record1.data.return_value = {"src": "http://example.com/s1", "rel": "http://example.com/p1", "dest": "literal1"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"src": "http://example.com/s2", "rel": "http://example.com/p2", "dest": "http://example.com/o2"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=None, + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different triples + assert len(result) == 2 + assert result[0].s.value == "http://example.com/s1" + assert result[0].p.value == "http://example.com/p1" + assert result[0].o.value == "literal1" + + assert result[1].s.value == "http://example.com/s2" + assert result[1].p.value == "http://example.com/p2" + assert result[1].o.value == "http://example.com/o2" + + @patch('trustgraph.query.triples.memgraph.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_exception_handling(self, mock_graph_db): + """Test exception handling during query processing""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock execute_query to raise exception + mock_driver.execute_query.side_effect = Exception("Database connection failed") + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=None, + o=None, + limit=100 + ) + + # Should raise the exception + with pytest.raises(Exception, match="Database connection failed"): + await processor.query_triples(query) + + def test_add_args_method(self): + """Test that add_args properly configures argument parser""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + # Mock the parent class add_args method + with patch('trustgraph.query.triples.memgraph.service.TriplesQueryService.add_args') as mock_parent_add_args: + Processor.add_args(parser) + + # Verify parent add_args was called + mock_parent_add_args.assert_called_once() + + # Verify our specific arguments were added + # Parse empty args to check defaults + args = parser.parse_args([]) + + assert hasattr(args, 'graph_host') + assert args.graph_host == 'bolt://memgraph:7687' + assert hasattr(args, 'username') + assert args.username == 'memgraph' + assert hasattr(args, 'password') + assert args.password == 'password' + assert hasattr(args, 'database') + assert args.database == 'memgraph' + + def test_add_args_with_custom_values(self): + """Test add_args with custom command line values""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.query.triples.memgraph.service.TriplesQueryService.add_args'): + Processor.add_args(parser) + + # Test parsing with custom values + args = parser.parse_args([ + '--graph-host', 'bolt://custom:7687', + '--username', 'queryuser', + '--password', 'querypass', + '--database', 'querydb' + ]) + + assert args.graph_host == 'bolt://custom:7687' + assert args.username == 'queryuser' + assert args.password == 'querypass' + assert args.database == 'querydb' + + def test_add_args_short_form(self): + """Test add_args with short form arguments""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.query.triples.memgraph.service.TriplesQueryService.add_args'): + Processor.add_args(parser) + + # Test parsing with short form + args = parser.parse_args(['-g', 'bolt://short:7687']) + + assert args.graph_host == 'bolt://short:7687' + + @patch('trustgraph.query.triples.memgraph.service.Processor.launch') + def test_run_function(self, mock_launch): + """Test the run function calls Processor.launch with correct parameters""" + from trustgraph.query.triples.memgraph.service import run, default_ident + + run() + + mock_launch.assert_called_once_with( + default_ident, + "\nTriples query service for memgraph.\nInput is a (s, p, o) triple, some values may be null. Output is a list of\ntriples.\n" + ) \ No newline at end of file diff --git a/tests/unit/test_query/test_triples_neo4j_query.py b/tests/unit/test_query/test_triples_neo4j_query.py new file mode 100644 index 00000000..320aed54 --- /dev/null +++ b/tests/unit/test_query/test_triples_neo4j_query.py @@ -0,0 +1,338 @@ +""" +Tests for Neo4j triples query service +""" + +import pytest +from unittest.mock import MagicMock, patch + +from trustgraph.query.triples.neo4j.service import Processor +from trustgraph.schema import Value, TriplesQueryRequest + + +class TestNeo4jQueryProcessor: + """Test cases for Neo4j query processor""" + + @pytest.fixture + def processor(self): + """Create a processor instance for testing""" + with patch('trustgraph.query.triples.neo4j.service.GraphDatabase'): + return Processor( + taskgroup=MagicMock(), + id='test-neo4j-query', + graph_host='bolt://localhost:7687' + ) + + def test_create_value_with_http_uri(self, processor): + """Test create_value with HTTP URI""" + result = processor.create_value("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): + """Test create_value with HTTPS URI""" + result = processor.create_value("https://example.com/resource") + + 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): + """Test create_value with literal value""" + result = processor.create_value("just a literal string") + + assert isinstance(result, Value) + assert result.value == "just a literal string" + assert result.is_uri is False + + def test_create_value_with_empty_string(self, processor): + """Test create_value with empty string""" + result = processor.create_value("") + + assert isinstance(result, Value) + assert result.value == "" + assert result.is_uri is False + + def test_create_value_with_partial_uri(self, processor): + """Test create_value with string that looks like URI but isn't complete""" + result = processor.create_value("http") + + assert isinstance(result, Value) + assert result.value == "http" + assert result.is_uri is False + + def test_create_value_with_ftp_uri(self, processor): + """Test create_value with FTP URI (should not be detected as URI)""" + result = processor.create_value("ftp://example.com/file") + + assert isinstance(result, Value) + assert result.value == "ftp://example.com/file" + assert result.is_uri is False + + @patch('trustgraph.query.triples.neo4j.service.GraphDatabase') + def test_processor_initialization_with_defaults(self, mock_graph_db): + """Test processor initialization with default parameters""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + processor = Processor(taskgroup=taskgroup_mock) + + assert processor.db == 'neo4j' + mock_graph_db.driver.assert_called_once_with( + 'bolt://neo4j:7687', + auth=('neo4j', 'password') + ) + + @patch('trustgraph.query.triples.neo4j.service.GraphDatabase') + def test_processor_initialization_with_custom_params(self, mock_graph_db): + """Test processor initialization with custom parameters""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + processor = Processor( + taskgroup=taskgroup_mock, + graph_host='bolt://custom:7687', + username='queryuser', + password='querypass', + database='customdb' + ) + + assert processor.db == 'customdb' + mock_graph_db.driver.assert_called_once_with( + 'bolt://custom:7687', + auth=('queryuser', 'querypass') + ) + + @patch('trustgraph.query.triples.neo4j.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_spo_query(self, mock_graph_db): + """Test SPO query (all values specified)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results - both queries return one record each + mock_records = [MagicMock()] + mock_driver.execute_query.return_value = (mock_records, None, None) + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=Value(value="http://example.com/predicate", is_uri=True), + o=Value(value="literal object", is_uri=False), + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify result contains the queried triple (appears twice - once from each query) + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal object" + + @patch('trustgraph.query.triples.neo4j.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_sp_query(self, mock_graph_db): + """Test SP query (subject and predicate specified)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results with different objects + mock_record1 = MagicMock() + mock_record1.data.return_value = {"dest": "literal result"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"dest": "http://example.com/uri_result"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=Value(value="http://example.com/predicate", is_uri=True), + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different objects + assert len(result) == 2 + assert result[0].s.value == "http://example.com/subject" + assert result[0].p.value == "http://example.com/predicate" + assert result[0].o.value == "literal result" + + assert result[1].s.value == "http://example.com/subject" + assert result[1].p.value == "http://example.com/predicate" + assert result[1].o.value == "http://example.com/uri_result" + + @patch('trustgraph.query.triples.neo4j.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_wildcard_query(self, mock_graph_db): + """Test wildcard query (no constraints)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock query results + mock_record1 = MagicMock() + mock_record1.data.return_value = {"src": "http://example.com/s1", "rel": "http://example.com/p1", "dest": "literal1"} + mock_record2 = MagicMock() + mock_record2.data.return_value = {"src": "http://example.com/s2", "rel": "http://example.com/p2", "dest": "http://example.com/o2"} + + mock_driver.execute_query.side_effect = [ + ([mock_record1], None, None), # Literal query + ([mock_record2], None, None) # URI query + ] + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=None, + p=None, + o=None, + limit=100 + ) + + result = await processor.query_triples(query) + + # Verify both literal and URI queries were executed + assert mock_driver.execute_query.call_count == 2 + + # Verify results contain different triples + assert len(result) == 2 + assert result[0].s.value == "http://example.com/s1" + assert result[0].p.value == "http://example.com/p1" + assert result[0].o.value == "literal1" + + assert result[1].s.value == "http://example.com/s2" + assert result[1].p.value == "http://example.com/p2" + assert result[1].o.value == "http://example.com/o2" + + @patch('trustgraph.query.triples.neo4j.service.GraphDatabase') + @pytest.mark.asyncio + async def test_query_triples_exception_handling(self, mock_graph_db): + """Test exception handling during query processing""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + + # Mock execute_query to raise exception + mock_driver.execute_query.side_effect = Exception("Database connection failed") + + processor = Processor(taskgroup=taskgroup_mock) + + # Create query request + query = TriplesQueryRequest( + user='test_user', + collection='test_collection', + s=Value(value="http://example.com/subject", is_uri=True), + p=None, + o=None, + limit=100 + ) + + # Should raise the exception + with pytest.raises(Exception, match="Database connection failed"): + await processor.query_triples(query) + + def test_add_args_method(self): + """Test that add_args properly configures argument parser""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + # Mock the parent class add_args method + with patch('trustgraph.query.triples.neo4j.service.TriplesQueryService.add_args') as mock_parent_add_args: + Processor.add_args(parser) + + # Verify parent add_args was called + mock_parent_add_args.assert_called_once() + + # Verify our specific arguments were added + # Parse empty args to check defaults + args = parser.parse_args([]) + + assert hasattr(args, 'graph_host') + assert args.graph_host == 'bolt://neo4j:7687' + assert hasattr(args, 'username') + assert args.username == 'neo4j' + assert hasattr(args, 'password') + assert args.password == 'password' + assert hasattr(args, 'database') + assert args.database == 'neo4j' + + def test_add_args_with_custom_values(self): + """Test add_args with custom command line values""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.query.triples.neo4j.service.TriplesQueryService.add_args'): + Processor.add_args(parser) + + # Test parsing with custom values + args = parser.parse_args([ + '--graph-host', 'bolt://custom:7687', + '--username', 'queryuser', + '--password', 'querypass', + '--database', 'querydb' + ]) + + assert args.graph_host == 'bolt://custom:7687' + assert args.username == 'queryuser' + assert args.password == 'querypass' + assert args.database == 'querydb' + + def test_add_args_short_form(self): + """Test add_args with short form arguments""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.query.triples.neo4j.service.TriplesQueryService.add_args'): + Processor.add_args(parser) + + # Test parsing with short form + args = parser.parse_args(['-g', 'bolt://short:7687']) + + assert args.graph_host == 'bolt://short:7687' + + @patch('trustgraph.query.triples.neo4j.service.Processor.launch') + def test_run_function(self, mock_launch): + """Test the run function calls Processor.launch with correct parameters""" + from trustgraph.query.triples.neo4j.service import run, default_ident + + run() + + mock_launch.assert_called_once_with( + default_ident, + "\nTriples query service for neo4j.\nInput is a (s, p, o) triple, some values may be null. Output is a list of\ntriples.\n" + ) \ No newline at end of file diff --git a/tests/unit/test_storage/test_triples_neo4j_storage.py b/tests/unit/test_storage/test_triples_neo4j_storage.py new file mode 100644 index 00000000..a84706ee --- /dev/null +++ b/tests/unit/test_storage/test_triples_neo4j_storage.py @@ -0,0 +1,548 @@ +""" +Tests for Neo4j triples storage service +""" + +import pytest +from unittest.mock import MagicMock, patch, AsyncMock + +from trustgraph.storage.triples.neo4j.write import Processor + + +class TestNeo4jStorageProcessor: + """Test cases for Neo4j storage processor""" + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_processor_initialization_with_defaults(self, mock_graph_db): + """Test processor initialization with default parameters""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + processor = Processor(taskgroup=taskgroup_mock) + + assert processor.db == 'neo4j' + mock_graph_db.driver.assert_called_once_with( + 'bolt://neo4j:7687', + auth=('neo4j', 'password') + ) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_processor_initialization_with_custom_params(self, mock_graph_db): + """Test processor initialization with custom parameters""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + processor = Processor( + taskgroup=taskgroup_mock, + graph_host='bolt://custom:7687', + username='testuser', + password='testpass', + database='testdb' + ) + + assert processor.db == 'testdb' + mock_graph_db.driver.assert_called_once_with( + 'bolt://custom:7687', + auth=('testuser', 'testpass') + ) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_create_indexes_success(self, mock_graph_db): + """Test successful index creation""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + processor = Processor(taskgroup=taskgroup_mock) + + # Verify index creation queries were executed + expected_calls = [ + "CREATE INDEX Node_uri FOR (n:Node) ON (n.uri)", + "CREATE INDEX Literal_value FOR (n:Literal) ON (n.value)", + "CREATE INDEX Rel_uri FOR ()-[r:Rel]-() ON (r.uri)" + ] + + assert mock_session.run.call_count == 3 + for expected_query in expected_calls: + mock_session.run.assert_any_call(expected_query) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_create_indexes_with_exceptions(self, mock_graph_db): + """Test index creation with exceptions (should be ignored)""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Make session.run raise exceptions + mock_session.run.side_effect = Exception("Index already exists") + + # Should not raise exception - they should be caught and ignored + processor = Processor(taskgroup=taskgroup_mock) + + # Should have tried to create all 3 indexes despite exceptions + assert mock_session.run.call_count == 3 + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_create_node(self, mock_graph_db): + """Test node creation""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 1 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Test create_node + processor.create_node("http://example.com/node") + + mock_driver.execute_query.assert_called_with( + "MERGE (n:Node {uri: $uri})", + uri="http://example.com/node", + database_="neo4j" + ) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_create_literal(self, mock_graph_db): + """Test literal creation""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 1 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Test create_literal + processor.create_literal("literal value") + + mock_driver.execute_query.assert_called_with( + "MERGE (n:Literal {value: $value})", + value="literal value", + database_="neo4j" + ) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_relate_node(self, mock_graph_db): + """Test node-to-node relationship creation""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 0 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Test relate_node + processor.relate_node( + "http://example.com/subject", + "http://example.com/predicate", + "http://example.com/object" + ) + + mock_driver.execute_query.assert_called_with( + "MATCH (src:Node {uri: $src}) " + "MATCH (dest:Node {uri: $dest}) " + "MERGE (src)-[:Rel {uri: $uri}]->(dest)", + src="http://example.com/subject", + dest="http://example.com/object", + uri="http://example.com/predicate", + database_="neo4j" + ) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + def test_relate_literal(self, mock_graph_db): + """Test node-to-literal relationship creation""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 0 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Test relate_literal + processor.relate_literal( + "http://example.com/subject", + "http://example.com/predicate", + "literal value" + ) + + mock_driver.execute_query.assert_called_with( + "MATCH (src:Node {uri: $src}) " + "MATCH (dest:Literal {value: $dest}) " + "MERGE (src)-[:Rel {uri: $uri}]->(dest)", + src="http://example.com/subject", + dest="literal value", + uri="http://example.com/predicate", + database_="neo4j" + ) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + @pytest.mark.asyncio + async def test_handle_triples_with_uri_object(self, mock_graph_db): + """Test handling triples message with URI object""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 1 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Create mock triple with URI object + triple = MagicMock() + triple.s.value = "http://example.com/subject" + triple.p.value = "http://example.com/predicate" + triple.o.value = "http://example.com/object" + triple.o.is_uri = True + + # Create mock message + mock_message = MagicMock() + mock_message.triples = [triple] + + await processor.store_triples(mock_message) + + # Verify create_node was called for subject and object + # Verify relate_node was called + expected_calls = [ + # Subject node creation + ( + "MERGE (n:Node {uri: $uri})", + {"uri": "http://example.com/subject", "database_": "neo4j"} + ), + # Object node creation + ( + "MERGE (n:Node {uri: $uri})", + {"uri": "http://example.com/object", "database_": "neo4j"} + ), + # Relationship creation + ( + "MATCH (src:Node {uri: $src}) " + "MATCH (dest:Node {uri: $dest}) " + "MERGE (src)-[:Rel {uri: $uri}]->(dest)", + { + "src": "http://example.com/subject", + "dest": "http://example.com/object", + "uri": "http://example.com/predicate", + "database_": "neo4j" + } + ) + ] + + assert mock_driver.execute_query.call_count == 3 + for expected_query, expected_params in expected_calls: + mock_driver.execute_query.assert_any_call(expected_query, **expected_params) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + @pytest.mark.asyncio + async def test_store_triples_with_literal_object(self, mock_graph_db): + """Test handling triples message with literal object""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 1 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Create mock triple with literal object + triple = MagicMock() + triple.s.value = "http://example.com/subject" + triple.p.value = "http://example.com/predicate" + triple.o.value = "literal value" + triple.o.is_uri = False + + # Create mock message + mock_message = MagicMock() + mock_message.triples = [triple] + + await processor.store_triples(mock_message) + + # Verify create_node was called for subject + # Verify create_literal was called for object + # Verify relate_literal was called + expected_calls = [ + # Subject node creation + ( + "MERGE (n:Node {uri: $uri})", + {"uri": "http://example.com/subject", "database_": "neo4j"} + ), + # Literal creation + ( + "MERGE (n:Literal {value: $value})", + {"value": "literal value", "database_": "neo4j"} + ), + # Relationship creation + ( + "MATCH (src:Node {uri: $src}) " + "MATCH (dest:Literal {value: $dest}) " + "MERGE (src)-[:Rel {uri: $uri}]->(dest)", + { + "src": "http://example.com/subject", + "dest": "literal value", + "uri": "http://example.com/predicate", + "database_": "neo4j" + } + ) + ] + + assert mock_driver.execute_query.call_count == 3 + for expected_query, expected_params in expected_calls: + mock_driver.execute_query.assert_any_call(expected_query, **expected_params) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + @pytest.mark.asyncio + async def test_store_multiple_triples(self, mock_graph_db): + """Test handling message with multiple triples""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 1 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Create mock triples + triple1 = MagicMock() + triple1.s.value = "http://example.com/subject1" + triple1.p.value = "http://example.com/predicate1" + triple1.o.value = "http://example.com/object1" + triple1.o.is_uri = True + + triple2 = MagicMock() + triple2.s.value = "http://example.com/subject2" + triple2.p.value = "http://example.com/predicate2" + triple2.o.value = "literal value" + triple2.o.is_uri = False + + # Create mock message + mock_message = MagicMock() + mock_message.triples = [triple1, triple2] + + await processor.store_triples(mock_message) + + # Should have processed both triples + # Triple1: 2 nodes + 1 relationship = 3 calls + # Triple2: 1 node + 1 literal + 1 relationship = 3 calls + # Total: 6 calls + assert mock_driver.execute_query.call_count == 6 + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + @pytest.mark.asyncio + async def test_store_empty_triples(self, mock_graph_db): + """Test handling message with no triples""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + processor = Processor(taskgroup=taskgroup_mock) + + # Create mock message with empty triples + mock_message = MagicMock() + mock_message.triples = [] + + await processor.store_triples(mock_message) + + # Should not have made any execute_query calls beyond index creation + # Only index creation calls should have been made during initialization + mock_driver.execute_query.assert_not_called() + + def test_add_args_method(self): + """Test that add_args properly configures argument parser""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + # Mock the parent class add_args method + with patch('trustgraph.storage.triples.neo4j.write.TriplesStoreService.add_args') as mock_parent_add_args: + Processor.add_args(parser) + + # Verify parent add_args was called + mock_parent_add_args.assert_called_once() + + # Verify our specific arguments were added + # Parse empty args to check defaults + args = parser.parse_args([]) + + assert hasattr(args, 'graph_host') + assert args.graph_host == 'bolt://neo4j:7687' + assert hasattr(args, 'username') + assert args.username == 'neo4j' + assert hasattr(args, 'password') + assert args.password == 'password' + assert hasattr(args, 'database') + assert args.database == 'neo4j' + + def test_add_args_with_custom_values(self): + """Test add_args with custom command line values""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.storage.triples.neo4j.write.TriplesStoreService.add_args'): + Processor.add_args(parser) + + # Test parsing with custom values + args = parser.parse_args([ + '--graph_host', 'bolt://custom:7687', + '--username', 'testuser', + '--password', 'testpass', + '--database', 'testdb' + ]) + + assert args.graph_host == 'bolt://custom:7687' + assert args.username == 'testuser' + assert args.password == 'testpass' + assert args.database == 'testdb' + + def test_add_args_short_form(self): + """Test add_args with short form arguments""" + from argparse import ArgumentParser + from unittest.mock import patch + + parser = ArgumentParser() + + with patch('trustgraph.storage.triples.neo4j.write.TriplesStoreService.add_args'): + Processor.add_args(parser) + + # Test parsing with short form + args = parser.parse_args(['-g', 'bolt://short:7687']) + + assert args.graph_host == 'bolt://short:7687' + + @patch('trustgraph.storage.triples.neo4j.write.Processor.launch') + def test_run_function(self, mock_launch): + """Test the run function calls Processor.launch with correct parameters""" + from trustgraph.storage.triples.neo4j.write import run, default_ident + + run() + + mock_launch.assert_called_once_with( + default_ident, + "\nGraph writer. Input is graph edge. Writes edges to Neo4j graph.\n" + ) + + @patch('trustgraph.storage.triples.neo4j.write.GraphDatabase') + @pytest.mark.asyncio + async def test_store_triples_with_special_characters(self, mock_graph_db): + """Test handling triples with special characters and unicode""" + taskgroup_mock = MagicMock() + mock_driver = MagicMock() + mock_graph_db.driver.return_value = mock_driver + mock_session = MagicMock() + mock_driver.session.return_value.__enter__.return_value = mock_session + + # Mock execute_query response + mock_result = MagicMock() + mock_summary = MagicMock() + mock_summary.counters.nodes_created = 1 + mock_summary.result_available_after = 10 + mock_result.summary = mock_summary + mock_driver.execute_query.return_value = mock_result + + processor = Processor(taskgroup=taskgroup_mock) + + # Create triple with special characters + triple = MagicMock() + triple.s.value = "http://example.com/subject with spaces" + triple.p.value = "http://example.com/predicate:with/symbols" + triple.o.value = 'literal with "quotes" and unicode: ñáéíóú' + triple.o.is_uri = False + + mock_message = MagicMock() + mock_message.triples = [triple] + + await processor.store_triples(mock_message) + + # Verify the triple was processed with special characters preserved + mock_driver.execute_query.assert_any_call( + "MERGE (n:Node {uri: $uri})", + uri="http://example.com/subject with spaces", + database_="neo4j" + ) + + mock_driver.execute_query.assert_any_call( + "MERGE (n:Literal {value: $value})", + value='literal with "quotes" and unicode: ñáéíóú', + database_="neo4j" + ) + + mock_driver.execute_query.assert_any_call( + "MATCH (src:Node {uri: $src}) " + "MATCH (dest:Literal {value: $dest}) " + "MERGE (src)-[:Rel {uri: $uri}]->(dest)", + src="http://example.com/subject with spaces", + dest='literal with "quotes" and unicode: ñáéíóú', + uri="http://example.com/predicate:with/symbols", + database_="neo4j" + ) diff --git a/trustgraph-flow/trustgraph/query/triples/falkordb/service.py b/trustgraph-flow/trustgraph/query/triples/falkordb/service.py index c62c28c1..2bbe5e2f 100755 --- a/trustgraph-flow/trustgraph/query/triples/falkordb/service.py +++ b/trustgraph-flow/trustgraph/query/triples/falkordb/service.py @@ -9,37 +9,24 @@ from falkordb import FalkorDB from .... schema import TriplesQueryRequest, TriplesQueryResponse, Error from .... schema import Value, Triple -from .... schema import triples_request_queue -from .... schema import triples_response_queue -from .... base import ConsumerProducer +from .... base import TriplesQueryService -module = "triples-query" - -default_input_queue = triples_request_queue -default_output_queue = triples_response_queue -default_subscriber = module +default_ident = "triples-query" default_graph_url = 'falkor://falkordb:6379' default_database = 'falkordb' -class Processor(ConsumerProducer): +class Processor(TriplesQueryService): def __init__(self, **params): - input_queue = params.get("input_queue", default_input_queue) - output_queue = params.get("output_queue", default_output_queue) - subscriber = params.get("subscriber", default_subscriber) - graph_url = params.get("graph_host", default_graph_url) + graph_url = params.get("graph_url", default_graph_url) database = params.get("database", default_database) super(Processor, self).__init__( **params | { - "input_queue": input_queue, - "output_queue": output_queue, - "subscriber": subscriber, - "input_schema": TriplesQueryRequest, - "output_schema": TriplesQueryResponse, "graph_url": graph_url, + "database": database, } ) @@ -54,50 +41,45 @@ class Processor(ConsumerProducer): else: return Value(value=ent, is_uri=False) - async def handle(self, msg): + async def query_triples(self, query): try: - v = msg.value() - - # Sender-produced ID - id = msg.properties()["id"] - - print(f"Handling input {id}...", flush=True) - triples = [] - if v.s is not None: - if v.p is not None: - if v.o is not None: + if query.s is not None: + if query.p is not None: + if query.o is not None: # SPO records = self.io.query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal {value: $value}) " - "RETURN $src as src", + "RETURN $src as src " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, - "rel": v.p.value, - "value": v.o.value, + "src": query.s.value, + "rel": query.p.value, + "value": query.o.value, }, ).result_set for rec in records: - triples.append((v.s.value, v.p.value, v.o.value)) + triples.append((query.s.value, query.p.value, query.o.value)) records = self.io.query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node {uri: $uri}) " - "RETURN $src as src", + "RETURN $src as src " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, - "rel": v.p.value, - "uri": v.o.value, + "src": query.s.value, + "rel": query.p.value, + "uri": query.o.value, }, ).result_set for rec in records: - triples.append((v.s.value, v.p.value, v.o.value)) + triples.append((query.s.value, query.p.value, query.o.value)) else: @@ -105,116 +87,124 @@ class Processor(ConsumerProducer): records = self.io.query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal) " - "RETURN dest.value as dest", + "RETURN dest.value as dest " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, - "rel": v.p.value, + "src": query.s.value, + "rel": query.p.value, }, ).result_set for rec in records: - triples.append((v.s.value, v.p.value, rec[0])) + triples.append((query.s.value, query.p.value, rec[0])) records = self.io.query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node) " - "RETURN dest.uri as dest", + "RETURN dest.uri as dest " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, - "rel": v.p.value, + "src": query.s.value, + "rel": query.p.value, }, ).result_set for rec in records: - triples.append((v.s.value, v.p.value, rec[0])) + triples.append((query.s.value, query.p.value, rec[0])) else: - if v.o is not None: + if query.o is not None: # SO records = self.io.query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal {value: $value}) " - "RETURN rel.uri as rel", + "RETURN rel.uri as rel " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, - "value": v.o.value, + "src": query.s.value, + "value": query.o.value, }, ).result_set for rec in records: - triples.append((v.s.value, rec[0], v.o.value)) + triples.append((query.s.value, rec[0], query.o.value)) records = self.io.query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node {uri: $uri}) " - "RETURN rel.uri as rel", + "RETURN rel.uri as rel " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, - "uri": v.o.value, + "src": query.s.value, + "uri": query.o.value, }, ).result_set for rec in records: - triples.append((v.s.value, rec[0], v.o.value)) + triples.append((query.s.value, rec[0], query.o.value)) else: # s records = self.io.query( - "match (src:node {uri: $src})-[rel:rel]->(dest:literal) " - "return rel.uri as rel, dest.value as dest", + "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal) " + "RETURN rel.uri as rel, dest.value as dest " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, + "src": query.s.value, }, ).result_set for rec in records: - triples.append((v.s.value, rec[0], rec[1])) + triples.append((query.s.value, rec[0], rec[1])) records = self.io.query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node) " - "RETURN rel.uri as rel, dest.uri as dest", + "RETURN rel.uri as rel, dest.uri as dest " + "LIMIT " + str(query.limit), params={ - "src": v.s.value, + "src": query.s.value, }, ).result_set for rec in records: - triples.append((v.s.value, rec[0], rec[1])) + triples.append((query.s.value, rec[0], rec[1])) else: - if v.p is not None: + if query.p is not None: - if v.o is not None: + if query.o is not None: # PO records = self.io.query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal {value: $value}) " - "RETURN src.uri as src", + "RETURN src.uri as src " + "LIMIT " + str(query.limit), params={ - "uri": v.p.value, - "value": v.o.value, + "uri": query.p.value, + "value": query.o.value, }, ).result_set for rec in records: - triples.append((rec[0], v.p.value, v.o.value)) + triples.append((rec[0], query.p.value, query.o.value)) records = self.io.query( - "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node {uri: $uri}) " - "RETURN src.uri as src", + "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node {uri: $dest}) " + "RETURN src.uri as src " + "LIMIT " + str(query.limit), params={ - "uri": v.p.value, - "dest": v.o.value, + "uri": query.p.value, + "dest": query.o.value, }, ).result_set for rec in records: - triples.append((rec[0], v.p.value, v.o.value)) + triples.append((rec[0], query.p.value, query.o.value)) else: @@ -222,53 +212,57 @@ class Processor(ConsumerProducer): records = self.io.query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal) " - "RETURN src.uri as src, dest.value as dest", + "RETURN src.uri as src, dest.value as dest " + "LIMIT " + str(query.limit), params={ - "uri": v.p.value, + "uri": query.p.value, }, ).result_set for rec in records: - triples.append((rec[0], v.p.value, rec[1])) + triples.append((rec[0], query.p.value, rec[1])) records = self.io.query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node) " - "RETURN src.uri as src, dest.uri as dest", + "RETURN src.uri as src, dest.uri as dest " + "LIMIT " + str(query.limit), params={ - "uri": v.p.value, + "uri": query.p.value, }, ).result_set for rec in records: - triples.append((rec[0], v.p.value, rec[1])) + triples.append((rec[0], query.p.value, rec[1])) else: - if v.o is not None: + if query.o is not None: # O records = self.io.query( "MATCH (src:Node)-[rel:Rel]->(dest:Literal {value: $value}) " - "RETURN src.uri as src, rel.uri as rel", + "RETURN src.uri as src, rel.uri as rel " + "LIMIT " + str(query.limit), params={ - "value": v.o.value, + "value": query.o.value, }, ).result_set for rec in records: - triples.append((rec[0], rec[1], v.o.value)) + triples.append((rec[0], rec[1], query.o.value)) records = self.io.query( "MATCH (src:Node)-[rel:Rel]->(dest:Node {uri: $uri}) " - "RETURN src.uri as src, rel.uri as rel", + "RETURN src.uri as src, rel.uri as rel " + "LIMIT " + str(query.limit), params={ - "uri": v.o.value, + "uri": query.o.value, }, ).result_set for rec in records: - triples.append((rec[0], rec[1], v.o.value)) + triples.append((rec[0], rec[1], query.o.value)) else: @@ -276,7 +270,8 @@ class Processor(ConsumerProducer): records = self.io.query( "MATCH (src:Node)-[rel:Rel]->(dest:Literal) " - "RETURN src.uri as src, rel.uri as rel, dest.value as dest", + "RETURN src.uri as src, rel.uri as rel, dest.value as dest " + "LIMIT " + str(query.limit), ).result_set for rec in records: @@ -284,7 +279,8 @@ class Processor(ConsumerProducer): records = self.io.query( "MATCH (src:Node)-[rel:Rel]->(dest:Node) " - "RETURN src.uri as src, rel.uri as rel, dest.uri as dest", + "RETURN src.uri as src, rel.uri as rel, dest.uri as dest " + "LIMIT " + str(query.limit), ).result_set for rec in records: @@ -296,40 +292,20 @@ class Processor(ConsumerProducer): p=self.create_value(t[1]), o=self.create_value(t[2]) ) - for t in triples + for t in triples[:query.limit] ] - print("Send response...", flush=True) - r = TriplesQueryResponse(triples=triples, error=None) - await self.send(r, properties={"id": id}) - - print("Done.", flush=True) + return triples except Exception as e: print(f"Exception: {e}") - - print("Send error response...", flush=True) - - r = TriplesQueryResponse( - error=Error( - type = "llm-error", - message = str(e), - ), - response=None, - ) - - await self.send(r, properties={"id": id}) - - self.consumer.acknowledge(msg) + raise e @staticmethod def add_args(parser): - ConsumerProducer.add_args( - parser, default_input_queue, default_subscriber, - default_output_queue, - ) + TriplesQueryService.add_args(parser) parser.add_argument( '-g', '--graph-url', @@ -345,5 +321,5 @@ class Processor(ConsumerProducer): def run(): - Processor.launch(module, __doc__) + Processor.launch(default_ident, __doc__) diff --git a/trustgraph-flow/trustgraph/query/triples/memgraph/service.py b/trustgraph-flow/trustgraph/query/triples/memgraph/service.py index 594c9130..bc75dd16 100755 --- a/trustgraph-flow/trustgraph/query/triples/memgraph/service.py +++ b/trustgraph-flow/trustgraph/query/triples/memgraph/service.py @@ -9,28 +9,19 @@ from neo4j import GraphDatabase from .... schema import TriplesQueryRequest, TriplesQueryResponse, Error from .... schema import Value, Triple -from .... schema import triples_request_queue -from .... schema import triples_response_queue -from .... base import ConsumerProducer +from .... base import TriplesQueryService -module = "triples-query" - -default_input_queue = triples_request_queue -default_output_queue = triples_response_queue -default_subscriber = module +default_ident = "triples-query" default_graph_host = 'bolt://memgraph:7687' default_username = 'memgraph' default_password = 'password' default_database = 'memgraph' -class Processor(ConsumerProducer): +class Processor(TriplesQueryService): def __init__(self, **params): - input_queue = params.get("input_queue", default_input_queue) - output_queue = params.get("output_queue", default_output_queue) - subscriber = params.get("subscriber", default_subscriber) graph_host = params.get("graph_host", default_graph_host) username = params.get("username", default_username) password = params.get("password", default_password) @@ -38,12 +29,9 @@ class Processor(ConsumerProducer): super(Processor, self).__init__( **params | { - "input_queue": input_queue, - "output_queue": output_queue, - "subscriber": subscriber, - "input_schema": TriplesQueryRequest, - "output_schema": TriplesQueryResponse, "graph_host": graph_host, + "username": username, + "database": database, } ) @@ -58,46 +46,39 @@ class Processor(ConsumerProducer): else: return Value(value=ent, is_uri=False) - async def handle(self, msg): + async def query_triples(self, query): try: - v = msg.value() - - # Sender-produced ID - id = msg.properties()["id"] - - print(f"Handling input {id}...", flush=True) - triples = [] - if v.s is not None: - if v.p is not None: - if v.o is not None: + if query.s is not None: + if query.p is not None: + if query.o is not None: # SPO records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal {value: $value}) " "RETURN $src as src " - "LIMIT " + str(v.limit), - src=v.s.value, rel=v.p.value, value=v.o.value, + "LIMIT " + str(query.limit), + src=query.s.value, rel=query.p.value, value=query.o.value, database_=self.db, ) for rec in records: - triples.append((v.s.value, v.p.value, v.o.value)) + triples.append((query.s.value, query.p.value, query.o.value)) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node {uri: $uri}) " "RETURN $src as src " - "LIMIT " + str(v.limit), - src=v.s.value, rel=v.p.value, uri=v.o.value, + "LIMIT " + str(query.limit), + src=query.s.value, rel=query.p.value, uri=query.o.value, database_=self.db, ) for rec in records: - triples.append((v.s.value, v.p.value, v.o.value)) + triples.append((query.s.value, query.p.value, query.o.value)) else: @@ -106,56 +87,56 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal) " "RETURN dest.value as dest " - "LIMIT " + str(v.limit), - src=v.s.value, rel=v.p.value, + "LIMIT " + str(query.limit), + src=query.s.value, rel=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, v.p.value, data["dest"])) + triples.append((query.s.value, query.p.value, data["dest"])) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node) " "RETURN dest.uri as dest " - "LIMIT " + str(v.limit), - src=v.s.value, rel=v.p.value, + "LIMIT " + str(query.limit), + src=query.s.value, rel=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, v.p.value, data["dest"])) + triples.append((query.s.value, query.p.value, data["dest"])) else: - if v.o is not None: + if query.o is not None: # SO records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal {value: $value}) " "RETURN rel.uri as rel " - "LIMIT " + str(v.limit), - src=v.s.value, value=v.o.value, + "LIMIT " + str(query.limit), + src=query.s.value, value=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], v.o.value)) + triples.append((query.s.value, data["rel"], query.o.value)) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node {uri: $uri}) " "RETURN rel.uri as rel " - "LIMIT " + str(v.limit), - src=v.s.value, uri=v.o.value, + "LIMIT " + str(query.limit), + src=query.s.value, uri=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], v.o.value)) + triples.append((query.s.value, data["rel"], query.o.value)) else: @@ -164,59 +145,59 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal) " "RETURN rel.uri as rel, dest.value as dest " - "LIMIT " + str(v.limit), - src=v.s.value, + "LIMIT " + str(query.limit), + src=query.s.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], data["dest"])) + triples.append((query.s.value, data["rel"], data["dest"])) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node) " "RETURN rel.uri as rel, dest.uri as dest " - "LIMIT " + str(v.limit), - src=v.s.value, + "LIMIT " + str(query.limit), + src=query.s.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], data["dest"])) + triples.append((query.s.value, data["rel"], data["dest"])) else: - if v.p is not None: + if query.p is not None: - if v.o is not None: + if query.o is not None: # PO records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal {value: $value}) " "RETURN src.uri as src " - "LIMIT " + str(v.limit), - uri=v.p.value, value=v.o.value, + "LIMIT " + str(query.limit), + uri=query.p.value, value=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, v.o.value)) + triples.append((data["src"], query.p.value, query.o.value)) records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node {uri: $uri}) " + "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node {uri: $dest}) " "RETURN src.uri as src " - "LIMIT " + str(v.limit), - uri=v.p.value, dest=v.o.value, + "LIMIT " + str(query.limit), + uri=query.p.value, dest=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, v.o.value)) + triples.append((data["src"], query.p.value, query.o.value)) else: @@ -225,56 +206,56 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal) " "RETURN src.uri as src, dest.value as dest " - "LIMIT " + str(v.limit), - uri=v.p.value, + "LIMIT " + str(query.limit), + uri=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, data["dest"])) + triples.append((data["src"], query.p.value, data["dest"])) records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node) " "RETURN src.uri as src, dest.uri as dest " - "LIMIT " + str(v.limit), - uri=v.p.value, + "LIMIT " + str(query.limit), + uri=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, data["dest"])) + triples.append((data["src"], query.p.value, data["dest"])) else: - if v.o is not None: + if query.o is not None: # O records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel]->(dest:Literal {value: $value}) " "RETURN src.uri as src, rel.uri as rel " - "LIMIT " + str(v.limit), - value=v.o.value, + "LIMIT " + str(query.limit), + value=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], data["rel"], v.o.value)) + triples.append((data["src"], data["rel"], query.o.value)) records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel]->(dest:Node {uri: $uri}) " "RETURN src.uri as src, rel.uri as rel " - "LIMIT " + str(v.limit), - uri=v.o.value, + "LIMIT " + str(query.limit), + uri=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], data["rel"], v.o.value)) + triples.append((data["src"], data["rel"], query.o.value)) else: @@ -283,7 +264,7 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel]->(dest:Literal) " "RETURN src.uri as src, rel.uri as rel, dest.value as dest " - "LIMIT " + str(v.limit), + "LIMIT " + str(query.limit), database_=self.db, ) @@ -294,7 +275,7 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel]->(dest:Node) " "RETURN src.uri as src, rel.uri as rel, dest.uri as dest " - "LIMIT " + str(v.limit), + "LIMIT " + str(query.limit), database_=self.db, ) @@ -308,40 +289,22 @@ class Processor(ConsumerProducer): p=self.create_value(t[1]), o=self.create_value(t[2]) ) - for t in triples[:v.limit] + for t in triples[:query.limit] ] - print("Send response...", flush=True) - r = TriplesQueryResponse(triples=triples, error=None) - await self.send(r, properties={"id": id}) - - print("Done.", flush=True) + return triples except Exception as e: print(f"Exception: {e}") - print("Send error response...", flush=True) - - r = TriplesQueryResponse( - error=Error( - type = "llm-error", - message = str(e), - ), - response=None, - ) - - await self.send(r, properties={"id": id}) - - self.consumer.acknowledge(msg) + print(f"Exception: {e}") + raise e @staticmethod def add_args(parser): - ConsumerProducer.add_args( - parser, default_input_queue, default_subscriber, - default_output_queue, - ) + TriplesQueryService.add_args(parser) parser.add_argument( '-g', '--graph-host', @@ -369,5 +332,5 @@ class Processor(ConsumerProducer): def run(): - Processor.launch(module, __doc__) + Processor.launch(default_ident, __doc__) diff --git a/trustgraph-flow/trustgraph/query/triples/neo4j/service.py b/trustgraph-flow/trustgraph/query/triples/neo4j/service.py index 591361ce..f65c0f56 100755 --- a/trustgraph-flow/trustgraph/query/triples/neo4j/service.py +++ b/trustgraph-flow/trustgraph/query/triples/neo4j/service.py @@ -9,28 +9,19 @@ from neo4j import GraphDatabase from .... schema import TriplesQueryRequest, TriplesQueryResponse, Error from .... schema import Value, Triple -from .... schema import triples_request_queue -from .... schema import triples_response_queue -from .... base import ConsumerProducer +from .... base import TriplesQueryService -module = "triples-query" - -default_input_queue = triples_request_queue -default_output_queue = triples_response_queue -default_subscriber = module +default_ident = "triples-query" default_graph_host = 'bolt://neo4j:7687' default_username = 'neo4j' default_password = 'password' default_database = 'neo4j' -class Processor(ConsumerProducer): +class Processor(TriplesQueryService): def __init__(self, **params): - input_queue = params.get("input_queue", default_input_queue) - output_queue = params.get("output_queue", default_output_queue) - subscriber = params.get("subscriber", default_subscriber) graph_host = params.get("graph_host", default_graph_host) username = params.get("username", default_username) password = params.get("password", default_password) @@ -38,12 +29,9 @@ class Processor(ConsumerProducer): super(Processor, self).__init__( **params | { - "input_queue": input_queue, - "output_queue": output_queue, - "subscriber": subscriber, - "input_schema": TriplesQueryRequest, - "output_schema": TriplesQueryResponse, "graph_host": graph_host, + "username": username, + "database": database, } ) @@ -58,44 +46,37 @@ class Processor(ConsumerProducer): else: return Value(value=ent, is_uri=False) - async def handle(self, msg): + async def query_triples(self, query): try: - v = msg.value() - - # Sender-produced ID - id = msg.properties()["id"] - - print(f"Handling input {id}...", flush=True) - triples = [] - if v.s is not None: - if v.p is not None: - if v.o is not None: + if query.s is not None: + if query.p is not None: + if query.o is not None: # SPO records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal {value: $value}) " "RETURN $src as src", - src=v.s.value, rel=v.p.value, value=v.o.value, + src=query.s.value, rel=query.p.value, value=query.o.value, database_=self.db, ) for rec in records: - triples.append((v.s.value, v.p.value, v.o.value)) + triples.append((query.s.value, query.p.value, query.o.value)) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node {uri: $uri}) " "RETURN $src as src", - src=v.s.value, rel=v.p.value, uri=v.o.value, + src=query.s.value, rel=query.p.value, uri=query.o.value, database_=self.db, ) for rec in records: - triples.append((v.s.value, v.p.value, v.o.value)) + triples.append((query.s.value, query.p.value, query.o.value)) else: @@ -104,52 +85,52 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal) " "RETURN dest.value as dest", - src=v.s.value, rel=v.p.value, + src=query.s.value, rel=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, v.p.value, data["dest"])) + triples.append((query.s.value, query.p.value, data["dest"])) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node) " "RETURN dest.uri as dest", - src=v.s.value, rel=v.p.value, + src=query.s.value, rel=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, v.p.value, data["dest"])) + triples.append((query.s.value, query.p.value, data["dest"])) else: - if v.o is not None: + if query.o is not None: # SO records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal {value: $value}) " "RETURN rel.uri as rel", - src=v.s.value, value=v.o.value, + src=query.s.value, value=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], v.o.value)) + triples.append((query.s.value, data["rel"], query.o.value)) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node {uri: $uri}) " "RETURN rel.uri as rel", - src=v.s.value, uri=v.o.value, + src=query.s.value, uri=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], v.o.value)) + triples.append((query.s.value, data["rel"], query.o.value)) else: @@ -158,55 +139,55 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal) " "RETURN rel.uri as rel, dest.value as dest", - src=v.s.value, + src=query.s.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], data["dest"])) + triples.append((query.s.value, data["rel"], data["dest"])) records, summary, keys = self.io.execute_query( "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node) " "RETURN rel.uri as rel, dest.uri as dest", - src=v.s.value, + src=query.s.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((v.s.value, data["rel"], data["dest"])) + triples.append((query.s.value, data["rel"], data["dest"])) else: - if v.p is not None: + if query.p is not None: - if v.o is not None: + if query.o is not None: # PO records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal {value: $value}) " "RETURN src.uri as src", - uri=v.p.value, value=v.o.value, + uri=query.p.value, value=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, v.o.value)) + triples.append((data["src"], query.p.value, query.o.value)) records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node {uri: $uri}) " + "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node {uri: $dest}) " "RETURN src.uri as src", - uri=v.p.value, dest=v.o.value, + uri=query.p.value, dest=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, v.o.value)) + triples.append((data["src"], query.p.value, query.o.value)) else: @@ -215,52 +196,52 @@ class Processor(ConsumerProducer): records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal) " "RETURN src.uri as src, dest.value as dest", - uri=v.p.value, + uri=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, data["dest"])) + triples.append((data["src"], query.p.value, data["dest"])) records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node) " "RETURN src.uri as src, dest.uri as dest", - uri=v.p.value, + uri=query.p.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], v.p.value, data["dest"])) + triples.append((data["src"], query.p.value, data["dest"])) else: - if v.o is not None: + if query.o is not None: # O records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel]->(dest:Literal {value: $value}) " "RETURN src.uri as src, rel.uri as rel", - value=v.o.value, + value=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], data["rel"], v.o.value)) + triples.append((data["src"], data["rel"], query.o.value)) records, summary, keys = self.io.execute_query( "MATCH (src:Node)-[rel:Rel]->(dest:Node {uri: $uri}) " "RETURN src.uri as src, rel.uri as rel", - uri=v.o.value, + uri=query.o.value, database_=self.db, ) for rec in records: data = rec.data() - triples.append((data["src"], data["rel"], v.o.value)) + triples.append((data["src"], data["rel"], query.o.value)) else: @@ -295,37 +276,17 @@ class Processor(ConsumerProducer): for t in triples ] - print("Send response...", flush=True) - r = TriplesQueryResponse(triples=triples, error=None) - await self.send(r, properties={"id": id}) - - print("Done.", flush=True) + return triples except Exception as e: print(f"Exception: {e}") - - print("Send error response...", flush=True) - - r = TriplesQueryResponse( - error=Error( - type = "llm-error", - message = str(e), - ), - response=None, - ) - - await self.send(r, properties={"id": id}) - - self.consumer.acknowledge(msg) + raise e @staticmethod def add_args(parser): - ConsumerProducer.add_args( - parser, default_input_queue, default_subscriber, - default_output_queue, - ) + TriplesQueryService.add_args(parser) parser.add_argument( '-g', '--graph-host', @@ -353,5 +314,5 @@ class Processor(ConsumerProducer): def run(): - Processor.launch(module, __doc__) + Processor.launch(default_ident, __doc__) diff --git a/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py b/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py index 84a4d923..5293ee1e 100755 --- a/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py @@ -10,28 +10,21 @@ import argparse import time from neo4j import GraphDatabase +from .... base import TriplesStoreService -from .... schema import Triples -from .... schema import triples_store_queue -from .... log_level import LogLevel -from .... base import Consumer - -module = "triples-write" - -default_input_queue = triples_store_queue -default_subscriber = module +default_ident = "triples-write" default_graph_host = 'bolt://neo4j:7687' default_username = 'neo4j' default_password = 'password' default_database = 'neo4j' -class Processor(Consumer): +class Processor(TriplesStoreService): def __init__(self, **params): - input_queue = params.get("input_queue", default_input_queue) - subscriber = params.get("subscriber", default_subscriber) + id = params.get("id", default_ident) + graph_host = params.get("graph_host", default_graph_host) username = params.get("username", default_username) password = params.get("password", default_password) @@ -39,10 +32,9 @@ class Processor(Consumer): super(Processor, self).__init__( **params | { - "input_queue": input_queue, - "subscriber": subscriber, - "input_schema": Triples, "graph_host": graph_host, + "username": username, + "database": database, } ) @@ -158,11 +150,9 @@ class Processor(Consumer): time=summary.result_available_after )) - async def handle(self, msg): + async def store_triples(self, message): - v = msg.value() - - for t in v.triples: + for t in message.triples: self.create_node(t.s.value) @@ -176,9 +166,7 @@ class Processor(Consumer): @staticmethod def add_args(parser): - Consumer.add_args( - parser, default_input_queue, default_subscriber, - ) + TriplesStoreService.add_args(parser) parser.add_argument( '-g', '--graph_host', @@ -206,5 +194,5 @@ class Processor(Consumer): def run(): - Processor.launch(module, __doc__) + Processor.launch(default_ident, __doc__)