Spaces:
Running
Running
File size: 10,550 Bytes
7d7a6cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 | """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
|