diff --git a/src/story_protocol_python_sdk/resources/IPAsset.py b/src/story_protocol_python_sdk/resources/IPAsset.py index 5066ad2..5328940 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 f1ec1bf..45b59fb 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 0000000..eec0fa3 --- /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