Coverage for scripts / test_webhook_delivery.py: 100.00%

1#!/usr/bin/env python3 

2""" 

3Test script for webhook delivery. 

4 

5This script tests the webhook dispatcher by: 

61. Starting a local HTTP server to receive webhooks 

72. Creating a mock job event 

83. Dispatching the webhook 

94. Verifying the payload was received correctly 

10 

11Usage: 

12 python scripts/test_webhook_delivery.py 

13 

14For testing with a real webhook.site endpoint: 

15 python scripts/test_webhook_delivery.py --url https://webhook.site/your-uuid 

16""" 

17 

18import argparse 

19import asyncio 

20import hashlib 

21import hmac 

22import json 

23import sys 

24import threading 

25from datetime import UTC, datetime 

26from http.server import BaseHTTPRequestHandler, HTTPServer 

27from pathlib import Path 

28from typing import Any 

29from unittest.mock import MagicMock, patch 

30 

31# Add project root to path 

32sys.path.insert(0, str(Path(__file__).parent.parent)) 

33 

34# Store received webhooks 

35received_webhooks: list[dict[str, Any]] = [] 

36 

37 

38class WebhookHandler(BaseHTTPRequestHandler): 

39 """Simple HTTP handler to receive webhook requests.""" 

40 

41 def do_POST(self) -> None: 

42 content_length = int(self.headers.get("Content-Length", 0)) 

43 body = self.rfile.read(content_length) 

44 

45 webhook_data = { 

46 "path": self.path, 

47 "headers": dict(self.headers), 

48 "body": body.decode("utf-8"), 

49 "timestamp": datetime.now(UTC).isoformat(), 

50 } 

51 received_webhooks.append(webhook_data) 

52 

53 print(f"\n{'=' * 60}") 

54 print("WEBHOOK RECEIVED!") 

55 print(f"{'=' * 60}") 

56 print(f"Event: {self.headers.get('X-GCO-Event', 'unknown')}") 

57 print(f"Cluster: {self.headers.get('X-GCO-Cluster', 'unknown')}") 

58 print(f"Region: {self.headers.get('X-GCO-Region', 'unknown')}") 

59 sig = self.headers.get("X-GCO-Signature") 

60 if sig: 

61 print(f"Signature: {sig[:50]}...") 

62 

63 try: 

64 payload = json.loads(body) 

65 print("\nPayload:") 

66 print(json.dumps(payload, indent=2)) 

67 except json.JSONDecodeError: 

68 print(f"\nRaw body: {body.decode('utf-8')}") 

69 

70 print(f"{'=' * 60}\n") 

71 

72 self.send_response(200) 

73 self.send_header("Content-Type", "application/json") 

74 self.end_headers() 

75 self.wfile.write(b'{"status": "received"}') 

76 

77 def log_message(self, format: str, *args: Any) -> None: 

78 # Suppress default logging 

79 pass 

80 

81 

82def start_local_server(port: int = 8888) -> HTTPServer: 

83 """Start a local HTTP server to receive webhooks.""" 

84 server = HTTPServer(("localhost", port), WebhookHandler) 

85 thread = threading.Thread(target=server.serve_forever, daemon=True) 

86 thread.start() 

87 print(f"Local webhook server started on http://localhost:{port}") 

88 return server 

89 

90 

91def create_mock_job() -> Any: 

92 """Create a mock Kubernetes job object.""" 

93 job = MagicMock() 

94 job.metadata.name = "test-webhook-job" 

95 job.metadata.namespace = "gco-jobs" 

96 job.metadata.uid = "test-job-uid-12345" 

97 job.metadata.labels = {"app": "webhook-test", "team": "platform"} 

98 job.status.conditions = [MagicMock(type="Complete", status="True")] 

99 job.status.active = 0 

100 job.status.succeeded = 1 

101 job.status.failed = 0 

102 job.status.start_time = datetime(2026, 2, 4, 12, 0, 0, tzinfo=UTC) 

103 job.status.completion_time = datetime(2026, 2, 4, 12, 5, 0, tzinfo=UTC) 

104 return job 

105 

106 

107async def test_with_local_server() -> bool: 

108 """Test webhook delivery with a local server.""" 

