763a1a2a16
archive_auditor.py not imported or used by main.py since archive metrics were removed from the monitoring loop. Remove the dead module along with its 10 tests and dead imports (ModelsMonitoringConfig, AppConfig, ArchiveEntry). Tests: 59 -> 48 (all pass)
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"
|