PapaTramp
64b942447b
Normalize empty-session placeholders; sync stats with report body after ingest. Co-authored-by: Cursor <cursoragent@cursor.com>
193 lines
7.2 KiB
Python
193 lines
7.2 KiB
Python
"""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"]
|