Загрузка данных


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()