Nexus/tests/test_notify_toggle_sync.py

"""Tests for notify toggle helpers and cross-worker settings broadcast."""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from server.config import settings
from server.infrastructure.settings_broadcast import apply_setting_override, publish_setting_change
from server.infrastructure.telegram import send_telegram_alert, send_telegram_recovery, send_telegram_system_alert
from server.utils.notify_toggles import (
    notify_attr_enabled,
    notify_enabled,
    resource_alert_notify_enabled,
    resource_recovery_notify_enabled,
)


@pytest.mark.parametrize(
    ("value", "expected"),
    [
        ("true", True),
        ("false", False),
        ("0", False),
        ("off", False),
        ("no", False),
        (False, False),
        (True, True),
        (None, True),
    ],
)
def test_notify_enabled(value, expected):
    assert notify_enabled(value) is expected


def test_resource_alert_notify_enabled_respects_cpu_toggle():
    settings.NOTIFY_ALERT_CPU = "false"
    settings.NOTIFY_ALERT_MEM = "true"
    settings.NOTIFY_ALERT_DISK = "true"
    assert resource_alert_notify_enabled("cpu") is False
    assert resource_alert_notify_enabled("mem") is True
    assert resource_alert_notify_enabled("disk") is True


def test_resource_recovery_requires_both_toggles():
    settings.NOTIFY_RECOVERY = "true"
    settings.NOTIFY_ALERT_CPU = "false"
    settings.NOTIFY_ALERT_MEM = "true"
    settings.NOTIFY_ALERT_DISK = "false"
    assert resource_recovery_notify_enabled("cpu") is False
    assert resource_recovery_notify_enabled("mem") is True
    assert resource_recovery_notify_enabled("disk") is False

    settings.NOTIFY_RECOVERY = "false"
    assert resource_recovery_notify_enabled("mem") is False


@pytest.mark.parametrize("metric,attr", [
    ("cpu", "NOTIFY_ALERT_CPU"),
    ("mem", "NOTIFY_ALERT_MEM"),
    ("disk", "NOTIFY_ALERT_DISK"),
])
@pytest.mark.asyncio
async def test_send_telegram_alert_skipped_when_metric_disabled(metric, attr):
    setattr(settings, attr, "false")
    settings.TELEGRAM_BOT_TOKEN = "123:ABC"
    settings.TELEGRAM_CHAT_ID = "1"

    with patch("server.infrastructure.telegram.send_telegram", new_callable=AsyncMock) as mock_send:
        result = await send_telegram_alert("srv-1", metric, 95.0, 1)

    assert result is False
    mock_send.assert_not_awaited()


def test_apply_db_override_notify_toggle():
    settings.NOTIFY_ALERT_CPU = "true"
    assert apply_setting_override("notify_alert_cpu", "false") is True
    assert settings.NOTIFY_ALERT_CPU == "false"


def test_apply_db_override_unknown_key():
    assert apply_setting_override("nonexistent_key", "x") is False


@pytest.mark.asyncio
async def test_publish_setting_change():
    mock_redis = AsyncMock()
    mock_redis.publish = AsyncMock(return_value=1)

    with patch("server.infrastructure.redis.client.get_redis", return_value=mock_redis):
        ok = await publish_setting_change("notify_alert_cpu", "false")

    assert ok is True
    mock_redis.publish.assert_awaited_once()
    channel, payload = mock_redis.publish.await_args.args
    assert channel == "nexus:settings"
    assert '"notify_alert_cpu"' in payload


@pytest.mark.asyncio
async def test_send_telegram_recovery_skipped_when_disabled():
    settings.NOTIFY_RECOVERY = "false"

    with patch("server.infrastructure.telegram.send_telegram", new_callable=AsyncMock) as mock_send:
        result = await send_telegram_recovery("srv-1", "cpu", 50.0, 1)

    assert result is False
    mock_send.assert_not_awaited()


