Support AMMClawback amendment

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [[Unreleased]]
 
 ### Added
+- Support `AMMClawback` amendment (XLS-73d).
 - Support for the `simulate` RPC ([XLS-69](https://github.com/XRPLF/XRPL-Standards/tree/master/XLS-0069d-simulate))
 
 ### Fixed

tests/integration/it_utils.py

@@ -321,6 +321,7 @@ def _get_non_decorator_code(function):
 
 def create_amm_pool(
     client: SyncClient = JSON_RPC_CLIENT,
+    enable_amm_clawback: bool = False,
 ) -> Dict[str, Any]:
     issuer_wallet = Wallet.create()
     fund_wallet(issuer_wallet)
@@ -337,6 +338,16 @@ def create_amm_pool(
         issuer_wallet,
     )
 
+    # The below flag is required for AMMClawback tests
+    if enable_amm_clawback:
+        sign_and_reliable_submission(
+            AccountSet(
+                account=issuer_wallet.classic_address,
+                set_flag=AccountSetAsfFlag.ASF_ALLOW_TRUSTLINE_CLAWBACK,
+            ),
+            issuer_wallet,
+        )
+
     sign_and_reliable_submission(
         TrustSet(
             account=lp_wallet.classic_address,
@@ -388,11 +399,13 @@ def create_amm_pool(
         "asset": asset,
         "asset2": asset2,
         "issuer_wallet": issuer_wallet,
+        "lp_wallet": lp_wallet,
     }
 
 
 async def create_amm_pool_async(
     client: AsyncClient = ASYNC_JSON_RPC_CLIENT,
+    enable_amm_clawback: bool = False,
 ) -> Dict[str, Any]:
     issuer_wallet = Wallet.create()
     await fund_wallet_async(issuer_wallet)
@@ -409,6 +422,16 @@ async def create_amm_pool_async(
         issuer_wallet,
     )
 
+    # The below flag is required for AMMClawback tests
+    if enable_amm_clawback:
+        await sign_and_reliable_submission_async(
+            AccountSet(
+                account=issuer_wallet.classic_address,
+                set_flag=AccountSetAsfFlag.ASF_ALLOW_TRUSTLINE_CLAWBACK,
+            ),
+            issuer_wallet,
+        )
+
     await sign_and_reliable_submission_async(
         TrustSet(
             account=lp_wallet.classic_address,
@@ -460,6 +483,7 @@ async def create_amm_pool_async(
         "asset": asset,
         "asset2": asset2,
         "issuer_wallet": issuer_wallet,
+        "lp_wallet": lp_wallet,
     }
 
 

tests/integration/transactions/test_amm_clawback.py

@@ -0,0 +1,61 @@
+from tests.integration.integration_test_case import IntegrationTestCase
+from tests.integration.it_utils import (
+    create_amm_pool_async,
+    sign_and_reliable_submission_async,
+    test_async_and_sync,
+)
+from xrpl.models.amounts.issued_currency_amount import IssuedCurrencyAmount
+from xrpl.models.currencies.issued_currency import IssuedCurrency
+from xrpl.models.currencies.xrp import XRP
+from xrpl.models.transactions import AMMDeposit
+from xrpl.models.transactions.amm_clawback import AMMClawback
+from xrpl.models.transactions.amm_deposit import AMMDepositFlag
+
+
+class TestAMMClawback(IntegrationTestCase):
+    @test_async_and_sync(globals())
+    async def test_positive_workflow(self, client):
+        amm_pool = await create_amm_pool_async(client, enable_amm_clawback=True)
+
+        # Asset-1 is XRP, Asset-2 is an IssuedCurrency titled "USD"
+        # The Issuer of Asset-2 is the issuer_wallet
+        # For the purposes of this test, the issuer_wallet has set the
+        # Allow Trust Line Clawback flag
+        issuer_wallet = amm_pool["issuer_wallet"]
+        holder_wallet = amm_pool["lp_wallet"]
+
+        # "holder" account deposits both assets into the AMM pool
+        # Deposit assets into AMM pool
+        amm_deposit = AMMDeposit(
+            account=holder_wallet.address,
+            asset=IssuedCurrency(
+                currency="USD",
+                issuer=issuer_wallet.address,
+            ),
+            asset2=XRP(),
+            amount=IssuedCurrencyAmount(
+                currency="USD",
+                issuer=issuer_wallet.address,
+                value="10",
+            ),
+            flags=AMMDepositFlag.TF_SINGLE_ASSET,
+        )
+        deposit_response = await sign_and_reliable_submission_async(
+            amm_deposit, holder_wallet, client
+        )
+        self.assertEqual(deposit_response.result["engine_result"], "tesSUCCESS")
+
+        # Clawback one of the assets from the AMM pool
+        amm_clawback = AMMClawback(
+            account=issuer_wallet.address,
+            holder=holder_wallet.address,
+            asset=IssuedCurrency(
+                currency="USD",
+                issuer=issuer_wallet.address,
+            ),
+            asset2=XRP(),
+        )
+        clawback_response = await sign_and_reliable_submission_async(
+            amm_clawback, issuer_wallet, client
+        )
+        self.assertEqual(clawback_response.result["engine_result"], "tesSUCCESS")

tests/unit/models/transactions/test_amm_clawback.py

@@ -0,0 +1,73 @@
+from unittest import TestCase
+
+from xrpl.models.amounts import IssuedCurrencyAmount
+from xrpl.models.currencies import XRP, IssuedCurrency
+from xrpl.models.exceptions import XRPLModelException
+from xrpl.models.transactions import AMMClawback
+from xrpl.models.transactions.amm_clawback import AMMClawbackFlag
+
+_ISSUER_ACCOUNT = "r9LqNeG6qHxjeUocjvVki2XR35weJ9mZgQ"
+_ASSET2 = XRP()
+_INVALID_ASSET = IssuedCurrency(
+    currency="ETH", issuer="rpGtkFRXhgVaBzC5XCR7gyE2AZN5SN3SEW"
+)
+_VALID_ASSET = IssuedCurrency(currency="ETH", issuer=_ISSUER_ACCOUNT)
+_HOLDER_ACCOUNT = "rNZdsTBP5tH1M6GHC6bTreHAp6ouP8iZSh"
+
+
+class TestAMMClawback(TestCase):
+    def test_identical_issuer_holder_wallets(self):
+        with self.assertRaises(XRPLModelException) as error:
+            AMMClawback(
+                account=_ISSUER_ACCOUNT,
+                holder=_ISSUER_ACCOUNT,
+                asset=_VALID_ASSET,
+                asset2=_ASSET2,
+            )
+        self.assertEqual(
+            error.exception.args[0],
+            "{'AMMClawback': 'Issuer and holder wallets must be distinct.'}",
+        )
+
+    def test_incorrect_asset_issuer(self):
+        with self.assertRaises(XRPLModelException) as error:
+            AMMClawback(
+                account=_ISSUER_ACCOUNT,
+                holder=_HOLDER_ACCOUNT,
+                asset=_INVALID_ASSET,
+                asset2=_ASSET2,
+            )
+        self.assertEqual(
+            error.exception.args[0],
+            "{'AMMClawback': 'Asset.issuer and AMMClawback transaction sender must be "
+            + "identical.'}",
+        )
+
+    def test_incorrect_asset_amount(self):
+        with self.assertRaises(XRPLModelException) as error:
+            AMMClawback(
+                account=_ISSUER_ACCOUNT,
+                holder=_HOLDER_ACCOUNT,
+                asset=_VALID_ASSET,
+                asset2=_ASSET2,
+                amount=IssuedCurrencyAmount(
+                    currency="BTC",
+                    issuer="rfpFv97Dwu89FTyUwPjtpZBbuZxTqqgTmH",
+                    value="100",
+                ),
+            )
+        self.assertEqual(
+            error.exception.args[0],
+            "{'AMMClawback': 'Amount.issuer and Amount.currency must match "
+            + "corresponding Asset fields.'}",
+        )
+
+    def test_valid_txn(self):
+        txn = AMMClawback(
+            account=_ISSUER_ACCOUNT,
+            holder=_HOLDER_ACCOUNT,
+            asset=_VALID_ASSET,
+            asset2=_ASSET2,
+            flags=AMMClawbackFlag.TF_CLAW_TWO_ASSETS,
+        )
+        self.assertTrue(txn.is_valid())

xrpl/core/binarycodec/definitions/definitions.json

@@ -3053,6 +3053,7 @@
   },
   "TRANSACTION_TYPES": {
     "AMMBid": 39,
+    "AMMClawback": 31,
     "AMMCreate": 35,
     "AMMDelete": 40,
     "AMMDeposit": 36,

xrpl/models/transactions/__init__.py

@@ -11,6 +11,7 @@
     AccountSetFlagInterface,
 )
 from xrpl.models.transactions.amm_bid import AMMBid, AuthAccount
+from xrpl.models.transactions.amm_clawback import AMMClawback
 from xrpl.models.transactions.amm_create import AMMCreate
 from xrpl.models.transactions.amm_delete import AMMDelete
 from xrpl.models.transactions.amm_deposit import (
@@ -119,6 +120,7 @@
     "AccountSetFlag",
     "AccountSetFlagInterface",
     "AMMBid",
+    "AMMClawback",
     "AMMCreate",
     "AMMDelete",
     "AMMDeposit",

xrpl/models/transactions/amm_clawback.py

@@ -0,0 +1,109 @@
+"""Model for AMMClawback transaction type."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Dict, Optional
+
+from typing_extensions import Self
+
+from xrpl.models.amounts import IssuedCurrencyAmount
+from xrpl.models.currencies import Currency
+from xrpl.models.currencies.issued_currency import IssuedCurrency
+from xrpl.models.flags import FlagInterface
+from xrpl.models.required import REQUIRED
+from xrpl.models.transactions.transaction import Transaction
+from xrpl.models.transactions.types import TransactionType
+from xrpl.models.utils import KW_ONLY_DATACLASS, require_kwargs_on_init
+
+
+class AMMClawbackFlag(int, Enum):
+    """
+    Claw back the specified amount of Asset, and a corresponding amount of Asset2 based
+    on the AMM pool's asset proportion; both assets must be issued by the issuer in the
+    Account field. If this flag isn't enabled, the issuer claws back the specified
+    amount of Asset, while a corresponding proportion of Asset2 goes back to the Holder.
+    """
+
+    TF_CLAW_TWO_ASSETS = 0x00000001
+
+
+class AMMClawbackFlagInterface(FlagInterface):
+    """
+    Claw back the specified amount of Asset, and a corresponding amount of Asset2 based
+    on the AMM pool's asset proportion; both assets must be issued by the issuer in the
+    Account field. If this flag isn't enabled, the issuer claws back the specified
+    amount of Asset, while a corresponding proportion of Asset2 goes back to the Holder.
+    """
+
+    TF_CLAW_TWO_ASSETS: bool
+
+
+@require_kwargs_on_init
+@dataclass(frozen=True, **KW_ONLY_DATACLASS)
+class AMMClawback(Transaction):
+    """
+    Claw back tokens from a holder who has deposited your issued tokens into an AMM
+    pool.
+    """
+
+    holder: str = REQUIRED  # type: ignore
+    """The account holding the asset to be clawed back."""
+
+    asset: IssuedCurrency = REQUIRED  # type: ignore
+    """
+    Specifies the asset that the issuer wants to claw back from the AMM pool. In JSON,
+    this is an object with currency and issuer fields. The issuer field must match with
+    Account.
+    """
+
+    asset2: Currency = REQUIRED  # type: ignore
+    """
+    Specifies the other asset in the AMM's pool. In JSON, this is an object with
+    currency and issuer fields (omit issuer for XRP).
+    """
+
+    amount: Optional[IssuedCurrencyAmount] = None
+    """
+    The maximum amount to claw back from the AMM account. The currency and issuer
+    subfields should match the Asset subfields. If this field isn't specified, or the
+    value subfield exceeds the holder's available tokens in the AMM, all of the
+    holder's tokens are clawed back.
+    """
+
+    transaction_type: TransactionType = field(
+        default=TransactionType.AMM_CLAWBACK,
+        init=False,
+    )
+
+    def _get_errors(self: Self) -> Dict[str, str]:
+        return {
+            key: value
+            for key, value in {
+                **super()._get_errors(),
+                "AMMClawback": self._validate_wallet_and_amount_fields(),
+            }.items()
+            if value is not None
+        }
+
+    def _validate_wallet_and_amount_fields(self: Self) -> Optional[str]:
+        errors = ""
+        if self.account == self.holder:
+            errors += "Issuer and holder wallets must be distinct."
+
+        if self.account != self.asset.issuer:
+            errors += (
+                "Asset.issuer and AMMClawback transaction sender must be identical."
+            )
+
+        if self.amount is not None and (
+            self.amount.issuer != self.asset.issuer
+            or self.amount.currency != self.asset.currency
+        ):
+            errors += (
+                "Amount.issuer and Amount.currency must match corresponding Asset "
+                + "fields."
+            )
+
+        return errors if errors else None

xrpl/models/transactions/types/transaction_type.py

@@ -10,6 +10,7 @@ class TransactionType(str, Enum):
     ACCOUNT_SET = "AccountSet"
     AMM_BID = "AMMBid"
     AMM_CREATE = "AMMCreate"
+    AMM_CLAWBACK = "AMMClawback"
     AMM_DELETE = "AMMDelete"
     AMM_DEPOSIT = "AMMDeposit"
     AMM_VOTE = "AMMVote"