MCP Memory Service

"""Tests for asymmetric relationship semantic correctness."""
import sys
from pathlib import Path
import pytest
import tempfile
import shutil
import os
import sqlite3
import importlib.util

# Load GraphStorage module directly without importing the package to avoid numpy issues
repo_root = Path(__file__).parent.parent.parent
graph_path = repo_root / "src" / "mcp_memory_service" / "storage" / "graph.py"
ontology_path = repo_root / "src" / "mcp_memory_service" / "models" / "ontology.py"

# Load ontology module first (dependency of graph)
spec = importlib.util.spec_from_file_location("ontology", ontology_path)
ontology_module = importlib.util.module_from_spec(spec)
sys.modules['mcp_memory_service.models.ontology'] = ontology_module
spec.loader.exec_module(ontology_module)

# Load graph module
spec = importlib.util.spec_from_file_location("graph", graph_path)
graph_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(graph_module)

GraphStorage = graph_module.GraphStorage


@pytest.fixture
def temp_graph_db():
    """Create a temporary database with graph table for testing."""
    # Create temporary directory
    temp_dir = tempfile.mkdtemp()
    db_path = os.path.join(temp_dir, "test_graph.db")

    # Initialize database with graph schema
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS memory_graph (
            source_hash TEXT NOT NULL,
            target_hash TEXT NOT NULL,
            similarity REAL NOT NULL,
            connection_types TEXT NOT NULL,
            metadata TEXT,
            created_at REAL NOT NULL,
            relationship_type TEXT DEFAULT 'related',
            PRIMARY KEY (source_hash, target_hash)
        )
    """)

    # Create indexes for performance
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_graph_source
        ON memory_graph(source_hash)
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_graph_target
        ON memory_graph(target_hash)
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_graph_relationship
        ON memory_graph(relationship_type)
    """)

    conn.commit()
    conn.close()

    yield db_path

    # Cleanup
    shutil.rmtree(temp_dir, ignore_errors=True)


@pytest.fixture
async def graph_storage(temp_graph_db):
    """Create GraphStorage instance with initialized database."""
    storage = GraphStorage(temp_graph_db)
    # Ensure connection is initialized
    await storage._get_connection()
    yield storage


