From c43f079c5048cbfa68c30624e106cb8aff75844b Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 3 Dec 2025 19:30:04 +0000 Subject: [PATCH] API tests --- .../test_python_api_integration.py | 384 --------------- tests/unit/test_python_api_client.py | 446 ++++++++++++++++++ 2 files changed, 446 insertions(+), 384 deletions(-) delete mode 100644 tests/integration/test_python_api_integration.py create mode 100644 tests/unit/test_python_api_client.py diff --git a/tests/integration/test_python_api_integration.py b/tests/integration/test_python_api_integration.py deleted file mode 100644 index e222681e..00000000 --- a/tests/integration/test_python_api_integration.py +++ /dev/null @@ -1,384 +0,0 @@ -""" -Integration tests for TrustGraph Python API - -These tests require a running TrustGraph Gateway API server. -Set environment variable TRUSTGRAPH_URL to point to your server. -Set TRUSTGRAPH_TOKEN if authentication is required. - -Example: - export TRUSTGRAPH_URL=http://localhost:8088/ - export TRUSTGRAPH_TOKEN=your-token-here - pytest tests/test_api_integration.py -v -""" - -import os -import pytest -import asyncio -from typing import List, Dict, Any - -from trustgraph.api import ( - Api, - Triple, - AgentThought, - AgentObservation, - AgentAnswer, - RAGChunk, -) - - -# Configuration from environment -GATEWAY_URL = os.getenv("TRUSTGRAPH_URL", "http://localhost:8088/") -AUTH_TOKEN = os.getenv("TRUSTGRAPH_TOKEN", None) -TEST_FLOW_ID = os.getenv("TRUSTGRAPH_TEST_FLOW", "test-flow") -TEST_USER = "test-user" -TEST_COLLECTION = "test-collection" - -# Skip tests if gateway is not available -SKIP_INTEGRATION = os.getenv("SKIP_INTEGRATION_TESTS", "false").lower() == "true" -skip_if_no_gateway = pytest.mark.skipif( - SKIP_INTEGRATION, - reason="Integration tests disabled (set SKIP_INTEGRATION_TESTS=false to enable)" -) - - -class TestBasicConnection: - """Test basic API instantiation and connectivity""" - - @skip_if_no_gateway - def test_api_instantiation(self): - """Test that Api class can be instantiated""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - assert api.url.endswith("api/v1/") - assert api.timeout == 60 - assert api.token == AUTH_TOKEN - - @skip_if_no_gateway - def test_api_with_context_manager(self): - """Test Api works as context manager""" - with Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) as api: - assert api is not None - # Context manager should work - # Should exit cleanly - - @skip_if_no_gateway - def test_client_lazy_initialization(self): - """Test that clients are lazily initialized""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - - # Clients should be None initially - assert api._socket_client is None - assert api._bulk_client is None - assert api._async_flow is None - - # Access should initialize - socket = api.socket() - assert api._socket_client is not None - assert socket is api._socket_client - - # Second access should return same instance - socket2 = api.socket() - assert socket2 is socket - - -class TestRESTAPI: - """Test REST API functionality""" - - @skip_if_no_gateway - def test_flow_list(self): - """Test listing flows via REST API""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - flow = api.flow() - - # Should be able to list flows - flows = flow.list() - assert isinstance(flows, list) - - @skip_if_no_gateway - def test_flow_class_list(self): - """Test listing flow classes""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - flow = api.flow() - - # Should be able to list classes - classes = flow.list_classes() - assert isinstance(classes, list) - - @skip_if_no_gateway - def test_flow_instance_creation(self): - """Test creating flow instance""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - flow = api.flow() - flow_instance = flow.id(TEST_FLOW_ID) - - assert flow_instance is not None - assert flow_instance.id == TEST_FLOW_ID - - @skip_if_no_gateway - def test_graph_embeddings_query_method_exists(self): - """Test that graph_embeddings_query method exists""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - flow = api.flow() - flow_instance = flow.id(TEST_FLOW_ID) - - assert hasattr(flow_instance, 'graph_embeddings_query') - - -class TestAsyncRESTAPI: - """Test asynchronous REST API functionality""" - - @skip_if_no_gateway - @pytest.mark.asyncio - async def test_async_flow_list(self): - """Test listing flows via async REST API""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_flow = api.async_flow() - - # Should be able to list flows - flows = await async_flow.list() - assert isinstance(flows, list) - - @skip_if_no_gateway - @pytest.mark.asyncio - async def test_async_flow_class_list(self): - """Test listing flow classes asynchronously""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_flow = api.async_flow() - - # Should be able to list classes - classes = await async_flow.list_classes() - assert isinstance(classes, list) - - @skip_if_no_gateway - @pytest.mark.asyncio - async def test_async_flow_instance_creation(self): - """Test creating async flow instance""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_flow = api.async_flow() - flow_instance = async_flow.id(TEST_FLOW_ID) - - assert flow_instance is not None - assert flow_instance.flow_id == TEST_FLOW_ID - - @skip_if_no_gateway - @pytest.mark.asyncio - async def test_async_context_manager(self): - """Test async context manager""" - async with Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) as api: - async_flow = api.async_flow() - flows = await async_flow.list() - assert isinstance(flows, list) - - -class TestWebSocketAPI: - """Test WebSocket API functionality""" - - @skip_if_no_gateway - def test_socket_client_creation(self): - """Test WebSocket client instantiation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - socket = api.socket() - - assert socket is not None - assert socket.url.startswith("ws://") or socket.url.startswith("wss://") - assert socket.token == AUTH_TOKEN - - @skip_if_no_gateway - def test_socket_flow_instance(self): - """Test WebSocket flow instance creation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - socket = api.socket() - flow_instance = socket.flow(TEST_FLOW_ID) - - assert flow_instance is not None - assert flow_instance.flow_id == TEST_FLOW_ID - - @skip_if_no_gateway - def test_socket_methods_exist(self): - """Test that all expected WebSocket methods exist""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - socket = api.socket() - flow_instance = socket.flow(TEST_FLOW_ID) - - # Check all methods exist - expected_methods = [ - 'agent', 'text_completion', 'graph_rag', 'document_rag', - 'prompt', 'graph_embeddings_query', 'embeddings', - 'triples_query', 'objects_query', 'mcp_tool' - ] - - for method in expected_methods: - assert hasattr(flow_instance, method), f"Missing method: {method}" - - -class TestAsyncWebSocketAPI: - """Test async WebSocket API functionality""" - - @skip_if_no_gateway - def test_async_socket_client_creation(self): - """Test async WebSocket client instantiation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_socket = api.async_socket() - - assert async_socket is not None - assert async_socket.url.startswith("ws://") or async_socket.url.startswith("wss://") - - @skip_if_no_gateway - def test_async_socket_flow_instance(self): - """Test async WebSocket flow instance creation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_socket = api.async_socket() - flow_instance = async_socket.flow(TEST_FLOW_ID) - - assert flow_instance is not None - assert flow_instance.flow_id == TEST_FLOW_ID - - @skip_if_no_gateway - def test_async_socket_methods_exist(self): - """Test that all expected async WebSocket methods exist""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_socket = api.async_socket() - flow_instance = async_socket.flow(TEST_FLOW_ID) - - # Check all methods exist - expected_methods = [ - 'agent', 'text_completion', 'graph_rag', 'document_rag', - 'prompt', 'graph_embeddings_query', 'embeddings', - 'triples_query', 'objects_query', 'mcp_tool' - ] - - for method in expected_methods: - assert hasattr(flow_instance, method), f"Missing method: {method}" - - -class TestBulkOperations: - """Test bulk operations functionality""" - - @skip_if_no_gateway - def test_bulk_client_creation(self): - """Test bulk client instantiation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - bulk = api.bulk() - - assert bulk is not None - assert bulk.url.startswith("ws://") or bulk.url.startswith("wss://") - - @skip_if_no_gateway - def test_bulk_methods_exist(self): - """Test that all expected bulk methods exist""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - bulk = api.bulk() - - # Check all methods exist - expected_methods = [ - 'import_triples', 'export_triples', - 'import_graph_embeddings', 'export_graph_embeddings', - 'import_document_embeddings', 'export_document_embeddings', - 'import_entity_contexts', 'export_entity_contexts', - 'import_objects' - ] - - for method in expected_methods: - assert hasattr(bulk, method), f"Missing method: {method}" - - @skip_if_no_gateway - def test_async_bulk_client_creation(self): - """Test async bulk client instantiation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_bulk = api.async_bulk() - - assert async_bulk is not None - assert async_bulk.url.startswith("ws://") or async_bulk.url.startswith("wss://") - - -class TestMetrics: - """Test metrics functionality""" - - @skip_if_no_gateway - def test_metrics_client_creation(self): - """Test metrics client instantiation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - metrics = api.metrics() - - assert metrics is not None - assert hasattr(metrics, 'get') - - @skip_if_no_gateway - def test_async_metrics_client_creation(self): - """Test async metrics client instantiation""" - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_metrics = api.async_metrics() - - assert async_metrics is not None - assert hasattr(async_metrics, 'get') - - @skip_if_no_gateway - @pytest.mark.asyncio - async def test_async_metrics_get(self): - """Test getting metrics asynchronously""" - try: - api = Api(url=GATEWAY_URL, timeout=60, token=AUTH_TOKEN) - async_metrics = api.async_metrics() - - metrics_text = await async_metrics.get() - assert isinstance(metrics_text, str) - # Prometheus metrics should contain "# HELP" lines - assert "# HELP" in metrics_text or len(metrics_text) > 0 - except Exception as e: - # Metrics endpoint might not be available - pytest.skip(f"Metrics endpoint not available: {e}") - - -class TestStreamingTypes: - """Test streaming chunk type classes""" - - def test_agent_thought_creation(self): - """Test AgentThought chunk creation""" - chunk = AgentThought(content="thinking...", end_of_message=False) - assert chunk.content == "thinking..." - assert chunk.end_of_message is False - assert chunk.chunk_type == "thought" - - def test_agent_observation_creation(self): - """Test AgentObservation chunk creation""" - chunk = AgentObservation(content="observing...", end_of_message=False) - assert chunk.content == "observing..." - assert chunk.chunk_type == "observation" - - def test_agent_answer_creation(self): - """Test AgentAnswer chunk creation""" - chunk = AgentAnswer( - content="answer", - end_of_message=True, - end_of_dialog=True - ) - assert chunk.content == "answer" - assert chunk.end_of_message is True - assert chunk.end_of_dialog is True - assert chunk.chunk_type == "final-answer" - - def test_rag_chunk_creation(self): - """Test RAGChunk creation""" - chunk = RAGChunk( - content="response chunk", - end_of_stream=False, - error=None - ) - assert chunk.content == "response chunk" - assert chunk.end_of_stream is False - assert chunk.error is None - - -class TestTripleType: - """Test Triple type""" - - def test_triple_creation(self): - """Test Triple creation""" - triple = Triple(s="subject", p="predicate", o="object") - assert triple.s == "subject" - assert triple.p == "predicate" - assert triple.o == "object" - - -# Run tests with: pytest tests/test_api_integration.py -v -if __name__ == "__main__": - pytest.main([__file__, "-v"]) diff --git a/tests/unit/test_python_api_client.py b/tests/unit/test_python_api_client.py new file mode 100644 index 00000000..f86ae3da --- /dev/null +++ b/tests/unit/test_python_api_client.py @@ -0,0 +1,446 @@ +""" +Unit tests for TrustGraph Python API client library + +These tests use mocks and do not require a running server. +""" + +import pytest +from unittest.mock import Mock, patch, MagicMock, call +import json + +from trustgraph.api import ( + Api, + Triple, + AgentThought, + AgentObservation, + AgentAnswer, + RAGChunk, +) + + +class TestApiInstantiation: + """Test Api class instantiation and configuration""" + + def test_api_instantiation_defaults(self): + """Test Api with default parameters""" + api = Api() + assert api.url == "http://localhost:8088/api/v1/" + assert api.timeout == 60 + assert api.token is None + + def test_api_instantiation_with_url(self): + """Test Api with custom URL""" + api = Api(url="http://test-server:9000/") + assert api.url == "http://test-server:9000/api/v1/" + + def test_api_instantiation_with_url_trailing_slash(self): + """Test Api adds trailing slash if missing""" + api = Api(url="http://test-server:9000") + assert api.url == "http://test-server:9000/api/v1/" + + def test_api_instantiation_with_token(self): + """Test Api with authentication token""" + api = Api(token="test-token-123") + assert api.token == "test-token-123" + + def test_api_instantiation_with_timeout(self): + """Test Api with custom timeout""" + api = Api(timeout=120) + assert api.timeout == 120 + + +class TestApiLazyInitialization: + """Test lazy initialization of client components""" + + def test_socket_client_lazy_init(self): + """Test socket client is created on first access""" + api = Api(url="http://test/", token="token") + + assert api._socket_client is None + socket = api.socket() + assert api._socket_client is not None + assert socket is api._socket_client + + # Second access returns same instance + socket2 = api.socket() + assert socket2 is socket + + def test_bulk_client_lazy_init(self): + """Test bulk client is created on first access""" + api = Api(url="http://test/") + + assert api._bulk_client is None + bulk = api.bulk() + assert api._bulk_client is not None + + def test_async_flow_lazy_init(self): + """Test async flow is created on first access""" + api = Api(url="http://test/") + + assert api._async_flow is None + async_flow = api.async_flow() + assert api._async_flow is not None + + def test_metrics_lazy_init(self): + """Test metrics client is created on first access""" + api = Api(url="http://test/") + + assert api._metrics is None + metrics = api.metrics() + assert api._metrics is not None + + +class TestApiContextManager: + """Test context manager functionality""" + + def test_sync_context_manager(self): + """Test synchronous context manager""" + with Api(url="http://test/") as api: + assert api is not None + assert isinstance(api, Api) + # Should exit cleanly + + @pytest.mark.asyncio + async def test_async_context_manager(self): + """Test asynchronous context manager""" + async with Api(url="http://test/") as api: + assert api is not None + assert isinstance(api, Api) + # Should exit cleanly + + +class TestFlowClient: + """Test Flow client functionality""" + + @patch('requests.post') + def test_flow_list(self, mock_post): + """Test listing flows""" + mock_post.return_value.status_code = 200 + mock_post.return_value.json.return_value = {"flow-ids": ["flow1", "flow2"]} + + api = Api(url="http://test/") + flows = api.flow().list() + + assert flows == ["flow1", "flow2"] + assert mock_post.called + + @patch('requests.post') + def test_flow_list_with_token(self, mock_post): + """Test flow listing includes auth token""" + mock_post.return_value.status_code = 200 + mock_post.return_value.json.return_value = {"flow-ids": []} + + api = Api(url="http://test/", token="my-token") + api.flow().list() + + # Verify Authorization header was set + call_args = mock_post.call_args + headers = call_args[1]['headers'] if 'headers' in call_args[1] else {} + assert 'Authorization' in headers + assert headers['Authorization'] == 'Bearer my-token' + + @patch('requests.post') + def test_flow_get(self, mock_post): + """Test getting flow definition""" + flow_def = {"name": "test-flow", "description": "Test"} + mock_post.return_value.status_code = 200 + mock_post.return_value.json.return_value = {"flow": json.dumps(flow_def)} + + api = Api(url="http://test/") + result = api.flow().get("test-flow") + + assert result == flow_def + + def test_flow_instance_creation(self): + """Test creating flow instance""" + api = Api(url="http://test/") + flow_instance = api.flow().id("my-flow") + + assert flow_instance is not None + assert flow_instance.id == "my-flow" + + def test_flow_instance_has_methods(self): + """Test flow instance has expected methods""" + api = Api(url="http://test/") + flow_instance = api.flow().id("my-flow") + + expected_methods = [ + 'text_completion', 'agent', 'graph_rag', 'document_rag', + 'graph_embeddings_query', 'embeddings', 'prompt', + 'triples_query', 'objects_query' + ] + + for method in expected_methods: + assert hasattr(flow_instance, method), f"Missing method: {method}" + + +class TestSocketClient: + """Test WebSocket client functionality""" + + def test_socket_client_url_conversion_http(self): + """Test HTTP URL converted to WebSocket""" + api = Api(url="http://test-server:8088/") + socket = api.socket() + + assert socket.url.startswith("ws://") + assert "test-server" in socket.url + + def test_socket_client_url_conversion_https(self): + """Test HTTPS URL converted to secure WebSocket""" + api = Api(url="https://test-server:8088/") + socket = api.socket() + + assert socket.url.startswith("wss://") + + def test_socket_client_token_passed(self): + """Test token is passed to socket client""" + api = Api(url="http://test/", token="socket-token") + socket = api.socket() + + assert socket.token == "socket-token" + + def test_socket_flow_instance(self): + """Test creating socket flow instance""" + api = Api(url="http://test/") + socket = api.socket() + flow_instance = socket.flow("test-flow") + + assert flow_instance is not None + assert flow_instance.flow_id == "test-flow" + + def test_socket_flow_has_methods(self): + """Test socket flow instance has expected methods""" + api = Api(url="http://test/") + flow_instance = api.socket().flow("test-flow") + + expected_methods = [ + 'agent', 'text_completion', 'graph_rag', 'document_rag', + 'prompt', 'graph_embeddings_query', 'embeddings', + 'triples_query', 'objects_query', 'mcp_tool' + ] + + for method in expected_methods: + assert hasattr(flow_instance, method), f"Missing method: {method}" + + +class TestBulkClient: + """Test bulk operations client""" + + def test_bulk_client_url_conversion(self): + """Test bulk client uses WebSocket URL""" + api = Api(url="http://test/") + bulk = api.bulk() + + assert bulk.url.startswith("ws://") + + def test_bulk_client_has_import_methods(self): + """Test bulk client has import methods""" + api = Api(url="http://test/") + bulk = api.bulk() + + import_methods = [ + 'import_triples', + 'import_graph_embeddings', + 'import_document_embeddings', + 'import_entity_contexts', + 'import_objects' + ] + + for method in import_methods: + assert hasattr(bulk, method), f"Missing method: {method}" + + def test_bulk_client_has_export_methods(self): + """Test bulk client has export methods""" + api = Api(url="http://test/") + bulk = api.bulk() + + export_methods = [ + 'export_triples', + 'export_graph_embeddings', + 'export_document_embeddings', + 'export_entity_contexts' + ] + + for method in export_methods: + assert hasattr(bulk, method), f"Missing method: {method}" + + +class TestMetricsClient: + """Test metrics client""" + + @patch('requests.get') + def test_metrics_get(self, mock_get): + """Test getting metrics""" + mock_get.return_value.status_code = 200 + mock_get.return_value.text = "# HELP metric_name\nmetric_name 42" + + api = Api(url="http://test/") + metrics_text = api.metrics().get() + + assert "metric_name" in metrics_text + assert mock_get.called + + @patch('requests.get') + def test_metrics_with_token(self, mock_get): + """Test metrics request includes token""" + mock_get.return_value.status_code = 200 + mock_get.return_value.text = "metrics" + + api = Api(url="http://test/", token="metrics-token") + api.metrics().get() + + # Verify token in headers + call_args = mock_get.call_args + headers = call_args[1].get('headers', {}) + assert 'Authorization' in headers + + +class TestStreamingTypes: + """Test streaming chunk types""" + + def test_agent_thought_creation(self): + """Test creating AgentThought chunk""" + chunk = AgentThought(content="thinking...", end_of_message=False) + + assert chunk.content == "thinking..." + assert chunk.end_of_message is False + assert chunk.chunk_type == "thought" + + def test_agent_observation_creation(self): + """Test creating AgentObservation chunk""" + chunk = AgentObservation(content="observing...", end_of_message=False) + + assert chunk.content == "observing..." + assert chunk.chunk_type == "observation" + + def test_agent_answer_creation(self): + """Test creating AgentAnswer chunk""" + chunk = AgentAnswer( + content="answer", + end_of_message=True, + end_of_dialog=True + ) + + assert chunk.content == "answer" + assert chunk.end_of_message is True + assert chunk.end_of_dialog is True + assert chunk.chunk_type == "final-answer" + + def test_rag_chunk_creation(self): + """Test creating RAGChunk""" + chunk = RAGChunk( + content="response chunk", + end_of_stream=False, + error=None + ) + + assert chunk.content == "response chunk" + assert chunk.end_of_stream is False + assert chunk.error is None + + def test_rag_chunk_with_error(self): + """Test RAGChunk with error""" + error_dict = {"type": "error", "message": "failed"} + chunk = RAGChunk( + content="", + end_of_stream=True, + error=error_dict + ) + + assert chunk.error == error_dict + + +class TestTripleType: + """Test Triple data structure""" + + def test_triple_creation(self): + """Test creating Triple""" + triple = Triple(s="subject", p="predicate", o="object") + + assert triple.s == "subject" + assert triple.p == "predicate" + assert triple.o == "object" + + def test_triple_with_uris(self): + """Test Triple with URI values""" + triple = Triple( + s="http://example.org/entity1", + p="http://example.org/relation", + o="http://example.org/entity2" + ) + + assert triple.s.startswith("http://") + assert triple.p.startswith("http://") + assert triple.o.startswith("http://") + + +class TestAsyncClients: + """Test async client availability""" + + def test_async_flow_creation(self): + """Test creating async flow client""" + api = Api(url="http://test/") + async_flow = api.async_flow() + + assert async_flow is not None + + def test_async_socket_creation(self): + """Test creating async socket client""" + api = Api(url="http://test/") + async_socket = api.async_socket() + + assert async_socket is not None + assert async_socket.url.startswith("ws://") + + def test_async_bulk_creation(self): + """Test creating async bulk client""" + api = Api(url="http://test/") + async_bulk = api.async_bulk() + + assert async_bulk is not None + + def test_async_metrics_creation(self): + """Test creating async metrics client""" + api = Api(url="http://test/") + async_metrics = api.async_metrics() + + assert async_metrics is not None + + +class TestErrorHandling: + """Test error handling""" + + @patch('requests.post') + def test_protocol_exception_on_non_200(self, mock_post): + """Test ProtocolException raised on non-200 status""" + from trustgraph.api.exceptions import ProtocolException + + mock_post.return_value.status_code = 500 + + api = Api(url="http://test/") + + with pytest.raises(ProtocolException): + api.flow().list() + + @patch('requests.post') + def test_application_exception_on_error_response(self, mock_post): + """Test ApplicationException on error in response""" + from trustgraph.api.exceptions import ApplicationException + + mock_post.return_value.status_code = 200 + mock_post.return_value.json.return_value = { + "error": { + "type": "ValidationError", + "message": "Invalid input" + } + } + + api = Api(url="http://test/") + + with pytest.raises(ApplicationException): + api.flow().list() + + +# Run tests with: pytest tests/unit/test_python_api_client.py -v +if __name__ == "__main__": + pytest.main([__file__, "-v"])