tests/ledger/test_ledger_core.py

"""Tests for the core ledger functionality."""

import pytest
from datetime import datetime
from uuid import uuid4

from fastmcp.ledger import ProvenanceLedger, LedgerEvent, EventType, LedgerEntry, LedgerBlock


class TestProvenanceLedger:
    """Test the ProvenanceLedger class."""
    
    @pytest.fixture
    def ledger(self):
        """Create a ledger instance for testing."""
        return ProvenanceLedger("sqlite:///:memory:")
    
    @pytest.fixture
    def sample_event(self):
        """Create a sample event for testing."""
        return LedgerEvent(
            event_type=EventType.TOOL_CALL,
            actor_id="user123",
            resource_id="resource456",
            action="execute_tool",
            metadata={"tool_name": "test_tool", "parameters": {"x": 1, "y": 2}},
            data_hash="sha256_hash_of_data"
        )
    
    def test_ledger_initialization(self, ledger):
        """Test ledger initialization."""
        assert ledger is not None
        assert ledger._current_sequence == 1
        assert ledger._current_block is None
        assert ledger._block_size == 100
    
    def test_append_event(self, ledger, sample_event):
        """Test appending an event to the ledger."""
        entry = ledger.append_event(sample_event)
        
        assert entry is not None
        assert entry.sequence_number == 1
        assert entry.previous_hash is None
        assert entry.entry_hash is not None
        assert entry.block_id is not None
        assert entry.is_verified is True
        
        # Verify the event data
        event = entry.get_event()
        assert event.event_type == sample_event.event_type
        assert event.actor_id == sample_event.actor_id
        assert event.action == sample_event.action
    
    def test_hash_chaining(self, ledger, sample_event):
        """Test that entries are properly hash-chained."""
        # Append first event
        entry1 = ledger.append_event(sample_event)
        
        # Append second event
        entry2 = ledger.append_event(sample_event)
        
        # Verify hash chaining
        assert entry2.previous_hash == entry1.entry_hash
        assert entry2.sequence_number == entry1.sequence_number + 1
    
    def test_entry_integrity_verification(self, ledger, sample_event):
        """Test entry integrity verification."""
        entry = ledger.append_event(sample_event)
        
        # Verify entry integrity
        assert entry.verify_integrity() is True
        
        # Tamper with the entry
        original_hash = entry.entry_hash
        entry.entry_hash = "tampered_hash"
        
        # Verify integrity fails
        assert entry.verify_integrity() is False
        
        # Restore original hash
        entry.entry_hash = original_hash
        assert entry.verify_integrity() is True
    
    def test_get_entry(self, ledger, sample_event):
        """Test retrieving an entry by sequence number."""
        entry = ledger.append_event(sample_event)
        
        retrieved_entry = ledger.get_entry(entry.sequence_number)
        assert retrieved_entry is not None
        assert retrieved_entry.id == entry.id
        assert retrieved_entry.sequence_number == entry.sequence_number
    
    def test_get_nonexistent_entry(self, ledger):
        """Test retrieving a non-existent entry."""
        entry = ledger.get_entry(999)
        assert entry is None
    
    def test_block_creation_and_sealing(self, ledger, sample_event):
        """Test block creation and sealing."""
        # Append events to fill a block
        entries = []
        for i in range(5):  # Use a smaller number for testing
            event = LedgerEvent(
                event_type=EventType.TOOL_CALL,
                actor_id=f"user{i}",
                action=f"action_{i}",
                metadata={"index": i}
            )
            entry = ledger.append_event(event)
            entries.append(entry)
        
        # Manually seal the block (normally done when block_size is reached)
        from sqlmodel import Session
        with Session(ledger.engine) as session:
            ledger._seal_block(session)
        
        # Get the block
        block = ledger.get_block(1)
        assert block is not None
        assert block.block_number == 1
        assert block.entry_count == 5
        assert block.merkle_root is not None
        assert block.sealed_at is not None
        assert block.is_verified is True
    
    def test_block_integrity_verification(self, ledger, sample_event):
        """Test block integrity verification."""
        # Create a block with multiple entries
        entries = []
        for i in range(3):
            event = LedgerEvent(
                event_type=EventType.TOOL_CALL,
                actor_id=f"user{i}",
                action=f"action_{i}"
            )
            entry = ledger.append_event(event)
            entries.append(entry)
        
        # Seal the block
        from sqlmodel import Session
        with Session(ledger.engine) as session:
            ledger._seal_block(session)
        
        # Verify block integrity
        assert ledger.verify_block_integrity(1) is True
        
        # Tamper with an entry
        entry = ledger.get_entry(1)
        original_hash = entry.entry_hash
        entry.entry_hash = "tampered_hash"
        
        # Update the entry in the database
        with Session(ledger.engine) as session:
            session.add(entry)
            session.commit()
        
        # Verify block integrity fails
        assert ledger.verify_block_integrity(1) is False
    
    def test_chain_integrity_verification(self, ledger, sample_event):
        """Test chain integrity verification."""
        # Append multiple events
        for i in range(5):
            event = LedgerEvent(
                event_type=EventType.TOOL_CALL,
                actor_id=f"user{i}",
                action=f"action_{i}"
            )
            ledger.append_event(event)
        
        # Verify entire chain
        assert ledger.verify_chain_integrity() is True
        
        # Verify partial chain
        assert ledger.verify_chain_integrity(start_sequence=2, end_sequence=4) is True
    
    def test_chain_integrity_with_tampering(self, ledger, sample_event):
        """Test chain integrity verification with tampered entries."""
        # Append multiple events
        entries = []
        for i in range(3):
            event = LedgerEvent(
                event_type=EventType.TOOL_CALL,
                actor_id=f"user{i}",
                action=f"action_{i}"
            )
            entry = ledger.append_event(event)
            entries.append(entry)
        
        # Verify chain is intact
        assert ledger.verify_chain_integrity() is True
        
        # Tamper with middle entry
        middle_entry = entries[1]
        original_hash = middle_entry.entry_hash
        middle_entry.entry_hash = "tampered_hash"
        
        # Update in database
        from sqlmodel import Session
        with Session(ledger.engine) as session:
            session.add(middle_entry)
            session.commit()
        
        # Verify chain integrity fails
        assert ledger.verify_chain_integrity() is False
    
    def test_ledger_statistics(self, ledger, sample_event):
        """Test ledger statistics."""
        # Initially empty
        stats = ledger.get_ledger_statistics()
        assert stats["total_entries"] == 0
        assert stats["total_blocks"] == 0
        assert stats["current_sequence"] == 0
        
        # Add some entries
        for i in range(3):
            event = LedgerEvent(
                event_type=EventType.TOOL_CALL,
                actor_id=f"user{i}",
                action=f"action_{i}"
            )
            ledger.append_event(event)
        
        # Check updated statistics
        stats = ledger.get_ledger_statistics()
        assert stats["total_entries"] == 3
        assert stats["total_blocks"] == 1
        assert stats["current_sequence"] == 3
    
    def test_different_event_types(self, ledger):
        """Test appending different types of events."""
        event_types = [
            EventType.TOOL_CALL,
            EventType.POLICY_DECISION,
            EventType.DATA_FLOW,
            EventType.CONTRACT_ACTION,
            EventType.AUTHENTICATION,
            EventType.AUTHORIZATION,
            EventType.SYSTEM_EVENT
        ]
        
        for event_type in event_types:
            event = LedgerEvent(
                event_type=event_type,
                actor_id="test_actor",
                action="test_action"
            )
            entry = ledger.append_event(event)
            
            retrieved_event = entry.get_event()
            assert retrieved_event.event_type == event_type
    
    def test_event_with_metadata(self, ledger):
        """Test events with complex metadata."""
        metadata = {
            "nested": {"key": "value"},
            "list": [1, 2, 3],
            "boolean": True,
            "null": None
        }
        
        event = LedgerEvent(
            event_type=EventType.TOOL_CALL,
            actor_id="test_actor",
            action="test_action",
            metadata=metadata
        )
        
        entry = ledger.append_event(event)
        retrieved_event = entry.get_event()
        
        assert retrieved_event.metadata == metadata
    
    def test_event_content_hash(self, ledger):
        """Test event content hash generation."""
        event = LedgerEvent(
            event_type=EventType.TOOL_CALL,
            actor_id="test_actor",
            action="test_action",
            metadata={"key": "value"}
        )
        
        content_hash = event.get_content_hash()
        assert content_hash is not None
        assert len(content_hash) == 64  # SHA-256 hex length
        
        # Same event should produce same hash
        event2 = LedgerEvent(
            event_type=EventType.TOOL_CALL,
            actor_id="test_actor",
            action="test_action",
            metadata={"key": "value"}
        )
        
        assert event2.get_content_hash() == content_hash
        
        # Different event should produce different hash
        event3 = LedgerEvent(
            event_type=EventType.TOOL_CALL,
            actor_id="test_actor",
            action="different_action",
            metadata={"key": "value"}
        )
        
        assert event3.get_content_hash() != content_hash