From c1c9d87be600afe5654a1731957a301774b619a0 Mon Sep 17 00:00:00 2001 From: Boris Polania Date: Fri, 24 Jan 2025 12:12:42 -0800 Subject: [PATCH] Integration Tests Update (#32) * Update test_integration_ip_asset.py * Update test_integration_ip_asset.py * Update test_integration_ip_asset.py IP Assets tests * Update test_integration_license.py License test cases * Update test_integration_ip_account.py Ip account tests * Update test_integration_nft_client.py Add NFT Client tests * Permission and Royalty tests added * Update test_integration_royalty.py * Update tests/integration/test_integration_ip_asset.py Co-authored-by: Draco * Update test_integration_license.py --------- Co-authored-by: Draco --- .../test_integration_ip_account.py | 185 ++++++ .../integration/test_integration_ip_asset.py | 567 ++++++++++++------ tests/integration/test_integration_license.py | 124 ++++ .../test_integration_nft_client.py | 116 +++- .../test_integration_permission.py | 121 +++- tests/integration/test_integration_royalty.py | 131 +++- 6 files changed, 1048 insertions(+), 196 deletions(-) diff --git a/tests/integration/test_integration_ip_account.py b/tests/integration/test_integration_ip_account.py index 93fd5db..18fb657 100644 --- a/tests/integration/test_integration_ip_account.py +++ b/tests/integration/test_integration_ip_account.py @@ -154,3 +154,188 @@ def test_executeWithSig(story_client): 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." + +# Boris tests + +def test_get_ip_account_nonce(story_client): + """Test getting IP Account nonce.""" + # First register an IP + 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'] + + # Get the nonce + state = story_client.IPAccount.getIpAccountNonce(ip_id) + + assert state is not None + assert isinstance(state, bytes) + +def test_execute_with_encoded_data(story_client): + """Test execute with pre-encoded function data.""" + # First register an IP + 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'] + + # Encode setPermission function data + data = story_client.IPAccount.access_controller_client.contract.encode_abi( + abi_element_identifier="setPermission", + args=[ + ip_id, + account.address, + "0x89630Ccf23277417FBdfd3076C702F5248267e78", # Module address + Web3.keccak(text="function execute(address,uint256,bytes,uint8)")[:4], # Function selector + 1 # Permission level + ] + ) + + 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 + +def test_execute_with_sig_multiple_permissions(story_client): + """Test executeWithSig setting multiple permissions.""" + # First register an IP + 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) + + # Encode multiple setPermission calls + function_signatures = [ + "function setAll(address,string,bytes32,bytes32)", + "function execute(address,uint256,bytes,uint8)", + "function registerDerivative(address,address[],uint256[],address,bytes)" + ] + + data_list = [] + for func_sig in function_signatures: + data = story_client.IPAccount.access_controller_client.contract.encode_abi( + abi_element_identifier="setPermission", + args=[ + ip_id, + account.address, + "0x6E81a25C99C6e8430aeC7353325EB138aFE5DC16", # Module address + Web3.keccak(text=func_sig)[:4], # Function selector + 1 # ALLOW + ] + ) + # Remove '0x' prefix before adding to list + if data.startswith('0x'): + data = data[2:] + data_list.append(data) + + # Combine all encoded data with '0x' prefix + combined_data = '0x' + ''.join(data_list) + + # Create and sign EIP-712 message + 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"}, + ], + } + + # Compute expected state + expected_state = Web3.keccak( + encode( + ["bytes32", "bytes"], + [ + state, + Web3.to_bytes(hexstr=combined_data) + ] + ) + ) + + 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 + +def test_execute_invalid_address(story_client): + """Test execute with invalid address should raise error.""" + # First register an IP + 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 + ) + + assert "is not a valid address" in str(exc_info.value) + +def test_execute_unregistered_ip(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 + ) + + assert "is not registered" in str(exc_info.value) \ No newline at end of file diff --git a/tests/integration/test_integration_ip_asset.py b/tests/integration/test_integration_ip_asset.py index 2296e07..578305e 100644 --- a/tests/integration/test_integration_ip_asset.py +++ b/tests/integration/test_integration_ip_asset.py @@ -3,8 +3,8 @@ import os import sys import pytest -from dotenv import load_dotenv from web3 import Web3 +from dotenv import load_dotenv # Ensure the src directory is in the Python path current_dir = os.path.dirname(__file__) @@ -31,119 +31,29 @@ 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(scope="module") def story_client(): return get_story_client_in_devnet(web3, account) +@pytest.fixture(scope="module") +def non_commercial_license_terms_id(story_client): + response = story_client.License.registerNonComSocialRemixingPIL() + return response['licenseTermsId'] + @pytest.fixture(scope="module") def parent_ip_id(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 ) - - assert 'txHash' in response - assert isinstance(response['txHash'], str) - - assert response is not None - assert 'ipId' in response - assert response['ipId'] is not None return response['ipId'] -def test_register_ip_asset(story_client, parent_ip_id): - assert parent_ip_id is not None - -def test_register_ip_asset_with_metadata(story_client): - token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - metadata = { - 'metadataURI': "test-uri", - 'metadataHash': web3.to_hex(web3.keccak(text="test-metadata-hash")), - 'nftMetadataHash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) - } - - response = story_client.IPAsset.register( - nft_contract=MockERC721, - token_id=token_id, - ip_metadata=metadata, - deadline=1000 - ) - - assert 'txHash' in response - assert isinstance(response['txHash'], str) - - assert response is not None - assert 'ipId' in response - assert response['ipId'] is not None - assert isinstance(response['ipId'], str) - -# @pytest.fixture(scope="module") -# def attach_non_commercial_license(story_client, parent_ip_id): -# license_template = "0x8BB1ADE72E21090Fc891e1d4b88AC5E57b27cB31" -# no_commercial_license_terms_id = 1 - -# attach_license_response = story_client.License.attachLicenseTerms( -# ip_id=parent_ip_id, -# license_template=license_template, -# license_terms_id=no_commercial_license_terms_id -# ) - -# def test_register_derivative(story_client, parent_ip_id, attach_non_commercial_license): -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# child_response = story_client.IPAsset.register( -# token_contract=MockERC721, -# token_id=token_id -# ) -# child_ip_id = child_response['ipId'] - -# response = story_client.IPAsset.registerDerivative( -# child_ip_id=child_ip_id, -# parent_ip_ids=[parent_ip_id], -# license_terms_ids=[2], -# license_template="0x8BB1ADE72E21090Fc891e1d4b88AC5E57b27cB31" -# ) - -# assert response is not None -# assert 'txHash' in response -# assert response['txHash'] is not None -# assert isinstance(response['txHash'], str) -# assert len(response['txHash']) > 0 - -# def test_registerDerivativeWithLicenseTokens(story_client, parent_ip_id, attach_non_commercial_license): -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) -# child_response = story_client.IPAsset.register( -# token_contract=MockERC721, -# token_id=token_id -# ) -# child_ip_id = child_response['ipId'] - -# license_token_response = story_client.License.mintLicenseTokens( -# licensor_ip_id=parent_ip_id, -# license_template="0x8BB1ADE72E21090Fc891e1d4b88AC5E57b27cB31", -# license_terms_id=1, -# amount=1, -# receiver=account.address -# ) -# licenseTokenIds = license_token_response['licenseTokenIds'] - -# response = story_client.IPAsset.registerDerivativeWithLicenseTokens( -# child_ip_id=child_ip_id, -# license_token_ids=licenseTokenIds -# ) - -# assert response is not None -# assert 'txHash' in response -# assert response['txHash'] is not None -# assert isinstance(response['txHash'], str) -# assert len(response['txHash']) > 0 - @pytest.fixture(scope="module") def nft_collection(story_client): - txData = story_client.NFTClient.createNFTCollection( + tx_data = story_client.NFTClient.createNFTCollection( name="test-collection", symbol="TEST", max_supply=100, @@ -152,22 +62,373 @@ def nft_collection(story_client): contract_uri="test-uri", mint_fee_recipient=account.address, ) - return txData['nftContract'] + return tx_data['nftContract'] + +class TestIPAssetCreation: + def test_register_ip_asset(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 + ) + assert 'ipId' in response + assert isinstance(response['ipId'], str) + assert len(response['ipId']) > 0 + + def test_register_derivative(self, story_client, parent_ip_id, no_commercial_license_terms_id): + + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + child_response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + child_ip_id = child_response['ipId'] + + + story_client.License.attachLicenseTerms( + ip_id=parent_ip_id, + license_template="0x2E896b0b2Fdb7457499B56AAaA4AE55BCB4Cd316", + license_terms_id=no_commercial_license_terms_id + ) + + # Register derivative + response = story_client.IPAsset.registerDerivative( + child_ip_id=child_ip_id, + parent_ip_ids=[parent_ip_id], + license_terms_ids=[no_commercial_license_terms_id] + ) + + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + + def test_register_derivative_with_license_tokens(self, story_client, parent_ip_id, no_commercial_license_terms_id): + # Register child IP + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + child_response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id, + + ) + child_ip_id = child_response['ipId'] + + # Mint license tokens + mint_response = story_client.License.mintLicenseTokens( + licensor_ip_id=parent_ip_id, + license_template="0x2E896b0b2Fdb7457499B56AAaA4AE55BCB4Cd316", + license_terms_id=no_commercial_license_terms_id, + amount=1, + receiver=account.address + ) + + # Register derivative with license tokens + response = story_client.IPAsset.registerDerivativeWithLicenseTokens( + child_ip_id=child_ip_id, + license_token_ids=mint_response['licenseTokenIds'] + ) + + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + + def test_is_registered(self, story_client, parent_ip_id): + is_registered = story_client.IPAsset._is_registered(parent_ip_id) + assert is_registered is True + + unregistered_ip_id = "0x1234567890123456789012345678901234567890" + is_registered = story_client.IPAsset._is_registered(unregistered_ip_id) + assert is_registered is False + +class TestNFTClientSPG: + @pytest.fixture(scope="class") + def setup(self, story_client, nft_collection): + result = story_client.IPAsset.mintAndRegisterIpAssetWithPilTerms( + spg_nft_contract=nft_collection, + terms=[{ + 'terms': { + 'transferable': True, + 'royalty_policy': ROYALTY_POLICY, + 'default_minting_fee': 1, + 'expiration': 0, + 'commercial_use': True, + 'commercial_attribution': False, + 'commercializer_checker': ZERO_ADDRESS, + 'commercializer_checker_data': ZERO_ADDRESS, + 'commercial_rev_share': 90, + 'commercial_rev_ceiling': 0, + 'derivatives_allowed': True, + 'derivatives_attribution': True, + 'derivatives_approval': False, + 'derivatives_reciprocal': True, + 'derivative_rev_ceiling': 0, + 'currency': MockERC20, + 'uri': "" + }, + 'licensing_config': { + 'is_set': True, + 'minting_fee': 1, + 'hook_data': "", + 'licensing_hook': ZERO_ADDRESS, + 'commercial_rev_share': 90, + 'disabled': False, + 'expect_minimum_group_reward_share': 0, + 'expect_group_reward_pool': ZERO_ADDRESS + } + }], + ) + return { + 'parent_ip_id': result['ipId'], + 'license_terms_id': result['licenseTermsIds'][0], + 'nft_contract': nft_collection + } + + def test_register_ip_with_metadata(self, story_client, setup): + token_id = get_token_id(setup['nft_contract'], story_client.web3, story_client.account) + response = story_client.IPAsset.register( + nft_contract=setup['nft_contract'], + token_id=token_id, + ip_metadata={ + 'ipMetadataURI': "test-uri", + 'ipMetadataHash': web3.to_hex(web3.keccak(text="test-metadata-hash")), + 'nftMetadataHash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) + }, + deadline=1000 + ) + + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert 'ipId' in response + assert isinstance(response['ipId'], str) + + def test_register_derivative_ip(self, story_client, setup): + token_id = get_token_id(setup['nft_contract'], story_client.web3, story_client.account) + + response = story_client.IPAsset.registerDerivativeIp( + nft_contract=setup['nft_contract'], + token_id=token_id, + deriv_data={ + 'parentIpIds': [setup['parent_ip_id']], + 'licenseTermsIds': [setup['license_terms_id']] + }, + metadata={ + 'metadataURI': "test-uri", + 'metadataHash': web3.to_hex(web3.keccak(text="test-metadata-hash")), + }, + deadline=1000 + ) + + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert 'ipId' in response + assert isinstance(response['ipId'], str) + + def test_register_ip_and_attach_pil_terms(self, story_client, setup): + token_id = get_token_id(setup['nft_contract'], story_client.web3, story_client.account) + + result = story_client.IPAsset.registerIpAndAttachPilTerms( + nft_contract=setup['nft_contract'], + token_id=token_id, + pil_type='non_commercial_remix', + metadata={ + 'metadataURI': "test-uri", + 'metadataHash': web3.to_hex(web3.keccak(text="test-metadata-hash")), + 'nftMetadataHash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) + }, + deadline=1000, + commercial_rev_share=10, + currency=MockERC20 + ) + + assert 'txHash' in result + assert isinstance(result['txHash'], str) + assert 'ipId' in result + assert isinstance(result['ipId'], str) + assert 'licenseTermsIds' in result + assert isinstance(result['licenseTermsIds'], list) + +def test_batch_register_ip_assets(story_client, nft_collection): + """Test batch registration of IP assets.""" + token_id1 = get_token_id(MockERC721, story_client.web3, story_client.account) + token_id2 = get_token_id(MockERC721, story_client.web3, story_client.account) + + response = story_client.IPAsset.batchRegister([ + { + 'nft_contract': MockERC721, + 'token_id': token_id1, + }, + { + 'nft_contract': MockERC721, + 'token_id': token_id2, + 'ip_metadata': { + 'ip_metadata_uri': "test-uri", + 'ip_metadata_hash': web3.to_hex(web3.keccak(text="test-metadata-hash")), + 'nft_metadata_hash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) + } + } + ]) + + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert 'results' in response + assert isinstance(response['results'], list) + assert len(response['results']) == 2 + +def test_batch_mint_and_register_ip_with_pil_terms(story_client, nft_collection): + """Test batch minting and registering IP assets with PIL terms.""" + response = story_client.IPAsset.batchMintAndRegisterIpAssetWithPilTerms([ + { + 'spg_nft_contract': nft_collection, + 'terms': [{ + 'terms': { + 'transferable': True, + 'royalty_policy': ROYALTY_POLICY, + 'default_minting_fee': 1, + 'expiration': 0, + 'commercial_use': False, + 'commercial_attribution': False, + 'commercializer_checker': ZERO_ADDRESS, + 'commercializer_checker_data': ZERO_ADDRESS, + 'commercial_rev_share': 0, + 'commercial_rev_ceiling': 0, + 'derivatives_allowed': True, + 'derivatives_attribution': True, + 'derivatives_approval': False, + 'derivatives_reciprocal': True, + 'derivative_rev_ceiling': 0, + 'currency': MockERC20, + 'uri': "" + }, + 'licensing_config': { + 'is_set': True, + 'minting_fee': 1, + 'hook_data': "", + 'licensing_hook': ZERO_ADDRESS, + 'commercial_rev_share': 0, + 'disabled': False, + 'expect_minimum_group_reward_share': 0, + 'expect_group_reward_pool': ZERO_ADDRESS + } + }] + }, + { + 'spg_nft_contract': nft_collection, + 'terms': [{ + 'terms': { + 'transferable': True, + 'royalty_policy': ROYALTY_POLICY, + 'default_minting_fee': 8, + 'expiration': 0, + 'commercial_use': True, + 'commercial_attribution': False, + 'commercializer_checker': ZERO_ADDRESS, + 'commercializer_checker_data': ZERO_ADDRESS, + 'commercial_rev_share': 90, + 'commercial_rev_ceiling': 0, + 'derivatives_allowed': True, + 'derivatives_attribution': True, + 'derivatives_approval': False, + 'derivatives_reciprocal': True, + 'derivative_rev_ceiling': 0, + 'currency': MockERC20, + 'uri': "" + }, + 'licensing_config': { + 'is_set': True, + 'minting_fee': 8, + 'hook_data': "", + 'licensing_hook': ZERO_ADDRESS, + 'commercial_rev_share': 90, + 'disabled': False, + 'expect_minimum_group_reward_share': 0, + 'expect_group_reward_pool': ZERO_ADDRESS + } + }] + } + ]) + + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert 'results' in response + assert isinstance(response['results'], list) + assert len(response['results']) == 2 + for result in response['results']: + assert 'ipId' in result + assert 'licenseTermsIds' in result + assert isinstance(result['licenseTermsIds'], list) + +def test_register_ip_with_royalty_distribution(story_client, nft_collection): + """Test registering IP with royalty distribution.""" + token_id = get_token_id(nft_collection, story_client.web3, story_client.account) + + response = story_client.IPAsset.registerIPAndAttachLicenseTermsAndDistributeRoyaltyTokens( + nft_contract=nft_collection, + token_id=token_id, + terms=[{ + 'terms': { + 'transferable': True, + 'royalty_policy': ROYALTY_POLICY, + 'default_minting_fee': 10000, + 'expiration': 1000, + 'commercial_use': True, + 'commercial_attribution': False, + 'commercializer_checker': ZERO_ADDRESS, + 'commercializer_checker_data': ZERO_ADDRESS, + 'commercial_rev_share': 0, + 'commercial_rev_ceiling': 0, + 'derivatives_allowed': True, + 'derivatives_attribution': True, + 'derivatives_approval': False, + 'derivatives_reciprocal': True, + 'derivative_rev_ceiling': 0, + 'currency': MockERC20, + 'uri': "test case" + }, + 'licensing_config': { + 'is_set': True, + 'minting_fee': 10000, + 'hook_data': "", + 'licensing_hook': ZERO_ADDRESS, + 'commercial_rev_share': 0, + 'disabled': False, + 'expect_minimum_group_reward_share': 0, + 'expect_group_reward_pool': ZERO_ADDRESS + } + }], + ip_metadata={ + 'ip_metadata_uri': "test-uri", + 'ip_metadata_hash': web3.to_hex(web3.keccak(text="test-metadata-hash")), + 'nft_metadata_hash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) + }, + royalty_shares=[{ + 'author': story_client.account.address, + 'percentage': 1 + }] + ) -def test_mint_register_attach_terms(story_client, nft_collection): - response = story_client.IPAsset.mintAndRegisterIpAssetWithPilTerms( + assert 'registerIpAndAttachPilTermsAndDeployRoyaltyVaultTxHash' in response + assert isinstance(response['registerIpAndAttachPilTermsAndDeployRoyaltyVaultTxHash'], str) + assert 'distributeRoyaltyTokensTxHash' in response + assert isinstance(response['distributeRoyaltyTokensTxHash'], str) + assert 'ipId' in response + assert isinstance(response['ipId'], str) + assert 'licenseTermsIds' in response + assert isinstance(response['licenseTermsIds'], list) + +def test_mint_register_and_distribute_royalty(story_client, nft_collection): + """Test minting, registering IP and distributing royalty tokens in one transaction.""" + response = story_client.IPAsset.mintAndRegisterIpAndAttachPilTermsAndDistributeRoyaltyTokens( spg_nft_contract=nft_collection, terms=[{ 'terms': { 'transferable': True, 'royalty_policy': ROYALTY_POLICY, - 'default_minting_fee': 1, - 'expiration': 0, + 'default_minting_fee': 10000, + 'expiration': 1000, 'commercial_use': True, 'commercial_attribution': False, 'commercializer_checker': ZERO_ADDRESS, 'commercializer_checker_data': ZERO_ADDRESS, - 'commercial_rev_share': 90, + 'commercial_rev_share': 0, 'commercial_rev_ceiling': 0, 'derivatives_allowed': True, 'derivatives_attribution': True, @@ -175,107 +436,35 @@ def test_mint_register_attach_terms(story_client, nft_collection): 'derivatives_reciprocal': True, 'derivative_rev_ceiling': 0, 'currency': MockERC20, - 'uri': "" + 'uri': "test case" }, 'licensing_config': { 'is_set': True, - 'minting_fee': 1, + 'minting_fee': 10000, 'hook_data': "", 'licensing_hook': ZERO_ADDRESS, - 'commercial_rev_share': 90, + 'commercial_rev_share': 0, 'disabled': False, 'expect_minimum_group_reward_share': 0, 'expect_group_reward_pool': ZERO_ADDRESS } }], + ip_metadata={ + 'ip_metadata_uri': "test-uri", + 'ip_metadata_hash': web3.to_hex(web3.keccak(text="test-metadata-hash")), + 'nft_metadata_hash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) + }, + royalty_shares=[{ + 'author': story_client.account.address, + 'percentage': 10 # 100% + }] ) assert 'txHash' in response assert isinstance(response['txHash'], str) - assert 'ipId' in response assert isinstance(response['ipId'], str) - assert response['ipId'].startswith("0x") - - assert 'tokenId' in response - assert isinstance(response['tokenId'], int) - assert 'licenseTermsIds' in response assert isinstance(response['licenseTermsIds'], list) - assert all(isinstance(id, int) for id in response['licenseTermsIds']) - -# def test_register_attach(story_client, nft_collection): -# token_id = get_token_id(nft_collection, story_client.web3, story_client.account) - -# pil_type = 'non_commercial_remix' -# metadata = { -# 'metadataURI': "test-uri", -# 'metadataHash': web3.to_hex(web3.keccak(text="test-metadata-hash")), -# 'nftMetadataHash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) -# } -# deadline = getBlockTimestamp(web3) + 1000 - -# response = story_client.IPAsset.registerIpAndAttachPilTerms( -# nft_contract=nft_collection, -# token_id=token_id, -# pil_type=pil_type, -# metadata=metadata, -# deadline=deadline, -# ) - -# assert 'txHash' in response -# assert isinstance(response['txHash'], str) -# assert response['txHash'].startswith("0x") - -# assert 'ipId' in response -# assert isinstance(response['ipId'], str) -# assert response['ipId'].startswith("0x") - -# assert 'licenseTermsId' in response -# assert isinstance(response['licenseTermsId'], int) - -# def test_register_ip_derivative(story_client, nft_collection): -# child_token_id = get_token_id(nft_collection, story_client.web3, story_client.account) - -# pil_type = 'non_commercial_remix' -# mint_metadata = { -# 'metadataHash': web3.to_hex(web3.keccak(text="test-metadata-hash")), -# 'nftMetadataHash': web3.to_hex(web3.keccak(text="test-nft-metadata-hash")) -# } - -# mint_response = story_client.IPAsset.mintAndRegisterIpAssetWithPilTerms( -# nft_contract=nft_collection, -# pil_type=pil_type, -# metadata=mint_metadata, -# minting_fee=100, -# commercial_rev_share=10, -# currency=MockERC20 -# ) - -# parent_ip_id = mint_response['ipId'] -# license_terms_id = mint_response['licenseTermsId'] - -# metadata = { -# 'metadataURI': "test-uri", -# 'metadataHash': web3.to_hex(web3.keccak(text="test-metadata-hash")), -# } -# derivData = { -# 'parentIpIds': [parent_ip_id], -# 'licenseTermsIds': [license_terms_id] -# } - -# response = story_client.IPAsset.registerDerivativeIp( -# nft_contract=nft_collection, -# token_id=child_token_id, -# metadata=metadata, -# deadline=1000, -# deriv_data=derivData -# ) - -# assert 'txHash' in response -# assert isinstance(response['txHash'], str) -# assert response['txHash'].startswith("0x") - -# assert 'ipId' in response -# assert isinstance(response['ipId'], str) -# assert response['ipId'].startswith("0x") + assert 'tokenId' in response + assert isinstance(response['tokenId'], int) \ No newline at end of file diff --git a/tests/integration/test_integration_license.py b/tests/integration/test_integration_license.py index dd17cd5..30b138c 100644 --- a/tests/integration/test_integration_license.py +++ b/tests/integration/test_integration_license.py @@ -202,3 +202,127 @@ def test_setLicensingConfig(story_client, ip_id): assert len(response['txHash']) > 0, "'txHash' is empty" assert 'success' in response, "Response does not contain 'success'" assert response['success'] is True, "'success' is not True" + +def test_register_pil_terms_with_no_minting_fee(story_client): + """Test registering PIL terms with no minting fee.""" + response = story_client.License.registerPILTerms( + transferable=False, + royalty_policy=story_client.web3.to_checksum_address("0x0000000000000000000000000000000000000000"), + default_minting_fee=0, # Minimal minting fee + expiration=0, + commercial_use=False, + commercial_attribution=False, + commercializer_checker=story_client.web3.to_checksum_address("0x0000000000000000000000000000000000000000"), + commercializer_checker_data="0x", + commercial_rev_share=0, + commercial_rev_ceiling=0, + derivatives_allowed=False, + derivatives_attribution=False, + derivatives_approval=False, + derivatives_reciprocal=False, + derivative_rev_ceiling=0, + currency=MockERC20, + uri="" + ) + + assert response is not None + assert 'licenseTermsId' in response + assert response['licenseTermsId'] is not None + assert isinstance(response['licenseTermsId'], int) + +def test_register_commercial_use_pil_without_royalty_policy(story_client): + """Test registering commercial use PIL without specifying royalty policy.""" + response = story_client.License.registerCommercialUsePIL( + default_minting_fee=1, + currency=MockERC20 + ) + + assert response is not None + assert 'licenseTermsId' in response + assert response['licenseTermsId'] is not None + assert isinstance(response['licenseTermsId'], int) + +@pytest.fixture(scope="module") +def setup_license_terms(story_client, ip_id): + """Fixture to set up license terms for testing.""" + response = story_client.License.registerCommercialRemixPIL( + default_minting_fee=1, + currency=MockERC20, + commercial_rev_share=100, + royalty_policy=royalty_policy + ) + license_id = response['licenseTermsId'] + + # Attach the license terms + story_client.License.attachLicenseTerms( + ip_id=ip_id, + license_template=license_template, + license_terms_id=license_id + ) + + return license_id + +def test_multi_token_minting(story_client, ip_id, setup_license_terms): + """Test minting multiple license tokens at once.""" + response = story_client.License.mintLicenseTokens( + licensor_ip_id=ip_id, + license_template=license_template, + license_terms_id=setup_license_terms, + amount=3, # Mint multiple tokens + receiver=account.address + ) + + assert response is not None + assert 'txHash' in response + assert response['txHash'] is not None + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert 'licenseTokenIds' in response + assert isinstance(response['licenseTokenIds'], list) + assert len(response['licenseTokenIds']) > 0 + +def test_set_licensing_config_with_hooks(story_client, ip_id): + """Test setting licensing configuration with hooks enabled.""" + licensing_config = { + 'mintingFee': 100, + 'isSet': True, + 'licensingHook': "0x0000000000000000000000000000000000000000", + 'hookData': "0x1234567890", # Different hook data + 'commercialRevShare': 50, # 50% revenue share + 'disabled': False, + 'expectMinimumGroupRewardShare': 10, # 10% minimum group reward + 'expectGroupRewardPool': "0x0000000000000000000000000000000000000000" + } + + response = story_client.License.setLicensingConfig( + ip_id=ip_id, + license_terms_id=0, + licensing_config=licensing_config, + license_template=license_template + ) + + assert response is not None + assert 'txHash' in response + assert response['txHash'] is not None + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert 'success' in response + assert response['success'] is True + +def test_predict_minting_fee_with_multiple_tokens(story_client, ip_id, setup_license_terms): + """Test predicting minting fee for multiple tokens.""" + response = story_client.License.predictMintingLicenseFee( + licensor_ip_id=ip_id, + license_terms_id=setup_license_terms, + amount=5 # Predict for 5 tokens + ) + + assert response is not None + assert 'currency' in response + assert response['currency'] is not None + assert isinstance(response['currency'], str) + assert len(response['currency']) > 0 + assert 'amount' in response + assert response['amount'] is not None + assert isinstance(response['amount'], int) + assert response['amount'] > 0 # Amount should be positive for multiple tokens diff --git a/tests/integration/test_integration_nft_client.py b/tests/integration/test_integration_nft_client.py index 5a2ef10..ab76b09 100644 --- a/tests/integration/test_integration_nft_client.py +++ b/tests/integration/test_integration_nft_client.py @@ -12,7 +12,7 @@ if src_path not in sys.path: sys.path.append(src_path) -from utils import get_story_client_in_odyssey +from utils import get_story_client_in_devnet load_dotenv() private_key = os.getenv('WALLET_PRIVATE_KEY') @@ -28,7 +28,7 @@ @pytest.fixture def story_client(): - return get_story_client_in_odyssey(web3, account) + return get_story_client_in_devnet(web3, account) def test_create_nft_collection(story_client): response = story_client.NFTClient.createNFTCollection( @@ -47,3 +47,115 @@ def test_create_nft_collection(story_client): assert isinstance(response['nftContract'], str) assert len(response['nftContract']) > 0 assert response['nftContract'].startswith('0x') + +def test_create_nft_collection_with_base_uri(story_client): + """Test creating NFT collection with base URI.""" + response = story_client.NFTClient.createNFTCollection( + name="test-collection-base-uri", + symbol="TESTURI", + max_supply=100, + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + base_uri="https://api.example.com/tokens/", + mint_fee_recipient=account.address + ) + + assert response is not None + assert 'txHash' in response + assert 'nftContract' in response + assert isinstance(response['nftContract'], str) + assert response['nftContract'].startswith('0x') + +def test_create_nft_collection_with_mint_fee(story_client): + """Test creating NFT collection with minting fee.""" + response = story_client.NFTClient.createNFTCollection( + name="test-collection-fee", + symbol="TESTFEE", + max_supply=100, + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + mint_fee=1000000, # Set a minting fee + mint_fee_token="0x0000000000000000000000000000000000000000", # Use zero address for native token + mint_fee_recipient=account.address + ) + + assert response is not None + assert 'txHash' in response + assert 'nftContract' in response + assert isinstance(response['nftContract'], str) + assert response['nftContract'].startswith('0x') + +def test_create_private_mint_collection(story_client): + """Test creating NFT collection with private minting.""" + response = story_client.NFTClient.createNFTCollection( + name="test-collection-private", + symbol="TESTPRIV", + max_supply=100, + is_public_minting=False, # Only allowed minters can mint + mint_open=True, + contract_uri="test-uri", + mint_fee_recipient=account.address + ) + + assert response is not None + assert 'txHash' in response + assert 'nftContract' in response + assert isinstance(response['nftContract'], str) + assert response['nftContract'].startswith('0x') + +def test_create_closed_mint_collection(story_client): + """Test creating NFT collection with minting closed.""" + response = story_client.NFTClient.createNFTCollection( + name="test-collection-closed", + symbol="TESTCLOSED", + max_supply=100, + is_public_minting=True, + mint_open=False, # Minting starts closed + contract_uri="test-uri", + mint_fee_recipient=account.address + ) + + assert response is not None + assert 'txHash' in response + assert 'nftContract' in response + assert isinstance(response['nftContract'], str) + assert response['nftContract'].startswith('0x') + +def test_create_collection_with_custom_owner(story_client): + """Test creating NFT collection with a custom owner.""" + custom_owner = "0x1234567890123456789012345678901234567890" + response = story_client.NFTClient.createNFTCollection( + name="test-collection-owner", + symbol="TESTOWNER", + max_supply=100, + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + owner=custom_owner, + mint_fee_recipient=account.address + ) + + assert response is not None + assert 'txHash' in response + assert 'nftContract' in response + assert isinstance(response['nftContract'], str) + assert response['nftContract'].startswith('0x') + +def test_create_collection_invalid_mint_fee(story_client): + """Test creating NFT collection with invalid mint fee configuration.""" + with pytest.raises(ValueError) as exc_info: + story_client.NFTClient.createNFTCollection( + name="test-collection-invalid", + symbol="TESTINV", + max_supply=100, + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + mint_fee=-1, # Invalid negative fee + mint_fee_token="0x0000000000000000000000000000000000000000", + mint_fee_recipient=account.address + ) + + assert "Invalid mint fee" in str(exc_info.value) diff --git a/tests/integration/test_integration_permission.py b/tests/integration/test_integration_permission.py index 7ebb9ac..3fbf72b 100644 --- a/tests/integration/test_integration_permission.py +++ b/tests/integration/test_integration_permission.py @@ -11,7 +11,7 @@ if src_path not in sys.path: sys.path.append(src_path) -from utils import get_token_id, get_story_client_in_sepolia, MockERC721, check_event_in_tx +from utils import get_token_id, get_story_client_in_devnet, MockERC721, check_event_in_tx load_dotenv() private_key = os.getenv('WALLET_PRIVATE_KEY') @@ -27,7 +27,7 @@ @pytest.fixture def story_client(): - return get_story_client_in_sepolia(web3, account) + return get_story_client_in_devnet(web3, account) def test_setPermission(story_client): token_id = get_token_id(MockERC721, story_client.web3, story_client.account) @@ -52,3 +52,120 @@ def test_setPermission(story_client): assert len(response['txHash']) > 0, "'txHash' is empty." assert check_event_in_tx(web3, response['txHash'], "PermissionSet(address,address,address,address,bytes4,uint8)") is True + +@pytest.fixture +def registered_ip(story_client): + """Fixture to create an IP for testing permissions.""" + token_id = get_token_id(MockERC721, story_client.web3, story_client.account) + response = story_client.IPAsset.register( + nft_contract=MockERC721, + token_id=token_id + ) + return response['ipId'] + +def test_set_permission_with_specific_function(story_client, registered_ip): + """Test setting permission for a specific function.""" + module_address = "0x2ac240293f12032E103458451dE8A8096c5A72E8" + function_selector = Web3.keccak(text="transfer(address,uint256)")[:4].hex() + + response = story_client.Permission.setPermission( + ip_asset=registered_ip, + signer=account.address, + to=module_address, + permission=1, # ALLOW + func=function_selector + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert check_event_in_tx(web3, response['txHash'], "PermissionSet(address,address,address,address,bytes4,uint8)") + +def test_set_all_permissions(story_client, registered_ip): + """Test setting all permissions for a signer.""" + response = story_client.Permission.setAllPermissions( + ip_asset=registered_ip, + signer=account.address, + permission=1 # ALLOW + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert check_event_in_tx(web3, response['txHash'], "PermissionSet(address,address,address,address,bytes4,uint8)") + +def test_set_batch_permissions(story_client, registered_ip): + """Test setting multiple permissions in a single transaction.""" + module_address = "0x2ac240293f12032E103458451dE8A8096c5A72E8" + permissions = [ + { + 'ip_asset': registered_ip, + 'signer': account.address, + 'to': module_address, + 'permission': 1, + 'func': "0x00000000" + }, + { + 'ip_asset': registered_ip, + 'signer': account.address, + 'to': module_address, + 'permission': 2, # DENY + 'func': Web3.keccak(text="transfer(address,uint256)")[:4].hex() + } + ] + + response = story_client.Permission.setBatchPermissions(permissions) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert check_event_in_tx(web3, response['txHash'], "PermissionSet(address,address,address,address,bytes4,uint8)") + +def test_set_permission_invalid_ip(story_client): + """Test setting permission for an unregistered IP.""" + unregistered_ip = "0x1234567890123456789012345678901234567890" + + with pytest.raises(ValueError) as exc_info: + story_client.Permission.setPermission( + ip_asset=unregistered_ip, + signer=account.address, + to="0x2ac240293f12032E103458451dE8A8096c5A72E8", + permission=1 + ) + + assert "is not registered" in str(exc_info.value) + +def test_set_permission_invalid_signer(story_client, registered_ip): + """Test setting permission with invalid signer address.""" + with pytest.raises(ValueError) as exc_info: + story_client.Permission.setPermission( + ip_asset=registered_ip, + signer="0xinvalid", + to="0x2ac240293f12032E103458451dE8A8096c5A72E8", + permission=1 + ) + + assert "is not a valid address" in str(exc_info.value) + +def test_create_permission_signature(story_client, registered_ip): + """Test creating and executing a permission signature.""" + module_address = "0x2ac240293f12032E103458451dE8A8096c5A72E8" + deadline = web3.eth.get_block('latest')['timestamp'] + 1000 + + response = story_client.Permission.createSetPermissionSignature( + ip_asset=registered_ip, + signer=account.address, + to=module_address, + permission=1, + func="0x00000000", + deadline=deadline + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert check_event_in_tx(web3, response['txHash'], "PermissionSet(address,address,address,address,bytes4,uint8)") \ No newline at end of file diff --git a/tests/integration/test_integration_royalty.py b/tests/integration/test_integration_royalty.py index da2a466..0b46e49 100644 --- a/tests/integration/test_integration_royalty.py +++ b/tests/integration/test_integration_royalty.py @@ -12,7 +12,7 @@ if src_path not in sys.path: sys.path.append(src_path) -from utils import get_story_client_in_sepolia, mint_tokens, approve, MockERC721, get_token_id, MockERC20 +from utils import get_story_client_in_devnet, mint_tokens, approve, MockERC721, get_token_id, MockERC20 load_dotenv() private_key = os.getenv('WALLET_PRIVATE_KEY') @@ -28,7 +28,7 @@ @pytest.fixture(scope="module") def story_client(): - return get_story_client_in_sepolia(web3, account) + return get_story_client_in_devnet(web3, account) @pytest.fixture(scope="module") def parent_ip_id(story_client): @@ -172,4 +172,129 @@ def test_claimRevenue(story_client, child_ip_id, snapshot_id): assert 'claimableToken' in response assert response['claimableToken'] is not None assert isinstance(response['claimableToken'], int) - assert response['claimableToken'] >= 0 \ No newline at end of file + assert response['claimableToken'] >= 0 + +def test_snapshot_and_claim_by_token_batch(story_client, child_ip_id): + """Test taking a snapshot and claiming revenue by token batch.""" + # First approve tokens for royalty payments + token_amount = 100000 * 10 ** 6 + mint_tokens( + erc20_contract_address=MockERC20, + web3=web3, + account=account, + to_address=account.address, + amount=token_amount + ) + approve( + erc20_contract_address=MockERC20, + web3=web3, + account=account, + spender_address="0xaabaf349c7a2a84564f9cc4ac130b3f19a718e86", + amount=token_amount + ) + + currency_tokens = [MockERC20] + + response = story_client.Royalty.snapshotAndClaimByTokenBatch( + royalty_vault_ip_id=child_ip_id, + currency_tokens=currency_tokens + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert 'snapshotId' in response + assert isinstance(response['snapshotId'], int) + assert response['snapshotId'] >= 0 + assert 'amountsClaimed' in response + assert isinstance(response['amountsClaimed'], int) + +def test_snapshot_and_claim_by_snapshot_batch(story_client, child_ip_id, snapshot_id): + """Test taking a snapshot and claiming revenue by snapshot batch.""" + currency_tokens = [MockERC20] + unclaimed_snapshot_ids = [snapshot_id] + + response = story_client.Royalty.snapshotAndClaimBySnapshotBatch( + royalty_vault_ip_id=child_ip_id, + currency_tokens=currency_tokens, + unclaimed_snapshot_ids=unclaimed_snapshot_ids + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert 'snapshotId' in response + assert isinstance(response['snapshotId'], int) + assert response['snapshotId'] >= 0 + assert 'amountsClaimed' in response + assert isinstance(response['amountsClaimed'], int) + +def test_transfer_to_vault_and_snapshot_and_claim_by_token_batch(story_client, parent_ip_id, child_ip_id): + """Test transferring to vault, taking snapshot, and claiming by token batch.""" + royalty_claim_details = [{ + 'child_ip_id': child_ip_id, + 'royalty_policy': "0xAAbaf349C7a2A84564F9CC4Ac130B3f19A718E86", + 'currency_token': MockERC20, + 'amount': 100 + }] + + response = story_client.Royalty.transferToVaultAndSnapshotAndClaimByTokenBatch( + ancestor_ip_id=parent_ip_id, + royalty_claim_details=royalty_claim_details + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert 'snapshotId' in response + assert isinstance(response['snapshotId'], int) + assert response['snapshotId'] >= 0 + assert 'amountsClaimed' in response + assert isinstance(response['amountsClaimed'], int) + +def test_transfer_to_vault_and_snapshot_and_claim_by_snapshot_batch(story_client, parent_ip_id, child_ip_id, snapshot_id): + """Test transferring to vault, taking snapshot, and claiming by snapshot batch.""" + royalty_claim_details = [{ + 'child_ip_id': child_ip_id, + 'royalty_policy': "0xAAbaf349C7a2A84564F9CC4Ac130B3f19A718E86", + 'currency_token': MockERC20, + 'amount': 100 + }] + unclaimed_snapshot_ids = [snapshot_id] + + response = story_client.Royalty.transferToVaultAndSnapshotAndClaimBySnapshotBatch( + ancestor_ip_id=parent_ip_id, + royalty_claim_details=royalty_claim_details, + unclaimed_snapshot_ids=unclaimed_snapshot_ids + ) + + assert response is not None + assert 'txHash' in response + assert isinstance(response['txHash'], str) + assert len(response['txHash']) > 0 + assert 'snapshotId' in response + assert isinstance(response['snapshotId'], int) + assert response['snapshotId'] >= 0 + assert 'amountsClaimed' in response + assert isinstance(response['amountsClaimed'], int) + +def test_royalty_vault_address(story_client, child_ip_id): + """Test getting royalty vault address for an IP.""" + vault_address = story_client.Royalty.getRoyaltyVaultAddress(child_ip_id) + + assert vault_address is not None + assert isinstance(vault_address, str) + assert vault_address.startswith('0x') + assert len(vault_address) == 42 # Valid Ethereum address length + +def test_get_royalty_vault_address_unregistered_ip(story_client): + """Test getting royalty vault address for unregistered IP.""" + unregistered_ip = "0x1234567890123456789012345678901234567890" + + with pytest.raises(ValueError) as exc_info: + story_client.Royalty.getRoyaltyVaultAddress(unregistered_ip) + + assert "is not registered" in str(exc_info.value) \ No newline at end of file