Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 19 additions & 10 deletions intentkit/wallets/cdp.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from intentkit.config.db import get_session
from intentkit.models.agent import Agent, AgentTable
from intentkit.models.agent_data import AgentData
from intentkit.utils.chain import SupportedNetwork, resolve_supported_network
from intentkit.utils.error import IntentKitAPIError
from intentkit.wallets.web3 import get_async_web3_client

Expand Down Expand Up @@ -190,27 +191,35 @@ async def get_evm_account(agent: Agent) -> EvmServerAccount:
return account


_SUPPORTED_NETWORK_TO_CDP: dict[SupportedNetwork, str] = {
SupportedNetwork.EthereumMainnet: "ethereum",
SupportedNetwork.BaseMainnet: "base",
SupportedNetwork.ArbitrumMainnet: "arbitrum",
SupportedNetwork.OptimismMainnet: "optimism",
SupportedNetwork.PolygonMainnet: "polygon",
SupportedNetwork.BaseSepolia: "base-sepolia",
SupportedNetwork.BnbMainnet: "bsc",
}


def get_cdp_network(agent: Agent) -> str:
if not agent.network_id:
raise IntentKitAPIError(
400,
"BadNetworkID",
"Your agent network ID is not set. Please set it in the agent config.",
)
mapping = {
"ethereum-mainnet": "ethereum",
"base-mainnet": "base",
"arbitrum-mainnet": "arbitrum",
"optimism-mainnet": "optimism",
"polygon-mainnet": "polygon",
"base-sepolia": "base-sepolia",
"bnb-mainnet": "bsc",
}
if agent.network_id == "solana":
raise IntentKitAPIError(
400, "BadNetworkID", "Solana is not supported by CDP EVM."
)
cdp_network = mapping.get(agent.network_id)
try:
supported_network = resolve_supported_network(agent.network_id)
except ValueError:
raise IntentKitAPIError(
400, "BadNetworkID", f"Unsupported network ID: {agent.network_id}"
)
cdp_network = _SUPPORTED_NETWORK_TO_CDP.get(supported_network)
if not cdp_network:
raise IntentKitAPIError(
400, "BadNetworkID", f"Unsupported network ID: {agent.network_id}"
Expand Down
88 changes: 88 additions & 0 deletions tests/wallets/test_cdp_network.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
"""Tests for get_cdp_network() in intentkit/wallets/cdp.py."""

from types import SimpleNamespace

import pytest

from intentkit.utils.error import IntentKitAPIError
from intentkit.wallets.cdp import get_cdp_network


def _agent(network_id):
"""Return a minimal agent-like object with the given network_id."""
return SimpleNamespace(network_id=network_id)


class TestGetCdpNetworkCanonicalNames:
"""Canonical network IDs that were already supported."""

def test_base_mainnet(self):
assert get_cdp_network(_agent("base-mainnet")) == "base"

def test_ethereum_mainnet(self):
assert get_cdp_network(_agent("ethereum-mainnet")) == "ethereum"

def test_arbitrum_mainnet(self):
assert get_cdp_network(_agent("arbitrum-mainnet")) == "arbitrum"

def test_optimism_mainnet(self):
assert get_cdp_network(_agent("optimism-mainnet")) == "optimism"

def test_polygon_mainnet(self):
assert get_cdp_network(_agent("polygon-mainnet")) == "polygon"

def test_base_sepolia(self):
assert get_cdp_network(_agent("base-sepolia")) == "base-sepolia"

def test_bnb_mainnet(self):
assert get_cdp_network(_agent("bnb-mainnet")) == "bsc"


class TestGetCdpNetworkAliases:
"""Network aliases that resolve via AGENT_NETWORK_TO_SUPPORTED_NETWORK."""

def test_ethereum_alias(self):
# "ethereum" is an alias for "ethereum-mainnet"
assert get_cdp_network(_agent("ethereum")) == "ethereum"

def test_polygon_alias(self):
# "polygon" is an alias for "polygon-mainnet"
assert get_cdp_network(_agent("polygon")) == "polygon"

def test_matic_alias(self):
# "matic" is an alias for "polygon-mainnet"
assert get_cdp_network(_agent("matic")) == "polygon"

def test_matic_mainnet_alias(self):
assert get_cdp_network(_agent("matic-mainnet")) == "polygon"

def test_bsc_mainnet_alias(self):
# "bsc-mainnet" is an alias for "bnb-mainnet"
assert get_cdp_network(_agent("bsc-mainnet")) == "bsc"

def test_binance_mainnet_alias(self):
# "binance-mainnet" is an alias for "bnb-mainnet"
assert get_cdp_network(_agent("binance-mainnet")) == "bsc"


class TestGetCdpNetworkErrors:
def test_missing_network_id_raises(self):
with pytest.raises(IntentKitAPIError) as exc_info:
get_cdp_network(_agent(None))
assert exc_info.value.status_code == 400

def test_empty_network_id_raises(self):
with pytest.raises(IntentKitAPIError) as exc_info:
get_cdp_network(_agent(""))
assert exc_info.value.status_code == 400

def test_solana_raises(self):
with pytest.raises(IntentKitAPIError) as exc_info:
get_cdp_network(_agent("solana"))
assert exc_info.value.status_code == 400
assert "Solana" in str(exc_info.value)

def test_unknown_network_raises(self):
with pytest.raises(IntentKitAPIError) as exc_info:
get_cdp_network(_agent("not-a-real-network"))
assert exc_info.value.status_code == 400