Coverage for cli / commands / webhooks_cmd.py: 96%

1"""Webhook commands.""" 

2 

3import sys 

4from typing import Any 

5 

6import click 

7 

8from ..config import GCOConfig 

9from ..output import get_output_formatter 

10 

11pass_config = click.make_pass_decorator(GCOConfig, ensure=True) 

12 

13 

14@click.group() 

15@pass_config 

16def webhooks(config: Any) -> None: 

17 """Manage webhooks for job event notifications. 

18 

19 Webhooks receive HTTP POST notifications when job events occur 

20 (job.started, job.completed, job.failed). 

21 """ 

22 pass 

23 

24 

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. 

31 

32 Examples: 

33 gco webhooks list 

34 gco webhooks list --namespace gco-jobs 

35 """ 

36 formatter = get_output_formatter(config) 

37 

38 try: 

39 from ..aws_client import get_aws_client 

40 

41 aws_client = get_aws_client(config) 

42 

43 query_region = region or config.default_region 

44 params = {} 

45 if namespace: 

46 params["namespace"] = namespace 

47 

48 result = aws_client.call_api( 

49 method="GET", 

50 path="/api/v1/webhooks", 

51 region=query_region, 

52 params=params, 

53 ) 

54 

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 

60 

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) 

75 

76 except Exception as e: 

77 formatter.print_error(f"Failed to list webhooks: {e}") 

78 sys.exit(1) 

79 

80 

81@webhooks.command("create") 

82@click.option("--url", "-u", required=True, help="Webhook URL") 

83@click.option( 

84 "--event", 

85 "-e", 

86 multiple=True, 

87 required=True, 

88 type=click.Choice(["job.started", "job.completed", "job.failed"]), 

89 help="Events to subscribe to", 

90) 

91@click.option("--namespace", "-n", help="Filter events by namespace") 

92@click.option("--secret", "-s", help="Secret for HMAC signature verification") 

93@click.option("--region", "-r", help="Region to use (any region works)") 

94@pass_config 

95def webhooks_create( 

96 config: Any, url: Any, event: Any, namespace: Any, secret: Any, region: Any 

97) -> None: 

98 """Register a new webhook for job events. 

99 

100 Examples: 

101 gco webhooks create --url https://example.com/webhook -e job.completed -e job.failed 

102 gco webhooks create -u https://slack.com/webhook -e job.failed -n gco-jobs 

103 """ 

104 formatter = get_output_formatter(config) 

105 

106 try: 

107 from ..aws_client import get_aws_client 

108 

109 aws_client = get_aws_client(config) 

110 

111 query_region = region or config.default_region 

112 result = aws_client.call_api( 

113 method="POST", 

114 path="/api/v1/webhooks", 

115 region=query_region, 

116 body={ 

117 "url": url, 

118 "events": list(event), 

119 "namespace": namespace, 

120 "secret": secret, 

121 }, 

122 ) 

123 

124 formatter.print_success("Webhook registered successfully") 

125 formatter.print(result) 

126 

127 except Exception as e: 

128 formatter.print_error(f"Failed to create webhook: {e}") 

129 sys.exit(1) 

130 

131 

132@webhooks.command("delete") 

133@click.argument("webhook_id") 

134@click.option("--region", "-r", help="Region to use (any region works)") 

135@click.option("--yes", "-y", is_flag=True, help="Skip confirmation") 

136@pass_config 

137def webhooks_delete(config: Any, webhook_id: Any, region: Any, yes: Any) -> None: 

138 """Delete a webhook. 

139 

140 Examples: 

141 gco webhooks delete abc123 

142 gco webhooks delete abc123 -y 

143 """ 

144 formatter = get_output_formatter(config) 

145 

146 if not yes: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 click.confirm(f"Delete webhook '{webhook_id}'?", abort=True) 

148 

149 try: 

150 from ..aws_client import get_aws_client 

151 

152 aws_client = get_aws_client(config) 

153 

154 query_region = region or config.default_region 

155 result = aws_client.call_api( 

156 method="DELETE", 

157 path=f"/api/v1/webhooks/{webhook_id}", 

158 region=query_region, 

159 ) 

160 

161 formatter.print_success(f"Webhook '{webhook_id}' deleted") 

162 formatter.print(result) 

163 

164 except Exception as e: 

165 formatter.print_error(f"Failed to delete webhook: {e}") 

166 sys.exit(1)