diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index 789367f7..8086d3e6 100644 --- a/examples/notification_configuration.py +++ b/examples/notification_configuration.py @@ -32,9 +32,15 @@ def main(): print("=== Python TFE Notification Configuration Example ===\n") - # Resolve workspace and team from environment (fallback to demo placeholders) - workspace_id = os.getenv("TFE_WORKSPACE_ID", "ws-example123456789") - workspace_name = os.getenv("TFE_ORG", "your-workspace-name") + # Resolve organization and workspace from environment variables + org_name = os.environ["TFE_ORG"] + workspace_name = os.getenv("TFE_WORKSPACE_NAME", "test-api") + workspace_id = os.getenv("TFE_WORKSPACE_ID", "") + if not workspace_id: + print(f"Looking up workspace '{workspace_name}' in org '{org_name}'...") + ws = client.workspaces.read(workspace_name, organization=org_name) + workspace_id = ws.id + print(f"Resolved workspace ID: {workspace_id}") print(f"Using workspace: {workspace_name} (ID: {workspace_id})") team_id = os.getenv("TFE_TEAM_ID", "team-example123456789") @@ -47,13 +53,12 @@ def main(): # ===== List notification configurations for workspace ===== print("1. Listing notification configurations for workspace...") try: - workspace_notifications = client.notification_configurations.list( + workspace_iter = client.notification_configurations.list( subscribable_id=workspace_id ) - print( - f"Found {len(workspace_notifications.items)} notification configurations" - ) - for nc in workspace_notifications.items: + workspace_notifications = list(workspace_iter) + print(f"Found {len(workspace_notifications)} notification configurations") + for nc in workspace_notifications: print(f"- {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})") except Exception as e: print(f"Error listing workspace notifications: {e}") @@ -69,13 +74,12 @@ def main(): options = NotificationConfigurationListOptions( subscribable_choice=team_choice ) - team_notifications = client.notification_configurations.list( + team_iter = 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: + team_notifications = list(team_iter) + print(f"Found {len(team_notifications)} team notification configurations") + for nc in team_notifications: print(f"- {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})") except Exception as e: error_msg = str(e).lower() @@ -93,16 +97,20 @@ def main(): workspace_choice = NotificationConfigurationSubscribableChoice( workspace={"id": workspace_id} ) - slack_url = os.getenv( + # Use GENERIC destination type with a URL that returns HTTP 200. + # SLACK/MICROSOFT_TEAMS destinations are auto-verified by HCP Terraform + # at creation time; a fake Slack URL returns 302 and causes the create + # call to fail immediately. GENERIC webhooks + httpbin always succeed. + webhook_url = os.getenv( "WEBHOOK_URL", - "https://hooks.slack.com/services/YOUR_SLACK_WORKSPACE/YOUR_CHANNEL/YOUR_WEBHOOK_TOKEN", + "https://httpbin.org/status/200", ) create_options = NotificationConfigurationCreateOptions( - destination_type=NotificationDestinationType.SLACK, + destination_type=NotificationDestinationType.GENERIC, enabled=True, - name="Python TFE Example Slack Notification", + name="Python TFE Example Generic Notification", subscribable_choice=workspace_choice, - url=slack_url, + url=webhook_url, triggers=[ NotificationTriggerType.COMPLETED, NotificationTriggerType.ERRORED, @@ -175,12 +183,17 @@ def main(): 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") + if "verification failed" in error_msg and ( + "404" in error_msg or "302" in error_msg + ): + print("Webhook verification failed (expected with fake URL)") + print( + "The URL returned a non-200 response - this is normal for testing" + ) + print("To test real verification, use a webhook from webhook.site,") + print( + "Slack, Teams, or Discord, or set WEBHOOK_URL=https://httpbin.org/status/200" + ) else: print(f" Error in workspace notification operations: {e}") diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index 0632a1e8..2e22d486 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -188,17 +188,14 @@ 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 @@ -206,8 +203,6 @@ 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 diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py index 4de32ea0..17341c76 100644 --- a/src/pytfe/resources/notification_configuration.py +++ b/src/pytfe/resources/notification_configuration.py @@ -6,6 +6,7 @@ from __future__ import annotations +from collections.abc import Iterator from typing import Any from ..errors import ( @@ -15,7 +16,6 @@ from ..models.notification_configuration import ( NotificationConfiguration, NotificationConfigurationCreateOptions, - NotificationConfigurationList, NotificationConfigurationListOptions, NotificationConfigurationUpdateOptions, ) @@ -30,35 +30,21 @@ def list( self, subscribable_id: str, options: NotificationConfigurationListOptions | None = None, - ) -> NotificationConfigurationList: + ) -> Iterator[NotificationConfiguration]: """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" + path = f"/api/v2/teams/{subscribable_id}/notification-configurations" else: - url = f"/api/v2/workspaces/{subscribable_id}/notification-configurations" + path = 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}, - } - ) + for d in self._list(path, params=params): + yield self._parse_notification_configuration(d) def create( self, subscribable_id: str, options: NotificationConfigurationCreateOptions diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py index 034d161e..3da8de97 100644 --- a/tests/units/test_notification_configuration.py +++ b/tests/units/test_notification_configuration.py @@ -73,20 +73,28 @@ def test_list_workspace_notifications(self): # Test list operation workspace_id = "ws-123456789" - result = self.notifications.list(workspace_id) + result_iter = self.notifications.list(workspace_id) + items = list(result_iter) - # Verify API call - self.mock_transport.request.assert_called_once_with( - "GET", - f"/api/v2/workspaces/{workspace_id}/notification-configurations", - params=None, + # Verify API call (occurs when iterator is consumed) + self.mock_transport.request.assert_called_once() + call_args = self.mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert ( + call_args[0][1] + == f"/api/v2/workspaces/{workspace_id}/notification-configurations" ) + params = call_args[1].get("params") + assert isinstance(params, dict) + assert "page[number]" in params and "page[size]" in params + assert params["page[number]"] == 1 + assert params["page[size]"] == 100 # 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" + assert len(items) == 1 + assert isinstance(items[0], NotificationConfiguration) + assert items[0].id == "nc-123456789" + assert items[0].name == "Test Notification" def test_list_team_notifications(self): """Test listing notification configurations for a team.""" @@ -105,16 +113,23 @@ def test_list_team_notifications(self): team_choice = NotificationConfigurationSubscribableChoice(team={"id": team_id}) options = NotificationConfigurationListOptions(subscribable_choice=team_choice) - result = self.notifications.list(team_id, options) + result_iter = self.notifications.list(team_id, options) + items = list(result_iter) - # Verify API call - self.mock_transport.request.assert_called_once_with( - "GET", f"/api/v2/teams/{team_id}/notification-configurations", params={} - ) + # Verify API call (occurs when iterator is consumed) + self.mock_transport.request.assert_called_once() + call_args = self.mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert call_args[0][1] == f"/api/v2/teams/{team_id}/notification-configurations" + params = call_args[1].get("params") + assert isinstance(params, dict) + assert "page[number]" in params and "page[size]" in params + assert params["page[number]"] == 1 + assert params["page[size]"] == 100 # Verify result - assert isinstance(result, NotificationConfigurationList) - assert len(result.items) == 1 + assert len(items) == 1 + assert isinstance(items[0], NotificationConfiguration) def test_list_with_pagination(self): """Test listing with pagination options.""" @@ -130,21 +145,29 @@ def test_list_with_pagination(self): # Test with pagination workspace_id = "ws-123456789" - options = NotificationConfigurationListOptions(page_number=2, page_size=50) + options = NotificationConfigurationListOptions(page_size=50) - self.notifications.list(workspace_id, options) + result_iter = self.notifications.list(workspace_id, options) + _ = list(result_iter) - # 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}, + # page_size from options is respected by _list(); page[number] is controlled by _list() + self.mock_transport.request.assert_called_once() + call_args = self.mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert ( + call_args[0][1] + == f"/api/v2/workspaces/{workspace_id}/notification-configurations" ) + params = call_args[1].get("params") + assert isinstance(params, dict) + assert "page[number]" in params and "page[size]" in params + assert params["page[number]"] == 1 + assert params["page[size]"] == 50 def test_list_invalid_id(self): """Test list with invalid subscribable ID.""" with pytest.raises(InvalidOrgError): - self.notifications.list("") + list(self.notifications.list("")) def test_create_workspace_notification(self): """Test creating a notification configuration for a workspace.""" @@ -619,10 +642,10 @@ def test_notification_configuration_list(self): def test_list_options_to_dict(self): """Test list options conversion to dictionary.""" - options = NotificationConfigurationListOptions(page_number=2, page_size=50) + options = NotificationConfigurationListOptions(page_size=50) result = options.to_dict() - assert result == {"page[number]": 2, "page[size]": 50} + assert result == {"page[size]": 50} def test_create_options_to_dict(self): """Test create options conversion to dictionary."""