diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py new file mode 100644 index 00000000..c02545dd --- /dev/null +++ b/examples/notification_configuration.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +""" +Example usage of Notification Configuration API + +This example demonstrates how to use the Python TFE library to manage +notification configurations for workspaces and teams. +""" + +import os +import sys + +# Add the src directory to the Python path so we can import the tfe module +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from pytfe.client import TFEClient +from pytfe.models.notification_configuration import ( + NotificationConfigurationCreateOptions, + NotificationConfigurationListOptions, + NotificationConfigurationSubscribableChoice, + NotificationConfigurationUpdateOptions, + NotificationDestinationType, + NotificationTriggerType, +) + + +def main(): + """Demonstrate notification configuration operations.""" + + # Initialize the TFE client + # Make sure to set TFE_ADDRESS and TFE_TOKEN environment variables + client = TFEClient() + + print("=== Python TFE Notification Configuration Example ===\n") + + # Use example workspace ID (replace with your actual workspace ID) + workspace_id = "ws-example123456789" # Replace with your workspace ID + workspace_name = "your-workspace-name" + print(f"Using workspace: {workspace_name} (ID: {workspace_id})") + + # Use fake team ID for demonstration (teams not available in free plan) + team_id = "team-example123456789" + print("Using fake team ID for demonstration (teams not available in free plan)") + + try: + # ===== List notification configurations for workspace ===== + print("1. Listing notification configurations for workspace...") + try: + workspace_notifications = client.notification_configurations.list( + subscribable_id=workspace_id + ) + print( + f"Found {len(workspace_notifications.items)} notification configurations" + ) + for nc in workspace_notifications.items: + print(f" - {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})") + except Exception as e: + print(f" Error listing workspace notifications: {e}") + + print() + + # ===== List notification configurations for team ===== + print("2. Listing notification configurations for team...") + try: + team_choice = NotificationConfigurationSubscribableChoice( + team={"id": team_id} + ) + options = NotificationConfigurationListOptions( + subscribable_choice=team_choice + ) + team_notifications = client.notification_configurations.list( + subscribable_id=team_id, options=options + ) + print( + f"Found {len(team_notifications.items)} team notification configurations" + ) + for nc in team_notifications.items: + print(f" - {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})") + except Exception as e: + error_msg = str(e).lower() + if "not found" in error_msg: + print(f" ⚠️ Team not found (expected with fake team ID): {team_id}") + print(" 💡 Teams are not available in HCP Terraform free plan") + else: + print(f" ❌ Error listing team notifications: {e}") + + print() + + # ===== Create a new workspace notification configuration ===== + print("3. Creating a new workspace notification configuration...") + try: + workspace_choice = NotificationConfigurationSubscribableChoice( + workspace={"id": workspace_id} + ) + create_options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.SLACK, + enabled=True, + name="Python TFE Example Slack Notification", + subscribable_choice=workspace_choice, + url="https://hooks.slack.com/services/YOUR_SLACK_WORKSPACE/YOUR_CHANNEL/YOUR_WEBHOOK_TOKEN", + triggers=[ + NotificationTriggerType.COMPLETED, + NotificationTriggerType.ERRORED, + ], + ) + + new_notification = client.notification_configurations.create( + workspace_id, create_options + ) + print( + f" Created notification: {new_notification.name} (ID: {new_notification.id})" + ) + + notification_id = new_notification.id + + # ===== Read the notification configuration ===== + print("\n4. Reading the notification configuration...") + read_notification = client.notification_configurations.read( + notification_config_id=notification_id + ) + print(f" Read notification: {read_notification.name}") + print(f" Destination type: {read_notification.destination_type}") + print(f" Enabled: {read_notification.enabled}") + print(f" Triggers: {read_notification.triggers}") + + # ===== Update the notification configuration ===== + print("\n5. Updating the notification configuration...") + update_options = NotificationConfigurationUpdateOptions( + name="Updated Python TFE Example Webhook", + enabled=False, + triggers=[NotificationTriggerType.ERRORED], # Only notify on errors + ) + + updated_notification = client.notification_configurations.update( + notification_config_id=notification_id, options=update_options + ) + print(f" Updated notification: {updated_notification.name}") + print(f" Enabled: {updated_notification.enabled}") + + # ===== Verify the notification configuration ===== + print("\n6. Verifying the notification configuration...") + print(" Note: This will fail with fake URLs - that's expected!") + try: + client.notification_configurations.verify( + notification_config_id=notification_id + ) + print( + f" ✅ Verification successful for notification ID: {notification_id}" + ) + print(" Note: Verification sends a test payload to the configured URL") + except Exception as e: + print(f" ⚠️ Verification failed (expected with fake URL): {e}") + print( + " 💡 To test verification, use a real webhook URL from Slack, Teams, or Discord" + ) + + # ===== Delete the notification configuration ===== + print("\n7. Deleting the notification configuration...") + client.notification_configurations.delete( + notification_config_id=notification_id + ) + print(f" Deleted notification configuration: {notification_id}") + + # Verify deletion + try: + client.notification_configurations.read( + notification_config_id=notification_id + ) + print(" ERROR: Notification still exists after deletion!") + except Exception: + print(" Confirmed: Notification configuration has been deleted") + + except Exception as e: + error_msg = str(e).lower() + if "verification failed" in error_msg and "404" in error_msg: + print(" ⚠️ Webhook verification failed (expected with fake URL)") + print( + " 💡 The fake Slack URL returns 404 - this is normal for testing" + ) + print(" 🔗 To test real verification, use a webhook from:") + print(" • webhook.site (instant test URL)") + print(" • Slack, Teams, or Discord webhook") + else: + print(f" ❌ Error in workspace notification operations: {e}") + + print() + + # ===== Create a team notification configuration ===== + print("8. Creating a team notification configuration...") + try: + if team_id != "team-example123456789": # Only try if we have a real team ID + team_choice = NotificationConfigurationSubscribableChoice( + team={"id": team_id} + ) + team_create_options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.SLACK, + enabled=True, + name="Team Slack Notifications", + subscribable_choice=team_choice, + url="https://hooks.slack.com/services/YOUR_SLACK_WORKSPACE/YOUR_CHANNEL/YOUR_WEBHOOK_TOKEN", + triggers=[NotificationTriggerType.COMPLETED], + ) + + team_notification = client.notification_configurations.create( + team_id, team_create_options + ) + print( + f" Created team notification: {team_notification.name} (ID: {team_notification.id})" + ) + + # Clean up team notification + client.notification_configurations.delete( + notification_config_id=team_notification.id + ) + print(f" Cleaned up team notification: {team_notification.id}") + else: + print( + f" Skipping team notifications - no real team ID available (using: {team_id})" + ) + + except Exception as e: + error_msg = str(e).lower() + if "not found" in error_msg or "team" in error_msg: + print(" ⚠️ Team operations not available (expected with fake team ID)") + print(" 💡 Teams require HCP Terraform paid plan or Enterprise") + else: + print(f" ❌ Error in team notification operations: {e}") + + print() + + # ===== Create a Microsoft Teams notification configuration ===== + print("9. Creating a Microsoft Teams notification configuration...") + try: + workspace_choice = NotificationConfigurationSubscribableChoice( + workspace={"id": workspace_id} + ) + teams_create_options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.MICROSOFT_TEAMS, + enabled=True, + name="Teams Notifications", + subscribable_choice=workspace_choice, + url="https://outlook.office.com/webhook/YOUR_TENANT_ID@YOUR_TENANT_ID/IncomingWebhook/YOUR_CONNECTOR_ID/YOUR_TEAMS_WEBHOOK_TOKEN", + triggers=[ + NotificationTriggerType.ERRORED, + NotificationTriggerType.NEEDS_ATTENTION, + ], + ) + + teams_notification = client.notification_configurations.create( + workspace_id, teams_create_options + ) + print( + f" Created Teams notification: {teams_notification.name} (ID: {teams_notification.id})" + ) + + # Clean up Teams notification + client.notification_configurations.delete( + notification_config_id=teams_notification.id + ) + print(f" Cleaned up Teams notification: {teams_notification.id}") + + except Exception as e: + print(f" Error in Teams notification operations: {e}") + + except Exception as e: + print(f"Error: {e}") + print("\nNote: Make sure to:") + print( + "1. Set TFE_ADDRESS environment variable (e.g., https://app.terraform.io)" + ) + print("2. Set TFE_TOKEN environment variable with your API token") + print( + "3. Replace workspace_id and team_id with actual values from your organization" + ) + + print("\n=== Notification Configuration Example Complete ===") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 8af2a385..0fda7767 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -6,6 +6,7 @@ from .resources.agents import Agents, AgentTokens from .resources.apply import Applies from .resources.configuration_version import ConfigurationVersions +from .resources.notification_configuration import NotificationConfigurations from .resources.oauth_client import OAuthClients from .resources.oauth_token import OAuthTokens from .resources.organizations import Organizations @@ -56,6 +57,7 @@ def __init__(self, config: TFEConfig | None = None): # Core resources self.configuration_versions = ConfigurationVersions(self._transport) + self.notification_configurations = NotificationConfigurations(self._transport) self.applies = Applies(self._transport) self.plans = Plans(self._transport) self.organizations = Organizations(self._transport) diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py new file mode 100644 index 00000000..0632a1e8 --- /dev/null +++ b/src/pytfe/models/notification_configuration.py @@ -0,0 +1,451 @@ +""" +Notification Configuration Models + +This module provides models for working with Terraform Cloud/Enterprise notification configurations. +""" + +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import Any + + +class NotificationTriggerType(Enum): + """Represents the different TFE notifications that can be sent as a run's progress transitions between different states.""" + + # Run triggers + CREATED = "run:created" + PLANNING = "run:planning" + NEEDS_ATTENTION = "run:needs_attention" + APPLYING = "run:applying" + COMPLETED = "run:completed" + ERRORED = "run:errored" + + # Assessment triggers + ASSESSMENT_DRIFTED = "assessment:drifted" + ASSESSMENT_FAILED = "assessment:failed" + ASSESSMENT_CHECK_FAILED = "assessment:check_failure" + + # Workspace triggers + WORKSPACE_AUTO_DESTROY_REMINDER = "workspace:auto_destroy_reminder" + WORKSPACE_AUTO_DESTROY_RUN_RESULTS = "workspace:auto_destroy_run_results" + + # Change request triggers + CHANGE_REQUEST_CREATED = "change_request:created" + + +class NotificationDestinationType(Enum): + """Represents the destination type of the notification configuration.""" + + EMAIL = "email" + GENERIC = "generic" + SLACK = "slack" + MICROSOFT_TEAMS = "microsoft-teams" + + +class DeliveryResponse: + """Represents a notification configuration delivery response.""" + + # Type annotations for instance attributes + body: str + code: str + headers: dict[str, Any] + sent_at: datetime | None + successful: str + url: str + + def __init__(self, data: dict[str, Any]): + self.body = data.get("body", "") + self.code = data.get("code", "") + self.headers = data.get("headers", {}) + self.sent_at = self._parse_datetime(data.get("sent-at")) + self.successful = data.get("successful", "") + self.url = data.get("url", "") + + def _parse_datetime(self, date_str: str | None) -> datetime | None: + """Parse ISO 8601 datetime string.""" + if not date_str: + return None + try: + return datetime.fromisoformat(date_str.replace("Z", "+00:00")) + except (ValueError, AttributeError): + return None + + def __repr__(self) -> str: + return f"DeliveryResponse(url='{self.url}', code='{self.code}', successful='{self.successful}')" + + +class NotificationConfigurationSubscribableChoice: + """Choice type struct that represents the possible values within a polymorphic relation.""" + + # Type annotations for instance attributes + team: Any | None + workspace: Any | None + + def __init__(self, team: Any | None = None, workspace: Any | None = None): + self.team = team + self.workspace = workspace + + def __repr__(self) -> str: + if self.team: + return f"NotificationConfigurationSubscribableChoice(team={self.team})" + elif self.workspace: + return f"NotificationConfigurationSubscribableChoice(workspace={self.workspace})" + return "NotificationConfigurationSubscribableChoice()" + + +class NotificationConfiguration: + """Represents a Notification Configuration.""" + + # Type annotations for instance attributes + id: str | None + created_at: datetime | None + updated_at: datetime | None + destination_type: str | None + enabled: bool + name: str + token: str + url: str + triggers: list[NotificationTriggerType] + delivery_responses: list[Any] + email_addresses: list[str] + email_users: list[Any] + subscribable: Any + subscribable_choice: Any | None + + def __init__(self, data: dict[str, Any]): + self.id = data.get("id") + self.created_at = self._parse_datetime(data.get("created-at")) + self.updated_at = self._parse_datetime(data.get("updated-at")) + + # Core attributes + self.destination_type = data.get("destination-type") + self.enabled = data.get("enabled", False) + self.name = data.get("name", "") + self.token = data.get("token", "") + self.url = data.get("url", "") + + # Triggers - convert from strings to enum values + self.triggers = self._parse_triggers(data.get("triggers", [])) + + # Delivery responses + delivery_responses_data = data.get("delivery-responses", []) + self.delivery_responses = [ + DeliveryResponse(dr) for dr in delivery_responses_data + ] + + # Email configuration + self.email_addresses = data.get("email-addresses", []) + self.email_users = data.get("email-users", []) + + # Relationships - using polymorphic relation pattern + self.subscribable = data.get( + "subscribable" + ) # Deprecated but maintained for compatibility + self.subscribable_choice = self._parse_subscribable_choice( + data.get("subscribable-choice") + ) + + def _parse_datetime(self, date_str: str | None) -> datetime | None: + """Parse ISO 8601 datetime string.""" + if not date_str: + return None + try: + return datetime.fromisoformat(date_str.replace("Z", "+00:00")) + except (ValueError, AttributeError): + return None + + def _parse_triggers(self, triggers: list[str]) -> list[NotificationTriggerType]: + """Parse trigger strings to enum values.""" + parsed_triggers = [] + for trigger in triggers: + try: + parsed_triggers.append(NotificationTriggerType(trigger)) + except ValueError: + # If trigger is not in enum, keep as string for backwards compatibility + pass + return parsed_triggers + + def _parse_subscribable_choice( + self, choice_data: dict[str, Any] | None + ) -> NotificationConfigurationSubscribableChoice | None: + """Parse subscribable choice data.""" + if not choice_data: + return None + + team = choice_data.get("team") + workspace = choice_data.get("workspace") + return NotificationConfigurationSubscribableChoice( + team=team, workspace=workspace + ) + + def __repr__(self) -> str: + return f"NotificationConfiguration(id='{self.id}', name='{self.name}', enabled={self.enabled})" + + +class NotificationConfigurationListOptions: + """Represents the options for listing notification configurations.""" + + # Type annotations for instance attributes + page_number: int | None + page_size: int | None + subscribable_choice: NotificationConfigurationSubscribableChoice | None + + def __init__( + self, + page_number: int | None = None, + page_size: int | None = None, + subscribable_choice: NotificationConfigurationSubscribableChoice | None = None, + ): + self.page_number = page_number + self.page_size = page_size + self.subscribable_choice = subscribable_choice + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary for API requests.""" + params = {} + + if self.page_number is not None: + params["page[number]"] = self.page_number + if self.page_size is not None: + params["page[size]"] = self.page_size + + return params + + +class NotificationConfigurationCreateOptions: + """Represents the options for creating a new notification configuration.""" + + # Type annotations for instance attributes + destination_type: NotificationDestinationType + enabled: bool + name: str + token: str | None + triggers: list[NotificationTriggerType] + url: str | None + email_addresses: list[str] + email_users: list[Any] + subscribable_choice: NotificationConfigurationSubscribableChoice | None + + def __init__( + self, + destination_type: NotificationDestinationType, + enabled: bool, + name: str, + token: str | None = None, + triggers: list[NotificationTriggerType] | None = None, + url: str | None = None, + email_addresses: list[str] | None = None, + email_users: list[Any] | None = None, + subscribable_choice: NotificationConfigurationSubscribableChoice | None = None, + ): + # Required fields + self.destination_type = destination_type + self.enabled = enabled + self.name = name + + # Optional fields + self.token = token + self.triggers = triggers or [] + self.url = url + self.email_addresses = email_addresses or [] + self.email_users = email_users or [] + self.subscribable_choice = subscribable_choice + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary for API requests.""" + data: dict[str, Any] = { + "type": "notification-configurations", + "attributes": { + "destination-type": self.destination_type.value, + "enabled": self.enabled, + "name": self.name, + }, + } + + # Add optional attributes + if self.token is not None: + data["attributes"]["token"] = self.token + + if self.triggers: + data["attributes"]["triggers"] = [ + trigger.value for trigger in self.triggers + ] + + if self.url is not None: + data["attributes"]["url"] = self.url + + if self.email_addresses: + data["attributes"]["email-addresses"] = self.email_addresses + + # Handle relationships + if self.email_users: + data["relationships"] = data.get("relationships", {}) + data["relationships"]["users"] = { + "data": [ + { + "type": "users", + "id": user.id if hasattr(user, "id") else str(user), + } + for user in self.email_users + ] + } + + return data + + def validate(self) -> list[str]: + """Validate the create options and return any errors.""" + errors = [] + + # Required field validation + if not self.name or not self.name.strip(): + errors.append("Name is required") + + if not isinstance(self.enabled, bool): + errors.append("Enabled must be a boolean") # type: ignore[unreachable] + + # URL validation for certain destination types + if self.destination_type in [ + NotificationDestinationType.GENERIC, + NotificationDestinationType.SLACK, + NotificationDestinationType.MICROSOFT_TEAMS, + ]: + if not self.url: + errors.append("URL is required for this destination type") + + # Trigger validation + for trigger in self.triggers: + if not isinstance(trigger, NotificationTriggerType): + errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] + + return errors + + +class NotificationConfigurationUpdateOptions: + """Represents the options for updating an existing notification configuration.""" + + # Type annotations for instance attributes + enabled: bool | None + name: str | None + token: str | None + triggers: list[NotificationTriggerType] | None + url: str | None + email_addresses: list[str] | None + email_users: list[Any] | None + + def __init__( + self, + enabled: bool | None = None, + name: str | None = None, + token: str | None = None, + triggers: list[NotificationTriggerType] | None = None, + url: str | None = None, + email_addresses: list[str] | None = None, + email_users: list[Any] | None = None, + ): + self.enabled = enabled + self.name = name + self.token = token + self.triggers = triggers + self.url = url + self.email_addresses = email_addresses + self.email_users = email_users + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary for API requests.""" + data: dict[str, Any] = {"type": "notification-configurations", "attributes": {}} + + # Add only specified attributes + if self.enabled is not None: + data["attributes"]["enabled"] = self.enabled + + if self.name is not None: + data["attributes"]["name"] = self.name + + if self.token is not None: + data["attributes"]["token"] = self.token + + if self.triggers is not None: + data["attributes"]["triggers"] = [ + trigger.value for trigger in self.triggers + ] + + if self.url is not None: + data["attributes"]["url"] = self.url + + if self.email_addresses is not None: + data["attributes"]["email-addresses"] = self.email_addresses + + # Handle relationships + if self.email_users is not None: + data["relationships"] = data.get("relationships", {}) + data["relationships"]["users"] = { + "data": [ + { + "type": "users", + "id": user.id if hasattr(user, "id") else str(user), + } + for user in self.email_users + ] + } + + return data + + def validate(self) -> list[str]: + """Validate the update options and return any errors.""" + errors = [] + + # Name validation (if provided) + if self.name is not None and (not self.name or not self.name.strip()): + errors.append("Name cannot be empty") + + # Trigger validation (if provided) + if self.triggers is not None: + for trigger in self.triggers: + if not isinstance(trigger, NotificationTriggerType): + errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] + + return errors + + +class NotificationConfigurationList: + """Represents a list of notification configurations with pagination.""" + + # Type annotations for instance attributes + items: list[NotificationConfiguration] + current_page: int + page_size: int + prev_page: int | None + next_page: int | None + total_pages: int + total_count: int + + def __init__(self, data: dict[str, Any]): + self.items = [ + NotificationConfiguration(item.get("attributes", {})) + for item in data.get("data", []) + ] + + # Pagination metadata + meta = data.get("meta", {}) + pagination = meta.get("pagination", {}) + + self.current_page = pagination.get("current-page", 0) + self.page_size = pagination.get("page-size", 20) + self.prev_page = pagination.get("prev-page") + self.next_page = pagination.get("next-page") + self.total_pages = pagination.get("total-pages", 0) + self.total_count = pagination.get("total-count", 0) + + def __len__(self) -> int: + return len(self.items) + + def __iter__(self) -> Any: + return iter(self.items) + + def __getitem__(self, index: int) -> NotificationConfiguration: + return self.items[index] + + def __repr__(self) -> str: + return f"NotificationConfigurationList(count={len(self.items)}, page={self.current_page}, total={self.total_count})" diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py new file mode 100644 index 00000000..4de32ea0 --- /dev/null +++ b/src/pytfe/resources/notification_configuration.py @@ -0,0 +1,216 @@ +""" +Notification Configuration Resources + +This module provides CRUD operations for Terraform Cloud/Enterprise notification configurations. +""" + +from __future__ import annotations + +from typing import Any + +from ..errors import ( + InvalidOrgError, + ValidationError, +) +from ..models.notification_configuration import ( + NotificationConfiguration, + NotificationConfigurationCreateOptions, + NotificationConfigurationList, + NotificationConfigurationListOptions, + NotificationConfigurationUpdateOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class NotificationConfigurations(_Service): + """Notification Configuration API for Terraform Enterprise.""" + + def list( + self, + subscribable_id: str, + options: NotificationConfigurationListOptions | None = None, + ) -> NotificationConfigurationList: + """List all notification configurations associated with a workspace or team.""" + if not valid_string_id(subscribable_id): + raise InvalidOrgError("Invalid subscribable ID") + + # Determine URL based on subscribable choice + if options and options.subscribable_choice and options.subscribable_choice.team: + url = f"/api/v2/teams/{subscribable_id}/notification-configurations" + else: + url = f"/api/v2/workspaces/{subscribable_id}/notification-configurations" + + params = options.to_dict() if options else None + + r = self.t.request("GET", url, params=params) + jd = r.json() + + items = [] + meta = jd.get("meta", {}) + pagination = meta.get("pagination", {}) + + for d in jd.get("data", []): + items.append(self._parse_notification_configuration(d)) + + return NotificationConfigurationList( + { + "data": [{"attributes": item.__dict__} for item in items], + "meta": {"pagination": pagination}, + } + ) + + def create( + self, subscribable_id: str, options: NotificationConfigurationCreateOptions + ) -> NotificationConfiguration: + """Create a new notification configuration.""" + if not valid_string_id(subscribable_id): + raise InvalidOrgError("Invalid subscribable ID provided") + + # Validate options + validation_errors = options.validate() + if validation_errors: + raise ValidationError( + f"Notification configuration validation failed: {', '.join(validation_errors)}" + ) + + # Determine URL based on subscribable choice + if options.subscribable_choice and options.subscribable_choice.team: + url = f"/api/v2/teams/{subscribable_id}/notification-configurations" + else: + url = f"/api/v2/workspaces/{subscribable_id}/notification-configurations" + + payload = {"data": options.to_dict()} + + try: + r = self.t.request("POST", url, json_body=payload) + jd = r.json() + + if "data" in jd: + return self._parse_notification_configuration(jd["data"]) + + raise ValidationError("Invalid response format from API") + except Exception as e: + # Enhance error messages for common scenarios + error_msg = str(e).lower() + if "verification failed" in error_msg and "404" in error_msg: + raise ValidationError( + "Webhook URL verification failed - check that the URL is reachable and accepts POST requests" + ) from e + elif "not found" in error_msg: + if "team" in url: + raise InvalidOrgError( + f"Team '{subscribable_id}' not found or teams not available in your plan" + ) from e + else: + raise InvalidOrgError( + f"Workspace '{subscribable_id}' not found" + ) from e + else: + raise + + def read(self, notification_config_id: str) -> NotificationConfiguration: + """Read a notification configuration by its ID.""" + if not valid_string_id(notification_config_id): + raise InvalidOrgError("Invalid notification configuration ID provided") + + url = f"/api/v2/notification-configurations/{notification_config_id}" + + try: + r = self.t.request("GET", url) + jd = r.json() + + if "data" in jd: + return self._parse_notification_configuration(jd["data"]) + + raise ValidationError("Invalid response format from API") + except Exception as e: + error_msg = str(e).lower() + if "not found" in error_msg: + raise InvalidOrgError( + f"Notification configuration '{notification_config_id}' not found" + ) from e + else: + raise + + def update( + self, + notification_config_id: str, + options: NotificationConfigurationUpdateOptions, + ) -> NotificationConfiguration: + """Update an existing notification configuration.""" + if not valid_string_id(notification_config_id): + raise InvalidOrgError("Invalid notification configuration ID") + + # Validate options + validation_errors = options.validate() + if validation_errors: + raise ValidationError(f"Invalid options: {', '.join(validation_errors)}") + + url = f"/api/v2/notification-configurations/{notification_config_id}" + + payload = {"data": options.to_dict()} + payload["data"]["id"] = notification_config_id + + r = self.t.request("PATCH", url, json_body=payload) + jd = r.json() + + if "data" in jd: + return self._parse_notification_configuration(jd["data"]) + + raise ValidationError("Invalid response format from API") + + def delete(self, notification_config_id: str) -> None: + """Delete a notification configuration by its ID.""" + if not valid_string_id(notification_config_id): + raise InvalidOrgError("Invalid notification configuration ID") + + url = f"/api/v2/notification-configurations/{notification_config_id}" + self.t.request("DELETE", url) + + def verify(self, notification_config_id: str) -> NotificationConfiguration: + """Verify a notification configuration by delivering a verification payload.""" + if not valid_string_id(notification_config_id): + raise InvalidOrgError("Invalid notification configuration ID provided") + + url = f"/api/v2/notification-configurations/{notification_config_id}/actions/verify" + + try: + r = self.t.request("POST", url, json_body={}) + jd = r.json() + + if "data" in jd: + return self._parse_notification_configuration(jd["data"]) + + raise ValidationError("Invalid response format from API") + except Exception as e: + error_msg = str(e).lower() + if "verification failed" in error_msg and "404" in error_msg: + raise ValidationError( + "Webhook verification failed: URL returned 404. Check that your webhook URL is correct and accessible." + ) from e + elif "not found" in error_msg: + raise InvalidOrgError( + f"Notification configuration '{notification_config_id}' not found" + ) from e + else: + raise + + def _parse_notification_configuration( + self, data: dict[str, Any] + ) -> NotificationConfiguration: + """Parse notification configuration data from API response.""" + attributes = data.get("attributes", {}) + attributes["id"] = data.get("id") + + # Handle relationships + relationships = data.get("relationships", {}) + if "subscribable" in relationships: + subscribable_data = relationships["subscribable"].get("data", {}) + attributes["subscribable-choice"] = subscribable_data + + if "users" in relationships: + users_data = relationships["users"].get("data", []) + attributes["email-users"] = users_data + + return NotificationConfiguration(attributes) diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py new file mode 100644 index 00000000..a85bba6e --- /dev/null +++ b/tests/units/test_notification_configuration.py @@ -0,0 +1,658 @@ +""" +Unit tests for Notification Configuration API. + +Tests all CRUD operations: List, Create, Read, Update, Delete, and Verify. +Based on the Go TFE notification_configuration_integration_test.go implementation. +""" + +from unittest.mock import Mock + +import pytest + +from pytfe.errors import InvalidOrgError, ValidationError +from pytfe.models.notification_configuration import ( + DeliveryResponse, + NotificationConfiguration, + NotificationConfigurationCreateOptions, + NotificationConfigurationList, + NotificationConfigurationListOptions, + NotificationConfigurationSubscribableChoice, + NotificationConfigurationUpdateOptions, + NotificationDestinationType, + NotificationTriggerType, +) +from pytfe.resources.notification_configuration import NotificationConfigurations + + +class TestNotificationConfigurations: + """Test suite for notification configuration operations.""" + + def setup_method(self): + """Set up test fixtures.""" + self.mock_transport = Mock() + self.notifications = NotificationConfigurations(self.mock_transport) + + # Sample notification configuration data + self.sample_nc_data = { + "id": "nc-123456789", + "attributes": { + "created-at": "2023-01-01T10:00:00Z", + "updated-at": "2023-01-01T10:00:00Z", + "destination-type": "generic", + "enabled": True, + "name": "Test Notification", + "token": "test-token", + "url": "https://example.com/webhook", + "triggers": ["run:created", "run:completed"], + "email-addresses": [], + "delivery-responses": [], + }, + "relationships": { + "subscribable": {"data": {"type": "workspaces", "id": "ws-123456789"}}, + "users": {"data": []}, + }, + } + + def test_list_workspace_notifications(self): + """Test listing notification configurations for a workspace.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": [self.sample_nc_data], + "meta": { + "pagination": { + "current-page": 1, + "page-size": 20, + "prev-page": None, + "next-page": None, + "total-pages": 1, + "total-count": 1, + } + }, + } + self.mock_transport.request.return_value = mock_response + + # Test list operation + workspace_id = "ws-123456789" + result = self.notifications.list(workspace_id) + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", + f"/api/v2/workspaces/{workspace_id}/notification-configurations", + params=None, + ) + + # Verify result + assert isinstance(result, NotificationConfigurationList) + assert len(result.items) == 1 + assert result.items[0].id == "nc-123456789" + assert result.items[0].name == "Test Notification" + + def test_list_team_notifications(self): + """Test listing notification configurations for a team.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": [self.sample_nc_data], + "meta": { + "pagination": {"current-page": 1, "page-size": 20, "total-count": 1} + }, + } + self.mock_transport.request.return_value = mock_response + + # Test list operation with team + team_id = "team-123456789" + team_choice = NotificationConfigurationSubscribableChoice(team={"id": team_id}) + options = NotificationConfigurationListOptions(subscribable_choice=team_choice) + + result = self.notifications.list(team_id, options) + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/teams/{team_id}/notification-configurations", params={} + ) + + # Verify result + assert isinstance(result, NotificationConfigurationList) + assert len(result.items) == 1 + + def test_list_with_pagination(self): + """Test listing with pagination options.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": [], + "meta": { + "pagination": {"current-page": 2, "page-size": 50, "total-count": 0} + }, + } + self.mock_transport.request.return_value = mock_response + + # Test with pagination + workspace_id = "ws-123456789" + options = NotificationConfigurationListOptions(page_number=2, page_size=50) + + self.notifications.list(workspace_id, options) + + # Verify API call with pagination + self.mock_transport.request.assert_called_once_with( + "GET", + f"/api/v2/workspaces/{workspace_id}/notification-configurations", + params={"page[number]": 2, "page[size]": 50}, + ) + + def test_list_invalid_id(self): + """Test list with invalid subscribable ID.""" + with pytest.raises(InvalidOrgError): + self.notifications.list("") + + def test_create_workspace_notification(self): + """Test creating a notification configuration for a workspace.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = {"data": self.sample_nc_data} + self.mock_transport.request.return_value = mock_response + + # Create options + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.GENERIC, + enabled=True, + name="Test Notification", + token="test-token", + url="https://example.com/webhook", + triggers=[ + NotificationTriggerType.CREATED, + NotificationTriggerType.COMPLETED, + ], + ) + + # Test create operation + workspace_id = "ws-123456789" + result = self.notifications.create(workspace_id, options) + + # Verify API call + call_args = self.mock_transport.request.call_args + assert call_args[0][0] == "POST" + assert ( + call_args[0][1] + == f"/api/v2/workspaces/{workspace_id}/notification-configurations" + ) + + payload = call_args[1]["json_body"] + assert payload["data"]["type"] == "notification-configurations" + assert payload["data"]["attributes"]["name"] == "Test Notification" + assert payload["data"]["attributes"]["destination-type"] == "generic" + + # Verify result + assert isinstance(result, NotificationConfiguration) + assert result.id == "nc-123456789" + assert result.name == "Test Notification" + + def test_create_team_notification(self): + """Test creating a notification configuration for a team.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = {"data": self.sample_nc_data} + self.mock_transport.request.return_value = mock_response + + # Create options with team choice + team_choice = NotificationConfigurationSubscribableChoice( + team={"id": "team-123456789"} + ) + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.SLACK, + enabled=False, + name="Team Slack Notification", + url="https://hooks.slack.com/webhook", + triggers=[NotificationTriggerType.CHANGE_REQUEST_CREATED], + subscribable_choice=team_choice, + ) + + # Test create operation + team_id = "team-123456789" + self.notifications.create(team_id, options) + + # Verify API call uses teams endpoint + call_args = self.mock_transport.request.call_args + assert call_args[0][1] == f"/api/v2/teams/{team_id}/notification-configurations" + + def test_create_email_notification(self): + """Test creating an email notification configuration.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = {"data": self.sample_nc_data} + self.mock_transport.request.return_value = mock_response + + # Create email notification options + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.EMAIL, + enabled=True, + name="Email Notification", + email_addresses=["admin@example.com"], + triggers=[NotificationTriggerType.ERRORED], + ) + + # Test create operation + workspace_id = "ws-123456789" + self.notifications.create(workspace_id, options) + + # Verify API call + call_args = self.mock_transport.request.call_args + payload = call_args[1]["json_body"] + assert payload["data"]["attributes"]["destination-type"] == "email" + assert payload["data"]["attributes"]["email-addresses"] == ["admin@example.com"] + + def test_create_validation_errors(self): + """Test create with validation errors.""" + # Test missing required fields + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.GENERIC, + enabled=True, + name="", # Empty name should fail validation + url="https://example.com", + ) + + with pytest.raises(ValidationError) as exc_info: + self.notifications.create("ws-123456789", options) + + assert "Name is required" in str(exc_info.value) + + def test_create_url_validation(self): + """Test create with URL validation for certain destination types.""" + # Test missing URL for generic destination + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.GENERIC, + enabled=True, + name="Test Notification", + # Missing URL + ) + + with pytest.raises(ValidationError) as exc_info: + self.notifications.create("ws-123456789", options) + + assert "URL is required" in str(exc_info.value) + + def test_create_invalid_id(self): + """Test create with invalid workspace ID.""" + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.GENERIC, + enabled=True, + name="Test Notification", + url="https://example.com", + ) + + with pytest.raises(InvalidOrgError): + self.notifications.create("", options) + + def test_read_notification_configuration(self): + """Test reading a notification configuration by ID.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = {"data": self.sample_nc_data} + self.mock_transport.request.return_value = mock_response + + # Test read operation + nc_id = "nc-123456789" + result = self.notifications.read(nc_id) + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/notification-configurations/{nc_id}" + ) + + # Verify result + assert isinstance(result, NotificationConfiguration) + assert result.id == "nc-123456789" + assert result.name == "Test Notification" + assert result.enabled is True + + def test_read_invalid_id(self): + """Test read with invalid notification configuration ID.""" + with pytest.raises(InvalidOrgError): + self.notifications.read("") + + def test_update_notification_configuration(self): + """Test updating a notification configuration.""" + # Mock API response + updated_data = self.sample_nc_data.copy() + updated_data["attributes"]["name"] = "Updated Notification" + updated_data["attributes"]["enabled"] = False + + mock_response = Mock() + mock_response.json.return_value = {"data": updated_data} + self.mock_transport.request.return_value = mock_response + + # Update options + options = NotificationConfigurationUpdateOptions( + name="Updated Notification", enabled=False + ) + + # Test update operation + nc_id = "nc-123456789" + result = self.notifications.update(nc_id, options) + + # Verify API call + call_args = self.mock_transport.request.call_args + assert call_args[0][0] == "PATCH" + assert call_args[0][1] == f"/api/v2/notification-configurations/{nc_id}" + + payload = call_args[1]["json_body"] + assert payload["data"]["id"] == nc_id + assert payload["data"]["attributes"]["name"] == "Updated Notification" + assert payload["data"]["attributes"]["enabled"] is False + + # Verify result + assert isinstance(result, NotificationConfiguration) + assert result.name == "Updated Notification" + assert result.enabled is False + + def test_update_triggers(self): + """Test updating notification triggers.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = {"data": self.sample_nc_data} + self.mock_transport.request.return_value = mock_response + + # Update triggers + options = NotificationConfigurationUpdateOptions( + triggers=[ + NotificationTriggerType.PLANNING, + NotificationTriggerType.APPLYING, + NotificationTriggerType.ERRORED, + ] + ) + + # Test update operation + nc_id = "nc-123456789" + self.notifications.update(nc_id, options) + + # Verify API call includes triggers + call_args = self.mock_transport.request.call_args + payload = call_args[1]["json_body"] + expected_triggers = ["run:planning", "run:applying", "run:errored"] + assert payload["data"]["attributes"]["triggers"] == expected_triggers + + def test_update_validation_errors(self): + """Test update with validation errors.""" + # Test empty name + options = NotificationConfigurationUpdateOptions(name="") + + with pytest.raises(ValidationError) as exc_info: + self.notifications.update("nc-123456789", options) + + assert "Name cannot be empty" in str(exc_info.value) + + def test_update_invalid_id(self): + """Test update with invalid notification configuration ID.""" + options = NotificationConfigurationUpdateOptions(name="New Name") + + with pytest.raises(InvalidOrgError): + self.notifications.update("", options) + + def test_delete_notification_configuration(self): + """Test deleting a notification configuration.""" + # Mock API response (DELETE returns no content) + mock_response = Mock() + self.mock_transport.request.return_value = mock_response + + # Test delete operation + nc_id = "nc-123456789" + self.notifications.delete(nc_id) + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "DELETE", f"/api/v2/notification-configurations/{nc_id}" + ) + + def test_delete_invalid_id(self): + """Test delete with invalid notification configuration ID.""" + with pytest.raises(InvalidOrgError): + self.notifications.delete("") + + def test_verify_notification_configuration(self): + """Test verifying a notification configuration.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = {"data": self.sample_nc_data} + self.mock_transport.request.return_value = mock_response + + # Test verify operation + nc_id = "nc-123456789" + result = self.notifications.verify(nc_id) + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "POST", + f"/api/v2/notification-configurations/{nc_id}/actions/verify", + json_body={}, + ) + + # Verify result + assert isinstance(result, NotificationConfiguration) + assert result.id == "nc-123456789" + + def test_verify_invalid_id(self): + """Test verify with invalid notification configuration ID.""" + with pytest.raises(InvalidOrgError): + self.notifications.verify("") + + +class TestNotificationConfigurationModels: + """Test suite for notification configuration models.""" + + def test_notification_configuration_parsing(self): + """Test parsing notification configuration from API data.""" + data = { + "id": "nc-123456789", + "created-at": "2023-01-01T10:00:00Z", + "updated-at": "2023-01-01T10:00:00Z", + "destination-type": "slack", + "enabled": True, + "name": "Slack Notification", + "token": "token-123", + "url": "https://hooks.slack.com/webhook", + "triggers": ["run:created", "run:errored"], + "email-addresses": [], + "delivery-responses": [ + { + "body": "OK", + "code": "200", + "headers": {}, + "sent-at": "2023-01-01T10:00:00Z", + "successful": "true", + "url": "https://hooks.slack.com/webhook", + } + ], + } + + nc = NotificationConfiguration(data) + + assert nc.id == "nc-123456789" + assert nc.name == "Slack Notification" + assert nc.enabled is True + assert nc.destination_type == "slack" + assert len(nc.triggers) == 2 + assert NotificationTriggerType.CREATED in nc.triggers + assert NotificationTriggerType.ERRORED in nc.triggers + assert len(nc.delivery_responses) == 1 + assert nc.delivery_responses[0].code == "200" + + def test_delivery_response_parsing(self): + """Test parsing delivery response data.""" + data = { + "body": "Success", + "code": "200", + "headers": {"Content-Type": ["application/json"]}, + "sent-at": "2023-01-01T10:00:00Z", + "successful": "true", + "url": "https://example.com/webhook", + } + + dr = DeliveryResponse(data) + + assert dr.body == "Success" + assert dr.code == "200" + assert dr.headers == {"Content-Type": ["application/json"]} + assert dr.successful == "true" + assert dr.url == "https://example.com/webhook" + assert dr.sent_at is not None + + def test_create_options_validation(self): + """Test validation of create options.""" + # Valid options + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.GENERIC, + enabled=True, + name="Test Notification", + url="https://example.com", + ) + errors = options.validate() + assert len(errors) == 0 + + # Invalid options - missing name + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.GENERIC, + enabled=True, + name="", + url="https://example.com", + ) + errors = options.validate() + assert "Name is required" in errors + + # Invalid options - missing URL for generic destination + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.GENERIC, + enabled=True, + name="Test Notification", + ) + errors = options.validate() + assert "URL is required for this destination type" in errors + + def test_update_options_validation(self): + """Test validation of update options.""" + # Valid options + options = NotificationConfigurationUpdateOptions( + name="Updated Name", enabled=False + ) + errors = options.validate() + assert len(errors) == 0 + + # Invalid options - empty name + options = NotificationConfigurationUpdateOptions(name="") + errors = options.validate() + assert "Name cannot be empty" in errors + + def test_trigger_type_enum(self): + """Test notification trigger type enum values.""" + assert NotificationTriggerType.CREATED.value == "run:created" + assert NotificationTriggerType.PLANNING.value == "run:planning" + assert NotificationTriggerType.NEEDS_ATTENTION.value == "run:needs_attention" + assert NotificationTriggerType.APPLYING.value == "run:applying" + assert NotificationTriggerType.COMPLETED.value == "run:completed" + assert NotificationTriggerType.ERRORED.value == "run:errored" + assert NotificationTriggerType.ASSESSMENT_DRIFTED.value == "assessment:drifted" + assert NotificationTriggerType.ASSESSMENT_FAILED.value == "assessment:failed" + assert ( + NotificationTriggerType.ASSESSMENT_CHECK_FAILED.value + == "assessment:check_failure" + ) + assert ( + NotificationTriggerType.WORKSPACE_AUTO_DESTROY_REMINDER.value + == "workspace:auto_destroy_reminder" + ) + assert ( + NotificationTriggerType.WORKSPACE_AUTO_DESTROY_RUN_RESULTS.value + == "workspace:auto_destroy_run_results" + ) + assert ( + NotificationTriggerType.CHANGE_REQUEST_CREATED.value + == "change_request:created" + ) + + def test_destination_type_enum(self): + """Test notification destination type enum values.""" + assert NotificationDestinationType.EMAIL.value == "email" + assert NotificationDestinationType.GENERIC.value == "generic" + assert NotificationDestinationType.SLACK.value == "slack" + assert NotificationDestinationType.MICROSOFT_TEAMS.value == "microsoft-teams" + + def test_subscribable_choice(self): + """Test notification configuration subscribable choice.""" + # Test workspace choice + workspace_choice = NotificationConfigurationSubscribableChoice( + workspace={"id": "ws-123456789"} + ) + assert workspace_choice.workspace == {"id": "ws-123456789"} + assert workspace_choice.team is None + + # Test team choice + team_choice = NotificationConfigurationSubscribableChoice( + team={"id": "team-123456789"} + ) + assert team_choice.team == {"id": "team-123456789"} + assert team_choice.workspace is None + + def test_notification_configuration_list(self): + """Test notification configuration list parsing.""" + data = { + "data": [ + {"attributes": {"id": "nc-1", "name": "NC 1"}}, + {"attributes": {"id": "nc-2", "name": "NC 2"}}, + ], + "meta": { + "pagination": { + "current-page": 1, + "page-size": 20, + "total-pages": 1, + "total-count": 2, + } + }, + } + + nc_list = NotificationConfigurationList(data) + + assert len(nc_list.items) == 2 + assert nc_list.current_page == 1 + assert nc_list.total_count == 2 + assert nc_list.items[0].name == "NC 1" + assert nc_list.items[1].name == "NC 2" + + def test_list_options_to_dict(self): + """Test list options conversion to dictionary.""" + options = NotificationConfigurationListOptions(page_number=2, page_size=50) + result = options.to_dict() + + assert result == {"page[number]": 2, "page[size]": 50} + + def test_create_options_to_dict(self): + """Test create options conversion to dictionary.""" + options = NotificationConfigurationCreateOptions( + destination_type=NotificationDestinationType.SLACK, + enabled=True, + name="Slack Notification", + url="https://hooks.slack.com/webhook", + triggers=[NotificationTriggerType.CREATED, NotificationTriggerType.ERRORED], + ) + result = options.to_dict() + + assert result["type"] == "notification-configurations" + assert result["attributes"]["destination-type"] == "slack" + assert result["attributes"]["enabled"] is True + assert result["attributes"]["name"] == "Slack Notification" + assert result["attributes"]["url"] == "https://hooks.slack.com/webhook" + assert result["attributes"]["triggers"] == ["run:created", "run:errored"] + + def test_update_options_to_dict(self): + """Test update options conversion to dictionary.""" + options = NotificationConfigurationUpdateOptions( + name="Updated Name", + enabled=False, + triggers=[NotificationTriggerType.COMPLETED], + ) + result = options.to_dict() + + assert result["type"] == "notification-configurations" + assert result["attributes"]["name"] == "Updated Name" + assert result["attributes"]["enabled"] is False + assert result["attributes"]["triggers"] == ["run:completed"]