100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
"""FCM push must not break event ingest."""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.models import Event, Host
|
|
from app.services.mobile_notify import MobileNotConfiguredError, _event_payload, _send_fcm_data
|
|
|
|
|
|
def test_event_payload_lifecycle_uses_telegram_wording():
|
|
host = Host(hostname="WIN01", display_name="K6A-DC3", os_family="windows")
|
|
event = Event(
|
|
event_id="00000000-0000-4000-8000-000000000511",
|
|
host_id=1,
|
|
host=host,
|
|
occurred_at=datetime(2026, 5, 29, 9, 0, tzinfo=timezone.utc),
|
|
category="agent",
|
|
type="agent.lifecycle",
|
|
severity="info",
|
|
title="started",
|
|
summary="short",
|
|
details={"lifecycle": "started", "trigger": "settings_reload", "telegram_via": "sac"},
|
|
payload={"source": {"product": "ssh-monitor", "product_version": "1.4.0-SAC"}},
|
|
)
|
|
title, body, data = _event_payload(event)
|
|
assert title == "✅ Агент запущен"
|
|
assert "graceful restart" in body
|
|
assert data["type"] == "agent.lifecycle"
|
|
|
|
|
|
def test_send_fcm_data_swallows_token_errors_by_default():
|
|
with patch(
|
|
"app.services.mobile_notify._fcm_access_token",
|
|
side_effect=MobileNotConfiguredError("FCM: install requests"),
|
|
):
|
|
with patch("app.services.mobile_notify.get_effective_fcm_config") as mock_cfg:
|
|
mock_cfg.return_value.enabled = True
|
|
mock_cfg.return_value.project_id = "proj"
|
|
mock_cfg.return_value.service_account_path = "/etc/fcm.json"
|
|
mock_cfg.return_value.configured = True
|
|
mock_cfg.return_value.active = True
|
|
_send_fcm_data(["token"], {"kind": "test"}, title="t", body="b")
|
|
|
|
|
|
def test_send_fcm_data_raises_token_errors_when_required():
|
|
with patch(
|
|
"app.services.mobile_notify._fcm_access_token",
|
|
side_effect=MobileNotConfiguredError("FCM: install requests"),
|
|
):
|
|
with patch("app.services.mobile_notify.get_effective_fcm_config") as mock_cfg:
|
|
mock_cfg.return_value.enabled = True
|
|
mock_cfg.return_value.project_id = "proj"
|
|
mock_cfg.return_value.service_account_path = "/etc/fcm.json"
|
|
mock_cfg.return_value.configured = True
|
|
mock_cfg.return_value.active = True
|
|
with pytest.raises(MobileNotConfiguredError):
|
|
_send_fcm_data(
|
|
["token"],
|
|
{"kind": "test"},
|
|
title="t",
|
|
body="b",
|
|
require_success=True,
|
|
)
|
|
|
|
|
|
def test_send_fcm_data_payload_is_data_only():
|
|
posted: list[dict] = []
|
|
|
|
def capture_post(_url, *, headers, json):
|
|
posted.append(json)
|
|
response = MagicMock()
|
|
response.raise_for_status.return_value = None
|
|
return response
|
|
|
|
with patch("app.services.mobile_notify._fcm_access_token", return_value="token"):
|
|
with patch("app.services.mobile_notify.get_effective_fcm_config") as mock_cfg:
|
|
mock_cfg.return_value.enabled = True
|
|
mock_cfg.return_value.project_id = "proj"
|
|
mock_cfg.return_value.service_account_path = "/etc/fcm.json"
|
|
mock_cfg.return_value.configured = True
|
|
mock_cfg.return_value.active = True
|
|
with patch("httpx.Client") as mock_client:
|
|
mock_client.return_value.__enter__.return_value.post = capture_post
|
|
_send_fcm_data(
|
|
["device-token"],
|
|
{"kind": "event", "id": "42", "severity": "high"},
|
|
title="Alert",
|
|
body="Summary text",
|
|
)
|
|
|
|
assert len(posted) == 1
|
|
message = posted[0]["message"]
|
|
assert "notification" not in message
|
|
assert message["data"]["title"] == "Alert"
|
|
assert message["data"]["body"] == "Summary text"
|
|
assert message["data"]["kind"] == "event"
|
|
assert message["data"]["id"] == "42"
|