Skip to content

Commit 9306d7f

Browse files
isivaselvanisivaselvan
authored
Merge pull request #109 from TanyaSingh369-svg/refactor/iterator/notify-config
Refactor NotificationConfigurations.list() to use iterator pattern
2 parents 2d2cbd4 + c2b1341 commit 9306d7f

4 files changed

Lines changed: 93 additions & 76 deletions

File tree

‎examples/notification_configuration.py‎

Lines changed: 37 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -32,9 +32,15 @@ def main():
3232

3333
print("=== Python TFE Notification Configuration Example ===\n")
3434

35-
# Resolve workspace and team from environment (fallback to demo placeholders)
36-
workspace_id = os.getenv("TFE_WORKSPACE_ID", "ws-example123456789")
37-
workspace_name = os.getenv("TFE_ORG", "your-workspace-name")
35+
# Resolve organization and workspace from environment variables
36+
org_name = os.environ["TFE_ORG"]
37+
workspace_name = os.getenv("TFE_WORKSPACE_NAME", "test-api")
38+
workspace_id = os.getenv("TFE_WORKSPACE_ID", "")
39+
if not workspace_id:
40+
print(f"Looking up workspace '{workspace_name}' in org '{org_name}'...")
41+
ws = client.workspaces.read(workspace_name, organization=org_name)
42+
workspace_id = ws.id
43+
print(f"Resolved workspace ID: {workspace_id}")
3844
print(f"Using workspace: {workspace_name} (ID: {workspace_id})")
3945

