From 1dece54e5713ce356735b5b03da761bddae44c4e Mon Sep 17 00:00:00 2001 From: chao-peng-story Date: Mon, 2 Mar 2026 10:57:30 +0800 Subject: [PATCH] feat(Royalty): add batch_claim_all_revenue functionality Implemented batch_claim_all_revenue method to claim revenue from multiple ancestor IPs in a single operation, with support for both sequential and multicall modes. Changes: - Added multicall and build_multicall_transaction methods to RoyaltyWorkflows_client - Implemented batch_claim_all_revenue in Royalty resource with: - Support for single or multiple ancestor IPs - Optional multicall mode for batching multiple claims - Token aggregation by claimer and token address - Auto-transfer and auto-unwrap functionality - Added _encode_transaction_data() for multicall encoding to avoid premature gas estimation - Added 4 unit tests covering various batch claim scenarios - Added 2 integration tests for single and multiple ancestor scenarios --- .../RoyaltyWorkflows_client.py | 6 + .../resources/Royalty.py | 161 ++++++++++ .../utils/transaction_utils.py | 4 + tests/integration/test_integration_royalty.py | 277 ++++++++++++++++++ .../resources/test_batch_claim_all_revenue.py | 199 +++++++++++++ 5 files changed, 647 insertions(+) create mode 100644 tests/unit/resources/test_batch_claim_all_revenue.py diff --git a/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py b/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py index 1a8270b5..d6c5ae89 100644 --- a/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py +++ b/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py @@ -55,3 +55,9 @@ def build_claimAllRevenue_transaction( return self.contract.functions.claimAllRevenue( ancestorIpId, claimer, childIpIds, royaltyPolicies, currencyTokens ).build_transaction(tx_params) + + def multicall(self, data): + return self.contract.functions.multicall(data).transact() + + def build_multicall_transaction(self, data, tx_params): + return self.contract.functions.multicall(data).build_transaction(tx_params) diff --git a/src/story_protocol_python_sdk/resources/Royalty.py b/src/story_protocol_python_sdk/resources/Royalty.py index 777febbf..735e775c 100644 --- a/src/story_protocol_python_sdk/resources/Royalty.py +++ b/src/story_protocol_python_sdk/resources/Royalty.py @@ -12,6 +12,7 @@ IpRoyaltyVaultImplClient, ) from story_protocol_python_sdk.abi.MockERC20.MockERC20_client import MockERC20Client +from story_protocol_python_sdk.abi.Multicall3.Multicall3_client import Multicall3Client from story_protocol_python_sdk.abi.RoyaltyModule.RoyaltyModule_client import ( RoyaltyModuleClient, ) @@ -56,6 +57,7 @@ def __init__(self, web3: Web3, account, chain_id: int): self.mock_erc20_client = MockERC20Client(web3) self.royalty_policy_lrp_client = RoyaltyPolicyLRPClient(web3) self.wrapped_ip_client = WrappedIPClient(web3) + self.multicall3_client = Multicall3Client(web3) def get_royalty_vault_address(self, ip_id: str) -> str: """ @@ -222,6 +224,165 @@ def claim_all_revenue( except Exception as e: raise ValueError(f"Failed to claim all revenue: {str(e)}") + def batch_claim_all_revenue( + self, + ancestor_ips: list[dict], + claim_options: dict | None = None, + options: dict | None = None, + tx_options: dict | None = None, + ) -> dict: + """ + Batch claims all revenue from the child IPs of multiple ancestor IPs. + If multicall is disabled, it will call claim_all_revenue for each ancestor IP. + Then transfer all claimed tokens to the wallet if the wallet owns the IP or is the claimer. + If claimed token is WIP, it will also be converted back to native tokens. + + Even if there are no child IPs, you must still populate `currency_tokens` in each ancestor IP + with the token addresses you wish to claim. This is required for the claim operation to know which + token balances to process. + + :param ancestor_ips list[dict]: List of ancestor IP configurations, each containing: + :param ip_id str: The IP ID of the ancestor. + :param claimer str: The address of the claimer. + :param child_ip_ids list: List of child IP IDs. + :param royalty_policies list: List of royalty policy addresses. + :param currency_tokens list: List of currency token addresses. + :param claim_options dict: [Optional] Options for auto_transfer_all_claimed_tokens_from_ip and auto_unwrap_ip_tokens. Default values are True. + :param options dict: [Optional] Options for use_multicall_when_possible. Default is True. + :param tx_options dict: [Optional] Transaction options. + :return dict: Dictionary with transaction hashes, receipts, and claimed tokens. + :return tx_hashes list[str]: List of transaction hashes. + :return receipts list[dict]: List of transaction receipts. + :return claimed_tokens list[dict]: Aggregated list of claimed tokens. + """ + try: + tx_hashes = [] + receipts = [] + claimed_tokens = [] + + use_multicall = options.get("use_multicall_when_possible", True) if options else True + + # If only 1 ancestor IP or multicall is disabled, call claim_all_revenue for each + if len(ancestor_ips) == 1 or not use_multicall: + for ancestor_ip in ancestor_ips: + result = self.claim_all_revenue( + ancestor_ip_id=ancestor_ip["ip_id"], + claimer=ancestor_ip["claimer"], + child_ip_ids=ancestor_ip["child_ip_ids"], + royalty_policies=ancestor_ip["royalty_policies"], + currency_tokens=ancestor_ip["currency_tokens"], + claim_options={ + "auto_transfer_all_claimed_tokens_from_ip": False, + "auto_unwrap_ip_tokens": False, + }, + tx_options=tx_options, + ) + tx_hashes.extend(result["tx_hashes"]) + receipts.append(result["receipt"]) + if result.get("claimed_tokens"): + claimed_tokens.extend(result["claimed_tokens"]) + else: + # Batch claimAllRevenue calls into a single multicall + encoded_txs = [] + for ancestor_ip in ancestor_ips: + encoded_data = self.royalty_workflows_client.contract.functions.claimAllRevenue( + validate_address(ancestor_ip["ip_id"]), + validate_address(ancestor_ip["claimer"]), + validate_addresses(ancestor_ip["child_ip_ids"]), + validate_addresses(ancestor_ip["royalty_policies"]), + validate_addresses(ancestor_ip["currency_tokens"]), + )._encode_transaction_data() + encoded_txs.append(encoded_data) + + response = build_and_send_transaction( + self.web3, + self.account, + self.royalty_workflows_client.build_multicall_transaction, + encoded_txs, + tx_options=tx_options, + ) + tx_hashes.append(response["tx_hash"]) + receipts.append(response["tx_receipt"]) + + # Parse claimed tokens from the receipt + claimed_token_logs = self._parse_tx_revenue_token_claimed_event( + response["tx_receipt"] + ) + claimed_tokens.extend(claimed_token_logs) + + # Aggregate claimed tokens by claimer and token address + aggregated_claimed_tokens = {} + for token in claimed_tokens: + key = f"{token['claimer']}_{token['token']}" + if key not in aggregated_claimed_tokens: + aggregated_claimed_tokens[key] = dict(token) + else: + aggregated_claimed_tokens[key]["amount"] += token["amount"] + + aggregated_claimed_tokens = list(aggregated_claimed_tokens.values()) + + # Get unique claimers + claimers = list(set(ancestor_ip["claimer"] for ancestor_ip in ancestor_ips)) + + auto_transfer = ( + claim_options.get("auto_transfer_all_claimed_tokens_from_ip", True) + if claim_options + else True + ) + auto_unwrap = ( + claim_options.get("auto_unwrap_ip_tokens", True) + if claim_options + else True + ) + + wip_claimable_amounts = 0 + + for claimer in claimers: + owns_claimer, is_claimer_ip, ip_account = self._get_claimer_info(claimer) + + # If ownsClaimer is false, skip + if not owns_claimer: + continue + + filter_claimed_tokens = [ + token for token in aggregated_claimed_tokens if token["claimer"] == claimer + ] + + # Transfer claimed tokens from IP to wallet if wallet owns IP + if auto_transfer and is_claimer_ip and owns_claimer: + hashes = self._transfer_claimed_tokens_from_ip_to_wallet( + ip_account, filter_claimed_tokens + ) + tx_hashes.extend(hashes) + + # Sum up the amount of WIP tokens claimed + for token in filter_claimed_tokens: + if token["token"] == WIP_TOKEN_ADDRESS: + wip_claimable_amounts += token["amount"] + + # Unwrap WIP tokens if needed + if wip_claimable_amounts > 0 and auto_unwrap: + hashes = self._unwrap_claimed_tokens_from_ip_to_wallet( + [ + { + "token": WIP_TOKEN_ADDRESS, + "amount": wip_claimable_amounts, + "claimer": self.account.address, + } + ] + ) + tx_hashes.extend(hashes) + + return { + "receipts": receipts, + "claimed_tokens": aggregated_claimed_tokens, + "tx_hashes": tx_hashes, + } + + except Exception as e: + error_msg = str(e).replace("Failed to claim all revenue: ", "").strip() + raise ValueError(f"Failed to batch claim all revenue: {error_msg}") + def transfer_to_vault( self, ip_id: str, diff --git a/src/story_protocol_python_sdk/utils/transaction_utils.py b/src/story_protocol_python_sdk/utils/transaction_utils.py index dfbc74ac..f44ae05e 100644 --- a/src/story_protocol_python_sdk/utils/transaction_utils.py +++ b/src/story_protocol_python_sdk/utils/transaction_utils.py @@ -55,6 +55,10 @@ def _get_transaction_options( if "maxFeePerGas" in tx_options: opts["maxFeePerGas"] = tx_options["maxFeePerGas"] + # Gas limit: use explicit gas if provided to avoid estimation + if "gas" in tx_options: + opts["gas"] = tx_options["gas"] + return opts diff --git a/tests/integration/test_integration_royalty.py b/tests/integration/test_integration_royalty.py index 6c489115..dafbc102 100644 --- a/tests/integration/test_integration_royalty.py +++ b/tests/integration/test_integration_royalty.py @@ -151,6 +151,283 @@ def test_pay_royalty_invalid_amount( amount=-1, ) + def test_batch_claim_all_revenue_single_ancestor(self, story_client: StoryClient): + """Test batch claiming revenue using the same pattern as test_claim_all_revenue + + This test verifies that batch_claim_all_revenue works correctly by: + 1. Creating a derivative chain A->B->C + 2. Using batch_claim_all_revenue to claim revenue for A + 3. Verifying the claimed amount matches expectations + """ + # Create NFT collection + collection_response = story_client.NFTClient.create_nft_collection( + name="batch-claim-test", + symbol="BCT", + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + mint_fee_recipient=ZERO_ADDRESS, + ) + spg_nft_contract = collection_response["nft_contract"] + + def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): + """Helper to create derivative with WIP tokens""" + minting_fee = story_client.License.predict_minting_license_fee( + licensor_ip_id=parent_ip_id, + license_terms_id=license_terms_id, + amount=1, + ) + amount = minting_fee["amount"] + + story_client.WIP.deposit(amount=amount) + story_client.WIP.approve(spender=spg_nft_contract, amount=amount) + derivative_workflows_address = DerivativeWorkflowsClient( + story_client.web3 + ).contract.address + story_client.WIP.approve( + spender=derivative_workflows_address, amount=amount + ) + + response = story_client.IPAsset.mint_and_register_ip_and_make_derivative( + spg_nft_contract=spg_nft_contract, + deriv_data=DerivativeDataInput( + parent_ip_ids=[parent_ip_id], + license_terms_ids=[license_terms_id], + ), + ) + return response["ip_id"] + + # Define license terms: 100 WIP minting fee + 10% royalty share + 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, + }, + } + ] + + # Register IP A with PIL terms + ip_a_response = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( + spg_nft_contract=spg_nft_contract, + terms=license_terms_template, + ) + ip_a = ip_a_response["ip_id"] + license_terms_id = ip_a_response["license_terms_ids"][0] + + # Build derivative chain: A -> B -> C -> D (same as test_claim_all_revenue) + ip_b = wrapper_derivative_with_wip(ip_a, license_terms_id) # B pays 100 WIP + ip_c = wrapper_derivative_with_wip(ip_b, license_terms_id) # C pays 100 WIP (10 to A, 90 to B) + wrapper_derivative_with_wip(ip_c, license_terms_id) # D pays 100 WIP (10 to A, 10 to B, 80 to C) + + # Batch claim revenue for IP A (should get 120 WIP: 100 from B + 10 from C + 10 from D) + # Note: Only pass [ip_b, ip_c] as child_ip_ids, not ip_d, matching test_claim_all_revenue + response = story_client.Royalty.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ip_a, + "claimer": ip_a, + "child_ip_ids": [ip_b, ip_c], + "royalty_policies": [ROYALTY_POLICY, ROYALTY_POLICY], + "currency_tokens": [WIP_TOKEN_ADDRESS, WIP_TOKEN_ADDRESS], + }, + ], + claim_options={ + "auto_transfer_all_claimed_tokens_from_ip": False, + "auto_unwrap_ip_tokens": False, + }, + ) + + # Verify response + assert response is not None + assert "tx_hashes" in response + assert len(response["tx_hashes"]) >= 1 + assert "receipts" in response + assert len(response["receipts"]) >= 1 + assert "claimed_tokens" in response + assert len(response["claimed_tokens"]) >= 1 + + # Verify IP A received 120 WIP tokens (100 from B + 10 from C + 10 from D) + assert response["claimed_tokens"][0]["amount"] == 120 + + def test_batch_claim_all_revenue_multiple_ancestors(self, story_client: StoryClient): + """Test batch claiming revenue from multiple ancestor IPs + + This test creates two independent derivative chains and claims revenue for both ancestors: + - Chain 1: A1 -> B1 -> C1 -> D1 (A1 gets 120 WIP) + - Chain 2: A2 -> B2 -> C2 (A2 gets 110 WIP) + """ + # Create NFT collection + collection_response = story_client.NFTClient.create_nft_collection( + name="multi-ancestor-test", + symbol="MAT", + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + mint_fee_recipient=ZERO_ADDRESS, + ) + spg_nft_contract = collection_response["nft_contract"] + + def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): + """Helper to create derivative with WIP tokens""" + minting_fee = story_client.License.predict_minting_license_fee( + licensor_ip_id=parent_ip_id, + license_terms_id=license_terms_id, + amount=1, + ) + amount = minting_fee["amount"] + + story_client.WIP.deposit(amount=amount) + story_client.WIP.approve(spender=spg_nft_contract, amount=amount) + derivative_workflows_address = DerivativeWorkflowsClient( + story_client.web3 + ).contract.address + story_client.WIP.approve( + spender=derivative_workflows_address, amount=amount + ) + + response = story_client.IPAsset.mint_and_register_ip_and_make_derivative( + spg_nft_contract=spg_nft_contract, + deriv_data=DerivativeDataInput( + parent_ip_ids=[parent_ip_id], + license_terms_ids=[license_terms_id], + ), + ) + return response["ip_id"] + + # Define license terms: 100 WIP minting fee + 10% royalty share + 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, + }, + } + ] + + # Register IP A1 with PIL terms + ip_a1_response = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( + spg_nft_contract=spg_nft_contract, + terms=license_terms_template, + ) + ip_a1 = ip_a1_response["ip_id"] + license_terms_id = ip_a1_response["license_terms_ids"][0] + + # Build derivative chain 1: A1 -> B1 -> C1 -> D1 + ip_b1 = wrapper_derivative_with_wip(ip_a1, license_terms_id) + ip_c1 = wrapper_derivative_with_wip(ip_b1, license_terms_id) + wrapper_derivative_with_wip(ip_c1, license_terms_id) # D1 + + # Register IP A2 and attach the same license terms (to avoid duplicate license terms error) + ip_a2_response = story_client.IPAsset.mint_and_register_ip( + spg_nft_contract=spg_nft_contract, + ) + ip_a2 = ip_a2_response["ip_id"] + + # Attach the same license terms to IP A2 + story_client.License.attach_license_terms( + ip_id=ip_a2, + license_template=PIL_LICENSE_TEMPLATE, + license_terms_id=license_terms_id, + ) + + # Build derivative chain 2: A2 -> B2 -> C2 + ip_b2 = wrapper_derivative_with_wip(ip_a2, license_terms_id) + ip_c2 = wrapper_derivative_with_wip(ip_b2, license_terms_id) + + # Batch claim revenue for both ancestors (disable multicall to avoid potential issues) + response = story_client.Royalty.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ip_a1, + "claimer": ip_a1, + "child_ip_ids": [ip_b1, ip_c1], + "royalty_policies": [ROYALTY_POLICY, ROYALTY_POLICY], + "currency_tokens": [WIP_TOKEN_ADDRESS, WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ip_a2, + "claimer": ip_a2, + "child_ip_ids": [ip_b2], + "royalty_policies": [ROYALTY_POLICY], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + claim_options={ + "auto_transfer_all_claimed_tokens_from_ip": False, + "auto_unwrap_ip_tokens": False, + }, + options={ + "use_multicall_when_possible": False, # Disable multicall for stability + }, + ) + + # Verify response + assert response is not None + assert "tx_hashes" in response + assert len(response["tx_hashes"]) >= 2 # Should have 2 separate txs + assert "receipts" in response + assert len(response["receipts"]) >= 2 + assert "claimed_tokens" in response + assert len(response["claimed_tokens"]) == 2 # Two ancestors claimed + + # Verify both ancestors received their expected amounts + # A1 should get 120 WIP (100 from B1 + 10 from C1 + 10 from D1) + # A2 should get 110 WIP (100 from B2 + 10 from C2) + claimed_amounts = {token["claimer"]: token["amount"] for token in response["claimed_tokens"]} + assert claimed_amounts[ip_a1] == 120 + assert claimed_amounts[ip_a2] == 110 + class TestClaimAllRevenue: def test_claim_all_revenue(self, story_client: StoryClient): diff --git a/tests/unit/resources/test_batch_claim_all_revenue.py b/tests/unit/resources/test_batch_claim_all_revenue.py new file mode 100644 index 00000000..698e7778 --- /dev/null +++ b/tests/unit/resources/test_batch_claim_all_revenue.py @@ -0,0 +1,199 @@ +from unittest.mock import patch + +import pytest + +from story_protocol_python_sdk.resources.Royalty import Royalty +from story_protocol_python_sdk.utils.constants import WIP_TOKEN_ADDRESS +from tests.unit.fixtures.data import ACCOUNT_ADDRESS, ADDRESS, TX_HASH + + +@pytest.fixture(scope="class") +def royalty_client(mock_web3, mock_account): + return Royalty(mock_web3, mock_account, 1) + + +class TestBatchClaimAllRevenue: + def test_batch_claim_all_revenue_single_ancestor(self, royalty_client): + """Test batch claim with single ancestor IP (should call claim_all_revenue)""" + with patch.object( + royalty_client, + "claim_all_revenue", + return_value={ + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + } + ], + }, + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + } + ], + ) + assert len(result["tx_hashes"]) >= 1 + assert len(result["receipts"]) == 1 + assert len(result["claimed_tokens"]) == 1 + + def test_batch_claim_all_revenue_multiple_ancestors_with_multicall( + self, royalty_client + ): + """Test batch claim with multiple ancestors using multicall""" + with patch.object( + royalty_client.royalty_workflows_client.contract.functions, + "claimAllRevenue", + return_value=type( + "MockFunction", + (), + { + "build_transaction": lambda self, opts: {"data": "0x1234"}, + "_encode_transaction_data": lambda self: "0x1234", + }, + )(), + ): + with patch.object( + royalty_client, + "_parse_tx_revenue_token_claimed_event", + return_value=[ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + }, + { + "claimer": ACCOUNT_ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 2000, + }, + ], + ): + with patch.object( + royalty_client, "_get_claimer_info", return_value=(False, False, None) + ): + with patch( + "story_protocol_python_sdk.resources.Royalty.build_and_send_transaction", + return_value={"tx_hash": TX_HASH.hex(), "tx_receipt": {"logs": []}}, + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ACCOUNT_ADDRESS, + "claimer": ACCOUNT_ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + ) + assert len(result["tx_hashes"]) == 1 + assert len(result["receipts"]) == 1 + assert len(result["claimed_tokens"]) == 2 + + def test_batch_claim_all_revenue_without_multicall(self, royalty_client): + """Test batch claim with multicall disabled""" + with patch.object( + royalty_client, + "claim_all_revenue", + return_value={ + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + } + ], + }, + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ACCOUNT_ADDRESS, + "claimer": ACCOUNT_ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + options={"use_multicall_when_possible": False}, + ) + assert len(result["tx_hashes"]) >= 2 + assert len(result["receipts"]) == 2 + + def test_batch_claim_all_revenue_aggregates_tokens(self, royalty_client): + """Test that claimed tokens are properly aggregated""" + with patch.object( + royalty_client, + "claim_all_revenue", + side_effect=[ + { + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + } + ], + }, + { + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 500, + } + ], + }, + ], + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ACCOUNT_ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + options={"use_multicall_when_possible": False}, + ) + # Should aggregate tokens for same claimer and token + assert len(result["claimed_tokens"]) == 1 + assert result["claimed_tokens"][0]["amount"] == 1500