From 512bc8d2432d1be2a1c567437becfe1e13086ddb Mon Sep 17 00:00:00 2001 From: chao-peng-story Date: Mon, 2 Mar 2026 16:02:24 +0800 Subject: [PATCH] feat(IPAsset): add batch_mint_and_register_ip_asset_with_pil_terms method Implement batch minting and registering of IP assets with PIL terms attached, based on TypeScript SDK implementation. This method uses multicall to batch multiple mint and register operations in a single transaction. Changes: - Add batch_mint_and_register_ip_asset_with_pil_terms to IPAsset resource - Support encodedTxDataOnly option in mint_and_register_ip_asset_with_pil_terms for multicall batching - Add _parse_tx_license_terms_attached_event_for_ip helper to parse license terms for specific IPs - Add unit tests covering successful batch minting, metadata, recipient, and empty args scenarios - Add integration tests for batch minting with PIL terms and metadata/recipient options - Use allow_duplicates=True in tests to handle duplicate license terms in same NFT collection --- .../resources/IPAsset.py | 151 +++++++++ .../integration/test_integration_ip_asset.py | 129 +++++++ ...nt_and_register_ip_asset_with_pil_terms.py | 320 ++++++++++++++++++ 3 files changed, 600 insertions(+) create mode 100644 tests/unit/resources/test_batch_mint_and_register_ip_asset_with_pil_terms.py diff --git a/src/story_protocol_python_sdk/resources/IPAsset.py b/src/story_protocol_python_sdk/resources/IPAsset.py index 5066ad28..53289402 100644 --- a/src/story_protocol_python_sdk/resources/IPAsset.py +++ b/src/story_protocol_python_sdk/resources/IPAsset.py @@ -557,6 +557,18 @@ def mint_and_register_ip_asset_with_pil_terms( } ) + # Check if only encoded transaction data is requested + if tx_options and tx_options.get("encodedTxDataOnly"): + # Build transaction to get encoded data + tx_data = self.license_attachment_workflows_client.contract.functions.mintAndRegisterIpAndAttachPILTerms( + spg_nft_contract, + self._validate_recipient(recipient), + metadata, + license_terms, + allow_duplicates, + ).build_transaction({"from": self.account.address, "gas": 0}) + return {"encoded_tx_data": tx_data["data"]} + response = build_and_send_transaction( self.web3, self.account, @@ -586,6 +598,120 @@ def mint_and_register_ip_asset_with_pil_terms( except Exception as e: raise e + def batch_mint_and_register_ip_asset_with_pil_terms( + self, + args: list[dict], + tx_options: dict | None = None, + ) -> dict: + """ + Batch mint NFTs from collections and register them as IPs with PIL terms attached. + + :param args list[dict]: List of mint and register configurations, each containing: + :param spg_nft_contract str: The address of the NFT collection. + :param terms list: An array of license terms to attach. + :param terms dict: The license terms configuration. + :param transferable bool: Transferability of the license. + :param royalty_policy str: Address of the royalty policy contract. + :param default_minting_fee int: Fee for minting a license. + :param expiration int: License expiration. + :param commercial_use bool: Whether commercial use is allowed. + :param commercial_attribution bool: Whether attribution is needed for commercial use. + :param commercializer_checker str: Allowed commercializers or zero address for none. + :param commercializer_checker_data str: Data for checker contract. + :param commercial_rev_share int: Percentage of revenue that must be shared with the licensor. Must be between 0 and 100 (where 100% represents 100,000,000). + :param commercial_rev_ceiling int: Maximum commercial revenue. + :param derivatives_allowed bool: Whether derivatives are allowed. + :param derivatives_attribution bool: Whether attribution is needed for derivatives. + :param derivatives_approval bool: Whether licensor approval is required for derivatives. + :param derivatives_reciprocal bool: Whether derivatives must use the same license terms. + :param derivative_rev_ceiling int: Max derivative revenue. + :param currency str: ERC20 token for the minting fee. + :param uri str: URI for offchain license terms. + :param licensing_config dict: The configuration for the license. + :param is_set bool: Whether the configuration is set or not. + :param minting_fee int: The fee to be paid when minting tokens. + :param hook_data str: The data used by the licensing hook. + :param licensing_hook str: The licensing hook contract address or address(0) if none. + :param commercial_rev_share int: Percentage of revenue that must be shared with the licensor. Must be between 0 and 100 (where 100% represents 100,000,000). + :param disabled bool: Whether the license is disabled. + :param expect_minimum_group_reward_share int: Minimum group reward share percentage. Must be between 0 and 100 (where 100% represents 100,000,000). + :param expect_group_reward_pool str: Address of the expected group reward pool. + :param ip_metadata dict: [Optional] NFT and IP metadata. + :param ip_metadata_uri str: [Optional] IP metadata URI. + :param ip_metadata_hash str: [Optional] IP metadata hash. + :param nft_metadata_uri str: [Optional] NFT metadata URI. + :param nft_metadata_hash str: [Optional] NFT metadata hash. + :param recipient str: [Optional] Recipient address (defaults to caller). + :param allow_duplicates bool: [Optional] Whether to allow duplicates. + :param tx_options dict: [Optional] Transaction options. + :return dict: Dictionary with tx hash and list of results for each minted IP. + :return tx_hash str: The transaction hash. + :return results list[dict]: List of results, each containing: + :return ip_id str: The ID of the registered IP. + :return token_id int: The ID of the minted NFT. + :return spg_nft_contract str: The address of the NFT collection. + :return license_terms_ids list[int]: The IDs of the attached license terms. + """ + try: + # Encode all mint and register calls + calldata = [] + for arg in args: + # Use encodedTxDataOnly to get the encoded transaction data + arg_with_encoded_option = arg.copy() + arg_with_encoded_option["tx_options"] = {"encodedTxDataOnly": True} + + result = self.mint_and_register_ip_asset_with_pil_terms( + spg_nft_contract=arg["spg_nft_contract"], + terms=arg["terms"], + ip_metadata=arg.get("ip_metadata"), + recipient=arg.get("recipient"), + allow_duplicates=arg.get("allow_duplicates", False), + tx_options=arg_with_encoded_option["tx_options"], + ) + calldata.append(result["encoded_tx_data"]) + + # Send multicall transaction + response = build_and_send_transaction( + self.web3, + self.account, + self.license_attachment_workflows_client.build_multicall_transaction, + calldata, + tx_options=tx_options, + ) + + # Parse IPRegistered events with full details + event_signature = self.web3.keccak( + text="IPRegistered(address,uint256,address,uint256,string,string,uint256)" + ).hex() + + results = [] + for log in response["tx_receipt"]["logs"]: + if log["topics"][0].hex() == event_signature: + event_result = self.ip_asset_registry_client.contract.events.IPRegistered.process_log(log) + ip_id = self.web3.to_checksum_address(event_result["args"]["ipId"]) + token_id = event_result["args"]["tokenId"] + token_contract = self.web3.to_checksum_address(event_result["args"]["tokenContract"]) + + # Parse license terms for this IP + license_terms_ids = self._parse_tx_license_terms_attached_event_for_ip( + response["tx_receipt"], ip_id + ) + + results.append({ + "ip_id": ip_id, + "token_id": token_id, + "spg_nft_contract": token_contract, + "license_terms_ids": license_terms_ids, + }) + + return { + "tx_hash": response["tx_hash"], + "results": results, + } + + except Exception as e: + raise ValueError(f"Failed to batch mint and register IP and attach PIL terms: {str(e)}") + @deprecated("Use register_ip_asset() instead.") def mint_and_register_ip( self, @@ -2214,6 +2340,31 @@ def _parse_tx_license_terms_attached_event(self, tx_receipt: dict) -> list[int]: return license_terms_ids + def _parse_tx_license_terms_attached_event_for_ip(self, tx_receipt: dict, ip_id: str) -> list[int]: + """ + Parse the LicenseTermsAttached events for a specific IP from a transaction receipt. + + :param tx_receipt dict: The transaction receipt. + :param ip_id str: The IP ID to filter events for. + :return list: A list of license terms IDs for the specified IP. + """ + event_signature = self.web3.keccak( + text="LicenseTermsAttached(address,address,address,uint256)" + ).hex() + license_terms_ids = [] + + for log in tx_receipt["logs"]: + if log["topics"][0].hex() == event_signature: + # Parse the full event to get ipId + event_result = self.licensing_module_client.contract.events.LicenseTermsAttached.process_log(log) + log_ip_id = event_result["args"]["ipId"] + + if log_ip_id.lower() == ip_id.lower(): + license_terms_id = event_result["args"]["licenseTermsId"] + license_terms_ids.append(license_terms_id) + + return license_terms_ids + def get_royalty_vault_address_by_ip_id( self, tx_receipt: dict, ipId: Address ) -> Address: diff --git a/tests/integration/test_integration_ip_asset.py b/tests/integration/test_integration_ip_asset.py index f1ec1bf8..45b59fb0 100644 --- a/tests/integration/test_integration_ip_asset.py +++ b/tests/integration/test_integration_ip_asset.py @@ -1186,6 +1186,135 @@ def test_batch_mint_and_register_ip( assert isinstance(ip_registered["ip_id"], str) and ip_registered["ip_id"] assert isinstance(ip_registered["token_id"], int) + def test_batch_mint_and_register_ip_asset_with_pil_terms( + self, story_client: StoryClient, public_nft_collection + ): + """Test batch minting and registering IP with PIL terms""" + + # Define license terms template + license_terms_template = { + "terms": { + "transferable": True, + "royalty_policy": ROYALTY_POLICY, + "default_minting_fee": 100, + "expiration": 0, + "commercial_use": True, + "commercial_attribution": False, + "commercializer_checker": ZERO_ADDRESS, + "commercializer_checker_data": ZERO_ADDRESS, + "commercial_rev_share": 10, + "commercial_rev_ceiling": 0, + "derivatives_allowed": True, + "derivatives_attribution": True, + "derivatives_approval": False, + "derivatives_reciprocal": True, + "derivative_rev_ceiling": 0, + "currency": WIP_TOKEN_ADDRESS, + "uri": "", + }, + "licensing_config": { + "is_set": True, + "minting_fee": 100, + "hook_data": ZERO_ADDRESS, + "licensing_hook": ZERO_ADDRESS, + "commercial_rev_share": 0, + "disabled": False, + "expect_minimum_group_reward_share": 0, + "expect_group_reward_pool": ZERO_ADDRESS, + }, + } + + # Test with two IPs (use allow_duplicates=True to avoid duplicate license terms error) + response = story_client.IPAsset.batch_mint_and_register_ip_asset_with_pil_terms( + args=[ + { + "spg_nft_contract": public_nft_collection, + "terms": [license_terms_template], + "allow_duplicates": True, + }, + { + "spg_nft_contract": public_nft_collection, + "terms": [license_terms_template], + "allow_duplicates": True, + }, + ] + ) + + # Verify response structure + assert isinstance(response["tx_hash"], str) and response["tx_hash"] + assert isinstance(response["results"], list) + assert len(response["results"]) == 2 + + # Verify each result + for result in response["results"]: + assert isinstance(result["ip_id"], str) and result["ip_id"] + assert isinstance(result["token_id"], int) + assert isinstance(result["spg_nft_contract"], str) + assert isinstance(result["license_terms_ids"], list) + assert len(result["license_terms_ids"]) >= 1 + + # Verify IPs are registered + for result in response["results"]: + is_registered = story_client.IPAsset.is_registered(result["ip_id"]) + assert is_registered is True + + def test_batch_mint_with_metadata_and_recipient( + self, story_client: StoryClient, public_nft_collection + ): + """Test batch minting with metadata and custom recipient""" + + license_terms_template = { + "terms": { + "transferable": True, + "royalty_policy": ROYALTY_POLICY, + "default_minting_fee": 100, + "expiration": 0, + "commercial_use": True, + "commercial_attribution": False, + "commercializer_checker": ZERO_ADDRESS, + "commercializer_checker_data": ZERO_ADDRESS, + "commercial_rev_share": 10, + "commercial_rev_ceiling": 0, + "derivatives_allowed": True, + "derivatives_attribution": True, + "derivatives_approval": False, + "derivatives_reciprocal": True, + "derivative_rev_ceiling": 0, + "currency": WIP_TOKEN_ADDRESS, + "uri": "", + }, + "licensing_config": { + "is_set": True, + "minting_fee": 100, + "hook_data": ZERO_ADDRESS, + "licensing_hook": ZERO_ADDRESS, + "commercial_rev_share": 0, + "disabled": False, + "expect_minimum_group_reward_share": 0, + "expect_group_reward_pool": ZERO_ADDRESS, + }, + } + + response = story_client.IPAsset.batch_mint_and_register_ip_asset_with_pil_terms( + args=[ + { + "spg_nft_contract": public_nft_collection, + "terms": [license_terms_template], + "ip_metadata": { + "ip_metadata_uri": "https://example.com/ip1", + "ip_metadata_hash": web3.keccak(text="ip1-metadata"), + }, + "recipient": account_2.address, + "allow_duplicates": True, + } + ] + ) + + assert isinstance(response["tx_hash"], str) and response["tx_hash"] + assert len(response["results"]) == 1 + assert isinstance(response["results"][0]["ip_id"], str) + assert isinstance(response["results"][0]["license_terms_ids"], list) + class TestRegisterIpAsset: """Test suite for the unified register_ip_asset method that supports 6 different workflows""" diff --git a/tests/unit/resources/test_batch_mint_and_register_ip_asset_with_pil_terms.py b/tests/unit/resources/test_batch_mint_and_register_ip_asset_with_pil_terms.py new file mode 100644 index 00000000..eec0fa32 --- /dev/null +++ b/tests/unit/resources/test_batch_mint_and_register_ip_asset_with_pil_terms.py @@ -0,0 +1,320 @@ +"""Unit tests for batch_mint_and_register_ip_asset_with_pil_terms functionality.""" + +from unittest.mock import Mock, patch + +import pytest + +from story_protocol_python_sdk.resources.IPAsset import IPAsset + +# Test constants +SPG_NFT_CONTRACT = "0x1234567890123456789012345678901234567890" +IP_ID_1 = "0xabcdef1234567890123456789012345678901234" +IP_ID_2 = "0xabcdef1234567890123456789012345678901235" +TX_HASH = "0x129f7dd802200f096221dd89d5b086e4bd3ad6eafb378a0c75e3b04fc375f997" +ZERO_HASH = "0x0000000000000000000000000000000000000000000000000000000000000000" +ROYALTY_POLICY = "0xBe54FB168b3c982b7AaE60dB6CF75Bd8447b390E" +ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" +WIP_TOKEN_ADDRESS = "0x1514000000000000000000000000000000000000" + + +@pytest.fixture +def ip_asset_client(mock_web3, mock_account): + """Create IPAsset client for testing.""" + return IPAsset(mock_web3, mock_account, chain_id=1516) + + +class TestBatchMintAndRegisterIpAssetWithPilTerms: + """Test suite for batch_mint_and_register_ip_asset_with_pil_terms.""" + + def test_batch_mint_successful_with_two_ips(self, ip_asset_client): + """Test successful batch minting of two IPs""" + license_terms_template = { + "terms": { + "transferable": True, + "royalty_policy": ROYALTY_POLICY, + "default_minting_fee": 100, + "expiration": 0, + "commercial_use": True, + "commercial_attribution": False, + "commercializer_checker": ZERO_ADDRESS, + "commercializer_checker_data": ZERO_ADDRESS, + "commercial_rev_share": 10, + "commercial_rev_ceiling": 0, + "derivatives_allowed": True, + "derivatives_attribution": True, + "derivatives_approval": False, + "derivatives_reciprocal": True, + "derivative_rev_ceiling": 0, + "currency": WIP_TOKEN_ADDRESS, + "uri": "", + }, + "licensing_config": { + "is_set": True, + "minting_fee": 100, + "hook_data": ZERO_ADDRESS, + "licensing_hook": ZERO_ADDRESS, + "commercial_rev_share": 0, + "disabled": False, + "expect_minimum_group_reward_share": 0, + "expect_group_reward_pool": ZERO_ADDRESS, + }, + } + + # Create mock logs for IPRegistered events + mock_topic = Mock() + mock_topic.hex.return_value = "0xip_registered_sig" + + mock_log_1 = {"topics": [mock_topic]} + mock_log_2 = {"topics": [mock_topic]} + + # Mock keccak for event signature + mock_keccak_result = Mock() + mock_keccak_result.hex.return_value = "0xip_registered_sig" + ip_asset_client.web3.keccak = Mock(return_value=mock_keccak_result) + + # Mock to_checksum_address + ip_asset_client.web3.to_checksum_address = Mock(side_effect=lambda x: x) + + # Mock event processing + mock_event_1 = {"args": {"ipId": IP_ID_1, "tokenId": 1, "tokenContract": SPG_NFT_CONTRACT}} + mock_event_2 = {"args": {"ipId": IP_ID_2, "tokenId": 2, "tokenContract": SPG_NFT_CONTRACT}} + ip_asset_client.ip_asset_registry_client.contract.events.IPRegistered.process_log = Mock( + side_effect=[mock_event_1, mock_event_2] + ) + + # Mock mint_and_register_ip_asset_with_pil_terms to return encoded data + with patch.object( + ip_asset_client, + "mint_and_register_ip_asset_with_pil_terms", + side_effect=[ + {"encoded_tx_data": "0x1234"}, + {"encoded_tx_data": "0x5678"}, + ], + ): + # Mock build_and_send_transaction + with patch( + "story_protocol_python_sdk.resources.IPAsset.build_and_send_transaction", + return_value={ + "tx_hash": TX_HASH, + "tx_receipt": {"logs": [mock_log_1, mock_log_2]}, + }, + ): + # Mock license terms parsing + with patch.object( + ip_asset_client, + "_parse_tx_license_terms_attached_event_for_ip", + side_effect=[[1, 2], [3]], + ): + result = ip_asset_client.batch_mint_and_register_ip_asset_with_pil_terms( + args=[ + { + "spg_nft_contract": SPG_NFT_CONTRACT, + "terms": [license_terms_template], + }, + { + "spg_nft_contract": SPG_NFT_CONTRACT, + "terms": [license_terms_template], + }, + ] + ) + + assert result["tx_hash"] == TX_HASH + assert len(result["results"]) == 2 + assert result["results"][0]["ip_id"] == IP_ID_1 + assert result["results"][0]["token_id"] == 1 + assert result["results"][0]["spg_nft_contract"] == SPG_NFT_CONTRACT + assert result["results"][0]["license_terms_ids"] == [1, 2] + assert result["results"][1]["ip_id"] == IP_ID_2 + assert result["results"][1]["token_id"] == 2 + assert result["results"][1]["license_terms_ids"] == [3] + + def test_batch_mint_with_metadata(self, ip_asset_client): + """Test batch minting with IP metadata""" + license_terms_template = { + "terms": { + "transferable": True, + "royalty_policy": ROYALTY_POLICY, + "default_minting_fee": 100, + "expiration": 0, + "commercial_use": True, + "commercial_attribution": False, + "commercializer_checker": ZERO_ADDRESS, + "commercializer_checker_data": ZERO_ADDRESS, + "commercial_rev_share": 10, + "commercial_rev_ceiling": 0, + "derivatives_allowed": True, + "derivatives_attribution": True, + "derivatives_approval": False, + "derivatives_reciprocal": True, + "derivative_rev_ceiling": 0, + "currency": WIP_TOKEN_ADDRESS, + "uri": "", + }, + "licensing_config": { + "is_set": True, + "minting_fee": 100, + "hook_data": ZERO_ADDRESS, + "licensing_hook": ZERO_ADDRESS, + "commercial_rev_share": 0, + "disabled": False, + "expect_minimum_group_reward_share": 0, + "expect_group_reward_pool": ZERO_ADDRESS, + }, + } + + # Create mock log for IPRegistered event + mock_topic = Mock() + mock_topic.hex.return_value = "0xip_registered_sig" + mock_log = {"topics": [mock_topic]} + + # Mock keccak for event signature + mock_keccak_result = Mock() + mock_keccak_result.hex.return_value = "0xip_registered_sig" + ip_asset_client.web3.keccak = Mock(return_value=mock_keccak_result) + + # Mock to_checksum_address + ip_asset_client.web3.to_checksum_address = Mock(side_effect=lambda x: x) + + # Mock event processing + mock_event = {"args": {"ipId": IP_ID_1, "tokenId": 1, "tokenContract": SPG_NFT_CONTRACT}} + ip_asset_client.ip_asset_registry_client.contract.events.IPRegistered.process_log = Mock( + return_value=mock_event + ) + + with patch.object( + ip_asset_client, + "mint_and_register_ip_asset_with_pil_terms", + return_value={"encoded_tx_data": "0x1234"}, + ): + with patch( + "story_protocol_python_sdk.resources.IPAsset.build_and_send_transaction", + return_value={ + "tx_hash": TX_HASH, + "tx_receipt": {"logs": [mock_log]}, + }, + ): + with patch.object( + ip_asset_client, + "_parse_tx_license_terms_attached_event_for_ip", + return_value=[1], + ): + result = ip_asset_client.batch_mint_and_register_ip_asset_with_pil_terms( + args=[ + { + "spg_nft_contract": SPG_NFT_CONTRACT, + "terms": [license_terms_template], + "ip_metadata": { + "ip_metadata_uri": "https://example.com/metadata", + "ip_metadata_hash": ZERO_HASH, + }, + } + ] + ) + + assert result["tx_hash"] == TX_HASH + assert len(result["results"]) == 1 + + def test_batch_mint_with_recipient(self, ip_asset_client): + """Test batch minting with custom recipient""" + recipient = "0x9999999999999999999999999999999999999999" + license_terms_template = { + "terms": { + "transferable": True, + "royalty_policy": ROYALTY_POLICY, + "default_minting_fee": 100, + "expiration": 0, + "commercial_use": True, + "commercial_attribution": False, + "commercializer_checker": ZERO_ADDRESS, + "commercializer_checker_data": ZERO_ADDRESS, + "commercial_rev_share": 10, + "commercial_rev_ceiling": 0, + "derivatives_allowed": True, + "derivatives_attribution": True, + "derivatives_approval": False, + "derivatives_reciprocal": True, + "derivative_rev_ceiling": 0, + "currency": WIP_TOKEN_ADDRESS, + "uri": "", + }, + "licensing_config": { + "is_set": True, + "minting_fee": 100, + "hook_data": ZERO_ADDRESS, + "licensing_hook": ZERO_ADDRESS, + "commercial_rev_share": 0, + "disabled": False, + "expect_minimum_group_reward_share": 0, + "expect_group_reward_pool": ZERO_ADDRESS, + }, + } + + # Create mock log for IPRegistered event + mock_topic = Mock() + mock_topic.hex.return_value = "0xip_registered_sig" + mock_log = {"topics": [mock_topic]} + + # Mock keccak for event signature + mock_keccak_result = Mock() + mock_keccak_result.hex.return_value = "0xip_registered_sig" + ip_asset_client.web3.keccak = Mock(return_value=mock_keccak_result) + + # Mock to_checksum_address + ip_asset_client.web3.to_checksum_address = Mock(side_effect=lambda x: x) + + # Mock event processing + mock_event = {"args": {"ipId": IP_ID_1, "tokenId": 1, "tokenContract": SPG_NFT_CONTRACT}} + ip_asset_client.ip_asset_registry_client.contract.events.IPRegistered.process_log = Mock( + return_value=mock_event + ) + + with patch.object( + ip_asset_client, + "mint_and_register_ip_asset_with_pil_terms", + return_value={"encoded_tx_data": "0x1234"}, + ): + with patch( + "story_protocol_python_sdk.resources.IPAsset.build_and_send_transaction", + return_value={ + "tx_hash": TX_HASH, + "tx_receipt": {"logs": [mock_log]}, + }, + ): + with patch.object( + ip_asset_client, + "_parse_tx_license_terms_attached_event_for_ip", + return_value=[1], + ): + result = ip_asset_client.batch_mint_and_register_ip_asset_with_pil_terms( + args=[ + { + "spg_nft_contract": SPG_NFT_CONTRACT, + "terms": [license_terms_template], + "recipient": recipient, + } + ] + ) + + assert result["tx_hash"] == TX_HASH + assert len(result["results"]) == 1 + + def test_batch_mint_empty_args(self, ip_asset_client): + """Test batch minting with empty args""" + with patch( + "story_protocol_python_sdk.resources.IPAsset.build_and_send_transaction", + return_value={ + "tx_hash": TX_HASH, + "tx_receipt": {"logs": []}, + }, + ): + with patch.object( + ip_asset_client, + "_parse_tx_ip_registered_event", + return_value=[], + ): + result = ip_asset_client.batch_mint_and_register_ip_asset_with_pil_terms( + args=[] + ) + + assert result["tx_hash"] == TX_HASH + assert len(result["results"]) == 0