diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f69fff..915995e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ ### Fixed - Fix bug in `Asset.from_model` where passed in asset ID was not used when creating a gwei or wei asset. +### Added + +- Add `FundOperation` and `FundQuote` classes to support wallet funding. + ## [0.10.3] - 2024-11-07 ### Added diff --git a/cdp/api_clients.py b/cdp/api_clients.py index 46c7532..e60e17b 100644 --- a/cdp/api_clients.py +++ b/cdp/api_clients.py @@ -4,6 +4,7 @@ from cdp.client.api.balance_history_api import BalanceHistoryApi from cdp.client.api.contract_invocations_api import ContractInvocationsApi from cdp.client.api.external_addresses_api import ExternalAddressesApi +from cdp.client.api.fund_api import FundApi from cdp.client.api.networks_api import NetworksApi from cdp.client.api.smart_contracts_api import SmartContractsApi from cdp.client.api.trades_api import TradesApi @@ -53,6 +54,7 @@ def __init__(self, cdp_client: CdpApiClient) -> None: self._smart_contracts: SmartContractsApi | None = None self._balance_history: BalanceHistoryApi | None = None self._transaction_history: TransactionHistoryApi | None = None + self._fund: FundApi | None = None @property def wallets(self) -> WalletsApi: @@ -233,3 +235,18 @@ def transaction_history(self) -> TransactionHistoryApi: if self._transaction_history is None: self._transaction_history = TransactionHistoryApi(api_client=self._cdp_client) return self._transaction_history + + @property + def fund(self) -> FundApi: + """Get the FundApi client instance. + + Returns: + FundApi: The FundApi client instance. + + Note: + This property lazily initializes the FundApi client on first access. + + """ + if self._fund is None: + self._fund = FundApi(api_client=self._cdp_client) + return self._fund diff --git a/cdp/client/__init__.py b/cdp/client/__init__.py index b0a5107..52e517f 100644 --- a/cdp/client/__init__.py +++ b/cdp/client/__init__.py @@ -27,6 +27,7 @@ from cdp.client.api.mpc_wallet_stake_api import MPCWalletStakeApi from cdp.client.api.networks_api import NetworksApi from cdp.client.api.onchain_identity_api import OnchainIdentityApi +from cdp.client.api.reputation_api import ReputationApi from cdp.client.api.server_signers_api import ServerSignersApi from cdp.client.api.smart_contracts_api import SmartContractsApi from cdp.client.api.stake_api import StakeApi @@ -53,6 +54,9 @@ from cdp.client.models.address_balance_list import AddressBalanceList from cdp.client.models.address_historical_balance_list import AddressHistoricalBalanceList from cdp.client.models.address_list import AddressList +from cdp.client.models.address_reputation import AddressReputation +from cdp.client.models.address_reputation_metadata import AddressReputationMetadata +from cdp.client.models.address_risk import AddressRisk from cdp.client.models.address_transaction_list import AddressTransactionList from cdp.client.models.asset import Asset from cdp.client.models.balance import Balance diff --git a/cdp/client/api/__init__.py b/cdp/client/api/__init__.py index f70084c..09e4751 100644 --- a/cdp/client/api/__init__.py +++ b/cdp/client/api/__init__.py @@ -11,6 +11,7 @@ from cdp.client.api.mpc_wallet_stake_api import MPCWalletStakeApi from cdp.client.api.networks_api import NetworksApi from cdp.client.api.onchain_identity_api import OnchainIdentityApi +from cdp.client.api.reputation_api import ReputationApi from cdp.client.api.server_signers_api import ServerSignersApi from cdp.client.api.smart_contracts_api import SmartContractsApi from cdp.client.api.stake_api import StakeApi diff --git a/cdp/client/api/reputation_api.py b/cdp/client/api/reputation_api.py new file mode 100644 index 0000000..5d77525 --- /dev/null +++ b/cdp/client/api/reputation_api.py @@ -0,0 +1,589 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + +import warnings +from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt +from typing import Any, Dict, List, Optional, Tuple, Union +from typing_extensions import Annotated + +from pydantic import Field, StrictStr +from typing_extensions import Annotated +from cdp.client.models.address_reputation import AddressReputation +from cdp.client.models.address_risk import AddressRisk + +from cdp.client.api_client import ApiClient, RequestSerialized +from cdp.client.api_response import ApiResponse +from cdp.client.rest import RESTResponseType + + +class ReputationApi: + """NOTE: This class is auto generated by OpenAPI Generator + Ref: https://openapi-generator.tech + + Do not edit the class manually. + """ + + def __init__(self, api_client=None) -> None: + if api_client is None: + api_client = ApiClient.get_default() + self.api_client = api_client + + + @validate_call + def get_address_reputation( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network.")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the reputation for.")], + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> AddressReputation: + """Get the onchain reputation of an external address + + Get the onchain reputation of an external address + + :param network_id: The ID of the blockchain network. (required) + :type network_id: str + :param address_id: The ID of the address to fetch the reputation for. (required) + :type address_id: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._get_address_reputation_serialize( + network_id=network_id, + address_id=address_id, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "AddressReputation", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ).data + + + @validate_call + def get_address_reputation_with_http_info( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network.")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the reputation for.")], + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> ApiResponse[AddressReputation]: + """Get the onchain reputation of an external address + + Get the onchain reputation of an external address + + :param network_id: The ID of the blockchain network. (required) + :type network_id: str + :param address_id: The ID of the address to fetch the reputation for. (required) + :type address_id: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._get_address_reputation_serialize( + network_id=network_id, + address_id=address_id, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "AddressReputation", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ) + + + @validate_call + def get_address_reputation_without_preload_content( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network.")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the reputation for.")], + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> RESTResponseType: + """Get the onchain reputation of an external address + + Get the onchain reputation of an external address + + :param network_id: The ID of the blockchain network. (required) + :type network_id: str + :param address_id: The ID of the address to fetch the reputation for. (required) + :type address_id: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._get_address_reputation_serialize( + network_id=network_id, + address_id=address_id, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "AddressReputation", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + return response_data.response + + + def _get_address_reputation_serialize( + self, + network_id, + address_id, + _request_auth, + _content_type, + _headers, + _host_index, + ) -> RequestSerialized: + + _host = None + + _collection_formats: Dict[str, str] = { + } + + _path_params: Dict[str, str] = {} + _query_params: List[Tuple[str, str]] = [] + _header_params: Dict[str, Optional[str]] = _headers or {} + _form_params: List[Tuple[str, str]] = [] + _files: Dict[ + str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] + ] = {} + _body_params: Optional[bytes] = None + + # process the path parameters + if network_id is not None: + _path_params['network_id'] = network_id + if address_id is not None: + _path_params['address_id'] = address_id + # process the query parameters + # process the header parameters + # process the form parameters + # process the body parameter + + + # set the HTTP header `Accept` + if 'Accept' not in _header_params: + _header_params['Accept'] = self.api_client.select_header_accept( + [ + 'application/json' + ] + ) + + + # authentication setting + _auth_settings: List[str] = [ + ] + + return self.api_client.param_serialize( + method='GET', + resource_path='/v1/networks/{network_id}/addresses/{address_id}/reputation', + path_params=_path_params, + query_params=_query_params, + header_params=_header_params, + body=_body_params, + post_params=_form_params, + files=_files, + auth_settings=_auth_settings, + collection_formats=_collection_formats, + _host=_host, + _request_auth=_request_auth + ) + + + + + @validate_call + def get_address_risk( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network.")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the risk for.")], + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> AddressRisk: + """Get the risk of an address + + Get the risk of an address + + :param network_id: The ID of the blockchain network. (required) + :type network_id: str + :param address_id: The ID of the address to fetch the risk for. (required) + :type address_id: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._get_address_risk_serialize( + network_id=network_id, + address_id=address_id, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "AddressRisk", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ).data + + + @validate_call + def get_address_risk_with_http_info( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network.")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the risk for.")], + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> ApiResponse[AddressRisk]: + """Get the risk of an address + + Get the risk of an address + + :param network_id: The ID of the blockchain network. (required) + :type network_id: str + :param address_id: The ID of the address to fetch the risk for. (required) + :type address_id: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._get_address_risk_serialize( + network_id=network_id, + address_id=address_id, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "AddressRisk", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ) + + + @validate_call + def get_address_risk_without_preload_content( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network.")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the risk for.")], + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> RESTResponseType: + """Get the risk of an address + + Get the risk of an address + + :param network_id: The ID of the blockchain network. (required) + :type network_id: str + :param address_id: The ID of the address to fetch the risk for. (required) + :type address_id: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._get_address_risk_serialize( + network_id=network_id, + address_id=address_id, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "AddressRisk", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + return response_data.response + + + def _get_address_risk_serialize( + self, + network_id, + address_id, + _request_auth, + _content_type, + _headers, + _host_index, + ) -> RequestSerialized: + + _host = None + + _collection_formats: Dict[str, str] = { + } + + _path_params: Dict[str, str] = {} + _query_params: List[Tuple[str, str]] = [] + _header_params: Dict[str, Optional[str]] = _headers or {} + _form_params: List[Tuple[str, str]] = [] + _files: Dict[ + str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] + ] = {} + _body_params: Optional[bytes] = None + + # process the path parameters + if network_id is not None: + _path_params['network_id'] = network_id + if address_id is not None: + _path_params['address_id'] = address_id + # process the query parameters + # process the header parameters + # process the form parameters + # process the body parameter + + + # set the HTTP header `Accept` + if 'Accept' not in _header_params: + _header_params['Accept'] = self.api_client.select_header_accept( + [ + 'application/json' + ] + ) + + + # authentication setting + _auth_settings: List[str] = [ + ] + + return self.api_client.param_serialize( + method='GET', + resource_path='/v1/networks/{network_id}/addresses/{address_id}/risk', + path_params=_path_params, + query_params=_query_params, + header_params=_header_params, + body=_body_params, + post_params=_form_params, + files=_files, + auth_settings=_auth_settings, + collection_formats=_collection_formats, + _host=_host, + _request_auth=_request_auth + ) + + diff --git a/cdp/client/models/__init__.py b/cdp/client/models/__init__.py index 188df88..db94dc4 100644 --- a/cdp/client/models/__init__.py +++ b/cdp/client/models/__init__.py @@ -18,6 +18,9 @@ from cdp.client.models.address_balance_list import AddressBalanceList from cdp.client.models.address_historical_balance_list import AddressHistoricalBalanceList from cdp.client.models.address_list import AddressList +from cdp.client.models.address_reputation import AddressReputation +from cdp.client.models.address_reputation_metadata import AddressReputationMetadata +from cdp.client.models.address_risk import AddressRisk from cdp.client.models.address_transaction_list import AddressTransactionList from cdp.client.models.asset import Asset from cdp.client.models.balance import Balance diff --git a/cdp/client/models/address_reputation.py b/cdp/client/models/address_reputation.py new file mode 100644 index 0000000..bbe381c --- /dev/null +++ b/cdp/client/models/address_reputation.py @@ -0,0 +1,93 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictInt +from typing import Any, ClassVar, Dict, List +from cdp.client.models.address_reputation_metadata import AddressReputationMetadata +from typing import Optional, Set +from typing_extensions import Self + +class AddressReputation(BaseModel): + """ + The reputation score with metadata of a blockchain address. + """ # noqa: E501 + reputation_score: StrictInt = Field(description="The reputation score of a wallet address which lie between 0 to 100.") + metadata: AddressReputationMetadata + __properties: ClassVar[List[str]] = ["reputation_score", "metadata"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of AddressReputation from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + # override the default output from pydantic by calling `to_dict()` of metadata + if self.metadata: + _dict['metadata'] = self.metadata.to_dict() + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of AddressReputation from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "reputation_score": obj.get("reputation_score"), + "metadata": AddressReputationMetadata.from_dict(obj["metadata"]) if obj.get("metadata") is not None else None + }) + return _obj + + diff --git a/cdp/client/models/address_reputation_metadata.py b/cdp/client/models/address_reputation_metadata.py new file mode 100644 index 0000000..950dbf8 --- /dev/null +++ b/cdp/client/models/address_reputation_metadata.py @@ -0,0 +1,105 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictInt +from typing import Any, ClassVar, Dict, List +from typing import Optional, Set +from typing_extensions import Self + +class AddressReputationMetadata(BaseModel): + """ + The metadata for the reputation score of onchain address. + """ # noqa: E501 + total_transactions: StrictInt = Field(description="The total number of transactions performed by the address.") + unique_days_active: StrictInt = Field(description="The number of unique days the address was active.") + longest_active_streak: StrictInt = Field(description="The longest streak of consecutive active days.") + current_active_streak: StrictInt = Field(description="The current streak of consecutive active days.") + activity_period_days: StrictInt = Field(description="The total number of days the address has been active.") + token_swaps_performed: StrictInt = Field(description="The number of token swaps performed by the address.") + bridge_transactions_performed: StrictInt = Field(description="The number of bridge transactions performed by the address.") + lend_borrow_stake_transactions: StrictInt = Field(description="The number of lend, borrow, or stake transactions performed by the address.") + ens_contract_interactions: StrictInt = Field(description="The number of interactions with ENS contracts.") + smart_contract_deployments: StrictInt = Field(description="The number of smart contracts deployed by the address.") + __properties: ClassVar[List[str]] = ["total_transactions", "unique_days_active", "longest_active_streak", "current_active_streak", "activity_period_days", "token_swaps_performed", "bridge_transactions_performed", "lend_borrow_stake_transactions", "ens_contract_interactions", "smart_contract_deployments"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of AddressReputationMetadata from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of AddressReputationMetadata from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "total_transactions": obj.get("total_transactions"), + "unique_days_active": obj.get("unique_days_active"), + "longest_active_streak": obj.get("longest_active_streak"), + "current_active_streak": obj.get("current_active_streak"), + "activity_period_days": obj.get("activity_period_days"), + "token_swaps_performed": obj.get("token_swaps_performed"), + "bridge_transactions_performed": obj.get("bridge_transactions_performed"), + "lend_borrow_stake_transactions": obj.get("lend_borrow_stake_transactions"), + "ens_contract_interactions": obj.get("ens_contract_interactions"), + "smart_contract_deployments": obj.get("smart_contract_deployments") + }) + return _obj + + diff --git a/cdp/client/models/address_risk.py b/cdp/client/models/address_risk.py new file mode 100644 index 0000000..b2db6e0 --- /dev/null +++ b/cdp/client/models/address_risk.py @@ -0,0 +1,87 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictInt +from typing import Any, ClassVar, Dict, List +from typing import Optional, Set +from typing_extensions import Self + +class AddressRisk(BaseModel): + """ + The risk score of a blockchain address. + """ # noqa: E501 + risk_score: StrictInt = Field(description="The lower the score is, the higher the risk is. The score lies between -100 to 0.") + __properties: ClassVar[List[str]] = ["risk_score"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of AddressRisk from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of AddressRisk from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "risk_score": obj.get("risk_score") + }) + return _obj + + diff --git a/cdp/crypto_amount.py b/cdp/crypto_amount.py new file mode 100644 index 0000000..63db82e --- /dev/null +++ b/cdp/crypto_amount.py @@ -0,0 +1,116 @@ +from decimal import Decimal + +from cdp.asset import Asset +from cdp.client.models.crypto_amount import CryptoAmount as CryptoAmountModel + + +class CryptoAmount: + """A representation of a CryptoAmount that includes the amount and asset.""" + + def __init__(self, amount: Decimal, asset: Asset, asset_id: str | None = None) -> None: + """Initialize a new CryptoAmount. Do not use this directly, use the from_model or from_model_and_asset_id methods instead. + + Args: + amount (Decimal): The amount of the Asset + asset (Asset): The Asset + asset_id (Optional[str]): The Asset ID + + """ + self._amount = amount + self._asset = asset + self._asset_id = asset_id or asset.asset_id + + @classmethod + def from_model(cls, crypto_amount_model: CryptoAmountModel) -> "CryptoAmount": + """Convert a CryptoAmount model to a CryptoAmount. + + Args: + crypto_amount_model (CryptoAmountModel): The crypto amount from the API. + + Returns: + CryptoAmount: The converted CryptoAmount object. + + """ + asset = Asset.from_model(crypto_amount_model.asset) + return cls(amount=asset.from_atomic_amount(crypto_amount_model.amount), asset=asset) + + @classmethod + def from_model_and_asset_id( + cls, crypto_amount_model: CryptoAmountModel, asset_id: str + ) -> "CryptoAmount": + """Convert a CryptoAmount model and asset ID to a CryptoAmount. + + This can be used to specify a non-primary denomination that we want the amount + to be converted to. + + Args: + crypto_amount_model (CryptoAmountModel): The crypto amount from the API. + asset_id (str): The Asset ID of the denomination we want returned. + + Returns: + CryptoAmount: The converted CryptoAmount object. + + """ + asset = Asset.from_model(crypto_amount_model.asset, asset_id=asset_id) + return cls( + amount=asset.from_atomic_amount(crypto_amount_model.amount), + asset=asset, + asset_id=asset_id, + ) + + @property + def amount(self) -> Decimal: + """Get the amount of the Asset. + + Returns: + Decimal: The amount of the Asset. + + """ + return self._amount + + @property + def asset(self) -> Asset: + """Get the Asset. + + Returns: + Asset: The Asset. + + """ + return self._asset + + @property + def asset_id(self) -> str: + """Get the Asset ID. + + Returns: + str: The Asset ID. + + """ + return self._asset_id + + def to_atomic_amount(self) -> Decimal: + """Convert the amount to atomic units. + + Returns: + Decimal: The amount in atomic units. + + """ + return self.asset.to_atomic_amount(self.amount) + + def __str__(self) -> str: + """Get a string representation of the CryptoAmount. + + Returns: + str: A string representation of the CryptoAmount. + + """ + return f"CryptoAmount(amount: '{int(self.amount)}', asset_id: '{self.asset_id}')" + + def __repr__(self) -> str: + """Get a string representation of the CryptoAmount. + + Returns: + str: A string representation of the CryptoAmount. + + """ + return self.__str__() diff --git a/cdp/fiat_amount.py b/cdp/fiat_amount.py new file mode 100644 index 0000000..7454c66 --- /dev/null +++ b/cdp/fiat_amount.py @@ -0,0 +1,69 @@ +from decimal import Decimal + +from cdp.client.models.fiat_amount import FiatAmount as FiatAmountModel + + +class FiatAmount: + """A representation of a FiatAmount that includes the amount and currency.""" + + def __init__(self, amount: Decimal, currency: str) -> None: + """Initialize a new FiatAmount. Do not use this directly, use the from_model method instead. + + Args: + amount (Decimal): The amount in the fiat currency + currency (str): The currency code (e.g. 'USD') + + """ + self._amount = amount + self._currency = currency + + @classmethod + def from_model(cls, fiat_amount_model: FiatAmountModel) -> "FiatAmount": + """Convert a FiatAmount model to a FiatAmount. + + Args: + fiat_amount_model (FiatAmountModel): The fiat amount from the API. + + Returns: + FiatAmount: The converted FiatAmount object. + + """ + return cls(amount=Decimal(fiat_amount_model.amount), currency=fiat_amount_model.currency) + + @property + def amount(self) -> Decimal: + """Get the amount in the fiat currency. + + Returns: + Decimal: The amount in the fiat currency. + + """ + return self._amount + + @property + def currency(self) -> str: + """Get the currency code. + + Returns: + str: The currency code. + + """ + return self._currency + + def __str__(self) -> str: + """Get a string representation of the FiatAmount. + + Returns: + str: A string representation of the FiatAmount. + + """ + return f"FiatAmount(amount: '{int(self.amount)}', currency: '{self.currency}')" + + def __repr__(self) -> str: + """Get a string representation of the FiatAmount. + + Returns: + str: A string representation of the FiatAmount. + + """ + return self.__str__() diff --git a/cdp/fund_operation.py b/cdp/fund_operation.py new file mode 100644 index 0000000..54e61e7 --- /dev/null +++ b/cdp/fund_operation.py @@ -0,0 +1,247 @@ +import time +from collections.abc import Iterator +from decimal import Decimal +from enum import Enum + +from cdp.asset import Asset +from cdp.cdp import Cdp +from cdp.client.models import FundOperation as FundOperationModel +from cdp.crypto_amount import CryptoAmount +from cdp.fiat_amount import FiatAmount +from cdp.fund_quote import FundQuote + + +class FundOperation: + """A representation of a Fund Operation.""" + + class Status(Enum): + """Fund Operation status constants.""" + + PENDING = "pending" + COMPLETE = "complete" + FAILED = "failed" + + @classmethod + def terminal_states(cls): + """Get the terminal states. + + Returns: + List[Status]: The terminal states. + + """ + return [cls.COMPLETE, cls.FAILED] + + def __str__(self) -> str: + """Return a string representation of the Status.""" + return self.value + + def __repr__(self) -> str: + """Return a string representation of the Status.""" + return str(self) + + def __init__(self, model: FundOperationModel) -> None: + """Initialize the FundOperation class. + + Args: + model (FundOperationModel): The model representing the fund operation. + + """ + self._model = model + self._network = None + self._asset = None + + @classmethod + def create( + cls, + wallet_id: str, + address_id: str, + amount: Decimal, + asset_id: str, + network_id: str, + quote: FundQuote | None = None, + ) -> "FundOperation": + """Create a new Fund Operation. + + Args: + wallet_id (str): The Wallet ID + address_id (str): The Address ID + amount (Decimal): The amount of the Asset + asset_id (str): The Asset ID + network_id (str): The Network ID + quote (Optional[FundQuote]): The optional Fund Quote + + Returns: + FundOperation: The new FundOperation object + + """ + asset = Asset.fetch(network_id, asset_id) + + create_request = { + "amount": str(int(asset.to_atomic_amount(amount))), + "asset_id": Asset.primary_denomination(asset.asset_id), + } + + if quote: + create_request["fund_quote_id"] = quote.id + + model = Cdp.api_clients.fund.create_fund_operation( + wallet_id=wallet_id, + address_id=address_id, + create_fund_operation_request=create_request, + ) + + return cls(model) + + @classmethod + def list(cls, wallet_id: str, address_id: str) -> Iterator["FundOperation"]: + """List fund operations. + + Args: + wallet_id (str): The wallet ID + address_id (str): The address ID + + Returns: + Iterator[FundOperation]: An iterator of fund operation objects + + """ + page = None + while True: + response = Cdp.api_clients.fund.list_fund_operations( + wallet_id=wallet_id, + address_id=address_id, + limit=100, + page=page, + ) + + for operation_model in response.data: + yield cls(operation_model) + + if not response.has_more: + break + + page = response.next_page + + @property + def id(self) -> str: + """Get the Fund Operation ID.""" + return self._model.fund_operation_id + + @property + def network_id(self) -> str: + """Get the Network ID.""" + return self._model.network_id + + @property + def wallet_id(self) -> str: + """Get the Wallet ID.""" + return self._model.wallet_id + + @property + def address_id(self) -> str: + """Get the Address ID.""" + return self._model.address_id + + @property + def asset(self) -> Asset: + """Get the Asset.""" + if self._asset is None: + self._asset = Asset.from_model(self._model.crypto_amount.asset) + return self._asset + + @property + def amount(self) -> CryptoAmount: + """Get the crypto amount.""" + return CryptoAmount.from_model(self._model.crypto_amount) + + @property + def fiat_amount(self) -> FiatAmount: + """Get the fiat amount.""" + return FiatAmount.from_model(self._model.fiat_amount) + + @property + def fiat_currency(self) -> str: + """Get the fiat currency.""" + return self._model.fiat_amount.currency + + @property + def buy_fee(self) -> dict: + """Get the buy fee. + + Returns: + dict: The buy fee information. + + """ + return { + "amount": self._model.fees.buy_fee.amount, + "currency": self._model.fees.buy_fee.currency, + } + + @property + def transfer_fee(self) -> CryptoAmount: + """Get the transfer fee. + + Returns: + CryptoAmount: The transfer fee. + + """ + return CryptoAmount.from_model(self._model.fees.transfer_fee) + + @property + def status(self) -> Status: + """Get the status.""" + return self.Status(self._model.status) + + def reload(self) -> "FundOperation": + """Reload the fund operation from the server. + + Returns: + FundOperation: The updated fund operation. + + """ + self._model = Cdp.api_clients.fund.get_fund_operation( + self.wallet_id, self.address_id, self.id + ) + return self + + def wait(self, interval_seconds: float = 0.2, timeout_seconds: float = 20) -> "FundOperation": + """Wait for the fund operation to complete. + + Args: + interval_seconds (float): The interval between checks + timeout_seconds (float): The maximum time to wait + + Returns: + FundOperation: The completed fund operation + + Raises: + TimeoutError: If the operation takes too long + + """ + start_time = time.time() + + while not self.terminal_state(): + self.reload() + + if time.time() - start_time > timeout_seconds: + raise TimeoutError("Fund operation timed out") + + time.sleep(interval_seconds) + + return self + + def terminal_state(self) -> bool: + """Check if the operation is in a terminal state.""" + return self.status in self.Status.terminal_states() + + def __str__(self) -> str: + """Get a string representation of the Fund Operation.""" + return ( + f"FundOperation(id: {self.id}, network_id: {self.network_id}, " + f"wallet_id: {self.wallet_id}, address_id: {self.address_id}, " + f"amount: {self.amount}, asset_id: {self.asset.asset_id}, " + f"status: {self.status.value})" + ) + + def __repr__(self) -> str: + """Get a string representation of the Fund Operation.""" + return self.__str__() diff --git a/cdp/fund_quote.py b/cdp/fund_quote.py new file mode 100644 index 0000000..36e318e --- /dev/null +++ b/cdp/fund_quote.py @@ -0,0 +1,199 @@ +from decimal import Decimal +from typing import TYPE_CHECKING + +from cdp.asset import Asset +from cdp.cdp import Cdp +from cdp.crypto_amount import CryptoAmount +from cdp.fiat_amount import FiatAmount + +if TYPE_CHECKING: + from cdp.fund_operation import FundOperation + +from cdp.client.models import FundQuote as FundQuoteModel + + +class FundQuote: + """A representation of a Fund Operation Quote.""" + + def __init__(self, model: FundQuoteModel) -> None: + """Initialize the FundQuote class. + + Args: + model (FundQuoteModel): The model representing the fund quote. + + """ + self._model = model + self._network = None + self._asset = None + + @classmethod + def create( + cls, + wallet_id: str, + address_id: str, + amount: Decimal, + asset_id: str, + network_id: str, + ) -> "FundQuote": + """Create a new Fund Operation Quote. + + Args: + wallet_id (str): The Wallet ID + address_id (str): The Address ID + amount (Decimal): The amount of the Asset + asset_id (str): The Asset ID + network_id (str): The Network ID + + Returns: + FundQuote: The new FundQuote object + + """ + asset = Asset.fetch(network_id, asset_id) + + model = Cdp.api_clients.fund.create_fund_quote( + wallet_id=wallet_id, + address_id=address_id, + create_fund_quote_request={ + "asset_id": Asset.primary_denomination(asset.asset_id), + "amount": str(int(asset.to_atomic_amount(amount))), + }, + ) + + return cls(model) + + @property + def id(self) -> str: + """Get the Fund Quote ID. + + Returns: + str: The Fund Quote ID. + + """ + return self._model.fund_quote_id + + @property + def network_id(self) -> str: + """Get the Network ID. + + Returns: + str: The Network ID. + + """ + return self._model.network_id + + @property + def wallet_id(self) -> str: + """Get the Wallet ID. + + Returns: + str: The Wallet ID. + + """ + return self._model.wallet_id + + @property + def address_id(self) -> str: + """Get the Address ID. + + Returns: + str: The Address ID. + + """ + return self._model.address_id + + @property + def asset(self) -> Asset: + """Get the Asset. + + Returns: + Asset: The Asset. + + """ + if self._asset is None: + self._asset = Asset.from_model(self._model.crypto_amount.asset) + return self._asset + + @property + def amount(self) -> CryptoAmount: + """Get the crypto amount. + + Returns: + CryptoAmount: The crypto amount. + + """ + return CryptoAmount.from_model(self._model.crypto_amount) + + @property + def fiat_amount(self) -> FiatAmount: + """Get the fiat amount. + + Returns: + Decimal: The fiat amount. + + """ + return FiatAmount.from_model(self._model.fiat_amount) + + @property + def fiat_currency(self) -> str: + """Get the fiat currency. + + Returns: + str: The fiat currency. + + """ + return self._model.fiat_amount.currency + + @property + def buy_fee(self) -> dict: + """Get the buy fee. + + Returns: + dict: The buy fee information. + + """ + return { + "amount": self._model.fees.buy_fee.amount, + "currency": self._model.fees.buy_fee.currency, + } + + @property + def transfer_fee(self) -> CryptoAmount: + """Get the transfer fee. + + Returns: + CryptoAmount: The transfer fee. + + """ + return CryptoAmount.from_model(self._model.fees.transfer_fee) + + def execute(self) -> "FundOperation": + """Execute the fund quote to create a fund operation. + + Returns: + FundOperation: The created fund operation. + + """ + from cdp.fund_operation import FundOperation + + return FundOperation.create( + wallet_id=self.wallet_id, + address_id=self.address_id, + amount=self.amount.amount, + asset_id=self.asset.asset_id, + network_id=self.network_id, + quote=self, + ) + + def __str__(self) -> str: + """Get a string representation of the Fund Quote.""" + return ( + f"FundQuote(network_id: {self.network_id}, wallet_id: {self.wallet_id}, " + f"address_id: {self.address_id}, crypto_amount: {self.amount.amount}, " + f"crypto_asset: {self.asset.asset_id}, fiat_amount: {self.fiat_amount.amount}, " + f"fiat_currency: {self.fiat_currency}, buy_fee: {{'amount': '{self.buy_fee['amount']}'}}, " + f"transfer_fee: {{'amount': '{self.transfer_fee.amount}'}})" + ) + + def __repr__(self) -> str: + """Get a string representation of the Fund Quote.""" + return self.__str__() diff --git a/cdp/wallet.py b/cdp/wallet.py index a35a22e..46958e0 100644 --- a/cdp/wallet.py +++ b/cdp/wallet.py @@ -29,6 +29,8 @@ from cdp.client.models.wallet_list import WalletList from cdp.contract_invocation import ContractInvocation from cdp.faucet_transaction import FaucetTransaction +from cdp.fund_operation import FundOperation +from cdp.fund_quote import FundQuote from cdp.payload_signature import PayloadSignature from cdp.smart_contract import SmartContract from cdp.trade import Trade @@ -801,3 +803,41 @@ def deploy_multi_token(self, uri: str) -> SmartContract: raise ValueError("Default address does not exist") return self.default_address.deploy_multi_token(uri) + + def fund(self, amount: Number | Decimal | str, asset_id: str) -> FundOperation: + """Fund the wallet from your account on the Coinbase Platform. + + Args: + amount (Union[Number, Decimal, str]): The amount of the Asset to fund the wallet with. + asset_id (str): The ID of the Asset to fund with. For Ether, 'eth', 'gwei', and 'wei' are supported. + + Returns: + FundOperation: The created fund operation object. + + Raises: + ValueError: If the default address does not exist. + + """ + if self.default_address is None: + raise ValueError("Default address does not exist") + + return self.default_address.fund(amount, asset_id) + + def quote_fund(self, amount: Number | Decimal | str, asset_id: str) -> FundQuote: + """Get a quote for funding the wallet from your Coinbase platform account. + + Args: + amount (Union[Number, Decimal, str]): The amount to fund. + asset_id (str): The ID of the Asset to fund with. For Ether, 'eth', 'gwei', and 'wei' are supported. + + Returns: + FundQuote: The fund quote object. + + Raises: + ValueError: If the default address does not exist. + + """ + if self.default_address is None: + raise ValueError("Default address does not exist") + + return self.default_address.quote_fund(amount, asset_id) diff --git a/cdp/wallet_address.py b/cdp/wallet_address.py index 19bfb83..e8d801d 100644 --- a/cdp/wallet_address.py +++ b/cdp/wallet_address.py @@ -11,6 +11,8 @@ from cdp.client.models.address import Address as AddressModel from cdp.contract_invocation import ContractInvocation from cdp.errors import InsufficientFundsError +from cdp.fund_operation import FundOperation +from cdp.fund_quote import FundQuote from cdp.payload_signature import PayloadSignature from cdp.smart_contract import SmartContract from cdp.trade import Trade @@ -323,6 +325,48 @@ def trades(self) -> Iterator[Trade]: """ return Trade.list(wallet_id=self.wallet_id, address_id=self.address_id) + def fund(self, amount: Number | Decimal | str, asset_id: str) -> FundOperation: + """Fund the address from your account on the Coinbase Platform. + + Args: + amount (Union[Number, Decimal, str]): The amount of the Asset to fund the wallet with. + asset_id (str): The ID of the Asset to fund with. For Ether, 'eth', 'gwei', and 'wei' are supported. + + Returns: + FundOperation: The created fund operation object. + + """ + normalized_amount = Decimal(amount) + + return FundOperation.create( + address_id=self.address_id, + amount=normalized_amount, + asset_id=asset_id, + network_id=self.network_id, + wallet_id=self.wallet_id, + ) + + def quote_fund(self, amount: Number | Decimal | str, asset_id: str) -> FundQuote: + """Get a quote for funding the address from your Coinbase platform account. + + Args: + amount (Union[Number, Decimal, str]): The amount to fund. + asset_id (str): The ID of the Asset to fund with. For Ether, 'eth', 'gwei', and 'wei' are supported. + + Returns: + FundQuote: The fund quote object. + + """ + normalized_amount = Decimal(amount) + + return FundQuote.create( + address_id=self.address_id, + amount=str(normalized_amount), + asset_id=asset_id, + network_id=self.network_id, + wallet_id=self.wallet_id, + ) + def _ensure_sufficient_balance(self, amount: Decimal, asset_id: str) -> None: """Ensure the wallet address has sufficient balance. diff --git a/tests/factories/crypto_amount_factory.py b/tests/factories/crypto_amount_factory.py new file mode 100644 index 0000000..cad725a --- /dev/null +++ b/tests/factories/crypto_amount_factory.py @@ -0,0 +1,28 @@ +from decimal import Decimal + +import pytest + +from cdp.client.models.crypto_amount import CryptoAmount as CryptoAmountModel +from cdp.crypto_amount import CryptoAmount + + +@pytest.fixture +def crypto_amount_model_factory(asset_model_factory): + """Create and return a factory for creating CryptoAmountModel fixtures.""" + + def _create_crypto_amount_model(asset_id="USDC", decimals=6, amount="1"): + asset_model = asset_model_factory("base-sepolia", asset_id, decimals) + return CryptoAmountModel(amount=amount, asset=asset_model) + + return _create_crypto_amount_model + + +@pytest.fixture +def crypto_amount_factory(asset_factory): + """Create and return a factory for creating CryptoAmount fixtures.""" + + def _create_crypto_amount(asset_id="USDC", decimals=6, amount="1"): + asset = asset_factory("base-sepolia", asset_id, decimals) + return CryptoAmount(Decimal(amount), asset) + + return _create_crypto_amount diff --git a/tests/factories/fund_operation_factory.py b/tests/factories/fund_operation_factory.py new file mode 100644 index 0000000..784368f --- /dev/null +++ b/tests/factories/fund_operation_factory.py @@ -0,0 +1,51 @@ +import pytest + +from cdp.client.models.fiat_amount import FiatAmount +from cdp.client.models.fund_operation import FundOperation as FundOperationModel +from cdp.client.models.fund_operation_fees import FundOperationFees +from cdp.fund_operation import FundOperation + + +@pytest.fixture +def fund_operation_model_factory( + crypto_amount_model_factory, transaction_model_factory, fund_quote_model_factory +): + """Create and return a factory for creating FundOperationModel fixtures.""" + + def _create_fund_operation_model( + status="complete", asset_id="eth", amount="2000000000000000000", decimals=18 + ): + crypto_amount_model = crypto_amount_model_factory(asset_id, decimals, amount) + transfer_fee_crypto_amount_model = crypto_amount_model_factory( + asset_id, 18, "10000000000000000" + ) # 0.01 ETH + return FundOperationModel( + fund_operation_id="test-operation-id", + network_id="base-sepolia", + wallet_id="test-wallet-id", + address_id="test-address-id", + crypto_amount=crypto_amount_model, + fiat_amount=FiatAmount(amount="100", currency="USD"), + fees=FundOperationFees( + buy_fee=FiatAmount(amount="1", currency="USD"), + transfer_fee=transfer_fee_crypto_amount_model, + ), + status=status, + ) + + return _create_fund_operation_model + + +@pytest.fixture +def fund_operation_factory(fund_operation_model_factory): + """Create and return a factory for creating FundOperation fixtures.""" + + def _create_fund_operation( + status="complete", asset_id="eth", amount="2000000000000000000", decimals=18 + ): + fund_operation_model = fund_operation_model_factory( + status=status, asset_id=asset_id, amount=amount, decimals=decimals + ) + return FundOperation(fund_operation_model) + + return _create_fund_operation diff --git a/tests/factories/fund_quote_factory.py b/tests/factories/fund_quote_factory.py new file mode 100644 index 0000000..849da6f --- /dev/null +++ b/tests/factories/fund_quote_factory.py @@ -0,0 +1,45 @@ +import pytest + +from cdp.client.models.fiat_amount import FiatAmount +from cdp.client.models.fund_operation_fees import FundOperationFees +from cdp.client.models.fund_quote import FundQuote as FundQuoteModel +from cdp.fund_quote import FundQuote + + +@pytest.fixture +def fund_quote_model_factory(crypto_amount_model_factory): + """Create and return a factory for creating FundQuoteModel fixtures.""" + + def _create_fund_quote_model(amount="2000000000000000000", decimals=18, asset_id="eth"): + crypto_amount_model = crypto_amount_model_factory(asset_id, decimals, amount) + transfer_fee_crypto_amount_model = crypto_amount_model_factory( + asset_id, 18, "10000000000000000" + ) # 0.01 ETH + return FundQuoteModel( + fund_quote_id="test-quote-id", + network_id="base-sepolia", + wallet_id="test-wallet-id", + address_id="test-address-id", + crypto_amount=crypto_amount_model, + fiat_amount=FiatAmount(amount="100", currency="USD"), + expires_at="2024-12-31T23:59:59Z", + fees=FundOperationFees( + buy_fee=FiatAmount(amount="1", currency="USD"), + transfer_fee=transfer_fee_crypto_amount_model, + ), + ) + + return _create_fund_quote_model + + +@pytest.fixture +def fund_quote_factory(fund_quote_model_factory): + """Create and return a factory for creating FundQuote fixtures.""" + + def _create_fund_quote(amount="2000000000000000000", decimals=18, asset_id="eth"): + fund_quote_model = fund_quote_model_factory( + amount=amount, decimals=decimals, asset_id=asset_id + ) + return FundQuote(fund_quote_model) + + return _create_fund_quote diff --git a/tests/test_crypto_amount.py b/tests/test_crypto_amount.py new file mode 100644 index 0000000..ebccd95 --- /dev/null +++ b/tests/test_crypto_amount.py @@ -0,0 +1,74 @@ +from decimal import Decimal + +from cdp.crypto_amount import CryptoAmount + + +def test_crypto_amount_initialization(crypto_amount_factory): + """Test crypto amount initialization.""" + crypto_amount = crypto_amount_factory("USDC", 6, "1") + assert isinstance(crypto_amount, CryptoAmount) + assert crypto_amount.amount == Decimal("1") + assert crypto_amount.asset.asset_id == "USDC" + assert crypto_amount.asset.network_id == "base-sepolia" + assert crypto_amount.asset.decimals == 6 + + +def test_crypto_amount_from_model(crypto_amount_model_factory): + """Test crypto amount from model.""" + crypto_amount_model = crypto_amount_model_factory("USDC", 6, "1") + crypto_amount = CryptoAmount.from_model(crypto_amount_model) + assert isinstance(crypto_amount, CryptoAmount) + assert crypto_amount.amount == ( + Decimal(crypto_amount_model.amount) / Decimal(10) ** crypto_amount_model.asset.decimals + ) + assert crypto_amount.asset.asset_id == "USDC" + assert crypto_amount.asset.network_id == "base-sepolia" + assert crypto_amount.asset.decimals == 6 + + +def test_crypto_amount_from_model_and_asset_id_with_gwei(crypto_amount_model_factory): + """Test crypto amount from model with gwei.""" + crypto_amount_model = crypto_amount_model_factory("eth", 18, "1") + crypto_amount = CryptoAmount.from_model_and_asset_id(crypto_amount_model, "gwei") + assert isinstance(crypto_amount, CryptoAmount) + assert crypto_amount.amount == (Decimal(crypto_amount_model.amount) / Decimal(10) ** 9) + assert crypto_amount.asset.asset_id == "gwei" + assert crypto_amount.asset.network_id == "base-sepolia" + assert crypto_amount.asset.decimals == 9 + + +def test_crypto_amount_from_model_and_asset_id_with_wei(crypto_amount_model_factory): + """Test crypto amount from model with wei.""" + crypto_amount_model = crypto_amount_model_factory("eth", 18, "1") + crypto_amount = CryptoAmount.from_model_and_asset_id(crypto_amount_model, "wei") + assert isinstance(crypto_amount, CryptoAmount) + assert crypto_amount.amount == Decimal(crypto_amount_model.amount) + assert crypto_amount.asset.asset_id == "wei" + assert crypto_amount.asset.network_id == "base-sepolia" + assert crypto_amount.asset.decimals == 0 + + +def test_crypto_amount_to_atomic_amount(crypto_amount_factory): + """Test crypto amount to atomic amount.""" + crypto_amount = crypto_amount_factory() + assert crypto_amount.to_atomic_amount() == ( + Decimal(crypto_amount.amount) * Decimal(10) ** crypto_amount.asset.decimals + ) + + +def test_crypto_amount_str_representation(crypto_amount_factory): + """Test crypto amount string representation.""" + crypto_amount = crypto_amount_factory() + assert ( + str(crypto_amount) + == f"CryptoAmount(amount: '{int(crypto_amount.amount)}', asset_id: '{crypto_amount.asset.asset_id}')" + ) + + +def test_crypto_amount_repr(crypto_amount_factory): + """Test crypto amount repr.""" + crypto_amount = crypto_amount_factory() + assert ( + repr(crypto_amount) + == f"CryptoAmount(amount: '{int(crypto_amount.amount)}', asset_id: '{crypto_amount.asset.asset_id}')" + ) diff --git a/tests/test_fiat_amount.py b/tests/test_fiat_amount.py new file mode 100644 index 0000000..ff802b9 --- /dev/null +++ b/tests/test_fiat_amount.py @@ -0,0 +1,36 @@ +from decimal import Decimal + +from cdp.client.models.fiat_amount import FiatAmount as FiatAmountModel +from cdp.fiat_amount import FiatAmount + + +def test_fiat_amount_from_model(): + """Test converting a FiatAmount model to a FiatAmount.""" + model = FiatAmountModel(amount="100.50", currency="USD") + fiat_amount = FiatAmount.from_model(model) + + assert fiat_amount.amount == Decimal("100.50") + assert fiat_amount.currency == "USD" + + +def test_fiat_amount_properties(): + """Test FiatAmount properties.""" + fiat_amount = FiatAmount(amount=Decimal("50.25"), currency="USD") + + assert fiat_amount.amount == Decimal("50.25") + assert fiat_amount.currency == "USD" + + +def test_fiat_amount_str_representation(): + """Test string representation of FiatAmount.""" + fiat_amount = FiatAmount(amount=Decimal("75.00"), currency="USD") + expected_str = "FiatAmount(amount: '75', currency: 'USD')" + + assert str(fiat_amount) == expected_str + assert repr(fiat_amount) == expected_str + + +def test_fiat_amount_repr(): + """Test repr of FiatAmount.""" + fiat_amount = FiatAmount(amount=Decimal("75.00"), currency="USD") + assert repr(fiat_amount) == "FiatAmount(amount: '75', currency: 'USD')" diff --git a/tests/test_fund_operation.py b/tests/test_fund_operation.py new file mode 100644 index 0000000..e3b7e53 --- /dev/null +++ b/tests/test_fund_operation.py @@ -0,0 +1,206 @@ +from decimal import Decimal +from unittest.mock import Mock, call, patch + +import pytest + +from cdp.asset import Asset +from cdp.fund_operation import FundOperation + + +def test_fund_operation_initialization(fund_operation_factory): + """Test the initialization of a FundOperation object.""" + fund_operation = fund_operation_factory() + assert isinstance(fund_operation, FundOperation) + + +def test_fund_operation_properties(fund_operation_factory): + """Test the properties of a FundOperation object.""" + fund_operation = fund_operation_factory() + assert fund_operation.amount.amount == Decimal("2") + assert fund_operation.fiat_amount.amount == Decimal("100") + assert fund_operation.buy_fee["amount"] == "1" + assert fund_operation.transfer_fee.amount == Decimal("0.01") + assert fund_operation.status.value == "complete" + assert isinstance(fund_operation.asset, Asset) + + +@patch("cdp.Cdp.api_clients") +@patch("cdp.fund_operation.Asset") +def test_fund_operation_create(mock_asset, mock_api_clients, asset_factory, fund_operation_factory): + """Test the creation of a FundOperation object without a quote.""" + mock_fetch = Mock() + mock_fetch.return_value = asset_factory(asset_id="eth", decimals=18) + mock_asset.fetch = mock_fetch + + mock_primary_denomination = Mock() + mock_primary_denomination.return_value = "eth" + mock_asset.primary_denomination = mock_primary_denomination + + mock_create_fund_operation = Mock() + mock_create_fund_operation.return_value = fund_operation_factory()._model + mock_api_clients.fund.create_fund_operation = mock_create_fund_operation + + fund_operation = FundOperation.create( + wallet_id="test-wallet-id", + address_id="test-address-id", + amount=Decimal("2"), + asset_id="eth", + network_id="base-sepolia", + ) + assert isinstance(fund_operation, FundOperation) + mock_fetch.assert_called_once_with("base-sepolia", "eth") + mock_primary_denomination.assert_called_once_with("eth") + mock_create_fund_operation.assert_called_once_with( + wallet_id="test-wallet-id", + address_id="test-address-id", + create_fund_operation_request={ + "amount": "2000000000000000000", + "asset_id": "eth", + }, + ) + + +@patch("cdp.Cdp.api_clients") +@patch("cdp.fund_operation.Asset") +def test_fund_operation_create_with_quote( + mock_asset, mock_api_clients, asset_factory, fund_operation_factory, fund_quote_factory +): + """Test the creation of a FundOperation object with a quote.""" + mock_fetch = Mock() + mock_fetch.return_value = asset_factory(asset_id="eth", decimals=18) + mock_asset.fetch = mock_fetch + + mock_primary_denomination = Mock() + mock_primary_denomination.return_value = "eth" + mock_asset.primary_denomination = mock_primary_denomination + + mock_create_fund_operation = Mock() + mock_create_fund_operation.return_value = fund_operation_factory()._model + mock_api_clients.fund.create_fund_operation = mock_create_fund_operation + + fund_operation = FundOperation.create( + wallet_id="test-wallet-id", + address_id="test-address-id", + amount=Decimal("2"), + asset_id="eth", + network_id="base-sepolia", + quote=fund_quote_factory(), + ) + assert isinstance(fund_operation, FundOperation) + mock_fetch.assert_called_once_with("base-sepolia", "eth") + mock_primary_denomination.assert_called_once_with("eth") + mock_create_fund_operation.assert_called_once_with( + wallet_id="test-wallet-id", + address_id="test-address-id", + create_fund_operation_request={ + "amount": "2000000000000000000", + "asset_id": "eth", + "fund_quote_id": "test-quote-id", + }, + ) + + +@patch("cdp.Cdp.api_clients") +def test_list_fund_operations(mock_api_clients, fund_operation_factory): + """Test the listing of fund operations.""" + mock_list_fund_operations = Mock() + mock_list_fund_operations.return_value = Mock( + data=[fund_operation_factory()._model], has_more=False + ) + mock_api_clients.fund.list_fund_operations = mock_list_fund_operations + fund_operations = FundOperation.list("test-wallet-id", "0xaddressid") + assert len(list(fund_operations)) == 1 + assert all(isinstance(f, FundOperation) for f in fund_operations) + mock_list_fund_operations.assert_called_once_with( + wallet_id="test-wallet-id", address_id="0xaddressid", limit=100, page=None + ) + + +@patch("cdp.Cdp.api_clients") +def test_fund_operation_reload(mock_api_clients, fund_operation_factory): + """Test the reloading of a FundOperation object.""" + mock_reload_fund_operation = Mock() + mock_reload_fund_operation.return_value = fund_operation_factory()._model + mock_api_clients.fund.get_fund_operation = mock_reload_fund_operation + + fund_operation = fund_operation_factory() + fund_operation.reload() + mock_reload_fund_operation.assert_called_once_with( + fund_operation.wallet_id, fund_operation.address_id, fund_operation.id + ) + assert fund_operation.status.value == "complete" + + +@patch("cdp.Cdp.api_clients") +@patch("cdp.fund_operation.time.sleep") +@patch("cdp.fund_operation.time.time") +def test_fund_operation_wait(mock_time, mock_sleep, mock_api_clients, fund_operation_factory): + """Test the waiting for a FundOperation object to complete.""" + pending_fund_operation = fund_operation_factory(status="pending") + complete_fund_operation = fund_operation_factory(status="complete") + mock_get_fund_operation = Mock() + mock_api_clients.fund.get_fund_operation = mock_get_fund_operation + mock_get_fund_operation.side_effect = [ + pending_fund_operation._model, + complete_fund_operation._model, + ] + + mock_time.side_effect = [0, 0.2, 0.4] + + result = pending_fund_operation.wait(interval_seconds=0.2, timeout_seconds=1) + + assert result.status.value == "complete" + mock_get_fund_operation.assert_called_with( + pending_fund_operation.wallet_id, + pending_fund_operation.address_id, + pending_fund_operation.id, + ) + assert mock_get_fund_operation.call_count == 2 + mock_sleep.assert_has_calls([call(0.2)] * 2) + assert mock_time.call_count == 3 + + +@patch("cdp.Cdp.api_clients") +@patch("cdp.fund_operation.time.sleep") +@patch("cdp.fund_operation.time.time") +def test_wait_for_fund_operation_timeout( + mock_time, mock_sleep, mock_api_clients, fund_operation_factory +): + """Test the waiting for a FundOperation object to complete with a timeout.""" + pending_fund_operation = fund_operation_factory(status="pending") + mock_get_fund_operation = Mock(return_value=pending_fund_operation._model) + mock_api_clients.fund.get_fund_operation = mock_get_fund_operation + + mock_time.side_effect = [0, 0.5, 1.0, 1.5, 2.0, 2.5] + + with pytest.raises(TimeoutError, match="Fund operation timed out"): + pending_fund_operation.wait(interval_seconds=0.5, timeout_seconds=2) + + mock_get_fund_operation.assert_called_with( + pending_fund_operation.wallet_id, + pending_fund_operation.address_id, + pending_fund_operation.id, + ) + assert mock_get_fund_operation.call_count == 5 + mock_sleep.assert_has_calls([call(0.5)] * 4) + assert mock_time.call_count == 6 + + +@pytest.mark.parametrize("status", ["pending", "complete", "failed"]) +def test_fund_operation_str_representation(fund_operation_factory, status): + """Test the string representation of a FundOperation object.""" + fund_operation = fund_operation_factory(status=status) + expected_str = ( + f"FundOperation(id: {fund_operation.id}, network_id: {fund_operation.network_id}, " + f"wallet_id: {fund_operation.wallet_id}, address_id: {fund_operation.address_id}, " + f"amount: {fund_operation.amount}, asset_id: {fund_operation.asset.asset_id}, " + f"status: {fund_operation.status.value})" + ) + assert str(fund_operation) == expected_str + + +@pytest.mark.parametrize("status", ["pending", "complete", "failed"]) +def test_fund_operation_repr(fund_operation_factory, status): + """Test the representation of a FundOperation object.""" + fund_operation = fund_operation_factory(status=status) + assert repr(fund_operation) == str(fund_operation) diff --git a/tests/test_fund_quote.py b/tests/test_fund_quote.py new file mode 100644 index 0000000..23084ab --- /dev/null +++ b/tests/test_fund_quote.py @@ -0,0 +1,93 @@ +from decimal import Decimal +from unittest.mock import Mock, patch + +from cdp.asset import Asset +from cdp.fund_quote import FundQuote + + +def test_fund_quote_initialization(fund_quote_factory): + """Test the initialization of a FundQuote object.""" + fund_quote = fund_quote_factory() + assert isinstance(fund_quote, FundQuote) + + +def test_fund_quote_properties(fund_quote_factory): + """Test the properties of a FundQuote object.""" + fund_quote = fund_quote_factory() + assert fund_quote.amount.amount == Decimal("2") + assert fund_quote.fiat_amount.amount == Decimal("100") + assert fund_quote.buy_fee["amount"] == "1" + assert fund_quote.transfer_fee.amount == Decimal("0.01") + assert isinstance(fund_quote.asset, Asset) + + +@patch("cdp.Cdp.api_clients") +@patch("cdp.fund_quote.Asset") +def test_fund_quote_create(mock_asset, mock_api_clients, asset_factory, fund_quote_factory): + """Test the creation of a FundQuote object.""" + mock_fetch = Mock() + mock_fetch.return_value = asset_factory(asset_id="eth", decimals=18) + mock_asset.fetch = mock_fetch + + mock_primary_denomination = Mock() + mock_primary_denomination.return_value = "eth" + mock_asset.primary_denomination = mock_primary_denomination + + mock_create_fund_quote = Mock() + mock_create_fund_quote.return_value = fund_quote_factory()._model + mock_api_clients.fund.create_fund_quote = mock_create_fund_quote + + fund_quote = FundQuote.create( + wallet_id="test-wallet-id", + address_id="test-address-id", + amount=Decimal("2"), + asset_id="eth", + network_id="base-sepolia", + ) + assert isinstance(fund_quote, FundQuote) + mock_fetch.assert_called_once_with("base-sepolia", "eth") + mock_primary_denomination.assert_called_once_with("eth") + mock_create_fund_quote.assert_called_once_with( + wallet_id="test-wallet-id", + address_id="test-address-id", + create_fund_quote_request={ + "asset_id": "eth", + "amount": "2000000000000000000", + }, + ) + + +@patch("cdp.fund_operation.FundOperation") +def test_fund_quote_execute(mock_fund_operation, fund_quote_factory): + """Test the execution of a FundQuote object.""" + mock_create = Mock() + mock_fund_operation.create = mock_create + + fund_quote = fund_quote_factory() + fund_quote.execute() + mock_create.assert_called_once_with( + wallet_id=fund_quote.wallet_id, + address_id=fund_quote.address_id, + amount=fund_quote.amount.amount, + asset_id=fund_quote.asset.asset_id, + network_id=fund_quote.network_id, + quote=fund_quote, + ) + + +def test_fund_quote_str(fund_quote_factory): + """Test the string representation of a FundQuote object.""" + fund_quote = fund_quote_factory() + assert ( + str(fund_quote) + == "FundQuote(network_id: base-sepolia, wallet_id: test-wallet-id, address_id: test-address-id, crypto_amount: 2, crypto_asset: eth, fiat_amount: 100, fiat_currency: USD, buy_fee: {'amount': '1'}, transfer_fee: {'amount': '0.01'})" + ) + + +def test_fund_quote_repr(fund_quote_factory): + """Test the string representation of a FundQuote object.""" + fund_quote = fund_quote_factory() + assert ( + repr(fund_quote) + == "FundQuote(network_id: base-sepolia, wallet_id: test-wallet-id, address_id: test-address-id, crypto_amount: 2, crypto_asset: eth, fiat_amount: 100, fiat_currency: USD, buy_fee: {'amount': '1'}, transfer_fee: {'amount': '0.01'})" + ) diff --git a/tests/test_wallet.py b/tests/test_wallet.py index 292397a..751d56b 100644 --- a/tests/test_wallet.py +++ b/tests/test_wallet.py @@ -2,13 +2,14 @@ from unittest.mock import ANY, Mock, PropertyMock, call, patch import pytest -from bip_utils import Bip32Slip10Secp256k1 from eth_account import Account from cdp.client.models.create_address_request import CreateAddressRequest from cdp.client.models.create_wallet_request import CreateWalletRequest, CreateWalletRequestWallet from cdp.client.models.create_wallet_webhook_request import CreateWalletWebhookRequest from cdp.contract_invocation import ContractInvocation +from cdp.fund_operation import FundOperation +from cdp.fund_quote import FundQuote from cdp.payload_signature import PayloadSignature from cdp.smart_contract import SmartContract from cdp.trade import Trade @@ -51,60 +52,6 @@ def test_wallet_initialization_with_server_signer(wallet_factory): assert not wallet.can_sign -@patch("cdp.Cdp.use_server_signer", False) -@patch("cdp.wallet.Account") -@patch("cdp.wallet.Bip32Slip10Secp256k1") -@patch("cdp.Cdp.api_clients") -def test_wallet_addresses( - mock_api_clients, mock_bip32, mock_account, wallet_factory, address_model_factory -): - """Test Wallet addresses method.""" - wallet = wallet_factory() - mock_list_addresses = Mock() - mock_list_addresses.return_value.data = [ - address_model_factory(address_id="0x1234"), - address_model_factory(address_id="0x5678"), - ] - mock_api_clients.addresses.list_addresses = mock_list_addresses - - mock_from_key = Mock( - side_effect=[Mock(spec=Account, address="0x1234"), Mock(spec=Account, address="0x5678")] - ) - mock_account.from_key = mock_from_key - - mock_derive_path = Mock(spec=Bip32Slip10Secp256k1) - mock_bip32.DerivePath = mock_derive_path - - addresses = wallet.addresses - - assert len(addresses) == 2 - assert all(isinstance(addr, WalletAddress) for addr in addresses) - assert addresses[0].address_id == "0x1234" - assert addresses[1].address_id == "0x5678" - - -@patch("cdp.Cdp.use_server_signer", True) -@patch("cdp.Cdp.api_clients") -def test_wallet_addresses_with_server_signer( - mock_api_clients, wallet_factory, address_model_factory -): - """Test Wallet addresses method with server-signer.""" - wallet = wallet_factory() - mock_list_addresses = Mock() - mock_list_addresses.return_value.data = [ - address_model_factory(address_id="0x1234"), - address_model_factory(address_id="0x5678"), - ] - mock_api_clients.addresses.list_addresses = mock_list_addresses - - addresses = wallet.addresses - - assert len(addresses) == 2 - assert all(isinstance(addr, WalletAddress) for addr in addresses) - assert addresses[0].address_id == "0x1234" - assert addresses[1].address_id == "0x5678" - - @patch("cdp.Cdp.use_server_signer", False) @patch("cdp.Cdp.api_clients") @patch("cdp.wallet.Bip32Slip10Secp256k1") @@ -645,3 +592,69 @@ def test_create_webhook(mock_api_clients, wallet_factory, webhook_factory): # Additional assertions to check the returned webhook object assert webhook.notification_uri == notification_uri + + +@patch("cdp.Cdp.use_server_signer", True) +def test_wallet_fund(wallet_factory): + """Test the fund method of a Wallet.""" + wallet = wallet_factory() + mock_default_address = Mock(spec=WalletAddress) + mock_fund_operation = Mock(spec=FundOperation) + mock_default_address.fund.return_value = mock_fund_operation + + with patch.object( + Wallet, "default_address", new_callable=PropertyMock + ) as mock_default_address_prop: + mock_default_address_prop.return_value = mock_default_address + + fund_operation = wallet.fund(amount="1.0", asset_id="eth") + + assert isinstance(fund_operation, FundOperation) + mock_default_address.fund.assert_called_once_with("1.0", "eth") + + +@patch("cdp.Cdp.use_server_signer", True) +def test_wallet_fund_no_default_address(wallet_factory): + """Test the fund method of a Wallet with no default address.""" + wallet = wallet_factory() + + with patch.object( + Wallet, "default_address", new_callable=PropertyMock + ) as mock_default_address_prop: + mock_default_address_prop.return_value = None + + with pytest.raises(ValueError, match="Default address does not exist"): + wallet.fund(amount="1.0", asset_id="eth") + + +@patch("cdp.Cdp.use_server_signer", True) +def test_wallet_quote_fund(wallet_factory): + """Test the quote_fund method of a Wallet.""" + wallet = wallet_factory() + mock_default_address = Mock(spec=WalletAddress) + mock_fund_quote = Mock(spec=FundQuote) + mock_default_address.quote_fund.return_value = mock_fund_quote + + with patch.object( + Wallet, "default_address", new_callable=PropertyMock + ) as mock_default_address_prop: + mock_default_address_prop.return_value = mock_default_address + + fund_quote = wallet.quote_fund(amount="1.0", asset_id="eth") + + assert isinstance(fund_quote, FundQuote) + mock_default_address.quote_fund.assert_called_once_with("1.0", "eth") + + +@patch("cdp.Cdp.use_server_signer", True) +def test_wallet_quote_fund_no_default_address(wallet_factory): + """Test the quote_fund method of a Wallet with no default address.""" + wallet = wallet_factory() + + with patch.object( + Wallet, "default_address", new_callable=PropertyMock + ) as mock_default_address_prop: + mock_default_address_prop.return_value = None + + with pytest.raises(ValueError, match="Default address does not exist"): + wallet.quote_fund(amount="1.0", asset_id="eth") diff --git a/tests/test_wallet_address.py b/tests/test_wallet_address.py index a04d6f2..2c82940 100644 --- a/tests/test_wallet_address.py +++ b/tests/test_wallet_address.py @@ -10,6 +10,8 @@ from cdp.contract_invocation import ContractInvocation from cdp.errors import InsufficientFundsError +from cdp.fund_operation import FundOperation +from cdp.fund_quote import FundQuote from cdp.payload_signature import PayloadSignature from cdp.smart_contract import SmartContract from cdp.trade import Trade @@ -1020,3 +1022,81 @@ def test_ensure_sufficient_balance_sufficient_full_amount( mock_get_balance.assert_called_once_with( network_id=wallet_address.network_id, address_id=wallet_address.address_id, asset_id="eth" ) + + +@patch("cdp.wallet_address.FundOperation") +def test_fund(mock_fund_operation, wallet_address_factory): + """Test the fund method.""" + wallet_address = wallet_address_factory() + + mock_fund_operation_instance = Mock(spec=FundOperation) + mock_fund_operation.create.return_value = mock_fund_operation_instance + + fund_operation = wallet_address.fund(amount="1.0", asset_id="eth") + + assert isinstance(fund_operation, FundOperation) + mock_fund_operation.create.assert_called_once_with( + address_id=wallet_address.address_id, + amount=Decimal("1.0"), + asset_id="eth", + network_id=wallet_address.network_id, + wallet_id=wallet_address.wallet_id, + ) + + +@patch("cdp.wallet_address.FundOperation") +def test_fund_api_error(mock_fund_operation, wallet_address_factory): + """Test the fund method raises an error when the API call fails.""" + wallet_address = wallet_address_factory() + + mock_fund_operation.create.side_effect = Exception("API Error") + + with pytest.raises(Exception, match="API Error"): + wallet_address.fund(amount="1.0", asset_id="eth") + + mock_fund_operation.create.assert_called_once_with( + address_id=wallet_address.address_id, + amount=Decimal("1.0"), + asset_id="eth", + network_id=wallet_address.network_id, + wallet_id=wallet_address.wallet_id, + ) + + +@patch("cdp.wallet_address.FundQuote") +def test_quote_fund(mock_fund_quote, wallet_address_factory): + """Test the quote_fund method.""" + wallet_address = wallet_address_factory() + + mock_fund_quote_instance = Mock(spec=FundQuote) + mock_fund_quote.create.return_value = mock_fund_quote_instance + + fund_quote = wallet_address.quote_fund(amount="1.0", asset_id="eth") + + assert isinstance(fund_quote, FundQuote) + mock_fund_quote.create.assert_called_once_with( + address_id=wallet_address.address_id, + amount="1.0", + asset_id="eth", + network_id=wallet_address.network_id, + wallet_id=wallet_address.wallet_id, + ) + + +@patch("cdp.wallet_address.FundQuote") +def test_quote_fund_api_error(mock_fund_quote, wallet_address_factory): + """Test the quote_fund method raises an error when the API call fails.""" + wallet_address = wallet_address_factory() + + mock_fund_quote.create.side_effect = Exception("API Error") + + with pytest.raises(Exception, match="API Error"): + wallet_address.quote_fund(amount="1.0", asset_id="eth") + + mock_fund_quote.create.assert_called_once_with( + address_id=wallet_address.address_id, + amount="1.0", + asset_id="eth", + network_id=wallet_address.network_id, + wallet_id=wallet_address.wallet_id, + )