Spaces:
Running
Running
| """Tests for the ledger adapters.""" | |
| import pytest | |
| from fastmcp.ledger.adapter import HyperledgerAdapter, OmniSealAdapter, StubAdapter | |
| class TestHyperledgerAdapter: | |
| """Test the HyperledgerAdapter class.""" | |
| def adapter(self): | |
| """Create a Hyperledger adapter for testing.""" | |
| return HyperledgerAdapter( | |
| network_config="test_config.json", | |
| channel_name="test-channel", | |
| chaincode_name="test-chaincode" | |
| ) | |
| async def test_adapter_initialization(self, adapter): | |
| """Test adapter initialization.""" | |
| assert adapter.network_config == "test_config.json" | |
| assert adapter.channel_name == "test-channel" | |
| assert adapter.chaincode_name == "test-chaincode" | |
| assert adapter.peer_endpoint == "localhost:7051" | |
| assert adapter.orderer_endpoint == "localhost:7050" | |
| async def test_submit_block(self, adapter): | |
| """Test block submission.""" | |
| block_data = { | |
| "block_number": 1, | |
| "merkle_root": "test_merkle_root", | |
| "entry_count": 5, | |
| "entries": ["entry1", "entry2", "entry3", "entry4", "entry5"] | |
| } | |
| tx_id = await adapter.submit_block(block_data) | |
| assert tx_id is not None | |
| assert len(tx_id) == 64 # SHA-256 hex length | |
| async def test_verify_block(self, adapter): | |
| """Test block verification.""" | |
| block_id = "test_block_1" | |
| result = await adapter.verify_block(block_id) | |
| # Stub implementation always returns True | |
| assert result is True | |
| async def test_get_block_proof(self, adapter): | |
| """Test getting block proof.""" | |
| block_id = "test_block_1" | |
| proof = await adapter.get_block_proof(block_id) | |
| assert proof is not None | |
| assert proof["block_id"] == block_id | |
| assert "block_hash" in proof | |
| assert "block_number" in proof | |
| assert "timestamp" in proof | |
| assert "proof_type" in proof | |
| assert proof["proof_type"] == "hyperledger_fabric" | |
| assert "signatures" in proof | |
| assert "merkle_root" in proof | |
| class TestOmniSealAdapter: | |
| """Test the OmniSealAdapter class.""" | |
| def adapter(self): | |
| """Create an OmniSeal adapter for testing.""" | |
| return OmniSealAdapter( | |
| api_endpoint="https://api.test.omniseal.com", | |
| api_key="test_api_key", | |
| network_id="testnet" | |
| ) | |
| async def test_adapter_initialization(self, adapter): | |
| """Test adapter initialization.""" | |
| assert adapter.api_endpoint == "https://api.test.omniseal.com" | |
| assert adapter.api_key == "test_api_key" | |
| assert adapter.network_id == "testnet" | |
| async def test_submit_block(self, adapter): | |
| """Test block submission.""" | |
| block_data = { | |
| "block_number": 1, | |
| "merkle_root": "test_merkle_root", | |
| "entry_count": 3, | |
| "entries": ["entry1", "entry2", "entry3"] | |
| } | |
| tx_id = await adapter.submit_block(block_data) | |
| assert tx_id is not None | |
| assert len(tx_id) == 64 # SHA-256 hex length | |
| async def test_verify_block(self, adapter): | |
| """Test block verification.""" | |
| block_id = "test_block_1" | |
| result = await adapter.verify_block(block_id) | |
| # Stub implementation always returns True | |
| assert result is True | |
| async def test_get_block_proof(self, adapter): | |
| """Test getting block proof.""" | |
| block_id = "test_block_1" | |
| proof = await adapter.get_block_proof(block_id) | |
| assert proof is not None | |
| assert proof["block_id"] == block_id | |
| assert "block_hash" in proof | |
| assert "block_number" in proof | |
| assert "timestamp" in proof | |
| assert "proof_type" in proof | |
| assert proof["proof_type"] == "omniseal" | |
| assert "network_id" in proof | |
| assert proof["network_id"] == "testnet" | |
| assert "merkle_root" in proof | |
| class TestStubAdapter: | |
| """Test the StubAdapter class.""" | |
| def adapter(self): | |
| """Create a stub adapter for testing.""" | |
| return StubAdapter() | |
| async def test_adapter_initialization(self, adapter): | |
| """Test adapter initialization.""" | |
| assert adapter.submitted_blocks == {} | |
| assert adapter.block_proofs == {} | |
| async def test_submit_block(self, adapter): | |
| """Test block submission.""" | |
| block_data = { | |
| "block_number": 1, | |
| "merkle_root": "test_merkle_root", | |
| "entry_count": 2, | |
| "entries": ["entry1", "entry2"] | |
| } | |
| block_id = await adapter.submit_block(block_data) | |
| assert block_id == "stub_block_1" | |
| assert block_id in adapter.submitted_blocks | |
| assert adapter.submitted_blocks[block_id] == block_data | |
| async def test_submit_multiple_blocks(self, adapter): | |
| """Test submitting multiple blocks.""" | |
| for i in range(3): | |
| block_data = { | |
| "block_number": i + 1, | |
| "merkle_root": f"merkle_root_{i}", | |
| "entry_count": 1, | |
| "entries": [f"entry_{i}"] | |
| } | |
| block_id = await adapter.submit_block(block_data) | |
| assert block_id == f"stub_block_{i + 1}" | |
| assert len(adapter.submitted_blocks) == 3 | |
| async def test_verify_block(self, adapter): | |
| """Test block verification.""" | |
| # Submit a block first | |
| block_data = {"block_number": 1, "merkle_root": "test"} | |
| block_id = await adapter.submit_block(block_data) | |
| # Verify existing block | |
| result = await adapter.verify_block(block_id) | |
| assert result is True | |
| # Verify non-existent block | |
| result = await adapter.verify_block("nonexistent_block") | |
| assert result is False | |
| async def test_get_block_proof(self, adapter): | |
| """Test getting block proof.""" | |
| # Submit a block first | |
| block_data = { | |
| "block_number": 1, | |
| "merkle_root": "test_merkle_root", | |
| "entry_count": 2 | |
| } | |
| block_id = await adapter.submit_block(block_data) | |
| # Get proof for existing block | |
| proof = await adapter.get_block_proof(block_id) | |
| assert proof is not None | |
| assert proof["block_id"] == block_id | |
| assert proof["block_hash"] == f"stub_hash_{block_id}" | |
| assert proof["block_number"] == 1 | |
| assert "timestamp" in proof | |
| assert proof["proof_type"] == "stub" | |
| assert proof["merkle_root"] == "test_merkle_root" | |
| # Get proof for non-existent block | |
| proof = await adapter.get_block_proof("nonexistent_block") | |
| assert proof is None | |
| async def test_adapter_state_persistence(self, adapter): | |
| """Test that adapter maintains state across operations.""" | |
| # Submit a block | |
| block_data = {"block_number": 1, "merkle_root": "test"} | |
| block_id = await adapter.submit_block(block_data) | |
| # Verify it exists | |
| assert await adapter.verify_block(block_id) is True | |
| # Get proof | |
| proof = await adapter.get_block_proof(block_id) | |
| assert proof is not None | |
| # Submit another block | |
| block_data2 = {"block_number": 2, "merkle_root": "test2"} | |
| block_id2 = await adapter.submit_block(block_data2) | |
| # Both blocks should exist | |
| assert await adapter.verify_block(block_id) is True | |
| assert await adapter.verify_block(block_id2) is True | |
| # Both proofs should be available | |
| proof1 = await adapter.get_block_proof(block_id) | |
| proof2 = await adapter.get_block_proof(block_id2) | |
| assert proof1 is not None | |
| assert proof2 is not None | |
| assert proof1["block_id"] != proof2["block_id"] | |