From e103c97be0fd59fa9ab6ef0472a1e83d92acf522 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Fri, 4 Apr 2025 00:01:48 +0900 Subject: [PATCH 1/5] Added setIpMetadata and respective integration test --- .../resources/IPAccount.py | 36 ++++++++++- .../test_integration_ip_account.py | 63 ++++++++++++------- 2 files changed, 73 insertions(+), 26 deletions(-) diff --git a/src/story_protocol_python_sdk/resources/IPAccount.py b/src/story_protocol_python_sdk/resources/IPAccount.py index 2179a60..356211a 100644 --- a/src/story_protocol_python_sdk/resources/IPAccount.py +++ b/src/story_protocol_python_sdk/resources/IPAccount.py @@ -1,11 +1,11 @@ """Module for handling IP Account operations and transactions.""" from web3 import Web3 -from web3.exceptions import InvalidAddress from story_protocol_python_sdk.abi.IPAccountImpl.IPAccountImpl_client import IPAccountImplClient from story_protocol_python_sdk.abi.IPAssetRegistry.IPAssetRegistry_client import IPAssetRegistryClient from story_protocol_python_sdk.abi.AccessController.AccessController_client import AccessControllerClient +from story_protocol_python_sdk.abi.CoreMetadataModule.CoreMetadataModule_client import CoreMetadataModuleClient from story_protocol_python_sdk.utils.transaction_utils import build_and_send_transaction @@ -25,6 +25,7 @@ def __init__(self, web3: Web3, account, chain_id: int): self.ip_asset_registry_client = IPAssetRegistryClient(web3) self.access_controller_client = AccessControllerClient(web3) self.ip_account_client = IPAccountImplClient(web3) + self.core_metadata_module_client = CoreMetadataModuleClient(web3) def getToken(self, ip_id: str) -> dict: """Retrieve token information associated with an IP account. @@ -153,4 +154,35 @@ def owner(self, ip_id: str) -> str: except ValueError: # Catch ValueError from to_checksum_address raise ValueError(f"Invalid IP id address: {ip_id}") except Exception as e: - raise e \ No newline at end of file + raise e + + def setIpMetadata(self, ip_id: str, metadata_uri: str, metadata_hash: str, tx_options: dict = None) -> dict: + """Sets the metadataURI for an IP asset. + + :param ip_id str: The IP ID to set metadata for. + :param metadata_uri str: The metadata URI to set. + :param metadata_hash str: The metadata hash. + :param tx_options dict: [Optional] The transaction options. + :returns dict: A dictionary with the transaction hash. + :raises ValueError: If the IP ID is invalid or not registered. + """ + try: + if not self._is_registered(ip_id): + raise ValueError(f"IP id {ip_id} is not registered") + + data = self.core_metadata_module_client.contract.encode_abi( + abi_element_identifier="setMetadataURI", + args=[Web3.to_checksum_address(ip_id), metadata_uri, metadata_hash] + ) + + response = self.execute( + to=self.core_metadata_module_client.contract.address, + value=0, + ip_id=ip_id, + data=data, + tx_options=tx_options + ) + + return response + except Exception as e: + raise e diff --git a/tests/integration/test_integration_ip_account.py b/tests/integration/test_integration_ip_account.py index bcfcae1..04f8071 100644 --- a/tests/integration/test_integration_ip_account.py +++ b/tests/integration/test_integration_ip_account.py @@ -8,29 +8,22 @@ from eth_account.messages import encode_typed_data from eth_abi.abi import encode -# Ensure the src directory is in the Python path -current_dir = os.path.dirname(__file__) -src_path = os.path.abspath(os.path.join(current_dir, '..', '..')) -if src_path not in sys.path: - sys.path.append(src_path) - -from utils import get_token_id, get_story_client_in_devnet, MockERC721, getBlockTimestamp - -load_dotenv(override=True) -private_key = os.getenv('WALLET_PRIVATE_KEY') -rpc_url = os.getenv('RPC_PROVIDER_URL') - -# Initialize Web3 -web3 = Web3(Web3.HTTPProvider(rpc_url)) -if not web3.is_connected(): - raise Exception("Failed to connect to Web3 provider") - -# Set up the account with the private key -account = web3.eth.account.from_key(private_key) - -@pytest.fixture -def story_client(): - return get_story_client_in_devnet(web3, account) +from setup_for_integration import ( + web3, + account, + story_client, + get_token_id, + mint_tokens, + approve, + getBlockTimestamp, + check_event_in_tx, + MockERC721, + MockERC20, + ZERO_ADDRESS, + ROYALTY_POLICY, + ROYALTY_MODULE, + PIL_LICENSE_TEMPLATE +) class TestBasicIPAccountOperations: """Basic IP Account operations like execute and nonce retrieval""" @@ -333,4 +326,26 @@ def test_execute_unregistered_ip(self, story_client): data=data ) - assert "is not registered" in str(exc_info.value) \ No newline at end of file + assert "is not registered" in str(exc_info.value) + +class TestSetIpMetadata: + """Tests for setting IP metadata""" + + def test_set_ip_metadata(self, story_client): + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + + response = story_client.IPAccount.setIpMetadata( + ip_id=response['ipId'], + metadata_uri="https://example.com", + metadata_hash=web3.to_hex(web3.keccak(text="test-metadata-hash")) + ) + + assert response is not None, "Response is None, indicating the contract interaction failed." + assert 'txHash' in response, "Response does not contain 'txHash'." + assert response['txHash'] is not None, "'txHash' is None." + assert isinstance(response['txHash'], str), "'txHash' is not a string." + assert len(response['txHash']) > 0, "'txHash' is empty." \ No newline at end of file From 2bf474bf117870473913ec586b566eb935fc560e Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Fri, 4 Apr 2025 10:15:32 +0900 Subject: [PATCH 2/5] Add transferERC20() to IPAccount module --- .../resources/IPAccount.py | 52 ++ .../test_integration_ip_account.py | 639 +++++++++--------- 2 files changed, 388 insertions(+), 303 deletions(-) diff --git a/src/story_protocol_python_sdk/resources/IPAccount.py b/src/story_protocol_python_sdk/resources/IPAccount.py index 356211a..8bd24ec 100644 --- a/src/story_protocol_python_sdk/resources/IPAccount.py +++ b/src/story_protocol_python_sdk/resources/IPAccount.py @@ -186,3 +186,55 @@ def setIpMetadata(self, ip_id: str, metadata_uri: str, metadata_hash: str, tx_op return response except Exception as e: raise e + + def transferERC20(self, ip_id: str, tokens: list, tx_options: dict = None) -> dict: + """Transfers ERC20 tokens from the IP Account to the target address. + + :param ip_id str: The IP ID to transfer tokens from. + :param tokens list: A list of dictionaries containing token transfer details. + Each dictionary should have 'address' (token contract address), + 'target' (recipient address), and 'amount' (token amount). + :param tx_options dict: [Optional] The transaction options. + :returns dict: A dictionary with the transaction hash. + :raises ValueError: If the IP ID is invalid or not registered, or if token parameters are invalid. + """ + try: + if not self._is_registered(ip_id): + raise ValueError(f"IP id {ip_id} is not registered") + + ip_account = IPAccountImplClient(self.web3, contract_address=ip_id) + + # Since we don't have executeBatch, we'll execute transfers one by one + results = [] + for token in tokens: + # Validate token parameters + if not all(key in token for key in ['address', 'target', 'amount']): + raise ValueError("Each token transfer must include 'address', 'target', and 'amount'") + + token_address = self.web3.to_checksum_address(token['address']) + target_address = self.web3.to_checksum_address(token['target']) + amount = int(token['amount']) + + # Build ERC20 transfer function data + data = self.mock_erc20_client.contract.encode_abi( + abi_element_identifier="transfer", + args=[target_address, amount] + ) + + response = build_and_send_transaction( + self.web3, + self.account, + ip_account.build_execute_transaction, + self.web3.to_checksum_address(token_address), + 0, + data, + 0 + ) + results.append(response) + + # Return the hash of the last transaction + # In a real implementation with executeBatch, we would return a single transaction hash + return results[-1] if results else {"txHash": None} + + except Exception as e: + raise ValueError(f"Failed to transfer ERC20: {str(e)}") diff --git a/tests/integration/test_integration_ip_account.py b/tests/integration/test_integration_ip_account.py index 04f8071..cb6d27d 100644 --- a/tests/integration/test_integration_ip_account.py +++ b/tests/integration/test_integration_ip_account.py @@ -25,327 +25,360 @@ PIL_LICENSE_TEMPLATE ) -class TestBasicIPAccountOperations: - """Basic IP Account operations like execute and nonce retrieval""" +# class TestBasicIPAccountOperations: +# """Basic IP Account operations like execute and nonce retrieval""" - def test_execute(self, story_client): - token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - response = story_client.IPAsset.register( - nft_contract=MockERC721, - token_id=token_id - ) - - data = story_client.IPAccount.access_controller_client.contract.encode_abi( - abi_element_identifier="setTransientPermission", - args=[response['ipId'], - account.address, - "0x89630Ccf23277417FBdfd3076C702F5248267e78", - Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], - 1] - ) - - response = story_client.IPAccount.execute( - to=story_client.IPAccount.access_controller_client.contract.address, - value=0, - ip_id=response['ipId'], - data=data - ) - - assert response is not None, "Response is None, indicating the contract interaction failed." - assert 'txHash' in response, "Response does not contain 'txHash'." - assert response['txHash'] is not None, "'txHash' is None." - assert isinstance(response['txHash'], str), "'txHash' is not a string." - assert len(response['txHash']) > 0, "'txHash' is empty." - - def test_get_ip_account_nonce(self, story_client): - """Test getting IP Account nonce.""" - token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - register_response = story_client.IPAsset.register( - nft_contract=MockERC721, - token_id=token_id - ) - ip_id = register_response['ipId'] +# def test_execute(self, story_client): +# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) +# response = story_client.IPAsset.register( +# nft_contract=MockERC721, +# token_id=token_id +# ) + +# data = story_client.IPAccount.access_controller_client.contract.encode_abi( +# abi_element_identifier="setTransientPermission", +# args=[response['ipId'], +# account.address, +# "0x89630Ccf23277417FBdfd3076C702F5248267e78", +# Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], +# 1] +# ) + +# response = story_client.IPAccount.execute( +# to=story_client.IPAccount.access_controller_client.contract.address, +# value=0, +# ip_id=response['ipId'], +# data=data +# ) + +# assert response is not None, "Response is None, indicating the contract interaction failed." +# assert 'txHash' in response, "Response does not contain 'txHash'." +# assert response['txHash'] is not None, "'txHash' is None." +# assert isinstance(response['txHash'], str), "'txHash' is not a string." +# assert len(response['txHash']) > 0, "'txHash' is empty." + +# def test_get_ip_account_nonce(self, story_client): +# """Test getting IP Account nonce.""" +# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) +# register_response = story_client.IPAsset.register( +# nft_contract=MockERC721, +# token_id=token_id +# ) +# ip_id = register_response['ipId'] - state = story_client.IPAccount.getIpAccountNonce(ip_id) +# state = story_client.IPAccount.getIpAccountNonce(ip_id) - assert state is not None - assert isinstance(state, bytes) - - def test_execute_with_encoded_data(self, story_client): - """Test execute with pre-encoded function data.""" - token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - register_response = story_client.IPAsset.register( - nft_contract=MockERC721, - token_id=token_id - ) - ip_id = register_response['ipId'] - - data = story_client.IPAccount.access_controller_client.contract.encode_abi( - abi_element_identifier="setTransientPermission", - args=[ - ip_id, - account.address, - "0x89630Ccf23277417FBdfd3076C702F5248267e78", - Web3.keccak(text="function execute(address,uint256,bytes,uint8)")[:4], - 1 - ] - ) - - response = story_client.IPAccount.execute( - to=story_client.IPAccount.access_controller_client.contract.address, - value=0, - ip_id=ip_id, - data=data - ) - - assert response is not None - assert 'txHash' in response - assert isinstance(response['txHash'], str) - assert len(response['txHash']) > 0 - -class TestSignatureOperations: - """Tests for operations involving signatures""" +# assert state is not None +# assert isinstance(state, bytes) + +# def test_execute_with_encoded_data(self, story_client): +# """Test execute with pre-encoded function data.""" +# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) +# register_response = story_client.IPAsset.register( +# nft_contract=MockERC721, +# token_id=token_id +# ) +# ip_id = register_response['ipId'] + +# data = story_client.IPAccount.access_controller_client.contract.encode_abi( +# abi_element_identifier="setTransientPermission", +# args=[ +# ip_id, +# account.address, +# "0x89630Ccf23277417FBdfd3076C702F5248267e78", +# Web3.keccak(text="function execute(address,uint256,bytes,uint8)")[:4], +# 1 +# ] +# ) + +# response = story_client.IPAccount.execute( +# to=story_client.IPAccount.access_controller_client.contract.address, +# value=0, +# ip_id=ip_id, +# data=data +# ) + +# assert response is not None +# assert 'txHash' in response +# assert isinstance(response['txHash'], str) +# assert len(response['txHash']) > 0 + +# class TestSignatureOperations: +# """Tests for operations involving signatures""" - def test_executeWithSig(self, story_client): - token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - response = story_client.IPAsset.register( - nft_contract=MockERC721, - token_id=token_id - ) - - ipId = response['ipId'] - deadline = getBlockTimestamp(web3) + 100 - state = story_client.IPAccount.getIpAccountNonce(ipId) - - core_data = story_client.IPAccount.access_controller_client.contract.encode_abi( - abi_element_identifier="setTransientPermission", - args=[ - ipId, - account.address, - "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", - Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], - 1 - ] - ) - - execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( - abi_element_identifier="execute", - args=[ - story_client.IPAccount.access_controller_client.contract.address, - 0, - core_data - ] - ) - - expected_state = Web3.keccak( - encode( - ["bytes32", "bytes"], - [state, Web3.to_bytes(hexstr=execute_data)] - ) - ) - - domain_data = { - "name": "Story Protocol IP Account", - "version": "1", - "chainId": 1315, - "verifyingContract": ipId, - } - - message_types = { - "Execute": [ - {"name": "to", "type": "address"}, - {"name": "value", "type": "uint256"}, - {"name": "data", "type": "bytes"}, - {"name": "nonce", "type": "bytes32"}, - {"name": "deadline", "type": "uint256"}, - ], - } - - message_data = { - "to": story_client.IPAccount.access_controller_client.contract.address, - "value": 0, - "data": core_data, - "nonce": expected_state, - "deadline": deadline, - } - - signable_message = encode_typed_data(domain_data, message_types, message_data) - signed_message = Account.sign_message(signable_message, private_key) - - response = story_client.IPAccount.executeWithSig( - to=story_client.IPAccount.access_controller_client.contract.address, - value=0, - ip_id=ipId, - data=core_data, - signer=account.address, - deadline=deadline, - signature=signed_message.signature - ) - - assert response is not None - assert 'txHash' in response - assert isinstance(response['txHash'], str) - assert len(response['txHash']) > 0 - - def test_execute_with_sig_multiple_permissions(self, story_client): - """Test executeWithSig setting multiple permissions.""" - token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - register_response = story_client.IPAsset.register( - nft_contract=MockERC721, - token_id=token_id - ) - ip_id = register_response['ipId'] - - deadline = getBlockTimestamp(web3) + 100 - state = story_client.IPAccount.getIpAccountNonce(ip_id) - - # Prepare all function signatures for permissions - function_signatures = [ - "function setAll(address,string,bytes32,bytes32)", - "function execute(address,uint256,bytes,uint8)", - "function registerDerivative(address,address[],uint256[],address,bytes)" - ] +# def test_executeWithSig(self, story_client): +# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) +# response = story_client.IPAsset.register( +# nft_contract=MockERC721, +# token_id=token_id +# ) + +# ipId = response['ipId'] +# deadline = getBlockTimestamp(web3) + 100 +# state = story_client.IPAccount.getIpAccountNonce(ipId) + +# core_data = story_client.IPAccount.access_controller_client.contract.encode_abi( +# abi_element_identifier="setTransientPermission", +# args=[ +# ipId, +# account.address, +# "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", +# Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], +# 1 +# ] +# ) + +# execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( +# abi_element_identifier="execute", +# args=[ +# story_client.IPAccount.access_controller_client.contract.address, +# 0, +# core_data +# ] +# ) + +# expected_state = Web3.keccak( +# encode( +# ["bytes32", "bytes"], +# [state, Web3.to_bytes(hexstr=execute_data)] +# ) +# ) + +# domain_data = { +# "name": "Story Protocol IP Account", +# "version": "1", +# "chainId": 1315, +# "verifyingContract": ipId, +# } + +# message_types = { +# "Execute": [ +# {"name": "to", "type": "address"}, +# {"name": "value", "type": "uint256"}, +# {"name": "data", "type": "bytes"}, +# {"name": "nonce", "type": "bytes32"}, +# {"name": "deadline", "type": "uint256"}, +# ], +# } + +# message_data = { +# "to": story_client.IPAccount.access_controller_client.contract.address, +# "value": 0, +# "data": core_data, +# "nonce": expected_state, +# "deadline": deadline, +# } + +# signable_message = encode_typed_data(domain_data, message_types, message_data) +# signed_message = Account.sign_message(signable_message, private_key) + +# response = story_client.IPAccount.executeWithSig( +# to=story_client.IPAccount.access_controller_client.contract.address, +# value=0, +# ip_id=ipId, +# data=core_data, +# signer=account.address, +# deadline=deadline, +# signature=signed_message.signature +# ) + +# assert response is not None +# assert 'txHash' in response +# assert isinstance(response['txHash'], str) +# assert len(response['txHash']) > 0 + +# def test_execute_with_sig_multiple_permissions(self, story_client): +# """Test executeWithSig setting multiple permissions.""" +# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) +# register_response = story_client.IPAsset.register( +# nft_contract=MockERC721, +# token_id=token_id +# ) +# ip_id = register_response['ipId'] + +# deadline = getBlockTimestamp(web3) + 100 +# state = story_client.IPAccount.getIpAccountNonce(ip_id) + +# # Prepare all function signatures for permissions +# function_signatures = [ +# "function setAll(address,string,bytes32,bytes32)", +# "function execute(address,uint256,bytes,uint8)", +# "function registerDerivative(address,address[],uint256[],address,bytes)" +# ] - # Create individual permission data and combine them - calls_data = [] - for func_sig in function_signatures: - data = story_client.IPAccount.access_controller_client.contract.encode_abi( - abi_element_identifier="setTransientPermission", - args=[ - ip_id, - account.address, - "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", - Web3.keccak(text=func_sig)[:4], - 1 - ] - ) - if data.startswith('0x'): - data = data[2:] - calls_data.append(data) - - # Combine all encoded data - combined_data = '0x' + ''.join(calls_data) - - # Create the execute data that would be signed - execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( - abi_element_identifier="execute", - args=[ - story_client.IPAccount.access_controller_client.contract.address, - 0, - combined_data - ] - ) - - # Calculate the expected state - expected_state = Web3.keccak( - encode( - ["bytes32", "bytes"], - [state, Web3.to_bytes(hexstr=execute_data)] - ) - ) - - # Prepare signature data - domain_data = { - "name": "Story Protocol IP Account", - "version": "1", - "chainId": 1315, - "verifyingContract": ip_id, - } - - message_types = { - "Execute": [ - {"name": "to", "type": "address"}, - {"name": "value", "type": "uint256"}, - {"name": "data", "type": "bytes"}, - {"name": "nonce", "type": "bytes32"}, - {"name": "deadline", "type": "uint256"}, - ], - } - - message_data = { - "to": story_client.IPAccount.access_controller_client.contract.address, - "value": 0, - "data": combined_data, - "nonce": expected_state, - "deadline": deadline, - } - - signable_message = encode_typed_data(domain_data, message_types, message_data) - signed_message = Account.sign_message(signable_message, private_key) - - response = story_client.IPAccount.executeWithSig( - ip_id=ip_id, - to=story_client.IPAccount.access_controller_client.contract.address, - value=0, - data=combined_data, - signer=account.address, - deadline=deadline, - signature=signed_message.signature - ) - - assert response is not None - assert 'txHash' in response - assert isinstance(response['txHash'], str) - assert len(response['txHash']) > 0 - -class TestErrorCases: - """Tests for error cases and validation""" +# # Create individual permission data and combine them +# calls_data = [] +# for func_sig in function_signatures: +# data = story_client.IPAccount.access_controller_client.contract.encode_abi( +# abi_element_identifier="setTransientPermission", +# args=[ +# ip_id, +# account.address, +# "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", +# Web3.keccak(text=func_sig)[:4], +# 1 +# ] +# ) +# if data.startswith('0x'): +# data = data[2:] +# calls_data.append(data) + +# # Combine all encoded data +# combined_data = '0x' + ''.join(calls_data) + +# # Create the execute data that would be signed +# execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( +# abi_element_identifier="execute", +# args=[ +# story_client.IPAccount.access_controller_client.contract.address, +# 0, +# combined_data +# ] +# ) + +# # Calculate the expected state +# expected_state = Web3.keccak( +# encode( +# ["bytes32", "bytes"], +# [state, Web3.to_bytes(hexstr=execute_data)] +# ) +# ) + +# # Prepare signature data +# domain_data = { +# "name": "Story Protocol IP Account", +# "version": "1", +# "chainId": 1315, +# "verifyingContract": ip_id, +# } + +# message_types = { +# "Execute": [ +# {"name": "to", "type": "address"}, +# {"name": "value", "type": "uint256"}, +# {"name": "data", "type": "bytes"}, +# {"name": "nonce", "type": "bytes32"}, +# {"name": "deadline", "type": "uint256"}, +# ], +# } + +# message_data = { +# "to": story_client.IPAccount.access_controller_client.contract.address, +# "value": 0, +# "data": combined_data, +# "nonce": expected_state, +# "deadline": deadline, +# } + +# signable_message = encode_typed_data(domain_data, message_types, message_data) +# signed_message = Account.sign_message(signable_message, private_key) + +# response = story_client.IPAccount.executeWithSig( +# ip_id=ip_id, +# to=story_client.IPAccount.access_controller_client.contract.address, +# value=0, +# data=combined_data, +# signer=account.address, +# deadline=deadline, +# signature=signed_message.signature +# ) + +# assert response is not None +# assert 'txHash' in response +# assert isinstance(response['txHash'], str) +# assert len(response['txHash']) > 0 + +# class TestErrorCases: +# """Tests for error cases and validation""" - def test_execute_invalid_address(self, story_client): - """Test execute with invalid address should raise error.""" - token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - register_response = story_client.IPAsset.register( - nft_contract=MockERC721, - token_id=token_id - ) - ip_id = register_response['ipId'] - - data = "0x" - invalid_address = "0xinvalid" +# def test_execute_invalid_address(self, story_client): +# """Test execute with invalid address should raise error.""" +# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) +# register_response = story_client.IPAsset.register( +# nft_contract=MockERC721, +# token_id=token_id +# ) +# ip_id = register_response['ipId'] + +# data = "0x" +# invalid_address = "0xinvalid" - with pytest.raises(ValueError) as exc_info: - story_client.IPAccount.execute( - to=invalid_address, - value=0, - ip_id=ip_id, - data=data - ) +# with pytest.raises(ValueError) as exc_info: +# story_client.IPAccount.execute( +# to=invalid_address, +# value=0, +# ip_id=ip_id, +# data=data +# ) - assert "is not a valid address" in str(exc_info.value) +# assert "is not a valid address" in str(exc_info.value) - def test_execute_unregistered_ip(self, story_client): - """Test execute with unregistered IP should raise error.""" - unregistered_ip = "0x1234567890123456789012345678901234567890" - data = "0x" +# def test_execute_unregistered_ip(self, story_client): +# """Test execute with unregistered IP should raise error.""" +# unregistered_ip = "0x1234567890123456789012345678901234567890" +# data = "0x" - with pytest.raises(ValueError) as exc_info: - story_client.IPAccount.execute( - to=story_client.IPAccount.access_controller_client.contract.address, - value=0, - ip_id=unregistered_ip, - data=data - ) +# with pytest.raises(ValueError) as exc_info: +# story_client.IPAccount.execute( +# to=story_client.IPAccount.access_controller_client.contract.address, +# value=0, +# ip_id=unregistered_ip, +# data=data +# ) - assert "is not registered" in str(exc_info.value) +# assert "is not registered" in str(exc_info.value) -class TestSetIpMetadata: - """Tests for setting IP metadata""" +# class TestSetIpMetadata: +# """Tests for setting IP metadata""" - def test_set_ip_metadata(self, story_client): +# def test_set_ip_metadata(self, story_client): +# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) +# response = story_client.IPAsset.register( +# nft_contract=MockERC721, +# token_id=token_id +# ) + +# response = story_client.IPAccount.setIpMetadata( +# ip_id=response['ipId'], +# metadata_uri="https://example.com", +# metadata_hash=web3.to_hex(web3.keccak(text="test-metadata-hash")) +# ) + +# assert response is not None, "Response is None, indicating the contract interaction failed." +# assert 'txHash' in response, "Response does not contain 'txHash'." +# assert response['txHash'] is not None, "'txHash' is None." +# assert isinstance(response['txHash'], str), "'txHash' is not a string." +# assert len(response['txHash']) > 0, "'txHash' is empty." + +class testTransferERC20: + """Tests for transferring ERC20 tokens""" + + def test_transfer_erc20(self, story_client): + """Test transferring ERC20 tokens""" token_id = get_token_id(MockERC721, story_client.web3, story_client.account) response = story_client.IPAsset.register( nft_contract=MockERC721, token_id=token_id ) + ip_id = response['ipId'] - response = story_client.IPAccount.setIpMetadata( - ip_id=response['ipId'], - metadata_uri="https://example.com", - metadata_hash=web3.to_hex(web3.keccak(text="test-metadata-hash")) + # 1. Query token balance of ipId and wallet before + initial_erc20_balance_of_ip_id = story_client.Royalty.mock_erc20_client.balanceOf( + account=ip_id + ) + initial_erc20_balance_of_wallet = story_client.Royalty.mock_erc20_client.balanceOf( + account=story_client.account.address ) + initial_wip_balance_of_ip_id = story_client.WIP.balanceOf( + account=ip_id + ) + initial_wip_balance_of_wallet = story_client.WIP.balanceOf( + account=story_client.account.address + ) + + print("initial_erc20_balance_of_ip_id", initial_erc20_balance_of_ip_id) + print("initial_erc20_balance_of_wallet", initial_erc20_balance_of_wallet) + print("initial_wip_balance_of_ip_id", initial_wip_balance_of_ip_id) + print("initial_wip_balance_of_wallet", initial_wip_balance_of_wallet) - assert response is not None, "Response is None, indicating the contract interaction failed." - assert 'txHash' in response, "Response does not contain 'txHash'." - assert response['txHash'] is not None, "'txHash' is None." - assert isinstance(response['txHash'], str), "'txHash' is not a string." - assert len(response['txHash']) > 0, "'txHash' is empty." \ No newline at end of file + pass From 3eda58891ea5fd8ebea284d19f1398f8ced5500f Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Sat, 5 Apr 2025 17:25:20 +0900 Subject: [PATCH 3/5] Added private_key to utils for ip account integration test --- tests/integration/setup_for_integration.py | 4 +++- tests/integration/utils.py | 1 - 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/integration/setup_for_integration.py b/tests/integration/setup_for_integration.py index 648fc55..ce49301 100644 --- a/tests/integration/setup_for_integration.py +++ b/tests/integration/setup_for_integration.py @@ -84,5 +84,7 @@ def story_client_2(): 'setup_royalty_vault', 'WIP_TOKEN_ADDRESS', 'wallet_address', - 'wallet_address_2' + 'wallet_address_2', + 'private_key', + 'private_key_2' ] \ No newline at end of file diff --git a/tests/integration/utils.py b/tests/integration/utils.py index 0de02e0..81b0bdf 100644 --- a/tests/integration/utils.py +++ b/tests/integration/utils.py @@ -196,7 +196,6 @@ def generate_cid() -> str: # Base58 encode return base58.b58encode(multihash).decode('utf-8') - def setup_royalty_vault(story_client, parent_ip_id, account): parent_ip_royalty_address = story_client.Royalty.getRoyaltyVaultAddress(parent_ip_id) From 8666200d360d1724563262637ea9ac7eaf06e14f Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Sat, 5 Apr 2025 17:25:33 +0900 Subject: [PATCH 4/5] Fixed transferERC20() bug --- .../resources/IPAccount.py | 4 +- .../test_integration_ip_account.py | 706 ++++++++++-------- 2 files changed, 380 insertions(+), 330 deletions(-) diff --git a/src/story_protocol_python_sdk/resources/IPAccount.py b/src/story_protocol_python_sdk/resources/IPAccount.py index 8bd24ec..9bc4237 100644 --- a/src/story_protocol_python_sdk/resources/IPAccount.py +++ b/src/story_protocol_python_sdk/resources/IPAccount.py @@ -6,6 +6,7 @@ from story_protocol_python_sdk.abi.IPAssetRegistry.IPAssetRegistry_client import IPAssetRegistryClient from story_protocol_python_sdk.abi.AccessController.AccessController_client import AccessControllerClient from story_protocol_python_sdk.abi.CoreMetadataModule.CoreMetadataModule_client import CoreMetadataModuleClient +from story_protocol_python_sdk.abi.MockERC20.MockERC20_client import MockERC20Client from story_protocol_python_sdk.utils.transaction_utils import build_and_send_transaction @@ -26,6 +27,7 @@ def __init__(self, web3: Web3, account, chain_id: int): self.access_controller_client = AccessControllerClient(web3) self.ip_account_client = IPAccountImplClient(web3) self.core_metadata_module_client = CoreMetadataModuleClient(web3) + self.mock_erc20_client = MockERC20Client(web3) def getToken(self, ip_id: str) -> dict: """Retrieve token information associated with an IP account. @@ -207,7 +209,6 @@ def transferERC20(self, ip_id: str, tokens: list, tx_options: dict = None) -> di # Since we don't have executeBatch, we'll execute transfers one by one results = [] for token in tokens: - # Validate token parameters if not all(key in token for key in ['address', 'target', 'amount']): raise ValueError("Each token transfer must include 'address', 'target', and 'amount'") @@ -215,7 +216,6 @@ def transferERC20(self, ip_id: str, tokens: list, tx_options: dict = None) -> di target_address = self.web3.to_checksum_address(token['target']) amount = int(token['amount']) - # Build ERC20 transfer function data data = self.mock_erc20_client.contract.encode_abi( abi_element_identifier="transfer", args=[target_address, amount] diff --git a/tests/integration/test_integration_ip_account.py b/tests/integration/test_integration_ip_account.py index cb6d27d..a1ec375 100644 --- a/tests/integration/test_integration_ip_account.py +++ b/tests/integration/test_integration_ip_account.py @@ -1,8 +1,6 @@ # tests/integration/test_integration_ip_account.py -import os, json, sys import pytest -from dotenv import load_dotenv from web3 import Web3 from eth_account import Account from eth_account.messages import encode_typed_data @@ -14,343 +12,338 @@ story_client, get_token_id, mint_tokens, - approve, getBlockTimestamp, - check_event_in_tx, MockERC721, MockERC20, - ZERO_ADDRESS, - ROYALTY_POLICY, - ROYALTY_MODULE, - PIL_LICENSE_TEMPLATE + private_key ) -# class TestBasicIPAccountOperations: -# """Basic IP Account operations like execute and nonce retrieval""" +class TestBasicIPAccountOperations: + """Basic IP Account operations like execute and nonce retrieval""" -# def test_execute(self, story_client): -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# response = story_client.IPAsset.register( -# nft_contract=MockERC721, -# token_id=token_id -# ) - -# data = story_client.IPAccount.access_controller_client.contract.encode_abi( -# abi_element_identifier="setTransientPermission", -# args=[response['ipId'], -# account.address, -# "0x89630Ccf23277417FBdfd3076C702F5248267e78", -# Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], -# 1] -# ) - -# response = story_client.IPAccount.execute( -# to=story_client.IPAccount.access_controller_client.contract.address, -# value=0, -# ip_id=response['ipId'], -# data=data -# ) - -# assert response is not None, "Response is None, indicating the contract interaction failed." -# assert 'txHash' in response, "Response does not contain 'txHash'." -# assert response['txHash'] is not None, "'txHash' is None." -# assert isinstance(response['txHash'], str), "'txHash' is not a string." -# assert len(response['txHash']) > 0, "'txHash' is empty." - -# def test_get_ip_account_nonce(self, story_client): -# """Test getting IP Account nonce.""" -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# register_response = story_client.IPAsset.register( -# nft_contract=MockERC721, -# token_id=token_id -# ) -# ip_id = register_response['ipId'] + def test_execute(self, story_client): + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + + data = story_client.IPAccount.access_controller_client.contract.encode_abi( + abi_element_identifier="setTransientPermission", + args=[response['ipId'], + account.address, + "0x89630Ccf23277417FBdfd3076C702F5248267e78", + Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], + 1] + ) + + response = story_client.IPAccount.execute( + to=story_client.IPAccount.access_controller_client.contract.address, + value=0, + ip_id=response['ipId'], + data=data + ) + + assert response is not None, "Response is None, indicating the contract interaction failed." + assert 'txHash' in response, "Response does not contain 'txHash'." + assert response['txHash'] is not None, "'txHash' is None." + assert isinstance(response['txHash'], str), "'txHash' is not a string." + assert len(response['txHash']) > 0, "'txHash' is empty." + + def test_get_ip_account_nonce(self, story_client): + """Test getting IP Account nonce.""" + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + register_response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + ip_id = register_response['ipId'] -# state = story_client.IPAccount.getIpAccountNonce(ip_id) + state = story_client.IPAccount.getIpAccountNonce(ip_id) -# assert state is not None -# assert isinstance(state, bytes) - -# def test_execute_with_encoded_data(self, story_client): -# """Test execute with pre-encoded function data.""" -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# register_response = story_client.IPAsset.register( -# nft_contract=MockERC721, -# token_id=token_id -# ) -# ip_id = register_response['ipId'] - -# data = story_client.IPAccount.access_controller_client.contract.encode_abi( -# abi_element_identifier="setTransientPermission", -# args=[ -# ip_id, -# account.address, -# "0x89630Ccf23277417FBdfd3076C702F5248267e78", -# Web3.keccak(text="function execute(address,uint256,bytes,uint8)")[:4], -# 1 -# ] -# ) - -# response = story_client.IPAccount.execute( -# to=story_client.IPAccount.access_controller_client.contract.address, -# value=0, -# ip_id=ip_id, -# data=data -# ) - -# assert response is not None -# assert 'txHash' in response -# assert isinstance(response['txHash'], str) -# assert len(response['txHash']) > 0 - -# class TestSignatureOperations: -# """Tests for operations involving signatures""" + assert state is not None + assert isinstance(state, bytes) + + def test_execute_with_encoded_data(self, story_client): + """Test execute with pre-encoded function data.""" + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + register_response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + ip_id = register_response['ipId'] + + data = story_client.IPAccount.access_controller_client.contract.encode_abi( + abi_element_identifier="setTransientPermission", + args=[ + ip_id, + account.address, + "0x89630Ccf23277417FBdfd3076C702F5248267e78", + Web3.keccak(text="function execute(address,uint256,bytes,uint8)")[:4], + 1 + ] + ) + + response = story_client.IPAccount.execute( + to=story_client.IPAccount.access_controller_client.contract.address, + value=0, + ip_id=ip_id, + data=data + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + +class TestSignatureOperations: + """Tests for operations involving signatures""" -# def test_executeWithSig(self, story_client): -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# response = story_client.IPAsset.register( -# nft_contract=MockERC721, -# token_id=token_id -# ) - -# ipId = response['ipId'] -# deadline = getBlockTimestamp(web3) + 100 -# state = story_client.IPAccount.getIpAccountNonce(ipId) - -# core_data = story_client.IPAccount.access_controller_client.contract.encode_abi( -# abi_element_identifier="setTransientPermission", -# args=[ -# ipId, -# account.address, -# "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", -# Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], -# 1 -# ] -# ) - -# execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( -# abi_element_identifier="execute", -# args=[ -# story_client.IPAccount.access_controller_client.contract.address, -# 0, -# core_data -# ] -# ) - -# expected_state = Web3.keccak( -# encode( -# ["bytes32", "bytes"], -# [state, Web3.to_bytes(hexstr=execute_data)] -# ) -# ) - -# domain_data = { -# "name": "Story Protocol IP Account", -# "version": "1", -# "chainId": 1315, -# "verifyingContract": ipId, -# } - -# message_types = { -# "Execute": [ -# {"name": "to", "type": "address"}, -# {"name": "value", "type": "uint256"}, -# {"name": "data", "type": "bytes"}, -# {"name": "nonce", "type": "bytes32"}, -# {"name": "deadline", "type": "uint256"}, -# ], -# } - -# message_data = { -# "to": story_client.IPAccount.access_controller_client.contract.address, -# "value": 0, -# "data": core_data, -# "nonce": expected_state, -# "deadline": deadline, -# } - -# signable_message = encode_typed_data(domain_data, message_types, message_data) -# signed_message = Account.sign_message(signable_message, private_key) - -# response = story_client.IPAccount.executeWithSig( -# to=story_client.IPAccount.access_controller_client.contract.address, -# value=0, -# ip_id=ipId, -# data=core_data, -# signer=account.address, -# deadline=deadline, -# signature=signed_message.signature -# ) - -# assert response is not None -# assert 'txHash' in response -# assert isinstance(response['txHash'], str) -# assert len(response['txHash']) > 0 - -# def test_execute_with_sig_multiple_permissions(self, story_client): -# """Test executeWithSig setting multiple permissions.""" -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# register_response = story_client.IPAsset.register( -# nft_contract=MockERC721, -# token_id=token_id -# ) -# ip_id = register_response['ipId'] - -# deadline = getBlockTimestamp(web3) + 100 -# state = story_client.IPAccount.getIpAccountNonce(ip_id) - -# # Prepare all function signatures for permissions -# function_signatures = [ -# "function setAll(address,string,bytes32,bytes32)", -# "function execute(address,uint256,bytes,uint8)", -# "function registerDerivative(address,address[],uint256[],address,bytes)" -# ] + def test_executeWithSig(self, story_client): + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + + ipId = response['ipId'] + deadline = getBlockTimestamp(web3) + 100 + state = story_client.IPAccount.getIpAccountNonce(ipId) + + core_data = story_client.IPAccount.access_controller_client.contract.encode_abi( + abi_element_identifier="setTransientPermission", + args=[ + ipId, + account.address, + "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", + Web3.keccak(text="function setAll(address,string,bytes32,bytes32)")[:4], + 1 + ] + ) + + execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( + abi_element_identifier="execute", + args=[ + story_client.IPAccount.access_controller_client.contract.address, + 0, + core_data + ] + ) + + expected_state = Web3.keccak( + encode( + ["bytes32", "bytes"], + [state, Web3.to_bytes(hexstr=execute_data)] + ) + ) + + domain_data = { + "name": "Story Protocol IP Account", + "version": "1", + "chainId": 1315, + "verifyingContract": ipId, + } + + message_types = { + "Execute": [ + {"name": "to", "type": "address"}, + {"name": "value", "type": "uint256"}, + {"name": "data", "type": "bytes"}, + {"name": "nonce", "type": "bytes32"}, + {"name": "deadline", "type": "uint256"}, + ], + } + + message_data = { + "to": story_client.IPAccount.access_controller_client.contract.address, + "value": 0, + "data": core_data, + "nonce": expected_state, + "deadline": deadline, + } + + signable_message = encode_typed_data(domain_data, message_types, message_data) + signed_message = Account.sign_message(signable_message, private_key) + + response = story_client.IPAccount.executeWithSig( + to=story_client.IPAccount.access_controller_client.contract.address, + value=0, + ip_id=ipId, + data=core_data, + signer=account.address, + deadline=deadline, + signature=signed_message.signature + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + + def test_execute_with_sig_multiple_permissions(self, story_client): + """Test executeWithSig setting multiple permissions.""" + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + register_response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + ip_id = register_response['ipId'] + + deadline = getBlockTimestamp(web3) + 100 + state = story_client.IPAccount.getIpAccountNonce(ip_id) + + # Prepare all function signatures for permissions + function_signatures = [ + "function setAll(address,string,bytes32,bytes32)", + "function execute(address,uint256,bytes,uint8)", + "function registerDerivative(address,address[],uint256[],address,bytes)" + ] -# # Create individual permission data and combine them -# calls_data = [] -# for func_sig in function_signatures: -# data = story_client.IPAccount.access_controller_client.contract.encode_abi( -# abi_element_identifier="setTransientPermission", -# args=[ -# ip_id, -# account.address, -# "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", -# Web3.keccak(text=func_sig)[:4], -# 1 -# ] -# ) -# if data.startswith('0x'): -# data = data[2:] -# calls_data.append(data) - -# # Combine all encoded data -# combined_data = '0x' + ''.join(calls_data) - -# # Create the execute data that would be signed -# execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( -# abi_element_identifier="execute", -# args=[ -# story_client.IPAccount.access_controller_client.contract.address, -# 0, -# combined_data -# ] -# ) - -# # Calculate the expected state -# expected_state = Web3.keccak( -# encode( -# ["bytes32", "bytes"], -# [state, Web3.to_bytes(hexstr=execute_data)] -# ) -# ) - -# # Prepare signature data -# domain_data = { -# "name": "Story Protocol IP Account", -# "version": "1", -# "chainId": 1315, -# "verifyingContract": ip_id, -# } - -# message_types = { -# "Execute": [ -# {"name": "to", "type": "address"}, -# {"name": "value", "type": "uint256"}, -# {"name": "data", "type": "bytes"}, -# {"name": "nonce", "type": "bytes32"}, -# {"name": "deadline", "type": "uint256"}, -# ], -# } - -# message_data = { -# "to": story_client.IPAccount.access_controller_client.contract.address, -# "value": 0, -# "data": combined_data, -# "nonce": expected_state, -# "deadline": deadline, -# } - -# signable_message = encode_typed_data(domain_data, message_types, message_data) -# signed_message = Account.sign_message(signable_message, private_key) - -# response = story_client.IPAccount.executeWithSig( -# ip_id=ip_id, -# to=story_client.IPAccount.access_controller_client.contract.address, -# value=0, -# data=combined_data, -# signer=account.address, -# deadline=deadline, -# signature=signed_message.signature -# ) - -# assert response is not None -# assert 'txHash' in response -# assert isinstance(response['txHash'], str) -# assert len(response['txHash']) > 0 - -# class TestErrorCases: -# """Tests for error cases and validation""" + # Create individual permission data and combine them + calls_data = [] + for func_sig in function_signatures: + data = story_client.IPAccount.access_controller_client.contract.encode_abi( + abi_element_identifier="setTransientPermission", + args=[ + ip_id, + account.address, + "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", + Web3.keccak(text=func_sig)[:4], + 1 + ] + ) + if data.startswith('0x'): + data = data[2:] + calls_data.append(data) + + # Combine all encoded data + combined_data = '0x' + ''.join(calls_data) + + # Create the execute data that would be signed + execute_data = story_client.IPAccount.ip_account_client.contract.encode_abi( + abi_element_identifier="execute", + args=[ + story_client.IPAccount.access_controller_client.contract.address, + 0, + combined_data + ] + ) + + # Calculate the expected state + expected_state = Web3.keccak( + encode( + ["bytes32", "bytes"], + [state, Web3.to_bytes(hexstr=execute_data)] + ) + ) + + # Prepare signature data + domain_data = { + "name": "Story Protocol IP Account", + "version": "1", + "chainId": 1315, + "verifyingContract": ip_id, + } + + message_types = { + "Execute": [ + {"name": "to", "type": "address"}, + {"name": "value", "type": "uint256"}, + {"name": "data", "type": "bytes"}, + {"name": "nonce", "type": "bytes32"}, + {"name": "deadline", "type": "uint256"}, + ], + } + + message_data = { + "to": story_client.IPAccount.access_controller_client.contract.address, + "value": 0, + "data": combined_data, + "nonce": expected_state, + "deadline": deadline, + } + + signable_message = encode_typed_data(domain_data, message_types, message_data) + signed_message = Account.sign_message(signable_message, private_key) + + response = story_client.IPAccount.executeWithSig( + ip_id=ip_id, + to=story_client.IPAccount.access_controller_client.contract.address, + value=0, + data=combined_data, + signer=account.address, + deadline=deadline, + signature=signed_message.signature + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + +class TestErrorCases: + """Tests for error cases and validation""" -# def test_execute_invalid_address(self, story_client): -# """Test execute with invalid address should raise error.""" -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# register_response = story_client.IPAsset.register( -# nft_contract=MockERC721, -# token_id=token_id -# ) -# ip_id = register_response['ipId'] - -# data = "0x" -# invalid_address = "0xinvalid" + def test_execute_invalid_address(self, story_client): + """Test execute with invalid address should raise error.""" + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + register_response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + ip_id = register_response['ipId'] + + data = "0x" + invalid_address = "0xinvalid" -# with pytest.raises(ValueError) as exc_info: -# story_client.IPAccount.execute( -# to=invalid_address, -# value=0, -# ip_id=ip_id, -# data=data -# ) + with pytest.raises(ValueError) as exc_info: + story_client.IPAccount.execute( + to=invalid_address, + value=0, + ip_id=ip_id, + data=data + ) -# assert "is not a valid address" in str(exc_info.value) + assert "is not a valid address" in str(exc_info.value) -# def test_execute_unregistered_ip(self, story_client): -# """Test execute with unregistered IP should raise error.""" -# unregistered_ip = "0x1234567890123456789012345678901234567890" -# data = "0x" + def test_execute_unregistered_ip(self, story_client): + """Test execute with unregistered IP should raise error.""" + unregistered_ip = "0x1234567890123456789012345678901234567890" + data = "0x" -# with pytest.raises(ValueError) as exc_info: -# story_client.IPAccount.execute( -# to=story_client.IPAccount.access_controller_client.contract.address, -# value=0, -# ip_id=unregistered_ip, -# data=data -# ) + with pytest.raises(ValueError) as exc_info: + story_client.IPAccount.execute( + to=story_client.IPAccount.access_controller_client.contract.address, + value=0, + ip_id=unregistered_ip, + data=data + ) -# assert "is not registered" in str(exc_info.value) + assert "is not registered" in str(exc_info.value) -# class TestSetIpMetadata: -# """Tests for setting IP metadata""" +class TestSetIpMetadata: + """Tests for setting IP metadata""" -# def test_set_ip_metadata(self, story_client): -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# response = story_client.IPAsset.register( -# nft_contract=MockERC721, -# token_id=token_id -# ) - -# response = story_client.IPAccount.setIpMetadata( -# ip_id=response['ipId'], -# metadata_uri="https://example.com", -# metadata_hash=web3.to_hex(web3.keccak(text="test-metadata-hash")) -# ) - -# assert response is not None, "Response is None, indicating the contract interaction failed." -# assert 'txHash' in response, "Response does not contain 'txHash'." -# assert response['txHash'] is not None, "'txHash' is None." -# assert isinstance(response['txHash'], str), "'txHash' is not a string." -# assert len(response['txHash']) > 0, "'txHash' is empty." - -class testTransferERC20: + def test_set_ip_metadata(self, story_client): + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + + response = story_client.IPAccount.setIpMetadata( + ip_id=response['ipId'], + metadata_uri="https://example.com", + metadata_hash=web3.to_hex(web3.keccak(text="test-metadata-hash")) + ) + + assert response is not None, "Response is None, indicating the contract interaction failed." + assert 'txHash' in response, "Response does not contain 'txHash'." + assert response['txHash'] is not None, "'txHash' is None." + assert isinstance(response['txHash'], str), "'txHash' is not a string." + assert len(response['txHash']) > 0, "'txHash' is empty." + +class TestTransferERC20: """Tests for transferring ERC20 tokens""" def test_transfer_erc20(self, story_client): @@ -370,15 +363,72 @@ def test_transfer_erc20(self, story_client): account=story_client.account.address ) initial_wip_balance_of_ip_id = story_client.WIP.balanceOf( - account=ip_id + address=ip_id ) initial_wip_balance_of_wallet = story_client.WIP.balanceOf( - account=story_client.account.address + address=story_client.account.address + ) + + # 2. Transfer ERC20 tokens to the IP account + amount_to_mint = 2000000 # Equivalent to 0.002 ether in wei + mint_receipt = mint_tokens( + erc20_contract_address=MockERC20, + web3=story_client.web3, + account=story_client.account, + to_address=ip_id, + amount=amount_to_mint + ) + + # 3. Transfer WIP to the IP account + # First deposit (wrap) IP to WIP + deposit_response = story_client.WIP.deposit( + amount=1 + ) + + # Then transfer WIP to the IP account + response = story_client.WIP.transfer( + to=ip_id, + amount=1 ) - print("initial_erc20_balance_of_ip_id", initial_erc20_balance_of_ip_id) - print("initial_erc20_balance_of_wallet", initial_erc20_balance_of_wallet) - print("initial_wip_balance_of_ip_id", initial_wip_balance_of_ip_id) - print("initial_wip_balance_of_wallet", initial_wip_balance_of_wallet) + # 4. Transfer tokens from IP account to wallet address + response = story_client.IPAccount.transferERC20( + ip_id=ip_id, + tokens=[ + { + "address": story_client.WIP.wip_client.contract.address, + "target": story_client.account.address, + "amount": 1 + }, + { + "address": MockERC20, + "target": story_client.account.address, + "amount": 1000000 # Equivalent to 0.001 ether + }, + { + "address": MockERC20, + "target": story_client.account.address, + "amount": 1000000 # Equivalent to 0.001 ether + } + ] + ) + + # 5. Query token balance of ipId and wallet address after transfer + final_erc20_balance_of_ip_id = story_client.Royalty.mock_erc20_client.balanceOf( + account=ip_id + ) + final_wip_balance_of_ip_id = story_client.WIP.balanceOf( + address=ip_id + ) + final_erc20_balance_of_wallet = story_client.Royalty.mock_erc20_client.balanceOf( + account=story_client.account.address + ) + final_wip_balance_of_wallet = story_client.WIP.balanceOf( + address=story_client.account.address + ) - pass + assert isinstance(response['txHash'], str) and response['txHash'] != "" + assert final_erc20_balance_of_ip_id == initial_erc20_balance_of_ip_id + assert final_wip_balance_of_ip_id == initial_wip_balance_of_ip_id + assert final_erc20_balance_of_wallet == initial_erc20_balance_of_wallet + 2000000 + assert final_wip_balance_of_wallet == initial_wip_balance_of_wallet + 1 From e3b960659fddc070c0e605a1a2d122720b76b87f Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Tue, 8 Apr 2025 03:21:57 +0900 Subject: [PATCH 5/5] Updated transferERC20() to use executeBatch() --- .../abi/IPAccountImpl/IPAccountImpl_client.py | 8 +++++ .../resources/IPAccount.py | 33 ++++++++++--------- .../scripts/config.json | 1 + 3 files changed, 26 insertions(+), 16 deletions(-) diff --git a/src/story_protocol_python_sdk/abi/IPAccountImpl/IPAccountImpl_client.py b/src/story_protocol_python_sdk/abi/IPAccountImpl/IPAccountImpl_client.py index 3d412be..2b868d9 100644 --- a/src/story_protocol_python_sdk/abi/IPAccountImpl/IPAccountImpl_client.py +++ b/src/story_protocol_python_sdk/abi/IPAccountImpl/IPAccountImpl_client.py @@ -27,6 +27,14 @@ def build_execute_transaction(self, to, value, data, operation, tx_params): # return self.contract.functions.execute(to, value, data).build_transaction(tx_params) + def executeBatch(self, calls, operation): + + return self.contract.functions.executeBatch(calls, operation).transact() + + def build_executeBatch_transaction(self, calls, operation, tx_params): + return self.contract.functions.executeBatch(calls, operation).build_transaction(tx_params) + + def executeWithSig(self, to, value, data, signer, deadline, signature): return self.contract.functions.executeWithSig(to, value, data, signer, deadline, signature).transact() diff --git a/src/story_protocol_python_sdk/resources/IPAccount.py b/src/story_protocol_python_sdk/resources/IPAccount.py index 9bc4237..cea4bd6 100644 --- a/src/story_protocol_python_sdk/resources/IPAccount.py +++ b/src/story_protocol_python_sdk/resources/IPAccount.py @@ -206,12 +206,12 @@ def transferERC20(self, ip_id: str, tokens: list, tx_options: dict = None) -> di ip_account = IPAccountImplClient(self.web3, contract_address=ip_id) - # Since we don't have executeBatch, we'll execute transfers one by one - results = [] for token in tokens: if not all(key in token for key in ['address', 'target', 'amount']): raise ValueError("Each token transfer must include 'address', 'target', and 'amount'") - + + calls = [] + for token in tokens: token_address = self.web3.to_checksum_address(token['address']) target_address = self.web3.to_checksum_address(token['target']) amount = int(token['amount']) @@ -221,20 +221,21 @@ def transferERC20(self, ip_id: str, tokens: list, tx_options: dict = None) -> di args=[target_address, amount] ) - response = build_and_send_transaction( - self.web3, - self.account, - ip_account.build_execute_transaction, - self.web3.to_checksum_address(token_address), - 0, - data, - 0 - ) - results.append(response) + calls.append({ + 'target': token_address, + 'data': data, + 'value': 0 + }) - # Return the hash of the last transaction - # In a real implementation with executeBatch, we would return a single transaction hash - return results[-1] if results else {"txHash": None} + response = build_and_send_transaction( + self.web3, + self.account, + ip_account.build_executeBatch_transaction, + calls, + 0, + tx_options=tx_options + ) + return response except Exception as e: raise ValueError(f"Failed to transfer ERC20: {str(e)}") diff --git a/src/story_protocol_python_sdk/scripts/config.json b/src/story_protocol_python_sdk/scripts/config.json index 9b1b5fd..e20f8af 100644 --- a/src/story_protocol_python_sdk/scripts/config.json +++ b/src/story_protocol_python_sdk/scripts/config.json @@ -30,6 +30,7 @@ "contract_address": "0xc93d49fEdED1A2fbE3B54223Df65f4edB3845eb0", "functions": [ "execute", + "executeBatch", "executeWithSig", "state", "token",