From 4758864f47ecbce51b52563fb14b79529077a6b0 Mon Sep 17 00:00:00 2001 From: Jair Rosa Date: Fri, 10 Jan 2025 18:56:40 -0300 Subject: [PATCH] Fix lint issues --- .../cdp_agentkit_core/actions/__init__.py | 4 +- .../cdp_agentkit_core/actions/webhook.py | 48 +++++++++++-------- .../tests/actions/test_webhook.py | 44 +++++++++-------- 3 files changed, 52 insertions(+), 44 deletions(-) diff --git a/cdp-agentkit-core/cdp_agentkit_core/actions/__init__.py b/cdp-agentkit-core/cdp_agentkit_core/actions/__init__.py index 44ce98f34..a77174ce1 100644 --- a/cdp-agentkit-core/cdp_agentkit_core/actions/__init__.py +++ b/cdp-agentkit-core/cdp_agentkit_core/actions/__init__.py @@ -8,10 +8,10 @@ from cdp_agentkit_core.actions.request_faucet_funds import RequestFaucetFundsAction from cdp_agentkit_core.actions.trade import TradeAction from cdp_agentkit_core.actions.transfer import TransferAction +from cdp_agentkit_core.actions.webhook import CreateWebhookAction from cdp_agentkit_core.actions.wow.buy_token import WowBuyTokenAction from cdp_agentkit_core.actions.wow.create_token import WowCreateTokenAction from cdp_agentkit_core.actions.wow.sell_token import WowSellTokenAction -from cdp_agentkit_core.actions.webhook import CreateWebhookAction # WARNING: All new CdpAction subclasses must be imported above, otherwise they will not be discovered @@ -29,6 +29,7 @@ def get_all_cdp_actions() -> list[type[CdpAction]]: __all__ = [ "CDP_ACTIONS", "CdpAction", + "CreateWebhookAction", "DeployNftAction", "DeployTokenAction", "GetBalanceAction", @@ -41,5 +42,4 @@ def get_all_cdp_actions() -> list[type[CdpAction]]: "WowBuyTokenAction", "WowCreateTokenAction", "WowSellTokenAction", - "CreateWebhookAction", ] diff --git a/cdp-agentkit-core/cdp_agentkit_core/actions/webhook.py b/cdp-agentkit-core/cdp_agentkit_core/actions/webhook.py index 6f59d5d47..65c82846e 100644 --- a/cdp-agentkit-core/cdp_agentkit_core/actions/webhook.py +++ b/cdp-agentkit-core/cdp_agentkit_core/actions/webhook.py @@ -1,16 +1,16 @@ from enum import Enum -from typing import Any, Dict, List, Optional, Union +from typing import Any -from cdp import Wallet, Webhook -from pydantic import BaseModel, Field, HttpUrl, model_validator, field_validator - -from cdp_agentkit_core.actions import CdpAction +from cdp import Webhook from cdp.client.models.webhook import WebhookEventTypeFilter -from cdp.client.models.webhook_wallet_activity_filter import WebhookWalletActivityFilter from cdp.client.models.webhook_smart_contract_event_filter import WebhookSmartContractEventFilter +from cdp.client.models.webhook_wallet_activity_filter import WebhookWalletActivityFilter +from pydantic import BaseModel, Field, HttpUrl, field_validator, model_validator + +from cdp_agentkit_core.actions import CdpAction CREATE_WEBHOOK_PROMPT = """ -Create a new webhook to receive real-time updates for on-chain events. +Create a new webhook to receive real-time updates for on-chain events. Supports monitoring wallet activity or smart contract events by specifying: - Callback URL for receiving events - Event type (wallet_activity, smart_contract_event_activity, erc20_transfer or erc721_transfer) @@ -24,6 +24,7 @@ class WebhookEventType(str, Enum): """Valid webhook event types.""" + WALLET_ACTIVITY = "wallet_activity" SMART_CONTRACT_EVENT_ACTIVITY = "smart_contract_event_activity" ERC20_TRANSFER = "erc20_transfer" @@ -31,14 +32,16 @@ class WebhookEventType(str, Enum): class WebhookNetworks(str, Enum): """Networks available for creating webhooks.""" + BASE_MAINNET = "base-mainnet" BASE_SEPOLIA = "base-sepolia" class EventFilter(BaseModel): """Schema for event filters.""" - from_address: Optional[str] = Field(None, description="Sender address for token transfers") - to_address: Optional[str] = Field(None, description="Recipient address for token transfers") - contract_address: Optional[str] = Field(None, description="Contract address for token transfers") + + from_address: str | None = Field(None, description="Sender address for token transfers") + to_address: str | None = Field(None, description="Recipient address for token transfers") + contract_address: str | None = Field(None, description="Contract address for token transfers") @model_validator(mode='after') def validate_at_least_one_filter(self) -> 'EventFilter': @@ -49,11 +52,12 @@ def validate_at_least_one_filter(self) -> 'EventFilter': class EventTypeFilter(BaseModel): """Schema for event type filter.""" - addresses: Optional[List[str]] = Field(None, description="List of wallet or contract addresses to monitor") + + addresses: list[str] | None = Field(None, description="List of wallet or contract addresses to monitor") @field_validator('addresses') @classmethod - def validate_addresses_not_empty(cls, v: Optional[List[str]]) -> Optional[List[str]]: + def validate_addresses_not_empty(cls, v: list[str] | None) -> list[str] | None: """Ensure addresses list is not empty when provided.""" if v is not None and len(v) == 0: raise ValueError("addresses must contain at least one value when provided") @@ -61,10 +65,11 @@ def validate_addresses_not_empty(cls, v: Optional[List[str]]) -> Optional[List[s class WebhookInput(BaseModel): """Input schema for create webhook action.""" + notification_uri: HttpUrl = Field(..., description="The callback URL where webhook events will be sent") event_type: WebhookEventType - event_type_filter: Optional[EventTypeFilter] = None - event_filters: Optional[List[EventFilter]] = None + event_type_filter: EventTypeFilter | None = None + event_filters: list[EventFilter] | None = None network_id: WebhookNetworks @model_validator(mode='after') @@ -95,13 +100,13 @@ def validate_filters(self) -> 'WebhookInput': return self def create_webhook( - notification_uri: Union[str, HttpUrl], + notification_uri: str | HttpUrl, event_type: str, network_id: str, - event_type_filter: Optional[Dict[str, Any]] = None, - event_filters: Optional[List[Dict[str, Any]]] = None, + event_type_filter: dict[str, Any] | None = None, + event_filters: list[dict[str, Any]] | None = None, ) -> str: - """Creates a new webhook for monitoring on-chain events. + """Create a new webhook for monitoring on-chain events. Args: notification_uri: The callback URL where webhook events will be sent @@ -112,8 +117,8 @@ def create_webhook( Returns: str: Details of the created webhook - """ + """ print(f"notification_uri: {notification_uri}") print(f"event_type_filter: {event_type_filter}") print(f"event_filters: {event_filters}") @@ -155,12 +160,13 @@ def create_webhook( print(f"webhook_options: {webhook_options}") webhook = Webhook.create(**webhook_options) return f"The webhook was successfully created: {webhook}\n\n" - + except Exception as error: - return f"Error: {str(error)}" + return f"Error: {error!s}" class CreateWebhookAction(CdpAction): """Create webhook action.""" + name: str = "create_webhook" description: str = CREATE_WEBHOOK_PROMPT args_schema: type[BaseModel] = WebhookInput diff --git a/cdp-agentkit-core/tests/actions/test_webhook.py b/cdp-agentkit-core/tests/actions/test_webhook.py index fb46b78fe..4337e2290 100644 --- a/cdp-agentkit-core/tests/actions/test_webhook.py +++ b/cdp-agentkit-core/tests/actions/test_webhook.py @@ -1,5 +1,6 @@ +from unittest.mock import Mock, patch + import pytest -from unittest.mock import patch, Mock from cdp_agentkit_core.actions.webhook import ( WebhookInput, @@ -15,13 +16,14 @@ @pytest.fixture def mock_webhook(): + """Provide a mocked Webhook instance for testing.""" with patch('cdp_agentkit_core.actions.webhook.Webhook') as mock: mock_instance = Mock() mock.create.return_value = mock_instance yield mock def test_webhook_input_valid_parsing(): - """Test successful parsing of valid webhook inputs""" + """Test successful parsing of valid webhook inputs.""" # Test wallet activity webhook input valid_input = { "notification_uri": MOCK_URL, @@ -31,7 +33,7 @@ def test_webhook_input_valid_parsing(): }, "network_id": MOCK_NETWORK } - + result = WebhookInput.model_validate(valid_input) assert str(result.notification_uri) == MOCK_URL assert result.event_type == MOCK_EVENT_TYPE @@ -47,20 +49,20 @@ def test_webhook_input_valid_parsing(): }], "network_id": MOCK_NETWORK } - + result = WebhookInput.model_validate(another_valid_input) assert str(result.notification_uri) == MOCK_URL assert result.event_type == "erc721_transfer" assert result.event_filters[0].from_address == MOCK_ADDRESS def test_webhook_input_invalid_parsing(): - """Test parsing failure for invalid webhook input""" + """Test parsing failure for invalid webhook input.""" empty_input = {} with pytest.raises(ValueError): WebhookInput.model_validate(empty_input) def test_create_wallet_activity_webhook(mock_webhook): - """Test creating wallet activity webhook""" + """Test creating wallet activity webhook.""" args = { "notification_uri": MOCK_URL, "event_type": MOCK_EVENT_TYPE, @@ -69,14 +71,14 @@ def test_create_wallet_activity_webhook(mock_webhook): }, "network_id": MOCK_NETWORK } - + response = create_webhook(**args) - + assert mock_webhook.create.call_count == 1 assert SUCCESS_MESSAGE in response def test_create_smart_contract_activity_webhook(mock_webhook): - """Test creating smart contract activity webhook""" + """Test creating smart contract activity webhook.""" args = { "notification_uri": MOCK_URL, "event_type": "smart_contract_event_activity", @@ -85,14 +87,14 @@ def test_create_smart_contract_activity_webhook(mock_webhook): }, "network_id": MOCK_NETWORK } - + response = create_webhook(**args) - + assert mock_webhook.create.call_count == 1 assert SUCCESS_MESSAGE in response def test_create_erc20_transfer_webhook(mock_webhook): - """Test creating ERC20 transfer webhook""" + """Test creating ERC20 transfer webhook.""" args = { "notification_uri": MOCK_URL, "event_type": "erc20_transfer", @@ -104,14 +106,14 @@ def test_create_erc20_transfer_webhook(mock_webhook): }], "network_id": MOCK_NETWORK } - + response = create_webhook(**args) - + assert mock_webhook.create.call_count == 1 assert SUCCESS_MESSAGE in response def test_create_erc721_transfer_webhook(mock_webhook): - """Test creating ERC721 transfer webhook""" + """Test creating ERC721 transfer webhook.""" args = { "notification_uri": MOCK_URL, "event_type": "erc721_transfer", @@ -120,17 +122,17 @@ def test_create_erc721_transfer_webhook(mock_webhook): }], "network_id": MOCK_NETWORK } - + response = create_webhook(**args) - + assert mock_webhook.create.call_count == 1 assert SUCCESS_MESSAGE in response def test_create_webhook_error_handling(mock_webhook): - """Test error handling when creating webhook fails""" + """Test error handling when creating webhook fails.""" error_msg = "Failed to create webhook" mock_webhook.create.side_effect = Exception(error_msg) - + args = { "notification_uri": MOCK_URL, "event_type": MOCK_EVENT_TYPE, @@ -139,8 +141,8 @@ def test_create_webhook_error_handling(mock_webhook): }, "network_id": MOCK_NETWORK } - + response = create_webhook(**args) - + assert mock_webhook.create.call_count == 1 assert f"Error: {error_msg}" in response