https://pastein.ru/t/Sf0
Загрузка данных
сцен
"""
Сценарии тестов по режимам МТ - функции-обёртки без pytest маркеров.
Каждая функция содержит логику одного теста.
Pytest маркеры и allure декораторы применяются в тестовых файлах.
"""
from datetime import datetime
from typing import Any
import allure
import pytest
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.enums import Direction, MessageType, StationaryStatus
from constants.test_constants import BaseTN3Constants as TestConst
from models.get_messages_model import Filtering, FilteringObjects, Pagination
from test_config.models_for_tests import BaseSuiteConfig, CaseData
from utils.helpers import ws_test_utils as t_utils
from utils.helpers.asserts import SoftAssertions, StepCheck
from utils.helpers.ws_message_parser import ws_message_parser as parser
def _unpack_stationary_status_test_data(
test_data: CaseData, require_control_points: bool = False
) -> tuple[Any, Any, list[str]]:
"""
Проверяет и распаковывает конфигурацию теста режима МТ.
"""
if not test_data or not test_data.expected_result:
pytest.fail("Не заполнены данные для теста режима МТ")
try:
expected_stationary_status, expected_stationary_status_reasons = test_data.expected_result
except (TypeError, ValueError) as error:
pytest.fail(
"Данные режима МТ должны содержать пару: "
f"(ожидаемый режим, ожидаемая причина). Получено: {test_data.expected_result}. Ошибка: {error}"
)
control_points = (test_data.params or {}).get(TestConst.CONTROL_POINTS_KEY, [])
if require_control_points and not control_points:
pytest.fail(f"Не заполнены обязательные параметры для теста: {TestConst.CONTROL_POINTS_KEY}")
return expected_stationary_status, expected_stationary_status_reasons, control_points
async def stationary_status_common_scheme(ws_client, cfg: BaseSuiteConfig, test_data: CaseData):
"""
Проверка режима работы и причины режима МТ в сообщении CommonSchemeContent
"""
# Распаковка данных для теста
expected_stationary_status, expected_stationary_status_reasons, _ = _unpack_stationary_status_test_data(test_data)
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_poll_subscribed_msg(
ws_client,
"CommonSchemeContent",
"SubscribeCommonSchemeRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_payload = parser.parse_common_scheme_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
flow_areas = t_utils.filter_flow_areas_with_available_flow(
getattr(parsed_payload.replyContent, 'flowAreas', [])
)
StepCheck(
"Проверка наличия участков карты течения с доступным течением CommonSchemeContent (ЭФ Схема)", "flowAreas"
).actual(flow_areas).is_not_empty()
longest_flow_area = t_utils.get_longest_flow_area_by_pipes(flow_areas)
StepCheck(
"Проверка наличия самого длинного пути течения CommonSchemeContent (ЭФ Схема)", "longest_flow_area"
).actual(longest_flow_area is not None).is_true_with_details(
expected_text="longest_flow_area определён",
actual_text=(
"longest_flow_area определён" if longest_flow_area is not None else "longest_flow_area не определён"
),
)
if longest_flow_area is not None:
allure.attach(
str(longest_flow_area),
name="CommonSchemeContent: самый длинный путь течения (ЭФ Схема)",
attachment_type=allure.attachment_type.TEXT,
)
diagnostic_areas = getattr(longest_flow_area, 'diagnosticAreas', [])
StepCheck(
"Проверка наличия данных диагностических участков CommonSchemeContent (ЭФ Схема)", "diagnosticAreas"
).actual(diagnostic_areas).is_not_empty()
common_scheme_stationary_statuses = []
for diagnostic_area in diagnostic_areas:
stationary_status_int = getattr(diagnostic_area, 'stationaryStatus', None)
if stationary_status_int is None:
continue
stationary_status = StationaryStatus(stationary_status_int)
common_scheme_stationary_statuses.append(stationary_status)
StepCheck(
"Проверка наличия режимов МТ на ДУ в CommonSchemeContent (ЭФ Схема)",
"stationaryStatus",
).actual(common_scheme_stationary_statuses).is_not_empty()
common_scheme_majority = t_utils.determine_stationary_status_by_majority(common_scheme_stationary_statuses)
StepCheck(
"CommonSchemeContent: общий режим МТ по большинству ДУ (ЭФ Схема)",
"stationaryStatus",
).actual(
common_scheme_majority
).expected(expected_stationary_status).equal_to()
majority_status_areas = []
# Получает все ДУ с ожидаемым режимом МТ
for diagnostic_area in diagnostic_areas:
if diagnostic_area.stationaryStatus == expected_stationary_status.value:
majority_status_areas.append(diagnostic_area)
with SoftAssertions() as soft_failures:
# Проверяет причину режима МТ у всех ДУ с ожидаемым режимом МТ
for diagnostic_area in majority_status_areas:
stationary_status_reasons = t_utils.parse_stationary_status_reasons(
diagnostic_area.stationaryStatus, diagnostic_area.stationaryStatusReasons, soft_failures
)
StepCheck(
f"Проверка причины режима работы МТ на ДУ с id:{diagnostic_area.id}",
"stationaryStatusReasons",
soft_failures,
).contains(stationary_status_reasons, expected_stationary_status_reasons)
async def stationary_status_main_page_info(ws_client, cfg: BaseSuiteConfig, test_data: CaseData):
"""
Проверка установки режима МТ в сообщении MainPageInfoContent
"""
expected_stationary_status, _, _ = _unpack_stationary_status_test_data(test_data)
with allure.step("Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent"):
payload = await t_utils.connect_and_poll_subscribed_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
tu_info = getattr(parsed_payload.replyContent, 'tuInfo', None)
StepCheck("Проверка наличия данных по ТУ", "tuInfo").actual(tu_info).is_not_none()
main_pipeline_stationary_status = (
StationaryStatus(tu_info.stationaryStatus) if tu_info.stationaryStatus else None
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck(
f"Проверка установки режима МТ для ТУ {cfg.tu_name}",
"stationaryStatus",
soft_failures,
).actual(
main_pipeline_stationary_status
).expected(expected_stationary_status).equal_to()
async def stationary_status_in_output_signals(ws_client, cfg: BaseSuiteConfig, test_data: CaseData):
"""
Проверка установки режима МТ в сообщении OutputSignalsInfo
"""
# Распаковка тестовых данных
expected_stationary_status, _, control_points = _unpack_stationary_status_test_data(
test_data, require_control_points=True
)
unknown_control_points = [name for name in control_points if name not in TestConst.CONTROLLED_SITE_SEGMENTS]
if unknown_control_points:
pytest.fail(
"Для контрольных участков не найдены controlledSiteId/segmentId "
f"в конфигурации стенда: {unknown_control_points}"
)
# получает часть словаря для всего списка controlled_sites_names
controlled_sites_dict = {
name: TestConst.CONTROLLED_SITE_SEGMENTS[name]
for name in control_points
if name in TestConst.CONTROLLED_SITE_SEGMENTS
}
controlled_sites = [
{'controlledSiteId': controlled_site_id, 'segmentId': segment_id}
for key, (controlled_site_id, segment_id) in controlled_sites_dict.items()
]
with allure.step("Подключение по ws, получение и обработка сообщения типа: OutputSignalsInfo"):
output_signals_payload = await t_utils.connect_and_poll_subscribed_msg(
ws_client,
"OutputSignalsInfo",
"SubscribeOutputSignalsRequest",
{
'objects': {
'linearParts': [],
'controlledSites': controlled_sites,
},
'signalTypes': 1023,
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_output_signals = parser.parse_output_signals_info_msg(output_signals_payload)
with allure.step("Извлечение и подготовка данных для проверки"):
controlled_site_signals = getattr(parsed_output_signals.replyContent, 'controlledSiteSignals', [])
StepCheck("Проверка наличия данных выходных сигналов", "controlledSiteSignals").actual(
controlled_site_signals
).is_not_empty()
output_signals_site_stationary_statuses, missed_controlled_points = (
t_utils.collect_stationary_statuses_from_output_signals(controlled_site_signals, controlled_sites_dict)
)
output_signals_details = "\n".join(
f"{controlled_point}: {stationary_status}"
for controlled_point, stationary_status in output_signals_site_stationary_statuses
)
allure.attach(
output_signals_details,
name="OutputSignalsInfo: режимы МТ по участкам КП-КП (ЭФ Выходные сигналы)",
attachment_type=allure.attachment_type.TEXT,
)
with SoftAssertions() as soft_failures:
for item in output_signals_site_stationary_statuses:
control_point, stationary_status = item
StepCheck(
f"Проверка режима МТ на контрольном участке: {control_point}", "stationaryStatus", soft_failures
).actual(stationary_status).expected(expected_stationary_status).equal_to()
missed_controlled_points_is_empty = not missed_controlled_points
StepCheck(
"При обработке Выходных Сигналов найдены записи режима МТ для всех выбранных контрольных участков.",
"участки КП-КП",
soft_failures,
).actual(missed_controlled_points_is_empty).is_true_with_details(
expected_text="Контрольные участки, для которых не найдены записи режима МТ missed_controlled_points = []",
actual_text="Контрольные участки, для которых не найдены записи режима МТ "
f"missed_controlled_points = {missed_controlled_points}",
)
def stationary_status_journal(http_client, cfg: BaseSuiteConfig, test_data: CaseData):
"""
Проверка наличия записи в журнале о режиме МТ.
"""
# Распаковка данных для теста
expected_stationary_status, expected_stationary_status_reasons, control_points = (
_unpack_stationary_status_test_data(test_data, require_control_points=True)
)
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=PUMPING_STATUS"):
end_time = datetime.now()
start_time = t_utils.datetime_minus_seconds(end_time, TestConst.JOURNAL_STATUS_TOTAL_WAIT)
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_STATUS_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(messageTypes=int(MessageType.PUMPING_STATUS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по времени и controlPoint"):
filter_start_msk = t_utils.localize_as_moscow(start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
stationary_msg_by_control_points = []
missed_control_points = []
time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
# Поиск нужных сообщений по КП
for control_point in control_points:
stationary_msg = next(
(msg for msg in time_filtered if msg.controlPoint == control_point),
None,
)
if stationary_msg:
stationary_msg_by_control_points.append(stationary_msg)
else:
missed_control_points.append(control_point)
StepCheck("Проверка наличия сообщений c controlPoint из списка в журнале", "messagesInfo").actual(
stationary_msg_by_control_points
).is_not_empty()
with SoftAssertions() as soft_failures:
for msg in stationary_msg_by_control_points:
msg_event = getattr(msg, 'event', None)
cp_stationary_status, cp_stationary_status_reasons = t_utils.parse_journal_event(msg_event)
StepCheck(f"Проверка режима работы МТ на КП:{msg.controlPoint}", "event", soft_failures).actual(
cp_stationary_status
).expected(expected_stationary_status).equal_to()
StepCheck(f"Проверка причины режима работы МТ на КП:{msg.controlPoint}", "event", soft_failures).contains(
cp_stationary_status_reasons, expected_stationary_status_reasons
)
missed_control_points_is_empty = not missed_control_points
StepCheck(
"При обработке сообщений журнала найдены сообщения для всех КП.",
"controlPoint",
soft_failures,
).actual(missed_control_points_is_empty).is_true_with_details(
expected_text="КП, для которых не найдены сообщения missed_control_points = []",
actual_text=f"КП, для которых не найдены сообщения missed_control_points = {missed_control_points}",
)
тест смоук
"""
Единый тестовый модуль для всех наборов данных (single-leak и multi-leak).
Архитектура параметризации:
- SUITE_PARAMS: тесты уровня набора (один раз на набор данных)
- LEAK_PARAMS: тесты уровня утечки (один раз на каждую утечку)
Для добавления нового набора:
1. Создать файл в test_config/datasets/
2. Тесты подхватятся автоматически
Запуск:
- Все тесты: pytest tests/test_smoke.py
- Один набор: pytest tests/test_smoke.py --suites=select_4
- Несколько наборов: pytest tests/test_smoke.py --suites=imitative_17,select_19_20
"""
import datetime
from typing import Any, List, Optional
import allure
import pytest
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from constants.test_constants import BaseTN3Constants as Base_const
from test_config.datasets import ALL_SMOKE_CONFIGS
from test_config.models_for_tests import CaseMarkers, LeakTestConfig, SmokeSuiteConfig
from test_scenarios import lds_status_scenarios
from test_scenarios import smoke_scenarios as scenarios
from test_scenarios import stationary_status_scenarios as stationary_scenarios
# ===== ГЕНЕРАЦИЯ ПАРАМЕТРОВ =====
def _get_suite_markers(config: SmokeSuiteConfig) -> List[pytest.MarkDecorator]:
"""Маркеры для тестового набора."""
return [
pytest.mark.test_suite_name(config.suite_name),
pytest.mark.test_suite_data_id(config.suite_data_id),
pytest.mark.test_data_name(config.archive_name),
pytest.mark.tu_id(config.technological_unit.id),
pytest.mark.ost_name(config.ost_name),
]
def _generate_suite_params() -> List[Any]:
"""
Генерирует параметры для тестов уровня набора данных.
Один параметр на каждый config.
"""
return [
pytest.param(config, id=config.suite_name, marks=_get_suite_markers(config)) for config in ALL_SMOKE_CONFIGS
]
def _generate_leak_params() -> List[Any]:
"""
Генерирует параметры для тестов уровня утечки.
Для single-leak конфигов: один параметр (config, leak, 1)
Для multi-leak конфигов: N параметров (config, leak_n, n) для каждой утечки
"""
params: List[Any] = []
for config in ALL_SMOKE_CONFIGS:
# Собираем все утечки из конфига
if config.has_multiple_leaks:
leaks = config.leaks
elif config.leak:
leaks = [config.leak]
else:
continue # Нет утечек в конфиге
# Создаём параметр для каждой утечки
for index, leak in enumerate(leaks):
leak_number = index + 1
# ID для Allure/pytest: select_4 или select_19_20_leak_1
if len(leaks) > 1:
param_id = f"{config.suite_name}_leak_{leak_number}"
else:
param_id = config.suite_name
params.append(
pytest.param(
config,
leak,
leak_number,
id=param_id,
marks=_get_suite_markers(config),
)
)
return params
# ===== ПАРАМЕТРЫ ДЛЯ ТЕСТОВ =====
SUITE_PARAMS: List[Any] = _generate_suite_params()
LEAK_PARAMS: List[Any] = _generate_leak_params()
# ===== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =====
def _apply_allure_markers(test_config: CaseMarkers, tag: str, title: str, description: Optional[str] = None) -> None:
"""Применяет allure-маркеры из конфига теста."""
if not test_config:
pytest.skip("Не заполнена конфигурация теста: тест пропущен")
allure.dynamic.tag(tag)
allure.dynamic.tag("SMOKE")
allure.dynamic.title(title)
if description:
allure.dynamic.description(description)
# ===== ТЕСТЫ УРОВНЯ НАБОРА =====
# Запускаются один раз для каждого config
@pytest.mark.parametrize("config", SUITE_PARAMS)
class TestSuiteScenarios:
"""
Тесты уровня набора данных.
Запускаются один раз для каждого конфига.
"""
@pytest.mark.critical_stop
def test_basic_info(self, http_client: StandHttpClient, config: SmokeSuiteConfig) -> None:
"""[BasicInfo] Проверка базовой информации СОУ: список ТУ"""
tag = "BasicInfo"
title = f"[{tag}] Проверка списка ТУ. ЭФ: Главная страница"
_apply_allure_markers(config.basic_info_test, tag, title)
scenarios.basic_info(http_client, config)
def test_journal_info(self, http_client: StandHttpClient, config: SmokeSuiteConfig) -> None:
"""[MessagesInfo] Проверка наличия сообщений в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка наличия сообщений в журнале. ЭФ: Журнал.Реальное время"
_apply_allure_markers(
config.journal_info_test,
tag,
title,
"Проверка наличия сообщений в журнале.\n" "Синхронный запрос типа: MessagesInfo",
)
scenarios.messages_exist_in_journal(http_client)
@pytest.mark.asyncio
async def test_lds_status_initialization(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[CommonScheme] Проверка режима работы СОУ: Инициализация"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Инициализация'. ЭФ: Схема"
_apply_allure_markers(
config.lds_status_initialization_test,
tag,
title,
(
f"Проверка режима работы СОУ на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.lds_status_initialization_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация"
),
)
await scenarios.lds_status_initialization(ws_client, config)
@pytest.mark.asyncio
async def test_diagnostics_of_signals_after_initialization(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig
) -> None:
"""[OutputSignalsInfo] Проверка выходных сигналов после выхода из инициализации"""
tag = "OutputSignalsInfo"
title = f"[{tag}] Проверка выходных сигналов после выхода из инициализации. ЭФ Диагностика сигналов"
_apply_allure_markers(
config.diagnostics_of_signals_after_initialization_test,
tag,
title,
(
f"Проверка состояния СОУ и перекачки на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.diagnostics_of_signals_after_initialization_test.offset} мин.\n"
),
)
test_data = config.controlled_sites_with_segment
await scenarios.diagnostics_of_signals_after_initialization(ws_client, config, test_data)
@pytest.mark.asyncio
def test_lds_status_init_in_journal(
self, http_client: StandHttpClient, config: SmokeSuiteConfig, imitator_start_time: datetime
) -> None:
"""[MessagesInfo] Проверка записи в журнале о входе СОУ в Инициализацию"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записи в журнале: СОУ в Инициализации. ЭФ: Журнал. Реальное время"
description = (
f"Проверка записи в журнале о входе СОУ в Инициализацию на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.lds_status_init_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
)
_apply_allure_markers(config.lds_status_init_in_journal_test, tag, title, description)
scenarios.lds_status_init_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_stationary_status_common_scheme(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig
) -> None:
"""[CommonScheme] Проверка режима работы и причины режима работы МТ на ЭФ: Схема"""
test_data = config.stationary_status_test_data
expected_stationary_status, expected_stationary_status_reasons = (
(test_data.expected_result if test_data else (None, None))
)
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы МТ: {expected_stationary_status},"
f" по причине: {expected_stationary_status_reasons}. ЭФ: Схема"
)
_apply_allure_markers(
config.stationary_status_common_scheme_test,
tag,
title,
(
f"Проверка режима работы МТ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.stationary_status_common_scheme_test.offset} мин.\n"
"Подписка на сообщения типа: CommonSchemeContent\n"
f"Ожидаемый режим работы МТ: {expected_stationary_status}\n "
f"Ожидаемая причина режима работы МТ: {expected_stationary_status_reasons}"
),
)
await stationary_scenarios.stationary_status_common_scheme(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_stationary_status_main_page_info(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig
) -> None:
"""[MainPageInfo] Проверка установки режима работы МТ. ЭФ: Главная страница. Контент таблицы с ТУ"""
test_data = config.stationary_status_test_data
expected_stationary_status, expected_stationary_status_reasons = (
(test_data.expected_result if test_data else (None, None))
)
tag = "MainPageInfo"
title = f"[{tag}] Проверка установки режима работы МТ: {expected_stationary_status}. " \
"ЭФ: Главная страница. Контент таблицы с ТУ"
_apply_allure_markers(
config.stationary_status_main_page_info_test,
tag,
title,
(
f"Проверка установки режима работы МТ на данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.stationary_status_main_page_info_test.offset} мин.\n"
"Подписка на сообщения типа: MainPageInfoContent\n"
f"Ожидаемый режим работы МТ: {expected_stationary_status}"
),
)
await stationary_scenarios.stationary_status_main_page_info(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_stationary_status_in_output_signals(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig
) -> None:
"""[OutputSignalsInfo] Проверка установки режима работы МТ. ЭФ: Диагностика сигналов. Выходные сигналы"""
test_data = config.stationary_status_test_data
expected_stationary_status, expected_stationary_status_reasons = (
(test_data.expected_result if test_data else (None, None))
)
tag = "OutputSignalsInfo"
title = f"[{tag}] Проверка режима работы МТ: {expected_stationary_status}. ЭФ: Диагностика сигналов. Выходные сигналы"
_apply_allure_markers(
config.stationary_status_in_output_signals_test,
tag,
title,
(
f"Проверка режима работы МТ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.stationary_status_in_output_signals_test.offset} мин.\n"
"Подписка на сообщения типа: OutputSignalsInfo\n"
f"Ожидаемый режим работы МТ: {expected_stationary_status}\n "
),
)
await stationary_scenarios.stationary_status_in_output_signals(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_main_page_info_signals(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[MainPageSignalsInfo] Проверка счетчиков состояния сигналов"""
tag = "MainPageSignalsInfo"
title = f"[{tag}] Проверка счетчиков состояния сигналов. ЭФ: Главная страница.Контент таблица по ТУ"
_apply_allure_markers(
config.main_page_info_signals_test,
tag,
title,
(
f"Проверка счетчиков состояния сигналов на данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.main_page_info_signals_test.offset} мин.\n"
"Подписка на сообщения типа: MainPageSignalsInfo\n"
),
)
await scenarios.main_page_info_signals(ws_client, config)
@pytest.mark.asyncio
async def test_imitate_flowmeter_signal(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
"""[ImitateSignal] Проверка имитации расходомера"""
tag = "ImitateSignal"
title = f"[{tag}] Проверка имитации сигнала расходомера. ЭФ: Схема, Входные сигналы, Журнал"
_apply_allure_markers(
config.imitate_flowmeter_signal_test,
tag,
title,
(
"Проверка работы имитации и снятия имитации c расходомера "
f"на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.imitate_flowmeter_signal_test.offset} мин.\n"
"Синхронные запросы типа: ImitateSignalRequest, UnImitateSignalRequest\n"
"Подписка на сообщения типа: InputSignalsContent\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос имитации.\n"
"Значения в полях isImitated, quality и imitation.value расходомера "
"во входных сигналах после имитации.\n"
"Статус-код ответа на синхронный запрос снятия имитации.\n"
"Значение в поле isImitated расходомера во входных сигналах после снятия имитации.\n"
"Примечание: что бы не повлиять на проверки утечек, тест на имитацию выполняется во время "
"инициализации"
),
)
test_data = config.imitate_flowmeter_signal_test_data
await scenarios.imitate_sensor_signal(ws_client, http_client, config, test_data, imitator_start_time)
@pytest.mark.asyncio
async def test_imitate_pressure_sensor_signal(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
"""[ImitateSignal] Проверка имитации датчика давления"""
tag = "ImitateSignal"
title = f"[{tag}] Проверка имитации сигнала датчика давления. ЭФ: Схема, Входные сигналы, Журнал"
_apply_allure_markers(
config.imitate_pressure_sensor_signal_test,
tag,
title,
(
"Проверка работы имитации и снятия имитации c датчика давления "
f"на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.imitate_pressure_sensor_signal_test.offset} мин.\n"
"Синхронные запросы типа: ImitateSignalRequest, UnImitateSignalRequest\n"
"Подписка на сообщения типа: InputSignalsContent\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос имитации.\n"
"Значения в полях isImitated, quality и imitation.value датчика давления "
"во входных сигналах после имитации.\n"
"Статус-код ответа на синхронный запрос снятия имитации.\n"
"Значение в поле isImitated датчика давления во входных сигналах после снятия имитации.\n"
"Примечание: что бы не повлиять на проверки утечек, тест на имитацию выполняется во время "
"инициализации"
),
)
test_data = config.imitate_pressure_sensor_signal_test_data
await scenarios.imitate_sensor_signal(ws_client, http_client, config, test_data, imitator_start_time)
@pytest.mark.asyncio
async def test_mask_signal(
self, ws_client: WebSocketClient, http_client: StandHttpClient, config: SmokeSuiteConfig
) -> None:
"""[MaskSignal] Проверка маскирования датчиков"""
tag = "MaskSignal"
title = f"[{tag}] проверка маскирования датчиков. ЭФ: Схема"
_apply_allure_markers(
config.mask_signal_test,
tag,
title,
(
f"Проверка работы маскирования и снятия маскирования на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.mask_signal_test.offset} мин.\n"
"Синхронные запросы типа: MaskSignalRequest, UnmaskSignalRequest\n"
"Подписка на сообщения типа: InputSignalsContent\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос маскирования.\n"
"Значение в поле isMasked сигнала во входных сигналах после маскирования.\n"
"Статус-код ответа на синхронный запрос снятия маскирования.\n"
"Значение в поле isMasked сигнала во входных сигналах после снятия маскирования.\n"
"Примечание: что бы не повлиять на проверки утечек, тест на маскирование выполняется во время "
"инициализации"
),
)
test_data = config.mask_signal_test_data
await scenarios.mask_signal_test(ws_client, http_client, config, test_data)
def test_mask_info_in_journal(
self, http_client: StandHttpClient, config: SmokeSuiteConfig, imitator_start_time: datetime
) -> None:
"""[MessagesInfo] Проверка записей журнала о маскировании и размаскировании"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записей журнала о маскировании и размаскировании. ЭФ: Журнал.Реальное время"
description = (
f"Проверка записей журнала о маскировании и размаскировании на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.mask_info_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo\n"
"Проверки: структура, фильтрация по времени, совпадение тегов между маскированием и снятием.\n"
"Примечание: тест выполняется после теста маскирования, чтобы записи успели попасть в журнал"
)
_apply_allure_markers(config.mask_info_in_journal_test, tag, title, description)
scenarios.mask_info_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_mask_du_on_mini_scheme(
self, ws_client: WebSocketClient, http_client: StandHttpClient, config: SmokeSuiteConfig
) -> None:
"""Проверка маскирования ДУ"""
tag = "MaskDuOnMiniScheme"
title = f"[{tag}] проверка маскирования линейного участка. ЭФ: Схема"
description = (
f"Проверка маскирования линейного участка, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.mask_signal_test.offset} мин.\n"
"Синхронные запросы типа: MaskLdsCommandRequest\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос маскирования.\n"
"Проверка маскирования ДУ в выходных сигналах.\n"
"Проверка маскирования на схеме.\n"
"Проверка сообщения о маскировании в журнале.\n"
)
_apply_allure_markers(config.mask_du_on_mini_scheme_test, tag, title, description)
await scenarios.mask_du_on_mini_scheme(ws_client, http_client, config)
@pytest.mark.asyncio
async def test_unmask_du_on_mini_scheme(
self, ws_client: WebSocketClient, http_client: StandHttpClient, config: SmokeSuiteConfig
) -> None:
"""Проверка размаскирования ДУ"""
tag = "UnmaskDuOnMiniScheme"
title = f"[{tag}] проверка снятия маскирования линейному участку. ЭФ: Схема"
description = (
f"Проверка снятия маскирования линейному участку, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.unmask_du_on_mini_scheme_test.offset} мин.\n"
"Синхронные запросы типа: UnmaskDuOnMiniScheme\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос снятия маски.\n"
"Проверка тега с маской в выходных сигналах.\n"
"Проверка сообщения о снятии маскирования в журнале.\n"
)
_apply_allure_markers(config.unmask_du_on_mini_scheme_test, tag, title, description)
await scenarios.unmask_du_on_mini_scheme(ws_client, http_client, config)
@pytest.mark.asyncio
async def test_lds_status_initialization_out(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[CommonScheme] Проверка выхода СОУ из Инициализации"""
tag = "CommonScheme"
title = f"[{tag}] Проверка выхода СОУ из Инициализации. ЭФ: Схема"
_apply_allure_markers(
config.lds_status_initialization_out_test,
tag,
title,
(
f"Проверка выхода СОУ из Инициализации на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.lds_status_initialization_out_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: не Инициализация"
),
)
await scenarios.lds_status_initialization_out(ws_client, config)
def test_lds_status_init_out_in_journal(
self, http_client: StandHttpClient, config: SmokeSuiteConfig, imitator_start_time: datetime
) -> None:
"""[MessagesInfo] Проверка записи в журнале о выходе СОУ из Инициализации"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записи в журнале: выход СОУ из Инициализации. ЭФ: Журнал. Реальное время"
description = (
f"Проверка записи в журнале о выходе СОУ из Инициализации на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.lds_status_init_out_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
)
_apply_allure_markers(config.lds_status_init_out_in_journal_test, tag, title, description)
scenarios.lds_status_init_out_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_main_page_info_unstationary(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[MainPageInfo] Проверка установки режима Нестационарный (для multi-leak)"""
tag = "MainPageInfo"
title = (
f"[{tag}] Проверка установки режима работы МТ: нестационарный. ЭФ: Главная страница. Контент таблица по ТУ"
)
_apply_allure_markers(
config.main_page_info_unstationary_test,
tag,
title,
(
f"Проверка установки режима работы МТ: нестационарный на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.main_page_info_unstationary_test.offset} мин.\n"
"Подписка на сообщения типа: MainPageInfo\n"
"Ожидаемый режим работы МТ: Нестационарный"
),
)
await scenarios.main_page_info_unstationary(ws_client, config)
def test_stationary_status_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
) -> None:
"""[MessagesInfo] Проверка режима работы МТ и причины режима работы МТ на ЭФ: Журнал. Реальное время"""
test_data = config.stationary_status_journal_test_data
expected_stationary_status, expected_stationary_status_reasons = (
(test_data.expected_result if test_data else (None, None))
)
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка режима работы МТ: {expected_stationary_status},"
f" по причине: {expected_stationary_status_reasons}. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.stationary_status_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.stationary_status_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=PUMPING_STATUS\n"
f"Ожидаемый режим работы МТ: {expected_stationary_status}\n "
f"Ожидаемая причина режима работы МТ: {expected_stationary_status_reasons}"
),
)
stationary_scenarios.stationary_status_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_export_lds_status_report(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
tag = "ExportReports"
title = f"[{tag}] Проверка формирования отчёта о режиме работы СОУ. ЭФ: Выпадашка отчётов"
_apply_allure_markers(
config.export_lds_status_report_test,
tag,
title,
(
f"Проверка формирования и содержимого xlsx-отчёта о режиме работы СОУ на наборе данных "
f"{config.suite_name},\n"
f"на технологическом участке {config.tu_name}\n"
f"Период отчёта: от старта имитатора до старта + сдвиг теста "
f"{config.export_lds_status_report_test.offset} мин.\n"
"Этапы сценария:\n"
"1) SubscribeReportsDataExportedRequest - подписка на пуш-нотификации\n"
"2) ExportReportsCommandRequest - запрос формирования отчёта (тип LdsStateReport, фильтр по периоду)\n"
"3) Ожидание ReportDataExportedNotification\n"
"4) Лонг-поллинг GetExportedDataListRequest - поиск отчёта в списке\n"
"5) DownloadExportedDataRequest (StreamInvocation) - скачивание по exportedDataId\n"
"6) Проверка xlsx: участки, длительности режимов СОУ, суммарное время работы\n"
"7) Проверка двойной шапки и названий колонок\n"
"8) Проверка имени файла (.xlsx, название отчёта, ТУ, период +-1 мин)\n"
"Во вложениях Allure xlsx прикладывается только при падении теста"
),
)
await scenarios.export_lds_status_report(ws_client, http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_export_mt_mode_report(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
tag = "ExportReports"
title = f"[{tag}] Проверка формирования отчёта о режиме работы МТ. ЭФ: Выпадашка отчётов"
_apply_allure_markers(
config.export_mt_mode_report_test,
tag,
title,
(
f"Проверка формирования и содержимого xlsx-отчёта о режиме работы МТ на наборе данных "
f"{config.suite_name},\n"
f"на технологическом участке {config.tu_name}\n"
f"Период отчёта: от старта имитатора до старта + сдвиг теста "
f"{config.export_mt_mode_report_test.offset} мин.\n"
"Этапы сценария:\n"
"1) SubscribeReportsDataExportedRequest - подписка на пуш-нотификации\n"
"2) ExportReportsCommandRequest - запрос формирования отчёта (тип StationaryStatusReport)\n"
"3) Ожидание ReportDataExportedNotification\n"
"4) Лонг-поллинг GetExportedDataListRequest - поиск отчёта в списке\n"
"5) DownloadExportedDataRequest (StreamInvocation) - скачивание по exportedDataId\n"
"6) Проверка xlsx: участки, длительности режимов МТ, суммарное время, доминирующий режим\n"
"7) Проверка двойной шапки и названий колонок\n"
"8) Проверка имени файла (.xlsx, название отчёта, ТУ, период +-1 мин)\n"
"Во вложениях Allure xlsx прикладывается только при падении теста"
),
)
await scenarios.export_mt_mode_report(ws_client, http_client, config, imitator_start_time)
# ===== ТЕСТЫ УРОВНЯ УТЕЧКИ =====
# Запускаются для каждой утечки в конфиге
@pytest.mark.parametrize("config, leak, leak_number", LEAK_PARAMS)
class TestLeakScenarios:
"""
Тесты уровня утечки.
Для single-leak: запускаются один раз.
Для multi-leak: запускаются для каждой утечки отдельно.
"""
@pytest.mark.asyncio
async def test_all_leaks_info(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[AllLeaksInfo] Проверка утечки в пуше"""
tag = "AllLeaksInfo"
title = f"[{tag}] Проверка сообщения об утечке. ЭФ: Пуш-сообщение об утечке на всех ЭФ"
_apply_allure_markers(
leak.all_leaks_info_test,
tag,
title,
(
f"Проверка пуш-сообщения об утечке на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.all_leaks_info_test.offset} мин.\n"
"Подписка на сообщения типа: AllLeaksInfo\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.all_leaks_info(ws_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_all_leaks_is_empty(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[AllLeaksInfo] Проверка отсутствия информации об утечке"""
tag = "AllLeaksInfo"
title = f"[{tag}] Проверка отсутствия информации об утечке в сообщении. ЭФ: Пуш-сообщение об утечке"
_apply_allure_markers(
leak.all_leaks_is_empty_test,
tag,
title,
(
"Проверка отсутствия информации об утечке в сообщении, \n"
f"Время проведения проверки: {leak.all_leaks_is_empty_test.offset} мин.\n"
"Подписка на сообщения типа: AllLeaksInfo\n"
"Примечание: проверяем, когда в КГ табличный вид список пуст"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.all_leaks_is_empty(ws_client, config)
@pytest.mark.asyncio
async def test_leaks_content(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[LeaksContent] Проверка сообщения об утечке в таблице КГ"""
tag = "LeaksContent"
title = f"[{tag}] Проверка сообщения об утечке. ЭФ: КГ.Табличное представление"
_apply_allure_markers(
leak.leaks_content_test,
tag,
title,
(
f"Проверка сообщения об утечке в таблице КГ на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leaks_content_test.offset} мин.\n"
"Подписка на сообщения типа: LeaksContent\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
# Добавляем номер утечки в title для multi-leak
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leaks_content(ws_client, config, leak, imitator_start_time)
def test_leak_info_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка сообщения об утечке в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка сообщения об утечке в журнале. ЭФ: Журнал.Реальное время"
_apply_allure_markers(
leak.leak_info_in_journal,
tag,
title,
(
f"Проверка сообщения об утечке в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_info_in_journal.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
# Добавляем номер утечки в title для multi-leak
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.leak_info_in_journal(http_client, config, leak, imitator_start_time)
def test_completed_leak_info_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка сообщения о завершенной утечке в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка сообщения о завершенной утечке в журнале. ЭФ: Журнал.Реальное время"
description = (
f"Проверка сообщения о завершенной утечке в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.completed_leak_info_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo\n"
)
_apply_allure_markers(
leak.completed_leak_info_in_journal_test,
tag,
title,
description,
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.completed_leak_info_in_journal(http_client, config, leak, imitator_start_time)
def test_possible_leak_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка наличия сообщения 'Возможна утечка' в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка сообщения 'Возможна утечка' в журнале. ЭФ: Журнал. Реальное время"
description = (
f"Проверка наличия сообщения 'Возможна утечка' в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.possible_leak_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo"
)
_apply_allure_markers(leak.possible_leak_in_journal_test, tag, title, description)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.possible_leak_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_tu_leaks_info(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[TuLeaksInfo] Проверка утечки на ТУ"""
tag = "TuLeaksInfo"
title = f"[{tag}] Проверка сообщения об утечке. Сообщения на ЭФ: Схема, Минисхема, Гидроуклон"
_apply_allure_markers(
leak.tu_leaks_info_test,
tag,
title,
(
f"Проверка сообщения об утечке на схемах и гидроуклоне на наборе данных {config.suite_name}, \n"
f"на технологическом участке {leak.diagnostic_area_name}\n"
f"Время проведения проверки: {leak.tu_leaks_info_test.offset} мин.\n"
"Подписка на сообщения типа: TuLeaksInfo\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.tu_leaks_info(ws_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_lds_status_during_leak(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[CommonScheme] Проверка режима работы СОУ во время утечки"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ во время утечки. ЭФ: Схема"
_apply_allure_markers(
leak.lds_status_during_leak_test,
tag,
title,
(
f"Проверка режима работы СОУ во время утечки на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.lds_status_during_leak_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Примечание: проверка режимов СОУ во время утечки должна выполняться раньше теста на квитирование\n"
"В рамках данного теста проверяется режим СОУ на ДУ с утечкой и на соседних ДУ"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.lds_status_during_leak(ws_client, config, leak)
@pytest.mark.asyncio
async def test_acknowledge_leak_info(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[AcknowledgeLeak] Проверка квитирования утечки"""
tag = "AcknowledgeLeak"
title = f"[{tag}] Проверка квитирования утечки. Отсутствие Пуш-сообщений об утечке на всех ЭФ"
_apply_allure_markers(
leak.acknowledge_leak_test,
tag,
title,
(
"Проверка квитирования утечки на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.acknowledge_leak_test.offset} мин.\n"
"Синхронный запрос типа: AcknowledgeLeak\n"
"Подписка на сообщения типа: TuLeaksInfo, AllLeaksInfo\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос о квитировании,\n"
"Отсутствие пуш-сообщений об утечках после квитирования"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.acknowledge_leak_info(ws_client, http_client, config, leak)
def test_acknowledge_leak_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка записи в журнале о квитировании утечки"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записи в журнале о квитировании утечки. ЭФ: Журнал. Реальное время"
description = (
f"Проверка записи о квитировании утечки в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.acknowledge_leak_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром userActions=LEAK_ACK\n"
)
_apply_allure_markers(leak.acknowledge_leak_in_journal_test, tag, title, description)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.acknowledge_leak_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_output_signals(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[OutputSignalsInfo] Проверка данных об утечке в выходных сигналах"""
tag = "OutputSignalsInfo"
title = (
f"[{tag}] Проверка наличия данных об утечке в выходных сигналах. ЭФ: Диагностика сигналов.Выходные "
f"сигналы"
)
_apply_allure_markers(
leak.output_signals_test,
tag,
title,
(
f"Проверка наличия данных об утечке в выходных сигналах на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.output_signals_test.offset} мин.\n"
"Синхронный запрос типа: GetOutputSignals\n"
"Подписка на сообщения типа: SubscribeOutputSignals\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: "
"В offset указано время проверок сообщения выходных сигналов + 1 минута "
"для корректной отработки проверок.\n"
"Данный тест так же проверяет квитирование, offset выставлять после запуска теста на квитирование "
"утечки"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.output_signals(ws_client, http_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_balance_algorithm_leak_waiting(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[BalanceAlgorithmResultsContent] Проверка сообщения 'подозрение на утечку' в таблице КГ"""
tag = "BalanceAlgorithmResultsContent"
title = f"[{tag}] Проверка сообщения 'подозрение на утечку'. ЭФ: КГ.Графическое представление"
_apply_allure_markers(
leak.balance_algorithm_leak_waiting_test,
tag,
title,
(
f"Проверка сообщения 'подозрение на утечку' в таблице КГ, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: "
f"{leak.balance_algorithm_leak_waiting_test.offset} - "
f"{leak.balance_algorithm_leak_waiting_test.offset + Base_const.BALANCE_ALGORITHM_TOTAL_WAIT / 60} "
f"мин.\n"
"Подписка на сообщения типа: BalanceAlgorithmResultsContent\n"
),
)
# Добавляем номер утечки в title для multi-leak
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.balance_algorithm_leak_waiting(ws_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_balance_algorithm_leak_detected(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[BalanceAlgorithmResultsContent] Проверка наличия утечки (isLeakDetected) в графическом представлении КГ"""
tag = "BalanceAlgorithmResultsContent"
title = f"[{tag}] Проверка наличия утечки (isLeakDetected). ЭФ: КГ Графическое представление"
_apply_allure_markers(
leak.balance_algorithm_leak_detected_test,
tag,
title,
(
f"Проверка наличия утечки (isLeakDetected) в графическом представлении КГ, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.balance_algorithm_leak_detected_test.offset} мин.\n"
"Подписка на сообщения типа: BalanceAlgorithmResultsContent\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.balance_algorithm_leak_detected(ws_client, config, leak)
@pytest.mark.asyncio
async def test_balance_algorithm_leak_completed(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""
[BalanceAlgorithmResultsContent] Проверка отсутствия утечки (isLeakDetected) в графическом представлении КГ
"""
tag = "BalanceAlgorithmResultsContent"
title = (
f"[{tag}] Проверка отсутствия утечки (isLeakDetected) после завершения утечки. "
f"ЭФ: КГ Графическое представление"
)
_apply_allure_markers(
leak.balance_algorithm_leak_detected_test,
tag,
title,
(
f"Проверка отсутствия утечки (isLeakDetected) в графическом представлении КГ, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.balance_algorithm_leak_detected_test.offset} мин.\n"
"Подписка на сообщения типа: BalanceAlgorithmResultsContent\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.balance_algorithm_leak_completed(ws_client, config)
@pytest.mark.asyncio
async def test_lds_status_after_confirming_leak(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig, leak: LeakTestConfig, leak_number: int
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация
Причина: Накопление данных
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: Инициализация, по причине: Накопление данных. ЭФ: Схема"
_apply_allure_markers(
leak.lds_status_after_confirming_leak_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{leak.lds_status_after_confirming_leak_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
test_data = leak.lds_status_after_confirming_leak_data
await lds_status_scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
await lds_status_scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_completed_leak(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig, leak: LeakTestConfig, leak_number: int
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ после завершения утечки: Инициализация
Причина: Накопление данных
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ после завершения утечки: Инициализация"
"Причина: Накопление данных. ЭФ: Схема"
)
_apply_allure_markers(
leak.lds_status_after_confirming_leak_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{leak.lds_status_completed_leak_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
test_data = leak.lds_status_after_completed_leak_data
await lds_status_scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_leak_is_complete_on_main_page(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка отсутствия утечки на ЭФ Состояние МТ"""
tag = "MainPageInfoContent"
title = f"[{tag}] Проверка отсутствия утечки на ЭФ Состояние МТ."
_apply_allure_markers(
leak.leak_is_complete_on_main_page_test,
tag,
title,
(
f"Проверка отсутствия утечки на ЭФ Состояние МТ\n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_is_complete_on_main_page_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_complete_on_main_page(ws_client, config)
@pytest.mark.asyncio
async def test_leak_is_confirm_on_main_page(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка подтвержденной утечки на ЭФ Состояние МТ"""
tag = "MainPageInfoContent"
title = f"[{tag}] Проверка подтвержденной утечки на ЭФ Состояние МТ"
_apply_allure_markers(
leak.leak_is_confirm_on_main_page_test,
tag,
title,
(
f"Проверка наличия утечки на ЭФ Состояние МТ \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_is_confirm_on_main_page_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_confirm_on_main_page(ws_client, config)
@pytest.mark.asyncio
async def test_the_leak_is_complete_on_kg(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка события завершения утечки в контроле герметичности"""
tag = "LeaksContent"
title = f"[{tag}] Проверка cобытия завершения утечки ЭФ КГ табличный вид."
_apply_allure_markers(
leak.the_leak_is_complete_on_kg_test,
tag,
title,
(
f"Проверка ЭФ КГ табличное представление, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.the_leak_is_complete_on_kg_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_complete_on_kg(ws_client, config, leak)
@pytest.mark.asyncio
async def test_leak_is_complete_in_output_signals(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка отсутствия утечки в выходных сигналах"""
tag = "OutputSignalsInfo"
title = f"[{tag}] Проверка отсутствия утечки в выходных сигналах."
_apply_allure_markers(
leak.leak_is_complete_in_output_signals_test,
tag,
title,
(
f"Проверка ЭФ Выходные сигналы \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_is_complete_in_output_signals_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_complete_in_output_signals(ws_client, http_client, config, leak)
@pytest.mark.asyncio
async def test_complete_tu_leaks_info_content(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка сообщения в журнале о завершении утечки"""
tag = "TuLeaksInfoContent"
title = f"[{tag}] Проверка отсутствия утечки на схеме"
_apply_allure_markers(
leak.complete_tu_leaks_info_content_test,
tag,
title,
(
f"Проверка ЭФ Схема, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.complete_tu_leaks_info_content_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.complete_tu_leaks_info_content(ws_client, config)
@pytest.mark.asyncio
async def test_export_leaks_report(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
tag = "ExportReports"
title = f"[{tag}] Проверка формирования отчёта об утечках. ЭФ: Выпадашка отчётов"
_apply_allure_markers(
leak.export_leaks_report_test,
tag,
title,
(
f"Проверка формирования и содержимого xlsx-отчёта об утечках на наборе данных "
f"{config.suite_name},\n"
f"на технологическом участке {config.tu_name}\n"
f"Период отчёта: от старта имитатора до старта + сдвиг теста"
f"{leak.export_leaks_report_test.offset} мин.\n"
"Этапы сценария:\n"
"1) SubscribeReportsDataExportedRequest - подписка на пуш-нотификации\n"
"2) ExportReportsCommandRequest - запрос формирования отчёта (тип LeaksReport, фильтр по периоду)\n"
"3) Ожидание ReportDataExportedNotification - пуш-нотификация о формировании отчёта\n"
"4) Лонг-поллинг GetExportedDataListRequest - поиск отчёта в списке "
"(имя, ТУ, период start/end с погрешностью +-1 мин)\n"
"5) DownloadExportedDataRequest (StreamInvocation) - скачивание по exportedDataId\n"
"6) Приём fileChunk, проверка формата xlsx (zip-сигнатура, контент существует)\n"
"7) Проверка имени файла: .xlsx, 'Отчет об утечках', название ТУ, "
"диапазон дат в имени (+-1 мин)\n"
"8) Проверка шапки xlsx: заголовок с периодом (+-1 мин), названия колонок\n"
"9) Проверка строки утечки: дата в периоде, объект, режим СОУ, маскирование, "
"координата, объём, режим работы МТ\n"
"Во вложениях Allure xlsx прикладывается только при падении теста"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.export_leaks_report(ws_client, http_client, config, leak, imitator_start_time)
