diff --git a/.gitignore b/.gitignore index 71ae301..87163b5 100644 --- a/.gitignore +++ b/.gitignore @@ -112,3 +112,4 @@ web3py.md # AI Assistant Configuration CLAUDE.md +.cursor/ diff --git a/src/story_protocol_python_sdk/__init__.py b/src/story_protocol_python_sdk/__init__.py index 6a095b3..7222ed6 100644 --- a/src/story_protocol_python_sdk/__init__.py +++ b/src/story_protocol_python_sdk/__init__.py @@ -8,7 +8,11 @@ from .resources.WIP import WIP from .story_client import StoryClient from .types.common import AccessPermission -from .types.resource.Group import ClaimReward, ClaimRewardsResponse +from .types.resource.Group import ( + ClaimReward, + ClaimRewardsResponse, + CollectRoyaltiesResponse, +) from .types.resource.IPAsset import ( RegisterPILTermsAndAttachResponse, RegistrationResponse, @@ -40,6 +44,7 @@ "RegistrationResponse", "ClaimRewardsResponse", "ClaimReward", + "CollectRoyaltiesResponse", "RegisterPILTermsAndAttachResponse", # Constants "ZERO_ADDRESS", diff --git a/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py b/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py index 1d647ed..e207d8d 100644 --- a/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py +++ b/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py @@ -51,6 +51,14 @@ def build_claimReward_transaction(self, groupId, token, ipIds, tx_params): groupId, token, ipIds ).build_transaction(tx_params) + def collectRoyalties(self, groupId, token): + return self.contract.functions.collectRoyalties(groupId, token).transact() + + def build_collectRoyalties_transaction(self, groupId, token, tx_params): + return self.contract.functions.collectRoyalties( + groupId, token + ).build_transaction(tx_params) + def registerGroup(self, groupPool): return self.contract.functions.registerGroup(groupPool).transact() diff --git a/src/story_protocol_python_sdk/resources/Group.py b/src/story_protocol_python_sdk/resources/Group.py index 1819234..18cb4d5 100644 --- a/src/story_protocol_python_sdk/resources/Group.py +++ b/src/story_protocol_python_sdk/resources/Group.py @@ -29,6 +29,7 @@ from story_protocol_python_sdk.types.resource.Group import ( ClaimReward, ClaimRewardsResponse, + CollectRoyaltiesResponse, ) from story_protocol_python_sdk.utils.constants import ZERO_ADDRESS, ZERO_HASH from story_protocol_python_sdk.utils.license_terms import LicenseTerms @@ -597,6 +598,55 @@ def claim_rewards( except Exception as e: raise ValueError(f"Failed to claim rewards: {str(e)}") + def collect_royalties( + self, + group_ip_id: Address, + currency_token: Address, + tx_options: dict | None = None, + ) -> CollectRoyaltiesResponse: + """ + Collects royalties into the pool, making them claimable by group member IPs. + + :param group_ip_id Address: The ID of the group IP. + :param currency_token Address: The address of the currency (revenue) token to collect. + :param tx_options dict: [Optional] The transaction options. + :return CollectRoyaltiesResponse: A response object with the transaction hash and collected royalties. + """ + try: + if not self.web3.is_address(group_ip_id): + raise ValueError(f"Invalid group IP ID: {group_ip_id}") + if not self.web3.is_address(currency_token): + raise ValueError(f"Invalid currency token: {currency_token}") + + response = build_and_send_transaction( + self.web3, + self.account, + self.grouping_module_client.build_collectRoyalties_transaction, + group_ip_id, + currency_token, + tx_options=tx_options, + ) + + event_signature = self.web3.keccak( + text="CollectedRoyaltiesToGroupPool(address,address,address,uint256)" + ).hex() + + collected_royalties = 0 + for log in response["tx_receipt"]["logs"]: + if log["topics"][0].hex() == event_signature: + event_results = self.grouping_module_client.contract.events.CollectedRoyaltiesToGroupPool.process_log( + log + ) + collected_royalties = event_results["args"]["amount"] + break + + return CollectRoyaltiesResponse( + tx_hash=response["tx_hash"], + collected_royalties=collected_royalties, + ) + except Exception as e: + raise ValueError(f"Failed to collect royalties: {str(e)}") + def _get_license_data(self, license_data: list) -> list: """ Process license data into the format expected by the contracts. diff --git a/src/story_protocol_python_sdk/scripts/config.json b/src/story_protocol_python_sdk/scripts/config.json index b1d9252..bf797ad 100644 --- a/src/story_protocol_python_sdk/scripts/config.json +++ b/src/story_protocol_python_sdk/scripts/config.json @@ -183,7 +183,8 @@ "registerGroup", "addIp", "IPGroupRegistered", - "claimReward" + "claimReward", + "collectRoyalties" ] }, { diff --git a/src/story_protocol_python_sdk/types/resource/Group.py b/src/story_protocol_python_sdk/types/resource/Group.py index e6c225e..4d262d1 100644 --- a/src/story_protocol_python_sdk/types/resource/Group.py +++ b/src/story_protocol_python_sdk/types/resource/Group.py @@ -21,3 +21,12 @@ class ClaimRewardsResponse(TypedDict): tx_hash: HexBytes claimed_rewards: ClaimReward + + +class CollectRoyaltiesResponse(TypedDict): + """ + Response structure for Group.collect_royalties method. + """ + + tx_hash: HexBytes + collected_royalties: int diff --git a/tests/integration/test_integration_group.py b/tests/integration/test_integration_group.py index e932781..2f81928 100644 --- a/tests/integration/test_integration_group.py +++ b/tests/integration/test_integration_group.py @@ -1,8 +1,7 @@ -# tests/integration/test_integration_group.py +import time +from typing import Any -import copy - -import pytest +from ens.ens import Address from story_protocol_python_sdk.story_client import StoryClient @@ -12,337 +11,18 @@ ROYALTY_POLICY_LRP, ZERO_ADDRESS, MockERC20, - account, web3, ) -# class TestGroupBasicOperations: -# def test_register_basic_group(self, story_client): -# response = story_client.Group.register_group( -# group_pool=EVEN_SPLIT_GROUP_POOL -# ) - -# assert 'tx_hash' in response -# assert isinstance(response['tx_hash'], str) -# assert len(response['tx_hash']) > 0 - -# assert 'group_id' in response -# assert isinstance(response['group_id'], str) -# assert response['group_id'].startswith("0x") - -# class TestGroupWithLicenseOperations: -# @pytest.fixture(scope="module") -# def nft_collection(self, story_client): -# tx_data = story_client.NFTClient.create_nft_collection( -# name="test-collection", -# symbol="TEST", -# max_supply=100, -# is_public_minting=True, -# mint_open=True, -# contract_uri="test-uri", -# mint_fee_recipient=account.address, -# ) -# return tx_data['nft_contract'] - -# @pytest.fixture(scope="module") -# def ip_with_license(self, story_client, nft_collection): -# # Create initial IP with license terms -# response = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( -# spg_nft_contract=nft_collection, -# terms=[{ -# 'terms': { -# 'transferable': True, -# 'royalty_policy': ROYALTY_POLICY_LRP, -# 'default_minting_fee': 0, -# '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': 0, -# 'hook_data': ZERO_ADDRESS, -# 'licensing_hook': ZERO_ADDRESS, -# 'commercial_rev_share': 0, -# 'disabled': False, -# 'expect_minimum_group_reward_share': 0, -# 'expect_group_reward_pool': EVEN_SPLIT_GROUP_POOL -# } -# }] -# ) - -# ip_id = response['ip_id'] -# license_terms_id = response['license_terms_ids'][0] - -# licensing_config = { -# 'isSet': True, -# 'mintingFee': 0, -# 'licensingHook': ZERO_ADDRESS, -# 'hookData': ZERO_ADDRESS, -# 'commercialRevShare': 0, -# 'disabled': False, -# 'expectMinimumGroupRewardShare': 0, -# 'expectGroupRewardPool': EVEN_SPLIT_GROUP_POOL -# } - -# # Set licensing config -# story_client.License.set_licensing_config( -# ip_id=ip_id, -# license_terms_id=license_terms_id, -# license_template=PIL_LICENSE_TEMPLATE, -# licensing_config=licensing_config -# ) - -# return { -# 'ip_id': ip_id, -# 'license_terms_id': license_terms_id -# } - -# @pytest.fixture(scope="module") -# def group_with_license(self, story_client, ip_with_license): -# response = story_client.Group.register_group_and_attach_license( -# group_pool=EVEN_SPLIT_GROUP_POOL, -# license_data={ -# 'license_terms_id': ip_with_license['license_terms_id'], -# 'licensing_config': { -# 'is_set': True, -# 'minting_fee': 0, -# '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 -# } -# } -# ) - -# assert 'tx_hash' in response -# assert isinstance(response['tx_hash'], str) - -# assert response is not None -# assert 'group_id' in response -# assert response['group_id'] is not None -# return response['group_id'] - -# def test_register_group_and_attach_license(self, group_with_license): -# assert group_with_license is not None - -# def test_mint_register_ip_attach_license_add_to_group(self, story_client, group_with_license, ip_with_license, nft_collection): -# response = story_client.Group.mint_and_register_ip_and_attach_license_and_add_to_group( -# group_id=group_with_license, -# spg_nft_contract=nft_collection, -# license_data=[{ -# 'license_terms_id': ip_with_license['license_terms_id'], -# 'licensing_config': { -# 'is_set': True, -# 'minting_fee': 0, -# 'hook_data': ZERO_ADDRESS, -# 'licensing_hook': ZERO_ADDRESS, -# 'commercial_rev_share': 0, -# 'disabled': False, -# 'expect_minimum_group_reward_share': 0, -# 'expect_group_reward_pool': EVEN_SPLIT_GROUP_POOL -# } -# }], -# max_allowed_reward_share=5 -# ) - -# assert 'tx_hash' in response -# assert isinstance(response['tx_hash'], str) -# assert len(response['tx_hash']) > 0 - -# assert 'ip_id' in response -# assert isinstance(response['ip_id'], str) -# assert response['ip_id'].startswith("0x") - -# class TestAdvancedGroupOperations: -# @pytest.fixture(scope="module") -# def nft_collection(self, story_client): -# tx_data = story_client.NFTClient.create_nft_collection( -# name="test-collection", -# symbol="TEST", -# max_supply=100, -# is_public_minting=True, -# mint_open=True, -# contract_uri="test-uri", -# mint_fee_recipient=account.address, -# ) -# return tx_data['nft_contract'] - -# @pytest.fixture(scope="module") -# def ip_with_license(self, story_client, nft_collection): -# response = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( -# spg_nft_contract=nft_collection, -# terms=[{ -# 'terms': { -# 'transferable': True, -# 'royalty_policy': ROYALTY_POLICY_LRP, -# 'default_minting_fee': 0, -# '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': 0, -# 'hook_data': ZERO_ADDRESS, -# 'licensing_hook': ZERO_ADDRESS, -# 'commercial_rev_share': 0, -# 'disabled': False, -# 'expect_minimum_group_reward_share': 0, -# 'expect_group_reward_pool': EVEN_SPLIT_GROUP_POOL -# } -# }] -# ) - -# return { -# 'ip_id': response['ip_id'], -# 'license_terms_id': response['license_terms_ids'][0] -# } - -# @pytest.fixture(scope="module") -# def group_id(self, story_client, ip_with_license): -# # Create a group with license attached -# response = story_client.Group.register_group_and_attach_license( -# group_pool=EVEN_SPLIT_GROUP_POOL, -# license_data={ -# 'license_terms_id': ip_with_license['license_terms_id'], -# 'license_template': PIL_LICENSE_TEMPLATE, -# 'licensing_config': { -# 'is_set': True, -# 'minting_fee': 0, -# '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 -# } -# } -# ) -# return response['group_id'] - -# def test_register_ip_and_attach_license_and_add_to_group(self, story_client, group_id, ip_with_license): -# token_id = get_token_id(MockERC721, story_client.web3, story_client.account) - -# response = story_client.Group.register_ip_and_attach_license_and_add_to_group( -# group_id=group_id, -# nft_contract=MockERC721, -# token_id=token_id, -# max_allowed_reward_share=5, -# license_data=[{ -# 'license_terms_id': ip_with_license['license_terms_id'], -# 'licensing_config': { -# 'is_set': True, -# 'minting_fee': 0, -# 'hook_data': ZERO_ADDRESS, -# 'licensing_hook': ZERO_ADDRESS, -# 'commercial_rev_share': 0, -# 'disabled': False, -# 'expect_minimum_group_reward_share': 0, -# 'expect_group_reward_pool': EVEN_SPLIT_GROUP_POOL -# } -# }] -# ) - -# assert 'tx_hash' in response -# assert isinstance(response['tx_hash'], str) -# assert len(response['tx_hash']) > 0 - -# assert 'ip_id' in response -# assert isinstance(response['ip_id'], str) -# assert response['ip_id'].startswith("0x") - -# def test_register_group_and_attach_license_and_add_ips(self, story_client, ip_with_license): -# response = story_client.Group.register_group_and_attach_license_and_add_ips( -# group_pool=EVEN_SPLIT_GROUP_POOL, -# max_allowed_reward_share=5, -# ip_ids=[ip_with_license['ip_id']], -# license_data={ -# 'license_terms_id': ip_with_license['license_terms_id'], -# 'licensing_config': { -# 'is_set': True, -# 'minting_fee': 0, -# '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 -# } -# } -# ) - -# assert 'tx_hash' in response -# assert isinstance(response['tx_hash'], str) -# assert len(response['tx_hash']) > 0 - -# assert 'group_id' in response -# assert isinstance(response['group_id'], str) -# assert response['group_id'].startswith("0x") - -# def test_fail_add_unregistered_ip_to_group(self, story_client, ip_with_license): -# with pytest.raises(ValueError, match="Failed to register group and attach license and add IPs"): -# story_client.Group.register_group_and_attach_license_and_add_ips( -# group_pool=EVEN_SPLIT_GROUP_POOL, -# max_allowed_reward_share=5, -# ip_ids=[ZERO_ADDRESS], # Invalid IP address -# license_data={ -# 'license_terms_id': ip_with_license['license_terms_id'], -# 'licensing_config': { -# 'is_set': True, -# 'minting_fee': 0, -# '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 -# } -# } -# ) - -class TestCollectRoyaltyAndClaimReward: - @pytest.fixture(scope="module") - def nft_collection(self, story_client): - tx_data = story_client.NFTClient.create_nft_collection( - name="test-collection", - symbol="TEST", - max_supply=100, - is_public_minting=True, - mint_open=True, - contract_uri="test-uri", - mint_fee_recipient=account.address, - ) - return tx_data["nft_contract"] +class GroupTestHelper: + """Helper class for Group integration tests.""" - @pytest.fixture(scope="module") - def setup_royalty_collection(self, story_client, nft_collection): - # Create license terms data + @staticmethod + def mint_and_register_ip_asset_with_pil_terms( + story_client: StoryClient, nft_collection: Address + ) -> dict[str, Any]: + """Helper to mint and register an IP asset with PIL terms.""" license_terms_data = [ { "terms": { @@ -371,52 +51,111 @@ def setup_royalty_collection(self, story_client, nft_collection): "licensing_hook": ZERO_ADDRESS, "commercial_rev_share": 10, "disabled": False, - "expect_minimum_group_reward_share": 10, + "expect_minimum_group_reward_share": 0, "expect_group_reward_pool": EVEN_SPLIT_GROUP_POOL, }, } ] - # Create unique metadata for each IP - metadata_1 = { - "ip_metadata_uri": "test-uri-1", - "ip_metadata_hash": web3.to_hex(web3.keccak(text="test-metadata-hash-1")), - "nft_metadata_uri": "test-nft-uri-1", + + # Create unique metadata + metadata = { + "ip_metadata_uri": f"test-uri-{int(time.time())}", + "ip_metadata_hash": web3.to_hex( + web3.keccak(text=f"test-metadata-hash-{int(time.time())}") + ), + "nft_metadata_uri": f"test-nft-uri-{int(time.time())}", "nft_metadata_hash": web3.to_hex( - web3.keccak(text="test-nft-metadata-hash-a") + web3.keccak(text=f"test-nft-metadata-hash-{int(time.time())}") ), } - metadata_2 = { - "ip_metadata_uri": "test-uri-2", - "ip_metadata_hash": web3.to_hex(web3.keccak(text="test-metadata-hash-2")), - "nft_metadata_uri": "test-nft-uri-2", + result = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( + spg_nft_contract=nft_collection, + terms=license_terms_data, + ip_metadata=metadata, + ) + + return { + "ip_id": result["ip_id"], + "license_terms_id": result["license_terms_ids"][0], + } + + @staticmethod + def mint_and_register_ip_and_make_derivative( + story_client: StoryClient, + nft_collection: Address, + group_id: Address, + license_id: int, + ) -> Address: + """Helper to mint and register an IP and make it a derivative of another IP.""" + # Step 1: Mint and register IP + metadata = { + "ip_metadata_uri": f"test-derivative-uri-{int(time.time())}", + "ip_metadata_hash": web3.to_hex( + web3.keccak(text=f"test-derivative-metadata-hash-{int(time.time())}") + ), + "nft_metadata_uri": f"test-derivative-nft-uri-{int(time.time())}", "nft_metadata_hash": web3.to_hex( - web3.keccak(text="test-nft-metadata-hash-2") + web3.keccak( + text=f"test-derivative-nft-metadata-hash-{int(time.time())}" + ) ), } - # Create two IPs - result1 = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( + result = story_client.IPAsset.mint_and_register_ip( spg_nft_contract=nft_collection, - terms=copy.deepcopy(license_terms_data), - ip_metadata=metadata_1, + ip_metadata=metadata, ) - result2 = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( - spg_nft_contract=nft_collection, - terms=copy.deepcopy(license_terms_data), - ip_metadata=metadata_2, + child_ip_id = result["ip_id"] + + # Step 2: Register as derivative + story_client.IPAsset.register_derivative( + child_ip_id=child_ip_id, + parent_ip_ids=[group_id], + license_terms_ids=[license_id], + max_minting_fee=0, + max_rts=10, + max_revenue_share=0, ) - ip_ids = [result1["ip_id"], result2["ip_id"]] - license_terms_id = result1["license_terms_ids"][0] + return child_ip_id + + @staticmethod + def pay_royalty_and_transfer_to_vault( + story_client: StoryClient, + child_ip_id: Address, + group_id: Address, + token: Address, + amount: int, + ) -> None: + """Helper to pay royalty on behalf and transfer to vault.""" + # Pay royalties from group IP id to child IP id + story_client.Royalty.pay_royalty_on_behalf( + receiver_ip_id=child_ip_id, + payer_ip_id=group_id, + token=token, + amount=amount, + ) - # Register group and add IPs - result3 = story_client.Group.register_group_and_attach_license_and_add_ips( + # Transfer to vault + story_client.Royalty.transfer_to_vault( + royalty_policy="LRP", + ip_id=child_ip_id, + ancestor_ip_id=group_id, + token=token, + ) + + @staticmethod + def register_group_and_attach_license( + story_client: StoryClient, license_id: int, ip_ids: list[Address] + ) -> Address: + """Helper to register a group and attach license.""" + result = story_client.Group.register_group_and_attach_license_and_add_ips( group_pool=EVEN_SPLIT_GROUP_POOL, max_allowed_reward_share=100, ip_ids=ip_ids, license_data={ - "license_terms_id": license_terms_id, + "license_terms_id": license_id, "license_template": PIL_LICENSE_TEMPLATE, "licensing_config": { "is_set": True, @@ -431,100 +170,129 @@ def setup_royalty_collection(self, story_client, nft_collection): }, ) - group_ip_id = result3["group_id"] + return result["group_id"] - # Create derivative IPs - Step 1: Mint and register - result4 = story_client.IPAsset.mint_and_register_ip( - spg_nft_contract=nft_collection, - ip_metadata={ - "ip_metadata_uri": "test-derivative-uri-4", - "ip_metadata_hash": web3.to_hex( - web3.keccak(text="test-derivative-metadata-hash-1") - ), - "nft_metadata_uri": "test-derivative-nft-uri-4", - "nft_metadata_hash": web3.to_hex( - web3.keccak(text="test-derivative-nft-metadata-hash-1") - ), - }, + +class TestCollectRoyaltyAndClaimReward: + """Test class for collecting royalties and claiming rewards functionality.""" + + def test_collect_royalties( + self, story_client: StoryClient, nft_collection: Address + ): + """Test collecting royalties into the pool.""" + # Register IP id + result1 = GroupTestHelper.mint_and_register_ip_asset_with_pil_terms( + story_client, nft_collection ) - child_ip_id1 = result4["ip_id"] + ip_id = result1["ip_id"] + license_terms_id = result1["license_terms_id"] - # Step 2: Register as derivative - story_client.IPAsset.register_derivative( - child_ip_id=child_ip_id1, - parent_ip_ids=[group_ip_id], - license_terms_ids=[license_terms_id], - max_minting_fee=0, - max_rts=10, - max_revenue_share=0, + # Register group id + group_ip_id = GroupTestHelper.register_group_and_attach_license( + story_client, license_terms_id, [ip_id] ) - # Create second derivative IP - Step 1: Mint and register - result5 = story_client.IPAsset.mint_and_register_ip( - spg_nft_contract=nft_collection, - ip_metadata={ - "ip_metadata_uri": "test-derivative-uri-5", - "ip_metadata_hash": web3.to_hex( - web3.keccak(text="test-derivative-metadata-hash-2") - ), - "nft_metadata_uri": "test-derivative-nft-uri-5", - "nft_metadata_hash": web3.to_hex( - web3.keccak(text="test-derivative-nft-metadata-hash-2") - ), - }, + # Mint and register child IP id + child_ip_id = GroupTestHelper.mint_and_register_ip_and_make_derivative( + story_client, nft_collection, group_ip_id, license_terms_id ) - child_ip_id2 = result5["ip_id"] - # Step 2: Register as derivative - story_client.IPAsset.register_derivative( - child_ip_id=child_ip_id2, - parent_ip_ids=[group_ip_id], - license_terms_ids=[license_terms_id], - max_minting_fee=0, - max_rts=10, - max_revenue_share=0, + # Pay royalties from group IP id to child IP id and transfer to vault + GroupTestHelper.pay_royalty_and_transfer_to_vault( + story_client, child_ip_id, group_ip_id, MockERC20, 100 ) - # Pay royalties from child IPs to group IP - story_client.Royalty.pay_royalty_on_behalf( - receiver_ip_id=child_ip_id1, - payer_ip_id=group_ip_id, - token=MockERC20, - amount=100, + # Collect royalties + result = story_client.Group.collect_royalties( + group_ip_id=group_ip_id, currency_token=MockERC20 ) - story_client.Royalty.pay_royalty_on_behalf( - receiver_ip_id=child_ip_id2, - payer_ip_id=group_ip_id, - token=MockERC20, - amount=100, + assert "tx_hash" in result + assert isinstance(result["tx_hash"], str) + assert len(result["tx_hash"]) > 0 + assert result["collected_royalties"] == 10 # 10% of 100 = 10 + + def test_claim_reward(self, story_client: StoryClient, nft_collection: Address): + """Test claiming rewards for group members.""" + # Register IP id + result1 = GroupTestHelper.mint_and_register_ip_asset_with_pil_terms( + story_client, nft_collection ) + ip_id = result1["ip_id"] + license_terms_id = result1["license_terms_id"] - # Transfer to vault - story_client.Royalty.transfer_to_vault( - royalty_policy="LRP", - ip_id=child_ip_id1, - ancestor_ip_id=group_ip_id, - token=MockERC20, + # Register group id + group_ip_id = GroupTestHelper.register_group_and_attach_license( + story_client, license_terms_id, [ip_id] ) - story_client.Royalty.transfer_to_vault( - royalty_policy="LRP", - ip_id=child_ip_id2, - ancestor_ip_id=group_ip_id, - token=MockERC20, + # Mint license tokens to the IP id which doesn't have a royalty vault + story_client.License.mint_license_tokens( + licensor_ip_id=ip_id, + license_template=PIL_LICENSE_TEMPLATE, + license_terms_id=license_terms_id, + amount=100, + receiver=ip_id, + max_minting_fee=1, + max_revenue_share=100, ) - return {"group_ip_id": group_ip_id, "ip_ids": ip_ids} + # Claim reward + result = story_client.Group.claim_rewards( + group_ip_id=group_ip_id, + currency_token=MockERC20, + member_ip_ids=[ip_id], + ) + + assert "tx_hash" in result + assert isinstance(result["tx_hash"], str) + assert "claimed_rewards" in result + assert len(result["claimed_rewards"]["ip_ids"]) == 1 + assert len(result["claimed_rewards"]["amounts"]) == 1 + assert result["claimed_rewards"]["token"] == MockERC20 + assert result["claimed_rewards"]["group_id"] == group_ip_id def test_collect_and_distribute_group_royalties( - self, story_client, setup_royalty_collection + self, story_client: StoryClient, nft_collection: Address ): - group_ip_id = setup_royalty_collection["group_ip_id"] - ip_ids = setup_royalty_collection["ip_ids"] + """Test collecting and distributing group royalties in one transaction.""" + ip_ids = [] + # Create two IPs + result1 = GroupTestHelper.mint_and_register_ip_asset_with_pil_terms( + story_client, nft_collection + ) + result2 = GroupTestHelper.mint_and_register_ip_asset_with_pil_terms( + story_client, nft_collection + ) + ip_ids.append(result1["ip_id"]) + ip_ids.append(result2["ip_id"]) + license_terms_id = result1["license_terms_id"] + + # Register group + group_id = GroupTestHelper.register_group_and_attach_license( + story_client, license_terms_id, ip_ids + ) + + # Create derivative IPs + child_ip_id1 = GroupTestHelper.mint_and_register_ip_and_make_derivative( + story_client, nft_collection, group_id, license_terms_id + ) + child_ip_id2 = GroupTestHelper.mint_and_register_ip_and_make_derivative( + story_client, nft_collection, group_id, license_terms_id + ) + + # Pay royalties + GroupTestHelper.pay_royalty_and_transfer_to_vault( + story_client, child_ip_id1, group_id, MockERC20, 100 + ) + GroupTestHelper.pay_royalty_and_transfer_to_vault( + story_client, child_ip_id2, group_id, MockERC20, 100 + ) + + # Collect and distribute royalties response = story_client.Group.collect_and_distribute_group_royalties( - group_ip_id=group_ip_id, currency_tokens=[MockERC20], member_ip_ids=ip_ids + group_ip_id=group_id, currency_tokens=[MockERC20], member_ip_ids=ip_ids ) assert "tx_hash" in response @@ -532,7 +300,6 @@ def test_collect_and_distribute_group_royalties( assert len(response["tx_hash"]) > 0 assert "collected_royalties" in response - assert len(response["collected_royalties"]) > 0 assert response["collected_royalties"][0]["amount"] == 20 @@ -540,25 +307,3 @@ def test_collect_and_distribute_group_royalties( assert len(response["royalties_distributed"]) == 2 assert response["royalties_distributed"][0]["amount"] == 10 assert response["royalties_distributed"][1]["amount"] == 10 - - def test_claim_rewards(self, story_client: StoryClient, setup_royalty_collection): - """Test claiming rewards for group members.""" - group_ip_id = setup_royalty_collection["group_ip_id"] - ip_ids = setup_royalty_collection["ip_ids"] - # Collect and distribute royalties to set up rewards for claiming - story_client.Group.collect_and_distribute_group_royalties( - group_ip_id=group_ip_id, currency_tokens=[MockERC20], member_ip_ids=ip_ids - ) - # Test claiming rewards for specific members - response = story_client.Group.claim_rewards( - group_ip_id=group_ip_id, - currency_token=MockERC20, - member_ip_ids=ip_ids, - ) - assert "tx_hash" in response - assert isinstance(response["tx_hash"], str) - assert "claimed_rewards" in response - assert len(response["claimed_rewards"]["ip_ids"]) == 2 - assert len(response["claimed_rewards"]["amounts"]) == 2 - assert response["claimed_rewards"]["token"] == MockERC20 - assert response["claimed_rewards"]["group_id"] == group_ip_id diff --git a/tests/unit/resources/test_group.py b/tests/unit/resources/test_group.py index 7ef683b..bfbcf45 100644 --- a/tests/unit/resources/test_group.py +++ b/tests/unit/resources/test_group.py @@ -6,6 +6,7 @@ from story_protocol_python_sdk.types.resource.Group import ( ClaimReward, ClaimRewardsResponse, + CollectRoyaltiesResponse, ) from tests.unit.fixtures.data import ADDRESS, CHAIN_ID, IP_ID, TX_HASH @@ -15,6 +16,135 @@ def group(mock_web3, mock_account): return Group(mock_web3, mock_account, CHAIN_ID) +class TestGroupCollectRoyalties: + """Test class for Group.collect_royalties method""" + + def test_collect_royalties_invalid_group_ip_id( + self, group: Group, mock_web3_is_address + ): + """Test collect_royalties with invalid group IP ID.""" + invalid_group_ip_id = "invalid_group_ip_id" + with mock_web3_is_address(False): + with pytest.raises( + ValueError, + match=f"Failed to collect royalties: Invalid group IP ID: {invalid_group_ip_id}", + ): + group.collect_royalties( + group_ip_id=invalid_group_ip_id, + currency_token=ADDRESS, + ) + + def test_collect_royalties_invalid_currency_token(self, group: Group, mock_web3): + """Test collect_royalties with invalid currency token.""" + invalid_currency_token = "invalid_currency_token" + with patch.object(mock_web3, "is_address") as mock_is_address: + # group_ip_id=True, currency_token=False + mock_is_address.side_effect = [True, False] + with pytest.raises( + ValueError, + match=f"Failed to collect royalties: Invalid currency token: {invalid_currency_token}", + ): + group.collect_royalties( + group_ip_id=IP_ID, + currency_token=invalid_currency_token, + ) + + def test_collect_royalties_success( + self, + group: Group, + mock_web3_is_address, + ): + """Test successful collect_royalties operation.""" + collected_amount = 100 + + with mock_web3_is_address(): + with patch( + "story_protocol_python_sdk.resources.Group.build_and_send_transaction", + return_value={ + "tx_hash": TX_HASH, + "tx_receipt": { + "logs": [ + { + "topics": [ + group.web3.keccak( + text="CollectedRoyaltiesToGroupPool(address,address,address,uint256)" + ) + ] + } + ] + }, + }, + ), patch.object( + group.grouping_module_client.contract.events.CollectedRoyaltiesToGroupPool, + "process_log", + return_value={"args": {"amount": collected_amount}}, + ): + result = group.collect_royalties( + group_ip_id=IP_ID, + currency_token=ADDRESS, + ) + + assert "tx_hash" in result + assert result["tx_hash"] == TX_HASH + assert "collected_royalties" in result + assert result["collected_royalties"] == collected_amount + assert result == CollectRoyaltiesResponse( + tx_hash=TX_HASH, + collected_royalties=collected_amount, + ) + + def test_collect_royalties_no_event_found( + self, + group: Group, + mock_web3_is_address, + ): + """Test collect_royalties when no CollectedRoyaltiesToGroupPool event is found.""" + with mock_web3_is_address(): + with patch( + "story_protocol_python_sdk.resources.Group.build_and_send_transaction", + return_value={ + "tx_hash": TX_HASH, + "tx_receipt": { + "logs": [ + {"topics": [group.web3.keccak(text="DifferentEvent()")]} + ] + }, + }, + ), patch.object( + group.grouping_module_client.contract.events.CollectedRoyaltiesToGroupPool, + "process_log", + return_value={"args": {"amount": 0}}, + ): + result = group.collect_royalties( + group_ip_id=IP_ID, + currency_token=ADDRESS, + ) + + # Should return 0 collected royalties when no event is found + assert "tx_hash" in result + assert result["tx_hash"] == TX_HASH + assert "collected_royalties" in result + assert result["collected_royalties"] == 0 + + def test_collect_royalties_transaction_build_failure( + self, group: Group, mock_web3_is_address + ): + """Test collect_royalties when transaction building fails.""" + with mock_web3_is_address(True): + with patch( + "story_protocol_python_sdk.resources.Group.build_and_send_transaction", + side_effect=Exception("Transaction build failed"), + ): + with pytest.raises( + ValueError, + match="Failed to collect royalties: Transaction build failed", + ): + group.collect_royalties( + group_ip_id=IP_ID, + currency_token=ADDRESS, + ) + + class TestGroupClaimRewards: """Test class for Group.claim_rewards method"""