Add FileSystemMessagingIntegration and related tests

elementary/messages/messaging_integrations/__init__.py

@@ -0,0 +1,9 @@
+from .file_system import FileSystemMessagingIntegration
+from .slack_webhook import SlackWebhookMessagingIntegration
+from .teams_webhook import TeamsWebhookMessagingIntegration
+
+__all__ = [
+    "FileSystemMessagingIntegration",
+    "TeamsWebhookMessagingIntegration",
+    "SlackWebhookMessagingIntegration",
+]

elementary/messages/messaging_integrations/file_system.py

@@ -0,0 +1,72 @@
+import os
+from datetime import datetime
+
+from elementary.messages.message_body import MessageBody
+from elementary.messages.messaging_integrations.base_messaging_integration import (
+    BaseMessagingIntegration,
+    MessageSendResult,
+)
+from elementary.messages.messaging_integrations.empty_message_context import (
+    EmptyMessageContext,
+)
+from elementary.messages.messaging_integrations.exceptions import (
+    MessagingIntegrationError,
+)
+from elementary.utils.log import get_logger
+
+logger = get_logger(__name__)
+
+
+class FileSystemMessagingIntegration(
+    BaseMessagingIntegration[str, EmptyMessageContext]
+):
+    def __init__(self, directory: str, create_if_missing: bool = True) -> None:
+        self.directory = os.path.abspath(directory)
+        self._create_if_missing = create_if_missing
+
+        if not os.path.exists(self.directory):
+            if self._create_if_missing:
+                logger.info(
+                    "Creating directory for FileSystemMessagingIntegration: %s",
+                    self.directory,
+                )
+                os.makedirs(self.directory, exist_ok=True)
+            else:
+                raise MessagingIntegrationError(
+                    f"Directory {self.directory} does not exist and create_if_missing is False"
+                )
+
+    def supports_reply(self) -> bool:
+        return False
+
+    def send_message(
+        self, destination: str, body: MessageBody
+    ) -> MessageSendResult[EmptyMessageContext]:
+        file_path = os.path.join(self.directory, destination)
+
+        if not os.path.exists(file_path) and not self._create_if_missing:
+            raise MessagingIntegrationError(
+                f"File {file_path} does not exist and create_if_missing is False"
+            )
+
+        try:
+            logger.info("Writing alert message to file %s", file_path)
+            with open(file_path, "a", encoding="utf-8") as fp:
+                fp.write(body.json())
+                fp.write("\n")
+        except Exception as exc:
+            logger.error(
+                "Failed to write alert message to file %s: %s",
+                file_path,
+                exc,
+                exc_info=True,
+            )
+            raise MessagingIntegrationError(
+                f"Failed writing alert message to file {file_path}"
+            ) from exc
+
+        return MessageSendResult(
+            timestamp=datetime.utcnow(),
+            message_format="json",
+            message_context=EmptyMessageContext(),
+        )

tests/unit/messages/messaging_integrations/test_file_system.py

@@ -0,0 +1,61 @@
+import json
+from pathlib import Path
+
+import pytest
+
+from elementary.messages.blocks import LineBlock, LinesBlock, TextBlock
+from elementary.messages.message_body import MessageBody
+from elementary.messages.messaging_integrations.exceptions import (
+    MessagingIntegrationError,
+)
+from elementary.messages.messaging_integrations.file_system import (
+    FileSystemMessagingIntegration,
+)
+
+
+def _build_body() -> MessageBody:
+    return MessageBody(
+        blocks=[LinesBlock(lines=[LineBlock(inlines=[TextBlock(text="hello")])])]
+    )
+
+
+def test_send_message_creates_file_and_appends(tmp_path: Path) -> None:
+    directory = tmp_path / "alerts"
+    integ = FileSystemMessagingIntegration(directory=str(directory))
+    body = _build_body()
+
+    integ.send_message("channel.json", body)
+
+    target_file = directory / "channel.json"
+    assert target_file.exists()
+
+    with target_file.open() as fp:
+        lines = fp.readlines()
+    assert len(lines) == 1
+    assert json.loads(lines[0]) == json.loads(body.json())
+
+
+def test_send_multiple_messages(tmp_path: Path) -> None:
+    directory = tmp_path / "alerts"
+    integ = FileSystemMessagingIntegration(directory=str(directory))
+    body1 = _build_body()
+    body2 = _build_body()
+
+    integ.send_message("channel.json", body1)
+    integ.send_message("channel.json", body2)
+
+    target_file = directory / "channel.json"
+    with target_file.open() as fp:
+        lines = fp.readlines()
+
+    assert len(lines) == 2
+    assert json.loads(lines[0]) == json.loads(body1.json())
+    assert json.loads(lines[1]) == json.loads(body2.json())
+
+
+def test_send_message_no_create_flag(tmp_path: Path) -> None:
+    directory = tmp_path / "alerts-no-create"
+    with pytest.raises(MessagingIntegrationError):
+        FileSystemMessagingIntegration(
+            directory=str(directory), create_if_missing=False
+        )