EmailToDiscord/tests/test_app.py

import os
import signal

import app

import unittest
from unittest.mock import Mock, patch

from app import EmailMessage, EmailMonitor


EMAIL_ENV = {
    "EMAIL_USER": "user@example.com",
    "EMAIL_PASSWORD": "password",
}


class EmailMonitorConfigTests(unittest.TestCase):
    @patch.dict(
        os.environ,
        {**EMAIL_ENV, "DISCORD_WEBHOOK_URL": "https://discord.example/webhook"},
        clear=True,
    )
    def test_discord_only_configuration_remains_supported(self):
        monitor = EmailMonitor()

        self.assertEqual(monitor.discord_webhook_url, "https://discord.example/webhook")
        self.assertIsNone(monitor.webhook_url)

    @patch.dict(
        os.environ,
        {
            **EMAIL_ENV,
            "WEBHOOK_URL": "https://agents.example/webhook",
            "WEBHOOK_TOKEN": "123:signed-token",
            "DISCORD_CHANNEL_ID": "456",
        },
        clear=True,
    )
    def test_agents_webhook_only_configuration_is_supported(self):
        monitor = EmailMonitor()

        self.assertIsNone(monitor.discord_webhook_url)
        self.assertEqual(monitor.discord_channel_id, 456)

    @patch.dict(
        os.environ,
        {**EMAIL_ENV, "WEBHOOK_URL": "https://agents.example/webhook"},
        clear=True,
    )
    def test_partial_agents_webhook_configuration_is_rejected(self):
        with self.assertRaises(SystemExit):
            EmailMonitor()


class EmailMonitorNotificationTests(unittest.TestCase):
    def setUp(self):
        self.email = EmailMessage(
            subject="テスト件名",
            sender="sender@example.com",
            date="Fri, 18 Jul 2026 12:00:00 +0900",
            body="メール本文",
            uid="42",
        )

    @patch.dict(
        os.environ,
        {**EMAIL_ENV, "DISCORD_WEBHOOK_URL": "https://discord.example/webhook"},
        clear=True,
    )
    @patch("app.requests.post")
    def test_existing_discord_delivery_remains_supported(self, post):
        post.return_value = Mock(status_code=204, text="")
        monitor = EmailMonitor()

        self.assertTrue(monitor.send_to_discord(self.email))

        post.assert_called_once()
        self.assertEqual(post.call_args.args[0], "https://discord.example/webhook")
        embed = post.call_args.kwargs["json"]["embeds"][0]
        self.assertEqual(embed["fields"][0]["value"], "テスト件名")

    @patch.dict(
        os.environ,
        {
            **EMAIL_ENV,
            "WEBHOOK_URL": "https://agents.example/webhook",
            "WEBHOOK_TOKEN": "123:signed-token",
            "DISCORD_CHANNEL_ID": "456",
            "WEBHOOK_SOURCE": "email-satellite",
        },
        clear=True,
    )
    @patch("app.requests.post")
    def test_agents_webhook_payload_matches_agents_contract(self, post):
        response = Mock(status_code=202, text="Accepted")
        post.return_value = response
        monitor = EmailMonitor()

        self.assertTrue(monitor.send_to_agents_webhook(self.email))

        post.assert_called_once()
        url = post.call_args.args[0]
        kwargs = post.call_args.kwargs
        self.assertEqual(url, "https://agents.example/webhook")
        self.assertEqual(kwargs["headers"]["Authorization"], "Bearer 123:signed-token")
        self.assertEqual(kwargs["json"]["source"], "email-satellite")
        self.assertEqual(kwargs["json"]["channel_id"], 456)
        self.assertIn("テスト件名", kwargs["json"]["prompt"])
        self.assertIn("メール本文", kwargs["json"]["prompt"])
        self.assertIn("信頼できない外部入力", kwargs["json"]["prompt"])

    @patch.dict(
        os.environ,
        {
            **EMAIL_ENV,
            "DISCORD_WEBHOOK_URL": "https://discord.example/webhook",
            "WEBHOOK_URL": "https://agents.example/webhook",
            "WEBHOOK_TOKEN": "123:signed-token",
            "DISCORD_CHANNEL_ID": "456",
        },
        clear=True,
    )
    def test_dual_mode_requires_both_destinations_to_succeed(self):
        monitor = EmailMonitor()

        with patch.object(monitor, "send_to_discord", return_value=True), patch.object(
            monitor, "send_to_agents_webhook", return_value=False
        ):
            self.assertFalse(monitor.send_notifications(self.email))

        with patch.object(monitor, "send_to_discord", return_value=True), patch.object(
            monitor, "send_to_agents_webhook", return_value=True
        ):
            self.assertTrue(monitor.send_notifications(self.email))

    @patch.dict(
        os.environ,
        {**EMAIL_ENV, "DISCORD_WEBHOOK_URL": "https://discord.example/webhook"},
        clear=True,
    )
    @patch("app.time.sleep")
    def test_email_is_marked_read_only_after_notifications_succeed(self, _sleep):
        monitor = EmailMonitor()

        with patch.object(monitor, "send_notifications", return_value=False), patch.object(
            monitor, "mark_email_as_read"
        ) as mark_read:
            monitor.send_notifications_with_rate_limit([self.email], rate_limit=0)
            mark_read.assert_not_called()

        with patch.object(monitor, "send_notifications", return_value=True), patch.object(
            monitor, "mark_email_as_read"
        ) as mark_read:
            monitor.send_notifications_with_rate_limit([self.email], rate_limit=0)
            mark_read.assert_called_once_with("42")


class ApplicationStartupTests(unittest.TestCase):
    @patch.dict(os.environ, {**EMAIL_ENV, "DISCORD_WEBHOOK_URL": "https://discord.example/webhook"}, clear=True)
    @patch("app.run_http_server")
    @patch("app.threading.Thread")
    @patch("app.signal.signal")
    def test_main_registers_signal_handlers_and_starts_services(
        self, register_signal, thread_class, run_http_server
    ):
        monitor_thread = thread_class.return_value

        app.main()

        registered_signals = [call.args[0] for call in register_signal.call_args_list]
        self.assertEqual(registered_signals, [signal.SIGINT, signal.SIGTERM])
        monitor_thread.start.assert_called_once_with()
        run_http_server.assert_called_once_with()


if __name__ == "__main__":
    unittest.main()