tests.py
1 import unittest 2 import sys 3 import os 4 from unittest.mock import MagicMock, patch 5 from decimal import Decimal 6 7 # Add parent directory to path to use local development code 8 # Must be first to override installed package 9 local_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) 10 if local_path not in sys.path: 11 sys.path.insert(0, local_path) 12 13 # Remove fx_sdk from cache if already imported (to force reload from local code) 14 if 'fx_sdk' in sys.modules: 15 del sys.modules['fx_sdk'] 16 if 'fx_sdk.client' in sys.modules: 17 del sys.modules['fx_sdk.client'] 18 if 'fx_sdk.constants' in sys.modules: 19 del sys.modules['fx_sdk.constants'] 20 if 'fx_sdk.exceptions' in sys.modules: 21 del sys.modules['fx_sdk.exceptions'] 22 if 'fx_sdk.utils' in sys.modules: 23 del sys.modules['fx_sdk.utils'] 24 25 from fx_sdk.client import ProtocolClient 26 from fx_sdk import utils 27 from fx_sdk.exceptions import ConfigurationError, TransactionFailedError, ContractCallError 28 29 class TestFXProtocolSDK(unittest.TestCase): 30 """Test suite for the f(x) Protocol Python SDK.""" 31 32 def setUp(self): 33 """Set up a client with mocked internals.""" 34 self.rpc_url = "http://localhost:8545" 35 self.private_key = "0x" + "1" * 64 36 37 # Patch Web3 inside the client module to avoid validation errors 38 with patch('fx_sdk.client.Web3') as MockWeb3: 39 self.mock_w3 = MockWeb3.return_value 40 self.mock_w3.is_connected.return_value = True 41 self.mock_w3.isConnected.return_value = True 42 43 self.mock_w3.eth = MagicMock() 44 self.mock_w3.eth.get_transaction_count.return_value = 0 45 self.mock_w3.eth.gas_price = 20000000000 46 self.mock_w3.eth.wait_for_transaction_receipt.return_value = MagicMock(status=1) 47 48 self.mock_w3.eth.account = MagicMock() 49 self.mock_w3.eth.account.sign_transaction.return_value = MagicMock( 50 rawTransaction=b"signed_tx_data" 51 ) 52 self.mock_w3.eth.send_raw_transaction.return_value = MagicMock( 53 hex=lambda: "0x" + "a" * 64 54 ) 55 56 self.client = ProtocolClient( 57 rpc_url=self.rpc_url, 58 private_key=self.private_key 59 ) 60 61 def test_unit_conversions(self): 62 """Test the utility conversion functions.""" 63 print("\n🧪 Testing Unit Conversions...") 64 wei = utils.decimal_to_wei(1.5, 18) 65 print(f" - 1.5 ETH -> {wei} Wei") 66 self.assertEqual(wei, 1500000000000000000) 67 68 dec = utils.wei_to_decimal(1500000000000000000, 18) 69 print(f" - {wei} Wei -> {dec} ETH") 70 self.assertEqual(dec, Decimal("1.5")) 71 72 def test_client_initialization(self): 73 """Test client setup and address parsing.""" 74 print("\n🧪 Testing Client Initialization...") 75 expected_addr = "0x19E7E376E7C213B7E7e7e46cc70A5dD086DAff2A" 76 print(f" - Derived Address: {self.client.address}") 77 self.assertEqual(self.client.address, expected_addr) 78 79 def test_read_balance(self): 80 """Test reading balance with mocked contract.""" 81 print("\n🧪 Testing Balance Fetching...") 82 mock_contract = MagicMock() 83 mock_contract.functions.balanceOf.return_value.call.return_value = 100 * 10**18 84 mock_contract.functions.decimals.return_value.call.return_value = 18 85 86 with patch.object(self.mock_w3.eth, 'contract', return_value=mock_contract): 87 balance = self.client.get_fxusd_balance() 88 print(f" - Mocked fxUSD Balance: {balance}") 89 self.assertEqual(balance, Decimal("100")) 90 91 def test_multi_output_parsing(self): 92 """Test methods that return dictionaries parsed from contract tuples.""" 93 print("\n🧪 Testing Multi-Output Tuple Parsing...") 94 95 # 1. Test Pool Manager Info 96 mock_pool_manager = MagicMock() 97 # Returns (collateralCapacity, collateralBalance, debtCapacity, debtBalance) 98 mock_pool_manager.functions.getPoolInfo.return_value.call.return_value = [ 99 1000 * 10**18, 500 * 10**18, 2000 * 10**18, 100 * 10**18 100 ] 101 102 with patch.object(self.client, '_get_contract', return_value=mock_pool_manager): 103 info = self.client.get_pool_manager_info("0x" + "5" * 40) 104 print(f" - Pool Info Parsed: {info}") 105 self.assertEqual(info['collateral_capacity'], Decimal("1000")) 106 self.assertEqual(info['debt_balance'], Decimal("100")) 107 108 # 2. Test veFXN Locked Info 109 # Returns (amount, end) 110 self.client.vefxn.functions.locked.return_value.call.return_value = [50 * 10**18, 1734652800] 111 lock_info = self.client.get_vefxn_locked_info() 112 print(f" - veFXN Lock Info Parsed: {lock_info}") 113 self.assertEqual(lock_info['amount'], Decimal("50")) 114 self.assertEqual(lock_info['end'], 1734652800) 115 116 def test_aggregation_logic(self): 117 """Test the balance aggregator across multiple tokens.""" 118 print("\n🧪 Testing Balance Aggregator (get_all_balances)...") 119 120 # Mock get_token_balance to return varying amounts 121 with patch.object(self.client, 'get_token_balance') as mock_balance: 122 mock_balance.side_effect = lambda addr, acc: Decimal("10.0") 123 124 balances = self.client.get_all_balances() 125 print(f" - Aggregated {len(balances)} tokens") 126 self.assertIn("fxUSD", balances) 127 self.assertEqual(balances["fxUSD"], Decimal("10.0")) 128 129 def test_complex_swap_params(self): 130 """Test the swap method parameter handling.""" 131 print("\n🧪 Testing Complex Swap Parameter Passing...") 132 133 with patch.object(self.client, '_build_and_send_transaction', return_value="0xswaphash") as mock_send: 134 # MultiPathConverter.convert(token_in, amount_in, encoding, routes) 135 tx_hash = self.client.swap( 136 token_in="0x" + "a" * 40, 137 amount_in=1.5, 138 encoding=123, 139 routes=[1, 2, 3] 140 ) 141 print(f" - Swap Transaction Hash: {tx_hash}") 142 self.assertEqual(tx_hash, "0xswaphash") 143 mock_send.assert_called_once() 144 145 def test_mint_f_token_via_gateway(self): 146 """Test minting fToken via Gateway renaming.""" 147 print("\n🧪 Testing stETH Gateway (mint_f_token_via_gateway)...") 148 with patch.object(self.client, '_build_and_send_transaction', return_value="0xgatewayhash") as mock_send: 149 tx_hash = self.client.mint_f_token_via_gateway(1.0) 150 print(f" - Gateway Tx Hash: {tx_hash}") 151 self.assertEqual(tx_hash, "0xgatewayhash") 152 mock_send.assert_called_once() 153 154 def test_position_and_peg_discovery(self): 155 """Test V2 position and peg keeper info fetching.""" 156 print("\n🧪 Testing V2 Position & Peg Keeper Discovery...") 157 158 # Mock Position Info 159 mock_pm = MagicMock() 160 mock_pm.functions.getPosition.return_value.call.return_value = [self.client.address, 100 * 10**18, 50 * 10**18] 161 162 # Mock Peg Keeper 163 mock_pk = MagicMock() 164 mock_pk.functions.isActive.return_value.call.return_value = True 165 mock_pk.functions.debtCeiling.return_value.call.return_value = 1000000 * 10**18 166 mock_pk.functions.totalDebt.return_value.call.return_value = 500000 * 10**18 167 168 def mock_get_contract_side_effect(name, address): 169 if name == "pool_manager": return mock_pm 170 if name == "peg_keeper": return mock_pk 171 return MagicMock() 172 173 with patch.object(self.client, '_get_contract', side_effect=mock_get_contract_side_effect): 174 pos = self.client.get_position_info(123) 175 print(f" - Position 123: {pos}") 176 self.assertEqual(pos['collateral'], Decimal("100")) 177 178 peg = self.client.get_peg_keeper_info() 179 print(f" - Peg Keeper Status: {peg}") 180 self.assertTrue(peg['is_active']) 181 182 def test_write_transaction(self): 183 """Test building and sending a transaction.""" 184 print("\n🧪 Testing Transaction Lifecycle (Build/Sign/Send)...") 185 mock_func = MagicMock() 186 mock_func.estimate_gas.return_value = 200000 187 188 built_tx = {'gas': 200000, 'nonce': 0} 189 mock_func.build_transaction.return_value = built_tx 190 mock_func.buildTransaction.return_value = built_tx 191 192 tx_hash = self.client._build_and_send_transaction(mock_func) 193 print(f" - Generated Tx Hash: {tx_hash}") 194 self.assertTrue(tx_hash.startswith("0x")) 195 196 def test_rebalance_pool_deposit(self): 197 """Test rebalance pool deposit flow.""" 198 print("\n🧪 Testing Rebalance Pool Flow...") 199 with patch.object(self.client, '_build_and_send_transaction', return_value="0xhash") as mock_send: 200 tx_hash = self.client.deposit_to_rebalance_pool( 201 pool_address="0x" + "2" * 40, 202 amount=10.5 203 ) 204 print(f" - Rebalance Deposit Hash: {tx_hash}") 205 self.assertEqual(tx_hash, "0xhash") 206 mock_send.assert_called_once() 207 208 def test_configuration_error(self): 209 """Test that write operations fail without a private key.""" 210 print("\n🧪 Testing Safety Checks (Read-Only Mode)...") 211 with patch('fx_sdk.client.Web3') as MockWeb3: 212 MockWeb3.return_value.is_connected.return_value = True 213 MockWeb3.return_value.isConnected.return_value = True 214 215 ro_client = ProtocolClient(rpc_url=self.rpc_url) 216 with self.assertRaises(ConfigurationError): 217 print(" - Attempting write operation without key (expecting error)...") 218 ro_client._build_and_send_transaction(MagicMock()) 219 print(" - Caught expected ConfigurationError ✅") 220 221 if __name__ == "__main__": 222 print("\n" + "="*50) 223 print("🚀 STARTING f(x) PROTOCOL SDK PRODUCTION TEST SUITE") 224 print("="*50) 225 unittest.main()