Coverage for cli/commands/webhooks_cmd.py: 94.74%
102 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Webhook commands."""
3import sys
4from typing import Any
6import click
8from ..config import GCOConfig
9from ..output import get_output_formatter
11pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
14@click.group()
15@pass_config
16def webhooks(config: Any) -> None:
17 """Manage webhooks for job event notifications.
19 Webhooks receive HTTP POST notifications when job events occur
20 (job.started, job.completed, job.failed).
21 """
22 pass
25@webhooks.command("list")
26@click.option("--namespace", "-n", help="Filter by namespace")
27@click.option("--region", "-r", help="Region to query (any region works)")
28@pass_config
29def webhooks_list(config: Any, namespace: Any, region: Any) -> None:
30 """List all registered webhooks.
32 Examples:
33 gco webhooks list
34 gco webhooks list --namespace gco-jobs
35 """
36 formatter = get_output_formatter(config)
38 try:
39 from ..aws_client import get_aws_client
41 aws_client = get_aws_client(config)
43 query_region = region or (config.default_region if config.use_regional_api else None)
44 params = {}
45 if namespace:
46 params["namespace"] = namespace
48 result = aws_client.call_api(
49 method="GET",
50 path="/api/v1/webhooks",
51 region=query_region,
52 params=params,
53 )
55 if config.output_format == "table": 55 ↛ 74line 55 didn't jump to line 74 because the condition on line 55 was always true
56 webhooks_data = result.get("webhooks", [])
57 if not webhooks_data:
58 formatter.print_info("No webhooks found")
59 return
61 print(f"\n Webhooks ({result.get('count', 0)} total)")
62 print(" " + "-" * 80)
63 print(
64 " ID URL EVENTS NAMESPACE"
65 )
66 print(" " + "-" * 80)
67 for w in webhooks_data:
68 wid = w.get("id", "")[:8]
69 url = w.get("url", "")[:40]
70 events = ",".join(w.get("events", []))[:18]
71 ns = (w.get("namespace") or "all")[:12]
72 print(f" {wid:<9} {url:<42} {events:<19} {ns}")
73 else:
74 formatter.print(result)
76 except Exception as e:
77 formatter.print_error(f"Failed to list webhooks: {e}")
78 sys.exit(1)
81@webhooks.command("get")
82@click.argument("webhook_id")
83@click.option("--region", "-r", help="Region to query (any region works)")
84@pass_config
85def webhooks_get(config: Any, webhook_id: Any, region: Any) -> None:
86 """Get a single webhook by id.
88 The webhooks API exposes list/create/delete but no fetch-by-id endpoint, so
89 this lists the region's webhooks and returns the one whose id matches.
91 Examples:
92 gco webhooks get abc12345
93 gco webhooks get abc12345 -r us-east-1
94 """
95 formatter = get_output_formatter(config)
97 try:
98 from ..aws_client import get_aws_client
100 aws_client = get_aws_client(config)
102 query_region = region or (config.default_region if config.use_regional_api else None)
103 result = aws_client.call_api(
104 method="GET",
105 path="/api/v1/webhooks",
106 region=query_region,
107 )
108 match = next(
109 (w for w in result.get("webhooks", []) if w.get("id") == webhook_id),
110 None,
111 )
112 if match is None: 112 ↛ 115line 112 didn't jump to line 115 because the condition on line 112 was always true
113 formatter.print_error(f"Webhook '{webhook_id}' not found")
114 sys.exit(1)
115 formatter.print(match)
117 except SystemExit:
118 raise
119 except Exception as e:
120 formatter.print_error(f"Failed to get webhook: {e}")
121 sys.exit(1)
124@webhooks.command("create")
125@click.option("--url", "-u", required=True, help="Webhook URL")
126@click.option(
127 "--event",
128 "-e",
129 multiple=True,
130 required=True,
131 type=click.Choice(["job.started", "job.completed", "job.failed"]),
132 help="Events to subscribe to",
133)
134@click.option("--namespace", "-n", help="Filter events by namespace")
135@click.option("--secret", "-s", help="Secret for HMAC signature verification")
136@click.option("--region", "-r", help="Region to use (any region works)")
137@pass_config
138def webhooks_create(
139 config: Any, url: Any, event: Any, namespace: Any, secret: Any, region: Any
140) -> None:
141 """Register a new webhook for job events.
143 Examples:
144 gco webhooks create --url https://example.com/webhook -e job.completed -e job.failed
145 gco webhooks create -u https://slack.com/webhook -e job.failed -n gco-jobs
146 """
147 formatter = get_output_formatter(config)
149 try:
150 from ..aws_client import get_aws_client
152 aws_client = get_aws_client(config)
154 query_region = region or (config.default_region if config.use_regional_api else None)
155 result = aws_client.call_api(
156 method="POST",
157 path="/api/v1/webhooks",
158 region=query_region,
159 body={
160 "url": url,
161 "events": list(event),
162 "namespace": namespace,
163 "secret": secret,
164 },
165 )
167 formatter.print_success("Webhook registered successfully")
168 formatter.print(result)
170 except Exception as e:
171 formatter.print_error(f"Failed to create webhook: {e}")
172 sys.exit(1)
175@webhooks.command("delete")
176@click.argument("webhook_id")
177@click.option("--region", "-r", help="Region to use (any region works)")
178@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
179@pass_config
180def webhooks_delete(config: Any, webhook_id: Any, region: Any, yes: Any) -> None:
181 """Delete a webhook.
183 Examples:
184 gco webhooks delete abc123
185 gco webhooks delete abc123 -y
186 """
187 formatter = get_output_formatter(config)
189 if not yes: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 click.confirm(f"Delete webhook '{webhook_id}'?", abort=True)
192 try:
193 from ..aws_client import get_aws_client
195 aws_client = get_aws_client(config)
197 query_region = region or (config.default_region if config.use_regional_api else None)
198 result = aws_client.call_api(
199 method="DELETE",
200 path=f"/api/v1/webhooks/{webhook_id}",
201 region=query_region,
202 )
204 formatter.print_success(f"Webhook '{webhook_id}' deleted")
205 formatter.print(result)
207 except Exception as e:
208 formatter.print_error(f"Failed to delete webhook: {e}")
209 sys.exit(1)