From f7c40d48c0727a96843c85990cc36ae5a9ac6888 Mon Sep 17 00:00:00 2001
From: Joel Grunbaum <joelgrun@gmail.com>
Date: Sat, 07 Feb 2026 23:42:43 +0000
Subject: [PATCH] Add integration test for binary
---
src/archbuild/notifications.py | 346 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 346 insertions(+), 0 deletions(-)
diff --git a/src/archbuild/notifications.py b/src/archbuild/notifications.py
new file mode 100644
index 0000000..be00edc
--- /dev/null
+++ b/src/archbuild/notifications.py
@@ -0,0 +1,346 @@
+"""Notification system with email and extensible webhook support."""
+
+import asyncio
+import smtplib
+import ssl
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from datetime import datetime
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from typing import Any
+
+import aiohttp
+
+from archbuild.builder import BuildResult, BuildStatus
+from archbuild.config import Config, EmailConfig, WebhookConfig
+from archbuild.logging import get_logger
+
+logger = get_logger("notifications")
+
+
+@dataclass
+class BuildSummary:
+ """Summary of build results for notifications."""
+
+ total: int
+ success: int
+ failed: int
+ skipped: int
+ failed_packages: list[str]
+ duration: float
+ timestamp: datetime
+
+ @classmethod
+ def from_results(cls, results: list[BuildResult]) -> "BuildSummary":
+ """Create summary from build results."""
+ return cls(
+ total=len(results),
+ success=sum(1 for r in results if r.status == BuildStatus.SUCCESS),
+ failed=sum(1 for r in results if r.status == BuildStatus.FAILED),
+ skipped=sum(1 for r in results if r.status == BuildStatus.SKIPPED),
+ failed_packages=[r.package for r in results if r.status == BuildStatus.FAILED],
+ duration=sum(r.duration for r in results),
+ timestamp=datetime.now(),
+ )
+
+
+class NotificationBackend(ABC):
+ """Abstract base class for notification backends."""
+
+ @abstractmethod
+ async def send(self, summary: BuildSummary, config: Config) -> bool:
+ """Send notification.
+
+ Args:
+ summary: Build summary
+ config: Application config
+
+ Returns:
+ True if sent successfully
+ """
+ pass
+
+
+class EmailBackend(NotificationBackend):
+ """Email notification backend."""
+
+ def __init__(self, email_config: EmailConfig):
+ """Initialize email backend.
+
+ Args:
+ email_config: Email configuration
+ """
+ self.config = email_config
+
+ def _format_message(self, summary: BuildSummary, repo_name: str) -> str:
+ """Format email message body.
+
+ Args:
+ summary: Build summary
+ repo_name: Repository name
+
+ Returns:
+ Formatted message string
+ """
+ lines = [
+ f"Build Report for {repo_name}",
+ f"Time: {summary.timestamp.strftime('%Y-%m-%d %H:%M:%S')}",
+ "",
+ f"Total packages: {summary.total}",
+ f" Successful: {summary.success}",
+ f" Failed: {summary.failed}",
+ f" Skipped: {summary.skipped}",
+ f"Total duration: {summary.duration:.1f}s",
+ ]
+
+ if summary.failed_packages:
+ lines.extend([
+ "",
+ "Failed packages:",
+ ])
+ for pkg in summary.failed_packages:
+ lines.append(f" - {pkg}")
+
+ return "\n".join(lines)
+
+ async def send(self, summary: BuildSummary, config: Config) -> bool:
+ """Send email notification."""
+ if not self.config.enabled:
+ return True
+
+ if not self.config.to:
+ logger.warning("Email notification enabled but no recipient configured")
+ return False
+
+ # Only send on failures
+ if summary.failed == 0:
+ logger.debug("No failures, skipping email notification")
+ return True
+
+ try:
+ msg = MIMEMultipart()
+ msg["From"] = self.config.from_addr or f"archbuild@localhost"
+ msg["To"] = self.config.to
+ msg["Subject"] = f"Build Errors - {config.repository.name}"
+
+ body = self._format_message(summary, config.repository.name)
+ msg.attach(MIMEText(body, "plain"))
+
+ # Send email
+ loop = asyncio.get_event_loop()
+ await loop.run_in_executor(None, self._send_email, msg)
+
+ logger.info(f"Sent email notification to {self.config.to}")
+ return True
+
+ except Exception as e:
+ logger.error(f"Failed to send email: {e}")
+ return False
+
+ def _send_email(self, msg: MIMEMultipart) -> None:
+ """Send email synchronously (called from executor)."""
+ if self.config.use_tls:
+ context = ssl.create_default_context()
+ with smtplib.SMTP_SSL(
+ self.config.smtp_host,
+ self.config.smtp_port,
+ context=context,
+ ) as server:
+ if self.config.username and self.config.password:
+ server.login(self.config.username, self.config.password)
+ server.send_message(msg)
+ else:
+ with smtplib.SMTP(self.config.smtp_host, self.config.smtp_port) as server:
+ if self.config.username and self.config.password:
+ server.login(self.config.username, self.config.password)
+ server.send_message(msg)
+
+
+class WebhookBackend(NotificationBackend):
+ """Generic webhook notification backend.
+
+ Extensible for Discord, Slack, ntfy, Gotify, etc.
+ """
+
+ def __init__(self, webhook_config: WebhookConfig):
+ """Initialize webhook backend.
+
+ Args:
+ webhook_config: Webhook configuration
+ """
+ self.config = webhook_config
+
+ def _format_payload(self, summary: BuildSummary, repo_name: str) -> dict[str, Any]:
+ """Format webhook payload based on webhook type.
+
+ Args:
+ summary: Build summary
+ repo_name: Repository name
+
+ Returns:
+ Payload dictionary
+ """
+ base_content = (
+ f"**Build Report for {repo_name}**\n"
+ f"✅ Success: {summary.success} | "
+ f"❌ Failed: {summary.failed} | "
+ f"⏭️ Skipped: {summary.skipped}"
+ )
+
+ if summary.failed_packages:
+ base_content += f"\n\nFailed: {', '.join(summary.failed_packages)}"
+
+ if self.config.type == "discord":
+ return {
+ "content": base_content,
+ "embeds": [{
+ "title": f"Build Report - {repo_name}",
+ "color": 0xFF0000 if summary.failed else 0x00FF00,
+ "fields": [
+ {"name": "Success", "value": str(summary.success), "inline": True},
+ {"name": "Failed", "value": str(summary.failed), "inline": True},
+ {"name": "Skipped", "value": str(summary.skipped), "inline": True},
+ ],
+ "timestamp": summary.timestamp.isoformat(),
+ }],
+ }
+
+ elif self.config.type == "slack":
+ return {
+ "text": base_content,
+ "blocks": [
+ {
+ "type": "section",
+ "text": {"type": "mrkdwn", "text": base_content},
+ }
+ ],
+ }
+
+ elif self.config.type == "ntfy":
+ return {
+ "topic": repo_name,
+ "title": f"Build Report - {repo_name}",
+ "message": base_content,
+ "priority": 5 if summary.failed else 3,
+ "tags": ["package", "failed"] if summary.failed else ["package"],
+ }
+
+ else: # generic
+ return {
+ "repository": repo_name,
+ "summary": {
+ "total": summary.total,
+ "success": summary.success,
+ "failed": summary.failed,
+ "skipped": summary.skipped,
+ },
+ "failed_packages": summary.failed_packages,
+ "timestamp": summary.timestamp.isoformat(),
+ }
+
+ async def send(self, summary: BuildSummary, config: Config) -> bool:
+ """Send webhook notification."""
+ if not self.config.enabled:
+ return True
+
+ if not self.config.url:
+ logger.warning("Webhook notification enabled but no URL configured")
+ return False
+
+ # Only send on failures (configurable in future)
+ if summary.failed == 0:
+ logger.debug("No failures, skipping webhook notification")
+ return True
+
+ try:
+ payload = self._format_payload(summary, config.repository.name)
+
+ async with aiohttp.ClientSession() as session:
+ async with session.post(
+ self.config.url,
+ json=payload,
+ timeout=aiohttp.ClientTimeout(total=30),
+ ) as response:
+ response.raise_for_status()
+
+ logger.info(f"Sent webhook notification to {self.config.url}")
+ return True
+
+ except Exception as e:
+ logger.error(f"Failed to send webhook: {e}")
+ return False
+
+
+class NotificationManager:
+ """Manage multiple notification backends."""
+
+ def __init__(self, config: Config):
+ """Initialize notification manager.
+
+ Args:
+ config: Application configuration
+ """
+ self.config = config
+ self.backends: list[NotificationBackend] = []
+
+ # Add email backend if configured
+ if config.notifications.email.enabled:
+ self.backends.append(EmailBackend(config.notifications.email))
+
+ # Add webhook backends
+ for webhook in config.notifications.webhooks:
+ if webhook.enabled:
+ self.backends.append(WebhookBackend(webhook))
+
+ async def notify(self, results: list[BuildResult]) -> dict[str, bool]:
+ """Send notifications for build results.
+
+ Args:
+ results: List of build results
+
+ Returns:
+ Dict mapping backend type to success status
+ """
+ summary = BuildSummary.from_results(results)
+ statuses: dict[str, bool] = {}
+
+ for backend in self.backends:
+ name = type(backend).__name__
+ try:
+ success = await backend.send(summary, self.config)
+ statuses[name] = success
+ except Exception as e:
+ logger.error(f"Notification backend {name} failed: {e}")
+ statuses[name] = False
+
+ return statuses
+
+ async def test(self) -> dict[str, bool]:
+ """Send test notification through all backends.
+
+ Returns:
+ Dict mapping backend type to success status
+ """
+ # Create dummy test summary
+ test_summary = BuildSummary(
+ total=3,
+ success=2,
+ failed=1,
+ skipped=0,
+ failed_packages=["test-package"],
+ duration=123.45,
+ timestamp=datetime.now(),
+ )
+
+ statuses: dict[str, bool] = {}
+ for backend in self.backends:
+ name = type(backend).__name__
+ try:
+ success = await backend.send(test_summary, self.config)
+ statuses[name] = success
+ except Exception as e:
+ logger.error(f"Test notification failed for {name}: {e}")
+ statuses[name] = False
+
+ return statuses
--
Gitblit v1.10.0