async def stationary_status_in_output_signals(ws_client, cfg: BaseSuiteConfig, test_data: CaseData):
"""
Проверка установки режима МТ в сообщении OutputSignalsInfo
"""
# Распаковка тестовых данных
if not test_data:
pytest.fail("Не заполнены данные для теста")
control_points = test_data.params.get(TestConst.CONTROL_POINTS_KEY, [])
if not control_points:
pytest.fail(f"Не заполнены обязательные параметры для теста: {TestConst.CONTROL_POINTS_KEY}")
expected_stationary_status, _ = test_data.expected_result
# получает часть словаря для всего списка 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': value[0], 'segmentId': value[1]} for key, value 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 = t_utils.collect_stationary_statuses_from_output_signals(
controlled_site_signals, controlled_sites_dict
)
output_signals_details = "\n".join(
f"{control_point}: {stationary_status}"
for control_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()