Coverage for scripts / test_webhook_delivery.py: 100.00%
179 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1#!/usr/bin/env python3
2"""
3Test script for webhook delivery.
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
11Usage:
12 python scripts/test_webhook_delivery.py
14For testing with a real webhook.site endpoint:
15 python scripts/test_webhook_delivery.py --url https://webhook.site/your-uuid
16"""
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
31# Add project root to path
32sys.path.insert(0, str(Path(__file__).parent.parent))
34# Store received webhooks
35received_webhooks: list[dict[str, Any]] = []
38class WebhookHandler(BaseHTTPRequestHandler):
39 """Simple HTTP handler to receive webhook requests."""
41 def do_POST(self) -> None:
42 content_length = int(self.headers.get("Content-Length", 0))
43 body = self.rfile.read(content_length)
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)
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]}...")
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')}")
70 print(f"{'=' * 60}\n")
72 self.send_response(200)
73 self.send_header("Content-Type", "application/json")
74 self.end_headers()
75 self.wfile.write(b'{"status": "received"}')
77 def log_message(self, format: str, *args: Any) -> None:
78 # Suppress default logging
79 pass
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
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
107async def test_with_local_server() -> bool:
108 """Test webhook delivery with a local server."""
109 from gco.services.webhook_dispatcher import WebhookDispatcher, WebhookEvent
111 print("\n" + "=" * 60)
112 print("WEBHOOK DELIVERY TEST - LOCAL SERVER")
113 print("=" * 60 + "\n")
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
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 ]
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
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 )
149 # Create mock job
150 job = create_mock_job()
152 print("Dispatching webhook for job.completed event...")
153 print(f"Target URL: {webhook_url}")
154 print("Secret configured: Yes")
155 print()
157 # Dispatch the event
158 results = await dispatcher._dispatch_event(WebhookEvent.JOB_COMPLETED, job)
160 # Check results
161 print("\n" + "-" * 40)
162 print("DELIVERY RESULTS:")
163 print("-" * 40)
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}")
176 # Verify signature
177 if received_webhooks:
178 print("\n" + "-" * 40)
179 print("SIGNATURE VERIFICATION:")
180 print("-" * 40)
182 webhook = received_webhooks[-1]
183 signature = webhook["headers"].get("X-GCO-Signature", "")
184 body = webhook["body"]
186 expected = hmac.new(
187 webhook_secret.encode("utf-8"),
188 body.encode("utf-8"),
189 hashlib.sha256,
190 ).hexdigest()
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}")
199 # Cleanup
200 server.shutdown()
202 print("\n" + "=" * 60)
203 print("TEST COMPLETE")
204 print("=" * 60 + "\n")
206 return len(results) > 0 and all(r.success for r in results)
209async def test_with_external_url(
210 url: str,
211 # Not a collected test (pytest's testpaths is tests/): this is the manual
212 # harness's CLI entry point, and the default mirrors its optional --secret.
213 secret: str | None = None, # noqa: PT028
214) -> bool:
215 """Test webhook delivery with an external URL (e.g., webhook.site)."""
216 from gco.services.webhook_dispatcher import WebhookDispatcher, WebhookEvent
218 print("\n" + "=" * 60)
219 print("WEBHOOK DELIVERY TEST - EXTERNAL URL")
220 print("=" * 60 + "\n")
222 print(f"Target URL: {url}")
223 print(f"Secret configured: {'Yes' if secret else 'No'}")
224 print()
226 # Create mock webhook store
227 mock_store = MagicMock()
228 webhook_config = {
229 "id": "external-webhook",
230 "url": url,
231 "events": ["job.completed", "job.failed", "job.started"],
232 "namespace": None, # Global webhook
233 }
234 if secret:
235 webhook_config["secret"] = secret
236 mock_store.get_webhooks_for_event.return_value = [webhook_config]
238 # Create dispatcher with mocked K8s config
239 with patch("gco.services.webhook_dispatcher.config") as mock_config:
240 mock_config.ConfigException = Exception
241 mock_config.load_incluster_config.side_effect = Exception()
242 mock_config.load_kube_config.return_value = None
244 with patch("gco.services.webhook_dispatcher.client"):
245 dispatcher = WebhookDispatcher(
246 cluster_id="gco-test-cluster",
247 region="us-east-1",
248 webhook_store=mock_store,
249 timeout=30,
250 max_retries=2,
251 retry_delay=2,
252 )
254 # Create mock job
255 job = create_mock_job()
257 # Test all three event types
258 events = [
259 (WebhookEvent.JOB_STARTED, "job.started"),
260 (WebhookEvent.JOB_COMPLETED, "job.completed"),
261 (WebhookEvent.JOB_FAILED, "job.failed"),
262 ]
264 all_success = True
266 for event, event_name in events:
267 # Adjust job status for the event
268 if event == WebhookEvent.JOB_STARTED:
269 job.status.conditions = []
270 job.status.active = 1
271 job.status.succeeded = 0
272 job.status.completion_time = None
273 elif event == WebhookEvent.JOB_COMPLETED:
274 job.status.conditions = [MagicMock(type="Complete", status="True")]
275 job.status.active = 0
276 job.status.succeeded = 1
277 job.status.completion_time = datetime(2026, 2, 4, 12, 5, 0, tzinfo=UTC)
278 else: # WebhookEvent.JOB_FAILED — the last of the three events above
279 job.status.conditions = [MagicMock(type="Failed", status="True")]
280 job.status.active = 0
281 job.status.succeeded = 0
282 job.status.failed = 1
283 job.status.completion_time = datetime(2026, 2, 4, 12, 5, 0, tzinfo=UTC)
285 print(f"Dispatching {event_name} event...")
286 results = await dispatcher._dispatch_event(event, job)
288 for result in results:
289 status = "✓" if result.success else "✗"
290 print(
291 f" {status} Status: {result.status_code}, "
292 f"Attempts: {result.attempts}, "
293 f"Duration: {result.duration_ms:.1f}ms"
294 )
295 if not result.success:
296 print(f" Error: {result.error}")
297 all_success = False
299 # Small delay between events
300 await asyncio.sleep(1)
302 print("\n" + "=" * 60)
303 if all_success:
304 print("ALL WEBHOOKS DELIVERED SUCCESSFULLY!")
305 print(f"Check your webhook receiver at: {url}")
306 else:
307 print("SOME WEBHOOKS FAILED - Check errors above")
308 print("=" * 60 + "\n")
310 return all_success
313async def main() -> int:
314 parser = argparse.ArgumentParser(description="Test webhook delivery")
315 parser.add_argument(
316 "--url",
317 help="External webhook URL (e.g., https://webhook.site/your-uuid)",
318 )
319 parser.add_argument(
320 "--secret",
321 help="HMAC secret for signature verification",
322 )
323 parser.add_argument(
324 "--local",
325 action="store_true",
326 help="Run test with local server (default if no URL provided)",
327 )
329 args = parser.parse_args()
331 if args.url:
332 success = await test_with_external_url(args.url, args.secret)
333 else:
334 success = await test_with_local_server()
336 return 0 if success else 1
339if __name__ == "__main__":
340 sys.exit(asyncio.run(main()))