class TestAsymmetricSemantics:
    """Verify semantic correctness of directional relationships."""

    @pytest.mark.asyncio
    async def test_causes_directionality(self, graph_storage):
        """A causes B should NOT imply B causes A."""
        await graph_storage.store_association(
            "decision_a", "error_b", 0.9, ["causal"],
            relationship_type="causes"
        )

        # Forward direction exists
        causes = await graph_storage.find_connected(
            "decision_a", relationship_type="causes", direction="outgoing"
        )
        assert "error_b" in [h for h, _ in causes], "decision_a should cause error_b"

        # Reverse direction does NOT exist
        reverse = await graph_storage.find_connected(
            "error_b", relationship_type="causes", direction="outgoing"
        )
        assert "decision_a" not in [h for h, _ in reverse], "error_b should NOT cause decision_a"

    @pytest.mark.asyncio
    async def test_fixes_directionality(self, graph_storage):
        """A fixes B should NOT imply B fixes A."""
        await graph_storage.store_association(
            "decision_fix", "error_bug", 0.95, ["remediation"],
            relationship_type="fixes"
        )

        # Forward direction exists
        fixes = await graph_storage.find_connected(
            "decision_fix", relationship_type="fixes", direction="outgoing"
        )
        assert "error_bug" in [h for h, _ in fixes], "decision_fix should fix error_bug"

        # Reverse does NOT exist
        reverse = await graph_storage.find_connected(
            "error_bug", relationship_type="fixes", direction="outgoing"
        )
        assert "decision_fix" not in [h for h, _ in reverse], "error_bug should NOT fix decision_fix"

    @pytest.mark.asyncio
    async def test_supports_directionality(self, graph_storage):
        """A supports B should NOT imply B supports A."""
        await graph_storage.store_association(
            "learning_a", "decision_b", 0.85, ["reinforcement"],
            relationship_type="supports"
        )

        # Forward direction exists
        supports = await graph_storage.find_connected(
            "learning_a", relationship_type="supports", direction="outgoing"
        )
        assert "decision_b" in [h for h, _ in supports], "learning_a should support decision_b"

        # Reverse does NOT exist
        reverse = await graph_storage.find_connected(
            "decision_b", relationship_type="supports", direction="outgoing"
        )
        assert "learning_a" not in [h for h, _ in reverse], "decision_b should NOT support learning_a"

    @pytest.mark.asyncio
    async def test_follows_directionality(self, graph_storage):
        """A follows B should NOT imply B follows A."""
        await graph_storage.store_association(
            "observation_a", "observation_b", 0.80, ["temporal"],
            relationship_type="follows"
        )

        # Forward direction exists
        follows = await graph_storage.find_connected(
            "observation_a", relationship_type="follows", direction="outgoing"
        )
        assert "observation_b" in [h for h, _ in follows], "observation_a should follow observation_b"

        # Reverse does NOT exist
        reverse = await graph_storage.find_connected(
            "observation_b", relationship_type="follows", direction="outgoing"
        )
        assert "observation_a" not in [h for h, _ in reverse], "observation_b should NOT follow observation_a"

    @pytest.mark.asyncio
    async def test_contradicts_bidirectional(self, graph_storage):
        """A contradicts B SHOULD imply B contradicts A (symmetric)."""
        await graph_storage.store_association(
            "learning_a", "learning_b", 0.85, ["semantic"],
            relationship_type="contradicts"
        )

        # Both directions should work
        from_a = await graph_storage.find_connected(
            "learning_a", relationship_type="contradicts"
        )
        from_b = await graph_storage.find_connected(
            "learning_b", relationship_type="contradicts"
        )

        assert "learning_b" in [h for h, _ in from_a], "learning_a should contradict learning_b"
        assert "learning_a" in [h for h, _ in from_b], "learning_b should contradict learning_a"

    @pytest.mark.asyncio
    async def test_related_bidirectional(self, graph_storage):
        """A related B SHOULD imply B related A (symmetric)."""
        await graph_storage.store_association(
            "observation_a", "observation_b", 0.75, ["semantic"],
            relationship_type="related"
        )

        # Both directions should work
        from_a = await graph_storage.find_connected(
            "observation_a", relationship_type="related"
        )
        from_b = await graph_storage.find_connected(
            "observation_b", relationship_type="related"
        )

        assert "observation_b" in [h for h, _ in from_a], "observation_a should be related to observation_b"
        assert "observation_a" in [h for h, _ in from_b], "observation_b should be related to observation_a"

    @pytest.mark.asyncio
    async def test_incoming_query_for_asymmetric(self, graph_storage):
        """Incoming queries should work for asymmetric relationships."""
        await graph_storage.store_association(
            "decision_a", "error_b", 0.90, ["causal"],
            relationship_type="causes"
        )

        # Can query incoming to find the cause
        incoming = await graph_storage.find_connected(
            "error_b", relationship_type="causes", direction="incoming"
        )
        assert "decision_a" in [h for h, _ in incoming], "Should find decision_a as incoming cause"

        # Reverse incoming query should be empty
        reverse_incoming = await graph_storage.find_connected(
            "decision_a", relationship_type="causes", direction="incoming"
        )
        assert "error_b" not in [h for h, _ in reverse_incoming], "error_b should not be an incoming cause"

    @pytest.mark.asyncio
    async def test_direction_both_with_asymmetric(self, graph_storage):
        """Direction='both' should work correctly with asymmetric relationships."""
        await graph_storage.store_association(
            "decision_a", "error_b", 0.90, ["causal"],
            relationship_type="causes"
        )

        # Query from decision_a with direction="both"
        from_a = await graph_storage.find_connected(
            "decision_a", relationship_type="causes", direction="both"
        )
        assert "error_b" in [h for h, _ in from_a], "Should find error_b from decision_a"

        # Query from error_b with direction="both"
        from_b = await graph_storage.find_connected(
            "error_b", relationship_type="causes", direction="both"
        )
        assert "decision_a" in [h for h, _ in from_b], "Should find decision_a from error_b"