109 from gco.services.webhook_dispatcher import WebhookDispatcher, WebhookEvent 

110 

111 print("\n" + "=" * 60) 

112 print("WEBHOOK DELIVERY TEST - LOCAL SERVER") 

113 print("=" * 60 + "\n") 

114 

115 # Start local server 

116 port = 8888 

117 server = start_local_server(port) 

118 webhook_url = f"http://localhost:{port}/webhook" 

119 webhook_secret = "test-secret-key" # nosec B105 — local test fixture, not a real secret 

120 

121 # Create mock webhook store 

122 mock_store = MagicMock() 

123 mock_store.get_webhooks_for_event.return_value = [ 

124 { 

125 "id": "test-webhook-1", 

126 "url": webhook_url, 

127 "events": ["job.completed"], 

128 "namespace": "gco-jobs", 

129 "secret": webhook_secret, 

130 } 

131 ] 

132 

133 # Create dispatcher with mocked K8s config 

134 with patch("gco.services.webhook_dispatcher.config") as mock_config: 

135 mock_config.ConfigException = Exception 

136 mock_config.load_incluster_config.side_effect = Exception() 

137 mock_config.load_kube_config.return_value = None 

138 

139 with patch("gco.services.webhook_dispatcher.client"): 

140 dispatcher = WebhookDispatcher( 

141 cluster_id="test-cluster", 

142 region="us-east-1", 

143 webhook_store=mock_store, 

144 timeout=10, 

145 max_retries=1, 

146 retry_delay=1, 

147 ) 

148 

149 # Create mock job 

150 job = create_mock_job() 

151 

152 print("Dispatching webhook for job.completed event...") 

153 print(f"Target URL: {webhook_url}") 

154 print("Secret configured: Yes") 

155 print() 

156 

157 # Dispatch the event 

158 results = await dispatcher._dispatch_event(WebhookEvent.JOB_COMPLETED, job) 

159 

160 # Check results 

161 print("\n" + "-" * 40) 

162 print("DELIVERY RESULTS:") 

163 print("-" * 40) 

164 

165 for result in results: 

166 status = "✓ SUCCESS" if result.success else "✗ FAILED" 

167 print(f"{status}") 

168 print(f" Webhook ID: {result.webhook_id}") 

169 print(f" URL: {result.url}") 

170 print(f" Status Code: {result.status_code}") 

171 print(f" Attempts: {result.attempts}") 

172 print(f" Duration: {result.duration_ms:.1f}ms") 

173 if result.error: 

174 print(f" Error: {result.error}") 

175 

176 # Verify signature 

177 if received_webhooks: 

178 print("\n" + "-" * 40) 

179 print("SIGNATURE VERIFICATION:") 

180 print("-" * 40) 

181 

182 webhook = received_webhooks[-1] 

183 signature = webhook["headers"].get("X-GCO-Signature", "") 

184 body = webhook["body"] 

185 

186 expected = hmac.new( 

187 webhook_secret.encode("utf-8"), 

188 body.encode("utf-8"), 

189 hashlib.sha256, 

190 ).hexdigest() 

191 

192 if signature == f"sha256={expected}": 

193 print("✓ Signature verified successfully!") 

194 else: 

195 print("✗ Signature verification failed!") 

196 print(f" Expected: sha256={expected}") 

197 print(f" Received: {signature}") 

198 

199 # Cleanup 

200 server.shutdown() 

201 

202 print("\n" + "=" * 60) 

203 print("TEST COMPLETE") 

204 print("=" * 60 + "\n") 

205 

206 return len(results) > 0 and all(r.success for r in results) 

207 

208 

209async def test_with_external_url(url: str, secret: str | None = None) -> bool: 

210 """Test webhook delivery with an external URL (e.g., webhook.site).""" 

211 from gco.services.webhook_dispatcher import WebhookDispatcher, WebhookEvent 

212 

213 print("\n" + "=" * 60) 

214 print("WEBHOOK DELIVERY TEST - EXTERNAL URL") 

215 print("=" * 60 + "\n") 

216 

217 print(f"Target URL: {url}") 

218 print(f"Secret configured: {'Yes' if secret else 'No'}") 

219 print() 

220 