4046
team_id = os.getenv("TFE_TEAM_ID", "team-example123456789")
@@ -47,13 +53,12 @@ def main():
4753
# ===== List notification configurations for workspace =====
4854
print("1. Listing notification configurations for workspace...")
4955
try:
50-
workspace_notifications = client.notification_configurations.list(
56+
workspace_iter = client.notification_configurations.list(
5157
subscribable_id=workspace_id
5258
)
53-
print(
54-
f"Found {len(workspace_notifications.items)} notification configurations"
55-
)
56-
for nc in workspace_notifications.items:
59+
workspace_notifications = list(workspace_iter)
60+
print(f"Found {len(workspace_notifications)} notification configurations")
61+
for nc in workspace_notifications:
5762
print(f"- {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})")
5863
except Exception as e:
5964
print(f"Error listing workspace notifications: {e}")
@@ -69,13 +74,12 @@ def main():
6974
options = NotificationConfigurationListOptions(
7075
subscribable_choice=team_choice
7176
)
72-
team_notifications = client.notification_configurations.list(
77+
team_iter = client.notification_configurations.list(
7378
subscribable_id=team_id, options=options
7479
)
75-
print(
76-
f"Found {len(team_notifications.items)} team notification configurations"
77-
)
78-
for nc in team_notifications.items:
80+
team_notifications = list(team_iter)
81+
print(f"Found {len(team_notifications)} team notification configurations")
82+
for nc in team_notifications:
7983
print(f"- {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})")
8084
except Exception as e:
8185
error_msg = str(e).lower()
@@ -93,16 +97,20 @@ def main():
9397
workspace_choice = NotificationConfigurationSubscribableChoice(
9498
workspace={"id": workspace_id}
9599
)
96-
slack_url = os.getenv(
100+
# Use GENERIC destination type with a URL that returns HTTP 200.
101+
# SLACK/MICROSOFT_TEAMS destinations are auto-verified by HCP Terraform
102+
# at creation time; a fake Slack URL returns 302 and causes the create
103+
# call to fail immediately. GENERIC webhooks + httpbin always succeed.
104+
webhook_url = os.getenv(
97105
"WEBHOOK_URL",
98-
"https://hooks.slack.com/services/YOUR_SLACK_WORKSPACE/YOUR_CHANNEL/YOUR_WEBHOOK_TOKEN",
106+
"https://httpbin.org/status/200",
99107
)
100108
create_options = NotificationConfigurationCreateOptions(
101-
destination_type=NotificationDestinationType.SLACK,
109+
destination_type=NotificationDestinationType.GENERIC,
102110
enabled=True,
103-
name="Python TFE Example Slack Notification",
111+
name="Python TFE Example Generic Notification",
104112
subscribable_choice=workspace_choice,
105-
url=slack_url,
113+
url=webhook_url,
106114
triggers=[
107115
NotificationTriggerType.COMPLETED,
108116
NotificationTriggerType.ERRORED,
@@ -175,12 +183,17 @@ def main():
175183

176184
except Exception as e:
177185
error_msg = str(e).lower()
178-
if "verification failed" in error_msg and "404" in error_msg:
179-
print(" Webhook verification failed (expected with fake URL)")
180-
print("The fake Slack URL returns 404 - this is normal for testing")
181-
print("To test real verification, use a webhook from:")
182-
print("webhook.site (instant test URL)")
183-
print("Slack, Teams, or Discord webhook")
186+
if "verification failed" in error_msg and (
187+
"404" in error_msg or "302" in error_msg
188+
):
189+
print("Webhook verification failed (expected with fake URL)")
190+
print(
191+
"The URL returned a non-200 response - this is normal for testing"
192+
)
193+
print("To test real verification, use a webhook from webhook.site,")
194+
print(
195+
"Slack, Teams, or Discord, or set WEBHOOK_URL=https://httpbin.org/status/200"
196+
)
184197
else:
185198
print(f" Error in workspace notification operations: {e}")
186199

‎src/pytfe/models/notification_configuration.py‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -188,26 +188,21 @@ class NotificationConfigurationListOptions:
188188
"""Represents the options for listing notification configurations."""
189189

190190
# Type annotations for instance attributes
191-
page_number: int | None
192191
page_size: int | None
193192
subscribable_choice: NotificationConfigurationSubscribableChoice | None
194193

195194
def __init__(
196195
self,
197-
page_number: int | None = None,
198196
page_size: int | None = None,
199197
subscribable_choice: NotificationConfigurationSubscribableChoice | None = None,
200198
):
201-
self.page_number = page_number
202199
self.page_size = page_size
203200
self.subscribable_choice = subscribable_choice
204201

205202
def to_dict(self) -> dict[str, Any]:
206203
"""Convert to dictionary for API requests."""
207204
params = {}
208205

209-
if self.page_number is not None:
210-
params["page[number]"] = self.page_number
211206
if self.page_size is not None:
212207
params["page[size]"] = self.page_size
213208

‎src/pytfe/resources/notification_configuration.py‎

Lines changed: 6 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
from __future__ import annotations
88

9+
from collections.abc import Iterator
910
from typing import Any
1011

1112
from ..errors import (
@@ -15,7 +16,6 @@
1516
from ..models.notification_configuration import (
1617
NotificationConfiguration,
1718
NotificationConfigurationCreateOptions,
18-
NotificationConfigurationList,
1919
NotificationConfigurationListOptions,
2020
NotificationConfigurationUpdateOptions,
2121
)
@@ -30,35 +30,21 @@ def list(
3030
self,
3131
subscribable_id: str,
3232
options: NotificationConfigurationListOptions | None = None,
33-
) -> NotificationConfigurationList:
33+
) -> Iterator[NotificationConfiguration]:
3434
"""List all notification configurations associated with a workspace or team."""
3535
if not valid_string_id(subscribable_id):
3636
raise InvalidOrgError("Invalid subscribable ID")
3737

3838
# Determine URL based on subscribable choice
3939
if options and options.subscribable_choice and options.subscribable_choice.team:
40-
url = f"/api/v2/teams/{subscribable_id}/notification-configurations"
40+
path = f"/api/v2/teams/{subscribable_id}/notification-configurations"
4141
else:
42-
url = f"/api/v2/workspaces/{subscribable_id}/notification-configurations"
42+
path = f"/api/v2/workspaces/{subscribable_id}/notification-configurations"
4343

4444
params = options.to_dict() if options else None
4545

46-
r = self.t.request("GET", url, params=params)
47-
jd = r.json()
48-
49-
items = []
50-
meta = jd.get("meta", {})
51-
pagination = meta.get("pagination", {})
52-
53-
for d in jd.get("data", []):
54-
items.append(self._parse_notification_configuration(d))
55-
56-
return NotificationConfigurationList(
57-
{
58-
"data": [{"attributes": item.__dict__} for item in items],
59-
"meta": {"pagination": pagination},
60-
}
61-
)
46+
for d in self._list(path, params=params):
47+
yield self._parse_notification_configuration(d)
6248

6349
def create(
6450
self, subscribable_id: str, options: NotificationConfigurationCreateOptions

‎tests/units/test_notification_configuration.py‎

Lines changed: 50 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -73,20 +73,28 @@ def test_list_workspace_notifications(self):
7373

7474
# Test list operation
7575
workspace_id = "ws-123456789"
76-
result = self.notifications.list(workspace_id)
76+
result_iter = self.notifications.list(workspace_id)
77+
items = list(result_iter)
7778

78-
# Verify API call
79-
self.mock_transport.request.assert_called_once_with(
80-
"GET",
81-
f"/api/v2/workspaces/{workspace_id}/notification-configurations",
82-
params=None,
79+
# Verify API call (occurs when iterator is consumed)
80+
self.mock_transport.request.assert_called_once()
81+
call_args = self.mock_transport.request.call_args
82+
assert call_args[0][0] == "GET"
83+
assert (
84+
call_args[0][1]
85+
== f"/api/v2/workspaces/{workspace_id}/notification-configurations"
8386
)
87+
params = call_args[1].get("params")
88+
assert isinstance(params, dict)
89+
assert "page[number]" in params and "page[size]" in params
90+
assert params["page[number]"] == 1
91+
assert params["page[size]"] == 100
8492

8593
# Verify result
86-
assert isinstance(result, NotificationConfigurationList)
87-
assert len(result.items) == 1
88-
assert result.items[0].id == "nc-123456789"
89-
assert result.items[0].name == "Test Notification"
94+
assert len(items) == 1
95+
assert isinstance(items[0], NotificationConfiguration)
96+
assert items[0].id == "nc-123456789"
97+
assert items[0].name == "Test Notification"
9098

9199
def test_list_team_notifications(self):
92100
"""Test listing notification configurations for a team."""
@@ -105,16 +113,23 @@ def test_list_team_notifications(self):
105113
team_choice = NotificationConfigurationSubscribableChoice(team={"id": team_id})
106114
options = NotificationConfigurationListOptions(subscribable_choice=team_choice)
107115

108-
result = self.notifications.list(team_id, options)
116+
result_iter = self.notifications.list(team_id, options)
117+
items = list(result_iter)
109118

110-
# Verify API call
111-
self.mock_transport.request.assert_called_once_with(
112-
"GET", f"/api/v2/teams/{team_id}/notification-configurations", params={}
113-
)
119+
# Verify API call (occurs when iterator is consumed)
120+
self.mock_transport.request.assert_called_once()
121+
call_args = self.mock_transport.request.call_args
122+
assert call_args[0][0] == "GET"
123+
assert call_args[0][1] == f"/api/v2/teams/{team_id}/notification-configurations"
124+
params = call_args[1].get("params")
125+
assert isinstance(params, dict)
126+
assert "page[number]" in params and "page[size]" in params
127+
assert params["page[number]"] == 1
128+
assert params["page[size]"] == 100
114129

115130
# Verify result
116-
assert isinstance(result, NotificationConfigurationList)
117-
assert len(result.items) == 1
131+
assert len(items) == 1
132+
assert isinstance(items[0], NotificationConfiguration)
118133

119134
def test_list_with_pagination(self):
120135
"""Test listing with pagination options."""
@@ -130,21 +145,29 @@ def test_list_with_pagination(self):
130145

131146
# Test with pagination
132147
workspace_id = "ws-123456789"
133-
options = NotificationConfigurationListOptions(page_number=2, page_size=50)
148+
options = NotificationConfigurationListOptions(page_size=50)
134149

135-
self.notifications.list(workspace_id, options)
150+
result_iter = self.notifications.list(workspace_id, options)
151+
_ = list(result_iter)
136152

137-
# Verify API call with pagination
138-
self.mock_transport.request.assert_called_once_with(
139-
"GET",
140-
f"/api/v2/workspaces/{workspace_id}/notification-configurations",
141-
params={"page[number]": 2, "page[size]": 50},
153+
# page_size from options is respected by _list(); page[number] is controlled by _list()
154+
self.mock_transport.request.assert_called_once()
155+
call_args = self.mock_transport.request.call_args
156+
assert call_args[0][0] == "GET"
157+
assert (
158+
call_args[0][1]
159+
== f"/api/v2/workspaces/{workspace_id}/notification-configurations"
142160
)
161+
params = call_args[1].get("params")
162+
assert isinstance(params, dict)
163+
assert "page[number]" in params and "page[size]" in params
164+
assert params["page[number]"] == 1
165+
assert params["page[size]"] == 50
143166

144167
def test_list_invalid_id(self):
145168
"""Test list with invalid subscribable ID."""
146169
with pytest.raises(InvalidOrgError):
147-
self.notifications.list("")
170+
list(self.notifications.list(""))
148171

149172
def test_create_workspace_notification(self):
150173
"""Test creating a notification configuration for a workspace."""
@@ -619,10 +642,10 @@ def test_notification_configuration_list(self):
619642

620643
def test_list_options_to_dict(self):
621644
"""Test list options conversion to dictionary."""
622-
options = NotificationConfigurationListOptions(page_number=2, page_size=50)
645+
options = NotificationConfigurationListOptions(page_size=50)
623646
result = options.to_dict()
624647

625-
assert result == {"page[number]": 2, "page[size]": 50}
648+
assert result == {"page[size]": 50}
626649

627650
def test_create_options_to_dict(self):
628651
"""Test create options conversion to dictionary."""

0 commit comments

Comments
 (0)