Release/v1.2 (#457)

* Bump setup.py versions for 1.1

* PoC MCP server (#419)

* Very initial MCP server PoC for TrustGraph

* Put service on port 8000

* Add MCP container and packages to buildout

* Update docs for API/CLI changes in 1.0 (#421)

* Update some API basics for the 0.23/1.0 API change

* Add MCP container push (#425)

* Add command args to the MCP server (#426)

* Host and port parameters

* Added websocket arg

* More docs

* MCP client support (#427)

- MCP client service
- Tool request/response schema
- API gateway support for mcp-tool
- Message translation for tool request & response
- Make mcp-tool using configuration service for information
  about where the MCP services are.

* Feature/react call mcp (#428)

Key Features

  - MCP Tool Integration: Added core MCP tool support with ToolClientSpec and ToolClient classes
  - API Enhancement: New mcp_tool method for flow-specific tool invocation
  - CLI Tooling: New tg-invoke-mcp-tool command for testing MCP integration
  - React Agent Enhancement: Fixed and improved multi-tool invocation capabilities
  - Tool Management: Enhanced CLI for tool configuration and management

Changes

  - Added MCP tool invocation to API with flow-specific integration
  - Implemented ToolClientSpec and ToolClient for tool call handling
  - Updated agent-manager-react to invoke MCP tools with configurable types
  - Enhanced CLI with new commands and improved help text
  - Added comprehensive documentation for new CLI commands
  - Improved tool configuration management

Testing

  - Added tg-invoke-mcp-tool CLI command for isolated MCP integration testing
  - Enhanced agent capability to invoke multiple tools simultaneously

* Test suite executed from CI pipeline (#433)

* Test strategy & test cases

* Unit tests

* Integration tests

* Extending test coverage (#434)

* Contract tests

* Testing embeedings

* Agent unit tests

* Knowledge pipeline tests

* Turn on contract tests

* Increase storage test coverage (#435)

* Fixing storage and adding tests

* PR pipeline only runs quick tests

* Empty configuration is returned as empty list, previously was not in response (#436)

* Update config util to take files as well as command-line text (#437)

* Updated CLI invocation and config model for tools and mcp (#438)

* Updated CLI invocation and config model for tools and mcp

* CLI anomalies

* Tweaked the MCP tool implementation for new model

* Update agent implementation to match the new model

* Fix agent tools, now all tested

* Fixed integration tests

* Fix MCP delete tool params

* Update Python deps to 1.2

* Update to enable knowledge extraction using the agent framework (#439)

* Implement KG extraction agent (kg-extract-agent)

* Using ReAct framework (agent-manager-react)
 
* ReAct manager had an issue when emitting JSON, which conflicts which ReAct manager's own JSON messages, so refactored ReAct manager to use traditional ReAct messages, non-JSON structure.
 
* Minor refactor to take the prompt template client out of prompt-template so it can be more readily used by other modules. kg-extract-agent uses this framework.

* Migrate from setup.py to pyproject.toml (#440)

* Converted setup.py to pyproject.toml

* Modern package infrastructure as recommended by py docs

* Install missing build deps (#441)

* Install missing build deps (#442)

* Implement logging strategy (#444)

* Logging strategy and convert all prints() to logging invocations

* Fix/startup failure (#445)

* Fix loggin startup problems

* Fix logging startup problems (#446)

* Fix logging startup problems (#447)

* Fixed Mistral OCR to use current API (#448)

* Fixed Mistral OCR to use current API

* Added PDF decoder tests

* Fix Mistral OCR ident to be standard pdf-decoder (#450)

* Fix Mistral OCR ident to be standard pdf-decoder

* Correct test

* Schema structure refactor (#451)

* Write schema refactor spec

* Implemented schema refactor spec

* Structure data mvp (#452)

* Structured data tech spec

* Architecture principles

* New schemas

* Updated schemas and specs

* Object extractor

* Add .coveragerc

* New tests

* Cassandra object storage

* Trying to object extraction working, issues exist

* Validate librarian collection (#453)

* Fix token chunker, broken API invocation (#454)

* Fix token chunker, broken API invocation (#455)

* Knowledge load utility CLI (#456)

* Knowledge loader

* More tests
This commit is contained in:
cybermaggedon 2025-08-18 20:56:09 +01:00 committed by GitHub
parent c85ba197be
commit 89be656990
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
509 changed files with 49632 additions and 5159 deletions

269
tests/integration/README.md Normal file
View file

@ -0,0 +1,269 @@
# Integration Test Pattern for TrustGraph
This directory contains integration tests that verify the coordination between multiple TrustGraph services and components, following the patterns outlined in [TEST_STRATEGY.md](../../TEST_STRATEGY.md).
## Integration Test Approach
Integration tests focus on **service-to-service communication patterns** and **end-to-end message flows** while still using mocks for external infrastructure.
### Key Principles
1. **Test Service Coordination**: Verify that services work together correctly
2. **Mock External Dependencies**: Use mocks for databases, APIs, and infrastructure
3. **Real Business Logic**: Exercise actual service logic and data transformations
4. **Error Propagation**: Test how errors flow through the system
5. **Configuration Testing**: Verify services respond correctly to different configurations
## Test Structure
### Fixtures (conftest.py)
Common fixtures for integration tests:
- `mock_pulsar_client`: Mock Pulsar messaging client
- `mock_flow_context`: Mock flow context for service coordination
- `integration_config`: Standard configuration for integration tests
- `sample_documents`: Test document collections
- `sample_embeddings`: Test embedding vectors
- `sample_queries`: Test query sets
### Test Patterns
#### 1. End-to-End Flow Testing
```python
@pytest.mark.integration
@pytest.mark.asyncio
async def test_service_end_to_end_flow(self, service_instance, mock_clients):
"""Test complete service pipeline from input to output"""
# Arrange - Set up realistic test data
# Act - Execute the full service workflow
# Assert - Verify coordination between all components
```
#### 2. Error Propagation Testing
```python
@pytest.mark.integration
@pytest.mark.asyncio
async def test_service_error_handling(self, service_instance, mock_clients):
"""Test how errors propagate through service coordination"""
# Arrange - Set up failure scenarios
# Act - Execute service with failing dependency
# Assert - Verify proper error handling and cleanup
```
#### 3. Configuration Testing
```python
@pytest.mark.integration
@pytest.mark.asyncio
async def test_service_configuration_scenarios(self, service_instance):
"""Test service behavior with different configurations"""
# Test multiple configuration scenarios
# Verify service adapts correctly to each configuration
```
## Running Integration Tests
### Run All Integration Tests
```bash
pytest tests/integration/ -m integration
```
### Run Specific Test
```bash
pytest tests/integration/test_document_rag_integration.py::TestDocumentRagIntegration::test_document_rag_end_to_end_flow -v
```
### Run with Coverage (Skip Coverage Requirement)
```bash
pytest tests/integration/ -m integration --cov=trustgraph --cov-fail-under=0
```
### Run Slow Tests
```bash
pytest tests/integration/ -m "integration and slow"
```
### Skip Slow Tests
```bash
pytest tests/integration/ -m "integration and not slow"
```
## Examples: Integration Test Implementations
### 1. Document RAG Integration Test
The `test_document_rag_integration.py` demonstrates the integration test pattern:
### What It Tests
- **Service Coordination**: Embeddings → Document Retrieval → Prompt Generation
- **Error Handling**: Failure scenarios for each service dependency
- **Configuration**: Different document limits, users, and collections
- **Performance**: Large document set handling
### Key Features
- **Realistic Data Flow**: Uses actual service logic with mocked dependencies
- **Multiple Scenarios**: Success, failure, and edge cases
- **Verbose Logging**: Tests logging functionality
- **Multi-User Support**: Tests user and collection isolation
### Test Coverage
- ✅ End-to-end happy path
- ✅ No documents found scenario
- ✅ Service failure scenarios (embeddings, documents, prompt)
- ✅ Configuration variations
- ✅ Multi-user isolation
- ✅ Performance testing
- ✅ Verbose logging
### 2. Text Completion Integration Test
The `test_text_completion_integration.py` demonstrates external API integration testing:
### What It Tests
- **External API Integration**: OpenAI API connectivity and authentication
- **Rate Limiting**: Proper handling of API rate limits and retries
- **Error Handling**: API failures, connection timeouts, and error propagation
- **Token Tracking**: Accurate input/output token counting and metrics
- **Configuration**: Different model parameters and settings
- **Concurrency**: Multiple simultaneous API requests
### Key Features
- **Realistic Mock Responses**: Uses actual OpenAI API response structures
- **Authentication Testing**: API key validation and base URL configuration
- **Error Scenarios**: Rate limits, connection failures, invalid requests
- **Performance Metrics**: Timing and token usage validation
- **Model Flexibility**: Tests different GPT models and parameters
### Test Coverage
- ✅ Successful text completion generation
- ✅ Multiple model configurations (GPT-3.5, GPT-4, GPT-4-turbo)
- ✅ Rate limit handling (RateLimitError → TooManyRequests)
- ✅ API error handling and propagation
- ✅ Token counting accuracy
- ✅ Prompt construction and parameter validation
- ✅ Authentication patterns and API key validation
- ✅ Concurrent request processing
- ✅ Response content extraction and validation
- ✅ Performance timing measurements
### 3. Agent Manager Integration Test
The `test_agent_manager_integration.py` demonstrates complex service coordination testing:
### What It Tests
- **ReAct Pattern**: Think-Act-Observe cycles with multi-step reasoning
- **Tool Coordination**: Selection and execution of different tools (knowledge query, text completion, MCP tools)
- **Conversation State**: Management of conversation history and context
- **Multi-Service Integration**: Coordination between prompt, graph RAG, and tool services
- **Error Handling**: Tool failures, unknown tools, and error propagation
- **Configuration Management**: Dynamic tool loading and configuration
### Key Features
- **Complex Coordination**: Tests agent reasoning with multiple tool options
- **Stateful Processing**: Maintains conversation history across interactions
- **Dynamic Tool Selection**: Tests tool selection based on context and reasoning
- **Callback Pattern**: Tests think/observe callback mechanisms
- **JSON Serialization**: Handles complex data structures in prompts
- **Performance Testing**: Large conversation history handling
### Test Coverage
- ✅ Basic reasoning cycle with tool selection
- ✅ Final answer generation (ending ReAct cycle)
- ✅ Full ReAct cycle with tool execution
- ✅ Conversation history management
- ✅ Multiple tool coordination and selection
- ✅ Tool argument validation and processing
- ✅ Error handling (unknown tools, execution failures)
- ✅ Context integration and additional prompting
- ✅ Empty tool configuration handling
- ✅ Tool response processing and cleanup
- ✅ Performance with large conversation history
- ✅ JSON serialization in complex prompts
### 4. Knowledge Graph Extract → Store Pipeline Integration Test
The `test_kg_extract_store_integration.py` demonstrates multi-stage pipeline testing:
### What It Tests
- **Text-to-Graph Transformation**: Complete pipeline from text chunks to graph triples
- **Entity Extraction**: Definition extraction with proper URI generation
- **Relationship Extraction**: Subject-predicate-object relationship extraction
- **Graph Database Integration**: Storage coordination with Cassandra knowledge store
- **Data Validation**: Entity filtering, validation, and consistency checks
- **Pipeline Coordination**: Multi-stage processing with proper data flow
### Key Features
- **Multi-Stage Pipeline**: Tests definitions → relationships → storage coordination
- **Graph Data Structures**: RDF triples, entity contexts, and graph embeddings
- **URI Generation**: Consistent entity URI creation across pipeline stages
- **Data Transformation**: Complex text analysis to structured graph data
- **Batch Processing**: Large document set processing performance
- **Error Resilience**: Graceful handling of extraction failures
### Test Coverage
- ✅ Definitions extraction pipeline (text → entities + definitions)
- ✅ Relationships extraction pipeline (text → subject-predicate-object)
- ✅ URI generation consistency between processors
- ✅ Triple generation from definitions and relationships
- ✅ Knowledge store integration (triples and embeddings storage)
- ✅ End-to-end pipeline coordination
- ✅ Error handling in extraction services
- ✅ Empty and invalid extraction results handling
- ✅ Entity filtering and validation
- ✅ Large batch processing performance
- ✅ Metadata propagation through pipeline stages
## Best Practices
### Test Organization
- Group related tests in classes
- Use descriptive test names that explain the scenario
- Follow the Arrange-Act-Assert pattern
- Use appropriate pytest markers (`@pytest.mark.integration`, `@pytest.mark.slow`)
### Mock Strategy
- Mock external services (databases, APIs, message brokers)
- Use real service logic and data transformations
- Create realistic mock responses that match actual service behavior
- Reset mocks between tests to ensure isolation
### Test Data
- Use realistic test data that reflects actual usage patterns
- Create reusable fixtures for common test scenarios
- Test with various data sizes and edge cases
- Include both success and failure scenarios
### Error Testing
- Test each dependency failure scenario
- Verify proper error propagation and cleanup
- Test timeout and retry mechanisms
- Validate error response formats
### Performance Testing
- Mark performance tests with `@pytest.mark.slow`
- Test with realistic data volumes
- Set reasonable performance expectations
- Monitor resource usage during tests
## Adding New Integration Tests
1. **Identify Service Dependencies**: Map out which services your target service coordinates with
2. **Create Mock Fixtures**: Set up mocks for each dependency in conftest.py
3. **Design Test Scenarios**: Plan happy path, error cases, and edge conditions
4. **Implement Tests**: Follow the established patterns in this directory
5. **Add Documentation**: Update this README with your new test patterns
## Test Markers
- `@pytest.mark.integration`: Marks tests as integration tests
- `@pytest.mark.slow`: Marks tests that take longer to run
- `@pytest.mark.asyncio`: Required for async test functions
## Future Enhancements
- Add tests with real test containers for database integration
- Implement contract testing for service interfaces
- Add performance benchmarking for critical paths
- Create integration test templates for common service patterns

View file

View file

@ -0,0 +1,112 @@
"""
Helper for managing Cassandra containers in integration tests
Alternative to testcontainers for Fedora/Podman compatibility
"""
import subprocess
import time
import socket
from contextlib import contextmanager
from cassandra.cluster import Cluster
from cassandra.policies import RetryPolicy
class CassandraTestContainer:
"""Simple Cassandra container manager using Podman"""
def __init__(self, image="docker.io/library/cassandra:4.1", port=9042):
self.image = image
self.port = port
self.container_name = f"test-cassandra-{int(time.time())}"
self.container_id = None
def start(self):
"""Start Cassandra container"""
# Remove any existing container with same name
subprocess.run([
"podman", "rm", "-f", self.container_name
], capture_output=True)
# Start new container with faster startup options
result = subprocess.run([
"podman", "run", "-d",
"--name", self.container_name,
"-p", f"{self.port}:9042",
"-e", "JVM_OPTS=-Dcassandra.skip_wait_for_gossip_to_settle=0",
self.image
], capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Failed to start container: {result.stderr}")
self.container_id = result.stdout.strip()
# Wait for Cassandra to be ready
self._wait_for_ready()
return self
def stop(self):
"""Stop and remove container"""
import time
if self.container_name:
# Small delay before stopping to ensure connections are closed
time.sleep(0.5)
subprocess.run([
"podman", "rm", "-f", self.container_name
], capture_output=True)
def get_connection_host_port(self):
"""Get host and port for connection"""
return "localhost", self.port
def _wait_for_ready(self, timeout=120):
"""Wait for Cassandra to be ready for CQL queries"""
start_time = time.time()
print(f"Waiting for Cassandra to be ready on port {self.port}...")
while time.time() - start_time < timeout:
try:
# First check if port is open
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
result = sock.connect_ex(("localhost", self.port))
sock.close()
if result == 0:
# Port is open, now try to connect with Cassandra driver
try:
cluster = Cluster(['localhost'], port=self.port)
cluster.connect_timeout = 5
session = cluster.connect()
# Try a simple query to verify Cassandra is ready
session.execute("SELECT release_version FROM system.local")
session.shutdown()
cluster.shutdown()
print("Cassandra is ready!")
return
except Exception as e:
print(f"Cassandra not ready yet: {e}")
pass
except Exception as e:
print(f"Connection check failed: {e}")
pass
time.sleep(3)
raise RuntimeError(f"Cassandra not ready after {timeout} seconds")
@contextmanager
def cassandra_container(image="docker.io/library/cassandra:4.1", port=9042):
"""Context manager for Cassandra container"""
container = CassandraTestContainer(image, port)
try:
container.start()
yield container
finally:
container.stop()

View file

@ -0,0 +1,404 @@
"""
Shared fixtures and configuration for integration tests
This file provides common fixtures and test configuration for integration tests.
Following the TEST_STRATEGY.md patterns for integration testing.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock
@pytest.fixture
def mock_pulsar_client():
"""Mock Pulsar client for integration tests"""
client = MagicMock()
client.create_producer.return_value = AsyncMock()
client.subscribe.return_value = AsyncMock()
return client
@pytest.fixture
def mock_flow_context():
"""Mock flow context for testing service coordination"""
context = MagicMock()
# Mock flow producers/consumers
context.return_value.send = AsyncMock()
context.return_value.receive = AsyncMock()
return context
@pytest.fixture
def integration_config():
"""Common configuration for integration tests"""
return {
"pulsar_host": "localhost",
"pulsar_port": 6650,
"test_timeout": 30.0,
"max_retries": 3,
"doc_limit": 10,
"embedding_dim": 5,
}
@pytest.fixture
def sample_documents():
"""Sample document collection for testing"""
return [
{
"id": "doc1",
"content": "Machine learning is a subset of artificial intelligence that focuses on algorithms that learn from data.",
"collection": "ml_knowledge",
"user": "test_user"
},
{
"id": "doc2",
"content": "Deep learning uses neural networks with multiple layers to model complex patterns in data.",
"collection": "ml_knowledge",
"user": "test_user"
},
{
"id": "doc3",
"content": "Supervised learning algorithms learn from labeled training data to make predictions on new data.",
"collection": "ml_knowledge",
"user": "test_user"
}
]
@pytest.fixture
def sample_embeddings():
"""Sample embedding vectors for testing"""
return [
[0.1, 0.2, 0.3, 0.4, 0.5],
[0.6, 0.7, 0.8, 0.9, 1.0],
[0.2, 0.3, 0.4, 0.5, 0.6],
[0.7, 0.8, 0.9, 1.0, 0.1],
[0.3, 0.4, 0.5, 0.6, 0.7]
]
@pytest.fixture
def sample_queries():
"""Sample queries for testing"""
return [
"What is machine learning?",
"How does deep learning work?",
"Explain supervised learning",
"What are neural networks?",
"How do algorithms learn from data?"
]
@pytest.fixture
def sample_text_completion_requests():
"""Sample text completion requests for testing"""
return [
{
"system": "You are a helpful assistant.",
"prompt": "What is artificial intelligence?",
"expected_keywords": ["artificial intelligence", "AI", "machine learning"]
},
{
"system": "You are a technical expert.",
"prompt": "Explain neural networks",
"expected_keywords": ["neural networks", "neurons", "layers"]
},
{
"system": "You are a teacher.",
"prompt": "What is supervised learning?",
"expected_keywords": ["supervised learning", "training", "labels"]
}
]
@pytest.fixture
def mock_openai_response():
"""Mock OpenAI API response structure"""
return {
"id": "chatcmpl-test123",
"object": "chat.completion",
"created": 1234567890,
"model": "gpt-3.5-turbo",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "This is a test response from the AI model."
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 50,
"completion_tokens": 100,
"total_tokens": 150
}
}
@pytest.fixture
def text_completion_configs():
"""Various text completion configurations for testing"""
return [
{
"model": "gpt-3.5-turbo",
"temperature": 0.0,
"max_output": 1024,
"description": "Conservative settings"
},
{
"model": "gpt-4",
"temperature": 0.7,
"max_output": 2048,
"description": "Balanced settings"
},
{
"model": "gpt-4-turbo",
"temperature": 1.0,
"max_output": 4096,
"description": "Creative settings"
}
]
@pytest.fixture
def sample_agent_tools():
"""Sample agent tools configuration for testing"""
return {
"knowledge_query": {
"name": "knowledge_query",
"description": "Query the knowledge graph for information",
"type": "knowledge-query",
"arguments": [
{
"name": "question",
"type": "string",
"description": "The question to ask the knowledge graph"
}
]
},
"text_completion": {
"name": "text_completion",
"description": "Generate text completion using LLM",
"type": "text-completion",
"arguments": [
{
"name": "question",
"type": "string",
"description": "The question to ask the LLM"
}
]
},
"web_search": {
"name": "web_search",
"description": "Search the web for information",
"type": "mcp-tool",
"arguments": [
{
"name": "query",
"type": "string",
"description": "The search query"
}
]
}
}
@pytest.fixture
def sample_agent_requests():
"""Sample agent requests for testing"""
return [
{
"question": "What is machine learning?",
"plan": "",
"state": "",
"history": [],
"expected_tool": "knowledge_query"
},
{
"question": "Can you explain neural networks in simple terms?",
"plan": "",
"state": "",
"history": [],
"expected_tool": "text_completion"
},
{
"question": "Search for the latest AI research papers",
"plan": "",
"state": "",
"history": [],
"expected_tool": "web_search"
}
]
@pytest.fixture
def sample_agent_responses():
"""Sample agent responses for testing"""
return [
{
"thought": "I need to search for information about machine learning",
"action": "knowledge_query",
"arguments": {"question": "What is machine learning?"}
},
{
"thought": "I can provide a direct answer about neural networks",
"final-answer": "Neural networks are computing systems inspired by biological neural networks."
},
{
"thought": "I should search the web for recent research",
"action": "web_search",
"arguments": {"query": "latest AI research papers 2024"}
}
]
@pytest.fixture
def sample_conversation_history():
"""Sample conversation history for testing"""
return [
{
"thought": "I need to search for basic information first",
"action": "knowledge_query",
"arguments": {"question": "What is artificial intelligence?"},
"observation": "AI is the simulation of human intelligence in machines."
},
{
"thought": "Now I can provide more specific information",
"action": "text_completion",
"arguments": {"question": "Explain machine learning within AI"},
"observation": "Machine learning is a subset of AI that enables computers to learn from data."
}
]
@pytest.fixture
def sample_kg_extraction_data():
"""Sample knowledge graph extraction data for testing"""
return {
"text_chunks": [
"Machine Learning is a subset of Artificial Intelligence that enables computers to learn from data.",
"Neural Networks are computing systems inspired by biological neural networks.",
"Deep Learning uses neural networks with multiple layers to model complex patterns."
],
"expected_entities": [
"Machine Learning",
"Artificial Intelligence",
"Neural Networks",
"Deep Learning"
],
"expected_relationships": [
{
"subject": "Machine Learning",
"predicate": "is_subset_of",
"object": "Artificial Intelligence"
},
{
"subject": "Deep Learning",
"predicate": "uses",
"object": "Neural Networks"
}
]
}
@pytest.fixture
def sample_kg_definitions():
"""Sample knowledge graph definitions for testing"""
return [
{
"entity": "Machine Learning",
"definition": "A subset of artificial intelligence that enables computers to learn from data without explicit programming."
},
{
"entity": "Artificial Intelligence",
"definition": "The simulation of human intelligence in machines that are programmed to think and act like humans."
},
{
"entity": "Neural Networks",
"definition": "Computing systems inspired by biological neural networks that process information using interconnected nodes."
},
{
"entity": "Deep Learning",
"definition": "A subset of machine learning that uses neural networks with multiple layers to model complex patterns in data."
}
]
@pytest.fixture
def sample_kg_relationships():
"""Sample knowledge graph relationships for testing"""
return [
{
"subject": "Machine Learning",
"predicate": "is_subset_of",
"object": "Artificial Intelligence",
"object-entity": True
},
{
"subject": "Deep Learning",
"predicate": "is_subset_of",
"object": "Machine Learning",
"object-entity": True
},
{
"subject": "Neural Networks",
"predicate": "is_used_in",
"object": "Deep Learning",
"object-entity": True
},
{
"subject": "Machine Learning",
"predicate": "processes",
"object": "data patterns",
"object-entity": False
}
]
@pytest.fixture
def sample_kg_triples():
"""Sample knowledge graph triples for testing"""
return [
{
"subject": "http://trustgraph.ai/e/machine-learning",
"predicate": "http://www.w3.org/2000/01/rdf-schema#label",
"object": "Machine Learning"
},
{
"subject": "http://trustgraph.ai/e/machine-learning",
"predicate": "http://trustgraph.ai/definition",
"object": "A subset of artificial intelligence that enables computers to learn from data."
},
{
"subject": "http://trustgraph.ai/e/machine-learning",
"predicate": "http://trustgraph.ai/e/is_subset_of",
"object": "http://trustgraph.ai/e/artificial-intelligence"
}
]
# Test markers for integration tests
pytestmark = pytest.mark.integration
def pytest_sessionfinish(session, exitstatus):
"""
Called after whole test run finished, right before returning the exit status.
This hook is used to ensure Cassandra driver threads have time to shut down
properly before pytest exits, preventing "cannot schedule new futures after
shutdown" errors.
"""
import time
import gc
# Force garbage collection to clean up any remaining objects
gc.collect()
# Give Cassandra driver threads more time to clean up
time.sleep(2)

View file

@ -0,0 +1,481 @@
"""
Integration tests for Agent-based Knowledge Graph Extraction
These tests verify the end-to-end functionality of the agent-driven knowledge graph
extraction pipeline, testing the integration between agent communication, prompt
rendering, JSON response processing, and knowledge graph generation.
Following the TEST_STRATEGY.md approach for integration testing.
"""
import pytest
import json
from unittest.mock import AsyncMock, MagicMock, patch
from trustgraph.extract.kg.agent.extract import Processor as AgentKgExtractor
from trustgraph.schema import Chunk, Triple, Triples, Metadata, Value, Error
from trustgraph.schema import EntityContext, EntityContexts, AgentRequest, AgentResponse
from trustgraph.rdf import TRUSTGRAPH_ENTITIES, DEFINITION, RDF_LABEL, SUBJECT_OF
from trustgraph.template.prompt_manager import PromptManager
@pytest.mark.integration
class TestAgentKgExtractionIntegration:
"""Integration tests for Agent-based Knowledge Graph Extraction"""
@pytest.fixture
def mock_flow_context(self):
"""Mock flow context for agent communication and output publishing"""
context = MagicMock()
# Mock agent client
agent_client = AsyncMock()
# Mock successful agent response
def mock_agent_response(recipient, question):
# Simulate agent processing and return structured response
mock_response = MagicMock()
mock_response.error = None
mock_response.answer = '''```json
{
"definitions": [
{
"entity": "Machine Learning",
"definition": "A subset of artificial intelligence that enables computers to learn from data without explicit programming."
},
{
"entity": "Neural Networks",
"definition": "Computing systems inspired by biological neural networks that process information."
}
],
"relationships": [
{
"subject": "Machine Learning",
"predicate": "is_subset_of",
"object": "Artificial Intelligence",
"object-entity": true
},
{
"subject": "Neural Networks",
"predicate": "used_in",
"object": "Machine Learning",
"object-entity": true
}
]
}
```'''
return mock_response.answer
agent_client.invoke = mock_agent_response
# Mock output publishers
triples_publisher = AsyncMock()
entity_contexts_publisher = AsyncMock()
def context_router(service_name):
if service_name == "agent-request":
return agent_client
elif service_name == "triples":
return triples_publisher
elif service_name == "entity-contexts":
return entity_contexts_publisher
else:
return AsyncMock()
context.side_effect = context_router
return context
@pytest.fixture
def sample_chunk(self):
"""Sample text chunk for knowledge extraction"""
text = """
Machine Learning is a subset of Artificial Intelligence that enables computers
to learn from data without explicit programming. Neural Networks are computing
systems inspired by biological neural networks that process information.
Neural Networks are commonly used in Machine Learning applications.
"""
return Chunk(
chunk=text.encode('utf-8'),
metadata=Metadata(
id="doc123",
metadata=[
Triple(
s=Value(value="doc123", is_uri=True),
p=Value(value="http://example.org/type", is_uri=True),
o=Value(value="document", is_uri=False)
)
]
)
)
@pytest.fixture
def configured_agent_extractor(self):
"""Mock agent extractor with loaded configuration for integration testing"""
# Create a mock extractor that simulates the real behavior
from trustgraph.extract.kg.agent.extract import Processor
# Create mock without calling __init__ to avoid FlowProcessor issues
extractor = MagicMock()
real_extractor = Processor.__new__(Processor)
# Copy the methods we want to test
extractor.to_uri = real_extractor.to_uri
extractor.parse_json = real_extractor.parse_json
extractor.process_extraction_data = real_extractor.process_extraction_data
extractor.emit_triples = real_extractor.emit_triples
extractor.emit_entity_contexts = real_extractor.emit_entity_contexts
# Set up the configuration and manager
extractor.manager = PromptManager()
extractor.template_id = "agent-kg-extract"
extractor.config_key = "prompt"
# Mock configuration
config = {
"system": json.dumps("You are a knowledge extraction agent."),
"template-index": json.dumps(["agent-kg-extract"]),
"template.agent-kg-extract": json.dumps({
"prompt": "Extract entities and relationships from: {{ text }}",
"response-type": "json"
})
}
# Load configuration
extractor.manager.load_config(config)
# Mock the on_message method to simulate real behavior
async def mock_on_message(msg, consumer, flow):
v = msg.value()
chunk_text = v.chunk.decode('utf-8')
# Render prompt
prompt = extractor.manager.render(extractor.template_id, {"text": chunk_text})
# Get agent response (the mock returns a string directly)
agent_client = flow("agent-request")
agent_response = agent_client.invoke(recipient=lambda x: True, question=prompt)
# Parse and process
extraction_data = extractor.parse_json(agent_response)
triples, entity_contexts = extractor.process_extraction_data(extraction_data, v.metadata)
# Add metadata triples
for t in v.metadata.metadata:
triples.append(t)
# Emit outputs
if triples:
await extractor.emit_triples(flow("triples"), v.metadata, triples)
if entity_contexts:
await extractor.emit_entity_contexts(flow("entity-contexts"), v.metadata, entity_contexts)
extractor.on_message = mock_on_message
return extractor
@pytest.mark.asyncio
async def test_end_to_end_knowledge_extraction(self, configured_agent_extractor, sample_chunk, mock_flow_context):
"""Test complete end-to-end knowledge extraction workflow"""
# Arrange
mock_message = MagicMock()
mock_message.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
# Assert
# Verify agent was called with rendered prompt
agent_client = mock_flow_context("agent-request")
# Check that the mock function was replaced and called
assert hasattr(agent_client, 'invoke')
# Verify triples were emitted
triples_publisher = mock_flow_context("triples")
triples_publisher.send.assert_called_once()
sent_triples = triples_publisher.send.call_args[0][0]
assert isinstance(sent_triples, Triples)
assert sent_triples.metadata.id == "doc123"
assert len(sent_triples.triples) > 0
# Check that we have definition triples
definition_triples = [t for t in sent_triples.triples if t.p.value == DEFINITION]
assert len(definition_triples) >= 2 # Should have definitions for ML and Neural Networks
# Check that we have label triples
label_triples = [t for t in sent_triples.triples if t.p.value == RDF_LABEL]
assert len(label_triples) >= 2 # Should have labels for entities
# Check subject-of relationships
subject_of_triples = [t for t in sent_triples.triples if t.p.value == SUBJECT_OF]
assert len(subject_of_triples) >= 2 # Entities should be linked to document
# Verify entity contexts were emitted
entity_contexts_publisher = mock_flow_context("entity-contexts")
entity_contexts_publisher.send.assert_called_once()
sent_contexts = entity_contexts_publisher.send.call_args[0][0]
assert isinstance(sent_contexts, EntityContexts)
assert len(sent_contexts.entities) >= 2 # Should have contexts for both entities
# Verify entity URIs are properly formed
entity_uris = [ec.entity.value for ec in sent_contexts.entities]
assert f"{TRUSTGRAPH_ENTITIES}Machine%20Learning" in entity_uris
assert f"{TRUSTGRAPH_ENTITIES}Neural%20Networks" in entity_uris
@pytest.mark.asyncio
async def test_agent_error_handling(self, configured_agent_extractor, sample_chunk, mock_flow_context):
"""Test handling of agent errors"""
# Arrange - mock agent error response
agent_client = mock_flow_context("agent-request")
def mock_error_response(recipient, question):
# Simulate agent error by raising an exception
raise RuntimeError("Agent processing failed")
agent_client.invoke = mock_error_response
mock_message = MagicMock()
mock_message.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act & Assert
with pytest.raises(RuntimeError) as exc_info:
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
assert "Agent processing failed" in str(exc_info.value)
@pytest.mark.asyncio
async def test_invalid_json_response_handling(self, configured_agent_extractor, sample_chunk, mock_flow_context):
"""Test handling of invalid JSON responses from agent"""
# Arrange - mock invalid JSON response
agent_client = mock_flow_context("agent-request")
def mock_invalid_json_response(recipient, question):
return "This is not valid JSON at all"
agent_client.invoke = mock_invalid_json_response
mock_message = MagicMock()
mock_message.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act & Assert
with pytest.raises((ValueError, json.JSONDecodeError)):
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
@pytest.mark.asyncio
async def test_empty_extraction_results(self, configured_agent_extractor, sample_chunk, mock_flow_context):
"""Test handling of empty extraction results"""
# Arrange - mock empty extraction response
agent_client = mock_flow_context("agent-request")
def mock_empty_response(recipient, question):
return '{"definitions": [], "relationships": []}'
agent_client.invoke = mock_empty_response
mock_message = MagicMock()
mock_message.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
# Assert
# Should still emit outputs (even if empty) to maintain flow consistency
triples_publisher = mock_flow_context("triples")
entity_contexts_publisher = mock_flow_context("entity-contexts")
# Triples should include metadata triples at minimum
triples_publisher.send.assert_called_once()
sent_triples = triples_publisher.send.call_args[0][0]
assert isinstance(sent_triples, Triples)
# Entity contexts should not be sent if empty
entity_contexts_publisher.send.assert_not_called()
@pytest.mark.asyncio
async def test_malformed_extraction_data(self, configured_agent_extractor, sample_chunk, mock_flow_context):
"""Test handling of malformed extraction data"""
# Arrange - mock malformed extraction response
agent_client = mock_flow_context("agent-request")
def mock_malformed_response(recipient, question):
return '''{"definitions": [{"entity": "Missing Definition"}], "relationships": [{"subject": "Missing Object"}]}'''
agent_client.invoke = mock_malformed_response
mock_message = MagicMock()
mock_message.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act & Assert
with pytest.raises(KeyError):
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
@pytest.mark.asyncio
async def test_prompt_rendering_integration(self, configured_agent_extractor, mock_flow_context):
"""Test integration with prompt template rendering"""
# Create a chunk with specific text
test_text = "Test text for prompt rendering"
chunk = Chunk(
chunk=test_text.encode('utf-8'),
metadata=Metadata(id="test-doc", metadata=[])
)
agent_client = mock_flow_context("agent-request")
def capture_prompt(recipient, question):
# Verify the prompt contains the test text
assert test_text in question
return '{"definitions": [], "relationships": []}'
agent_client.invoke = capture_prompt
mock_message = MagicMock()
mock_message.value.return_value = chunk
mock_consumer = MagicMock()
# Act
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
# Assert - prompt should have been rendered with the text
# The agent_client.invoke is a function, not a mock, so we verify it was called by checking the flow worked
assert hasattr(agent_client, 'invoke')
@pytest.mark.asyncio
async def test_concurrent_processing_simulation(self, configured_agent_extractor, mock_flow_context):
"""Test simulation of concurrent chunk processing"""
# Create multiple chunks
chunks = []
for i in range(3):
text = f"Test document {i} content"
chunks.append(Chunk(
chunk=text.encode('utf-8'),
metadata=Metadata(id=f"doc{i}", metadata=[])
))
agent_client = mock_flow_context("agent-request")
responses = []
def mock_response(recipient, question):
response = f'{{"definitions": [{{"entity": "Entity {len(responses)}", "definition": "Definition {len(responses)}"}}], "relationships": []}}'
responses.append(response)
return response
agent_client.invoke = mock_response
# Process chunks sequentially (simulating concurrent processing)
for chunk in chunks:
mock_message = MagicMock()
mock_message.value.return_value = chunk
mock_consumer = MagicMock()
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
# Assert
assert len(responses) == 3
# Verify all chunks were processed
triples_publisher = mock_flow_context("triples")
assert triples_publisher.send.call_count == 3
@pytest.mark.asyncio
async def test_unicode_text_handling(self, configured_agent_extractor, mock_flow_context):
"""Test handling of text with unicode characters"""
# Create chunk with unicode text
unicode_text = "Machine Learning (学习机器) は人工知能の一分野です。"
chunk = Chunk(
chunk=unicode_text.encode('utf-8'),
metadata=Metadata(id="unicode-doc", metadata=[])
)
agent_client = mock_flow_context("agent-request")
def mock_unicode_response(recipient, question):
# Verify unicode text was properly decoded and included
assert "学习机器" in question
assert "人工知能" in question
return '''{"definitions": [{"entity": "機械学習", "definition": "人工知能の一分野"}], "relationships": []}'''
agent_client.invoke = mock_unicode_response
mock_message = MagicMock()
mock_message.value.return_value = chunk
mock_consumer = MagicMock()
# Act
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
# Assert - should handle unicode properly
triples_publisher = mock_flow_context("triples")
triples_publisher.send.assert_called_once()
sent_triples = triples_publisher.send.call_args[0][0]
# Check that unicode entity was properly processed
entity_labels = [t for t in sent_triples.triples if t.p.value == RDF_LABEL and t.o.value == "機械学習"]
assert len(entity_labels) > 0
@pytest.mark.asyncio
async def test_large_text_chunk_processing(self, configured_agent_extractor, mock_flow_context):
"""Test processing of large text chunks"""
# Create a large text chunk
large_text = "Machine Learning is important. " * 1000 # Repeat to create large text
chunk = Chunk(
chunk=large_text.encode('utf-8'),
metadata=Metadata(id="large-doc", metadata=[])
)
agent_client = mock_flow_context("agent-request")
def mock_large_text_response(recipient, question):
# Verify large text was included
assert len(question) > 10000
return '''{"definitions": [{"entity": "Machine Learning", "definition": "Important AI technique"}], "relationships": []}'''
agent_client.invoke = mock_large_text_response
mock_message = MagicMock()
mock_message.value.return_value = chunk
mock_consumer = MagicMock()
# Act
await configured_agent_extractor.on_message(mock_message, mock_consumer, mock_flow_context)
# Assert - should handle large text without issues
triples_publisher = mock_flow_context("triples")
triples_publisher.send.assert_called_once()
def test_configuration_parameter_validation(self):
"""Test parameter validation logic"""
# Test that default parameter logic would work
default_template_id = "agent-kg-extract"
default_config_type = "prompt"
default_concurrency = 1
# Simulate parameter handling
params = {}
template_id = params.get("template-id", default_template_id)
config_key = params.get("config-type", default_config_type)
concurrency = params.get("concurrency", default_concurrency)
assert template_id == "agent-kg-extract"
assert config_key == "prompt"
assert concurrency == 1
# Test with custom parameters
custom_params = {
"template-id": "custom-template",
"config-type": "custom-config",
"concurrency": 10
}
template_id = custom_params.get("template-id", default_template_id)
config_key = custom_params.get("config-type", default_config_type)
concurrency = custom_params.get("concurrency", default_concurrency)
assert template_id == "custom-template"
assert config_key == "custom-config"
assert concurrency == 10

View file

@ -0,0 +1,716 @@
"""
Integration tests for Agent Manager (ReAct Pattern) Service
These tests verify the end-to-end functionality of the Agent Manager service,
testing the ReAct pattern (Think-Act-Observe), tool coordination, multi-step reasoning,
and conversation state management.
Following the TEST_STRATEGY.md approach for integration testing.
"""
import pytest
import json
from unittest.mock import AsyncMock, MagicMock, patch
from trustgraph.agent.react.agent_manager import AgentManager
from trustgraph.agent.react.tools import KnowledgeQueryImpl, TextCompletionImpl, McpToolImpl
from trustgraph.agent.react.types import Action, Final, Tool, Argument
from trustgraph.schema import AgentRequest, AgentResponse, AgentStep, Error
@pytest.mark.integration
class TestAgentManagerIntegration:
"""Integration tests for Agent Manager ReAct pattern coordination"""
@pytest.fixture
def mock_flow_context(self):
"""Mock flow context for service coordination"""
context = MagicMock()
# Mock prompt client
prompt_client = AsyncMock()
prompt_client.agent_react.return_value = """Thought: I need to search for information about machine learning
Action: knowledge_query
Args: {
"question": "What is machine learning?"
}"""
# Mock graph RAG client
graph_rag_client = AsyncMock()
graph_rag_client.rag.return_value = "Machine learning is a subset of AI that enables computers to learn from data."
# Mock text completion client
text_completion_client = AsyncMock()
text_completion_client.question.return_value = "Machine learning involves algorithms that improve through experience."
# Mock MCP tool client
mcp_tool_client = AsyncMock()
mcp_tool_client.invoke.return_value = "Tool execution successful"
# Configure context to return appropriate clients
def context_router(service_name):
if service_name == "prompt-request":
return prompt_client
elif service_name == "graph-rag-request":
return graph_rag_client
elif service_name == "prompt-request":
return text_completion_client
elif service_name == "mcp-tool-request":
return mcp_tool_client
else:
return AsyncMock()
context.side_effect = context_router
return context
@pytest.fixture
def sample_tools(self):
"""Sample tool configuration for testing"""
return {
"knowledge_query": Tool(
name="knowledge_query",
description="Query the knowledge graph for information",
arguments=[
Argument(
name="question",
type="string",
description="The question to ask the knowledge graph"
)
],
implementation=KnowledgeQueryImpl,
config={}
),
"text_completion": Tool(
name="text_completion",
description="Generate text completion using LLM",
arguments=[
Argument(
name="question",
type="string",
description="The question to ask the LLM"
)
],
implementation=TextCompletionImpl,
config={}
),
"web_search": Tool(
name="web_search",
description="Search the web for information",
arguments=[
Argument(
name="query",
type="string",
description="The search query"
)
],
implementation=lambda context: AsyncMock(invoke=AsyncMock(return_value="Web search results")),
config={}
)
}
@pytest.fixture
def agent_manager(self, sample_tools):
"""Create agent manager with sample tools"""
return AgentManager(
tools=sample_tools,
additional_context="You are a helpful AI assistant with access to knowledge and tools."
)
@pytest.mark.asyncio
async def test_agent_manager_reasoning_cycle(self, agent_manager, mock_flow_context):
"""Test basic reasoning cycle with tool selection"""
# Arrange
question = "What is machine learning?"
history = []
# Act
action = await agent_manager.reason(question, history, mock_flow_context)
# Assert
assert isinstance(action, Action)
assert action.thought == "I need to search for information about machine learning"
assert action.name == "knowledge_query"
assert action.arguments == {"question": "What is machine learning?"}
assert action.observation == ""
# Verify prompt client was called correctly
prompt_client = mock_flow_context("prompt-request")
prompt_client.agent_react.assert_called_once()
# Verify the prompt variables passed to agent_react
call_args = prompt_client.agent_react.call_args
variables = call_args[0][0]
assert variables["question"] == question
assert len(variables["tools"]) == 3 # knowledge_query, text_completion, web_search
assert variables["context"] == "You are a helpful AI assistant with access to knowledge and tools."
@pytest.mark.asyncio
async def test_agent_manager_final_answer(self, agent_manager, mock_flow_context):
"""Test agent manager returning final answer"""
# Arrange
mock_flow_context("prompt-request").agent_react.return_value = """Thought: I have enough information to answer the question
Final Answer: Machine learning is a field of AI that enables computers to learn from data."""
question = "What is machine learning?"
history = []
# Act
action = await agent_manager.reason(question, history, mock_flow_context)
# Assert
assert isinstance(action, Final)
assert action.thought == "I have enough information to answer the question"
assert action.final == "Machine learning is a field of AI that enables computers to learn from data."
@pytest.mark.asyncio
async def test_agent_manager_react_with_tool_execution(self, agent_manager, mock_flow_context):
"""Test full ReAct cycle with tool execution"""
# Arrange
question = "What is machine learning?"
history = []
think_callback = AsyncMock()
observe_callback = AsyncMock()
# Act
action = await agent_manager.react(question, history, think_callback, observe_callback, mock_flow_context)
# Assert
assert isinstance(action, Action)
assert action.thought == "I need to search for information about machine learning"
assert action.name == "knowledge_query"
assert action.arguments == {"question": "What is machine learning?"}
assert action.observation == "Machine learning is a subset of AI that enables computers to learn from data."
# Verify callbacks were called
think_callback.assert_called_once_with("I need to search for information about machine learning")
observe_callback.assert_called_once_with("Machine learning is a subset of AI that enables computers to learn from data.")
# Verify tool was executed
graph_rag_client = mock_flow_context("graph-rag-request")
graph_rag_client.rag.assert_called_once_with("What is machine learning?")
@pytest.mark.asyncio
async def test_agent_manager_react_with_final_answer(self, agent_manager, mock_flow_context):
"""Test ReAct cycle ending with final answer"""
# Arrange
mock_flow_context("prompt-request").agent_react.return_value = """Thought: I can provide a direct answer
Final Answer: Machine learning is a branch of artificial intelligence."""
question = "What is machine learning?"
history = []
think_callback = AsyncMock()
observe_callback = AsyncMock()
# Act
action = await agent_manager.react(question, history, think_callback, observe_callback, mock_flow_context)
# Assert
assert isinstance(action, Final)
assert action.thought == "I can provide a direct answer"
assert action.final == "Machine learning is a branch of artificial intelligence."
# Verify only think callback was called (no observation for final answer)
think_callback.assert_called_once_with("I can provide a direct answer")
observe_callback.assert_not_called()
@pytest.mark.asyncio
async def test_agent_manager_with_conversation_history(self, agent_manager, mock_flow_context):
"""Test agent manager with conversation history"""
# Arrange
question = "Can you tell me more about neural networks?"
history = [
Action(
thought="I need to search for information about machine learning",
name="knowledge_query",
arguments={"question": "What is machine learning?"},
observation="Machine learning is a subset of AI that enables computers to learn from data."
)
]
# Act
action = await agent_manager.reason(question, history, mock_flow_context)
# Assert
assert isinstance(action, Action)
# Verify history was included in prompt variables
prompt_client = mock_flow_context("prompt-request")
call_args = prompt_client.agent_react.call_args
variables = call_args[0][0]
assert len(variables["history"]) == 1
assert variables["history"][0]["thought"] == "I need to search for information about machine learning"
assert variables["history"][0]["action"] == "knowledge_query"
assert variables["history"][0]["observation"] == "Machine learning is a subset of AI that enables computers to learn from data."
@pytest.mark.asyncio
async def test_agent_manager_tool_selection(self, agent_manager, mock_flow_context):
"""Test agent manager selecting different tools"""
# Test different tool selections
tool_scenarios = [
("knowledge_query", "graph-rag-request"),
("text_completion", "prompt-request"),
]
for tool_name, expected_service in tool_scenarios:
# Arrange
mock_flow_context("prompt-request").agent_react.return_value = f"""Thought: I need to use {tool_name}
Action: {tool_name}
Args: {{
"question": "test question"
}}"""
think_callback = AsyncMock()
observe_callback = AsyncMock()
# Act
action = await agent_manager.react("test question", [], think_callback, observe_callback, mock_flow_context)
# Assert
assert isinstance(action, Action)
assert action.name == tool_name
# Verify correct service was called
if tool_name == "knowledge_query":
mock_flow_context("graph-rag-request").rag.assert_called()
elif tool_name == "text_completion":
mock_flow_context("prompt-request").question.assert_called()
# Reset mocks for next iteration
for service in ["prompt-request", "graph-rag-request", "prompt-request"]:
mock_flow_context(service).reset_mock()
@pytest.mark.asyncio
async def test_agent_manager_unknown_tool_error(self, agent_manager, mock_flow_context):
"""Test agent manager error handling for unknown tool"""
# Arrange
mock_flow_context("prompt-request").agent_react.return_value = """Thought: I need to use an unknown tool
Action: unknown_tool
Args: {
"param": "value"
}"""
think_callback = AsyncMock()
observe_callback = AsyncMock()
# Act & Assert
with pytest.raises(RuntimeError) as exc_info:
await agent_manager.react("test question", [], think_callback, observe_callback, mock_flow_context)
assert "No action for unknown_tool!" in str(exc_info.value)
@pytest.mark.asyncio
async def test_agent_manager_tool_execution_error(self, agent_manager, mock_flow_context):
"""Test agent manager handling tool execution errors"""
# Arrange
mock_flow_context("graph-rag-request").rag.side_effect = Exception("Tool execution failed")
think_callback = AsyncMock()
observe_callback = AsyncMock()
# Act & Assert
with pytest.raises(Exception) as exc_info:
await agent_manager.react("test question", [], think_callback, observe_callback, mock_flow_context)
assert "Tool execution failed" in str(exc_info.value)
@pytest.mark.asyncio
async def test_agent_manager_multiple_tools_coordination(self, agent_manager, mock_flow_context):
"""Test agent manager coordination with multiple available tools"""
# Arrange
question = "Find information about AI and summarize it"
# Mock multi-step reasoning
mock_flow_context("prompt-request").agent_react.return_value = """Thought: I need to search for AI information first
Action: knowledge_query
Args: {
"question": "What is artificial intelligence?"
}"""
# Act
action = await agent_manager.reason(question, [], mock_flow_context)
# Assert
assert isinstance(action, Action)
assert action.name == "knowledge_query"
# Verify tool information was passed to prompt
prompt_client = mock_flow_context("prompt-request")
call_args = prompt_client.agent_react.call_args
variables = call_args[0][0]
# Should have all 3 tools available
tool_names = [tool["name"] for tool in variables["tools"]]
assert "knowledge_query" in tool_names
assert "text_completion" in tool_names
assert "web_search" in tool_names
@pytest.mark.asyncio
async def test_agent_manager_tool_argument_validation(self, agent_manager, mock_flow_context):
"""Test agent manager with various tool argument patterns"""
# Arrange
test_cases = [
{
"action": "knowledge_query",
"arguments": {"question": "What is deep learning?"},
"expected_service": "graph-rag-request"
},
{
"action": "text_completion",
"arguments": {"question": "Explain neural networks"},
"expected_service": "prompt-request"
},
{
"action": "web_search",
"arguments": {"query": "latest AI research"},
"expected_service": None # Custom mock
}
]
for test_case in test_cases:
# Arrange
# Format arguments as JSON
import json
args_json = json.dumps(test_case['arguments'], indent=4)
mock_flow_context("prompt-request").agent_react.return_value = f"""Thought: Using {test_case['action']}
Action: {test_case['action']}
Args: {args_json}"""
think_callback = AsyncMock()
observe_callback = AsyncMock()
# Act
action = await agent_manager.react("test", [], think_callback, observe_callback, mock_flow_context)
# Assert
assert isinstance(action, Action)
assert action.name == test_case['action']
assert action.arguments == test_case['arguments']
# Reset mocks
for service in ["prompt-request", "graph-rag-request", "prompt-request"]:
mock_flow_context(service).reset_mock()
@pytest.mark.asyncio
async def test_agent_manager_context_integration(self, agent_manager, mock_flow_context):
"""Test agent manager integration with additional context"""
# Arrange
agent_with_context = AgentManager(
tools={"knowledge_query": agent_manager.tools["knowledge_query"]},
additional_context="You are an expert in machine learning research."
)
question = "What are the latest developments in AI?"
# Act
action = await agent_with_context.reason(question, [], mock_flow_context)
# Assert
prompt_client = mock_flow_context("prompt-request")
call_args = prompt_client.agent_react.call_args
variables = call_args[0][0]
assert variables["context"] == "You are an expert in machine learning research."
assert variables["question"] == question
@pytest.mark.asyncio
async def test_agent_manager_empty_tools(self, mock_flow_context):
"""Test agent manager with no tools available"""
# Arrange
agent_no_tools = AgentManager(tools={}, additional_context="")
question = "What is machine learning?"
# Act
action = await agent_no_tools.reason(question, [], mock_flow_context)
# Assert
prompt_client = mock_flow_context("prompt-request")
call_args = prompt_client.agent_react.call_args
variables = call_args[0][0]
assert len(variables["tools"]) == 0
assert variables["tool_names"] == ""
@pytest.mark.asyncio
async def test_agent_manager_tool_response_processing(self, agent_manager, mock_flow_context):
"""Test agent manager processing different tool response types"""
# Arrange
response_scenarios = [
"Simple text response",
"Multi-line response\nwith several lines\nof information",
"Response with special characters: @#$%^&*()_+-=[]{}|;':\",./<>?",
" Response with whitespace ",
"" # Empty response
]
for expected_response in response_scenarios:
# Set up mock response
mock_flow_context("graph-rag-request").rag.return_value = expected_response
think_callback = AsyncMock()
observe_callback = AsyncMock()
# Act
action = await agent_manager.react("test question", [], think_callback, observe_callback, mock_flow_context)
# Assert
assert isinstance(action, Action)
assert action.observation == expected_response.strip()
observe_callback.assert_called_with(expected_response.strip())
# Reset mocks
mock_flow_context("graph-rag-request").reset_mock()
@pytest.mark.asyncio
async def test_agent_manager_malformed_response_handling(self, agent_manager, mock_flow_context):
"""Test agent manager handling of malformed text responses"""
# Test cases with expected error messages
test_cases = [
# Missing action/final answer
{
"response": "Thought: I need to do something",
"error_contains": "Response has thought but no action or final answer"
},
# Invalid JSON in Args
{
"response": """Thought: I need to search
Action: knowledge_query
Args: {invalid json}""",
"error_contains": "Invalid JSON in Args"
},
# Empty response
{
"response": "",
"error_contains": "Could not parse response"
},
# Only whitespace
{
"response": " \n\t ",
"error_contains": "Could not parse response"
},
# Missing Args for action (should create empty args dict)
{
"response": """Thought: I need to search
Action: knowledge_query""",
"error_contains": None # This should actually succeed with empty args
},
# Incomplete JSON
{
"response": """Thought: I need to search
Action: knowledge_query
Args: {
"question": "test"
""",
"error_contains": "Invalid JSON in Args"
},
]
for test_case in test_cases:
mock_flow_context("prompt-request").agent_react.return_value = test_case["response"]
if test_case["error_contains"]:
# Should raise an error
with pytest.raises(RuntimeError) as exc_info:
await agent_manager.reason("test question", [], mock_flow_context)
assert "Failed to parse agent response" in str(exc_info.value)
assert test_case["error_contains"] in str(exc_info.value)
else:
# Should succeed
action = await agent_manager.reason("test question", [], mock_flow_context)
assert isinstance(action, Action)
assert action.name == "knowledge_query"
assert action.arguments == {}
@pytest.mark.asyncio
async def test_agent_manager_text_parsing_edge_cases(self, agent_manager, mock_flow_context):
"""Test edge cases in text parsing"""
# Test response with markdown code blocks
mock_flow_context("prompt-request").agent_react.return_value = """```
Thought: I need to search for information
Action: knowledge_query
Args: {
"question": "What is AI?"
}
```"""
action = await agent_manager.reason("test", [], mock_flow_context)
assert isinstance(action, Action)
assert action.thought == "I need to search for information"
assert action.name == "knowledge_query"
# Test response with extra whitespace
mock_flow_context("prompt-request").agent_react.return_value = """
Thought: I need to think about this
Action: knowledge_query
Args: {
"question": "test"
}
"""
action = await agent_manager.reason("test", [], mock_flow_context)
assert isinstance(action, Action)
assert action.thought == "I need to think about this"
assert action.name == "knowledge_query"
@pytest.mark.asyncio
async def test_agent_manager_multiline_content(self, agent_manager, mock_flow_context):
"""Test handling of multi-line thoughts and final answers"""
# Multi-line thought
mock_flow_context("prompt-request").agent_react.return_value = """Thought: I need to consider multiple factors:
1. The user's question is complex
2. I should search for comprehensive information
3. This requires using the knowledge query tool
Action: knowledge_query
Args: {
"question": "complex query"
}"""
action = await agent_manager.reason("test", [], mock_flow_context)
assert isinstance(action, Action)
assert "multiple factors" in action.thought
assert "knowledge query tool" in action.thought
# Multi-line final answer
mock_flow_context("prompt-request").agent_react.return_value = """Thought: I have gathered enough information
Final Answer: Here is a comprehensive answer:
1. First point about the topic
2. Second point with details
3. Final conclusion
This covers all aspects of the question."""
action = await agent_manager.reason("test", [], mock_flow_context)
assert isinstance(action, Final)
assert "First point" in action.final
assert "Final conclusion" in action.final
assert "all aspects" in action.final
@pytest.mark.asyncio
async def test_agent_manager_json_args_special_characters(self, agent_manager, mock_flow_context):
"""Test JSON arguments with special characters and edge cases"""
# Test with special characters in JSON (properly escaped)
mock_flow_context("prompt-request").agent_react.return_value = """Thought: Processing special characters
Action: knowledge_query
Args: {
"question": "What about \\"quotes\\" and 'apostrophes'?",
"context": "Line 1\\nLine 2\\tTabbed",
"special": "Symbols: @#$%^&*()_+-=[]{}|;':,.<>?"
}"""
action = await agent_manager.reason("test", [], mock_flow_context)
assert isinstance(action, Action)
assert action.arguments["question"] == 'What about "quotes" and \'apostrophes\'?'
assert action.arguments["context"] == "Line 1\nLine 2\tTabbed"
assert "@#$%^&*" in action.arguments["special"]
# Test with nested JSON
mock_flow_context("prompt-request").agent_react.return_value = """Thought: Complex arguments
Action: web_search
Args: {
"query": "test",
"options": {
"limit": 10,
"filters": ["recent", "relevant"],
"metadata": {
"source": "user",
"timestamp": "2024-01-01"
}
}
}"""
action = await agent_manager.reason("test", [], mock_flow_context)
assert isinstance(action, Action)
assert action.arguments["options"]["limit"] == 10
assert "recent" in action.arguments["options"]["filters"]
assert action.arguments["options"]["metadata"]["source"] == "user"
@pytest.mark.asyncio
async def test_agent_manager_final_answer_json_format(self, agent_manager, mock_flow_context):
"""Test final answers that contain JSON-like content"""
# Final answer with JSON content
mock_flow_context("prompt-request").agent_react.return_value = """Thought: I can provide the data in JSON format
Final Answer: {
"result": "success",
"data": {
"name": "Machine Learning",
"type": "AI Technology",
"applications": ["NLP", "Computer Vision", "Robotics"]
},
"confidence": 0.95
}"""
action = await agent_manager.reason("test", [], mock_flow_context)
assert isinstance(action, Final)
# The final answer should preserve the JSON structure as a string
assert '"result": "success"' in action.final
assert '"applications":' in action.final
@pytest.mark.asyncio
@pytest.mark.slow
async def test_agent_manager_performance_with_large_history(self, agent_manager, mock_flow_context):
"""Test agent manager performance with large conversation history"""
# Arrange
large_history = [
Action(
thought=f"Step {i} thinking",
name="knowledge_query",
arguments={"question": f"Question {i}"},
observation=f"Observation {i}"
)
for i in range(50) # Large history
]
question = "Final question"
# Act
import time
start_time = time.time()
action = await agent_manager.reason(question, large_history, mock_flow_context)
end_time = time.time()
execution_time = end_time - start_time
# Assert
assert isinstance(action, Action)
assert execution_time < 5.0 # Should complete within reasonable time
# Verify history was processed correctly
prompt_client = mock_flow_context("prompt-request")
call_args = prompt_client.agent_react.call_args
variables = call_args[0][0]
assert len(variables["history"]) == 50
@pytest.mark.asyncio
async def test_agent_manager_json_serialization(self, agent_manager, mock_flow_context):
"""Test agent manager handling of JSON serialization in prompts"""
# Arrange
complex_history = [
Action(
thought="Complex thinking with special characters: \"quotes\", 'apostrophes', and symbols",
name="knowledge_query",
arguments={"question": "What about JSON serialization?", "complex": {"nested": "value"}},
observation="Response with JSON: {\"key\": \"value\"}"
)
]
question = "Handle JSON properly"
# Act
action = await agent_manager.reason(question, complex_history, mock_flow_context)
# Assert
assert isinstance(action, Action)
# Verify JSON was properly serialized in prompt
prompt_client = mock_flow_context("prompt-request")
call_args = prompt_client.agent_react.call_args
variables = call_args[0][0]
# Should not raise JSON serialization errors
json_str = json.dumps(variables, indent=4)
assert len(json_str) > 0

View file

@ -0,0 +1,411 @@
"""
Cassandra integration tests using Podman containers
These tests verify end-to-end functionality of Cassandra storage and query processors
with real database instances. Compatible with Fedora Linux and Podman.
Uses a single container for all tests to minimize startup time.
"""
import pytest
import asyncio
import time
from unittest.mock import MagicMock
from .cassandra_test_helper import cassandra_container
from trustgraph.direct.cassandra import TrustGraph
from trustgraph.storage.triples.cassandra.write import Processor as StorageProcessor
from trustgraph.query.triples.cassandra.service import Processor as QueryProcessor
from trustgraph.schema import Triple, Value, Metadata, Triples, TriplesQueryRequest
@pytest.mark.integration
@pytest.mark.slow
class TestCassandraIntegration:
"""Integration tests for Cassandra using a single shared container"""
@pytest.fixture(scope="class")
def cassandra_shared_container(self):
"""Class-level fixture: single Cassandra container for all tests"""
with cassandra_container() as container:
yield container
def setup_method(self):
"""Track all created clients for cleanup"""
self.clients_to_close = []
def teardown_method(self):
"""Clean up all Cassandra connections"""
import gc
for client in self.clients_to_close:
try:
client.close()
except Exception:
pass # Ignore errors during cleanup
# Clear the list and force garbage collection
self.clients_to_close.clear()
gc.collect()
# Small delay to let threads finish
time.sleep(0.5)
@pytest.mark.asyncio
async def test_complete_cassandra_integration(self, cassandra_shared_container):
"""Complete integration test covering all Cassandra functionality"""
container = cassandra_shared_container
host, port = container.get_connection_host_port()
print("=" * 60)
print("RUNNING COMPLETE CASSANDRA INTEGRATION TEST")
print("=" * 60)
# =====================================================
# Test 1: Basic TrustGraph Operations
# =====================================================
print("\n1. Testing basic TrustGraph operations...")
client = TrustGraph(
hosts=[host],
keyspace="test_basic",
table="test_table"
)
self.clients_to_close.append(client)
# Insert test data
client.insert("http://example.org/alice", "knows", "http://example.org/bob")
client.insert("http://example.org/alice", "age", "25")
client.insert("http://example.org/bob", "age", "30")
# Test get_all
all_results = list(client.get_all(limit=10))
assert len(all_results) == 3
print(f"✓ Stored and retrieved {len(all_results)} triples")
# Test get_s (subject query)
alice_results = list(client.get_s("http://example.org/alice", limit=10))
assert len(alice_results) == 2
alice_predicates = [r.p for r in alice_results]
assert "knows" in alice_predicates
assert "age" in alice_predicates
print("✓ Subject queries working")
# Test get_p (predicate query)
age_results = list(client.get_p("age", limit=10))
assert len(age_results) == 2
age_subjects = [r.s for r in age_results]
assert "http://example.org/alice" in age_subjects
assert "http://example.org/bob" in age_subjects
print("✓ Predicate queries working")
# =====================================================
# Test 2: Storage Processor Integration
# =====================================================
print("\n2. Testing storage processor integration...")
storage_processor = StorageProcessor(
taskgroup=MagicMock(),
hosts=[host],
keyspace="test_storage",
table="test_triples"
)
# Track the TrustGraph instance that will be created
self.storage_processor = storage_processor
# Create test message
storage_message = Triples(
metadata=Metadata(user="testuser", collection="testcol"),
triples=[
Triple(
s=Value(value="http://example.org/person1", is_uri=True),
p=Value(value="http://example.org/name", is_uri=True),
o=Value(value="Alice Smith", is_uri=False)
),
Triple(
s=Value(value="http://example.org/person1", is_uri=True),
p=Value(value="http://example.org/age", is_uri=True),
o=Value(value="25", is_uri=False)
),
Triple(
s=Value(value="http://example.org/person1", is_uri=True),
p=Value(value="http://example.org/department", is_uri=True),
o=Value(value="Engineering", is_uri=False)
)
]
)
# Store triples via processor
await storage_processor.store_triples(storage_message)
# Track the created TrustGraph instance
if hasattr(storage_processor, 'tg'):
self.clients_to_close.append(storage_processor.tg)
# Verify data was stored
storage_results = list(storage_processor.tg.get_s("http://example.org/person1", limit=10))
assert len(storage_results) == 3
predicates = [row.p for row in storage_results]
objects = [row.o for row in storage_results]
assert "http://example.org/name" in predicates
assert "http://example.org/age" in predicates
assert "http://example.org/department" in predicates
assert "Alice Smith" in objects
assert "25" in objects
assert "Engineering" in objects
print("✓ Storage processor working")
# =====================================================
# Test 3: Query Processor Integration
# =====================================================
print("\n3. Testing query processor integration...")
query_processor = QueryProcessor(
taskgroup=MagicMock(),
hosts=[host],
keyspace="test_query",
table="test_triples"
)
# Use same storage processor for the query keyspace
query_storage_processor = StorageProcessor(
taskgroup=MagicMock(),
hosts=[host],
keyspace="test_query",
table="test_triples"
)
# Store test data for querying
query_test_message = Triples(
metadata=Metadata(user="testuser", collection="testcol"),
triples=[
Triple(
s=Value(value="http://example.org/alice", is_uri=True),
p=Value(value="http://example.org/knows", is_uri=True),
o=Value(value="http://example.org/bob", is_uri=True)
),
Triple(
s=Value(value="http://example.org/alice", is_uri=True),
p=Value(value="http://example.org/age", is_uri=True),
o=Value(value="30", is_uri=False)
),
Triple(
s=Value(value="http://example.org/bob", is_uri=True),
p=Value(value="http://example.org/knows", is_uri=True),
o=Value(value="http://example.org/charlie", is_uri=True)
)
]
)
await query_storage_processor.store_triples(query_test_message)
# Debug: Check what was actually stored
print("Debug: Checking what was stored for Alice...")
direct_results = list(query_storage_processor.tg.get_s("http://example.org/alice", limit=10))
print(f"Direct TrustGraph results: {len(direct_results)}")
for result in direct_results:
print(f" S=http://example.org/alice, P={result.p}, O={result.o}")
# Test S query (find all relationships for Alice)
s_query = TriplesQueryRequest(
s=Value(value="http://example.org/alice", is_uri=True),
p=None, # None for wildcard
o=None, # None for wildcard
limit=10,
user="testuser",
collection="testcol"
)
s_results = await query_processor.query_triples(s_query)
print(f"Query processor results: {len(s_results)}")
for result in s_results:
print(f" S={result.s.value}, P={result.p.value}, O={result.o.value}")
assert len(s_results) == 2
s_predicates = [t.p.value for t in s_results]
assert "http://example.org/knows" in s_predicates
assert "http://example.org/age" in s_predicates
print("✓ Subject queries via processor working")
# Test P query (find all "knows" relationships)
p_query = TriplesQueryRequest(
s=None, # None for wildcard
p=Value(value="http://example.org/knows", is_uri=True),
o=None, # None for wildcard
limit=10,
user="testuser",
collection="testcol"
)
p_results = await query_processor.query_triples(p_query)
print(p_results)
assert len(p_results) == 2 # Alice knows Bob, Bob knows Charlie
p_subjects = [t.s.value for t in p_results]
assert "http://example.org/alice" in p_subjects
assert "http://example.org/bob" in p_subjects
print("✓ Predicate queries via processor working")
# =====================================================
# Test 4: Concurrent Operations
# =====================================================
print("\n4. Testing concurrent operations...")
concurrent_processor = StorageProcessor(
taskgroup=MagicMock(),
hosts=[host],
keyspace="test_concurrent",
table="test_triples"
)
# Create multiple coroutines for concurrent storage
async def store_person_data(person_id, name, age, department):
message = Triples(
metadata=Metadata(user="concurrent_test", collection="people"),
triples=[
Triple(
s=Value(value=f"http://example.org/{person_id}", is_uri=True),
p=Value(value="http://example.org/name", is_uri=True),
o=Value(value=name, is_uri=False)
),
Triple(
s=Value(value=f"http://example.org/{person_id}", is_uri=True),
p=Value(value="http://example.org/age", is_uri=True),
o=Value(value=str(age), is_uri=False)
),
Triple(
s=Value(value=f"http://example.org/{person_id}", is_uri=True),
p=Value(value="http://example.org/department", is_uri=True),
o=Value(value=department, is_uri=False)
)
]
)
await concurrent_processor.store_triples(message)
# Store data for multiple people concurrently
people_data = [
("person1", "John Doe", 25, "Engineering"),
("person2", "Jane Smith", 30, "Marketing"),
("person3", "Bob Wilson", 35, "Engineering"),
("person4", "Alice Brown", 28, "Sales"),
]
# Run storage operations concurrently
store_tasks = [store_person_data(pid, name, age, dept) for pid, name, age, dept in people_data]
await asyncio.gather(*store_tasks)
# Track the created TrustGraph instance
if hasattr(concurrent_processor, 'tg'):
self.clients_to_close.append(concurrent_processor.tg)
# Verify all names were stored
name_results = list(concurrent_processor.tg.get_p("http://example.org/name", limit=10))
assert len(name_results) == 4
stored_names = [r.o for r in name_results]
expected_names = ["John Doe", "Jane Smith", "Bob Wilson", "Alice Brown"]
for name in expected_names:
assert name in stored_names
# Verify department data
dept_results = list(concurrent_processor.tg.get_p("http://example.org/department", limit=10))
assert len(dept_results) == 4
stored_depts = [r.o for r in dept_results]
assert "Engineering" in stored_depts
assert "Marketing" in stored_depts
assert "Sales" in stored_depts
print("✓ Concurrent operations working")
# =====================================================
# Test 5: Complex Queries and Data Integrity
# =====================================================
print("\n5. Testing complex queries and data integrity...")
complex_processor = StorageProcessor(
taskgroup=MagicMock(),
hosts=[host],
keyspace="test_complex",
table="test_triples"
)
# Create a knowledge graph about a company
company_graph = Triples(
metadata=Metadata(user="integration_test", collection="company"),
triples=[
# People and their types
Triple(
s=Value(value="http://company.org/alice", is_uri=True),
p=Value(value="http://www.w3.org/1999/02/22-rdf-syntax-ns#type", is_uri=True),
o=Value(value="http://company.org/Employee", is_uri=True)
),
Triple(
s=Value(value="http://company.org/bob", is_uri=True),
p=Value(value="http://www.w3.org/1999/02/22-rdf-syntax-ns#type", is_uri=True),
o=Value(value="http://company.org/Employee", is_uri=True)
),
# Relationships
Triple(
s=Value(value="http://company.org/alice", is_uri=True),
p=Value(value="http://company.org/reportsTo", is_uri=True),
o=Value(value="http://company.org/bob", is_uri=True)
),
Triple(
s=Value(value="http://company.org/alice", is_uri=True),
p=Value(value="http://company.org/worksIn", is_uri=True),
o=Value(value="http://company.org/engineering", is_uri=True)
),
# Personal info
Triple(
s=Value(value="http://company.org/alice", is_uri=True),
p=Value(value="http://company.org/fullName", is_uri=True),
o=Value(value="Alice Johnson", is_uri=False)
),
Triple(
s=Value(value="http://company.org/alice", is_uri=True),
p=Value(value="http://company.org/email", is_uri=True),
o=Value(value="alice@company.org", is_uri=False)
),
]
)
# Store the company knowledge graph
await complex_processor.store_triples(company_graph)
# Track the created TrustGraph instance
if hasattr(complex_processor, 'tg'):
self.clients_to_close.append(complex_processor.tg)
# Verify all Alice's data
alice_data = list(complex_processor.tg.get_s("http://company.org/alice", limit=20))
assert len(alice_data) == 5
alice_predicates = [r.p for r in alice_data]
expected_predicates = [
"http://www.w3.org/1999/02/22-rdf-syntax-ns#type",
"http://company.org/reportsTo",
"http://company.org/worksIn",
"http://company.org/fullName",
"http://company.org/email"
]
for pred in expected_predicates:
assert pred in alice_predicates
# Test type-based queries
employee_results = list(complex_processor.tg.get_p("http://www.w3.org/1999/02/22-rdf-syntax-ns#type", limit=10))
print(employee_results)
assert len(employee_results) == 2
employees = [r.s for r in employee_results]
assert "http://company.org/alice" in employees
assert "http://company.org/bob" in employees
print("✓ Complex queries and data integrity working")
# =====================================================
# Summary
# =====================================================
print("\n" + "=" * 60)
print("✅ ALL CASSANDRA INTEGRATION TESTS PASSED!")
print("✅ Basic operations: PASSED")
print("✅ Storage processor: PASSED")
print("✅ Query processor: PASSED")
print("✅ Concurrent operations: PASSED")
print("✅ Complex queries: PASSED")
print("=" * 60)

View file

@ -0,0 +1,312 @@
"""
Integration tests for DocumentRAG retrieval system
These tests verify the end-to-end functionality of the DocumentRAG system,
testing the coordination between embeddings, document retrieval, and prompt services.
Following the TEST_STRATEGY.md approach for integration testing.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock
from trustgraph.retrieval.document_rag.document_rag import DocumentRag
@pytest.mark.integration
class TestDocumentRagIntegration:
"""Integration tests for DocumentRAG system coordination"""
@pytest.fixture
def mock_embeddings_client(self):
"""Mock embeddings client that returns realistic vector embeddings"""
client = AsyncMock()
client.embed.return_value = [
[0.1, 0.2, 0.3, 0.4, 0.5], # Realistic 5-dimensional embedding
[0.6, 0.7, 0.8, 0.9, 1.0] # Second embedding for testing
]
return client
@pytest.fixture
def mock_doc_embeddings_client(self):
"""Mock document embeddings client that returns realistic document chunks"""
client = AsyncMock()
client.query.return_value = [
"Machine learning is a subset of artificial intelligence that focuses on algorithms that learn from data.",
"Deep learning uses neural networks with multiple layers to model complex patterns in data.",
"Supervised learning algorithms learn from labeled training data to make predictions on new data."
]
return client
@pytest.fixture
def mock_prompt_client(self):
"""Mock prompt client that generates realistic responses"""
client = AsyncMock()
client.document_prompt.return_value = (
"Machine learning is a field of artificial intelligence that enables computers to learn "
"and improve from experience without being explicitly programmed. It uses algorithms "
"to find patterns in data and make predictions or decisions."
)
return client
@pytest.fixture
def document_rag(self, mock_embeddings_client, mock_doc_embeddings_client, mock_prompt_client):
"""Create DocumentRag instance with mocked dependencies"""
return DocumentRag(
embeddings_client=mock_embeddings_client,
doc_embeddings_client=mock_doc_embeddings_client,
prompt_client=mock_prompt_client,
verbose=True
)
@pytest.mark.asyncio
async def test_document_rag_end_to_end_flow(self, document_rag, mock_embeddings_client,
mock_doc_embeddings_client, mock_prompt_client):
"""Test complete DocumentRAG pipeline from query to response"""
# Arrange
query = "What is machine learning?"
user = "test_user"
collection = "ml_knowledge"
doc_limit = 10
# Act
result = await document_rag.query(
query=query,
user=user,
collection=collection,
doc_limit=doc_limit
)
# Assert - Verify service coordination
mock_embeddings_client.embed.assert_called_once_with(query)
mock_doc_embeddings_client.query.assert_called_once_with(
[[0.1, 0.2, 0.3, 0.4, 0.5], [0.6, 0.7, 0.8, 0.9, 1.0]],
limit=doc_limit,
user=user,
collection=collection
)
mock_prompt_client.document_prompt.assert_called_once_with(
query=query,
documents=[
"Machine learning is a subset of artificial intelligence that focuses on algorithms that learn from data.",
"Deep learning uses neural networks with multiple layers to model complex patterns in data.",
"Supervised learning algorithms learn from labeled training data to make predictions on new data."
]
)
# Verify final response
assert result is not None
assert isinstance(result, str)
assert "machine learning" in result.lower()
assert "artificial intelligence" in result.lower()
@pytest.mark.asyncio
async def test_document_rag_with_no_documents_found(self, mock_embeddings_client,
mock_doc_embeddings_client, mock_prompt_client):
"""Test DocumentRAG behavior when no documents are retrieved"""
# Arrange
mock_doc_embeddings_client.query.return_value = [] # No documents found
mock_prompt_client.document_prompt.return_value = "I couldn't find any relevant documents for your query."
document_rag = DocumentRag(
embeddings_client=mock_embeddings_client,
doc_embeddings_client=mock_doc_embeddings_client,
prompt_client=mock_prompt_client,
verbose=False
)
# Act
result = await document_rag.query("very obscure query")
# Assert
mock_embeddings_client.embed.assert_called_once()
mock_doc_embeddings_client.query.assert_called_once()
mock_prompt_client.document_prompt.assert_called_once_with(
query="very obscure query",
documents=[]
)
assert result == "I couldn't find any relevant documents for your query."
@pytest.mark.asyncio
async def test_document_rag_embeddings_service_failure(self, mock_embeddings_client,
mock_doc_embeddings_client, mock_prompt_client):
"""Test DocumentRAG error handling when embeddings service fails"""
# Arrange
mock_embeddings_client.embed.side_effect = Exception("Embeddings service unavailable")
document_rag = DocumentRag(
embeddings_client=mock_embeddings_client,
doc_embeddings_client=mock_doc_embeddings_client,
prompt_client=mock_prompt_client,
verbose=False
)
# Act & Assert
with pytest.raises(Exception) as exc_info:
await document_rag.query("test query")
assert "Embeddings service unavailable" in str(exc_info.value)
mock_embeddings_client.embed.assert_called_once()
mock_doc_embeddings_client.query.assert_not_called()
mock_prompt_client.document_prompt.assert_not_called()
@pytest.mark.asyncio
async def test_document_rag_document_service_failure(self, mock_embeddings_client,
mock_doc_embeddings_client, mock_prompt_client):
"""Test DocumentRAG error handling when document service fails"""
# Arrange
mock_doc_embeddings_client.query.side_effect = Exception("Document service connection failed")
document_rag = DocumentRag(
embeddings_client=mock_embeddings_client,
doc_embeddings_client=mock_doc_embeddings_client,
prompt_client=mock_prompt_client,
verbose=False
)
# Act & Assert
with pytest.raises(Exception) as exc_info:
await document_rag.query("test query")
assert "Document service connection failed" in str(exc_info.value)
mock_embeddings_client.embed.assert_called_once()
mock_doc_embeddings_client.query.assert_called_once()
mock_prompt_client.document_prompt.assert_not_called()
@pytest.mark.asyncio
async def test_document_rag_prompt_service_failure(self, mock_embeddings_client,
mock_doc_embeddings_client, mock_prompt_client):
"""Test DocumentRAG error handling when prompt service fails"""
# Arrange
mock_prompt_client.document_prompt.side_effect = Exception("LLM service rate limited")
document_rag = DocumentRag(
embeddings_client=mock_embeddings_client,
doc_embeddings_client=mock_doc_embeddings_client,
prompt_client=mock_prompt_client,
verbose=False
)
# Act & Assert
with pytest.raises(Exception) as exc_info:
await document_rag.query("test query")
assert "LLM service rate limited" in str(exc_info.value)
mock_embeddings_client.embed.assert_called_once()
mock_doc_embeddings_client.query.assert_called_once()
mock_prompt_client.document_prompt.assert_called_once()
@pytest.mark.asyncio
async def test_document_rag_with_different_document_limits(self, document_rag,
mock_doc_embeddings_client):
"""Test DocumentRAG with various document limit configurations"""
# Test different document limits
test_cases = [1, 5, 10, 25, 50]
for limit in test_cases:
# Reset mock call history
mock_doc_embeddings_client.reset_mock()
# Act
await document_rag.query(f"query with limit {limit}", doc_limit=limit)
# Assert
mock_doc_embeddings_client.query.assert_called_once()
call_args = mock_doc_embeddings_client.query.call_args
assert call_args.kwargs['limit'] == limit
@pytest.mark.asyncio
async def test_document_rag_multi_user_isolation(self, document_rag, mock_doc_embeddings_client):
"""Test DocumentRAG properly isolates queries by user and collection"""
# Arrange
test_scenarios = [
("user1", "collection1"),
("user2", "collection2"),
("user1", "collection2"), # Same user, different collection
("user2", "collection1"), # Different user, same collection
]
for user, collection in test_scenarios:
# Reset mock call history
mock_doc_embeddings_client.reset_mock()
# Act
await document_rag.query(
f"query from {user} in {collection}",
user=user,
collection=collection
)
# Assert
mock_doc_embeddings_client.query.assert_called_once()
call_args = mock_doc_embeddings_client.query.call_args
assert call_args.kwargs['user'] == user
assert call_args.kwargs['collection'] == collection
@pytest.mark.asyncio
async def test_document_rag_verbose_logging(self, mock_embeddings_client,
mock_doc_embeddings_client, mock_prompt_client,
caplog):
"""Test DocumentRAG verbose logging functionality"""
import logging
# Arrange - Configure logging to capture debug messages
caplog.set_level(logging.DEBUG)
document_rag = DocumentRag(
embeddings_client=mock_embeddings_client,
doc_embeddings_client=mock_doc_embeddings_client,
prompt_client=mock_prompt_client,
verbose=True
)
# Act
await document_rag.query("test query for verbose logging")
# Assert - Check for new logging messages
log_messages = caplog.text
assert "DocumentRag initialized" in log_messages
assert "Constructing prompt..." in log_messages
assert "Computing embeddings..." in log_messages
assert "Getting documents..." in log_messages
assert "Invoking LLM..." in log_messages
assert "Query processing complete" in log_messages
@pytest.mark.asyncio
@pytest.mark.slow
async def test_document_rag_performance_with_large_document_set(self, document_rag,
mock_doc_embeddings_client):
"""Test DocumentRAG performance with large document retrieval"""
# Arrange - Mock large document set (100 documents)
large_doc_set = [f"Document {i} content about machine learning and AI" for i in range(100)]
mock_doc_embeddings_client.query.return_value = large_doc_set
# Act
import time
start_time = time.time()
result = await document_rag.query("performance test query", doc_limit=100)
end_time = time.time()
execution_time = end_time - start_time
# Assert
assert result is not None
assert execution_time < 5.0 # Should complete within 5 seconds
mock_doc_embeddings_client.query.assert_called_once()
call_args = mock_doc_embeddings_client.query.call_args
assert call_args.kwargs['limit'] == 100
@pytest.mark.asyncio
async def test_document_rag_default_parameters(self, document_rag, mock_doc_embeddings_client):
"""Test DocumentRAG uses correct default parameters"""
# Act
await document_rag.query("test query with defaults")
# Assert
mock_doc_embeddings_client.query.assert_called_once()
call_args = mock_doc_embeddings_client.query.call_args
assert call_args.kwargs['user'] == "trustgraph"
assert call_args.kwargs['collection'] == "default"
assert call_args.kwargs['limit'] == 20

View file

@ -0,0 +1,642 @@
"""
Integration tests for Knowledge Graph Extract Store Pipeline
These tests verify the end-to-end functionality of the knowledge graph extraction
and storage pipeline, testing text-to-graph transformation, entity extraction,
relationship extraction, and graph database storage.
Following the TEST_STRATEGY.md approach for integration testing.
"""
import pytest
import json
import urllib.parse
from unittest.mock import AsyncMock, MagicMock, patch
from trustgraph.extract.kg.definitions.extract import Processor as DefinitionsProcessor
from trustgraph.extract.kg.relationships.extract import Processor as RelationshipsProcessor
from trustgraph.storage.knowledge.store import Processor as KnowledgeStoreProcessor
from trustgraph.schema import Chunk, Triple, Triples, Metadata, Value, Error
from trustgraph.schema import EntityContext, EntityContexts, GraphEmbeddings
from trustgraph.rdf import TRUSTGRAPH_ENTITIES, DEFINITION, RDF_LABEL, SUBJECT_OF
@pytest.mark.integration
class TestKnowledgeGraphPipelineIntegration:
"""Integration tests for Knowledge Graph Extract → Store Pipeline"""
@pytest.fixture
def mock_flow_context(self):
"""Mock flow context for service coordination"""
context = MagicMock()
# Mock prompt client for definitions extraction
prompt_client = AsyncMock()
prompt_client.extract_definitions.return_value = [
{
"entity": "Machine Learning",
"definition": "A subset of artificial intelligence that enables computers to learn from data without explicit programming."
},
{
"entity": "Neural Networks",
"definition": "Computing systems inspired by biological neural networks that process information."
}
]
# Mock prompt client for relationships extraction
prompt_client.extract_relationships.return_value = [
{
"subject": "Machine Learning",
"predicate": "is_subset_of",
"object": "Artificial Intelligence",
"object-entity": True
},
{
"subject": "Neural Networks",
"predicate": "is_used_in",
"object": "Machine Learning",
"object-entity": True
}
]
# Mock producers for output streams
triples_producer = AsyncMock()
entity_contexts_producer = AsyncMock()
# Configure context routing
def context_router(service_name):
if service_name == "prompt-request":
return prompt_client
elif service_name == "triples":
return triples_producer
elif service_name == "entity-contexts":
return entity_contexts_producer
else:
return AsyncMock()
context.side_effect = context_router
return context
@pytest.fixture
def mock_cassandra_store(self):
"""Mock Cassandra knowledge table store"""
store = AsyncMock()
store.add_triples.return_value = None
store.add_graph_embeddings.return_value = None
return store
@pytest.fixture
def sample_chunk(self):
"""Sample text chunk for processing"""
return Chunk(
metadata=Metadata(
id="doc-123",
user="test_user",
collection="test_collection",
metadata=[]
),
chunk=b"Machine Learning is a subset of Artificial Intelligence. Neural Networks are used in Machine Learning to process complex patterns."
)
@pytest.fixture
def sample_definitions_response(self):
"""Sample definitions extraction response"""
return [
{
"entity": "Machine Learning",
"definition": "A subset of artificial intelligence that enables computers to learn from data."
},
{
"entity": "Artificial Intelligence",
"definition": "The simulation of human intelligence in machines."
},
{
"entity": "Neural Networks",
"definition": "Computing systems inspired by biological neural networks."
}
]
@pytest.fixture
def sample_relationships_response(self):
"""Sample relationships extraction response"""
return [
{
"subject": "Machine Learning",
"predicate": "is_subset_of",
"object": "Artificial Intelligence",
"object-entity": True
},
{
"subject": "Neural Networks",
"predicate": "is_used_in",
"object": "Machine Learning",
"object-entity": True
},
{
"subject": "Machine Learning",
"predicate": "processes",
"object": "data patterns",
"object-entity": False
}
]
@pytest.fixture
def definitions_processor(self):
"""Create definitions processor with minimal configuration"""
processor = MagicMock()
processor.to_uri = DefinitionsProcessor.to_uri.__get__(processor, DefinitionsProcessor)
processor.emit_triples = DefinitionsProcessor.emit_triples.__get__(processor, DefinitionsProcessor)
processor.emit_ecs = DefinitionsProcessor.emit_ecs.__get__(processor, DefinitionsProcessor)
processor.on_message = DefinitionsProcessor.on_message.__get__(processor, DefinitionsProcessor)
return processor
@pytest.fixture
def relationships_processor(self):
"""Create relationships processor with minimal configuration"""
processor = MagicMock()
processor.to_uri = RelationshipsProcessor.to_uri.__get__(processor, RelationshipsProcessor)
processor.emit_triples = RelationshipsProcessor.emit_triples.__get__(processor, RelationshipsProcessor)
processor.on_message = RelationshipsProcessor.on_message.__get__(processor, RelationshipsProcessor)
return processor
@pytest.mark.asyncio
async def test_definitions_extraction_pipeline(self, definitions_processor, mock_flow_context, sample_chunk):
"""Test definitions extraction from text chunk to graph triples"""
# Arrange
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Assert
# Verify prompt client was called for definitions extraction
prompt_client = mock_flow_context("prompt-request")
prompt_client.extract_definitions.assert_called_once()
call_args = prompt_client.extract_definitions.call_args
assert "Machine Learning" in call_args.kwargs['text']
assert "Neural Networks" in call_args.kwargs['text']
# Verify triples producer was called
triples_producer = mock_flow_context("triples")
triples_producer.send.assert_called_once()
# Verify entity contexts producer was called
entity_contexts_producer = mock_flow_context("entity-contexts")
entity_contexts_producer.send.assert_called_once()
@pytest.mark.asyncio
async def test_relationships_extraction_pipeline(self, relationships_processor, mock_flow_context, sample_chunk):
"""Test relationships extraction from text chunk to graph triples"""
# Arrange
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act
await relationships_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Assert
# Verify prompt client was called for relationships extraction
prompt_client = mock_flow_context("prompt-request")
prompt_client.extract_relationships.assert_called_once()
call_args = prompt_client.extract_relationships.call_args
assert "Machine Learning" in call_args.kwargs['text']
# Verify triples producer was called
triples_producer = mock_flow_context("triples")
triples_producer.send.assert_called_once()
@pytest.mark.asyncio
async def test_uri_generation_consistency(self, definitions_processor, relationships_processor):
"""Test URI generation consistency between processors"""
# Arrange
test_entities = [
"Machine Learning",
"Artificial Intelligence",
"Neural Networks",
"Deep Learning",
"Natural Language Processing"
]
# Act & Assert
for entity in test_entities:
def_uri = definitions_processor.to_uri(entity)
rel_uri = relationships_processor.to_uri(entity)
# URIs should be identical between processors
assert def_uri == rel_uri
# URI should be properly encoded
assert def_uri.startswith(TRUSTGRAPH_ENTITIES)
assert " " not in def_uri
assert def_uri.endswith(urllib.parse.quote(entity.replace(" ", "-").lower().encode("utf-8")))
@pytest.mark.asyncio
async def test_definitions_triple_generation(self, definitions_processor, sample_definitions_response):
"""Test triple generation from definitions extraction"""
# Arrange
metadata = Metadata(
id="test-doc",
user="test_user",
collection="test_collection",
metadata=[]
)
# Act
triples = []
entities = []
for defn in sample_definitions_response:
s = defn["entity"]
o = defn["definition"]
if s and o:
s_uri = definitions_processor.to_uri(s)
s_value = Value(value=str(s_uri), is_uri=True)
o_value = Value(value=str(o), is_uri=False)
# Generate triples as the processor would
triples.append(Triple(
s=s_value,
p=Value(value=RDF_LABEL, is_uri=True),
o=Value(value=s, is_uri=False)
))
triples.append(Triple(
s=s_value,
p=Value(value=DEFINITION, is_uri=True),
o=o_value
))
entities.append(EntityContext(
entity=s_value,
context=defn["definition"]
))
# Assert
assert len(triples) == 6 # 2 triples per entity * 3 entities
assert len(entities) == 3 # 1 entity context per entity
# Verify triple structure
label_triples = [t for t in triples if t.p.value == RDF_LABEL]
definition_triples = [t for t in triples if t.p.value == DEFINITION]
assert len(label_triples) == 3
assert len(definition_triples) == 3
# Verify entity contexts
for entity in entities:
assert entity.entity.is_uri is True
assert entity.entity.value.startswith(TRUSTGRAPH_ENTITIES)
assert len(entity.context) > 0
@pytest.mark.asyncio
async def test_relationships_triple_generation(self, relationships_processor, sample_relationships_response):
"""Test triple generation from relationships extraction"""
# Arrange
metadata = Metadata(
id="test-doc",
user="test_user",
collection="test_collection",
metadata=[]
)
# Act
triples = []
for rel in sample_relationships_response:
s = rel["subject"]
p = rel["predicate"]
o = rel["object"]
if s and p and o:
s_uri = relationships_processor.to_uri(s)
s_value = Value(value=str(s_uri), is_uri=True)
p_uri = relationships_processor.to_uri(p)
p_value = Value(value=str(p_uri), is_uri=True)
if rel["object-entity"]:
o_uri = relationships_processor.to_uri(o)
o_value = Value(value=str(o_uri), is_uri=True)
else:
o_value = Value(value=str(o), is_uri=False)
# Main relationship triple
triples.append(Triple(s=s_value, p=p_value, o=o_value))
# Label triples
triples.append(Triple(
s=s_value,
p=Value(value=RDF_LABEL, is_uri=True),
o=Value(value=str(s), is_uri=False)
))
triples.append(Triple(
s=p_value,
p=Value(value=RDF_LABEL, is_uri=True),
o=Value(value=str(p), is_uri=False)
))
if rel["object-entity"]:
triples.append(Triple(
s=o_value,
p=Value(value=RDF_LABEL, is_uri=True),
o=Value(value=str(o), is_uri=False)
))
# Assert
assert len(triples) > 0
# Verify relationship triples exist
relationship_triples = [t for t in triples if t.p.value.endswith("is_subset_of") or t.p.value.endswith("is_used_in")]
assert len(relationship_triples) >= 2
# Verify label triples
label_triples = [t for t in triples if t.p.value == RDF_LABEL]
assert len(label_triples) > 0
@pytest.mark.asyncio
async def test_knowledge_store_triples_storage(self, mock_cassandra_store):
"""Test knowledge store triples storage integration"""
# Arrange
processor = MagicMock()
processor.table_store = mock_cassandra_store
processor.on_triples = KnowledgeStoreProcessor.on_triples.__get__(processor, KnowledgeStoreProcessor)
sample_triples = Triples(
metadata=Metadata(
id="test-doc",
user="test_user",
collection="test_collection",
metadata=[]
),
triples=[
Triple(
s=Value(value="http://trustgraph.ai/e/machine-learning", is_uri=True),
p=Value(value=DEFINITION, is_uri=True),
o=Value(value="A subset of AI", is_uri=False)
)
]
)
mock_msg = MagicMock()
mock_msg.value.return_value = sample_triples
# Act
await processor.on_triples(mock_msg, None, None)
# Assert
mock_cassandra_store.add_triples.assert_called_once_with(sample_triples)
@pytest.mark.asyncio
async def test_knowledge_store_graph_embeddings_storage(self, mock_cassandra_store):
"""Test knowledge store graph embeddings storage integration"""
# Arrange
processor = MagicMock()
processor.table_store = mock_cassandra_store
processor.on_graph_embeddings = KnowledgeStoreProcessor.on_graph_embeddings.__get__(processor, KnowledgeStoreProcessor)
sample_embeddings = GraphEmbeddings(
metadata=Metadata(
id="test-doc",
user="test_user",
collection="test_collection",
metadata=[]
),
entities=[]
)
mock_msg = MagicMock()
mock_msg.value.return_value = sample_embeddings
# Act
await processor.on_graph_embeddings(mock_msg, None, None)
# Assert
mock_cassandra_store.add_graph_embeddings.assert_called_once_with(sample_embeddings)
@pytest.mark.asyncio
async def test_end_to_end_pipeline_coordination(self, definitions_processor, relationships_processor,
mock_flow_context, sample_chunk):
"""Test end-to-end pipeline coordination from chunk to storage"""
# Arrange
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act - Process through definitions extractor
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Act - Process through relationships extractor
await relationships_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Assert
# Verify both extractors called prompt service
prompt_client = mock_flow_context("prompt-request")
prompt_client.extract_definitions.assert_called_once()
prompt_client.extract_relationships.assert_called_once()
# Verify triples were produced from both extractors
triples_producer = mock_flow_context("triples")
assert triples_producer.send.call_count == 2 # One from each extractor
# Verify entity contexts were produced from definitions extractor
entity_contexts_producer = mock_flow_context("entity-contexts")
entity_contexts_producer.send.assert_called_once()
@pytest.mark.asyncio
async def test_error_handling_in_definitions_extraction(self, definitions_processor, mock_flow_context, sample_chunk):
"""Test error handling in definitions extraction"""
# Arrange
mock_flow_context("prompt-request").extract_definitions.side_effect = Exception("Prompt service unavailable")
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act & Assert
# Should not raise exception, but should handle it gracefully
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Verify prompt was attempted
prompt_client = mock_flow_context("prompt-request")
prompt_client.extract_definitions.assert_called_once()
@pytest.mark.asyncio
async def test_error_handling_in_relationships_extraction(self, relationships_processor, mock_flow_context, sample_chunk):
"""Test error handling in relationships extraction"""
# Arrange
mock_flow_context("prompt-request").extract_relationships.side_effect = Exception("Prompt service unavailable")
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act & Assert
# Should not raise exception, but should handle it gracefully
await relationships_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Verify prompt was attempted
prompt_client = mock_flow_context("prompt-request")
prompt_client.extract_relationships.assert_called_once()
@pytest.mark.asyncio
async def test_empty_extraction_results_handling(self, definitions_processor, mock_flow_context, sample_chunk):
"""Test handling of empty extraction results"""
# Arrange
mock_flow_context("prompt-request").extract_definitions.return_value = []
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Assert
# Should still call producers but with empty results
triples_producer = mock_flow_context("triples")
entity_contexts_producer = mock_flow_context("entity-contexts")
triples_producer.send.assert_called_once()
entity_contexts_producer.send.assert_called_once()
@pytest.mark.asyncio
async def test_invalid_extraction_format_handling(self, definitions_processor, mock_flow_context, sample_chunk):
"""Test handling of invalid extraction response format"""
# Arrange
mock_flow_context("prompt-request").extract_definitions.return_value = "invalid format" # Should be list
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act & Assert
# Should handle invalid format gracefully
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Verify prompt was attempted
prompt_client = mock_flow_context("prompt-request")
prompt_client.extract_definitions.assert_called_once()
@pytest.mark.asyncio
async def test_entity_filtering_and_validation(self, definitions_processor, mock_flow_context):
"""Test entity filtering and validation in extraction"""
# Arrange
mock_flow_context("prompt-request").extract_definitions.return_value = [
{"entity": "Valid Entity", "definition": "Valid definition"},
{"entity": "", "definition": "Empty entity"}, # Should be filtered
{"entity": "Valid Entity 2", "definition": ""}, # Should be filtered
{"entity": None, "definition": "None entity"}, # Should be filtered
{"entity": "Valid Entity 3", "definition": None}, # Should be filtered
]
sample_chunk = Chunk(
metadata=Metadata(id="test", user="user", collection="collection", metadata=[]),
chunk=b"Test chunk"
)
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Assert
# Should only process valid entities
triples_producer = mock_flow_context("triples")
entity_contexts_producer = mock_flow_context("entity-contexts")
triples_producer.send.assert_called_once()
entity_contexts_producer.send.assert_called_once()
@pytest.mark.asyncio
@pytest.mark.slow
async def test_large_batch_processing_performance(self, definitions_processor, relationships_processor,
mock_flow_context):
"""Test performance with large batch of chunks"""
# Arrange
large_chunk_batch = [
Chunk(
metadata=Metadata(id=f"doc-{i}", user="user", collection="collection", metadata=[]),
chunk=f"Document {i} contains machine learning and AI content.".encode("utf-8")
)
for i in range(100) # Large batch
]
mock_consumer = MagicMock()
# Act
import time
start_time = time.time()
for chunk in large_chunk_batch:
mock_msg = MagicMock()
mock_msg.value.return_value = chunk
# Process through both extractors
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
await relationships_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
end_time = time.time()
execution_time = end_time - start_time
# Assert
assert execution_time < 30.0 # Should complete within reasonable time
# Verify all chunks were processed
prompt_client = mock_flow_context("prompt-request")
assert prompt_client.extract_definitions.call_count == 100
assert prompt_client.extract_relationships.call_count == 100
@pytest.mark.asyncio
async def test_metadata_propagation_through_pipeline(self, definitions_processor, mock_flow_context):
"""Test metadata propagation through the pipeline"""
# Arrange
original_metadata = Metadata(
id="test-doc-123",
user="test_user",
collection="test_collection",
metadata=[
Triple(
s=Value(value="doc:test", is_uri=True),
p=Value(value="dc:title", is_uri=True),
o=Value(value="Test Document", is_uri=False)
)
]
)
sample_chunk = Chunk(
metadata=original_metadata,
chunk=b"Test content for metadata propagation"
)
mock_msg = MagicMock()
mock_msg.value.return_value = sample_chunk
mock_consumer = MagicMock()
# Act
await definitions_processor.on_message(mock_msg, mock_consumer, mock_flow_context)
# Assert
# Verify metadata was propagated to output
triples_producer = mock_flow_context("triples")
entity_contexts_producer = mock_flow_context("entity-contexts")
triples_producer.send.assert_called_once()
entity_contexts_producer.send.assert_called_once()
# Check that metadata was included in the calls
triples_call = triples_producer.send.call_args[0][0]
entity_contexts_call = entity_contexts_producer.send.call_args[0][0]
assert triples_call.metadata.id == "test-doc-123"
assert triples_call.metadata.user == "test_user"
assert triples_call.metadata.collection == "test_collection"
assert entity_contexts_call.metadata.id == "test-doc-123"
assert entity_contexts_call.metadata.user == "test_user"
assert entity_contexts_call.metadata.collection == "test_collection"

View file

@ -0,0 +1,540 @@
"""
Integration tests for Object Extraction Service
These tests verify the end-to-end functionality of the object extraction service,
testing configuration management, text-to-object transformation, and service coordination.
Following the TEST_STRATEGY.md approach for integration testing.
"""
import pytest
import json
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from trustgraph.extract.kg.objects.processor import Processor
from trustgraph.schema import (
Chunk, ExtractedObject, Metadata, RowSchema, Field,
PromptRequest, PromptResponse
)
@pytest.mark.integration
class TestObjectExtractionServiceIntegration:
"""Integration tests for Object Extraction Service"""
@pytest.fixture
def integration_config(self):
"""Integration test configuration with multiple schemas"""
customer_schema = {
"name": "customer_records",
"description": "Customer information schema",
"fields": [
{
"name": "customer_id",
"type": "string",
"primary_key": True,
"required": True,
"indexed": True,
"description": "Unique customer identifier"
},
{
"name": "name",
"type": "string",
"required": True,
"description": "Customer full name"
},
{
"name": "email",
"type": "string",
"required": True,
"indexed": True,
"description": "Customer email address"
},
{
"name": "phone",
"type": "string",
"required": False,
"description": "Customer phone number"
}
]
}
product_schema = {
"name": "product_catalog",
"description": "Product catalog schema",
"fields": [
{
"name": "product_id",
"type": "string",
"primary_key": True,
"required": True,
"indexed": True,
"description": "Unique product identifier"
},
{
"name": "name",
"type": "string",
"required": True,
"description": "Product name"
},
{
"name": "price",
"type": "double",
"required": True,
"description": "Product price"
},
{
"name": "category",
"type": "string",
"required": False,
"enum": ["electronics", "clothing", "books", "home"],
"description": "Product category"
}
]
}
return {
"schema": {
"customer_records": json.dumps(customer_schema),
"product_catalog": json.dumps(product_schema)
}
}
@pytest.fixture
def mock_integrated_flow(self):
"""Mock integrated flow context with realistic prompt responses"""
context = MagicMock()
# Mock prompt client with realistic responses
prompt_client = AsyncMock()
def mock_extract_objects(schema, text):
"""Mock extract_objects with schema-aware responses"""
# Schema is now a dict (converted by row_schema_translator)
schema_name = schema.get("name") if isinstance(schema, dict) else schema.name
if schema_name == "customer_records":
if "john" in text.lower():
return [
{
"customer_id": "CUST001",
"name": "John Smith",
"email": "john.smith@email.com",
"phone": "555-0123"
}
]
elif "jane" in text.lower():
return [
{
"customer_id": "CUST002",
"name": "Jane Doe",
"email": "jane.doe@email.com",
"phone": ""
}
]
else:
return []
elif schema_name == "product_catalog":
if "laptop" in text.lower():
return [
{
"product_id": "PROD001",
"name": "Gaming Laptop",
"price": "1299.99",
"category": "electronics"
}
]
elif "book" in text.lower():
return [
{
"product_id": "PROD002",
"name": "Python Programming Guide",
"price": "49.99",
"category": "books"
}
]
else:
return []
return []
prompt_client.extract_objects.side_effect = mock_extract_objects
# Mock output producer
output_producer = AsyncMock()
def context_router(service_name):
if service_name == "prompt-request":
return prompt_client
elif service_name == "output":
return output_producer
else:
return AsyncMock()
context.side_effect = context_router
return context
@pytest.mark.asyncio
async def test_multi_schema_configuration_integration(self, integration_config):
"""Test integration with multiple schema configurations"""
# Arrange - Create mock processor with actual methods
processor = MagicMock()
processor.schemas = {}
processor.config_key = "schema"
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
# Act
await processor.on_schema_config(integration_config, version=1)
# Assert
assert len(processor.schemas) == 2
assert "customer_records" in processor.schemas
assert "product_catalog" in processor.schemas
# Verify customer schema
customer_schema = processor.schemas["customer_records"]
assert customer_schema.name == "customer_records"
assert len(customer_schema.fields) == 4
# Verify product schema
product_schema = processor.schemas["product_catalog"]
assert product_schema.name == "product_catalog"
assert len(product_schema.fields) == 4
# Check enum field in product schema
category_field = next((f for f in product_schema.fields if f.name == "category"), None)
assert category_field is not None
assert len(category_field.enum_values) == 4
assert "electronics" in category_field.enum_values
@pytest.mark.asyncio
async def test_full_service_integration_customer_extraction(self, integration_config, mock_integrated_flow):
"""Test full service integration for customer data extraction"""
# Arrange - Create mock processor with actual methods
processor = MagicMock()
processor.schemas = {}
processor.config_key = "schema"
processor.flow = mock_integrated_flow
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
processor.on_chunk = Processor.on_chunk.__get__(processor, Processor)
processor.extract_objects_for_schema = Processor.extract_objects_for_schema.__get__(processor, Processor)
# Import and bind the convert_values_to_strings function
from trustgraph.extract.kg.objects.processor import convert_values_to_strings
processor.convert_values_to_strings = convert_values_to_strings
# Load configuration
await processor.on_schema_config(integration_config, version=1)
# Create realistic customer data chunk
metadata = Metadata(
id="customer-doc-001",
user="integration_test",
collection="test_documents",
metadata=[]
)
chunk_text = """
Customer Registration Form
Name: John Smith
Email: john.smith@email.com
Phone: 555-0123
Customer ID: CUST001
Registration completed successfully.
"""
chunk = Chunk(metadata=metadata, chunk=chunk_text.encode('utf-8'))
# Mock message
mock_msg = MagicMock()
mock_msg.value.return_value = chunk
# Act
await processor.on_chunk(mock_msg, None, mock_integrated_flow)
# Assert
output_producer = mock_integrated_flow("output")
# Should have calls for both schemas (even if one returns empty)
assert output_producer.send.call_count >= 1
# Find customer extraction
customer_calls = []
for call in output_producer.send.call_args_list:
extracted_obj = call[0][0]
if extracted_obj.schema_name == "customer_records":
customer_calls.append(extracted_obj)
assert len(customer_calls) == 1
customer_obj = customer_calls[0]
assert customer_obj.values["customer_id"] == "CUST001"
assert customer_obj.values["name"] == "John Smith"
assert customer_obj.values["email"] == "john.smith@email.com"
assert customer_obj.confidence > 0.5
@pytest.mark.asyncio
async def test_full_service_integration_product_extraction(self, integration_config, mock_integrated_flow):
"""Test full service integration for product data extraction"""
# Arrange - Create mock processor with actual methods
processor = MagicMock()
processor.schemas = {}
processor.config_key = "schema"
processor.flow = mock_integrated_flow
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
processor.on_chunk = Processor.on_chunk.__get__(processor, Processor)
processor.extract_objects_for_schema = Processor.extract_objects_for_schema.__get__(processor, Processor)
# Import and bind the convert_values_to_strings function
from trustgraph.extract.kg.objects.processor import convert_values_to_strings
processor.convert_values_to_strings = convert_values_to_strings
# Load configuration
await processor.on_schema_config(integration_config, version=1)
# Create realistic product data chunk
metadata = Metadata(
id="product-doc-001",
user="integration_test",
collection="test_documents",
metadata=[]
)
chunk_text = """
Product Specification Sheet
Product Name: Gaming Laptop
Product ID: PROD001
Price: $1,299.99
Category: Electronics
High-performance gaming laptop with latest specifications.
"""
chunk = Chunk(metadata=metadata, chunk=chunk_text.encode('utf-8'))
# Mock message
mock_msg = MagicMock()
mock_msg.value.return_value = chunk
# Act
await processor.on_chunk(mock_msg, None, mock_integrated_flow)
# Assert
output_producer = mock_integrated_flow("output")
# Find product extraction
product_calls = []
for call in output_producer.send.call_args_list:
extracted_obj = call[0][0]
if extracted_obj.schema_name == "product_catalog":
product_calls.append(extracted_obj)
assert len(product_calls) == 1
product_obj = product_calls[0]
assert product_obj.values["product_id"] == "PROD001"
assert product_obj.values["name"] == "Gaming Laptop"
assert product_obj.values["price"] == "1299.99"
assert product_obj.values["category"] == "electronics"
@pytest.mark.asyncio
async def test_concurrent_extraction_integration(self, integration_config, mock_integrated_flow):
"""Test concurrent processing of multiple chunks"""
# Arrange - Create mock processor with actual methods
processor = MagicMock()
processor.schemas = {}
processor.config_key = "schema"
processor.flow = mock_integrated_flow
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
processor.on_chunk = Processor.on_chunk.__get__(processor, Processor)
processor.extract_objects_for_schema = Processor.extract_objects_for_schema.__get__(processor, Processor)
# Import and bind the convert_values_to_strings function
from trustgraph.extract.kg.objects.processor import convert_values_to_strings
processor.convert_values_to_strings = convert_values_to_strings
# Load configuration
await processor.on_schema_config(integration_config, version=1)
# Create multiple test chunks
chunks_data = [
("customer-chunk-1", "Customer: John Smith, email: john.smith@email.com, ID: CUST001"),
("customer-chunk-2", "Customer: Jane Doe, email: jane.doe@email.com, ID: CUST002"),
("product-chunk-1", "Product: Gaming Laptop, ID: PROD001, Price: $1299.99, Category: electronics"),
("product-chunk-2", "Product: Python Programming Guide, ID: PROD002, Price: $49.99, Category: books")
]
chunks = []
for chunk_id, text in chunks_data:
metadata = Metadata(
id=chunk_id,
user="concurrent_test",
collection="test_collection",
metadata=[]
)
chunk = Chunk(metadata=metadata, chunk=text.encode('utf-8'))
chunks.append(chunk)
# Act - Process chunks concurrently
tasks = []
for chunk in chunks:
mock_msg = MagicMock()
mock_msg.value.return_value = chunk
task = processor.on_chunk(mock_msg, None, mock_integrated_flow)
tasks.append(task)
await asyncio.gather(*tasks)
# Assert
output_producer = mock_integrated_flow("output")
# Should have processed all chunks (some may produce objects, some may not)
assert output_producer.send.call_count >= 2 # At least customer and product extractions
# Verify we got both types of objects
extracted_objects = []
for call in output_producer.send.call_args_list:
extracted_objects.append(call[0][0])
customer_objects = [obj for obj in extracted_objects if obj.schema_name == "customer_records"]
product_objects = [obj for obj in extracted_objects if obj.schema_name == "product_catalog"]
assert len(customer_objects) >= 1
assert len(product_objects) >= 1
@pytest.mark.asyncio
async def test_configuration_reload_integration(self, integration_config, mock_integrated_flow):
"""Test configuration reload during service operation"""
# Arrange - Create mock processor with actual methods
processor = MagicMock()
processor.schemas = {}
processor.config_key = "schema"
processor.flow = mock_integrated_flow
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
# Load initial configuration (only customer schema)
initial_config = {
"schema": {
"customer_records": integration_config["schema"]["customer_records"]
}
}
await processor.on_schema_config(initial_config, version=1)
assert len(processor.schemas) == 1
assert "customer_records" in processor.schemas
assert "product_catalog" not in processor.schemas
# Act - Reload with full configuration
await processor.on_schema_config(integration_config, version=2)
# Assert
assert len(processor.schemas) == 2
assert "customer_records" in processor.schemas
assert "product_catalog" in processor.schemas
@pytest.mark.asyncio
async def test_error_resilience_integration(self, integration_config):
"""Test service resilience to various error conditions"""
# Arrange - Create mock processor with actual methods
processor = MagicMock()
processor.schemas = {}
processor.config_key = "schema"
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
processor.on_chunk = Processor.on_chunk.__get__(processor, Processor)
processor.extract_objects_for_schema = Processor.extract_objects_for_schema.__get__(processor, Processor)
# Import and bind the convert_values_to_strings function
from trustgraph.extract.kg.objects.processor import convert_values_to_strings
processor.convert_values_to_strings = convert_values_to_strings
# Mock flow with failing prompt service
failing_flow = MagicMock()
failing_prompt = AsyncMock()
failing_prompt.extract_rows.side_effect = Exception("Prompt service unavailable")
def failing_context_router(service_name):
if service_name == "prompt-request":
return failing_prompt
elif service_name == "output":
return AsyncMock()
else:
return AsyncMock()
failing_flow.side_effect = failing_context_router
processor.flow = failing_flow
# Load configuration
await processor.on_schema_config(integration_config, version=1)
# Create test chunk
metadata = Metadata(id="error-test", user="test", collection="test", metadata=[])
chunk = Chunk(metadata=metadata, chunk=b"Some text that will fail to process")
mock_msg = MagicMock()
mock_msg.value.return_value = chunk
# Act & Assert - Should not raise exception
try:
await processor.on_chunk(mock_msg, None, failing_flow)
# Should complete without throwing exception
except Exception as e:
pytest.fail(f"Service should handle errors gracefully, but raised: {e}")
@pytest.mark.asyncio
async def test_metadata_propagation_integration(self, integration_config, mock_integrated_flow):
"""Test proper metadata propagation through extraction pipeline"""
# Arrange - Create mock processor with actual methods
processor = MagicMock()
processor.schemas = {}
processor.config_key = "schema"
processor.flow = mock_integrated_flow
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
processor.on_chunk = Processor.on_chunk.__get__(processor, Processor)
processor.extract_objects_for_schema = Processor.extract_objects_for_schema.__get__(processor, Processor)
# Import and bind the convert_values_to_strings function
from trustgraph.extract.kg.objects.processor import convert_values_to_strings
processor.convert_values_to_strings = convert_values_to_strings
# Load configuration
await processor.on_schema_config(integration_config, version=1)
# Create chunk with rich metadata
original_metadata = Metadata(
id="metadata-test-chunk",
user="test_user",
collection="test_collection",
metadata=[] # Could include source document metadata
)
chunk = Chunk(
metadata=original_metadata,
chunk=b"Customer: John Smith, ID: CUST001, email: john.smith@email.com"
)
mock_msg = MagicMock()
mock_msg.value.return_value = chunk
# Act
await processor.on_chunk(mock_msg, None, mock_integrated_flow)
# Assert
output_producer = mock_integrated_flow("output")
# Find extracted object
extracted_obj = None
for call in output_producer.send.call_args_list:
obj = call[0][0]
if obj.schema_name == "customer_records":
extracted_obj = obj
break
assert extracted_obj is not None
# Verify metadata propagation
assert extracted_obj.metadata.user == "test_user"
assert extracted_obj.metadata.collection == "test_collection"
assert "metadata-test-chunk" in extracted_obj.metadata.id # Should include source reference

View file

@ -0,0 +1,384 @@
"""
Integration tests for Cassandra Object Storage
These tests verify the end-to-end functionality of storing ExtractedObjects
in Cassandra, including table creation, data insertion, and error handling.
"""
import pytest
from unittest.mock import MagicMock, AsyncMock, patch
import json
import uuid
from trustgraph.storage.objects.cassandra.write import Processor
from trustgraph.schema import ExtractedObject, Metadata, RowSchema, Field
@pytest.mark.integration
class TestObjectsCassandraIntegration:
"""Integration tests for Cassandra object storage"""
@pytest.fixture
def mock_cassandra_session(self):
"""Mock Cassandra session for integration tests"""
session = MagicMock()
session.execute = MagicMock()
return session
@pytest.fixture
def mock_cassandra_cluster(self, mock_cassandra_session):
"""Mock Cassandra cluster"""
cluster = MagicMock()
cluster.connect.return_value = mock_cassandra_session
cluster.shutdown = MagicMock()
return cluster
@pytest.fixture
def processor_with_mocks(self, mock_cassandra_cluster, mock_cassandra_session):
"""Create processor with mocked Cassandra dependencies"""
processor = MagicMock()
processor.graph_host = "localhost"
processor.graph_username = None
processor.graph_password = None
processor.config_key = "schema"
processor.schemas = {}
processor.known_keyspaces = set()
processor.known_tables = {}
processor.cluster = None
processor.session = None
# Bind actual methods
processor.connect_cassandra = Processor.connect_cassandra.__get__(processor, Processor)
processor.ensure_keyspace = Processor.ensure_keyspace.__get__(processor, Processor)
processor.ensure_table = Processor.ensure_table.__get__(processor, Processor)
processor.sanitize_name = Processor.sanitize_name.__get__(processor, Processor)
processor.sanitize_table = Processor.sanitize_table.__get__(processor, Processor)
processor.get_cassandra_type = Processor.get_cassandra_type.__get__(processor, Processor)
processor.convert_value = Processor.convert_value.__get__(processor, Processor)
processor.on_schema_config = Processor.on_schema_config.__get__(processor, Processor)
processor.on_object = Processor.on_object.__get__(processor, Processor)
return processor, mock_cassandra_cluster, mock_cassandra_session
@pytest.mark.asyncio
async def test_end_to_end_object_storage(self, processor_with_mocks):
"""Test complete flow from schema config to object storage"""
processor, mock_cluster, mock_session = processor_with_mocks
# Mock Cluster creation
with patch('trustgraph.storage.objects.cassandra.write.Cluster', return_value=mock_cluster):
# Step 1: Configure schema
config = {
"schema": {
"customer_records": json.dumps({
"name": "customer_records",
"description": "Customer information",
"fields": [
{"name": "customer_id", "type": "string", "primary_key": True},
{"name": "name", "type": "string", "required": True},
{"name": "email", "type": "string", "indexed": True},
{"name": "age", "type": "integer"}
]
})
}
}
await processor.on_schema_config(config, version=1)
assert "customer_records" in processor.schemas
# Step 2: Process an ExtractedObject
test_obj = ExtractedObject(
metadata=Metadata(
id="doc-001",
user="test_user",
collection="import_2024",
metadata=[]
),
schema_name="customer_records",
values={
"customer_id": "CUST001",
"name": "John Doe",
"email": "john@example.com",
"age": "30"
},
confidence=0.95,
source_span="Customer: John Doe..."
)
msg = MagicMock()
msg.value.return_value = test_obj
await processor.on_object(msg, None, None)
# Verify Cassandra interactions
assert mock_cluster.connect.called
# Verify keyspace creation
keyspace_calls = [call for call in mock_session.execute.call_args_list
if "CREATE KEYSPACE" in str(call)]
assert len(keyspace_calls) == 1
assert "test_user" in str(keyspace_calls[0])
# Verify table creation
table_calls = [call for call in mock_session.execute.call_args_list
if "CREATE TABLE" in str(call)]
assert len(table_calls) == 1
assert "o_customer_records" in str(table_calls[0]) # Table gets o_ prefix
assert "collection text" in str(table_calls[0])
assert "PRIMARY KEY ((collection, customer_id))" in str(table_calls[0])
# Verify index creation
index_calls = [call for call in mock_session.execute.call_args_list
if "CREATE INDEX" in str(call)]
assert len(index_calls) == 1
assert "email" in str(index_calls[0])
# Verify data insertion
insert_calls = [call for call in mock_session.execute.call_args_list
if "INSERT INTO" in str(call)]
assert len(insert_calls) == 1
insert_call = insert_calls[0]
assert "test_user.o_customer_records" in str(insert_call) # Table gets o_ prefix
# Check inserted values
values = insert_call[0][1]
assert "import_2024" in values # collection
assert "CUST001" in values # customer_id
assert "John Doe" in values # name
assert "john@example.com" in values # email
assert 30 in values # age (converted to int)
@pytest.mark.asyncio
async def test_multi_schema_handling(self, processor_with_mocks):
"""Test handling multiple schemas and objects"""
processor, mock_cluster, mock_session = processor_with_mocks
with patch('trustgraph.storage.objects.cassandra.write.Cluster', return_value=mock_cluster):
# Configure multiple schemas
config = {
"schema": {
"products": json.dumps({
"name": "products",
"fields": [
{"name": "product_id", "type": "string", "primary_key": True},
{"name": "name", "type": "string"},
{"name": "price", "type": "float"}
]
}),
"orders": json.dumps({
"name": "orders",
"fields": [
{"name": "order_id", "type": "string", "primary_key": True},
{"name": "customer_id", "type": "string"},
{"name": "total", "type": "float"}
]
})
}
}
await processor.on_schema_config(config, version=1)
assert len(processor.schemas) == 2
# Process objects for different schemas
product_obj = ExtractedObject(
metadata=Metadata(id="p1", user="shop", collection="catalog", metadata=[]),
schema_name="products",
values={"product_id": "P001", "name": "Widget", "price": "19.99"},
confidence=0.9,
source_span="Product..."
)
order_obj = ExtractedObject(
metadata=Metadata(id="o1", user="shop", collection="sales", metadata=[]),
schema_name="orders",
values={"order_id": "O001", "customer_id": "C001", "total": "59.97"},
confidence=0.85,
source_span="Order..."
)
# Process both objects
for obj in [product_obj, order_obj]:
msg = MagicMock()
msg.value.return_value = obj
await processor.on_object(msg, None, None)
# Verify separate tables were created
table_calls = [call for call in mock_session.execute.call_args_list
if "CREATE TABLE" in str(call)]
assert len(table_calls) == 2
assert any("o_products" in str(call) for call in table_calls) # Tables get o_ prefix
assert any("o_orders" in str(call) for call in table_calls) # Tables get o_ prefix
@pytest.mark.asyncio
async def test_missing_required_fields(self, processor_with_mocks):
"""Test handling of objects with missing required fields"""
processor, mock_cluster, mock_session = processor_with_mocks
with patch('trustgraph.storage.objects.cassandra.write.Cluster', return_value=mock_cluster):
# Configure schema with required field
processor.schemas["test_schema"] = RowSchema(
name="test_schema",
description="Test",
fields=[
Field(name="id", type="string", size=50, primary=True, required=True),
Field(name="required_field", type="string", size=100, required=True)
]
)
# Create object missing required field
test_obj = ExtractedObject(
metadata=Metadata(id="t1", user="test", collection="test", metadata=[]),
schema_name="test_schema",
values={"id": "123"}, # missing required_field
confidence=0.8,
source_span="Test"
)
msg = MagicMock()
msg.value.return_value = test_obj
# Should still process (Cassandra doesn't enforce NOT NULL)
await processor.on_object(msg, None, None)
# Verify insert was attempted
insert_calls = [call for call in mock_session.execute.call_args_list
if "INSERT INTO" in str(call)]
assert len(insert_calls) == 1
@pytest.mark.asyncio
async def test_schema_without_primary_key(self, processor_with_mocks):
"""Test handling schemas without defined primary keys"""
processor, mock_cluster, mock_session = processor_with_mocks
with patch('trustgraph.storage.objects.cassandra.write.Cluster', return_value=mock_cluster):
# Configure schema without primary key
processor.schemas["events"] = RowSchema(
name="events",
description="Event log",
fields=[
Field(name="event_type", type="string", size=50),
Field(name="timestamp", type="timestamp", size=0)
]
)
# Process object
test_obj = ExtractedObject(
metadata=Metadata(id="e1", user="logger", collection="app_events", metadata=[]),
schema_name="events",
values={"event_type": "login", "timestamp": "2024-01-01T10:00:00Z"},
confidence=1.0,
source_span="Event"
)
msg = MagicMock()
msg.value.return_value = test_obj
await processor.on_object(msg, None, None)
# Verify synthetic_id was added
table_calls = [call for call in mock_session.execute.call_args_list
if "CREATE TABLE" in str(call)]
assert len(table_calls) == 1
assert "synthetic_id uuid" in str(table_calls[0])
# Verify insert includes UUID
insert_calls = [call for call in mock_session.execute.call_args_list
if "INSERT INTO" in str(call)]
assert len(insert_calls) == 1
values = insert_calls[0][0][1]
# Check that a UUID was generated (will be in values list)
uuid_found = any(isinstance(v, uuid.UUID) for v in values)
assert uuid_found
@pytest.mark.asyncio
async def test_authentication_handling(self, processor_with_mocks):
"""Test Cassandra authentication"""
processor, mock_cluster, mock_session = processor_with_mocks
processor.graph_username = "cassandra_user"
processor.graph_password = "cassandra_pass"
with patch('trustgraph.storage.objects.cassandra.write.Cluster') as mock_cluster_class:
with patch('trustgraph.storage.objects.cassandra.write.PlainTextAuthProvider') as mock_auth:
mock_cluster_class.return_value = mock_cluster
# Trigger connection
processor.connect_cassandra()
# Verify authentication was configured
mock_auth.assert_called_once_with(
username="cassandra_user",
password="cassandra_pass"
)
mock_cluster_class.assert_called_once()
call_kwargs = mock_cluster_class.call_args[1]
assert 'auth_provider' in call_kwargs
@pytest.mark.asyncio
async def test_error_handling_during_insert(self, processor_with_mocks):
"""Test error handling when insertion fails"""
processor, mock_cluster, mock_session = processor_with_mocks
with patch('trustgraph.storage.objects.cassandra.write.Cluster', return_value=mock_cluster):
processor.schemas["test"] = RowSchema(
name="test",
fields=[Field(name="id", type="string", size=50, primary=True)]
)
# Make insert fail
mock_session.execute.side_effect = [
None, # keyspace creation succeeds
None, # table creation succeeds
Exception("Connection timeout") # insert fails
]
test_obj = ExtractedObject(
metadata=Metadata(id="t1", user="test", collection="test", metadata=[]),
schema_name="test",
values={"id": "123"},
confidence=0.9,
source_span="Test"
)
msg = MagicMock()
msg.value.return_value = test_obj
# Should raise the exception
with pytest.raises(Exception, match="Connection timeout"):
await processor.on_object(msg, None, None)
@pytest.mark.asyncio
async def test_collection_partitioning(self, processor_with_mocks):
"""Test that objects are properly partitioned by collection"""
processor, mock_cluster, mock_session = processor_with_mocks
with patch('trustgraph.storage.objects.cassandra.write.Cluster', return_value=mock_cluster):
processor.schemas["data"] = RowSchema(
name="data",
fields=[Field(name="id", type="string", size=50, primary=True)]
)
# Process objects from different collections
collections = ["import_jan", "import_feb", "import_mar"]
for coll in collections:
obj = ExtractedObject(
metadata=Metadata(id=f"{coll}-1", user="analytics", collection=coll, metadata=[]),
schema_name="data",
values={"id": f"ID-{coll}"},
confidence=0.9,
source_span="Data"
)
msg = MagicMock()
msg.value.return_value = obj
await processor.on_object(msg, None, None)
# Verify all inserts include collection in values
insert_calls = [call for call in mock_session.execute.call_args_list
if "INSERT INTO" in str(call)]
assert len(insert_calls) == 3
# Check each insert has the correct collection
for i, call in enumerate(insert_calls):
values = call[0][1]
assert collections[i] in values

View file

@ -0,0 +1,205 @@
"""
Simplified integration tests for Template Service
These tests verify the basic functionality of the template service
without the full message queue infrastructure.
"""
import pytest
import json
from unittest.mock import AsyncMock, MagicMock
from trustgraph.schema import PromptRequest, PromptResponse
from trustgraph.template.prompt_manager import PromptManager
@pytest.mark.integration
class TestTemplateServiceSimple:
"""Simplified integration tests for Template Service components"""
@pytest.fixture
def sample_config(self):
"""Sample configuration for testing"""
return {
"system": json.dumps("You are a helpful assistant."),
"template-index": json.dumps(["greeting", "json_test"]),
"template.greeting": json.dumps({
"prompt": "Hello {{ name }}, welcome to {{ system_name }}!",
"response-type": "text"
}),
"template.json_test": json.dumps({
"prompt": "Generate profile for {{ username }}",
"response-type": "json",
"schema": {
"type": "object",
"properties": {
"name": {"type": "string"},
"role": {"type": "string"}
},
"required": ["name", "role"]
}
})
}
@pytest.fixture
def prompt_manager(self, sample_config):
"""Create a configured PromptManager"""
pm = PromptManager()
pm.load_config(sample_config)
pm.terms["system_name"] = "TrustGraph"
return pm
@pytest.mark.asyncio
async def test_prompt_manager_text_invocation(self, prompt_manager):
"""Test PromptManager text response invocation"""
# Mock LLM function
async def mock_llm(system, prompt):
assert system == "You are a helpful assistant."
assert "Hello Alice, welcome to TrustGraph!" in prompt
return "Welcome message processed!"
result = await prompt_manager.invoke("greeting", {"name": "Alice"}, mock_llm)
assert result == "Welcome message processed!"
@pytest.mark.asyncio
async def test_prompt_manager_json_invocation(self, prompt_manager):
"""Test PromptManager JSON response invocation"""
# Mock LLM function
async def mock_llm(system, prompt):
assert "Generate profile for johndoe" in prompt
return '{"name": "John Doe", "role": "user"}'
result = await prompt_manager.invoke("json_test", {"username": "johndoe"}, mock_llm)
assert isinstance(result, dict)
assert result["name"] == "John Doe"
assert result["role"] == "user"
@pytest.mark.asyncio
async def test_prompt_manager_json_validation_error(self, prompt_manager):
"""Test JSON schema validation failure"""
# Mock LLM function that returns invalid JSON
async def mock_llm(system, prompt):
return '{"name": "John Doe"}' # Missing required "role"
with pytest.raises(RuntimeError) as exc_info:
await prompt_manager.invoke("json_test", {"username": "johndoe"}, mock_llm)
assert "Schema validation fail" in str(exc_info.value)
@pytest.mark.asyncio
async def test_prompt_manager_json_parse_error(self, prompt_manager):
"""Test JSON parsing failure"""
# Mock LLM function that returns non-JSON
async def mock_llm(system, prompt):
return "This is not JSON at all"
with pytest.raises(RuntimeError) as exc_info:
await prompt_manager.invoke("json_test", {"username": "johndoe"}, mock_llm)
assert "JSON parse fail" in str(exc_info.value)
@pytest.mark.asyncio
async def test_prompt_manager_unknown_prompt(self, prompt_manager):
"""Test unknown prompt ID handling"""
async def mock_llm(system, prompt):
return "Response"
with pytest.raises(KeyError):
await prompt_manager.invoke("unknown_prompt", {}, mock_llm)
@pytest.mark.asyncio
async def test_prompt_manager_term_merging(self, prompt_manager):
"""Test proper term merging (global + prompt + input)"""
# Add prompt-specific terms
prompt_manager.prompts["greeting"].terms = {"greeting_prefix": "Hi"}
async def mock_llm(system, prompt):
# Should have global term (system_name), input term (name), and any prompt terms
assert "TrustGraph" in prompt # Global term
assert "Bob" in prompt # Input term
return "Merged correctly"
result = await prompt_manager.invoke("greeting", {"name": "Bob"}, mock_llm)
assert result == "Merged correctly"
def test_prompt_manager_template_rendering(self, prompt_manager):
"""Test direct template rendering"""
result = prompt_manager.render("greeting", {"name": "Charlie"})
assert "Hello Charlie, welcome to TrustGraph!" == result.strip()
def test_prompt_manager_configuration_loading(self):
"""Test configuration loading with various formats"""
pm = PromptManager()
# Test empty configuration
pm.load_config({})
assert pm.config.system_template == "Be helpful."
assert len(pm.prompts) == 0
# Test configuration with single prompt
config = {
"system": json.dumps("Test system"),
"template-index": json.dumps(["test"]),
"template.test": json.dumps({
"prompt": "Test {{ value }}",
"response-type": "text"
})
}
pm.load_config(config)
assert pm.config.system_template == "Test system"
assert "test" in pm.prompts
assert pm.prompts["test"].response_type == "text"
@pytest.mark.asyncio
async def test_prompt_manager_json_with_markdown(self, prompt_manager):
"""Test JSON extraction from markdown code blocks"""
async def mock_llm(system, prompt):
return '''
Here's the profile:
```json
{"name": "Jane Smith", "role": "admin"}
```
'''
result = await prompt_manager.invoke("json_test", {"username": "jane"}, mock_llm)
assert isinstance(result, dict)
assert result["name"] == "Jane Smith"
assert result["role"] == "admin"
def test_prompt_manager_error_handling_in_templates(self, prompt_manager):
"""Test error handling in template rendering"""
# Test with missing variable - ibis might handle this differently than Jinja2
try:
result = prompt_manager.render("greeting", {}) # Missing 'name'
# If no exception, check that result is still a string
assert isinstance(result, str)
except Exception as e:
# If exception is raised, that's also acceptable
assert "name" in str(e) or "undefined" in str(e).lower() or "variable" in str(e).lower()
@pytest.mark.asyncio
async def test_concurrent_prompt_invocations(self, prompt_manager):
"""Test concurrent invocations"""
async def mock_llm(system, prompt):
# Extract name from prompt for response
if "Alice" in prompt:
return "Alice response"
elif "Bob" in prompt:
return "Bob response"
else:
return "Default response"
# Run concurrent invocations
import asyncio
results = await asyncio.gather(
prompt_manager.invoke("greeting", {"name": "Alice"}, mock_llm),
prompt_manager.invoke("greeting", {"name": "Bob"}, mock_llm),
)
assert "Alice response" in results
assert "Bob response" in results

View file

@ -0,0 +1,429 @@
"""
Integration tests for Text Completion Service (OpenAI)
These tests verify the end-to-end functionality of the OpenAI text completion service,
testing API connectivity, authentication, rate limiting, error handling, and token tracking.
Following the TEST_STRATEGY.md approach for integration testing.
"""
import pytest
import os
from unittest.mock import AsyncMock, MagicMock, patch
from openai import OpenAI, RateLimitError
from openai.types.chat import ChatCompletion, ChatCompletionMessage
from openai.types.chat.chat_completion import Choice
from openai.types.completion_usage import CompletionUsage
from trustgraph.model.text_completion.openai.llm import Processor
from trustgraph.exceptions import TooManyRequests
from trustgraph.base import LlmResult
from trustgraph.schema import TextCompletionRequest, TextCompletionResponse, Error
@pytest.mark.integration
class TestTextCompletionIntegration:
"""Integration tests for OpenAI text completion service coordination"""
@pytest.fixture
def mock_openai_client(self):
"""Mock OpenAI client that returns realistic responses"""
client = MagicMock(spec=OpenAI)
# Mock chat completion response
usage = CompletionUsage(prompt_tokens=50, completion_tokens=100, total_tokens=150)
message = ChatCompletionMessage(role="assistant", content="This is a test response from the AI model.")
choice = Choice(index=0, message=message, finish_reason="stop")
completion = ChatCompletion(
id="chatcmpl-test123",
choices=[choice],
created=1234567890,
model="gpt-3.5-turbo",
object="chat.completion",
usage=usage
)
client.chat.completions.create.return_value = completion
return client
@pytest.fixture
def processor_config(self):
"""Configuration for processor testing"""
return {
"model": "gpt-3.5-turbo",
"temperature": 0.7,
"max_output": 1024,
}
@pytest.fixture
def text_completion_processor(self, processor_config):
"""Create text completion processor with test configuration"""
# Create a minimal processor instance for testing generate_content
processor = MagicMock()
processor.model = processor_config["model"]
processor.temperature = processor_config["temperature"]
processor.max_output = processor_config["max_output"]
# Add the actual generate_content method from Processor class
processor.generate_content = Processor.generate_content.__get__(processor, Processor)
return processor
@pytest.mark.asyncio
async def test_text_completion_successful_generation(self, text_completion_processor, mock_openai_client):
"""Test successful text completion generation"""
# Arrange
text_completion_processor.openai = mock_openai_client
system_prompt = "You are a helpful assistant."
user_prompt = "What is machine learning?"
# Act
result = await text_completion_processor.generate_content(system_prompt, user_prompt)
# Assert
assert isinstance(result, LlmResult)
assert result.text == "This is a test response from the AI model."
assert result.in_token == 50
assert result.out_token == 100
assert result.model == "gpt-3.5-turbo"
# Verify OpenAI API was called correctly
mock_openai_client.chat.completions.create.assert_called_once()
call_args = mock_openai_client.chat.completions.create.call_args
assert call_args.kwargs['model'] == "gpt-3.5-turbo"
assert call_args.kwargs['temperature'] == 0.7
assert call_args.kwargs['max_tokens'] == 1024
assert len(call_args.kwargs['messages']) == 1
assert call_args.kwargs['messages'][0]['role'] == "user"
assert "You are a helpful assistant." in call_args.kwargs['messages'][0]['content'][0]['text']
assert "What is machine learning?" in call_args.kwargs['messages'][0]['content'][0]['text']
@pytest.mark.asyncio
async def test_text_completion_with_different_configurations(self, mock_openai_client):
"""Test text completion with various configuration parameters"""
# Test different configurations
test_configs = [
{"model": "gpt-4", "temperature": 0.0, "max_output": 512},
{"model": "gpt-3.5-turbo", "temperature": 1.0, "max_output": 2048},
{"model": "gpt-4-turbo", "temperature": 0.5, "max_output": 4096}
]
for config in test_configs:
# Arrange - Create minimal processor mock
processor = MagicMock()
processor.model = config['model']
processor.temperature = config['temperature']
processor.max_output = config['max_output']
processor.openai = mock_openai_client
# Add the actual generate_content method
processor.generate_content = Processor.generate_content.__get__(processor, Processor)
# Act
result = await processor.generate_content("System prompt", "User prompt")
# Assert
assert isinstance(result, LlmResult)
assert result.text == "This is a test response from the AI model."
assert result.in_token == 50
assert result.out_token == 100
# Note: result.model comes from mock response, not processor config
# Verify configuration was applied
call_args = mock_openai_client.chat.completions.create.call_args
assert call_args.kwargs['model'] == config['model']
assert call_args.kwargs['temperature'] == config['temperature']
assert call_args.kwargs['max_tokens'] == config['max_output']
# Reset mock for next iteration
mock_openai_client.reset_mock()
@pytest.mark.asyncio
async def test_text_completion_rate_limit_handling(self, text_completion_processor, mock_openai_client):
"""Test proper rate limit error handling"""
# Arrange
mock_openai_client.chat.completions.create.side_effect = RateLimitError(
"Rate limit exceeded",
response=MagicMock(status_code=429),
body={}
)
text_completion_processor.openai = mock_openai_client
# Act & Assert
with pytest.raises(TooManyRequests):
await text_completion_processor.generate_content("System prompt", "User prompt")
# Verify OpenAI API was called
mock_openai_client.chat.completions.create.assert_called_once()
@pytest.mark.asyncio
async def test_text_completion_api_error_handling(self, text_completion_processor, mock_openai_client):
"""Test handling of general API errors"""
# Arrange
mock_openai_client.chat.completions.create.side_effect = Exception("API connection failed")
text_completion_processor.openai = mock_openai_client
# Act & Assert
with pytest.raises(Exception) as exc_info:
await text_completion_processor.generate_content("System prompt", "User prompt")
assert "API connection failed" in str(exc_info.value)
mock_openai_client.chat.completions.create.assert_called_once()
@pytest.mark.asyncio
async def test_text_completion_token_tracking(self, text_completion_processor, mock_openai_client):
"""Test accurate token counting and tracking"""
# Arrange - Different token counts for multiple requests
test_cases = [
(25, 75), # Small request
(100, 200), # Medium request
(500, 1000) # Large request
]
for input_tokens, output_tokens in test_cases:
# Update mock response with different token counts
usage = CompletionUsage(
prompt_tokens=input_tokens,
completion_tokens=output_tokens,
total_tokens=input_tokens + output_tokens
)
message = ChatCompletionMessage(role="assistant", content="Test response")
choice = Choice(index=0, message=message, finish_reason="stop")
completion = ChatCompletion(
id="chatcmpl-test123",
choices=[choice],
created=1234567890,
model="gpt-3.5-turbo",
object="chat.completion",
usage=usage
)
mock_openai_client.chat.completions.create.return_value = completion
text_completion_processor.openai = mock_openai_client
# Act
result = await text_completion_processor.generate_content("System", "Prompt")
# Assert
assert result.in_token == input_tokens
assert result.out_token == output_tokens
assert result.model == "gpt-3.5-turbo"
# Reset mock for next iteration
mock_openai_client.reset_mock()
@pytest.mark.asyncio
async def test_text_completion_prompt_construction(self, text_completion_processor, mock_openai_client):
"""Test proper prompt construction with system and user prompts"""
# Arrange
text_completion_processor.openai = mock_openai_client
system_prompt = "You are an expert in artificial intelligence."
user_prompt = "Explain neural networks in simple terms."
# Act
result = await text_completion_processor.generate_content(system_prompt, user_prompt)
# Assert
call_args = mock_openai_client.chat.completions.create.call_args
sent_message = call_args.kwargs['messages'][0]['content'][0]['text']
# Verify system and user prompts are combined correctly
assert system_prompt in sent_message
assert user_prompt in sent_message
assert sent_message.startswith(system_prompt)
assert user_prompt in sent_message
@pytest.mark.asyncio
async def test_text_completion_concurrent_requests(self, processor_config, mock_openai_client):
"""Test handling of concurrent requests"""
# Arrange
processors = []
for i in range(5):
processor = MagicMock()
processor.model = processor_config["model"]
processor.temperature = processor_config["temperature"]
processor.max_output = processor_config["max_output"]
processor.openai = mock_openai_client
processor.generate_content = Processor.generate_content.__get__(processor, Processor)
processors.append(processor)
# Simulate multiple concurrent requests
tasks = []
for i, processor in enumerate(processors):
task = processor.generate_content(f"System {i}", f"Prompt {i}")
tasks.append(task)
# Act
import asyncio
results = await asyncio.gather(*tasks)
# Assert
assert len(results) == 5
for result in results:
assert isinstance(result, LlmResult)
assert result.text == "This is a test response from the AI model."
assert result.in_token == 50
assert result.out_token == 100
# Verify all requests were processed
assert mock_openai_client.chat.completions.create.call_count == 5
@pytest.mark.asyncio
async def test_text_completion_response_format_validation(self, text_completion_processor, mock_openai_client):
"""Test response format and structure validation"""
# Arrange
text_completion_processor.openai = mock_openai_client
# Act
result = await text_completion_processor.generate_content("System", "Prompt")
# Assert
# Verify OpenAI API call parameters
call_args = mock_openai_client.chat.completions.create.call_args
assert call_args.kwargs['response_format'] == {"type": "text"}
assert call_args.kwargs['top_p'] == 1
assert call_args.kwargs['frequency_penalty'] == 0
assert call_args.kwargs['presence_penalty'] == 0
# Verify result structure
assert hasattr(result, 'text')
assert hasattr(result, 'in_token')
assert hasattr(result, 'out_token')
assert hasattr(result, 'model')
@pytest.mark.asyncio
async def test_text_completion_authentication_patterns(self):
"""Test different authentication configurations"""
# Test missing API key first (this should fail early)
with pytest.raises(RuntimeError) as exc_info:
Processor(id="test-no-key", api_key=None)
assert "OpenAI API key not specified" in str(exc_info.value)
# Test authentication pattern by examining the initialization logic
# Since we can't fully instantiate due to taskgroup requirements,
# we'll test the authentication logic directly
from trustgraph.model.text_completion.openai.llm import default_api_key, default_base_url
# Test default values
assert default_base_url == "https://api.openai.com/v1"
# Test configuration parameters
test_configs = [
{"api_key": "test-key-1", "url": "https://api.openai.com/v1"},
{"api_key": "test-key-2", "url": "https://custom.openai.com/v1"},
]
for config in test_configs:
# We can't fully test instantiation due to taskgroup,
# but we can verify the authentication logic would work
assert config["api_key"] is not None
assert config["url"] is not None
@pytest.mark.asyncio
async def test_text_completion_error_propagation(self, text_completion_processor, mock_openai_client):
"""Test error propagation through the service"""
# Test different error types
error_cases = [
(RateLimitError("Rate limit", response=MagicMock(status_code=429), body={}), TooManyRequests),
(Exception("Connection timeout"), Exception),
(ValueError("Invalid request"), ValueError),
]
for error_input, expected_error in error_cases:
# Arrange
mock_openai_client.chat.completions.create.side_effect = error_input
text_completion_processor.openai = mock_openai_client
# Act & Assert
with pytest.raises(expected_error):
await text_completion_processor.generate_content("System", "Prompt")
# Reset mock for next iteration
mock_openai_client.reset_mock()
@pytest.mark.asyncio
async def test_text_completion_model_parameter_validation(self, mock_openai_client):
"""Test that model parameters are correctly passed to OpenAI API"""
# Arrange
processor = MagicMock()
processor.model = "gpt-4"
processor.temperature = 0.8
processor.max_output = 2048
processor.openai = mock_openai_client
processor.generate_content = Processor.generate_content.__get__(processor, Processor)
# Act
await processor.generate_content("System prompt", "User prompt")
# Assert
call_args = mock_openai_client.chat.completions.create.call_args
assert call_args.kwargs['model'] == "gpt-4"
assert call_args.kwargs['temperature'] == 0.8
assert call_args.kwargs['max_tokens'] == 2048
assert call_args.kwargs['top_p'] == 1
assert call_args.kwargs['frequency_penalty'] == 0
assert call_args.kwargs['presence_penalty'] == 0
@pytest.mark.asyncio
@pytest.mark.slow
async def test_text_completion_performance_timing(self, text_completion_processor, mock_openai_client):
"""Test performance timing for text completion"""
# Arrange
text_completion_processor.openai = mock_openai_client
# Act
import time
start_time = time.time()
result = await text_completion_processor.generate_content("System", "Prompt")
end_time = time.time()
execution_time = end_time - start_time
# Assert
assert isinstance(result, LlmResult)
assert execution_time < 1.0 # Should complete quickly with mocked API
mock_openai_client.chat.completions.create.assert_called_once()
@pytest.mark.asyncio
async def test_text_completion_response_content_extraction(self, text_completion_processor, mock_openai_client):
"""Test proper extraction of response content from OpenAI API"""
# Arrange
test_responses = [
"This is a simple response.",
"This is a multi-line response.\nWith multiple lines.\nAnd more content.",
"Response with special characters: @#$%^&*()_+-=[]{}|;':\",./<>?",
"" # Empty response
]
for test_content in test_responses:
# Update mock response
usage = CompletionUsage(prompt_tokens=10, completion_tokens=20, total_tokens=30)
message = ChatCompletionMessage(role="assistant", content=test_content)
choice = Choice(index=0, message=message, finish_reason="stop")
completion = ChatCompletion(
id="chatcmpl-test123",
choices=[choice],
created=1234567890,
model="gpt-3.5-turbo",
object="chat.completion",
usage=usage
)
mock_openai_client.chat.completions.create.return_value = completion
text_completion_processor.openai = mock_openai_client
# Act
result = await text_completion_processor.generate_content("System", "Prompt")
# Assert
assert result.text == test_content
assert result.in_token == 10
assert result.out_token == 20
assert result.model == "gpt-3.5-turbo"
# Reset mock for next iteration
mock_openai_client.reset_mock()