221 # Create mock webhook store 

222 mock_store = MagicMock() 

223 webhook_config = { 

224 "id": "external-webhook", 

225 "url": url, 

226 "events": ["job.completed", "job.failed", "job.started"], 

227 "namespace": None, # Global webhook 

228 } 

229 if secret: 

230 webhook_config["secret"] = secret 

231 mock_store.get_webhooks_for_event.return_value = [webhook_config] 

232 

233 # Create dispatcher with mocked K8s config 

234 with patch("gco.services.webhook_dispatcher.config") as mock_config: 

235 mock_config.ConfigException = Exception 

236 mock_config.load_incluster_config.side_effect = Exception() 

237 mock_config.load_kube_config.return_value = None 

238 

239 with patch("gco.services.webhook_dispatcher.client"): 

240 dispatcher = WebhookDispatcher( 

241 cluster_id="gco-test-cluster", 

242 region="us-east-1", 

243 webhook_store=mock_store, 

244 timeout=30, 

245 max_retries=2, 

246 retry_delay=2, 

247 ) 

248 

249 # Create mock job 

250 job = create_mock_job() 

251 

252 # Test all three event types 

253 events = [ 

254 (WebhookEvent.JOB_STARTED, "job.started"), 

255 (WebhookEvent.JOB_COMPLETED, "job.completed"), 

256 (WebhookEvent.JOB_FAILED, "job.failed"), 

257 ] 

258 

259 all_success = True 

260 

261 for event, event_name in events: 

262 # Adjust job status for the event 

263 if event == WebhookEvent.JOB_STARTED: 

264 job.status.conditions = [] 

265 job.status.active = 1 

266 job.status.succeeded = 0 

267 job.status.completion_time = None 

268 elif event == WebhookEvent.JOB_COMPLETED: 

269 job.status.conditions = [MagicMock(type="Complete", status="True")] 

270 job.status.active = 0 

271 job.status.succeeded = 1 

272 job.status.completion_time = datetime(2026, 2, 4, 12, 5, 0, tzinfo=UTC) 

273 else: # WebhookEvent.JOB_FAILED — the last of the three events above 

274 job.status.conditions = [MagicMock(type="Failed", status="True")] 

275 job.status.active = 0 

276 job.status.succeeded = 0 

277 job.status.failed = 1 

278 job.status.completion_time = datetime(2026, 2, 4, 12, 5, 0, tzinfo=UTC) 

279 

280 print(f"Dispatching {event_name} event...") 

281 results = await dispatcher._dispatch_event(event, job) 

282 

283 for result in results: 

284 status = "✓" if result.success else "✗" 

285 print( 

286 f" {status} Status: {result.status_code}, " 

287 f"Attempts: {result.attempts}, " 

288 f"Duration: {result.duration_ms:.1f}ms" 

289 ) 

290 if not result.success: 

291 print(f" Error: {result.error}") 

292 all_success = False 

293 

294 # Small delay between events 

295 await asyncio.sleep(1) 

296 

297 print("\n" + "=" * 60) 

298 if all_success: 

299 print("ALL WEBHOOKS DELIVERED SUCCESSFULLY!") 

300 print(f"Check your webhook receiver at: {url}") 

301 else: 

302 print("SOME WEBHOOKS FAILED - Check errors above") 

303 print("=" * 60 + "\n") 

304 

305 return all_success 

306 

307 

308async def main() -> int: 

309 parser = argparse.ArgumentParser(description="Test webhook delivery") 

310 parser.add_argument( 

311 "--url", 

312 help="External webhook URL (e.g., https://webhook.site/your-uuid)", 

313 ) 

314 parser.add_argument( 

315 "--secret", 

316 help="HMAC secret for signature verification", 

317 ) 

318 parser.add_argument( 

319 "--local", 

320 action="store_true", 

321 help="Run test with local server (default if no URL provided)", 

322 ) 

323 

324 args = parser.parse_args() 

325 

326 if args.url: 

327 success = await test_with_external_url(args.url, args.secret) 

328 else: 

329 success = await test_with_local_server() 

330 

331 return 0 if success else 1 

332 

333 

334if __name__ == "__main__": 

335 sys.exit(asyncio.run(main()))