8dca7eb46b
Watchdog-демон для непрерывного мониторинга SES. - Liveness-проверки API, лицензии, списка роботов - Диалоговые сценарии из YAML с агрегацией в 6 фиксированных Zabbix-метрик - Доставка: zabbix_sender (push) или UserParameter (poll) - Hot-reload сценариев без перезапуска
695 lines
24 KiB
Python
695 lines
24 KiB
Python
"""
|
||
Тесты для основных компонентов системы мониторинга SES.
|
||
|
||
Проверяют:
|
||
- Загрузку конфигурации из YAML
|
||
- Извлечение JSON-значений по пути
|
||
- Проверку условий ожидания (expect)
|
||
- Парсинг ответов SES API
|
||
- Выполнение шагов сценария (с мок-клиентом)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import tempfile
|
||
from pathlib import Path
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
import yaml
|
||
|
||
from ses_monitor.config import (
|
||
ScenarioConfig,
|
||
ScenarioStep,
|
||
load_scenarios,
|
||
)
|
||
from ses_monitor.utils import (
|
||
check_all_expectations,
|
||
check_expectation,
|
||
get_json_value,
|
||
setup_logging,
|
||
timer,
|
||
)
|
||
from ses_monitor.session_client import AskResponse, HostnameResponse, LicenseResponse
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: config.py
|
||
# =============================================================================
|
||
|
||
class TestConfigLoading:
|
||
"""Тесты загрузки конфигурации."""
|
||
|
||
def test_load_scenarios(self):
|
||
"""Проверяет загрузку YAML-сценариев из каталога."""
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
# Создаём тестовый сценарий
|
||
scenario_yaml = {
|
||
"name": "Тестовый сценарий",
|
||
"description": "Проверка загрузки",
|
||
"enabled": True,
|
||
"interval": 120,
|
||
"steps": [
|
||
{
|
||
"send": "привет",
|
||
"expect": [
|
||
{"path": "error", "equals": 0},
|
||
{"path": "answer[0].messages[0]", "contains": "при"},
|
||
],
|
||
}
|
||
],
|
||
}
|
||
with open(Path(tmpdir) / "test_scenario.yaml", "w") as f:
|
||
yaml.dump(scenario_yaml, f)
|
||
|
||
scenarios = load_scenarios(tmpdir)
|
||
assert len(scenarios) == 1
|
||
s = scenarios[0]
|
||
assert s.name == "Тестовый сценарий"
|
||
assert s.interval == 120
|
||
assert len(s.steps) == 1
|
||
assert s.steps[0].send == "привет"
|
||
assert len(s.steps[0].expect) == 2
|
||
|
||
def test_load_scenarios_disabled(self):
|
||
"""Проверяет что отключённые сценарии не загружаются."""
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
scenario_yaml = {
|
||
"name": "Отключённый",
|
||
"enabled": False,
|
||
"steps": [],
|
||
}
|
||
with open(Path(tmpdir) / "disabled.yaml", "w") as f:
|
||
yaml.dump(scenario_yaml, f)
|
||
|
||
scenarios = load_scenarios(tmpdir)
|
||
assert len(scenarios) == 0
|
||
|
||
def test_load_scenarios_empty_dir(self):
|
||
"""Проверяет загрузку из пустого каталога."""
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
scenarios = load_scenarios(tmpdir)
|
||
assert scenarios == []
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: utils.py — get_json_value
|
||
# =============================================================================
|
||
|
||
class TestGetJsonValue:
|
||
"""Тесты извлечения значений из JSON по пути."""
|
||
|
||
def test_simple_key(self):
|
||
data = {"error": 0, "message": "ok"}
|
||
assert get_json_value(data, "error") == 0
|
||
assert get_json_value(data, "message") == "ok"
|
||
|
||
def test_nested_key(self):
|
||
data = {"answer": [{"messages": ["здравствуйте"]}]}
|
||
assert get_json_value(data, "answer[0].messages[0]") == "здравствуйте"
|
||
|
||
def test_missing_key_returns_none(self):
|
||
data = {"error": 0}
|
||
assert get_json_value(data, "nonexistent") is None
|
||
assert get_json_value(data, "answer[0].text") is None
|
||
|
||
def test_array_index_out_of_bounds(self):
|
||
data = {"items": [1, 2]}
|
||
assert get_json_value(data, "items[5]") is None
|
||
|
||
def test_deeply_nested(self):
|
||
data = {
|
||
"request": {
|
||
"data": [
|
||
{
|
||
"reply": {
|
||
"confidence": 0.95,
|
||
}
|
||
}
|
||
]
|
||
}
|
||
}
|
||
assert get_json_value(data, "request.data[0].reply.confidence") == 0.95
|
||
|
||
def test_non_dict_data(self):
|
||
assert get_json_value([1, 2, 3], "items[0]") is None
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: utils.py — check_expectation
|
||
# =============================================================================
|
||
|
||
class TestCheckExpectation:
|
||
"""Тесты проверки условий ожидания."""
|
||
|
||
def test_equals_int(self):
|
||
ok, msg = check_expectation({"error": 0}, "error", "equals", 0)
|
||
assert ok
|
||
assert msg == ""
|
||
|
||
def test_equals_fail(self):
|
||
ok, msg = check_expectation({"error": 1}, "error", "equals", 0)
|
||
assert not ok
|
||
assert "не равно" in msg
|
||
|
||
def test_contains_str(self):
|
||
data = {"answer": [{"messages": ["введите фамилию"]}]}
|
||
ok, msg = check_expectation(
|
||
data, "answer[0].messages[0]", "contains", "фамилию"
|
||
)
|
||
assert ok
|
||
|
||
def test_contains_case_insensitive(self):
|
||
data = {"answer": [{"messages": ["Введите ФАМИЛИЮ"]}]}
|
||
ok, msg = check_expectation(
|
||
data, "answer[0].messages[0]", "contains", "фамилию"
|
||
)
|
||
assert ok
|
||
|
||
def test_not_contains(self):
|
||
data = {"answer": [{"messages": ["всё хорошо"]}]}
|
||
ok, msg = check_expectation(
|
||
data, "answer[0].messages[0]", "not_contains", "ошибка"
|
||
)
|
||
assert ok
|
||
|
||
def test_not_contains_fail(self):
|
||
data = {"answer": [{"messages": ["произошла ошибка"]}]}
|
||
ok, msg = check_expectation(
|
||
data, "answer[0].messages[0]", "not_contains", "ошибка"
|
||
)
|
||
assert not ok
|
||
|
||
def test_not_empty(self):
|
||
ok, msg = check_expectation(
|
||
{"text": "привет"}, "text", "not_empty", True
|
||
)
|
||
assert ok
|
||
|
||
def test_not_empty_fail(self):
|
||
ok, msg = check_expectation({}, "text", "not_empty", True)
|
||
assert not ok
|
||
|
||
def test_not_empty_empty_string(self):
|
||
ok, msg = check_expectation({"text": ""}, "text", "not_empty", True)
|
||
assert not ok
|
||
|
||
def test_regex_match(self):
|
||
data = {"phone": "+7-913-123-45-67"}
|
||
ok, msg = check_expectation(data, "phone", "regex", r"\+7[\d\-]+")
|
||
assert ok
|
||
|
||
def test_regex_no_match(self):
|
||
data = {"phone": "abc"}
|
||
ok, msg = check_expectation(data, "phone", "regex", r"^\+\d+$")
|
||
assert not ok
|
||
|
||
def test_exists(self):
|
||
ok, msg = check_expectation({"name": "test"}, "name", "exists")
|
||
assert ok
|
||
|
||
def test_exists_fail(self):
|
||
ok, msg = check_expectation({}, "name", "exists")
|
||
assert not ok
|
||
|
||
def test_missing_path(self):
|
||
ok, msg = check_expectation({}, "nonexistent[0].field", "equals", 0)
|
||
assert not ok
|
||
assert "не найдено" in msg or "не найд" in msg
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: utils.py — check_all_expectations
|
||
# =============================================================================
|
||
|
||
class TestCheckAllExpectations:
|
||
"""Тесты групповой проверки ожиданий."""
|
||
|
||
def test_all_pass(self):
|
||
data = {"error": 0, "answer": [{"messages": ["здравствуйте"]}]}
|
||
expectations = [
|
||
{"path": "error", "equals": 0},
|
||
{"path": "answer[0].messages[0]", "contains": "здрав"},
|
||
]
|
||
passed, errors = check_all_expectations(data, expectations)
|
||
assert passed
|
||
assert errors == []
|
||
|
||
def test_one_fails(self):
|
||
data = {"error": 1}
|
||
expectations = [
|
||
{"path": "error", "equals": 0},
|
||
{"path": "answer", "not_empty": True},
|
||
]
|
||
passed, errors = check_all_expectations(data, expectations)
|
||
assert not passed
|
||
assert len(errors) >= 1 # Первое или оба могут упасть
|
||
|
||
def test_no_expectations(self):
|
||
passed, errors = check_all_expectations({}, [])
|
||
assert passed
|
||
assert errors == []
|
||
|
||
def test_missing_path_in_expectation(self):
|
||
passed, errors = check_all_expectations({}, [{"contains": "test"}])
|
||
assert not passed
|
||
assert any("path" in e for e in errors)
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: utils.py — timer
|
||
# =============================================================================
|
||
|
||
class TestTimer:
|
||
"""Тесты контекстного менеджера timer."""
|
||
|
||
def test_measures_time(self):
|
||
import time
|
||
|
||
with timer() as t:
|
||
time.sleep(0.05)
|
||
|
||
assert "start" in t
|
||
assert "elapsed" in t
|
||
assert t["elapsed"] >= 0.04 # Допускаем небольшую погрешность
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: session_client.py — парсинг ответов
|
||
# =============================================================================
|
||
|
||
class TestAskResponse:
|
||
"""Тесты парсинга ответа POST /ses/ask."""
|
||
|
||
def test_from_valid_json(self):
|
||
data = {
|
||
"error": 0,
|
||
"question": "привет",
|
||
"answer": [{"messages": ["здравствуйте"], "voice": "Коля"}],
|
||
"session": "monitoring_test_123",
|
||
"uuid": "abc-def-123",
|
||
}
|
||
resp = AskResponse.from_json(data)
|
||
assert resp.error == 0
|
||
assert resp.question == "привет"
|
||
assert len(resp.answer) == 1
|
||
assert resp.session == "monitoring_test_123"
|
||
assert resp.uuid == "abc-def-123"
|
||
assert resp.raw == data
|
||
|
||
def test_from_error_json(self):
|
||
data = {"error": 1, "message": "robot not found"}
|
||
resp = AskResponse.from_json(data)
|
||
assert resp.error == 1
|
||
assert resp.question == ""
|
||
assert resp.answer == []
|
||
|
||
|
||
class TestHostnameResponse:
|
||
"""Тесты парсинга ответа GET /ses/hostname."""
|
||
|
||
def test_from_valid_json(self):
|
||
data = {
|
||
"error": 0,
|
||
"message": "Success",
|
||
"hostname": "ses-host.example.com",
|
||
"version": "5.046 release",
|
||
}
|
||
resp = HostnameResponse.from_json(data)
|
||
assert resp.error == 0
|
||
assert resp.hostname == "ses-host.example.com"
|
||
assert "5.046" in resp.version
|
||
|
||
|
||
class TestLicenseResponse:
|
||
"""Тесты парсинга ответа GET /license/check."""
|
||
|
||
def test_from_valid_json(self):
|
||
data = {
|
||
"error": 0,
|
||
"uuid": "398e8f18a83a6535d2bade81e9ee52a3",
|
||
"remaining_licenses": "infinity",
|
||
"threshold": 0,
|
||
}
|
||
resp = LicenseResponse.from_json(data)
|
||
assert resp.error == 0
|
||
assert resp.remaining_licenses == "infinity"
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: scenario_runner.py — с мок-клиентом
|
||
# =============================================================================
|
||
|
||
class TestScenarioRunner:
|
||
"""Тесты выполнения сценариев с мок-клиентом."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_simple_scenario(self):
|
||
"""Проверяет выполнение простого сценария с успешным ответом."""
|
||
from ses_monitor.config import SESConfig
|
||
from ses_monitor.scenario_runner import ScenarioRunner
|
||
|
||
scenario = ScenarioConfig(
|
||
name="test",
|
||
interval=10,
|
||
steps=[
|
||
ScenarioStep(
|
||
send="привет",
|
||
expect=[{"path": "error", "equals": 0}],
|
||
),
|
||
],
|
||
)
|
||
|
||
runner = ScenarioRunner(SESConfig(), [scenario])
|
||
|
||
# Мок HTTP-клиента
|
||
mock_client = AsyncMock()
|
||
mock_client.ask.return_value = AskResponse.from_json({
|
||
"error": 0,
|
||
"question": "привет",
|
||
"answer": [{"messages": ["здравствуйте"], "voice": "Коля"}],
|
||
"session": "monitoring_test_1",
|
||
"uuid": "test-uuid",
|
||
})
|
||
|
||
results = await runner.run_all(mock_client)
|
||
assert len(results) == 1
|
||
assert results[0].passed
|
||
assert len(results[0].step_results) == 1
|
||
assert results[0].step_results[0].passed
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_scenario_api_error(self):
|
||
"""Проверяет что сценарий фейлится при error != 0."""
|
||
from ses_monitor.config import SESConfig
|
||
from ses_monitor.scenario_runner import ScenarioRunner
|
||
|
||
scenario = ScenarioConfig(
|
||
name="test",
|
||
interval=10,
|
||
steps=[
|
||
ScenarioStep(
|
||
send="привет",
|
||
expect=[{"path": "error", "equals": 0}],
|
||
),
|
||
],
|
||
)
|
||
|
||
runner = ScenarioRunner(SESConfig(), [scenario])
|
||
|
||
mock_client = AsyncMock()
|
||
mock_client.ask.return_value = AskResponse.from_json({
|
||
"error": 1,
|
||
"question": "привет",
|
||
"answer": [],
|
||
"session": "",
|
||
"uuid": "",
|
||
})
|
||
|
||
results = await runner.run_all(mock_client)
|
||
assert len(results) == 1
|
||
assert not results[0].passed
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_scenario_expectation_fail(self):
|
||
"""Проверяет что сценарий фейлится при несовпадении ожиданий."""
|
||
from ses_monitor.config import SESConfig
|
||
from ses_monitor.scenario_runner import ScenarioRunner
|
||
|
||
scenario = ScenarioConfig(
|
||
name="test",
|
||
interval=10,
|
||
steps=[
|
||
ScenarioStep(
|
||
send="привет",
|
||
expect=[{"path": "answer[0].messages[0]", "contains": "ОЖИДАЕМЫЙ_ТЕКСТ"}],
|
||
),
|
||
],
|
||
)
|
||
|
||
runner = ScenarioRunner(SESConfig(), [scenario])
|
||
|
||
mock_client = AsyncMock()
|
||
mock_client.ask.return_value = AskResponse.from_json({
|
||
"error": 0,
|
||
"question": "привет",
|
||
"answer": [{"messages": ["другой текст"], "voice": "Коля"}],
|
||
"session": "test_1",
|
||
"uuid": "test-uuid",
|
||
})
|
||
|
||
results = await runner.run_all(mock_client)
|
||
assert len(results) == 1
|
||
assert not results[0].passed
|
||
|
||
|
||
# =============================================================================
|
||
# Тесты: aggregate_scenario_results
|
||
# =============================================================================
|
||
|
||
class TestAggregateScenarioResults:
|
||
"""Тесты агрегации результатов сценариев."""
|
||
|
||
def _make_passed(self, name: str = "test", time_ms: float = 100.0):
|
||
"""Создаёт успешный результат."""
|
||
from ses_monitor.scenario_runner import ScenarioResult
|
||
return ScenarioResult(
|
||
scenario_name=name,
|
||
passed=True,
|
||
total_time_ms=time_ms,
|
||
)
|
||
|
||
def _make_failed(self, name: str = "test", time_ms: float = 100.0):
|
||
"""Создаёт упавший результат."""
|
||
from ses_monitor.scenario_runner import ScenarioResult
|
||
return ScenarioResult(
|
||
scenario_name=name,
|
||
passed=False,
|
||
total_time_ms=time_ms,
|
||
error_message="тестовая ошибка",
|
||
)
|
||
|
||
def test_empty_list_returns_zeroes(self):
|
||
"""Пустой список — ok=1, всё нули."""
|
||
from ses_monitor.scenario_runner import aggregate_scenario_results
|
||
agg = aggregate_scenario_results([])
|
||
assert agg["ok"] == 1
|
||
assert agg["total"] == 0
|
||
assert agg["failed"] == 0
|
||
assert agg["avg_time"] == 0
|
||
assert agg["max_time"] == 0
|
||
assert agg["error"] == ""
|
||
|
||
def test_all_pass(self):
|
||
"""Все сценарии прошли — ok=1, failed=0, error пуст."""
|
||
from ses_monitor.scenario_runner import aggregate_scenario_results
|
||
results = [
|
||
self._make_passed("a", 100.0),
|
||
self._make_passed("b", 200.0),
|
||
self._make_passed("c", 300.0),
|
||
]
|
||
agg = aggregate_scenario_results(results)
|
||
assert agg["ok"] == 1
|
||
assert agg["total"] == 3
|
||
assert agg["failed"] == 0
|
||
assert agg["error"] == ""
|
||
assert agg["avg_time"] == 200 # (100+200+300)/3
|
||
assert agg["max_time"] == 300
|
||
|
||
def test_some_fail(self):
|
||
"""Часть сценариев упала — ok=0, failed=N, error содержит счётчик."""
|
||
from ses_monitor.scenario_runner import aggregate_scenario_results
|
||
results = [
|
||
self._make_passed("a", 100.0),
|
||
self._make_failed("b", 200.0),
|
||
self._make_failed("c", 300.0),
|
||
]
|
||
agg = aggregate_scenario_results(results)
|
||
assert agg["ok"] == 0
|
||
assert agg["total"] == 3
|
||
assert agg["failed"] == 2
|
||
assert agg["error"] == "2 сценариев упали"
|
||
assert agg["avg_time"] == 200
|
||
assert agg["max_time"] == 300
|
||
|
||
def test_all_fail(self):
|
||
"""Все сценарии упали — ok=0, failed=total."""
|
||
from ses_monitor.scenario_runner import aggregate_scenario_results
|
||
results = [
|
||
self._make_failed("a", 50.0),
|
||
self._make_failed("b", 150.0),
|
||
]
|
||
agg = aggregate_scenario_results(results)
|
||
assert agg["ok"] == 0
|
||
assert agg["total"] == 2
|
||
assert agg["failed"] == 2
|
||
assert agg["error"] == "2 сценариев упали"
|
||
|
||
def test_avg_time_calculation(self):
|
||
"""Среднее время считается корректно (включая упавшие)."""
|
||
from ses_monitor.scenario_runner import aggregate_scenario_results
|
||
results = [
|
||
self._make_passed("fast", 50.0),
|
||
self._make_failed("slow", 450.0),
|
||
]
|
||
agg = aggregate_scenario_results(results)
|
||
assert agg["avg_time"] == 250 # (50+450)/2
|
||
assert agg["max_time"] == 450
|
||
|
||
def test_max_time_with_single_result(self):
|
||
"""Один сценарий — avg_time == max_time."""
|
||
from ses_monitor.scenario_runner import aggregate_scenario_results
|
||
results = [self._make_passed("x", 123.0)]
|
||
agg = aggregate_scenario_results(results)
|
||
assert agg["ok"] == 1
|
||
assert agg["total"] == 1
|
||
assert agg["failed"] == 0
|
||
assert agg["avg_time"] == 123
|
||
assert agg["max_time"] == 123
|
||
|
||
def test_rounding_behaviour(self):
|
||
"""Время округляется до целых миллисекунд."""
|
||
from ses_monitor.scenario_runner import aggregate_scenario_results
|
||
results = [
|
||
self._make_passed("a", 10.6),
|
||
self._make_passed("b", 10.3),
|
||
]
|
||
agg = aggregate_scenario_results(results)
|
||
assert agg["avg_time"] == 10 # round(20.9/2) = round(10.45) = 10
|
||
assert agg["max_time"] == 11 # round(10.6) = 11
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
class TestUserParameterWriter:
|
||
"""Тесты для UserParameterWriter."""
|
||
|
||
async def test_write_json_file(self, tmp_path):
|
||
"""Проверяет что JSON-файл пишется атомарно и читается."""
|
||
from ses_monitor.zabbix import UserParameterWriter
|
||
from ses_monitor.config import ZabbixConfig, ZabbixMode
|
||
|
||
metrics_file = str(tmp_path / "metrics.json")
|
||
config = ZabbixConfig(
|
||
enabled=True,
|
||
mode=ZabbixMode.USERPARAMETER,
|
||
metrics_file=metrics_file,
|
||
)
|
||
writer = UserParameterWriter(config)
|
||
|
||
writer.add("ses.api.liveness", 1)
|
||
writer.add("ses.api.response_time", 42)
|
||
writer.add("ses.license.ok", True)
|
||
|
||
ok = await writer.flush()
|
||
assert ok, "flush должен вернуть True"
|
||
|
||
import json
|
||
with open(metrics_file, "r") as f:
|
||
data = json.load(f)
|
||
|
||
assert data["ses.api.liveness"] == "1"
|
||
assert data["ses.api.response_time"] == "42"
|
||
assert data["ses.license.ok"] == "1"
|
||
|
||
async def test_disabled_no_write(self, tmp_path):
|
||
"""Проверяет что при отключённом Zabbix файл не пишется."""
|
||
from ses_monitor.zabbix import UserParameterWriter
|
||
from ses_monitor.config import ZabbixConfig, ZabbixMode
|
||
|
||
metrics_file = str(tmp_path / "metrics.json")
|
||
config = ZabbixConfig(
|
||
enabled=False,
|
||
mode=ZabbixMode.USERPARAMETER,
|
||
metrics_file=metrics_file,
|
||
)
|
||
writer = UserParameterWriter(config)
|
||
|
||
writer.add("ses.test", 1)
|
||
ok = await writer.flush()
|
||
assert ok
|
||
|
||
import os
|
||
assert not os.path.exists(metrics_file), "файл не должен создаваться"
|
||
|
||
async def test_empty_no_write(self, tmp_path):
|
||
"""Проверяет что при пустом буфере файл не пишется."""
|
||
from ses_monitor.zabbix import UserParameterWriter
|
||
from ses_monitor.config import ZabbixConfig, ZabbixMode
|
||
|
||
metrics_file = str(tmp_path / "metrics.json")
|
||
config = ZabbixConfig(
|
||
enabled=True,
|
||
mode=ZabbixMode.USERPARAMETER,
|
||
metrics_file=metrics_file,
|
||
)
|
||
writer = UserParameterWriter(config)
|
||
|
||
ok = await writer.flush()
|
||
assert ok
|
||
|
||
import os
|
||
assert not os.path.exists(metrics_file), "файл не должен создаваться"
|
||
|
||
async def test_accumulates_in_single_cycle(self, tmp_path):
|
||
"""Проверяет что все метрики одного цикла попадают в файл."""
|
||
from ses_monitor.zabbix import UserParameterWriter
|
||
from ses_monitor.config import ZabbixConfig, ZabbixMode
|
||
|
||
metrics_file = str(tmp_path / "metrics.json")
|
||
config = ZabbixConfig(
|
||
enabled=True,
|
||
mode=ZabbixMode.USERPARAMETER,
|
||
metrics_file=metrics_file,
|
||
)
|
||
writer = UserParameterWriter(config)
|
||
|
||
writer.add("ses.metric.one", 1)
|
||
writer.add("ses.metric.two", 2)
|
||
await writer.flush()
|
||
|
||
import json
|
||
with open(metrics_file, "r") as f:
|
||
data = json.load(f)
|
||
|
||
assert data.get("ses.metric.one") == "1"
|
||
assert data.get("ses.metric.two") == "2"
|
||
|
||
async def test_generate_agentd_config(self):
|
||
"""Проверяет генерацию конфига Zabbix агента."""
|
||
from ses_monitor.zabbix import UserParameterWriter
|
||
from ses_monitor.config import ZabbixConfig, ZabbixMode
|
||
|
||
config = ZabbixConfig(mode=ZabbixMode.USERPARAMETER)
|
||
writer = UserParameterWriter(config)
|
||
cfg = writer.generate_agentd_config()
|
||
|
||
assert "UserParameter=ses.metrics," in cfg
|
||
assert "zbx_get.py" in cfg
|
||
|
||
async def test_atomic_write(self, tmp_path):
|
||
"""Проверяет что запись атомарна (файл не остаётся в повреждённом состоянии)."""
|
||
from ses_monitor.zabbix import UserParameterWriter
|
||
from ses_monitor.config import ZabbixConfig, ZabbixMode
|
||
|
||
metrics_file = str(tmp_path / "metrics.json")
|
||
config = ZabbixConfig(
|
||
enabled=True,
|
||
mode=ZabbixMode.USERPARAMETER,
|
||
metrics_file=metrics_file,
|
||
)
|
||
writer = UserParameterWriter(config)
|
||
|
||
for i in range(100):
|
||
writer.add(f"ses.test.{i}", i)
|
||
|
||
ok = await writer.flush()
|
||
assert ok
|
||
|
||
import json
|
||
with open(metrics_file, "r") as f:
|
||
data = json.load(f)
|
||
|
||
assert len(data) == 100
|
||
assert data["ses.test.99"] == "99"
|