# Copyright 2024 Heinrich Krupp
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Tests for Hybrid backend graph method delegation (v9.2.0).
Validates that HybridStorage correctly delegates graph queries to primary storage:
- get_relationship_type_distribution()
- get_graph_visualization_data()
"""
import pytest
import pytest_asyncio
import tempfile
import os
from unittest.mock import AsyncMock, patch, MagicMock
from mcp_memory_service.storage.hybrid import HybridMemoryStorage
from mcp_memory_service.storage.sqlite_vec import SqliteVecMemoryStorage
from mcp_memory_service.storage.graph import GraphStorage
from mcp_memory_service.models.memory import Memory
from mcp_memory_service.utils.hashing import generate_content_hash
@pytest.fixture
def temp_db():
"""Create a temporary database for testing."""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = os.path.join(tmpdir, "test_hybrid_graph.db")
yield db_path
@pytest_asyncio.fixture
async def primary_storage(temp_db):
"""Create and initialize primary SQLite storage."""
storage = SqliteVecMemoryStorage(temp_db)
await storage.initialize()
yield storage
storage.close()
@pytest_asyncio.fixture
async def hybrid_storage_with_mock_cloudflare(primary_storage):
"""Create HybridStorage with real primary and mocked Cloudflare secondary.
This allows testing delegation without requiring Cloudflare credentials.
"""
# Mock Cloudflare storage
mock_cloudflare = AsyncMock()
mock_cloudflare.initialize = AsyncMock()
mock_cloudflare.store = AsyncMock()
mock_cloudflare.close = AsyncMock()
# Create hybrid storage
hybrid = HybridMemoryStorage(primary_storage.db_path)
hybrid.primary = primary_storage
hybrid.secondary = mock_cloudflare
yield hybrid
@pytest_asyncio.fixture
async def hybrid_with_graph_data(hybrid_storage_with_mock_cloudflare):
"""Populate hybrid storage with graph data for testing."""
storage = hybrid_storage_with_mock_cloudflare
primary = storage.primary
# Create memories
memories = []
for i in range(6):
content = f"Memory {i}"
memory = Memory(
content=content,
content_hash=generate_content_hash(content),
tags=["test"],
memory_type="note"
)
success, message = await primary.store(memory)
assert success, f"Failed to store memory: {message}"
memories.append(memory.content_hash)
# Create graph with typed relationships
graph = GraphStorage(primary.db_path)
await graph._get_connection()
# All 6 relationship types
await graph.store_association(
memories[0], memories[1], 0.8, ["semantic"], relationship_type="related"
)
await graph.store_association(
memories[1], memories[2], 0.9, ["causal"], relationship_type="causes"
)
await graph.store_association(
memories[2], memories[3], 0.85, ["resolution"], relationship_type="fixes"
)
await graph.store_association(
memories[3], memories[4], 0.7, ["semantic"], relationship_type="contradicts"
)
await graph.store_association(
memories[4], memories[5], 0.75, ["semantic"], relationship_type="supports"
)
await graph.store_association(
memories[5], memories[0], 0.65, ["temporal"], relationship_type="follows"
)
yield storage
# Tests for get_relationship_type_distribution() delegation
@pytest.mark.asyncio
async def test_hybrid_delegates_relationship_type_distribution(hybrid_storage_with_mock_cloudflare):
"""Test that hybrid storage delegates to primary storage.
Validates:
- Method exists on HybridStorage
- Calls primary.get_relationship_type_distribution()
- Returns result from primary
"""
storage = hybrid_storage_with_mock_cloudflare
# Call method
distribution = await storage.get_relationship_type_distribution()
# Should return dict (empty if no data)
assert isinstance(distribution, dict)
@pytest.mark.asyncio
async def test_hybrid_relationship_distribution_returns_primary_data(hybrid_with_graph_data):
"""Test that distribution data comes from primary storage, not Cloudflare.
Validates:
- Results match primary storage state
- Secondary (Cloudflare) is not queried for graph data
"""
storage = hybrid_with_graph_data
distribution = await storage.get_relationship_type_distribution()
# Should have all 6 types plus bidirectional edges
assert "related" in distribution
assert "causes" in distribution
assert "fixes" in distribution
assert "contradicts" in distribution
assert "supports" in distribution
assert "follows" in distribution
# Verify counts (symmetric relationships = 2 edges)
assert distribution["related"] == 2 # Bidirectional
assert distribution["contradicts"] == 2 # Bidirectional
assert distribution["causes"] == 1 # Directed
assert distribution["fixes"] == 1 # Directed
assert distribution["supports"] == 1 # Directed
assert distribution["follows"] == 1 # Directed
@pytest.mark.xfail(reason="Pre-existing bug: Permission denied accessing '/fake' path")
@pytest.mark.asyncio
async def test_hybrid_relationship_distribution_with_mock_primary():
"""Test delegation using fully mocked primary storage.
Validates:
- Hybrid correctly calls primary method
- Returns mocked result without modification
"""
# Mock primary storage
mock_primary = AsyncMock()
mock_distribution = {"related": 10, "causes": 5, "fixes": 3}
mock_primary.get_relationship_type_distribution = AsyncMock(return_value=mock_distribution)
# Mock secondary
mock_secondary = AsyncMock()
# Create hybrid with mocks
hybrid = HybridMemoryStorage("/fake/path")
hybrid.primary = mock_primary
hybrid.secondary = mock_secondary
# Call method
result = await hybrid.get_relationship_type_distribution()
# Verify delegation
mock_primary.get_relationship_type_distribution.assert_called_once()
assert result == mock_distribution
@pytest.mark.asyncio
async def test_hybrid_relationship_distribution_empty_graph(hybrid_storage_with_mock_cloudflare):
"""Test that empty graph returns empty dict through hybrid storage."""
storage = hybrid_storage_with_mock_cloudflare
distribution = await storage.get_relationship_type_distribution()
assert isinstance(distribution, dict)
assert len(distribution) == 0
# Tests for get_graph_visualization_data() delegation
@pytest.mark.asyncio
async def test_hybrid_delegates_graph_visualization_data(hybrid_storage_with_mock_cloudflare):
"""Test that hybrid storage delegates visualization query to primary.
Validates:
- Method exists on HybridStorage
- Accepts limit and min_connections parameters
- Returns dict with nodes/edges/meta structure
"""
storage = hybrid_storage_with_mock_cloudflare
data = await storage.get_graph_visualization_data(limit=50, min_connections=1)
assert isinstance(data, dict)
assert "nodes" in data
assert "edges" in data
assert "meta" in data
@pytest.mark.asyncio
async def test_hybrid_graph_visualization_returns_primary_data(hybrid_with_graph_data):
"""Test that visualization data comes from primary storage.
Validates:
- Nodes and edges reflect primary storage state
- Relationship types are correctly included
"""
storage = hybrid_with_graph_data
data = await storage.get_graph_visualization_data(limit=10, min_connections=1)
# Should have nodes and edges
assert len(data["nodes"]) > 0
assert len(data["edges"]) > 0
# Verify relationship types in edges
rel_types = [edge["relationship_type"] for edge in data["edges"]]
assert "related" in rel_types
assert "causes" in rel_types
assert "fixes" in rel_types
@pytest.mark.xfail(reason="Pre-existing bug: Permission denied accessing '/fake' path")
@pytest.mark.asyncio
async def test_hybrid_graph_visualization_with_mock_primary():
"""Test delegation using fully mocked primary storage.
Validates:
- Parameters are passed through correctly
- Returns mocked result
"""
# Mock primary storage
mock_primary = AsyncMock()
mock_data = {
"nodes": [{"id": "hash1", "type": "note", "content": "Test", "connections": 2}],
"edges": [{"source": "hash1", "target": "hash2", "relationship_type": "related"}],
"meta": {"total_nodes": 1, "total_edges": 1, "limit": 100, "min_connections": 1}
}
mock_primary.get_graph_visualization_data = AsyncMock(return_value=mock_data)
# Mock secondary
mock_secondary = AsyncMock()
# Create hybrid with mocks
hybrid = HybridMemoryStorage("/fake/path")
hybrid.primary = mock_primary
hybrid.secondary = mock_secondary
# Call method with parameters
result = await hybrid.get_graph_visualization_data(limit=100, min_connections=1)
# Verify delegation with correct parameters
mock_primary.get_graph_visualization_data.assert_called_once_with(100, 1)
assert result == mock_data
@pytest.mark.asyncio
async def test_hybrid_graph_visualization_limit_parameter_delegation(hybrid_with_graph_data):
"""Test that limit parameter is correctly passed to primary storage."""
storage = hybrid_with_graph_data
# Test with different limits
data_small = await storage.get_graph_visualization_data(limit=2)
data_large = await storage.get_graph_visualization_data(limit=10)
# Smaller limit should return fewer or equal nodes
assert len(data_small["nodes"]) <= 2
assert len(data_large["nodes"]) <= 10
assert len(data_small["nodes"]) <= len(data_large["nodes"])
@pytest.mark.asyncio
async def test_hybrid_graph_visualization_min_connections_delegation(hybrid_with_graph_data):
"""Test that min_connections parameter is correctly passed to primary."""
storage = hybrid_with_graph_data
# Create hub with many connections
primary = storage.primary
graph = GraphStorage(primary.db_path)
content = "Hub"
hub_memory = Memory(
content=content,
content_hash=generate_content_hash(content),
tags=["test"]
)
success, message = await primary.store(hub_memory)
assert success, f"Failed to store hub memory: {message}"
hub = hub_memory.content_hash
for i in range(5):
spoke_content = f"Spoke {i}"
spoke_memory = Memory(
content=spoke_content,
content_hash=generate_content_hash(spoke_content),
tags=["test"]
)
success, message = await primary.store(spoke_memory)
assert success, f"Failed to store spoke memory: {message}"
spoke = spoke_memory.content_hash
await graph.store_association(hub, spoke, 0.7, ["semantic"], relationship_type="related")
# Test with high min_connections (should filter out low-connectivity nodes)
data = await storage.get_graph_visualization_data(min_connections=3)
# Only highly connected nodes should remain
for node in data["nodes"]:
assert node["connections"] >= 3
@pytest.mark.asyncio
async def test_hybrid_graph_visualization_empty_graph(hybrid_storage_with_mock_cloudflare):
"""Test that empty graph returns empty structure through hybrid storage."""
storage = hybrid_storage_with_mock_cloudflare
data = await storage.get_graph_visualization_data()
assert data["nodes"] == []
assert data["edges"] == []
assert data["meta"]["total_nodes"] == 0
assert data["meta"]["total_edges"] == 0
@pytest.mark.asyncio
async def test_hybrid_does_not_query_cloudflare_for_graph_data(hybrid_with_graph_data):
"""Test that Cloudflare secondary storage is NOT queried for graph operations.
Validates:
- Only primary storage handles graph queries
- Secondary (Cloudflare) is never called for graph methods
"""
storage = hybrid_with_graph_data
mock_secondary = storage.secondary
# Call both graph methods
await storage.get_relationship_type_distribution()
await storage.get_graph_visualization_data(limit=10, min_connections=1)
# Verify Cloudflare was not queried
# (No graph-related methods should have been called on secondary)
# Check that no graph-related methods exist on the mock
assert not hasattr(mock_secondary, 'get_relationship_type_distribution') or \
not mock_secondary.get_relationship_type_distribution.called
assert not hasattr(mock_secondary, 'get_graph_visualization_data') or \
not mock_secondary.get_graph_visualization_data.called
# Integration tests
@pytest.mark.asyncio
async def test_hybrid_graph_methods_match_primary_directly(hybrid_with_graph_data):
"""Test that calling through hybrid returns same results as calling primary directly.
Validates:
- Hybrid delegation is transparent
- Results are identical to direct primary calls
"""
storage = hybrid_with_graph_data
primary = storage.primary
# Get results through hybrid
hybrid_distribution = await storage.get_relationship_type_distribution()
hybrid_viz = await storage.get_graph_visualization_data(limit=10, min_connections=1)
# Get results directly from primary
primary_distribution = await primary.get_relationship_type_distribution()
primary_viz = await primary.get_graph_visualization_data(limit=10, min_connections=1)
# Should be identical
assert hybrid_distribution == primary_distribution
assert hybrid_viz["nodes"] == primary_viz["nodes"]
assert hybrid_viz["edges"] == primary_viz["edges"]
assert hybrid_viz["meta"] == primary_viz["meta"]
@pytest.mark.asyncio
async def test_hybrid_graph_methods_consistency_across_calls(hybrid_with_graph_data):
"""Test that repeated calls through hybrid return consistent results.
Validates:
- No caching issues
- Consistent behavior
"""
storage = hybrid_with_graph_data
# Call multiple times
dist1 = await storage.get_relationship_type_distribution()
dist2 = await storage.get_relationship_type_distribution()
viz1 = await storage.get_graph_visualization_data(limit=10)
viz2 = await storage.get_graph_visualization_data(limit=10)
# Results should be identical
assert dist1 == dist2
assert viz1["nodes"] == viz2["nodes"]
assert viz1["edges"] == viz2["edges"]
@pytest.mark.xfail(reason="Pre-existing bug: NameError - 'HybridStorage' not defined (should be 'HybridMemoryStorage')")
@pytest.mark.asyncio
async def test_hybrid_graph_methods_error_handling():
"""Test that errors from primary storage are propagated correctly.
Validates:
- Exceptions from primary are not swallowed
- Error handling is transparent
"""
# Mock primary that raises exception
mock_primary = AsyncMock()
mock_primary.get_relationship_type_distribution = AsyncMock(
side_effect=Exception("Database error")
)
mock_primary.get_graph_visualization_data = AsyncMock(
side_effect=Exception("Query error")
)
mock_secondary = AsyncMock()
hybrid = HybridStorage("/fake/path")
hybrid.primary = mock_primary
hybrid.secondary = mock_secondary
# Exceptions should propagate
with pytest.raises(Exception, match="Database error"):
await hybrid.get_relationship_type_distribution()
with pytest.raises(Exception, match="Query error"):
await hybrid.get_graph_visualization_data()
