"""Telegram HTML templates (notif-30)."""
from datetime import datetime, timezone
from app.models import Event, Host
from app.services.telegram_templates import (
format_event_mobile_push,
format_event_telegram_html,
format_rdp_login_html,
html_escape,
telegram_html_to_plaintext,
)
def test_html_escape_special_chars():
assert html_escape("a&") == "a<b>&"
def test_sanitize_telegram_html_keeps_bold():
from app.services.telegram_templates import sanitize_telegram_html
assert sanitize_telegram_html("title
line") == "title\nline"
def test_rdp_failed_template_includes_user_ip_logon():
host = Host(hostname="K6A-DC3", display_name="UNMS Kalina", os_family="windows")
event = Event(
event_id="00000000-0000-4000-8000-000000000501",
host_id=1,
host=host,
occurred_at=datetime(2026, 5, 29, 10, 30, 0, tzinfo=timezone.utc),
category="auth",
type="rdp.login.failed",
severity="warning",
title="RDP login failed",
summary="4625 user from 10.0.0.5",
details={
"user": "DOMAIN\\admin",
"ip_address": "10.0.0.5",
"logon_type": 10,
"event_id_windows": 4625,
"workstation_name": "CLIENT01",
},
payload={},
)
text = format_rdp_login_html(event)
assert "НЕУДАЧНАЯ" in text
assert "DOMAIN\\admin" in text or "DOMAIN" in text
assert "10.0.0.5" in text
assert "Удаленный интерактивный" in text
assert "(10)" in text
assert "UNMS Kalina" in text
assert "CLIENT01" in text
assert "4625" in text
assert "" in text
def test_rdp_login_hides_redundant_workstation_when_matches_hostname():
host = Host(
hostname="ITIS198",
display_name="ITIS (192.168.160.198)",
os_family="windows",
)
event = Event(
event_id="00000000-0000-4000-8000-000000000508",
host_id=1,
host=host,
occurred_at=datetime(2026, 5, 31, 18, 48, 52, tzinfo=timezone.utc),
category="auth",
type="rdp.login.success",
severity="info",
title="RDP login",
summary="papatramp",
details={
"user": "papatramp",
"ip_address": "192.168.160.3",
"logon_type": 10,
"workstation_name": "ITIS198",
"event_id_windows": 4624,
},
payload={},
)
text = format_rdp_login_html(event)
assert "ITIS" in text
assert "192.168.160.3" in text
assert "Рабочая станция" not in text
assert "ITIS198" not in text
def test_rdp_login_shows_workstation_when_client_differs():
host = Host(hostname="ITIS198", display_name="ITIS", os_family="windows")
event = Event(
event_id="00000000-0000-4000-8000-000000000509",
host_id=1,
host=host,
occurred_at=datetime(2026, 5, 31, 18, 48, 52, tzinfo=timezone.utc),
category="auth",
type="rdp.login.success",
severity="info",
title="RDP login",
summary="papatramp",
details={
"user": "papatramp",
"ip_address": "192.168.160.3",
"logon_type": 10,
"workstation_name": "NEW-ADMIN-PC",
"event_id_windows": 4624,
},
payload={},
)
text = format_rdp_login_html(event)
assert "Рабочая станция" in text
assert "NEW-ADMIN-PC" in text
def test_rdp_success_template():
event = Event(
event_id="00000000-0000-4000-8000-000000000502",
host_id=1,
occurred_at=datetime(2026, 5, 29, 11, 0, tzinfo=timezone.utc),
category="auth",
type="rdp.login.success",
severity="info",
title="ok",
summary="ok",
details={"user": "u1", "ip_address": "1.2.3.4", "logon_type": 3},
payload={},
)
text = format_event_telegram_html(event)
assert "УСПЕШНЫЙ" in text
assert "Сеть/RDP" in text
assert "📡 Оповещение: SAC (Security Alert Center)" in text
def test_event_telegram_includes_sac_source_for_sac_daily_report():
host = Host(hostname="h1", display_name="H1", os_family="linux", product="ssh-monitor")
event = Event(
event_id="00000000-0000-4000-8000-000000000504",
host_id=1,
host=host,
occurred_at=datetime(2026, 5, 29, 9, 0, tzinfo=timezone.utc),
category="report",
type="report.daily.ssh",
severity="info",
title="Отчёт",
summary="short",
details={"generated_by": "sac", "report_body": "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА\nline"},
payload={},
)
text = format_event_telegram_html(event)
assert "📡 Оповещение: SAC (Security Alert Center)" in text
def test_telegram_html_to_plaintext_strips_tags():
assert telegram_html_to_plaintext("✅ Агент запущен\n🏢 Хост: WIN01") == (
"✅ Агент запущен\n🏢 Хост: WIN01"
)
def test_format_event_mobile_push_lifecycle_includes_trigger():
host = Host(hostname="WIN01", display_name="K6A-DC3", os_family="windows")
event = Event(
event_id="00000000-0000-4000-8000-000000000509",
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="Мониторинг запущен",
details={"lifecycle": "started", "trigger": "deploy_recycle", "telegram_via": "sac"},
payload={"source": {"product": "rdp-login-monitor", "product_version": "1.2.30-SAC"}},
)
title, body = format_event_mobile_push(event)
assert title == "✅ Агент запущен"
assert "K6A-DC3" in body
assert "1.2.30-SAC" in body
assert "обновление скрипта" in body
assert "📡 Оповещение: SAC" in body
def test_format_event_mobile_push_uses_notification_body():
host = Host(hostname="WIN01", display_name="K6A-DC3", os_family="windows")
body_text = "✅ Мониторинг логинов ЗАПУЩЕН\n🖥️ Сервер: K6A-DC3 (10.0.0.1)"
event = Event(
event_id="00000000-0000-4000-8000-000000000510",
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="Мониторинг запущен",
details={
"lifecycle": "started",
"trigger": "boot",
"notification_body": body_text,
"telegram_via": "sac",
},
payload={"source": {"product": "rdp-login-monitor", "product_version": "1.2.31-SAC"}},
)
title, body = format_event_mobile_push(event)
assert title == "✅ Мониторинг логинов ЗАПУЩЕН"
assert "K6A-DC3 (10.0.0.1)" in body
def test_lifecycle_started_template():
host = Host(hostname="WIN01", display_name="K6A-DC3", os_family="windows")
event = Event(
event_id="00000000-0000-4000-8000-000000000505",
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="Мониторинг запущен",
details={"lifecycle": "started", "trigger": "boot", "telegram_via": "sac"},
payload={"source": {"product": "rdp-login-monitor", "product_version": "1.2.30-SAC"}},
)
text = format_event_telegram_html(event)
assert "Агент запущен" in text
assert "K6A-DC3" in text
assert "1.2.30-SAC" in text
assert "загрузка ОС" in text
assert "📡 Оповещение: SAC (Security Alert Center)" in text
def test_lifecycle_notification_body_renders_full_text():
host = Host(hostname="WIN01", display_name="K6A-DC3", os_family="windows")
body = "✅ Мониторинг логинов ЗАПУЩЕН\n🖥️ Сервер: K6A-DC3 (10.0.0.1)\n🕐 Время запуска: 28.05.2026 12:00:00"
event = Event(
event_id="00000000-0000-4000-8000-000000000507",
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="Мониторинг запущен",
details={"lifecycle": "started", "trigger": "boot", "notification_body": body, "telegram_via": "sac"},
payload={"source": {"product": "rdp-login-monitor", "product_version": "1.2.31-SAC"}},
)
text = format_event_telegram_html(event)
assert "✅ Мониторинг логинов ЗАПУЩЕН" in text
assert "K6A-DC3 (10.0.0.1)" in text
assert "Агент запущен" not in text
assert "📡 Оповещение: SAC (Security Alert Center)" in text
def test_lifecycle_footer_uses_telegram_via_agent_when_present():
host = Host(hostname="WIN01", display_name="H1", os_family="windows")
event = Event(
event_id="00000000-0000-4000-8000-000000000506",
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="x",
details={"lifecycle": "started", "telegram_via": "agent"},
payload={"source": {"product": "rdp-login-monitor", "product_version": "1.2.30-SAC"}},
)
text = format_event_telegram_html(event)
assert "📡 Оповещение: агент (rdp-login-monitor 1.2.30-SAC)" in text
def test_daily_report_summary_fallback_includes_host():
host = Host(hostname="srv1", display_name="K6A-DC5", ipv4="192.168.160.91", os_family="linux")
event = Event(
event_id="00000000-0000-4000-8000-000000000508",
host_id=1,
host=host,
occurred_at=datetime(2026, 6, 3, 9, 0, tzinfo=timezone.utc),
category="report",
type="report.daily.ssh",
severity="info",
title="Ежедневный отчёт SSH",
summary="SSH 24ч: успех 0, неудач 0, банов 0",
details={"generated_by": "sac"},
payload={},
)
text = format_event_telegram_html(event)
assert "K6A-DC5" in text
assert "SSH 24ч: успех 0" in text
assert "ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH" in text
def test_rdp_shadow_control_template():
host = Host(hostname="RDS01", display_name="RDS Farm", ipv4="10.0.0.5", os_family="windows")
event = Event(
event_id="00000000-0000-4000-8000-000000000701",
host_id=1,
host=host,
occurred_at=datetime(2026, 5, 30, 10, 0, 0, tzinfo=timezone.utc),
category="session",
type="rdp.shadow.control.started",
severity="warning",
title="RDS Shadow Control 20506",
summary="Shadow admin -> user",
details={
"event_id_windows": 20506,
"shadower_user": "DOMAIN\\admin",
"target_user": "DOMAIN\\user",
"session_id": "3",
"shadow_action": "control_started",
},
payload={},
)
text = format_event_telegram_html(event)
assert "SHADOW CONTROL" in text
assert "DOMAIN" in text
assert "20506" in text
assert "RDS Farm" in text
def test_winrm_session_template():
host = Host(hostname="SRV01", display_name="App Server", os_family="windows")
event = Event(
event_id="00000000-0000-4000-8000-000000000702",
host_id=1,
host=host,
occurred_at=datetime(2026, 5, 30, 11, 0, 0, tzinfo=timezone.utc),
category="session",
type="winrm.session.started",
severity="warning",
title="WinRM shell",
summary="WinRM 91",
details={
"event_id_windows": 91,
"user": "DOMAIN\\admin",
"source_ip": "192.168.160.3",
"resource_uri": "http://schemas.microsoft.com/powershell/Microsoft.PowerShell",
},
payload={},
)
text = format_event_telegram_html(event)
assert "Enter-PSSession" in text or "WinRM" in text
assert "192.168.160.3" in text
assert "91" in text
def test_ssh_failed_template():
event = Event(
event_id="00000000-0000-4000-8000-000000000503",
host_id=1,
occurred_at=datetime(2026, 5, 29, 12, 0, tzinfo=timezone.utc),
category="auth",
type="ssh.login.failed",
severity="warning",
title="ssh fail",
summary="fail",
details={"user": "root", "source_ip": "10.10.36.9", "port": 22, "attempt_number": 3, "max_attempts": 5},
payload={},
)
text = format_event_telegram_html(event)
assert "SSH" in text
assert "root" in text
assert "10.10.36.9" in text
assert "3 / 5" in text
def test_agent_inventory_hardware_change_template():
event = Event(
event_id="00000000-0000-4000-8000-000000000504",
host_id=1,
occurred_at=datetime(2026, 6, 4, 12, 0, tzinfo=timezone.utc),
category="agent",
type="agent.inventory",
severity="warning",
title="Hardware changed",
summary="memory_gb changed",
details={
"hardware_changes": [
{"field": "memory_gb", "old": 16.0, "new": 32.0},
]
},
payload={},
)
text = format_event_telegram_html(event)
assert "железо" in text.lower() or "Изменилось" in text
assert "memory_gb" in text
assert "32" in text
def test_format_problem_host_silence_scan_without_event():
from app.models import Host, Problem
from app.services.problem_rules import RULE_HOST_SILENCE
from app.services.telegram_templates import format_problem_telegram_html
host = Host(
hostname="YURKOV-PC",
os_family="windows",
product="rdp-login-monitor",
last_seen_at=datetime(2026, 6, 11, 9, 0, tzinfo=timezone.utc),
)
problem = Problem(
host_id=1,
host=host,
title="Host silence: YURKOV-PC",
summary="Нет свежего agent.heartbeat более 300 мин",
severity="high",
status="open",
rule_id=RULE_HOST_SILENCE,
fingerprint="fp-silence",
event_count=0,
last_seen_at=datetime(2026, 6, 11, 9, 0, tzinfo=timezone.utc),
)
text = format_problem_telegram_html(problem, event=None)
assert "периодическая проверка SAC" in text
assert "host_silence scan" in text
assert "YURKOV-PC" in text or "UNMS" not in text # host label uses hostname