@pytest.mark.parametrize("metric,attr", [
    ("cpu", "NOTIFY_ALERT_CPU"),
    ("mem", "NOTIFY_ALERT_MEM"),
    ("disk", "NOTIFY_ALERT_DISK"),
])
@pytest.mark.asyncio
async def test_send_telegram_recovery_skipped_when_metric_alert_disabled(metric, attr):
    settings.NOTIFY_RECOVERY = "true"
    setattr(settings, attr, "false")

    with patch("server.infrastructure.telegram.send_telegram", new_callable=AsyncMock) as mock_send:
        result = await send_telegram_recovery("srv-1", metric, 50.0, 1)

    assert result is False
    mock_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_send_telegram_recovery_allowed_when_cpu_enabled():
    settings.NOTIFY_RECOVERY = "true"
    settings.NOTIFY_ALERT_CPU = "true"
    settings.TELEGRAM_BOT_TOKEN = "123:ABC"
    settings.TELEGRAM_CHAT_ID = "1"

    with patch("server.infrastructure.telegram.send_telegram", new_callable=AsyncMock) as mock_send:
        mock_send.return_value = True
        result = await send_telegram_recovery("srv-1", "cpu", 50.0, 1)

    assert result is True
    mock_send.assert_awaited_once()


@pytest.mark.parametrize("notify_key", ["NOTIFY_SYSTEM_REDIS", "NOTIFY_SYSTEM_MYSQL"])
@pytest.mark.asyncio
async def test_send_telegram_system_recovery_skipped_when_disabled(notify_key):
    setattr(settings, notify_key, "false")

    with patch("server.infrastructure.telegram.send_telegram", new_callable=AsyncMock) as mock_send:
        result = await send_telegram_system_alert(
            summary="连接已恢复",
            title="服务恢复",
            notify_key=notify_key,
            icon="🟢",
        )

    assert result is False
    mock_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_send_telegram_system_alert_skipped_when_disabled():
    settings.NOTIFY_TIME_DRIFT = "false"

    with patch("server.infrastructure.telegram.send_telegram", new_callable=AsyncMock) as mock_send:
        result = await send_telegram_system_alert(
            summary="test",
            notify_key="NOTIFY_TIME_DRIFT",
        )

    assert result is False
    mock_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_set_setting_publishes_broadcast():
    from server.api.schemas import SettingUpdatePayload
    from server.api.settings import set_setting
    from server.domain.models import Setting

    settings.NOTIFY_ALERT_CPU = "true"
    admin = MagicMock(username="admin")
    request = MagicMock()
    request.client.host = "127.0.0.1"
    db = MagicMock()
    mock_setting = Setting(key="notify_alert_cpu", value="false")

    with patch("server.api.settings.SettingRepositoryImpl") as repo_cls, patch(
        "server.api.settings.AuditLogRepositoryImpl"
    ) as audit_cls, patch(
        "server.infrastructure.settings_broadcast.publish_setting_change",
        new_callable=AsyncMock,
    ) as mock_publish:
        repo = repo_cls.return_value
        repo.set = AsyncMock(return_value=mock_setting)
        audit_cls.return_value.create = AsyncMock()

        await set_setting(
            "notify_alert_cpu",
            SettingUpdatePayload(value="false"),
            request,
            admin,
            db,
        )

    assert settings.NOTIFY_ALERT_CPU == "false"
    mock_publish.assert_awaited_once_with("notify_alert_cpu", "false")


@pytest.mark.parametrize(
    "attr",
    [
        "NOTIFY_ALERT_CPU",
        "NOTIFY_ALERT_MEM",
        "NOTIFY_ALERT_DISK",
        "NOTIFY_RECOVERY",
        "NOTIFY_TIME_DRIFT",
        "NOTIFY_SYSTEM_REDIS",
        "NOTIFY_SYSTEM_MYSQL",
        "NOTIFY_RESTART",
        "NOTIFY_SCRIPT_COMPLETE",
    ],
)
def test_all_notify_attrs_disable(attr):
    setattr(settings, attr, "false")
    assert notify_attr_enabled(attr) is False