This is a test email from the Python CCAI client.
", sender_email="noreply@cloudcontactai.com", diff --git a/src/ccai_python/email_service.py b/src/ccai_python/email_service.py index 85a3412..d2aaa64 100644 --- a/src/ccai_python/email_service.py +++ b/src/ccai_python/email_service.py @@ -7,7 +7,7 @@ """ from typing import Any, Dict, List, Optional, Callable -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator import requests @@ -17,6 +17,15 @@ class EmailAccount(BaseModel): last_name: str = Field(..., description="Recipient's last name") email: str = Field(..., description="Recipient's email address") phone: str = Field(default="", description="Phone number (required by base Account type)") + custom_account_id: Optional[str] = Field( + default=None, + description="External ID to link this account to an external system. Sent as 'customAccountId'." + ) + data: Optional[Dict[str, str]] = Field( + default=None, + description="Additional key-value pairs for variable substitution in email templates. " + "Sent to the API as 'data'." + ) class EmailCampaign(BaseModel): @@ -24,6 +33,7 @@ class EmailCampaign(BaseModel): subject: str = Field(..., description="Email subject") title: str = Field(..., description="Campaign title") message: str = Field(..., description="HTML message content") + text_content: Optional[str] = Field(default=None, description="Plain-text version of the email body") sender_email: str = Field(..., description="Sender's email address") reply_email: str = Field(..., description="Reply-to email address") sender_name: str = Field(..., description="Sender's name") @@ -51,6 +61,21 @@ class EmailResponse(BaseModel): campaign_id: Optional[str] = Field(default=None, description="Campaign ID") messages_sent: Optional[int] = Field(default=None, description="Number of messages sent") timestamp: Optional[str] = Field(default=None, description="Timestamp") + message: Optional[str] = Field(default=None, description="Human-readable message from the API") + response_id: Optional[str] = Field(default=None, description="Unique response identifier") + + @model_validator(mode="before") + @classmethod + def normalize_fields(cls, values): + if "responseId" in values and "response_id" not in values: + values["response_id"] = values.pop("responseId") + if "campaignId" in values and "campaign_id" not in values: + values["campaign_id"] = values.pop("campaignId") + if "messagesSent" in values and "messages_sent" not in values: + values["messages_sent"] = values.pop("messagesSent") + return values + + model_config = {"extra": "allow"} class EmailOptions(BaseModel): @@ -65,11 +90,62 @@ class Config: class Email: """Email service for sending email campaigns""" - + def __init__(self, ccai): self.ccai = ccai - self.base_url = "https://email-campaigns-test-cloudcontactai.allcode.com/api/v1" - + + def _build_payload(self, campaign: "EmailCampaign") -> dict: + """Build camelCase API payload from the campaign model.""" + def account_to_dict(acc: "EmailAccount") -> dict: + d: dict = { + "firstName": acc.first_name, + "lastName": acc.last_name, + "email": acc.email, + } + if acc.phone: + d["phone"] = acc.phone + if acc.custom_account_id is not None: + d["customAccountId"] = acc.custom_account_id + if acc.data is not None: + d["data"] = acc.data + return d + + payload: dict = { + "subject": campaign.subject, + "title": campaign.title, + "message": campaign.message, + "senderEmail": campaign.sender_email, + "replyEmail": campaign.reply_email, + "senderName": campaign.sender_name, + "accounts": [account_to_dict(a) for a in campaign.accounts], + "campaignType": campaign.campaign_type, + "addToList": campaign.add_to_list, + "contactInput": campaign.contact_input, + "fromType": campaign.from_type, + "senders": campaign.senders, + } + if campaign.scheduled_timestamp is not None: + payload["scheduledTimestamp"] = campaign.scheduled_timestamp + if campaign.scheduled_timezone is not None: + payload["scheduledTimezone"] = campaign.scheduled_timezone + if campaign.selected_list is not None: + payload["selectedList"] = campaign.selected_list + if campaign.list_id is not None: + payload["listId"] = campaign.list_id + if campaign.replace_contacts is not None: + payload["replaceContacts"] = campaign.replace_contacts + if campaign.email_template_id is not None: + payload["emailTemplateId"] = campaign.email_template_id + if campaign.flux_id is not None: + payload["fluxId"] = campaign.flux_id + if campaign.editor is not None: + payload["editor"] = campaign.editor + if campaign.file_key is not None: + payload["fileKey"] = campaign.file_key + if campaign.text_content is not None: + payload["textContent"] = campaign.text_content + return payload + def make_email_request( self, method: str, @@ -77,31 +153,16 @@ def make_email_request( data: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """Make an authenticated API request to the email campaigns API with required headers""" - url = f"{self.base_url}{endpoint}" - - # Print curl command for debugging (matching Node.js behavior) - curl_cmd = f"curl -X {method.upper()} \"{url}\" \\" - curl_cmd += f"\n -H \"Authorization: Bearer {self.ccai.api_key}\" \\" - curl_cmd += f"\n -H \"Content-Type: application/json\" \\" - curl_cmd += f"\n -H \"Accept: */*\" \\" - curl_cmd += f"\n -H \"clientId: {self.ccai.client_id}\" \\" - curl_cmd += f"\n -H \"accountId: 1223\"" - if data: - import json - curl_cmd += f" \\\n -d '{json.dumps(data)}'" - - print("\nš” Equivalent curl command:") - print(curl_cmd) - print("") - + url = f"{self.ccai.email_base_url}{endpoint}" + headers = { "Authorization": f"Bearer {self.ccai.api_key}", "Content-Type": "application/json", "Accept": "*/*", - "clientId": str(self.ccai.client_id), - "accountId": "1223" + "AccountId": str(self.ccai.client_id), + "ClientId": str(self.ccai.client_id) } - + try: response = requests.request( method=method.upper(), @@ -123,7 +184,30 @@ def make_email_request( raise except requests.RequestException as e: raise Exception(f"Network error: {str(e)}") - + + def send( + self, + accounts: List[EmailAccount], + subject: str, + message: str, + sender_email: str, + reply_email: str, + sender_name: str, + title: Optional[str] = None, + options: Optional[EmailOptions] = None + ) -> EmailResponse: + """Send an email campaign to multiple recipients""" + campaign = EmailCampaign( + subject=subject, + title=title or subject, + message=message, + sender_email=sender_email, + reply_email=reply_email, + sender_name=sender_name, + accounts=accounts + ) + return self.send_campaign(campaign, options) + def send_campaign( self, campaign: EmailCampaign, @@ -132,7 +216,7 @@ def send_campaign( """Send an email campaign to multiple recipients""" if not campaign.accounts: raise ValueError("At least one account is required") - + if not campaign.subject: raise ValueError("Subject is required") if not campaign.title: @@ -145,29 +229,29 @@ def send_campaign( raise ValueError("Reply email is required") if not campaign.sender_name: raise ValueError("Sender name is required") - + if options and options.on_progress: options.on_progress("Preparing to send email campaign") - + try: if options and options.on_progress: options.on_progress("Sending email campaign") - + response = self.make_email_request( "POST", "/campaigns", - campaign.dict() + self._build_payload(campaign) ) - + if options and options.on_progress: options.on_progress("Email campaign sent successfully") - + return EmailResponse(**response) except Exception as error: if options and options.on_progress: options.on_progress("Email campaign sending failed") raise error - + def send_single( self, first_name: str, @@ -175,10 +259,11 @@ def send_single( email: str, subject: str, message: str, - sender_email: str, - reply_email: str, - sender_name: str, - title: str, + text_content: Optional[str] = None, + sender_email: str = 'noreply@cloudcontactai.com', + reply_email: str = 'noreply@cloudcontactai.com', + sender_name: str = 'CloudContactAI', + title: Optional[str] = None, options: Optional[EmailOptions] = None ) -> EmailResponse: """Send a single email to one recipient""" @@ -188,15 +273,16 @@ def send_single( email=email, phone="" ) - + campaign = EmailCampaign( subject=subject, - title=title, + title=title or subject, message=message, + text_content=text_content, sender_email=sender_email, reply_email=reply_email, sender_name=sender_name, accounts=[account] ) - - return self.send_campaign(campaign, options) \ No newline at end of file + + return self.send_campaign(campaign, options) diff --git a/src/ccai_python/examples/webhook_examples.py b/src/ccai_python/examples/webhook_examples.py index 430d9df..b57e6e6 100644 --- a/src/ccai_python/examples/webhook_examples.py +++ b/src/ccai_python/examples/webhook_examples.py @@ -19,16 +19,29 @@ ) def register_webhook_example(): - """Example: Register a new webhook""" + """Example: Register a new webhook with auto-generated secret""" try: + # Example 1: Auto-generated secret (server will generate and return it) config = WebhookConfig( url="https://your-domain.com/api/ccai-webhook", - events=[WebhookEventType.MESSAGE_SENT, WebhookEventType.MESSAGE_RECEIVED], - secret="your-webhook-secret" + events=[WebhookEventType.MESSAGE_SENT, WebhookEventType.MESSAGE_RECEIVED] + # secret not provided - server will auto-generate ) - + response = ccai.webhook.register(config) - print("Webhook registered successfully:", response) + print("Webhook registered successfully with auto-generated secret:", response) + print(f"Auto-generated Secret Key: {getattr(response, 'secretKey', 'N/A')}") + + # Example 2: Custom secret + config_custom = WebhookConfig( + url="https://your-domain.com/api/ccai-webhook-v2", + events=[WebhookEventType.MESSAGE_SENT, WebhookEventType.MESSAGE_RECEIVED], + secret="my-custom-secret-key" + ) + + response_custom = ccai.webhook.register(config_custom) + print("Webhook registered with custom secret:", response_custom) + return response.id except Exception as error: print("Error registering webhook:", error) diff --git a/src/ccai_python/mms_send.py b/src/ccai_python/mms_send.py index 8ad0c08..2db1bd7 100644 --- a/src/ccai_python/mms_send.py +++ b/src/ccai_python/mms_send.py @@ -1,20 +1,21 @@ import os from dotenv import load_dotenv -from ccai_python import CCAI, Account +from ccai_python import CCAI +from ccai_python.sms.sms import Account # Load environment variables load_dotenv() ccai = CCAI( - client_id=os.getenv('CCAI_CLIENT_ID'), - api_key=os.getenv('CCAI_API_KEY'), + client_id=os.environ['CCAI_CLIENT_ID'], + api_key=os.environ['CCAI_API_KEY'], use_test=True ) account = Account( - first_name=os.getenv('TEST_FIRST_NAME'), - last_name=os.getenv('TEST_LAST_NAME'), - phone=os.getenv('TEST_PHONE_NUMBER') + first_name=os.environ['TEST_FIRST_NAME'], + last_name=os.environ['TEST_LAST_NAME'], + phone=os.environ['TEST_PHONE_NUMBER'] ) response = ccai.mms.send_with_image( diff --git a/src/ccai_python/sms/mms.py b/src/ccai_python/sms/mms.py index 3974fe1..a954616 100644 --- a/src/ccai_python/sms/mms.py +++ b/src/ccai_python/sms/mms.py @@ -39,12 +39,12 @@ def file_base_url(self) -> str: ... def request( - self, - method: str, - endpoint: str, - data: Optional[Dict[str, Any]] = None, - timeout: Optional[int] = None - ) -> Dict[str, Any]: + self, + method: str, + endpoint: str, + data: Optional[Union[Dict[str, Any], List[Any]]] = None, + timeout: int = 30 + ) -> Any: ... @@ -201,6 +201,7 @@ def send( accounts: List[Union[Account, Dict[str, str]]], message: str, title: str, + sender_phone: Optional[str] = None, options: Optional[SMSOptions] = None, force_new_campaign: bool = True ) -> SMSResponse: @@ -270,14 +271,18 @@ def send( endpoint = f"/clients/{self._ccai.client_id}/campaigns/direct" # Convert Account objects to dictionaries with camelCase keys for API compatibility - accounts_data = [ - { + accounts_data = [] + for account in normalized_accounts: + acc: Dict[str, Any] = { "firstName": account.first_name, "lastName": account.last_name, - "phone": account.phone + "phone": account.phone, } - for account in normalized_accounts - ] + if account.data: + acc["data"] = account.data + if account.message_data: + acc["messageData"] = account.message_data + accounts_data.append(acc) campaign_data = { "pictureFileKey": picture_file_key, @@ -285,6 +290,8 @@ def send( "message": message, "title": title } + if sender_phone: + campaign_data["senderPhone"] = sender_phone try: # Notify progress if callback provided @@ -332,6 +339,8 @@ def send_single( phone: str, message: str, title: str, + custom_data: Optional[str] = None, + sender_phone: Optional[str] = None, options: Optional[SMSOptions] = None, force_new_campaign: bool = True ) -> SMSResponse: @@ -345,6 +354,8 @@ def send_single( phone: Recipient's phone number (E.164 format) message: Message content (can include ${first_name} and ${last_name} variables) title: Campaign title + custom_data: Optional arbitrary string forwarded to your webhook handler (sent as messageData) + sender_phone: Optional sender phone number options: Optional settings for the MMS send operation force_new_campaign: Whether to force a new campaign (default: True) @@ -354,7 +365,8 @@ def send_single( account = Account( first_name=first_name, last_name=last_name, - phone=phone + phone=phone, + message_data=custom_data ) return self.send( @@ -362,6 +374,7 @@ def send_single( accounts=[account], message=message, title=title, + sender_phone=sender_phone, options=options, force_new_campaign=force_new_campaign ) @@ -373,6 +386,7 @@ def send_with_image( accounts: List[Union[Account, Dict[str, str]]], message: str, title: str, + sender_phone: Optional[str] = None, options: Optional[SMSOptions] = None, force_new_campaign: bool = True ) -> SMSResponse: @@ -414,6 +428,7 @@ def send_with_image( accounts=accounts, message=message, title=title, + sender_phone=sender_phone, options=options, force_new_campaign=force_new_campaign ) @@ -447,6 +462,7 @@ def send_with_image( accounts=accounts, message=message, title=title, + sender_phone=sender_phone, options=options, force_new_campaign=force_new_campaign ) diff --git a/src/ccai_python/sms/sms.py b/src/ccai_python/sms/sms.py index 54c38fe..38e3252 100644 --- a/src/ccai_python/sms/sms.py +++ b/src/ccai_python/sms/sms.py @@ -15,6 +15,16 @@ class Account(BaseModel): first_name: str = Field(..., description="Recipient's first name") last_name: str = Field(..., description="Recipient's last name") phone: str = Field(..., description="Recipient's phone number in E.164 format") + data: Optional[Dict[str, str]] = Field( + default=None, + description="Additional key-value pairs for variable substitution in message templates. " + "Use ${key} in your message. Sent to the API as 'data'." + ) + message_data: Optional[str] = Field( + default=None, + description="Arbitrary string forwarded as-is to your webhook handler. " + "Not used in the message body. Sent to the API as 'messageData'." + ) class SMSCampaign(BaseModel): @@ -31,11 +41,20 @@ class SMSResponse(BaseModel): campaign_id: Optional[str] = Field(None, description="Campaign ID") messages_sent: Optional[int] = Field(None, description="Number of messages sent") timestamp: Optional[str] = Field(None, description="Timestamp of the operation") + message: Optional[str] = Field(None, description="Human-readable message from the API") + response_id: Optional[str] = Field(None, description="Unique response identifier") @model_validator(mode="before") - def coerce_id(cls, values): + def normalize_fields(cls, values): if "id" in values and isinstance(values["id"], int): values["id"] = str(values["id"]) + # Handle camelCase keys from real API responses + if "campaignId" in values and "campaign_id" not in values: + values["campaign_id"] = values.pop("campaignId") + if "messagesSent" in values and "messages_sent" not in values: + values["messages_sent"] = values.pop("messagesSent") + if "responseId" in values and "response_id" not in values: + values["response_id"] = values.pop("responseId") return values model_config = { @@ -45,11 +64,11 @@ def coerce_id(cls, values): class SMSOptions(BaseModel): """Options for SMS operations""" - timeout: Optional[int] = Field(None, description="Request timeout in seconds") - retries: Optional[int] = Field(None, description="Number of retry attempts") - on_progress: Optional[Callable[[str], None]] = Field( - None, description="Callback for tracking progress" - ) + model_config = {"arbitrary_types_allowed": True} + + timeout: Optional[int] = Field(default=None, description="Request timeout in seconds") + retries: Optional[int] = Field(default=None, description="Number of retry attempts") + on_progress: Optional[Callable[[str], None]] = Field(default=None, description="Callback for tracking progress") class CCAIProtocol(Protocol): @@ -62,9 +81,9 @@ def request( self, method: str, endpoint: str, - data: Optional[Dict[str, Any]] = None, - timeout: Optional[int] = None - ) -> Dict[str, Any]: + data: Optional[Union[Dict[str, Any], List[Any]]] = None, + timeout: int = 30 + ) -> Any: ... @@ -79,6 +98,7 @@ def send( accounts: List[Union[Account, Dict[str, str]]], message: str, title: str, + sender_phone: Optional[str] = None, options: Optional[SMSOptions] = None ) -> SMSResponse: if not accounts: @@ -111,25 +131,32 @@ def send( options.on_progress("Preparing to send SMS") endpoint = f"/clients/{self._ccai.client_id}/campaigns/direct" - accounts_data = [ - { + accounts_data = [] + for acct in normalized_accounts: + account_dict: Dict[str, Any] = { "firstName": acct.first_name, "lastName": acct.last_name, - "phone": acct.phone - } for acct in normalized_accounts - ] - - payload = { + "phone": acct.phone, + } + if acct.data: + account_dict["data"] = acct.data + if acct.message_data: + account_dict["messageData"] = acct.message_data + accounts_data.append(account_dict) + + payload: Dict[str, Any] = { "accounts": accounts_data, "message": message, "title": title } + if sender_phone: + payload["senderPhone"] = sender_phone try: if options and options.on_progress: options.on_progress("Sending SMS") - timeout = options.timeout if options else None + timeout = (options.timeout if options and options.timeout else None) or 30 response_data = self._ccai.request( method="post", endpoint=endpoint, @@ -154,12 +181,15 @@ def send_single( phone: str, message: str, title: str, + custom_data: Optional[str] = None, + sender_phone: Optional[str] = None, options: Optional[SMSOptions] = None ) -> SMSResponse: account = Account( first_name=first_name, last_name=last_name, - phone=phone + phone=phone, + message_data=custom_data ) - return self.send([account], message, title, options) + return self.send([account], message, title, sender_phone, options) diff --git a/src/ccai_python/sms_send.py b/src/ccai_python/sms_send.py index 8e68c18..21659f7 100644 --- a/src/ccai_python/sms_send.py +++ b/src/ccai_python/sms_send.py @@ -1,20 +1,21 @@ import os from dotenv import load_dotenv -from ccai_python import CCAI, Account +from ccai_python import CCAI +from ccai_python.sms.sms import Account # Load environment variables load_dotenv() ccai = CCAI( - client_id=os.getenv('CCAI_CLIENT_ID'), - api_key=os.getenv('CCAI_API_KEY'), + client_id=os.environ['CCAI_CLIENT_ID'], + api_key=os.environ['CCAI_API_KEY'], use_test=True ) account = Account( - first_name=os.getenv('TEST_FIRST_NAME'), - last_name=os.getenv('TEST_LAST_NAME'), - phone=os.getenv('TEST_PHONE_NUMBER') + first_name=os.environ['TEST_FIRST_NAME'], + last_name=os.environ['TEST_LAST_NAME'], + phone=os.environ['TEST_PHONE_NUMBER'] ) response = ccai.sms.send( diff --git a/src/ccai_python/webhook.py b/src/ccai_python/webhook.py index 3b66361..5cd96d4 100644 --- a/src/ccai_python/webhook.py +++ b/src/ccai_python/webhook.py @@ -6,7 +6,10 @@ :copyright: 2025 CloudContactAI LLC """ -from typing import Any, Dict, List, Optional +import hmac +import hashlib +import base64 +from typing import Any, Callable, Dict, List, Optional from pydantic import BaseModel, Field from enum import Enum @@ -14,35 +17,20 @@ class WebhookEventType(str, Enum): """Event types supported by CloudContactAI webhooks""" MESSAGE_SENT = "message.sent" + MESSAGE_INCOMING = "message.incoming" MESSAGE_RECEIVED = "message.received" + MESSAGE_EXCLUDED = "message.excluded" + MESSAGE_ERROR_CARRIER = "message.error.carrier" + MESSAGE_ERROR_CLOUDCONTACT = "message.error.cloudcontact" -class WebhookCampaign(BaseModel): - """Campaign information included in webhook events""" - id: int = Field(..., description="Campaign ID") - title: str = Field(..., description="Campaign title") - message: str = Field(..., description="Campaign message") - sender_phone: str = Field(..., description="Sender phone number") - created_at: str = Field(..., description="Creation timestamp") - run_at: str = Field(..., description="Run timestamp") +class WebhookEvent(BaseModel): + """Webhook event sent by CloudContactAI server""" + model_config = {"extra": "allow"} - -class WebhookEventBase(BaseModel): - """Base interface for all webhook events""" - campaign: WebhookCampaign = Field(..., description="Campaign information") - from_: str = Field(..., alias="from", description="Sender") - to: str = Field(..., description="Recipient") - message: str = Field(..., description="Message content") - - -class MessageSentEvent(WebhookEventBase): - """Message Sent (Outbound) webhook event""" - type: WebhookEventType = Field(default=WebhookEventType.MESSAGE_SENT, description="Event type") - - -class MessageReceivedEvent(WebhookEventBase): - """Message Received (Inbound) webhook event""" - type: WebhookEventType = Field(default=WebhookEventType.MESSAGE_RECEIVED, description="Event type") + event_type: str = Field(..., alias="eventType", description="Type of the event") + data: Dict[str, Any] = Field(..., description="Event-specific data") + event_hash: str = Field(..., alias="eventHash", description="Hash computed by the backend for signature verification") class WebhookConfig(BaseModel): @@ -50,61 +38,134 @@ class WebhookConfig(BaseModel): url: str = Field(..., description="Webhook URL") events: List[WebhookEventType] = Field(..., description="List of events to subscribe to") secret: Optional[str] = Field(default=None, description="Optional secret for signature verification") + secret_key: Optional[str] = Field(default=None, description="Alternative secret key name") + method: Optional[str] = Field(default="POST", description="HTTP method") + integration_type: Optional[str] = Field(default="ALL", description="Integration type") class WebhookResponse(BaseModel): """Response from webhook API operations""" - id: str = Field(..., description="Webhook ID") - url: str = Field(..., description="Webhook URL") - events: List[WebhookEventType] = Field(..., description="Subscribed events") + model_config = {"extra": "allow"} + + id: Optional[str] = Field(default=None, description="Webhook ID") + url: Optional[str] = Field(default=None, description="Webhook URL") + events: Optional[List[WebhookEventType]] = Field(default=None, description="Subscribed events") class Webhook: """Webhook service for handling CloudContactAI webhook events""" - + def __init__(self, ccai): self.ccai = ccai - + def register(self, config: WebhookConfig) -> WebhookResponse: - """Register a new webhook endpoint""" - response = self.ccai.request('POST', '/webhooks', config.dict()) - return WebhookResponse(**response) - + """Register a new webhook endpoint. + If secret is not provided, the server will auto-generate one. + The API expects an array of webhook objects and returns an array. + """ + payload = [{ + "url": config.url, + "method": config.method or "POST", + "integrationType": config.integration_type or "ALL", + }] + + # Only include secretKey if explicitly provided + secret = config.secret_key or config.secret + if secret: + payload[0]["secretKey"] = secret + + endpoint = f"/v1/client/{self.ccai.client_id}/integration" + response = self.ccai.request('POST', endpoint, payload) + + # API returns an array ā return the first element + if isinstance(response, list) and len(response) > 0: + data: Dict[str, Any] = dict(response[0]) + else: + data = dict(response) + return WebhookResponse(**data) + def update(self, webhook_id: str, config: Dict[str, Any]) -> WebhookResponse: - """Update an existing webhook configuration""" - response = self.ccai.request('PUT', f'/webhooks/{webhook_id}', config) - return WebhookResponse(**response) - + """Update an existing webhook configuration. + If secret is not provided, the server will keep the existing secret. + Uses POST to the same endpoint as register, with id in the payload. + """ + # Try to convert to int, fall back to string ID + try: + webhook_int_id = int(webhook_id) + except ValueError: + webhook_int_id = webhook_id + + payload = [{ + "id": webhook_int_id, + "url": config.get("url", ""), + "method": config.get("method", "POST"), + "integrationType": config.get("integration_type", "ALL"), + }] + + # Only include secretKey if explicitly provided + secret = config.get("secret_key") or config.get("secret") + if secret: + payload[0]["secretKey"] = secret + + endpoint = f"/v1/client/{self.ccai.client_id}/integration" + response = self.ccai.request('POST', endpoint, payload) + + # API returns an array ā return the first element + if isinstance(response, list) and len(response) > 0: + data: Dict[str, Any] = dict(response[0]) + else: + data = dict(response) + return WebhookResponse(**data) + def list(self) -> List[WebhookResponse]: """List all registered webhooks""" - response = self.ccai.request('GET', '/webhooks') - return [WebhookResponse(**webhook) for webhook in response] - + endpoint = f"/v1/client/{self.ccai.client_id}/integration" + response = self.ccai.request('GET', endpoint) + if isinstance(response, list): + return [WebhookResponse(**dict(webhook)) for webhook in response] + return [WebhookResponse(**dict(response))] + def delete(self, webhook_id: str) -> Dict[str, Any]: """Delete a webhook""" - return self.ccai.request('DELETE', f'/webhooks/{webhook_id}') - - def verify_signature(self, signature: str, body: str, secret: str) -> bool: - """Verify a webhook signature""" - # Placeholder for signature verification logic - # In production, this should implement proper HMAC verification - return True - + endpoint = f"/v1/client/{self.ccai.client_id}/integration/{webhook_id}" + return self.ccai.request('DELETE', endpoint) + + def verify_signature(self, signature: str, client_id: str, event_hash: str, secret: str) -> bool: + """Verify a webhook signature using HMAC-SHA256. + + Signature is computed as: HMAC-SHA256(secretKey, clientId:eventHash) encoded in Base64 + + :param signature: Signature from the X-CCAI-Signature header (Base64 encoded) + :param client_id: Client ID + :param event_hash: Event hash from the webhook payload + :param secret: Webhook secret key + :return: True if the signature is valid + """ + if not signature or not client_id or not event_hash or not secret: + return False + + # Compute: HMAC-SHA256(secretKey, "$clientId:$eventHash") + data = f"{client_id}:{event_hash}" + computed = hmac.new( + secret.encode('utf-8'), + data.encode('utf-8'), + hashlib.sha256 + ).digest() # raw bytes + + computed_base64 = base64.b64encode(computed).decode('utf-8') + + # Constant-time comparison to prevent timing attacks + return hmac.compare_digest(computed_base64, signature) + @staticmethod - def create_handler(handlers: Dict[str, Any]) -> callable: + def create_handler(handlers: Dict[str, Any]) -> Callable[[Dict[str, Any]], Dict[str, Any]]: """Create a webhook handler function for web frameworks""" def handler(request_body: Dict[str, Any]) -> Dict[str, Any]: - event_type = request_body.get('type') - - if event_type == WebhookEventType.MESSAGE_SENT: - event = MessageSentEvent(**request_body) - if 'on_message_sent' in handlers: - handlers['on_message_sent'](event) - elif event_type == WebhookEventType.MESSAGE_RECEIVED: - event = MessageReceivedEvent(**request_body) - if 'on_message_received' in handlers: - handlers['on_message_received'](event) - + event = WebhookEvent(**request_body) + + if 'on_event' in handlers: + handlers['on_event'](event) + return {'received': True} - - return handler \ No newline at end of file + + return handler diff --git a/src/ccai_python/webhook_handler_example.py b/src/ccai_python/webhook_handler_example.py index 7cf88bd..1eb16f4 100644 --- a/src/ccai_python/webhook_handler_example.py +++ b/src/ccai_python/webhook_handler_example.py @@ -2,45 +2,90 @@ import os sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) -from ccai_python import Webhook, WebhookEventType, MessageSentEvent, MessageReceivedEvent - -# Example webhook handler -def handle_message_sent(event: MessageSentEvent): - print(f"ā Message sent event received:") - print(f" Campaign: {event.campaign.title} (ID: {event.campaign.id})") - print(f" From: {event.from_}") - print(f" To: {event.to}") - print(f" Message: {event.message}") - -def handle_message_received(event: MessageReceivedEvent): - print(f"šØ Message received event received:") - print(f" Campaign: {event.campaign.title} (ID: {event.campaign.id})") - print(f" From: {event.from_}") - print(f" To: {event.to}") - print(f" Message: {event.message}") +from ccai_python import Webhook, WebhookEventType, WebhookEvent + +# Example webhook event handler +def handle_webhook_event(event: WebhookEvent): + """Handle CloudContactAI webhook events""" + print(f"šØ Webhook Event Received: {event.event_type}") + print(f" Event Hash: {event.event_hash}") + + # Handle different event types + if event.event_type == WebhookEventType.MESSAGE_SENT: + print(f"ā Message sent successfully") + if 'To' in event.data: + print(f" Recipient: {event.data['To']}") + if 'TotalPrice' in event.data: + print(f" Cost: ${event.data['TotalPrice']}") + if 'Segments' in event.data: + print(f" Segments: {event.data['Segments']}") + + elif event.event_type == WebhookEventType.MESSAGE_INCOMING: + print(f"š„ Message received (reply)") + if 'From' in event.data: + print(f" From: {event.data['From']}") + if 'Message' in event.data: + print(f" Message: {event.data['Message']}") + + elif event.event_type == WebhookEventType.MESSAGE_EXCLUDED: + print(f"ā ļø Message excluded") + if 'ExcludedReason' in event.data: + print(f" Reason: {event.data['ExcludedReason']}") + + elif event.event_type == WebhookEventType.MESSAGE_ERROR_CARRIER: + print(f"ā Carrier error") + if 'ErrorCode' in event.data: + print(f" Code: {event.data['ErrorCode']}") + if 'ErrorMessage' in event.data: + print(f" Message: {event.data['ErrorMessage']}") + + elif event.event_type == WebhookEventType.MESSAGE_ERROR_CLOUDCONTACT: + print(f"šØ System error") + if 'ErrorCode' in event.data: + print(f" Code: {event.data['ErrorCode']}") + if 'ErrorMessage' in event.data: + print(f" Message: {event.data['ErrorMessage']}") + + # Handle custom data if present + if 'CustomData' in event.data and event.data['CustomData']: + print(f" š Custom Data: {event.data['CustomData']}") + # Create webhook handler webhook_handler = Webhook.create_handler({ - 'on_message_sent': handle_message_sent, - 'on_message_received': handle_message_received + 'on_event': handle_webhook_event }) -# Example webhook payload (this would come from CloudContactAI) -test_payload = { - 'type': 'message.sent', - 'campaign': { - 'id': 123, - 'title': 'Test Campaign', - 'message': '', - 'sender_phone': '+11234567894', - 'created_at': '2025-01-14 22:18:28.273', - 'run_at': '' +# Example webhook payloads (these would come from CloudContactAI) +test_payloads = [ + { + 'eventType': 'message.sent', + 'eventHash': 'hash-abc123', + 'data': { + 'To': '+15551234567', + 'TotalPrice': 0.07, + 'Segments': 1, + 'CustomData': 'order-12345' + } }, - 'from': '+11234567894', - 'to': '+15551234567', - 'message': 'Hello Test User, this is a test message!' -} - -print("š§ Testing webhook handler with sample payload:") -result = webhook_handler(test_payload) -print(f"š¤ Handler result: {result}") \ No newline at end of file + { + 'eventType': 'message.incoming', + 'eventHash': 'hash-def456', + 'data': { + 'From': '+15551234567', + 'Message': 'Thanks for the message!' + } + }, + { + 'eventType': 'message.excluded', + 'eventHash': 'hash-ghi789', + 'data': { + 'ExcludedReason': 'Invalid phone number' + } + } +] + +print("š§ Testing webhook handler with sample payloads:\n") +for payload in test_payloads: + result = webhook_handler(payload) + print(f"ā Result: {result}\n") \ No newline at end of file diff --git a/src/ccai_python/webhook_test.py b/src/ccai_python/webhook_test.py index d37a6a4..0f9e911 100644 --- a/src/ccai_python/webhook_test.py +++ b/src/ccai_python/webhook_test.py @@ -4,14 +4,15 @@ import os from dotenv import load_dotenv -from ccai_python import CCAI, Account +from ccai_python import CCAI +from ccai_python.sms.sms import Account # Load environment variables load_dotenv(os.path.join(os.path.dirname(__file__), '..', '..', '.env')) ccai = CCAI( - client_id=os.getenv('CCAI_CLIENT_ID'), - api_key=os.getenv('CCAI_API_KEY'), + client_id=os.environ['CCAI_CLIENT_ID'], + api_key=os.environ['CCAI_API_KEY'], use_test=True ) @@ -19,9 +20,9 @@ print("š Sending test SMS to trigger webhook...") account = Account( - first_name=os.getenv('TEST_FIRST_NAME'), - last_name=os.getenv('TEST_LAST_NAME'), - phone=os.getenv('TEST_PHONE_NUMBER') + first_name=os.environ['TEST_FIRST_NAME'], + last_name=os.environ['TEST_LAST_NAME'], + phone=os.environ['TEST_PHONE_NUMBER'] ) try: diff --git a/tests/test_ccai.py b/tests/test_ccai.py index afbe865..07f54e4 100644 --- a/tests/test_ccai.py +++ b/tests/test_ccai.py @@ -5,6 +5,7 @@ :copyright: 2025 CloudContactAI LLC """ +import os import unittest from unittest.mock import patch, MagicMock @@ -13,32 +14,54 @@ class TestCCAI(unittest.TestCase): """Test cases for the CCAI client""" - + def setUp(self): """Set up test fixtures""" self.client_id = "test-client-id" self.api_key = "test-api-key" self.ccai = CCAI(client_id=self.client_id, api_key=self.api_key) - + def test_initialization(self): """Test client initialization""" self.assertEqual(self.ccai.client_id, self.client_id) self.assertEqual(self.ccai.api_key, self.api_key) self.assertEqual(self.ccai.base_url, "https://core.cloudcontactai.com/api") - + # Test custom base URL custom_url = "https://custom.api.example.com" ccai = CCAI(client_id=self.client_id, api_key=self.api_key, base_url=custom_url) self.assertEqual(ccai.base_url, custom_url) - + + def test_initialization_with_test_environment(self): + """Test client initialization with test environment""" + ccai = CCAI(client_id=self.client_id, api_key=self.api_key, use_test=True) + self.assertEqual(ccai.base_url, "https://core-test-cloudcontactai.allcode.com/api") + self.assertEqual(ccai.email_base_url, "https://email-campaigns-test-cloudcontactai.allcode.com/api/v1") + self.assertEqual(ccai.file_base_url, "https://files-test-cloudcontactai.allcode.com") + self.assertTrue(ccai.use_test) + + def test_initialization_production_default(self): + """Test production URLs are default""" + ccai = CCAI(client_id=self.client_id, api_key=self.api_key) + self.assertEqual(ccai.base_url, "https://core.cloudcontactai.com/api") + self.assertEqual(ccai.email_base_url, "https://email-campaigns.cloudcontactai.com/api/v1") + self.assertEqual(ccai.file_base_url, "https://files.cloudcontactai.com") + self.assertFalse(ccai.use_test) + def test_initialization_validation(self): """Test validation during initialization""" with self.assertRaises(ValueError): CCAI(client_id="", api_key=self.api_key) - + with self.assertRaises(ValueError): CCAI(client_id=self.client_id, api_key="") - + + @patch.dict(os.environ, {"CCAI_BASE_URL": "https://env-base.example.com"}) + def test_env_var_override(self): + """Test environment variable URL override""" + ccai = CCAI(client_id=self.client_id, api_key=self.api_key) + self.assertEqual(ccai.base_url, "https://env-base.example.com") + @patch('requests.request') def test_request(self, mock_request): """Test the request method""" @@ -47,11 +70,11 @@ def test_request(self, mock_request): mock_response.json.return_value = {"status": "success"} mock_response.raise_for_status.return_value = None mock_request.return_value = mock_response - + # Test GET request result = self.ccai.request("get", "/test-endpoint") self.assertEqual(result, {"status": "success"}) - + # Verify request was made correctly mock_request.assert_called_with( method="GET", @@ -64,12 +87,12 @@ def test_request(self, mock_request): json=None, timeout=30 ) - + # Test POST request with data data = {"key": "value"} result = self.ccai.request("post", "/test-endpoint", data=data) self.assertEqual(result, {"status": "success"}) - + # Verify request was made correctly mock_request.assert_called_with( method="POST", diff --git a/tests/test_email.py b/tests/test_email.py new file mode 100644 index 0000000..a6e5ba8 --- /dev/null +++ b/tests/test_email.py @@ -0,0 +1,135 @@ +""" +Tests for the email service + +:license: MIT +:copyright: 2025 CloudContactAI LLC +""" + +import unittest +from unittest.mock import patch, MagicMock + +from ccai_python import CCAI +from ccai_python.email_service import Email, EmailCampaign, EmailAccount + + +class TestEmail(unittest.TestCase): + """Test cases for the Email service""" + + def setUp(self): + """Set up test fixtures""" + self.client_id = "test-client-id" + self.api_key = "test-api-key" + self.ccai = CCAI(client_id=self.client_id, api_key=self.api_key) + + @patch('requests.request') + def test_send_single_uses_client_email_url(self, mock_request): + """Test that email uses the client's email_base_url, not hardcoded""" + mock_response = MagicMock() + mock_response.json.return_value = {"id": 123, "status": "PENDING"} + mock_response.raise_for_status.return_value = None + mock_request.return_value = mock_response + + self.ccai.email.send_single( + first_name="John", + last_name="Doe", + email="john@example.com", + subject="Test Subject", + message="Test
", + sender_email="sender@example.com", + reply_email="reply@example.com", + sender_name="Test Sender", + title="Test Campaign" + ) + + # Verify the URL uses client's email_base_url, not hardcoded test URL + call_url = mock_request.call_args.kwargs.get('url') or mock_request.call_args[1].get('url') + self.assertIn("email-campaigns.cloudcontactai.com/api/v1", call_url) + # Verify headers use client_id, not hardcoded 1223 + call_headers = mock_request.call_args.kwargs.get('headers') or mock_request.call_args[1].get('headers') + self.assertEqual(call_headers.get('AccountId'), 'test-client-id') + self.assertEqual(call_headers.get('ClientId'), 'test-client-id') + self.assertNotEqual(call_headers.get('AccountId'), '1223') + + @patch('requests.request') + def test_send_single_test_environment(self, mock_request): + """Test email URL in test environment""" + mock_response = MagicMock() + mock_response.json.return_value = {"id": 456, "status": "PENDING"} + mock_response.raise_for_status.return_value = None + mock_request.return_value = mock_response + + ccai = CCAI(client_id=self.client_id, api_key=self.api_key, use_test=True) + ccai.email.send_single( + first_name="John", last_name="Doe", email="john@example.com", + subject="Test", message="Test
", + sender_email="s@e.com", reply_email="r@e.com", + sender_name="Sender", title="Title" + ) + + call_url = mock_request.call_args.kwargs.get('url') or mock_request.call_args[1].get('url') + self.assertIn("email-campaigns-test-cloudcontactai.allcode.com/api/v1", call_url) + + def test_send_validation_empty_accounts(self): + """Test validation for empty accounts""" + campaign = EmailCampaign( + subject="Test", title="Test", message="Test
", + sender_email="s@e.com", reply_email="r@e.com", sender_name="Sender", + accounts=[] + ) + with self.assertRaises(ValueError) as ctx: + self.ccai.email.send_campaign(campaign) + self.assertIn("At least one account is required", str(ctx.exception)) + + def test_send_validation_missing_fields(self): + """Test validation for missing required fields""" + with self.assertRaises(ValueError): + self.ccai.email.send_single( + first_name="John", last_name="Doe", email="john@example.com", + subject="", message="Test
", + sender_email="s@e.com", reply_email="r@e.com", + sender_name="Sender", title="Title" + ) + + + @patch('requests.request') + def test_email_account_custom_fields_and_id(self, mock_request): + """Test that customAccountId and data are sent in the request""" + mock_response = MagicMock() + mock_response.json.return_value = { + "id": 123, + "status": "PENDING", + "message": "Email campaign sent successfully", + "responseId": "resp-email-xyz", + } + mock_response.raise_for_status.return_value = None + mock_request.return_value = mock_response + + campaign = EmailCampaign( + subject="Test", title="Test", message="Test
", + sender_email="s@e.com", reply_email="r@e.com", sender_name="Sender", + accounts=[EmailAccount( + first_name="John", last_name="Doe", email="john@example.com", + custom_account_id="ext-id-123", + data={"tier": "gold", "locale": "en-US"}, + )] + ) + result = self.ccai.email.send_campaign(campaign) + + # Verify request body contains the custom fields + call_json = mock_request.call_args.kwargs.get("json") or mock_request.call_args[1].get("json") + account_in_payload = call_json["accounts"][0] + self.assertEqual(account_in_payload.get("customAccountId"), "ext-id-123") + self.assertEqual(account_in_payload.get("data"), {"tier": "gold", "locale": "en-US"}) + # Verify wire format uses camelCase for campaign fields + self.assertIn("senderEmail", call_json) + self.assertIn("replyEmail", call_json) + self.assertNotIn("sender_email", call_json) + self.assertNotIn("reply_email", call_json) + + # Verify response fields + self.assertEqual(result.message, "Email campaign sent successfully") + self.assertEqual(result.response_id, "resp-email-xyz") + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_mms.py b/tests/test_mms.py index 4987473..863253f 100644 --- a/tests/test_mms.py +++ b/tests/test_mms.py @@ -523,5 +523,30 @@ def test_check_file_uploaded_not_found(self): self.assertEqual(result.url, "") + @patch('requests.Session.post') + def test_send_maps_data_and_message_data(self, mock_post): + """Test that data and message_data are mapped to wire format data/messageData""" + mock_response = MagicMock() + mock_response.json.return_value = {"id": "mms-123", "status": "sent"} + mock_response.status_code = 200 + mock_post.return_value = mock_response + + account = Account( + first_name="John", + last_name="Doe", + phone="+15551234567", + data={"city": "Miami", "plan": "premium"}, + message_data='{"source":"mms-test"}' + ) + + self.ccai.mms.send("file-key", [account], "Hello from ${city}!", "Test") + + call_json = mock_post.call_args.kwargs.get("json") or mock_post.call_args[1].get("json") + sent_account = call_json["accounts"][0] + self.assertEqual(sent_account["data"], {"city": "Miami", "plan": "premium"}) + self.assertEqual(sent_account["messageData"], '{"source":"mms-test"}') + self.assertNotIn("message_data", sent_account) + + if __name__ == '__main__': unittest.main() diff --git a/tests/test_sms.py b/tests/test_sms.py index cd53c4b..99b4ece 100644 --- a/tests/test_sms.py +++ b/tests/test_sms.py @@ -73,7 +73,7 @@ def test_send(self, mock_request): "message": self.message, "title": self.title }, - timeout=None + timeout=30 ) @patch.object(CCAI, 'request') @@ -112,7 +112,7 @@ def test_send_with_dict_accounts(self, mock_request): "message": self.message, "title": self.title }, - timeout=None + timeout=30 ) @patch.object(CCAI, 'request') @@ -149,7 +149,7 @@ def test_send_single(self, mock_request): "message": "Hi ${first_name}, thanks for your interest!", "title": "Single Message Test" }, - timeout=None + timeout=30 ) @patch.object(CCAI, 'request') @@ -207,6 +207,61 @@ def track_progress(status: str): timeout=60 ) + @patch.object(CCAI, 'request') + def test_send_with_data_and_message_data(self, mock_request): + """Test that data and messageData are included in the request payload""" + mock_request.return_value = { + "id": "msg-cf-123", + "status": "sent", + "message": "SMS sent successfully", + "responseId": "resp-abc-456", + } + + account = Account( + first_name="John", + last_name="Doe", + phone="+14156961732", + data={"city": "Miami", "country": "USA", "plan": "premium"}, + message_data='{"source":"python-sdk-test"}', + ) + + response = self.ccai.sms.send( + accounts=[account], + message="Hello ${firstName} from ${city}!", + title="Test data fields" + ) + + # Verify payload contains "data" and "messageData" keys (wire format) + call_args = mock_request.call_args + payload = call_args.kwargs.get("data") or call_args[1].get("data") + sent_account = payload["accounts"][0] + self.assertEqual(sent_account["data"], {"city": "Miami", "country": "USA", "plan": "premium"}) + self.assertEqual(sent_account["messageData"], '{"source":"python-sdk-test"}') + self.assertNotIn("message_data", sent_account) + + # Verify response fields + self.assertEqual(response.message, "SMS sent successfully") + self.assertEqual(response.response_id, "resp-abc-456") + + @patch.object(CCAI, 'request') + def test_response_message_and_response_id(self, mock_request): + """Test that message and response_id are returned from API response""" + mock_request.return_value = { + "id": "msg-123", + "status": "sent", + "message": "SMS sent successfully", + "responseId": "resp-id-xyz", + } + + response = self.ccai.sms.send( + accounts=[self.account], + message=self.message, + title=self.title + ) + + self.assertEqual(response.message, "SMS sent successfully") + self.assertEqual(response.response_id, "resp-id-xyz") + def test_validation(self): """Test input validation""" # Test empty accounts diff --git a/tests/test_webhook.py b/tests/test_webhook.py new file mode 100644 index 0000000..09bf678 --- /dev/null +++ b/tests/test_webhook.py @@ -0,0 +1,178 @@ +""" +Tests for the webhook service + +:license: MIT +:copyright: 2025 CloudContactAI LLC +""" + +import hmac +import hashlib +import base64 +import unittest +from unittest.mock import patch, MagicMock + +from ccai_python import CCAI +from ccai_python.webhook import Webhook, WebhookConfig, WebhookEventType + + +class TestWebhook(unittest.TestCase): + """Test cases for the Webhook service""" + + def setUp(self): + """Set up test fixtures""" + self.client_id = "test-client-id" + self.api_key = "test-api-key" + self.ccai = CCAI(client_id=self.client_id, api_key=self.api_key) + + @patch.object(CCAI, 'request') + def test_register_webhook_auto_generated_secret(self, mock_request): + """Test webhook registration with auto-generated secret""" + mock_request.return_value = [{ + "id": "webhook_123", + "url": "https://example.com/webhook", + "events": ["message.sent"], + "secretKey": "sk_live_auto_generated" + }] + + config = WebhookConfig( + url="https://example.com/webhook", + events=[WebhookEventType.MESSAGE_SENT] + # secret not provided - server should auto-generate + ) + result = self.ccai.webhook.register(config) + + self.assertEqual(result.id, "webhook_123") + self.assertEqual(result.url, "https://example.com/webhook") + mock_request.assert_called_once() + call_args = mock_request.call_args + self.assertEqual(call_args[0][0], 'POST') + self.assertIn('/v1/client/', call_args[0][1]) + # Payload should be an array WITHOUT secretKey + self.assertIsInstance(call_args[0][2], list) + self.assertNotIn("secretKey", call_args[0][2][0]) + + @patch.object(CCAI, 'request') + def test_register_webhook_custom_secret(self, mock_request): + """Test webhook registration with custom secret""" + mock_request.return_value = [{ + "id": "webhook_124", + "url": "https://example.com/webhook-custom", + "events": ["message.sent"], + "secretKey": "my-custom-secret" + }] + + config = WebhookConfig( + url="https://example.com/webhook-custom", + events=[WebhookEventType.MESSAGE_SENT], + secret="my-custom-secret" + ) + result = self.ccai.webhook.register(config) + + self.assertEqual(result.id, "webhook_124") + mock_request.assert_called_once() + call_args = mock_request.call_args + # Payload should be an array WITH secretKey + self.assertIsInstance(call_args[0][2], list) + self.assertIn("secretKey", call_args[0][2][0]) + self.assertEqual(call_args[0][2][0]["secretKey"], "my-custom-secret") + + @patch.object(CCAI, 'request') + def test_update_webhook(self, mock_request): + """Test webhook update without secret""" + mock_request.return_value = [{ + "id": "webhook_123", + "url": "https://example.com/updated", + "events": ["message.received"] + }] + + result = self.ccai.webhook.update("webhook_123", {"url": "https://example.com/updated"}) + + self.assertEqual(result.url, "https://example.com/updated") + call_args = mock_request.call_args + self.assertEqual(call_args[0][0], 'POST') + # Payload should be an array with id (string IDs pass through as-is) + self.assertIsInstance(call_args[0][2], list) + self.assertEqual(call_args[0][2][0]["id"], "webhook_123") + # Payload should NOT contain secretKey when not provided + self.assertNotIn("secretKey", call_args[0][2][0]) + + @patch.object(CCAI, 'request') + def test_update_webhook_with_secret(self, mock_request): + """Test webhook update with custom secret""" + mock_request.return_value = [{ + "id": "webhook_123", + "url": "https://example.com/updated", + "events": ["message.received"], + "secretKey": "new-secret" + }] + + result = self.ccai.webhook.update("webhook_123", { + "url": "https://example.com/updated", + "secret": "new-secret" + }) + + self.assertEqual(result.url, "https://example.com/updated") + call_args = mock_request.call_args + # Payload should contain secretKey when provided + self.assertIn("secretKey", call_args[0][2][0]) + self.assertEqual(call_args[0][2][0]["secretKey"], "new-secret") + + @patch.object(CCAI, 'request') + def test_list_webhooks(self, mock_request): + """Test listing webhooks""" + mock_request.return_value = [{ + "id": "webhook_123", + "url": "https://example.com/webhook", + "events": ["message.sent"] + }] + + result = self.ccai.webhook.list() + + self.assertEqual(len(result), 1) + self.assertEqual(result[0].id, "webhook_123") + + @patch.object(CCAI, 'request') + def test_delete_webhook(self, mock_request): + """Test deleting a webhook""" + mock_request.return_value = {"success": True, "message": "Webhook deleted"} + + result = self.ccai.webhook.delete("webhook_123") + + self.assertTrue(result["success"]) + call_args = mock_request.call_args + self.assertEqual(call_args[0][0], 'DELETE') + self.assertIn('webhook_123', call_args[0][1]) + + def test_verify_signature_valid(self): + """Test valid signature verification""" + client_id = "test-client-id" + event_hash = "event-hash-abc123" + secret = "test-secret" + + # Compute expected signature: HMAC-SHA256(secretKey, clientId:eventHash) in Base64 + data = f"{client_id}:{event_hash}" + computed = hmac.new( + secret.encode('utf-8'), + data.encode('utf-8'), + hashlib.sha256 + ).digest() # raw bytes + expected = base64.b64encode(computed).decode('utf-8') + + result = self.ccai.webhook.verify_signature(expected, client_id, event_hash, secret) + self.assertTrue(result) + + def test_verify_signature_invalid(self): + """Test invalid signature rejection""" + result = self.ccai.webhook.verify_signature("bad-signature", "client-id", "event-hash", "test-secret") + self.assertFalse(result) + + def test_verify_signature_empty_params(self): + """Test empty parameter rejection""" + self.assertFalse(self.ccai.webhook.verify_signature("", "client-id", "event-hash", "test-secret")) + self.assertFalse(self.ccai.webhook.verify_signature("sig", "", "event-hash", "test-secret")) + self.assertFalse(self.ccai.webhook.verify_signature("sig", "client-id", "", "test-secret")) + self.assertFalse(self.ccai.webhook.verify_signature("sig", "client-id", "event-hash", "")) + + +if __name__ == '__main__': + unittest.main()