diff --git a/meshai/config.py b/meshai/config.py index 7aa291d..9ee8c43 100644 --- a/meshai/config.py +++ b/meshai/config.py @@ -499,6 +499,53 @@ class EnvironmentalConfig: geocoder: GeocoderConfig = field(default_factory=GeocoderConfig) +@dataclass +class SinkConfig: + """Named notification sink — transport defined once, referenced by name. + + Phase A of routing revamp: transports defined once in sinks block, + referenced by name in toggles/rules (Phase B). + """ + + type: str = "mesh_broadcast" # mesh_broadcast|mesh_dm|email|webhook + + # Mesh broadcast + channel: int = 0 # Channel index (>= 0, 0 is valid primary channel) + + # Mesh DM + node_ids: list = field(default_factory=list) + + # Email (SMTP) + smtp_host: str = "" + smtp_port: int = 587 + smtp_user: str = "" + smtp_password: str = "" + smtp_tls: bool = True + from_address: str = "" + recipients: list = field(default_factory=list) + + # Webhook + webhook_url: str = "" + webhook_headers: dict = field(default_factory=dict) + + def validate(self) -> list[str]: + """Validate sink config, return list of errors.""" + errors = [] + if self.type not in ("mesh_broadcast", "mesh_dm", "email", "webhook"): + errors.append(f"Invalid sink type: {self.type}") + if self.type == "mesh_broadcast" and self.channel < 0: + errors.append(f"Channel must be >= 0, got {self.channel}") + if self.type == "mesh_dm" and not self.node_ids: + errors.append("mesh_dm sink requires node_ids") + if self.type == "email" and not self.smtp_host: + errors.append("email sink requires smtp_host") + if self.type == "email" and not self.recipients: + errors.append("email sink requires recipients") + if self.type == "webhook" and not self.webhook_url: + errors.append("webhook sink requires webhook_url") + return errors + + @dataclass class NotificationRuleConfig: """Self-contained notification rule with inline delivery config.""" @@ -642,6 +689,7 @@ class NotificationsConfig: toggles: dict = field(default_factory=_default_toggles) # family -> NotificationToggle digest: DigestConfig = field(default_factory=DigestConfig) rules: list = field(default_factory=list) # List of NotificationRuleConfig + sinks: dict = field(default_factory=dict) # name -> SinkConfig @dataclass class DashboardConfig: @@ -785,6 +833,11 @@ def _dict_to_dataclass(cls, data: dict): name: _dict_to_dataclass(NotificationToggle, t) if isinstance(t, dict) else t for name, t in value["toggles"].items() } + if "sinks" in value and isinstance(value["sinks"], dict): + notifications.sinks = { + name: _dict_to_dataclass(SinkConfig, s) if isinstance(s, dict) else s + for name, s in value["sinks"].items() + } if "channels" in value and isinstance(value["channels"], list) and value["channels"]: _migrate_legacy_channels(notifications, value) kwargs[key] = notifications diff --git a/meshai/dashboard/api/notification_routes.py b/meshai/dashboard/api/notification_routes.py index a3f0e55..fb61dc9 100644 --- a/meshai/dashboard/api/notification_routes.py +++ b/meshai/dashboard/api/notification_routes.py @@ -33,6 +33,11 @@ class ChannelTestRequest(BaseModel): headers: Optional[Dict[str, str]] = {} +class SinkTestRequest(BaseModel): + """Request body for sink connectivity test.""" + name: str # Sink name from config + + class RuleSourcesRequest(BaseModel): """Request body for rule sources health check.""" categories: List[str] = [] @@ -303,3 +308,94 @@ async def send_rule_live(request: Request, rule_index: int): ) return result + + +# ============================================================================= +# SINKS ENDPOINTS (Phase A of routing revamp) +# ============================================================================= + +@router.get("/sinks") +async def get_sinks(request: Request): + """Get configured notification sinks. + + Returns list of named sinks with their type and config. + Phase A: read-only list; Phase B adds edit endpoints. + """ + config = getattr(request.app.state, "config", None) + if not config or not hasattr(config, "notifications"): + return [] + + sinks = getattr(config.notifications, "sinks", {}) + if not sinks: + return [] + + result = [] + for name, sink in sinks.items(): + # Convert dataclass to dict if needed + if hasattr(sink, "__dataclass_fields__"): + sink_dict = { + "name": name, + "type": sink.type, + "channel": getattr(sink, "channel", 0), + "node_ids": getattr(sink, "node_ids", []), + "smtp_host": getattr(sink, "smtp_host", ""), + "smtp_port": getattr(sink, "smtp_port", 587), + "from_address": getattr(sink, "from_address", ""), + "recipients": getattr(sink, "recipients", []), + "webhook_url": getattr(sink, "webhook_url", ""), + } + else: + sink_dict = {"name": name, **sink} + + result.append(sink_dict) + + return result + + +@router.post("/sinks/test") +async def test_sink(request: Request, body: SinkTestRequest): + """Test a named sink's connectivity. + + Uses the existing channel test_connection() method. + """ + config = getattr(request.app.state, "config", None) + if not config or not hasattr(config, "notifications"): + raise HTTPException(status_code=404, detail="Notifications not configured") + + sinks = getattr(config.notifications, "sinks", {}) + if not sinks: + raise HTTPException(status_code=404, detail="No sinks configured") + + if body.name not in sinks: + raise HTTPException(status_code=404, detail=f"Sink not found: {body.name}") + + sink = sinks[body.name] + + # Get connector for mesh channels + connector = getattr(request.app.state, "connector", None) + + try: + from ...notifications.channels import create_channel_from_sink + channel = create_channel_from_sink(sink, connector=connector) + result = await channel.test_connection() + return { + "sink": body.name, + "type": sink.type, + **result + } + except ValueError as e: + return { + "sink": body.name, + "type": getattr(sink, "type", "unknown"), + "success": False, + "message": str(e), + "error": str(e), + } + except Exception as e: + return { + "sink": body.name, + "type": getattr(sink, "type", "unknown"), + "success": False, + "message": f"Test failed: {str(e)}", + "error": str(e), + } diff --git a/meshai/notifications/channels.py b/meshai/notifications/channels.py index 6ac39ff..08113de 100644 --- a/meshai/notifications/channels.py +++ b/meshai/notifications/channels.py @@ -14,7 +14,7 @@ import httpx if TYPE_CHECKING: from ..connector import MeshConnector - from ..config import NotificationRuleConfig + from ..config import NotificationRuleConfig, SinkConfig from .events import NotificationPayload from meshai.notifications.renderers import MeshRenderer, EmailRenderer, WebhookRenderer @@ -837,3 +837,51 @@ def create_channel_from_dict(config: dict, connector=None) -> NotificationChanne ) else: raise ValueError("Unknown channel type: %s" % channel_type) + +def create_channel_from_sink(sink: "SinkConfig", connector=None) -> NotificationChannel: + """Create a channel instance from a SinkConfig dataclass. + + Phase A of routing revamp: sinks are named transports defined once, + this factory creates the channel instance for delivery. + + Args: + sink: SinkConfig dataclass instance + connector: MeshConnector instance (required for mesh channels) + + Returns: + NotificationChannel instance + + Raises: + ValueError: If sink type is unknown or channel < 0 + """ + sink_type = sink.type + + if sink_type == "mesh_broadcast": + if sink.channel < 0: + raise ValueError(f"Channel must be >= 0, got {sink.channel}") + return MeshBroadcastChannel( + connector=connector, + channel_index=sink.channel, + ) + elif sink_type == "mesh_dm": + return MeshDMChannel( + connector=connector, + node_ids=sink.node_ids, + ) + elif sink_type == "email": + return EmailChannel( + smtp_host=sink.smtp_host, + smtp_port=sink.smtp_port, + smtp_user=sink.smtp_user, + smtp_password=sink.smtp_password, + smtp_tls=sink.smtp_tls, + from_address=sink.from_address, + recipients=sink.recipients, + ) + elif sink_type == "webhook": + return WebhookChannel( + url=sink.webhook_url, + headers=sink.webhook_headers, + ) + else: + raise ValueError(f"Unknown sink type: {sink_type}") diff --git a/meshai/scripts/migrate_config_routing.py b/meshai/scripts/migrate_config_routing.py new file mode 100644 index 0000000..b4e4baf --- /dev/null +++ b/meshai/scripts/migrate_config_routing.py @@ -0,0 +1,349 @@ +#!/usr/bin/env python3 +"""Migration script for MeshAI routing revamp Phase A: synthesize sinks. + +This script reads existing notification toggles and rules, extracts their +inline transport configurations, and synthesizes named sinks. + +Run manually: python -m meshai.scripts.migrate_config_routing [--dry-run] + +The migration: +1. Backs up the config to .pre-sinks..bak +2. For each toggle with inline transport config, synthesizes a named sink +3. For each enabled rule with inline transport config, synthesizes a named sink +4. Deduplicates identical transports into one sink +5. Writes the sinks block to the config +6. Does NOT remove inline fields (Phase B does that) + +Idempotent: refuses to run if a sinks block already exists. +""" + +import argparse +import hashlib +import json +import logging +import os +import sys +import time +from pathlib import Path +from typing import Any, Optional +from urllib.parse import urlparse + +import yaml + +# Setup logging +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger(__name__) + + +def compute_sink_hash(sink_dict: dict) -> str: + """Compute a stable hash of sink config for deduplication.""" + # Sort keys for stable comparison + canonical = json.dumps(sink_dict, sort_keys=True) + return hashlib.sha256(canonical.encode()).hexdigest()[:8] + + +def generate_sink_name(sink_type: str, sink_dict: dict) -> str: + """Generate a human-readable sink name from its config.""" + if sink_type == "mesh_broadcast": + channel = sink_dict.get("channel", 0) + return f"mesh-ch{channel}" + elif sink_type == "mesh_dm": + node_ids = sink_dict.get("node_ids", []) + if node_ids: + first_node = node_ids[0].lstrip("!")[:8] + return f"dm-{first_node}" + return "dm-unknown" + elif sink_type == "email": + host = sink_dict.get("smtp_host", "") + if host: + # Extract domain + return f"email-{host.split('.')[0]}" + return "email-unknown" + elif sink_type == "webhook": + url = sink_dict.get("webhook_url", "") + if url: + parsed = urlparse(url) + return f"webhook-{parsed.netloc.split('.')[0]}" + return "webhook-unknown" + return f"sink-{sink_type}" + + +def extract_sinks_from_toggle(toggle: dict) -> list[dict]: + """Extract ALL sink configs from a NotificationToggle's inline fields. + + Returns a list of sink dicts, one per configured transport type. + No precedence — every transport with non-empty config is extracted. + """ + sinks = [] + + # Check for mesh_broadcast (broadcast_channel field) + broadcast_channel = toggle.get("broadcast_channel") + if broadcast_channel is not None: + sinks.append({ + "type": "mesh_broadcast", + "channel": int(broadcast_channel), + }) + + # Check for mesh_dm (node_ids field) + node_ids = toggle.get("node_ids", []) + if node_ids: + sinks.append({ + "type": "mesh_dm", + "node_ids": node_ids, + }) + + # Check for email (smtp_host field) + smtp_host = toggle.get("smtp_host", "") + if smtp_host: + sinks.append({ + "type": "email", + "smtp_host": smtp_host, + "smtp_port": toggle.get("smtp_port", 587), + "smtp_user": toggle.get("smtp_user", ""), + "smtp_password": toggle.get("smtp_password", ""), + "smtp_tls": toggle.get("smtp_tls", True), + "from_address": toggle.get("from_address", ""), + "recipients": toggle.get("recipients", []), + }) + + # Check for webhook (webhook_url field) + webhook_url = toggle.get("webhook_url", "") + if webhook_url: + sinks.append({ + "type": "webhook", + "webhook_url": webhook_url, + "webhook_headers": toggle.get("webhook_headers", {}), + }) + + return sinks + + +def extract_sink_from_toggle(toggle: dict) -> Optional[dict]: + """Legacy wrapper - returns first sink or None. Use extract_sinks_from_toggle instead.""" + sinks = extract_sinks_from_toggle(toggle) + return sinks[0] if sinks else None + + +def extract_sink_from_rule(rule: dict) -> Optional[dict]: + """Extract sink config from a NotificationRuleConfig's inline fields.""" + delivery_type = rule.get("delivery_type", "") + + if delivery_type == "mesh_broadcast": + return { + "type": "mesh_broadcast", + "channel": rule.get("broadcast_channel", 0), + } + elif delivery_type == "mesh_dm": + return { + "type": "mesh_dm", + "node_ids": rule.get("node_ids", []), + } + elif delivery_type == "email": + return { + "type": "email", + "smtp_host": rule.get("smtp_host", ""), + "smtp_port": rule.get("smtp_port", 587), + "smtp_user": rule.get("smtp_user", ""), + "smtp_password": rule.get("smtp_password", ""), + "smtp_tls": rule.get("smtp_tls", True), + "from_address": rule.get("from_address", ""), + "recipients": rule.get("recipients", []), + } + elif delivery_type == "webhook": + return { + "type": "webhook", + "webhook_url": rule.get("webhook_url", ""), + "webhook_headers": rule.get("webhook_headers", {}), + } + + return None + + +def synthesize_sinks(notifications: dict) -> dict: + """Synthesize named sinks from toggles and rules. + + Returns: + dict mapping sink names to sink configs + """ + sinks = {} + hash_to_name = {} # For deduplication + + # Process toggles - extract ALL configured transports per toggle + toggles = notifications.get("toggles", {}) + for toggle_name, toggle in toggles.items(): + if not isinstance(toggle, dict): + continue + + sink_dicts = extract_sinks_from_toggle(toggle) + if not sink_dicts: + continue + + for sink_dict in sink_dicts: + sink_hash = compute_sink_hash(sink_dict) + if sink_hash in hash_to_name: + logger.info(f" Toggle '{toggle_name}' reuses existing sink '{hash_to_name[sink_hash]}'") + continue + + sink_name = generate_sink_name(sink_dict["type"], sink_dict) + # Handle name collisions + base_name = sink_name + counter = 2 + while sink_name in sinks: + sink_name = f"{base_name}-{counter}" + counter += 1 + + sinks[sink_name] = sink_dict + hash_to_name[sink_hash] = sink_name + logger.info(f" Toggle '{toggle_name}' → sink '{sink_name}'") + + # Process rules + rules = notifications.get("rules", []) + for i, rule in enumerate(rules): + if not isinstance(rule, dict): + continue + + # Only process enabled rules + if not rule.get("enabled", True): + continue + + sink_dict = extract_sink_from_rule(rule) + if not sink_dict: + continue + + sink_hash = compute_sink_hash(sink_dict) + if sink_hash in hash_to_name: + rule_name = rule.get("name", f"rule-{i}") + logger.info(f" Rule '{rule_name}' reuses existing sink '{hash_to_name[sink_hash]}'") + continue + + sink_name = generate_sink_name(sink_dict["type"], sink_dict) + # Handle name collisions + base_name = sink_name + counter = 2 + while sink_name in sinks: + sink_name = f"{base_name}-{counter}" + counter += 1 + + sinks[sink_name] = sink_dict + hash_to_name[sink_hash] = sink_name + rule_name = rule.get("name", f"rule-{i}") + logger.info(f" Rule '{rule_name}' → sink '{sink_name}'") + + return sinks + + +def load_notifications_config(config_path: Path) -> tuple[dict, dict]: + """Load notifications config from file. + + Returns: + (full_config_dict, notifications_dict) + """ + with open(config_path, "r") as f: + config = yaml.safe_load(f) or {} + + # Handle both monolithic (has notifications: key) and multi-file (IS notifications) layouts + if "notifications" in config: + notifications = config["notifications"] + elif "toggles" in config or "rules" in config: + notifications = config + else: + notifications = {} + return config, notifications + + +def backup_config(config_path: Path) -> Path: + """Create a timestamped backup of the config file.""" + epoch = int(time.time()) + backup_path = config_path.with_suffix(f".pre-sinks.{epoch}.bak") + import shutil + shutil.copy2(config_path, backup_path) + return backup_path + + +def write_sinks_to_config(config_path: Path, sinks: dict): + """Write synthesized sinks block to the config file.""" + with open(config_path, "r") as f: + content = f.read() + + # Parse YAML to find where to insert + # Insert sinks after notifications.rules or at end of notifications block + # This is a simplified approach - in production you'd want proper YAML manipulation + + # For now, read as dict, add sinks, write back + config = yaml.safe_load(content) or {} + + if "notifications" not in config: + config["notifications"] = {} + + config["notifications"]["sinks"] = sinks + + with open(config_path, "w") as f: + yaml.dump(config, f, default_flow_style=False, sort_keys=False, allow_unicode=True) + + +def main(): + parser = argparse.ArgumentParser( + description="Migrate MeshAI config to use named sinks (Phase A)" + ) + parser.add_argument( + "--config", + type=Path, + default=Path("/data/config/notifications.yaml"), + help="Path to notifications config file", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be done without making changes", + ) + args = parser.parse_args() + + config_path = args.config + + if not config_path.exists(): + logger.error(f"Config file not found: {config_path}") + sys.exit(1) + + logger.info(f"Loading config from {config_path}") + full_config, notifications = load_notifications_config(config_path) + + # Check if sinks already exist + if notifications.get("sinks"): + logger.error("Sinks block already exists in config. Migration already complete.") + logger.info("Existing sinks: %s", list(notifications["sinks"].keys())) + sys.exit(1) + + # Synthesize sinks + logger.info("Synthesizing sinks from toggles and rules...") + sinks = synthesize_sinks(notifications) + + if not sinks: + logger.info("No sinks to synthesize (no inline transport configs found)") + sys.exit(0) + + logger.info(f"Synthesized {len(sinks)} sink(s):") + for name, sink in sinks.items(): + logger.info(f" {name}: {sink}") + + if args.dry_run: + logger.info("DRY RUN - no changes made") + print("\n--- Would write sinks block: ---") + print(yaml.dump({"sinks": sinks}, default_flow_style=False)) + return + + # Backup and write + backup_path = backup_config(config_path) + logger.info(f"Backed up config to {backup_path}") + + write_sinks_to_config(config_path, sinks) + logger.info(f"Wrote sinks block to {config_path}") + + logger.info("Migration complete!") + + +if __name__ == "__main__": + main() diff --git a/tests/test_sinks.py b/tests/test_sinks.py new file mode 100644 index 0000000..d5f0cc9 --- /dev/null +++ b/tests/test_sinks.py @@ -0,0 +1,491 @@ +"""Tests for Phase A routing revamp: SinkConfig and sink utilities. + +Tests cover: +1. SinkConfig dataclass conversion from dict +2. Channel factory per sink type +3. Migration synthesis + idempotence + dry-run +""" + +import pytest +import tempfile +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import yaml + + +class TestSinkConfigDataclass: + """Tests for SinkConfig dataclass and dict conversion.""" + + def test_sink_config_defaults(self): + """SinkConfig has correct defaults.""" + from meshai.config import SinkConfig + + sink = SinkConfig() + assert sink.type == "mesh_broadcast" + assert sink.channel == 0 + assert sink.node_ids == [] + assert sink.smtp_host == "" + assert sink.recipients == [] + assert sink.webhook_url == "" + + def test_sink_config_mesh_broadcast(self): + """SinkConfig correctly stores mesh_broadcast config.""" + from meshai.config import SinkConfig + + sink = SinkConfig(type="mesh_broadcast", channel=2) + assert sink.type == "mesh_broadcast" + assert sink.channel == 2 + + def test_sink_config_mesh_dm(self): + """SinkConfig correctly stores mesh_dm config.""" + from meshai.config import SinkConfig + + sink = SinkConfig(type="mesh_dm", node_ids=["!abc123", "!def456"]) + assert sink.type == "mesh_dm" + assert sink.node_ids == ["!abc123", "!def456"] + + def test_sink_config_email(self): + """SinkConfig correctly stores email config.""" + from meshai.config import SinkConfig + + sink = SinkConfig( + type="email", + smtp_host="smtp.example.com", + smtp_port=465, + smtp_user="user", + smtp_password="pass", + smtp_tls=True, + from_address="alerts@example.com", + recipients=["ops@example.com"], + ) + assert sink.type == "email" + assert sink.smtp_host == "smtp.example.com" + assert sink.smtp_port == 465 + assert sink.recipients == ["ops@example.com"] + + def test_sink_config_webhook(self): + """SinkConfig correctly stores webhook config.""" + from meshai.config import SinkConfig + + sink = SinkConfig( + type="webhook", + webhook_url="https://hooks.example.com/alert", + webhook_headers={"Authorization": "Bearer token"}, + ) + assert sink.type == "webhook" + assert sink.webhook_url == "https://hooks.example.com/alert" + assert sink.webhook_headers == {"Authorization": "Bearer token"} + + def test_sink_config_validation_valid(self): + """SinkConfig.validate() returns empty list for valid configs.""" + from meshai.config import SinkConfig + + # Valid mesh_broadcast + sink = SinkConfig(type="mesh_broadcast", channel=0) + assert sink.validate() == [] + + # Valid mesh_dm + sink = SinkConfig(type="mesh_dm", node_ids=["!abc123"]) + assert sink.validate() == [] + + # Valid email + sink = SinkConfig(type="email", smtp_host="smtp.test.com", recipients=["a@b.com"]) + assert sink.validate() == [] + + # Valid webhook + sink = SinkConfig(type="webhook", webhook_url="https://example.com") + assert sink.validate() == [] + + def test_sink_config_validation_invalid_type(self): + """SinkConfig.validate() catches invalid type.""" + from meshai.config import SinkConfig + + sink = SinkConfig(type="invalid_type") + errors = sink.validate() + assert len(errors) == 1 + assert "Invalid sink type" in errors[0] + + def test_sink_config_validation_negative_channel(self): + """SinkConfig.validate() catches negative channel.""" + from meshai.config import SinkConfig + + sink = SinkConfig(type="mesh_broadcast", channel=-1) + errors = sink.validate() + assert len(errors) == 1 + assert "must be >= 0" in errors[0] + + def test_sink_config_validation_channel_zero_valid(self): + """SinkConfig.validate() accepts channel 0 (B6 fix verification).""" + from meshai.config import SinkConfig + + sink = SinkConfig(type="mesh_broadcast", channel=0) + errors = sink.validate() + assert errors == [] + + def test_dict_to_dataclass_converts_sinks(self): + """_dict_to_dataclass correctly converts sinks dict to SinkConfig instances.""" + from meshai.config import _dict_to_dataclass, Config, SinkConfig + + config_dict = { + "notifications": { + "enabled": True, + "sinks": { + "mesh-primary": {"type": "mesh_broadcast", "channel": 0}, + "mesh-alerts": {"type": "mesh_broadcast", "channel": 2}, + "dm-ops": {"type": "mesh_dm", "node_ids": ["!abcd1234"]}, + }, + }, + } + + config = _dict_to_dataclass(Config, config_dict) + + assert hasattr(config.notifications, "sinks") + sinks = config.notifications.sinks + + assert "mesh-primary" in sinks + assert isinstance(sinks["mesh-primary"], SinkConfig) + assert sinks["mesh-primary"].type == "mesh_broadcast" + assert sinks["mesh-primary"].channel == 0 + + assert "mesh-alerts" in sinks + assert isinstance(sinks["mesh-alerts"], SinkConfig) + assert sinks["mesh-alerts"].channel == 2 + + assert "dm-ops" in sinks + assert isinstance(sinks["dm-ops"], SinkConfig) + assert sinks["dm-ops"].type == "mesh_dm" + assert sinks["dm-ops"].node_ids == ["!abcd1234"] + + +class TestCreateChannelFromSink: + """Tests for create_channel_from_sink factory function.""" + + def test_create_mesh_broadcast_channel(self): + """create_channel_from_sink creates MeshBroadcastChannel.""" + from meshai.config import SinkConfig + from meshai.notifications.channels import ( + create_channel_from_sink, + MeshBroadcastChannel, + ) + + sink = SinkConfig(type="mesh_broadcast", channel=2) + mock_connector = MagicMock() + + channel = create_channel_from_sink(sink, connector=mock_connector) + + assert isinstance(channel, MeshBroadcastChannel) + assert channel._channel == 2 + assert channel._connector == mock_connector + + def test_create_mesh_dm_channel(self): + """create_channel_from_sink creates MeshDMChannel.""" + from meshai.config import SinkConfig + from meshai.notifications.channels import ( + create_channel_from_sink, + MeshDMChannel, + ) + + sink = SinkConfig(type="mesh_dm", node_ids=["!abc123", "!def456"]) + mock_connector = MagicMock() + + channel = create_channel_from_sink(sink, connector=mock_connector) + + assert isinstance(channel, MeshDMChannel) + assert channel._node_ids == ["!abc123", "!def456"] + + def test_create_email_channel(self): + """create_channel_from_sink creates EmailChannel.""" + from meshai.config import SinkConfig + from meshai.notifications.channels import ( + create_channel_from_sink, + EmailChannel, + ) + + sink = SinkConfig( + type="email", + smtp_host="smtp.test.com", + smtp_port=587, + smtp_user="user", + smtp_password="pass", + smtp_tls=True, + from_address="alerts@test.com", + recipients=["ops@test.com"], + ) + + channel = create_channel_from_sink(sink) + + assert isinstance(channel, EmailChannel) + assert channel._host == "smtp.test.com" + assert channel._port == 587 + assert channel._recipients == ["ops@test.com"] + + def test_create_webhook_channel(self): + """create_channel_from_sink creates WebhookChannel.""" + from meshai.config import SinkConfig + from meshai.notifications.channels import ( + create_channel_from_sink, + WebhookChannel, + ) + + sink = SinkConfig( + type="webhook", + webhook_url="https://hooks.test.com/alert", + webhook_headers={"X-Token": "secret"}, + ) + + channel = create_channel_from_sink(sink) + + assert isinstance(channel, WebhookChannel) + assert channel._url == "https://hooks.test.com/alert" + assert channel._headers == {"X-Token": "secret"} + + def test_create_channel_invalid_type_raises(self): + """create_channel_from_sink raises ValueError for invalid type.""" + from meshai.config import SinkConfig + from meshai.notifications.channels import create_channel_from_sink + + sink = SinkConfig(type="invalid_type") + + with pytest.raises(ValueError, match="Unknown sink type"): + create_channel_from_sink(sink) + + def test_create_channel_negative_channel_raises(self): + """create_channel_from_sink raises ValueError for negative channel.""" + from meshai.config import SinkConfig + from meshai.notifications.channels import create_channel_from_sink + + sink = SinkConfig(type="mesh_broadcast", channel=-1) + + with pytest.raises(ValueError, match="must be >= 0"): + create_channel_from_sink(sink) + + def test_create_channel_zero_channel_valid(self): + """create_channel_from_sink accepts channel 0 (B6 fix verification).""" + from meshai.config import SinkConfig + from meshai.notifications.channels import ( + create_channel_from_sink, + MeshBroadcastChannel, + ) + + sink = SinkConfig(type="mesh_broadcast", channel=0) + mock_connector = MagicMock() + + channel = create_channel_from_sink(sink, connector=mock_connector) + + assert isinstance(channel, MeshBroadcastChannel) + assert channel._channel == 0 + + +class TestMigrationSynthesis: + """Tests for migration script sink synthesis logic.""" + + def test_extract_sink_from_toggle_mesh_broadcast(self): + """extract_sink_from_toggle handles broadcast_channel.""" + from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle + + toggle = {"broadcast_channel": 2, "enabled": True} + sink = extract_sink_from_toggle(toggle) + + assert sink == {"type": "mesh_broadcast", "channel": 2} + + def test_extract_sink_from_toggle_mesh_dm(self): + """extract_sink_from_toggle handles node_ids.""" + from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle + + toggle = {"node_ids": ["!abc123"], "enabled": True} + sink = extract_sink_from_toggle(toggle) + + assert sink == {"type": "mesh_dm", "node_ids": ["!abc123"]} + + def test_extract_sink_from_toggle_email(self): + """extract_sink_from_toggle handles smtp_host.""" + from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle + + toggle = { + "smtp_host": "smtp.test.com", + "smtp_port": 587, + "smtp_user": "user", + "smtp_password": "pass", + "smtp_tls": True, + "from_address": "alerts@test.com", + "recipients": ["ops@test.com"], + } + sink = extract_sink_from_toggle(toggle) + + assert sink["type"] == "email" + assert sink["smtp_host"] == "smtp.test.com" + assert sink["recipients"] == ["ops@test.com"] + + def test_extract_sink_from_toggle_webhook(self): + """extract_sink_from_toggle handles webhook_url.""" + from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle + + toggle = { + "webhook_url": "https://hooks.test.com", + "webhook_headers": {"X-Token": "secret"}, + } + sink = extract_sink_from_toggle(toggle) + + assert sink == { + "type": "webhook", + "webhook_url": "https://hooks.test.com", + "webhook_headers": {"X-Token": "secret"}, + } + + def test_extract_sink_from_toggle_none(self): + """extract_sink_from_toggle returns None for toggle without transport.""" + from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle + + toggle = {"enabled": True, "min_severity": "priority"} + sink = extract_sink_from_toggle(toggle) + + assert sink is None + + + def test_extract_sinks_from_toggle_multiple_transports(self): + """extract_sinks_from_toggle returns ALL configured transports.""" + from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle + + toggle = { + "broadcast_channel": 1, + "node_ids": ["!abc123"], + "smtp_host": "", # Empty = not configured + "webhook_url": "", # Empty = not configured + } + sinks = extract_sinks_from_toggle(toggle) + + assert len(sinks) == 2 + types = {s["type"] for s in sinks} + assert types == {"mesh_broadcast", "mesh_dm"} + + def test_extract_sinks_from_toggle_all_four_types(self): + """extract_sinks_from_toggle extracts all four transport types.""" + from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle + + toggle = { + "broadcast_channel": 2, + "node_ids": ["!abc123"], + "smtp_host": "smtp.test.com", + "recipients": ["ops@test.com"], + "webhook_url": "https://hooks.test.com", + } + sinks = extract_sinks_from_toggle(toggle) + + assert len(sinks) == 4 + types = {s["type"] for s in sinks} + assert types == {"mesh_broadcast", "mesh_dm", "email", "webhook"} + + def test_extract_sink_from_rule(self): + """extract_sink_from_rule handles delivery_type.""" + from meshai.scripts.migrate_config_routing import extract_sink_from_rule + + rule = {"delivery_type": "mesh_broadcast", "broadcast_channel": 3} + sink = extract_sink_from_rule(rule) + + assert sink == {"type": "mesh_broadcast", "channel": 3} + + def test_generate_sink_name_mesh_broadcast(self): + """generate_sink_name creates readable names for mesh_broadcast.""" + from meshai.scripts.migrate_config_routing import generate_sink_name + + name = generate_sink_name("mesh_broadcast", {"channel": 2}) + assert name == "mesh-ch2" + + def test_generate_sink_name_mesh_dm(self): + """generate_sink_name creates readable names for mesh_dm.""" + from meshai.scripts.migrate_config_routing import generate_sink_name + + name = generate_sink_name("mesh_dm", {"node_ids": ["!abcd1234"]}) + assert name == "dm-abcd1234" + + def test_generate_sink_name_email(self): + """generate_sink_name creates readable names for email.""" + from meshai.scripts.migrate_config_routing import generate_sink_name + + name = generate_sink_name("email", {"smtp_host": "smtp.example.com"}) + assert name == "email-smtp" + + def test_generate_sink_name_webhook(self): + """generate_sink_name creates readable names for webhook.""" + from meshai.scripts.migrate_config_routing import generate_sink_name + + name = generate_sink_name("webhook", {"webhook_url": "https://hooks.slack.com/abc"}) + assert name == "webhook-hooks" + + def test_synthesize_sinks_deduplicates(self): + """synthesize_sinks deduplicates identical transports.""" + from meshai.scripts.migrate_config_routing import synthesize_sinks + + notifications = { + "toggles": { + "fire": {"broadcast_channel": 2}, + "weather": {"broadcast_channel": 2}, # Same channel + } + } + + sinks = synthesize_sinks(notifications) + + # Should only have one sink for channel 2 + assert len(sinks) == 1 + assert "mesh-ch2" in sinks + + def test_synthesize_sinks_handles_collisions(self): + """synthesize_sinks handles name collisions.""" + from meshai.scripts.migrate_config_routing import synthesize_sinks + + notifications = { + "toggles": { + "fire": {"broadcast_channel": 0}, + "weather": {"broadcast_channel": 1}, + "roads": {"broadcast_channel": 2}, + } + } + + sinks = synthesize_sinks(notifications) + + # Should have three unique sinks + assert len(sinks) == 3 + + +class TestMigrationIdempotence: + """Tests for migration script idempotence.""" + + def test_migration_refuses_if_sinks_exist(self): + """Migration refuses to run if sinks block already exists.""" + from meshai.scripts.migrate_config_routing import load_notifications_config + + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: + yaml.dump({ + "notifications": { + "enabled": True, + "sinks": {"mesh-primary": {"type": "mesh_broadcast"}}, + } + }, f) + config_path = Path(f.name) + + try: + _, notifications = load_notifications_config(config_path) + assert notifications.get("sinks") is not None + # The main() function checks this and exits + finally: + os.unlink(config_path) + + def test_backup_creates_timestamped_file(self): + """backup_config creates properly named backup.""" + from meshai.scripts.migrate_config_routing import backup_config + + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: + f.write("test: true\n") + config_path = Path(f.name) + + try: + backup_path = backup_config(config_path) + assert backup_path.exists() + assert ".pre-sinks." in str(backup_path) + assert backup_path.suffix == ".bak" + finally: + os.unlink(config_path) + if backup_path.exists(): + os.unlink(backup_path)