From 8c95423c975c719c8bc2b3241dd091156d4cae93 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Mon, 13 Oct 2025 14:06:48 +0530 Subject: [PATCH 1/9] feat: Add notification configuration API support - Support workspace and team notification configurations - Add comprehensive unit tests with 31 test cases - Include example usage with sanitized webhook URLs - Based on Go TFE notification_configuration implementation --- examples/notification_configuration.py | 281 ++++++++ src/pytfe/client.py | 2 + .../models/notification_configuration.py | 390 +++++++++++ .../resources/notification_configuration.py | 217 ++++++ src/tfe/models/__init__.py | 336 +++++++++ .../units/test_notification_configuration.py | 658 ++++++++++++++++++ 6 files changed, 1884 insertions(+) create mode 100644 examples/notification_configuration.py create mode 100644 src/pytfe/models/notification_configuration.py create mode 100644 src/pytfe/resources/notification_configuration.py create mode 100644 src/tfe/models/__init__.py create mode 100644 tests/units/test_notification_configuration.py diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py new file mode 100644 index 0000000..fe089fd --- /dev/null +++ b/examples/notification_configuration.py @@ -0,0 +1,281 @@ +#!/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. + +Based on the Go TFE notification_configuration.go implementation. +""" + +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 8af2a38..0fda776 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 0000000..8d37767 --- /dev/null +++ b/src/pytfe/models/notification_configuration.py @@ -0,0 +1,390 @@ +""" +Notification Configuration Models + +This module provides models for working with Terraform Cloud/Enterprise notification configurations. +Based on the Go TFE notification_configuration.go implementation. +""" + +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.""" + + 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): + 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.""" + + def __init__(self, team: Any | None = None, workspace: Any | None = None): + self.team = team + self.workspace = workspace + + def __repr__(self): + 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.""" + + 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): + return f"NotificationConfiguration(id='{self.id}', name='{self.name}', enabled={self.enabled})" + + +class NotificationConfigurationListOptions: + """Represents the options for listing notification configurations.""" + + 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.""" + + 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 = { + "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") + + # 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}") + + return errors + + +class NotificationConfigurationUpdateOptions: + """Represents the options for updating an existing notification configuration.""" + + 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 = {"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}") + + return errors + + +class NotificationConfigurationList: + """Represents a list of notification configurations with pagination.""" + + 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): + return len(self.items) + + def __iter__(self): + return iter(self.items) + + def __getitem__(self, index): + return self.items[index] + + def __repr__(self): + 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 0000000..a4a860f --- /dev/null +++ b/src/pytfe/resources/notification_configuration.py @@ -0,0 +1,217 @@ +""" +Notification Configuration Resources + +This module provides CRUD operations for Terraform Cloud/Enterprise notification configurations. +Based on the Go TFE notification_configuration.go implementation. +""" + +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/src/tfe/models/__init__.py b/src/tfe/models/__init__.py new file mode 100644 index 0000000..2ab5d03 --- /dev/null +++ b/src/tfe/models/__init__.py @@ -0,0 +1,336 @@ +"""Types package for TFE client.""" + +# Import all types from the main types module by using importlib to avoid circular imports +import importlib.util +import os + +# Re-export all agent and agent pool types +from .agent import ( + Agent, + AgentListOptions, + AgentPool, + AgentPoolAllowedWorkspacePolicy, + AgentPoolAssignToWorkspacesOptions, + AgentPoolCreateOptions, + AgentPoolListOptions, + AgentPoolReadOptions, + AgentPoolRemoveFromWorkspacesOptions, + AgentPoolUpdateOptions, + AgentReadOptions, + AgentStatus, + AgentToken, + AgentTokenCreateOptions, + AgentTokenListOptions, +) + +# Re-export all configuration version types +from .configuration_version_types import ( + ConfigurationSource, + ConfigurationStatus, + ConfigurationVersion, + ConfigurationVersionCreateOptions, + ConfigurationVersionList, + ConfigurationVersionListOptions, + ConfigurationVersionReadOptions, + ConfigurationVersionUpload, + ConfigVerIncludeOpt, + IngressAttributes, +) + +# Re-export all notification configuration types +from .notification_configuration import ( + DeliveryResponse, + NotificationConfiguration, + NotificationConfigurationCreateOptions, + NotificationConfigurationList, + NotificationConfigurationListOptions, + NotificationConfigurationSubscribableChoice, + NotificationConfigurationUpdateOptions, + NotificationDestinationType, + NotificationTriggerType, +) + +# Re-export all OAuth client types +from .oauth_client import ( + OAuthClient, + OAuthClientAddProjectsOptions, + OAuthClientCreateOptions, + OAuthClientIncludeOpt, + OAuthClientList, + OAuthClientListOptions, + OAuthClientReadOptions, + OAuthClientRemoveProjectsOptions, + OAuthClientUpdateOptions, + ServiceProviderType, +) + +# Re-export all OAuth token types +from .oauth_token import ( + OAuthToken, + OAuthTokenList, + OAuthTokenListOptions, + OAuthTokenUpdateOptions, +) + +# Re-export all query run types +from .query_run import ( + QueryRun, + QueryRunCancelOptions, + QueryRunCreateOptions, + QueryRunForceCancelOptions, + QueryRunList, + QueryRunListOptions, + QueryRunLogs, + QueryRunReadOptions, + QueryRunResults, + QueryRunStatus, + QueryRunType, +) + +# Re-export all registry module types +from .registry_module_types import ( + AgentExecutionMode, + Commit, + CommitList, + Input, + Output, + ProviderDependency, + PublishingMechanism, + RegistryModule, + RegistryModuleCreateOptions, + RegistryModuleCreateVersionOptions, + RegistryModuleCreateWithVCSConnectionOptions, + RegistryModuleID, + RegistryModuleList, + RegistryModuleListIncludeOpt, + RegistryModuleListOptions, + RegistryModulePermissions, + RegistryModuleStatus, + RegistryModuleUpdateOptions, + RegistryModuleVCSRepo, + RegistryModuleVCSRepoOptions, + RegistryModuleVCSRepoUpdateOptions, + RegistryModuleVersion, + RegistryModuleVersionStatus, + RegistryModuleVersionStatuses, + RegistryName, + Resource, + Root, + TerraformRegistryModule, + TestConfig, +) + +# Re-export all registry provider types +from .registry_provider_types import ( + RegistryProvider, + RegistryProviderCreateOptions, + RegistryProviderID, + RegistryProviderIncludeOps, + RegistryProviderList, + RegistryProviderListOptions, + RegistryProviderPermissions, + RegistryProviderReadOptions, +) + +# Re-export all reserved tag key types +from .reserved_tag_key import ( + ReservedTagKey, + ReservedTagKeyCreateOptions, + ReservedTagKeyList, + ReservedTagKeyListOptions, + ReservedTagKeyUpdateOptions, +) + +# Re-export all SSH key types +from .ssh_key import ( + SSHKey, + SSHKeyCreateOptions, + SSHKeyList, + SSHKeyListOptions, + SSHKeyUpdateOptions, +) + +# Define what should be available when importing with * +__all__ = [ + # Notification configuration types + "DeliveryResponse", + "NotificationConfiguration", + "NotificationConfigurationCreateOptions", + "NotificationConfigurationList", + "NotificationConfigurationListOptions", + "NotificationConfigurationSubscribableChoice", + "NotificationConfigurationUpdateOptions", + "NotificationDestinationType", + "NotificationTriggerType", + # OAuth client types + "OAuthClient", + "OAuthClientAddProjectsOptions", + "OAuthClientCreateOptions", + "OAuthClientIncludeOpt", + "OAuthClientList", + "OAuthClientListOptions", + "OAuthClientReadOptions", + "OAuthClientRemoveProjectsOptions", + "OAuthClientUpdateOptions", + "ServiceProviderType", + # OAuth token types + "OAuthToken", + "OAuthTokenList", + "OAuthTokenListOptions", + "OAuthTokenUpdateOptions", + # SSH key types + "SSHKey", + "SSHKeyCreateOptions", + "SSHKeyList", + "SSHKeyListOptions", + "SSHKeyUpdateOptions", + # Reserved tag key types + "ReservedTagKey", + "ReservedTagKeyCreateOptions", + "ReservedTagKeyList", + "ReservedTagKeyListOptions", + "ReservedTagKeyUpdateOptions", + # Agent and agent pool types + "Agent", + "AgentPool", + "AgentPoolAllowedWorkspacePolicy", + "AgentPoolAssignToWorkspacesOptions", + "AgentPoolCreateOptions", + "AgentPoolListOptions", + "AgentPoolReadOptions", + "AgentPoolRemoveFromWorkspacesOptions", + "AgentPoolUpdateOptions", + "AgentStatus", + "AgentListOptions", + "AgentReadOptions", + "AgentToken", + "AgentTokenCreateOptions", + "AgentTokenListOptions", + # Configuration version types + "ConfigurationSource", + "ConfigurationStatus", + "ConfigurationVersion", + "ConfigurationVersionCreateOptions", + "ConfigurationVersionList", + "ConfigurationVersionListOptions", + "ConfigurationVersionReadOptions", + "ConfigurationVersionUpload", + "ConfigVerIncludeOpt", + "IngressAttributes", + # Registry module types + "AgentExecutionMode", + "Commit", + "CommitList", + "Input", + "Output", + "ProviderDependency", + "PublishingMechanism", + "RegistryModule", + "RegistryModuleCreateOptions", + "RegistryModuleCreateVersionOptions", + "RegistryModuleCreateWithVCSConnectionOptions", + "RegistryModuleID", + "RegistryModuleList", + "RegistryModuleListIncludeOpt", + "RegistryModuleListOptions", + "RegistryModulePermissions", + "RegistryModuleStatus", + "RegistryModuleUpdateOptions", + "RegistryModuleVCSRepo", + "RegistryModuleVCSRepoOptions", + "RegistryModuleVCSRepoUpdateOptions", + "RegistryModuleVersion", + "RegistryModuleVersionStatus", + "RegistryModuleVersionStatuses", + "RegistryName", + "Resource", + "Root", + "TestConfig", + "TerraformRegistryModule", + # Registry provider types + "RegistryProvider", + "RegistryProviderCreateOptions", + "RegistryProviderID", + "RegistryProviderIncludeOps", + "RegistryProviderList", + "RegistryProviderListOptions", + "RegistryProviderPermissions", + "RegistryProviderReadOptions", + # Query run types + "QueryRun", + "QueryRunCancelOptions", + "QueryRunCreateOptions", + "QueryRunForceCancelOptions", + "QueryRunList", + "QueryRunListOptions", + "QueryRunLogs", + "QueryRunReadOptions", + "QueryRunResults", + "QueryRunStatus", + "QueryRunType", + # Main types from types.py (will be dynamically added below) + "Capacity", + "DataRetentionPolicy", + "DataRetentionPolicyChoice", + "DataRetentionPolicyDeleteOlder", + "DataRetentionPolicyDeleteOlderSetOptions", + "DataRetentionPolicyDontDelete", + "DataRetentionPolicyDontDeleteSetOptions", + "DataRetentionPolicySetOptions", + "EffectiveTagBinding", + "Entitlements", + "ExecutionMode", + "LockedByChoice", + "Organization", + "OrganizationCreateOptions", + "OrganizationUpdateOptions", + "Pagination", + "Project", + "ReadRunQueueOptions", + "Run", + "RunQueue", + "RunStatus", + "Tag", + "TagBinding", + "TagList", + "Variable", + "VariableCreateOptions", + "VariableListOptions", + "VariableUpdateOptions", + "VCSRepo", + "Workspace", + "WorkspaceActions", + "WorkspaceAddRemoteStateConsumersOptions", + "WorkspaceAddTagBindingsOptions", + "WorkspaceAddTagsOptions", + "WorkspaceAssignSSHKeyOptions", + "WorkspaceCreateOptions", + "WorkspaceIncludeOpt", + "WorkspaceList", + "WorkspaceListOptions", + "WorkspaceListRemoteStateConsumersOptions", + "WorkspaceLockOptions", + "WorkspaceOutputs", + "WorkspacePermissions", + "WorkspaceReadOptions", + "WorkspaceRemoveRemoteStateConsumersOptions", + "WorkspaceRemoveTagsOptions", + "WorkspaceRemoveVCSConnectionOptions", + "WorkspaceSettingOverwrites", + "WorkspaceSource", + "WorkspaceTagListOptions", + "WorkspaceUpdateOptions", + "WorkspaceUpdateRemoteStateConsumersOptions", +] + +# Load the main types.py file that's at the same level as this types/ directory +types_py_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "types.py") +spec = importlib.util.spec_from_file_location("main_types", types_py_path) +if spec is not None and spec.loader is not None: + main_types = importlib.util.module_from_spec(spec) + spec.loader.exec_module(main_types) + + # Re-export all main types + for name in dir(main_types): + if not name.startswith("_"): + globals()[name] = getattr(main_types, name) diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py new file mode 100644 index 0000000..a85bba6 --- /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"] From 218beabb27c7e8ed8550fa241605e8c9e0f44da9 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 9 Oct 2025 16:11:22 +0530 Subject: [PATCH 2/9] Fix MyPy type annotations for notification configuration - Add explicit type annotations for all class attributes - Update to modern Python type syntax (list instead of List, | None instead of Optional) - Fix Collection[str] issues by adding explicit type annotations - All 31 unit tests still passing - Ruff compliance maintained - Only 3 MyPy unreachable warnings remain (false positives) --- .../models/notification_configuration.py | 80 ++++++++++++++++--- src/pytfe/utils.py | 2 +- 2 files changed, 72 insertions(+), 10 deletions(-) diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index 8d37767..d3ed859 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -48,6 +48,14 @@ class NotificationDestinationType(Enum): 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", "") @@ -65,18 +73,22 @@ def _parse_datetime(self, date_str: str | None) -> datetime | None: except (ValueError, AttributeError): return None - def __repr__(self): + 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): + def __repr__(self) -> str: if self.team: return f"NotificationConfigurationSubscribableChoice(team={self.team})" elif self.workspace: @@ -87,6 +99,22 @@ def __repr__(self): 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")) @@ -153,13 +181,18 @@ def _parse_subscribable_choice( team=team, workspace=workspace ) - def __repr__(self): + 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, @@ -185,6 +218,17 @@ def to_dict(self) -> dict[str, Any]: 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, @@ -212,7 +256,7 @@ def __init__( def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" - data = { + data: dict[str, Any] = { "type": "notification-configurations", "attributes": { "destination-type": self.destination_type.value, @@ -282,6 +326,15 @@ def validate(self) -> list[str]: 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, @@ -302,7 +355,7 @@ def __init__( def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" - data = {"type": "notification-configurations", "attributes": {}} + data: dict[str, Any] = {"type": "notification-configurations", "attributes": {}} # Add only specified attributes if self.enabled is not None: @@ -360,6 +413,15 @@ def validate(self) -> list[str]: 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", {})) @@ -377,14 +439,14 @@ def __init__(self, data: dict[str, Any]): self.total_pages = pagination.get("total-pages", 0) self.total_count = pagination.get("total-count", 0) - def __len__(self): + def __len__(self) -> int: return len(self.items) - def __iter__(self): + def __iter__(self) -> Any: return iter(self.items) - def __getitem__(self, index): + def __getitem__(self, index: int) -> NotificationConfiguration: return self.items[index] - def __repr__(self): + def __repr__(self) -> str: return f"NotificationConfigurationList(count={len(self.items)}, page={self.current_page}, total={self.total_count})" diff --git a/src/pytfe/utils.py b/src/pytfe/utils.py index 9adf58d..bab0eb0 100644 --- a/src/pytfe/utils.py +++ b/src/pytfe/utils.py @@ -16,7 +16,7 @@ from urllib.parse import urlparse try: - import slug # type: ignore[import-not-found] + import slug except ImportError: slug = None From aec5b53f90e92401653b25b8e97120864c2c77f4 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 9 Oct 2025 16:21:18 +0530 Subject: [PATCH 3/9] Python TFE - Notifications implementation --- src/pytfe/models/notification_configuration.py | 6 +++--- src/pytfe/utils.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index d3ed859..11824e7 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -304,7 +304,7 @@ def validate(self) -> list[str]: errors.append("Name is required") if not isinstance(self.enabled, bool): - errors.append("Enabled must be a boolean") + errors.append("Enabled must be a boolean") # type: ignore[unreachable] # URL validation for certain destination types if self.destination_type in [ @@ -318,7 +318,7 @@ def validate(self) -> list[str]: # Trigger validation for trigger in self.triggers: if not isinstance(trigger, NotificationTriggerType): - errors.append(f"Invalid trigger type: {trigger}") + errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] return errors @@ -405,7 +405,7 @@ def validate(self) -> list[str]: if self.triggers is not None: for trigger in self.triggers: if not isinstance(trigger, NotificationTriggerType): - errors.append(f"Invalid trigger type: {trigger}") + errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] return errors diff --git a/src/pytfe/utils.py b/src/pytfe/utils.py index bab0eb0..9adf58d 100644 --- a/src/pytfe/utils.py +++ b/src/pytfe/utils.py @@ -16,7 +16,7 @@ from urllib.parse import urlparse try: - import slug + import slug # type: ignore[import-not-found] except ImportError: slug = None From a93a64a5f6b3ab0d56d7a438dc52ba0cb53fa314 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Mon, 13 Oct 2025 14:19:08 +0530 Subject: [PATCH 4/9] Remove Go TFE references from notification configuration - Clean up documentation to remove Go TFE implementation references - Make codebase more self-contained and focused on Python implementation - Update all notification configuration files --- examples/notification_configuration.py | 2 -- src/pytfe/models/notification_configuration.py | 1 - src/pytfe/resources/notification_configuration.py | 1 - tests/units/test_notification_configuration.py | 1 - 4 files changed, 5 deletions(-) diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index fe089fd..c02545d 100644 --- a/examples/notification_configuration.py +++ b/examples/notification_configuration.py @@ -4,8 +4,6 @@ This example demonstrates how to use the Python TFE library to manage notification configurations for workspaces and teams. - -Based on the Go TFE notification_configuration.go implementation. """ import os diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index 11824e7..0632a1e 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -2,7 +2,6 @@ Notification Configuration Models This module provides models for working with Terraform Cloud/Enterprise notification configurations. -Based on the Go TFE notification_configuration.go implementation. """ from __future__ import annotations diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py index a4a860f..4de32ea 100644 --- a/src/pytfe/resources/notification_configuration.py +++ b/src/pytfe/resources/notification_configuration.py @@ -2,7 +2,6 @@ Notification Configuration Resources This module provides CRUD operations for Terraform Cloud/Enterprise notification configurations. -Based on the Go TFE notification_configuration.go implementation. """ from __future__ import annotations diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py index a85bba6..034d161 100644 --- a/tests/units/test_notification_configuration.py +++ b/tests/units/test_notification_configuration.py @@ -2,7 +2,6 @@ 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 From 72166a801c39cbf17ae8b43961f83b5802c25a16 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Mon, 13 Oct 2025 14:19:36 +0530 Subject: [PATCH 5/9] Revert "Remove Go TFE references from notification configuration" This reverts commit a93a64a5f6b3ab0d56d7a438dc52ba0cb53fa314. --- examples/notification_configuration.py | 2 ++ src/pytfe/models/notification_configuration.py | 1 + src/pytfe/resources/notification_configuration.py | 1 + tests/units/test_notification_configuration.py | 1 + 4 files changed, 5 insertions(+) diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index c02545d..fe089fd 100644 --- a/examples/notification_configuration.py +++ b/examples/notification_configuration.py @@ -4,6 +4,8 @@ This example demonstrates how to use the Python TFE library to manage notification configurations for workspaces and teams. + +Based on the Go TFE notification_configuration.go implementation. """ import os diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index 0632a1e..11824e7 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -2,6 +2,7 @@ Notification Configuration Models This module provides models for working with Terraform Cloud/Enterprise notification configurations. +Based on the Go TFE notification_configuration.go implementation. """ from __future__ import annotations diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py index 4de32ea..a4a860f 100644 --- a/src/pytfe/resources/notification_configuration.py +++ b/src/pytfe/resources/notification_configuration.py @@ -2,6 +2,7 @@ Notification Configuration Resources This module provides CRUD operations for Terraform Cloud/Enterprise notification configurations. +Based on the Go TFE notification_configuration.go implementation. """ from __future__ import annotations diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py index 034d161..a85bba6 100644 --- a/tests/units/test_notification_configuration.py +++ b/tests/units/test_notification_configuration.py @@ -2,6 +2,7 @@ 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 From 9681f1c8c2cf8ffa79e519809ae5c669dcdb942a Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Mon, 13 Oct 2025 14:23:10 +0530 Subject: [PATCH 6/9] Pythin TFE - Notification Rebased --- src/tfe/models/notification_configuration.py | 451 ++++++++++++++++++ .../resources/notification_configuration.py | 216 +++++++++ 2 files changed, 667 insertions(+) create mode 100644 src/tfe/models/notification_configuration.py create mode 100644 src/tfe/resources/notification_configuration.py diff --git a/src/tfe/models/notification_configuration.py b/src/tfe/models/notification_configuration.py new file mode 100644 index 0000000..0632a1e --- /dev/null +++ b/src/tfe/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/tfe/resources/notification_configuration.py b/src/tfe/resources/notification_configuration.py new file mode 100644 index 0000000..4de32ea --- /dev/null +++ b/src/tfe/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) From 97fcbe860b6c527f5c1003ea19bc615890a0bf77 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Mon, 13 Oct 2025 14:36:12 +0530 Subject: [PATCH 7/9] Pythin TFE - Notification Rebased --- examples/notification_configuration.py | 2 -- src/pytfe/models/notification_configuration.py | 1 - src/pytfe/resources/notification_configuration.py | 1 - 3 files changed, 4 deletions(-) diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index fe089fd..c02545d 100644 --- a/examples/notification_configuration.py +++ b/examples/notification_configuration.py @@ -4,8 +4,6 @@ This example demonstrates how to use the Python TFE library to manage notification configurations for workspaces and teams. - -Based on the Go TFE notification_configuration.go implementation. """ import os diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index 11824e7..0632a1e 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -2,7 +2,6 @@ Notification Configuration Models This module provides models for working with Terraform Cloud/Enterprise notification configurations. -Based on the Go TFE notification_configuration.go implementation. """ from __future__ import annotations diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py index a4a860f..4de32ea 100644 --- a/src/pytfe/resources/notification_configuration.py +++ b/src/pytfe/resources/notification_configuration.py @@ -2,7 +2,6 @@ Notification Configuration Resources This module provides CRUD operations for Terraform Cloud/Enterprise notification configurations. -Based on the Go TFE notification_configuration.go implementation. """ from __future__ import annotations From 8d1a82ed621e25725279827022d57869cac0cb03 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Wed, 15 Oct 2025 14:15:01 +0530 Subject: [PATCH 8/9] Pythin TFE - Notification Rebased --- src/tfe/models/__init__.py | 13 +- src/tfe/models/notification_configuration.py | 451 ------------------ .../resources/notification_configuration.py | 216 --------- 3 files changed, 1 insertion(+), 679 deletions(-) delete mode 100644 src/tfe/models/notification_configuration.py delete mode 100644 src/tfe/resources/notification_configuration.py diff --git a/src/tfe/models/__init__.py b/src/tfe/models/__init__.py index 2ab5d03..64b23d0 100644 --- a/src/tfe/models/__init__.py +++ b/src/tfe/models/__init__.py @@ -37,18 +37,7 @@ IngressAttributes, ) -# Re-export all notification configuration types -from .notification_configuration import ( - DeliveryResponse, - NotificationConfiguration, - NotificationConfigurationCreateOptions, - NotificationConfigurationList, - NotificationConfigurationListOptions, - NotificationConfigurationSubscribableChoice, - NotificationConfigurationUpdateOptions, - NotificationDestinationType, - NotificationTriggerType, -) + # Re-export all OAuth client types from .oauth_client import ( diff --git a/src/tfe/models/notification_configuration.py b/src/tfe/models/notification_configuration.py deleted file mode 100644 index 0632a1e..0000000 --- a/src/tfe/models/notification_configuration.py +++ /dev/null @@ -1,451 +0,0 @@ -""" -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/tfe/resources/notification_configuration.py b/src/tfe/resources/notification_configuration.py deleted file mode 100644 index 4de32ea..0000000 --- a/src/tfe/resources/notification_configuration.py +++ /dev/null @@ -1,216 +0,0 @@ -""" -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) From b1cf98ae27e4f1d1545ed9064ae37efc91784f34 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Wed, 15 Oct 2025 14:15:08 +0530 Subject: [PATCH 9/9] Pythin TFE - Notification Rebased --- src/tfe/models/__init__.py | 325 ------------------------------------- 1 file changed, 325 deletions(-) delete mode 100644 src/tfe/models/__init__.py diff --git a/src/tfe/models/__init__.py b/src/tfe/models/__init__.py deleted file mode 100644 index 64b23d0..0000000 --- a/src/tfe/models/__init__.py +++ /dev/null @@ -1,325 +0,0 @@ -"""Types package for TFE client.""" - -# Import all types from the main types module by using importlib to avoid circular imports -import importlib.util -import os - -# Re-export all agent and agent pool types -from .agent import ( - Agent, - AgentListOptions, - AgentPool, - AgentPoolAllowedWorkspacePolicy, - AgentPoolAssignToWorkspacesOptions, - AgentPoolCreateOptions, - AgentPoolListOptions, - AgentPoolReadOptions, - AgentPoolRemoveFromWorkspacesOptions, - AgentPoolUpdateOptions, - AgentReadOptions, - AgentStatus, - AgentToken, - AgentTokenCreateOptions, - AgentTokenListOptions, -) - -# Re-export all configuration version types -from .configuration_version_types import ( - ConfigurationSource, - ConfigurationStatus, - ConfigurationVersion, - ConfigurationVersionCreateOptions, - ConfigurationVersionList, - ConfigurationVersionListOptions, - ConfigurationVersionReadOptions, - ConfigurationVersionUpload, - ConfigVerIncludeOpt, - IngressAttributes, -) - - - -# Re-export all OAuth client types -from .oauth_client import ( - OAuthClient, - OAuthClientAddProjectsOptions, - OAuthClientCreateOptions, - OAuthClientIncludeOpt, - OAuthClientList, - OAuthClientListOptions, - OAuthClientReadOptions, - OAuthClientRemoveProjectsOptions, - OAuthClientUpdateOptions, - ServiceProviderType, -) - -# Re-export all OAuth token types -from .oauth_token import ( - OAuthToken, - OAuthTokenList, - OAuthTokenListOptions, - OAuthTokenUpdateOptions, -) - -# Re-export all query run types -from .query_run import ( - QueryRun, - QueryRunCancelOptions, - QueryRunCreateOptions, - QueryRunForceCancelOptions, - QueryRunList, - QueryRunListOptions, - QueryRunLogs, - QueryRunReadOptions, - QueryRunResults, - QueryRunStatus, - QueryRunType, -) - -# Re-export all registry module types -from .registry_module_types import ( - AgentExecutionMode, - Commit, - CommitList, - Input, - Output, - ProviderDependency, - PublishingMechanism, - RegistryModule, - RegistryModuleCreateOptions, - RegistryModuleCreateVersionOptions, - RegistryModuleCreateWithVCSConnectionOptions, - RegistryModuleID, - RegistryModuleList, - RegistryModuleListIncludeOpt, - RegistryModuleListOptions, - RegistryModulePermissions, - RegistryModuleStatus, - RegistryModuleUpdateOptions, - RegistryModuleVCSRepo, - RegistryModuleVCSRepoOptions, - RegistryModuleVCSRepoUpdateOptions, - RegistryModuleVersion, - RegistryModuleVersionStatus, - RegistryModuleVersionStatuses, - RegistryName, - Resource, - Root, - TerraformRegistryModule, - TestConfig, -) - -# Re-export all registry provider types -from .registry_provider_types import ( - RegistryProvider, - RegistryProviderCreateOptions, - RegistryProviderID, - RegistryProviderIncludeOps, - RegistryProviderList, - RegistryProviderListOptions, - RegistryProviderPermissions, - RegistryProviderReadOptions, -) - -# Re-export all reserved tag key types -from .reserved_tag_key import ( - ReservedTagKey, - ReservedTagKeyCreateOptions, - ReservedTagKeyList, - ReservedTagKeyListOptions, - ReservedTagKeyUpdateOptions, -) - -# Re-export all SSH key types -from .ssh_key import ( - SSHKey, - SSHKeyCreateOptions, - SSHKeyList, - SSHKeyListOptions, - SSHKeyUpdateOptions, -) - -# Define what should be available when importing with * -__all__ = [ - # Notification configuration types - "DeliveryResponse", - "NotificationConfiguration", - "NotificationConfigurationCreateOptions", - "NotificationConfigurationList", - "NotificationConfigurationListOptions", - "NotificationConfigurationSubscribableChoice", - "NotificationConfigurationUpdateOptions", - "NotificationDestinationType", - "NotificationTriggerType", - # OAuth client types - "OAuthClient", - "OAuthClientAddProjectsOptions", - "OAuthClientCreateOptions", - "OAuthClientIncludeOpt", - "OAuthClientList", - "OAuthClientListOptions", - "OAuthClientReadOptions", - "OAuthClientRemoveProjectsOptions", - "OAuthClientUpdateOptions", - "ServiceProviderType", - # OAuth token types - "OAuthToken", - "OAuthTokenList", - "OAuthTokenListOptions", - "OAuthTokenUpdateOptions", - # SSH key types - "SSHKey", - "SSHKeyCreateOptions", - "SSHKeyList", - "SSHKeyListOptions", - "SSHKeyUpdateOptions", - # Reserved tag key types - "ReservedTagKey", - "ReservedTagKeyCreateOptions", - "ReservedTagKeyList", - "ReservedTagKeyListOptions", - "ReservedTagKeyUpdateOptions", - # Agent and agent pool types - "Agent", - "AgentPool", - "AgentPoolAllowedWorkspacePolicy", - "AgentPoolAssignToWorkspacesOptions", - "AgentPoolCreateOptions", - "AgentPoolListOptions", - "AgentPoolReadOptions", - "AgentPoolRemoveFromWorkspacesOptions", - "AgentPoolUpdateOptions", - "AgentStatus", - "AgentListOptions", - "AgentReadOptions", - "AgentToken", - "AgentTokenCreateOptions", - "AgentTokenListOptions", - # Configuration version types - "ConfigurationSource", - "ConfigurationStatus", - "ConfigurationVersion", - "ConfigurationVersionCreateOptions", - "ConfigurationVersionList", - "ConfigurationVersionListOptions", - "ConfigurationVersionReadOptions", - "ConfigurationVersionUpload", - "ConfigVerIncludeOpt", - "IngressAttributes", - # Registry module types - "AgentExecutionMode", - "Commit", - "CommitList", - "Input", - "Output", - "ProviderDependency", - "PublishingMechanism", - "RegistryModule", - "RegistryModuleCreateOptions", - "RegistryModuleCreateVersionOptions", - "RegistryModuleCreateWithVCSConnectionOptions", - "RegistryModuleID", - "RegistryModuleList", - "RegistryModuleListIncludeOpt", - "RegistryModuleListOptions", - "RegistryModulePermissions", - "RegistryModuleStatus", - "RegistryModuleUpdateOptions", - "RegistryModuleVCSRepo", - "RegistryModuleVCSRepoOptions", - "RegistryModuleVCSRepoUpdateOptions", - "RegistryModuleVersion", - "RegistryModuleVersionStatus", - "RegistryModuleVersionStatuses", - "RegistryName", - "Resource", - "Root", - "TestConfig", - "TerraformRegistryModule", - # Registry provider types - "RegistryProvider", - "RegistryProviderCreateOptions", - "RegistryProviderID", - "RegistryProviderIncludeOps", - "RegistryProviderList", - "RegistryProviderListOptions", - "RegistryProviderPermissions", - "RegistryProviderReadOptions", - # Query run types - "QueryRun", - "QueryRunCancelOptions", - "QueryRunCreateOptions", - "QueryRunForceCancelOptions", - "QueryRunList", - "QueryRunListOptions", - "QueryRunLogs", - "QueryRunReadOptions", - "QueryRunResults", - "QueryRunStatus", - "QueryRunType", - # Main types from types.py (will be dynamically added below) - "Capacity", - "DataRetentionPolicy", - "DataRetentionPolicyChoice", - "DataRetentionPolicyDeleteOlder", - "DataRetentionPolicyDeleteOlderSetOptions", - "DataRetentionPolicyDontDelete", - "DataRetentionPolicyDontDeleteSetOptions", - "DataRetentionPolicySetOptions", - "EffectiveTagBinding", - "Entitlements", - "ExecutionMode", - "LockedByChoice", - "Organization", - "OrganizationCreateOptions", - "OrganizationUpdateOptions", - "Pagination", - "Project", - "ReadRunQueueOptions", - "Run", - "RunQueue", - "RunStatus", - "Tag", - "TagBinding", - "TagList", - "Variable", - "VariableCreateOptions", - "VariableListOptions", - "VariableUpdateOptions", - "VCSRepo", - "Workspace", - "WorkspaceActions", - "WorkspaceAddRemoteStateConsumersOptions", - "WorkspaceAddTagBindingsOptions", - "WorkspaceAddTagsOptions", - "WorkspaceAssignSSHKeyOptions", - "WorkspaceCreateOptions", - "WorkspaceIncludeOpt", - "WorkspaceList", - "WorkspaceListOptions", - "WorkspaceListRemoteStateConsumersOptions", - "WorkspaceLockOptions", - "WorkspaceOutputs", - "WorkspacePermissions", - "WorkspaceReadOptions", - "WorkspaceRemoveRemoteStateConsumersOptions", - "WorkspaceRemoveTagsOptions", - "WorkspaceRemoveVCSConnectionOptions", - "WorkspaceSettingOverwrites", - "WorkspaceSource", - "WorkspaceTagListOptions", - "WorkspaceUpdateOptions", - "WorkspaceUpdateRemoteStateConsumersOptions", -] - -# Load the main types.py file that's at the same level as this types/ directory -types_py_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "types.py") -spec = importlib.util.spec_from_file_location("main_types", types_py_path) -if spec is not None and spec.loader is not None: - main_types = importlib.util.module_from_spec(spec) - spec.loader.exec_module(main_types) - - # Re-export all main types - for name in dir(main_types): - if not name.startswith("_"): - globals()[name] = getattr(main_types, name)