feat: add e2e test for native asset transfer functionality (#576)

linconvidal · web-flow · commit e5fce1aaa23e · 2025-09-03T12:12:47.000-03:00
diff --git a/e2e_tests/README.md b/e2e_tests/README.md
@@ -5,6 +5,7 @@ End-to-end testing framework for Cardano's Rosetta API implementation.
 ## Features
 
 - Basic ADA transfers
+- Native token transfers (tokenBundle)
 - UTXO management
 - Transaction construction and signing
 - Multi-input/output transactions
@@ -29,6 +30,7 @@ e2e_tests/
 │   └── pycardano_wallet.py  # PyCardano wallet wrapper
 └── tests/                   # Test cases
     ├── test_multi_io_transactions.py  # Multi-input/output tests
+    ├── test_native_asset_transfer.py  # Native asset transfer test
     └── test_stake_operations.py       # Stake operations tests
 ```
 
@@ -39,7 +41,8 @@ e2e_tests/
 - Python 3.11+
 - pip
 - Cardano Rosetta API endpoint
-- Test wallet with funds and at least 11 ada-only utxos
+- Test wallet with funds and at least 11 ada-only UTXOs
+- For native asset tests: at least 1 UTXO containing a token bundle and at least 1 ADA-only UTXO (≥ 5 ADA recommended to cover fee)
 
 ### Quick Start
 
@@ -69,7 +72,6 @@ DREP_SCRIPT_HASH_ID=2d4cb680b5f400d3521d272b4295d61150e0eff3950ef4285406a953 # R
 POOL_REGISTRATION_CERT=<hex-encoded-cert> # Required for poolRegistrationWithCert test
 POOL_GOVERNANCE_PROPOSAL_ID=df58f714c0765f3489afb6909384a16c31d600695be7e86ff9c59cf2e8a48c7900 # Required for pool governance vote tests
 POOL_VOTE_CHOICE=yes # Optional: Vote choice for pool governance vote (yes/no/abstain, default: yes)
-```
 
 ## Usage
 
@@ -140,6 +142,11 @@ pytest --log-cli-level=INFO tests/test_pool_operations.py::test_pool_registratio
 pytest --log-cli-level=INFO tests/test_pool_operations.py::test_pool_registration_with_cert
 pytest --log-cli-level=INFO tests/test_pool_operations.py::test_pool_governance_vote
 pytest --log-cli-level=INFO tests/test_pool_operations.py::test_pool_retirement
+
+# Native asset tests
+
+# Run the native asset self-transfer test (spends a token UTXO and recreates the token bundle to self)
+pytest --log-cli-level=INFO tests/test_native_asset_transfer.py::test_native_asset_self_transfer
 ```
 
 ## Test Scenarios
@@ -151,6 +158,13 @@ pytest --log-cli-level=INFO tests/test_pool_operations.py::test_pool_retirement
 3. **Fan-out**: Single input → Multiple outputs
 4. **Complex**: Multiple inputs → Multiple outputs
 5. **Fixed Fee**: Transaction with a fixed 4 ADA fee (4,000,000 lovelaces) that bypasses suggested fee
+6. **Native Asset Self-Transfer**: Spend a UTXO carrying native tokens (tokenBundle) and recreate the same bundle to self; add an ADA-only UTXO to comfortably cover fees.
+
+### Notes for Native Asset Tests
+
+- The Rosetta response for `/account/coins` is expected to include token bundles on UTXOs in `coin.metadata`.
+- The test attaches a `tokenBundle` to the input (with negative values) and to the token output (with positive values), matching the documented format in `docs/user-guides/multi-assets.md`.
+- Ensure your wallet has at least one UTXO that carries a token bundle and a separate ADA-only UTXO with enough ADA to pay fees.
 
 ### Stake Operations
 
diff --git a/e2e_tests/test_utils/utxo_selector.py b/e2e_tests/test_utils/utxo_selector.py
@@ -167,4 +167,49 @@ def select_utxos(
         total_selected = sum(int(utxo["amount"]["value"]) for utxo in selected_utxos)
         logger.debug(f"Selected {len(selected_utxos)} UTXOs with total {total_selected} lovelace")
         
-        return selected_utxos 
+        return selected_utxos 
+
+    @staticmethod
+    def find_first_asset_utxo(client: RosettaClient, address: str, exclude_utxos: Optional[List[str]] = None) -> Dict:
+        """
+        Find the first UTXO that carries native assets (token bundle) for the given address.
+
+        Supports the Cardano Rosetta representation where `coin.metadata` contains
+        a map keyed by the UTXO identifier with a list of policyId + tokens.
+
+        Args:
+            client: RosettaClient instance
+            address: Address to inspect
+            exclude_utxos: Optional list of UTXO IDs to exclude
+
+        Returns:
+            The first matching UTXO dict containing assets
+
+        Raises:
+            ValidationError: If no asset UTXO found
+        """
+        coins = client.get_utxos(address)
+        if not coins:
+            raise ValidationError(f"No UTXOs found for address {address}")
+
+        for utxo in coins:
+            utxo_id = utxo.get("coin_identifier", {}).get("identifier")
+            if not utxo_id:
+                continue
+            if exclude_utxos and utxo_id in exclude_utxos:
+                continue
+
+            md = utxo.get("metadata") or {}
+            # Newer shape: metadata keyed by utxo id -> [ { policyId, tokens: [...] }, ... ]
+            bundle_list = md.get(utxo_id)
+            if isinstance(bundle_list, list) and len(bundle_list) > 0:
+                # Double-check presence of tokens in the first bundle
+                tokens = bundle_list[0].get("tokens") if isinstance(bundle_list[0], dict) else None
+                if isinstance(tokens, list) and len(tokens) > 0:
+                    return utxo
+
+            # Fallback for older shape (if any): metadata["assets"] exists and non-empty
+            if isinstance(md.get("assets"), list) and len(md.get("assets")) > 0:
+                return utxo
+
+        raise ValidationError(f"No native-asset UTXOs found for address {address}")
diff --git a/e2e_tests/tests/test_native_asset_transfer.py b/e2e_tests/tests/test_native_asset_transfer.py
@@ -0,0 +1,177 @@
+import logging
+import pytest
+
+from e2e_tests.test_utils.utxo_selector import UtxoSelector
+
+logger = logging.getLogger(__name__)
+
+
+def _extract_token_bundle_from_utxo(utxo: dict) -> list:
+    """
+    Extract the tokenBundle list from an account/coins UTXO entry.
+
+    Supports the Rosetta representation where coin.metadata is a map keyed by
+    the UTXO identifier. Falls back to legacy shapes if necessary.
+    """
+    md = utxo.get("metadata") or {}
+    utxo_id = utxo.get("coin_identifier", {}).get("identifier")
+    if utxo_id and isinstance(md.get(utxo_id), list):
+        return md[utxo_id]
+    # Fallback (legacy): a flat assets list
+    if isinstance(md.get("assets"), list):
+        # Convert a legacy flat list into a tokenBundle-like list if possible
+        # Expect entries like { policyId, tokens: [...] }
+        return md["assets"]
+    return []
+
+
+def _negate_token_bundle_values(bundle: list) -> list:
+    """Return a deep-copied tokenBundle list with token values negated (for inputs)."""
+    out = []
+    for policy in bundle:
+        if not isinstance(policy, dict):
+            continue
+        tokens = []
+        for t in policy.get("tokens", []) or []:
+            try:
+                v = int(str(t.get("value", "0")))
+            except Exception:
+                v = 0
+            tokens.append({
+                "value": str(-v),
+                "currency": t.get("currency")
+            })
+        out.append({
+            "policyId": policy.get("policyId"),
+            "tokens": tokens
+        })
+    return out
+
+
+@pytest.mark.e2e
+def test_native_asset_self_transfer(
+    rosetta_client,
+    test_wallet,
+    transaction_orchestrator,
+    signing_handler,
+    utxo_selector,
+):
+    """
+    E2E: Spend a UTXO carrying native tokens and recreate the token bundle in an output to self.
+
+    - Select a UTXO with a tokenBundle (native asset)
+    - Add one ADA-only UTXO to comfortably pay the fee
+    - Create input ops for both UTXOs; attach tokenBundle metadata only to the asset input
+    - Create two outputs to self: one reproducing the token bundle with its ADA; one change output for remaining ADA
+    - Build, sign, submit, and wait for confirmation
+    - Verify the confirmed transaction includes the tokenBundle on an output
+    """
+    sender_addr = test_wallet.get_address()
+
+    # 1) Select asset UTXO
+    asset_utxo = UtxoSelector.find_first_asset_utxo(rosetta_client, sender_addr)
+    asset_utxo_id = asset_utxo["coin_identifier"]["identifier"]
+    asset_utxo_ada = int(asset_utxo["amount"]["value"])  # ADA contained in the asset UTXO
+    token_bundle = _extract_token_bundle_from_utxo(asset_utxo)
+    assert token_bundle, "Selected UTXO must contain token bundle"
+
+    logger.info(f"Using asset UTXO {asset_utxo_id} with {asset_utxo_ada} lovelace and token bundle")
+
+    # 2) Add an ADA-only UTXO to cover fee and keep change positive
+    ada_only_inputs = utxo_selector.select_utxos(
+        rosetta_client,
+        sender_addr,
+        required_amount=5_000_000,
+        exclude_utxos=[asset_utxo_id],
+        strategy="single",
+        utxo_count=1,
+    )
+    ada_utxo = ada_only_inputs[0]
+    ada_utxo_id = ada_utxo["coin_identifier"]["identifier"]
+    ada_utxo_ada = int(ada_utxo["amount"]["value"])  # ADA-only amount
+
+    total_input_ada = asset_utxo_ada + ada_utxo_ada
+
+    # 3) Build operations
+    ops = []
+
+    # Input: asset UTXO (ADA negative; tokenBundle values negative)
+    ops.append({
+        "operation_identifier": {"index": len(ops)},
+        "type": "input",
+        "account": {"address": sender_addr},
+        "amount": {"value": f"-{asset_utxo_ada}", "currency": {"symbol": "ADA", "decimals": 6}},
+        "coin_change": {
+            "coin_identifier": {"identifier": asset_utxo_id},
+            "coin_action": "coin_spent",
+        },
+        "metadata": {
+            "tokenBundle": _negate_token_bundle_values(token_bundle)
+        },
+    })
+
+    # Input: ADA-only UTXO
+    ops.append({
+        "operation_identifier": {"index": len(ops)},
+        "type": "input",
+        "account": {"address": sender_addr},
+        "amount": {"value": f"-{ada_utxo_ada}", "currency": {"symbol": "ADA", "decimals": 6}},
+        "coin_change": {
+            "coin_identifier": {"identifier": ada_utxo_id},
+            "coin_action": "coin_spent",
+        },
+    })
+
+    # Output: token bundle to self, ADA equals the ADA in the asset UTXO
+    ops.append({
+        "operation_identifier": {"index": len(ops)},
+        "type": "output",
+        "account": {"address": sender_addr},
+        "amount": {"value": str(asset_utxo_ada), "currency": {"symbol": "ADA", "decimals": 6}},
+        "metadata": {
+            "tokenBundle": token_bundle
+        },
+    })
+
+    # Output: change to self, ADA is remaining; fee will be deducted by orchestrator
+    change_ada = total_input_ada - asset_utxo_ada
+    assert change_ada > 0, "Change must be positive before fee deduction"
+    ops.append({
+        "operation_identifier": {"index": len(ops)},
+        "type": "output",
+        "account": {"address": sender_addr},
+        "amount": {"value": str(change_ada), "currency": {"symbol": "ADA", "decimals": 6}},
+    })
+
+    # 4) Build unsigned tx (let orchestrator fetch suggested fee and adjust change)
+    unsigned_tx, payloads, metadata, fee = transaction_orchestrator.build_transaction(ops)
+    logger.info(f"Unsigned tx built. Suggested fee: {fee} lovelace")
+
+    # 5) Sign and submit; wait for confirmation
+    sign_fn = signing_handler.get_signing_function()  # payment key
+    tx_hash, tx_details = transaction_orchestrator.sign_and_submit(
+        unsigned_transaction=unsigned_tx,
+        payloads=payloads,
+        signing_function=sign_fn,
+        wait_for_confirmation=True,
+        confirmation_timeout=180,
+    )
+
+    logger.info(f"Native asset tx confirmed · {tx_hash}")
+
+    # 6) Verify the confirmed tx includes an output with the tokenBundle
+    assert tx_details and "transaction" in tx_details, "Missing transaction details after confirmation"
+    parsed_ops = tx_details["transaction"].get("operations", [])
+    assert parsed_ops, "Confirmed transaction should include operations"
+
+    has_token_output = False
+    for op in parsed_ops:
+        if op.get("type") == "output":
+            md = op.get("metadata") or {}
+            tb = md.get("tokenBundle")
+            if isinstance(tb, list) and len(tb) > 0:
+                has_token_output = True
+                break
+
+    assert has_token_output, "Expected an output operation with tokenBundle metadata"
+
