Загрузка данных
async def main_page_info(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка установки режима МТ по MainPageInfo, OutputSignalsInfo и CommonSchemeContent.
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent"):
main_page_payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_main_page = parser.parse_main_page_msg(main_page_payload)
with allure.step("Подключение по ws, получение и обработка сообщения типа: OutputSignalsInfo"):
output_signals_payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"OutputSignalsInfo",
"SubscribeOutputSignalsRequest",
{
'objects': {
'linearParts': [],
'controlledSites': [site.controlledSiteSegmentDict for site in SiteKpKp],
},
'signalTypes': 1023,
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_output_signals = parser.parse_output_signals_info_msg(output_signals_payload)
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
common_scheme_payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"CommonSchemeContent",
"SubscribeCommonSchemeRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_common_scheme = parser.parse_common_scheme_info_msg(common_scheme_payload)
with allure.step("Извлечение и подготовка данных для проверки"):
tu_info = getattr(parsed_main_page.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
)
controlled_site_signals = getattr(parsed_output_signals.replyContent, 'controlledSiteSignals', [])
output_signals_site_stationary_statuses = t_utils.collect_stationary_statuses_from_output_signals(
controlled_site_signals
)
output_signals_stationary_statuses = [
stationary_status for _, stationary_status in output_signals_site_stationary_statuses
]
StepCheck(
"Проверка наличия режимов МТ в OutputSignalsInfo",
"controlledSiteSignals",
).actual(output_signals_stationary_statuses).is_not_empty()
output_signals_majority = t_utils.determine_stationary_status_by_majority(
output_signals_stationary_statuses
)
output_signals_details = "\n".join(
f"{site.name}: {stationary_status}"
for site, stationary_status in output_signals_site_stationary_statuses
)
allure.attach(
output_signals_details,
name="OutputSignalsInfo: режимы МТ по участкам КП-КП",
attachment_type=allure.attachment_type.TEXT,
)
flow_areas = t_utils.filter_flow_areas_with_available_flow(
getattr(parsed_common_scheme.replyContent, 'flowAreas', [])
)
StepCheck("Проверка наличия участков карты течения с доступным течением", "flowAreas").actual(
flow_areas
).is_not_empty()
longest_flow_area = t_utils.get_longest_flow_area_by_pipes(flow_areas)
StepCheck("Проверка наличия самого длинного пути течения", "longest_flow_area").actual(
longest_flow_area
).is_not_none()
diagnostic_areas = getattr(longest_flow_area, 'diagnosticAreas', [])
StepCheck("Проверка наличия данных диагностических участков", "diagnosticAreas").actual(
diagnostic_areas
).is_not_empty()
base_ids = set(t_utils.get_diagnostic_area_base_ids())
excluded_ids = set(TestConst.DIAGNOSTIC_AREA_IDS_EXCLUDED_GRAVITY) | set(
TestConst.DIAGNOSTIC_AREA_IDS_EXCLUDED_NPS
)
common_scheme_stationary_statuses = []
common_scheme_details = []
for diagnostic_area in diagnostic_areas:
if diagnostic_area.id not in base_ids or diagnostic_area.id in excluded_ids:
continue
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)
common_scheme_details.append(
f"{t_utils.diagnostic_area_title(diagnostic_area.id)}: {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
)
allure.attach(
f"Самый длинный путь течения: {longest_flow_area}\n" + "\n".join(common_scheme_details),
name="CommonSchemeContent: режимы МТ на базовых ДУ",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Проверка режима работы МТ"):
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_main_page.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck(
f"MainPageInfo: режим МТ для ТУ {cfg.tu_name}",
"stationaryStatus",
soft_failures,
).actual(main_pipeline_stationary_status).expected(cfg.expected_stationary_status).equal_to()
StepCheck(
"OutputSignalsInfo: общий режим МТ по большинству участков",
"stationaryStatus",
soft_failures,
).actual(output_signals_majority).expected(cfg.expected_stationary_status).equal_to()
StepCheck(
"CommonSchemeContent: общий режим МТ по большинству базовых ДУ",
"stationaryStatus",
soft_failures,
).actual(common_scheme_majority).expected(cfg.expected_stationary_status).equal_to()