"""Unified daily report text format.""" from datetime import datetime, timezone from app.models import Host from app.services.daily_report_format import ( append_notification_source_plain, build_report_body, enrich_stats_for_storage, format_notification_source_plain, normalize_active_users_list, normalize_report_body, ) def test_build_report_body_ssh_matches_agent_layout(): host = Host( hostname="haproxy", display_name="HaProxy Kalina", ipv4="192.168.160.117", os_family="linux", product="ssh-monitor", ) when = datetime(2026, 5, 29, 9, 0, 5, tzinfo=timezone.utc) stats = enrich_stats_for_storage( "ssh", { "successful_ssh": 0, "failed_ssh": 0, "sudo_commands": 5, "active_bans": 0, "top_failed_ips": [], "active_users": [ "👤 papatramp | pts/0 | с 2026-05-27 12:28 | 🌐 192.168.160.3", ], }, ) body = build_report_body("ssh", host, stats, when, sac_generated=True) assert "ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА" in body assert "Agent version sac" in body assert "🖥️ Сервер: HaProxy Kalina (192.168.160.117)" in body assert " 📈 СТАТИСТИКА" in body assert "Команд через sudo: 5" in body assert " 👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (1)" in body assert "papatramp" in body assert "\n\n\n" not in body def test_build_report_body_windows_layout(): host = Host( hostname="WIN01", display_name="K6A-DC3", ipv4="10.0.0.10", os_family="windows", product="rdp-login-monitor", ) when = datetime(2026, 5, 29, 9, 0, 5, tzinfo=timezone.utc) stats = enrich_stats_for_storage( "windows", { "rdp_success": 1, "rdp_failed": 2, "active_bans": 0, "top_failed_ips": ["1.2.3.4 — 2"], "active_users": ["👤 DOMAIN\\user1", "👤 user2"], }, ) body = build_report_body("windows", host, stats, when, sac_generated=True) assert "ЕЖЕДНЕВНЫЙ ОТЧЕТ МОНИТОРИНГА WINDOWS" in body assert "Agent version sac" in body assert "K6A-DC3 (10.0.0.10)" in body assert "Успешных RDP подключений: 1" in body assert " 👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (2)" in body assert "user2" in body lines = [ln for ln in body.split("\n") if ln.strip().startswith("👤")] assert len(lines) == 2 def test_build_report_body_includes_notification_source(): host = Host(hostname="h1", display_name="H1", os_family="linux", product="ssh-monitor") when = datetime(2026, 5, 29, 9, 0, 5, tzinfo=timezone.utc) stats = enrich_stats_for_storage("ssh", {"successful_ssh": 0, "failed_ssh": 0, "sudo_commands": 0, "active_bans": 0}) sac_body = build_report_body("ssh", host, stats, when, sac_generated=True) assert "📡 Оповещение: SAC (Security Alert Center)" in sac_body agent_body = build_report_body( "ssh", host, {**stats, "agent_version": "1.2.11-SAC"}, when, sac_generated=False, ) assert "📡 Оповещение: агент (ssh-monitor 1.2.11-SAC)" in agent_body def test_append_notification_source_plain_dedupes(): line = format_notification_source_plain(generated_by="sac") body = append_notification_source_plain(f"title\n\n{line}", generated_by="sac") assert body.count("📡 Оповещение:") == 1 def test_normalize_active_users_list_splits_combined_line(): users = normalize_active_users_list(["👤 k.khodasevich 👤 papatramp"]) assert len(users) == 2 assert users[0].strip().startswith("👤 k.khodasevich") assert users[1].strip().startswith("👤 papatramp") def test_normalize_report_body_adds_server_and_collapses_blanks(): host = Host( hostname="srv", display_name="Unimus Kalina", ipv4="192.168.160.17", product_version="1.2.10-SAC", ) raw = "\n".join( [ "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА", "", "", "🕐 Время отчета: 30.05.2026 09:00:00", "", "", " 📈 СТАТИСТИКА ЗА ПОСЛЕДНИЕ 24 ЧАСА:", " ✅ Успешных SSH подключений: 0", "", " 👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (1):", " 👤 u1 👤 u2", ] ) body = normalize_report_body(raw, host, "ssh") assert "Agent version 1.2.10-SAC" in body assert "🖥️ Сервер: Unimus Kalina (192.168.160.17)" in body assert "\n\n\n" not in body user_lines = [ln for ln in body.split("\n") if ln.strip().startswith("👤")] assert len(user_lines) == 2 assert " 👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (2)" in body def test_normalize_ssh_active_users_mismatch_count_and_empty_body(): host = Host(hostname="srv", display_name="SSH", ipv4="10.0.0.1", product="ssh-monitor") raw = "\n".join( [ "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА", "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (2):", " (нет данных)", ] ) body = normalize_report_body(raw, host, "ssh") assert "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0)" in body assert "(нет данных)" in body assert not any(ln.strip().startswith("👤") for ln in body.split("\n")) def test_normalize_rdp_empty_active_users_placeholder(): host = Host(hostname="gw", display_name="K6A-DC3", ipv4="192.168.160.40", product="rdp-login-monitor") raw = "\n".join( [ "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ МОНИТОРИНГА WINDOWS", "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0):", " (нет активных пользователей / RDP-сессий)", ] ) body = normalize_report_body(raw, host, "windows") assert "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0)" in body assert "нет активных пользователей" in body assert "👤 (нет активных" not in body def test_normalize_daily_report_reconciles_stats_from_body(): from app.services.daily_report_format import normalize_daily_report_details host = Host(hostname="srv", display_name="SSH", ipv4="10.0.0.2", product="ssh-monitor") details = { "report_body": "\n".join( [ "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА", "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (9):", " (нет данных)", ] ), "stats": { "successful_ssh": 1, "active_sessions": 9, "active_users": [], "generated_by": "agent", }, } out = normalize_daily_report_details(details, host, "report.daily.ssh") assert out is not None assert out["stats"]["active_sessions"] == 0 assert out["stats"]["active_users"] == [] assert "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0)" in out["report_body"]