From e41d040c9aa18f49734f23abf181f078f370d594 Mon Sep 17 00:00:00 2001 From: Austin Date: Thu, 14 May 2026 13:55:44 -0700 Subject: [PATCH] Added GitHub notification support for Slack, Discord, Teams, Email, and SMS --- .github/workflows/README.md | 177 ++++++++++ .github/workflows/testsPython.yml | 49 ++- README.md | 2 + scripts/github_actions_notify.py | 551 ++++++++++++++++++++++++++++++ 4 files changed, 773 insertions(+), 6 deletions(-) create mode 100644 .github/workflows/README.md create mode 100644 scripts/github_actions_notify.py diff --git a/.github/workflows/README.md b/.github/workflows/README.md new file mode 100644 index 0000000..8e814cc --- /dev/null +++ b/.github/workflows/README.md @@ -0,0 +1,177 @@ +# GitHub Actions Notifications + +The `testsPython.yml` workflow can notify Slack, Discord, Microsoft Teams, Mailgun, and Twilio when the Python unit test job fails. It can also send manual test notifications or print dry-run payloads from the GitHub Actions UI. + +The workflow calls `scripts/github_actions_notify.py`. The script uses only the Python standard library, so the GitHub Actions runner does not need to install vendor SDKs. + +## Behavior + +- Notifications are sent when `python-unit-tests` fails. +- The notification job uses `if: ${{ always() }}` so it still runs after a test failure. +- Each provider is optional. A provider is skipped when its required configuration or addressees are not configured. +- Each provider supports JSON arrays where multiple addressees make sense. +- Slack and Discord support target objects so mentions can be set per webhook destination. +- Notification failures are allowed to continue so a broken notification provider does not block CI. +- Dry-run output masks webhook URLs, email addresses, and phone numbers in logs. + +## GitHub Secrets and Variables + +Add secrets and variables from the repository settings: + +1. Open the repository in GitHub. +2. Go to `Settings` -> `Secrets and variables` -> `Actions`. +3. Add sensitive values under `Secrets`. +4. Add non-sensitive values under `Variables`. + +Treat webhook URLs, API keys, email addresses, phone numbers, Account SIDs, and Messaging Service SIDs as secrets. Dry-run logs mask these values, but GitHub Actions can still print workflow environment values that are stored as variables. + +Use JSON arrays for multiple addressees: + +```json +["first@example.com", "second@example.com"] +``` + +Provider addressee fields use JSON arrays so one workflow configuration can send to multiple destinations. + +Slack and Discord use target objects so each webhook destination can have its own optional mentions: + +```json +[ + { + "url": "https://example.invalid/webhook", + "mentions": ["<@U0123456789>", ""] + }, + { + "url": "https://example.invalid/another-webhook" + } +] +``` + +## Manual Testing + +Open the `Actions` tab, select `Python Unit Tests`, choose `Run workflow`, then set `notification-mode`. + +### `notify-on-failure` + +Use `notify-on-failure` for normal workflow behavior. Notifications are sent only when the Python unit test job fails. + +### `dry-run` + +Use `dry-run` to print parsed addressee lists and payloads without sending notifications. This is the safest first test after adding configuration. + +### `test-notification` + +Use `test-notification` to send configured notifications even if the Python tests pass. Use this after dry-run output looks correct. + +## Slack + +Get credentials from Slack by creating or selecting a Slack app and enabling Incoming Webhooks: + +- Slack Incoming Webhooks: https://api.slack.com/messaging/webhooks +- Slack GitHub Action docs: https://docs.slack.dev/tools/slack-github-action/ + +After you have webhook URLs, configure this secret: + +| Type | Name | Example | +| --- | --- | --- | +| Secret | `SLACK_WEBHOOK_TARGETS_JSON` | `[{"url": "https://hooks.slack.com/services/T000/B000/XXX", "mentions": ["<@U0123456789>", ""]}, {"url": "https://hooks.slack.com/services/T111/B111/YYY"}]` | + +Each target object requires `url` and may include an optional `mentions` array. + +Each Slack webhook is tied to the channel selected when the webhook is created, so a list of webhook URLs is the Slack addressee list. + +## Discord + +Get credentials by creating webhooks in the Discord channels that should receive alerts: + +- Discord webhook resources: https://discord.com/developers/docs/resources/webhook +- Discord webhook guide: https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks + +After you have webhook URLs, configure this secret: + +| Type | Name | Example | +| --- | --- | --- | +| Secret | `DISCORD_WEBHOOK_TARGETS_JSON` | `[{"url": "https://discord.com/api/webhooks/111/aaa", "mentions": ["<@everyone>"]}, {"url": "https://discord.com/api/webhooks/222/bbb"}]` | + +Each target object requires `url` and may include an optional `mentions` array. + +Each Discord webhook is tied to one channel. + +## Microsoft Teams + +Get a Teams webhook URL from the Microsoft Teams Workflows app or from the incoming webhook option supported by your tenant: + +- Teams incoming webhook setup: https://support.microsoft.com/en-US/Workflows/send-messages-in-teams-using-incoming-webhooks + +After you have webhook URLs, configure this secret: + +| Type | Name | Example | +| --- | --- | --- | +| Secret | `TEAMS_WEBHOOK_URLS_JSON` | `["https://example.webhook.office.com/webhookb2/...", "https://prod-00.westus.logic.azure.com/..."]` | + +Each Teams webhook maps to the team, channel, or workflow destination selected during setup. + +## Mailgun + +Get credentials from Mailgun: + +- Mailgun dashboard: https://app.mailgun.com/ +- Mailgun API keys: https://app.mailgun.com/app/account/security/api_keys +- Mailgun sending domains: https://app.mailgun.com/app/sending/domains +- Mailgun Messages API: https://documentation.mailgun.com/docs/mailgun/api-reference/send/mailgun/messages + +After you have an API key and a sending domain, configure: + +| Type | Name | Example | +| --- | --- | --- | +| Secret | `MAILGUN_API_KEY` | `xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx` | +| Variable | `MAILGUN_DOMAIN` | `mg.example.com` | +| Secret | `MAILGUN_FROM_EMAIL` | `GitHub Actions ` | +| Secret | `MAILGUN_TO_EMAILS_JSON` | `["first@example.com", "second@example.com"]` | +| Variable [1] | `MAILGUN_API_BASE_URL` | `https://api.mailgun.net/v3` | + +[1] `MAILGUN_API_BASE_URL` is optional. The default is `https://api.mailgun.net/v3`. Use `https://api.eu.mailgun.net/v3` for Mailgun EU domains. + +## Twilio + +Get credentials from Twilio: + +- Twilio Console: https://console.twilio.com/ +- Twilio API credentials: https://www.twilio.com/docs/iam/api +- Twilio Messaging API: https://www.twilio.com/docs/messaging/api/message-resource +- Twilio phone numbers: https://console.twilio.com/us1/develop/phone-numbers/manage/incoming + +After you have an Account SID, Auth Token, and a sending phone number or Messaging Service SID, configure: + +| Type | Name | Example | +| --- | --- | --- | +| Secret | `TWILIO_ACCOUNT_SID` | `ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx` | +| Secret | `TWILIO_AUTH_TOKEN` | `xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx` | +| Secret | `TWILIO_TO_PHONES_JSON` | `["+16045550123", "+12505550123"]` | +| Secret [1] | `TWILIO_FROM_PHONE` | `+16045550999` | +| Secret [1] | `TWILIO_MESSAGING_SERVICE_SID` | `MGxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx` | + +[1] Set either `TWILIO_FROM_PHONE` or `TWILIO_MESSAGING_SERVICE_SID`. SMS providers may charge for sent messages, phone numbers, carrier fees, and message segments. + +## Example Dry-Run Configuration + +For a safe dry run, configure personal addresses and phone numbers as secrets and non-sensitive routing values as variables. + +Secrets: + +```text +MAILGUN_FROM_EMAIL=GitHub Actions +MAILGUN_TO_EMAILS_JSON=["first@example.com", "second@example.com"] +TWILIO_TO_PHONES_JSON=["+16045550123", "+12505550123"] +TWILIO_FROM_PHONE=+16045550999 +``` + +Variables: + +```text +MAILGUN_DOMAIN=mg.example.com +``` + +Then manually run `Python Unit Tests` with `notification-mode=dry-run`. + +For webhook providers, the webhook URL is both the addressee and the credential, so store webhook lists as GitHub secrets before testing. diff --git a/.github/workflows/testsPython.yml b/.github/workflows/testsPython.yml index 452f71d..5706ff9 100644 --- a/.github/workflows/testsPython.yml +++ b/.github/workflows/testsPython.yml @@ -22,6 +22,16 @@ name: Python Unit Tests on: workflow_dispatch: # Allows the workflow to be manually triggered from the GitHub Actions tab + inputs: + notification-mode: + description: "Notification behavior for this manual run" + required: false + type: choice + default: notify-on-failure + options: + - notify-on-failure + - dry-run + - test-notification pull_request: # paths: # Trigger workflow on pull requests, but - "**.py" # only if Python files are changed @@ -67,13 +77,40 @@ jobs: # is scaffolded to facilitate sending notifications based # on the test results. notifications: + if: ${{ always() }} needs: python-unit-tests runs-on: ubuntu-latest steps: + - name: Checkout code + id: checkout + uses: actions/checkout@v6 + - name: Notify on test results - run: | - if [ "${{ needs.python-unit-tests.result }}" == "success" ]; then - echo "success notifications go here" - else - echo "failure notifications go here" - fi + continue-on-error: true + env: + PYTHON_UNIT_TEST_RESULT: ${{ needs.python-unit-tests.result }} + NOTIFICATION_TEST: ${{ github.event.inputs['notification-mode'] == 'test-notification' && 'true' || 'false' }} + NOTIFICATION_DRY_RUN: ${{ github.event.inputs['notification-mode'] == 'dry-run' && 'true' || 'false' }} + GITHUB_REPOSITORY_NAME: ${{ github.repository }} + GITHUB_WORKFLOW_NAME: ${{ github.workflow }} + GITHUB_RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + GITHUB_REF_NAME: ${{ github.ref_name }} + GITHUB_SHA: ${{ github.sha }} + GITHUB_ACTOR: ${{ github.actor }} + GITHUB_EVENT_NAME: ${{ github.event_name }} + GITHUB_RUN_ID: ${{ github.run_id }} + GITHUB_RUN_ATTEMPT: ${{ github.run_attempt }} + SLACK_WEBHOOK_TARGETS_JSON: ${{ secrets.SLACK_WEBHOOK_TARGETS_JSON }} + DISCORD_WEBHOOK_TARGETS_JSON: ${{ secrets.DISCORD_WEBHOOK_TARGETS_JSON }} + TEAMS_WEBHOOK_URLS_JSON: ${{ secrets.TEAMS_WEBHOOK_URLS_JSON }} + MAILGUN_API_KEY: ${{ secrets.MAILGUN_API_KEY }} + MAILGUN_API_BASE_URL: ${{ vars.MAILGUN_API_BASE_URL }} + MAILGUN_DOMAIN: ${{ vars.MAILGUN_DOMAIN }} + MAILGUN_FROM_EMAIL: ${{ secrets.MAILGUN_FROM_EMAIL }} + MAILGUN_TO_EMAILS_JSON: ${{ secrets.MAILGUN_TO_EMAILS_JSON }} + TWILIO_ACCOUNT_SID: ${{ secrets.TWILIO_ACCOUNT_SID }} + TWILIO_AUTH_TOKEN: ${{ secrets.TWILIO_AUTH_TOKEN }} + TWILIO_FROM_PHONE: ${{ secrets.TWILIO_FROM_PHONE }} + TWILIO_MESSAGING_SERVICE_SID: ${{ secrets.TWILIO_MESSAGING_SERVICE_SID }} + TWILIO_TO_PHONES_JSON: ${{ secrets.TWILIO_TO_PHONES_JSON }} + run: python3 scripts/github_actions_notify.py diff --git a/README.md b/README.md index 28065c8..b0c5fd2 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,8 @@ make docker-prune # prune (permanently delete) all existing data in Docker: co Documentation is available here: [Documentation](./doc/) +GitHub Actions notification setup is available here: [Workflow Notifications](./.github/workflows/README.md) + ## Support To get community support, go to the official [Issues Page](https://github.com/FullStackWithLawrence/agentic-ai-workflow/issues) for this project. diff --git a/scripts/github_actions_notify.py b/scripts/github_actions_notify.py new file mode 100644 index 0000000..ccceac7 --- /dev/null +++ b/scripts/github_actions_notify.py @@ -0,0 +1,551 @@ +#!/usr/bin/env python3 +"""Send GitHub Actions test-result notifications to configured vendors.""" + +from __future__ import annotations + +import base64 +import json +import os +import sys +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from typing import Any, Callable + + +USER_AGENT = "agentic-ai-workflow-github-actions-notifier/1.0" + + +class NotificationError(Exception): + """Raised when a notification provider cannot complete its work.""" + + +@dataclass(frozen=True) +class NotificationContext: + """GitHub Actions context used to build notification messages.""" + + repository: str + workflow: str + run_id: str + run_attempt: str + run_url: str + ref_name: str + sha: str + actor: str + event_name: str + test_result: str + notification_test: bool + dry_run: bool + + @property + def short_sha(self) -> str: + return self.sha[:7] if self.sha else "unknown" + + @property + def should_notify(self) -> bool: + return self.dry_run or self.notification_test or self.test_result == "failure" + + @property + def mode(self) -> str: + if self.dry_run: + return "dry-run" + if self.notification_test: + return "test" + if self.test_result == "failure": + return "failure" + return "none" + + @property + def title(self) -> str: + if self.dry_run: + return "GitHub Actions notification dry run" + if self.notification_test: + return "GitHub Actions test notification" + if self.test_result == "failure": + return "Python unit tests failed" + return "Python unit tests completed" + + @property + def subject(self) -> str: + return f"{self.title}: {self.repository}" + + @property + def message(self) -> str: + return "\n".join( + [ + self.title, + f"Repository: {self.repository}", + f"Workflow: {self.workflow}", + f"Result: {self.test_result}", + f"Branch: {self.ref_name}", + f"Commit: {self.short_sha}", + f"Actor: {self.actor}", + f"Event: {self.event_name}", + f"Run attempt: {self.run_attempt}", + f"Run: {self.run_url}", + ] + ) + + @property + def sms_message(self) -> str: + return ( + f"{self.title}: {self.repository} " + f"{self.ref_name}@{self.short_sha} result={self.test_result}. " + f"Run: {self.run_url}" + ) + + +@dataclass(frozen=True) +class WebhookTarget: + """Webhook destination with optional provider-specific mentions.""" + + url: str + mentions: tuple[str, ...] = () + + +def getenv(name: str, default: str = "") -> str: + value = os.environ.get(name) + if value is None: + return default + return value.strip() or default + + +def truthy(value: str) -> bool: + return value.strip().lower() in {"1", "true", "yes", "y", "on"} + + +def read_json_list(name: str) -> list[str]: + """Read a JSON list from an environment variable.""" + + raw = getenv(name) + if raw: + try: + values = json.loads(raw) + except json.JSONDecodeError as exc: + raise NotificationError(f"{name} must be a JSON list") from exc + if not isinstance(values, list): + raise NotificationError(f"{name} must be a JSON list") + cleaned = [str(value).strip() for value in values if str(value).strip()] + if len(cleaned) != len(values): + raise NotificationError(f"{name} cannot contain empty values") + return cleaned + + return [] + + +def read_webhook_targets( + targets_name: str, +) -> list[WebhookTarget]: + """Read webhook targets with optional mentions.""" + + raw = getenv(targets_name) + if not raw: + return [] + + try: + values = json.loads(raw) + except json.JSONDecodeError as exc: + raise NotificationError(f"{targets_name} must be a JSON list of objects") from exc + + if not isinstance(values, list): + raise NotificationError(f"{targets_name} must be a JSON list of objects") + + targets = [] + for index, value in enumerate(values, start=1): + if not isinstance(value, dict): + raise NotificationError(f"{targets_name}[{index}] must be an object") + + url = str(value.get("url", "")).strip() + if not url: + raise NotificationError(f"{targets_name}[{index}].url is required") + + mentions = value.get("mentions", []) + if mentions is None: + mentions = [] + if not isinstance(mentions, list): + raise NotificationError(f"{targets_name}[{index}].mentions must be a JSON list") + + cleaned_mentions = [str(mention).strip() for mention in mentions if str(mention).strip()] + if len(cleaned_mentions) != len(mentions): + raise NotificationError(f"{targets_name}[{index}].mentions cannot contain empty values") + + targets.append(WebhookTarget(url=url, mentions=tuple(cleaned_mentions))) + + return targets + + +def mask_value(value: str, kind: str) -> str: + if not value: + return "" + + if kind == "url": + parsed = urllib.parse.urlparse(value) + if parsed.scheme and parsed.netloc: + return f"{parsed.scheme}://{parsed.netloc}/" + return "" + + if kind == "sensitive_url": + return "" + + if kind == "email": + if "<" in value and ">" in value: + value = value[value.find("<") + 1 : value.rfind(">")].strip() + if "@" not in value: + return "" + name, domain = value.split("@", 1) + visible = name[:1] if name else "*" + return f"{visible}***@{domain}" + + if kind == "sensitive_email": + if "<" in value and ">" in value: + value = value[value.find("<") + 1 : value.rfind(">")].strip() + if "@" not in value: + return "" + name, _domain = value.split("@", 1) + visible = name[:1] if name else "*" + return f"{visible}***@" + + if kind == "phone": + digits = "".join(character for character in value if character.isdigit()) + suffix = digits[-4:] if len(digits) >= 4 else "****" + return f"***{suffix}" + + if len(value) <= 8: + return "" + return f"{value[:4]}...{value[-4:]}" + + +def masked_list(values: list[str], kind: str) -> list[str]: + return [mask_value(value, kind) for value in values] + + +def mask_mailgun_url(api_base_url: str) -> str: + parsed = urllib.parse.urlparse(api_base_url) + if parsed.scheme and parsed.netloc: + base_path = parsed.path.rstrip("/") + return f"{parsed.scheme}://{parsed.netloc}{base_path}//messages" + return "" + + +def masked_webhook_targets(targets: list[WebhookTarget]) -> list[dict[str, Any]]: + return [ + { + "url": mask_value(target.url, "url"), + "mentions": list(target.mentions), + } + for target in targets + ] + + +def print_json(label: str, payload: Any) -> None: + print(label) + print(json.dumps(payload, indent=2, sort_keys=True)) + + +def http_post_json(url: str, payload: dict[str, Any]) -> tuple[int, str]: + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={ + "Content-Type": "application/json", + "User-Agent": USER_AGENT, + }, + method="POST", + ) + return send_request(request) + + +def http_post_form( + url: str, + form_values: list[tuple[str, str]], + username: str, + password: str, +) -> tuple[int, str]: + auth = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") + request = urllib.request.Request( + url, + data=urllib.parse.urlencode(form_values).encode("utf-8"), + headers={ + "Authorization": f"Basic {auth}", + "Content-Type": "application/x-www-form-urlencoded", + "User-Agent": USER_AGENT, + }, + method="POST", + ) + return send_request(request) + + +def send_request(request: urllib.request.Request) -> tuple[int, str]: + try: + with urllib.request.urlopen(request, timeout=20) as response: + body = response.read().decode("utf-8", errors="replace") + return response.status, body + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + raise NotificationError(f"HTTP {exc.code}: {body[:1000]}") from exc + except urllib.error.URLError as exc: + raise NotificationError(f"Request failed: {exc.reason}") from exc + + +def send_webhook_notifications( + provider: str, + urls: list[str], + payload: dict[str, Any], + ctx: NotificationContext, + url_mask_kind: str = "url", +) -> None: + if not urls: + print(f"{provider}: skipped, no webhook URLs configured") + return + + print(f"{provider}: parsed addressees: {masked_list(urls, url_mask_kind)}") + for index, url in enumerate(urls, start=1): + masked_url = mask_value(url, url_mask_kind) + if ctx.dry_run: + print_json(f"{provider}: dry-run payload for addressee {index} ({masked_url})", payload) + continue + status, body = http_post_json(url, payload) + print(f"{provider}: sent to addressee {index} ({masked_url}), status {status}, response {body[:200]}") + + +def send_targeted_webhook_notifications( + provider: str, + targets: list[WebhookTarget], + ctx: NotificationContext, + payload_builder: Callable[[WebhookTarget], dict[str, Any]], +) -> None: + if not targets: + print(f"{provider}: skipped, no webhook targets configured") + return + + print_json(f"{provider}: parsed addressees", masked_webhook_targets(targets)) + for index, target in enumerate(targets, start=1): + payload = payload_builder(target) + masked_url = mask_value(target.url, "url") + if ctx.dry_run: + print_json(f"{provider}: dry-run payload for addressee {index} ({masked_url})", payload) + continue + status, body = http_post_json(target.url, payload) + print(f"{provider}: sent to addressee {index} ({masked_url}), status {status}, response {body[:200]}") + + +def prepend_mentions(message: str, mentions: tuple[str, ...]) -> str: + if not mentions: + return message + return f"{' '.join(mentions)}\n{message}" + + +def notify_slack(ctx: NotificationContext) -> None: + targets = read_webhook_targets("SLACK_WEBHOOK_TARGETS_JSON") + + def build_payload(target: WebhookTarget) -> dict[str, Any]: + return {"text": prepend_mentions(ctx.message, target.mentions)} + + send_targeted_webhook_notifications("Slack", targets, ctx, build_payload) + + +def notify_discord(ctx: NotificationContext) -> None: + targets = read_webhook_targets("DISCORD_WEBHOOK_TARGETS_JSON") + + def build_payload(target: WebhookTarget) -> dict[str, Any]: + return {"content": prepend_mentions(ctx.message, target.mentions)} + + send_targeted_webhook_notifications("Discord", targets, ctx, build_payload) + + +def build_teams_adaptive_card(ctx: NotificationContext) -> dict[str, Any]: + facts = [ + {"title": "Repository", "value": ctx.repository}, + {"title": "Workflow", "value": ctx.workflow}, + {"title": "Result", "value": ctx.test_result}, + {"title": "Branch", "value": ctx.ref_name}, + {"title": "Commit", "value": ctx.short_sha}, + {"title": "Actor", "value": ctx.actor}, + {"title": "Event", "value": ctx.event_name}, + {"title": "Run attempt", "value": ctx.run_attempt}, + ] + return { + "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", + "type": "AdaptiveCard", + "version": "1.4", + "body": [ + { + "type": "TextBlock", + "text": ctx.title, + "weight": "Bolder", + "size": "Medium", + "wrap": True, + }, + { + "type": "FactSet", + "facts": facts, + }, + ], + "actions": [ + { + "type": "Action.OpenUrl", + "title": "Open workflow run", + "url": ctx.run_url, + } + ], + } + + +def notify_teams(ctx: NotificationContext) -> None: + urls = read_json_list("TEAMS_WEBHOOK_URLS_JSON") + payload = build_teams_adaptive_card(ctx) + send_webhook_notifications("Microsoft Teams", urls, payload, ctx, "sensitive_url") + + +def notify_mailgun(ctx: NotificationContext) -> None: + recipients = read_json_list("MAILGUN_TO_EMAILS_JSON") + if not recipients: + print("Mailgun: skipped, no email recipients configured") + return + + domain = getenv("MAILGUN_DOMAIN") + from_email = getenv("MAILGUN_FROM_EMAIL") + api_key = getenv("MAILGUN_API_KEY") + api_base_url = getenv("MAILGUN_API_BASE_URL", "https://api.mailgun.net/v3").rstrip("/") + + missing = [] + if not domain: + missing.append("MAILGUN_DOMAIN") + if not from_email: + missing.append("MAILGUN_FROM_EMAIL") + if not ctx.dry_run and not api_key: + missing.append("MAILGUN_API_KEY") + if missing: + raise NotificationError(f"Mailgun missing required configuration: {', '.join(missing)}") + + form_values = [ + ("from", from_email), + ("subject", ctx.subject), + ("text", ctx.message), + ] + form_values.extend(("to", recipient) for recipient in recipients) + + print(f"Mailgun: parsed addressees: {masked_list(recipients, 'email')}") + if ctx.dry_run: + payload = { + "from": mask_value(from_email, "sensitive_email"), + "to": masked_list(recipients, "email"), + "subject": ctx.subject, + "text": ctx.message, + "url": mask_mailgun_url(api_base_url), + } + print_json("Mailgun: dry-run payload", payload) + return + + url = f"{api_base_url}/{urllib.parse.quote(domain, safe='')}/messages" + status, body = http_post_form(url, form_values, "api", api_key) + print(f"Mailgun: sent to {len(recipients)} recipients, status {status}, response {body[:200]}") + + +def notify_twilio(ctx: NotificationContext) -> None: + recipients = read_json_list("TWILIO_TO_PHONES_JSON") + if not recipients: + print("Twilio: skipped, no phone recipients configured") + return + + account_sid = getenv("TWILIO_ACCOUNT_SID") + auth_token = getenv("TWILIO_AUTH_TOKEN") + from_phone = getenv("TWILIO_FROM_PHONE") + messaging_service_sid = getenv("TWILIO_MESSAGING_SERVICE_SID") + + missing = [] + if not ctx.dry_run and not account_sid: + missing.append("TWILIO_ACCOUNT_SID") + if not ctx.dry_run and not auth_token: + missing.append("TWILIO_AUTH_TOKEN") + if not from_phone and not messaging_service_sid: + missing.append("TWILIO_FROM_PHONE or TWILIO_MESSAGING_SERVICE_SID") + if missing: + raise NotificationError(f"Twilio missing required configuration: {', '.join(missing)}") + + endpoint_sid = account_sid or "" + url = f"https://api.twilio.com/2010-04-01/Accounts/{urllib.parse.quote(endpoint_sid, safe='')}/Messages.json" + + print(f"Twilio: parsed addressees: {masked_list(recipients, 'phone')}") + for index, recipient in enumerate(recipients, start=1): + form_values = [ + ("To", recipient), + ("Body", ctx.sms_message), + ] + if messaging_service_sid: + form_values.append(("MessagingServiceSid", messaging_service_sid)) + else: + form_values.append(("From", from_phone)) + + if ctx.dry_run: + payload = { + "to": mask_value(recipient, "phone"), + "body": ctx.sms_message, + } + if messaging_service_sid: + payload["messaging_service_sid"] = mask_value(messaging_service_sid, "generic") + else: + payload["from"] = mask_value(from_phone, "phone") + print_json(f"Twilio: dry-run payload for recipient {index}", payload) + continue + + status, body = http_post_form(url, form_values, account_sid, auth_token) + print( + "Twilio: sent to recipient " + f"{index} ({mask_value(recipient, 'phone')}), status {status}, response {body[:200]}" + ) + + +def build_context() -> NotificationContext: + return NotificationContext( + repository=getenv("GITHUB_REPOSITORY_NAME", getenv("GITHUB_REPOSITORY", "unknown")), + workflow=getenv("GITHUB_WORKFLOW_NAME", getenv("GITHUB_WORKFLOW", "unknown")), + run_id=getenv("GITHUB_RUN_ID", "unknown"), + run_attempt=getenv("GITHUB_RUN_ATTEMPT", "unknown"), + run_url=getenv("GITHUB_RUN_URL"), + ref_name=getenv("GITHUB_REF_NAME"), + sha=getenv("GITHUB_SHA"), + actor=getenv("GITHUB_ACTOR"), + event_name=getenv("GITHUB_EVENT_NAME"), + test_result=getenv("PYTHON_UNIT_TEST_RESULT", "unknown"), + notification_test=truthy(getenv("NOTIFICATION_TEST")), + dry_run=truthy(getenv("NOTIFICATION_DRY_RUN")), + ) + + +def main() -> int: + ctx = build_context() + print(f"Notification mode: {ctx.mode}") + print(f"Python unit test result: {ctx.test_result}") + + if not ctx.should_notify: + print("Notifications skipped because tests did not fail and no manual notification mode was requested.") + return 0 + + providers = [ + notify_slack, + notify_discord, + notify_teams, + notify_mailgun, + notify_twilio, + ] + + failures = 0 + for provider in providers: + try: + provider(ctx) + except NotificationError as exc: + failures += 1 + print(f"{provider.__name__}: failed: {exc}", file=sys.stderr) + + if failures: + print(f"Notification providers completed with {failures} failure(s).", file=sys.stderr) + return 1 + + print("Notification providers completed.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())