https://pastein.ru/t/KFN
Загрузка данных
clients/opc_ua_client.py
"""
OPC UA клиент для чтения выходных сигналов СОУ (обратный обмен СДКУ).
Подключается к OPC UA серверу по opc.tcp и читает значения тегов
(Value + StatusCode + SourceTimestamp). Используется как в setup-части
(проверка доступности OPC), так и в сценариях имитации выходных сигналов.
"""
import asyncio
import logging
from typing import Any, Optional
from asyncua import Client, ua
from constants.architecture_constants import OpcUaConstants as OpcConst
logger = logging.getLogger(__name__)
def build_opc_node_id(
ost_name: str,
stand_name: str,
address: str,
suffix: str,
namespace_index: int = OpcConst.NAMESPACE_INDEX,
) -> str:
"""
Собирает NodeId вида ns=2;s=CHTNTEST4:CHTN_linearParts_9.isLeakDetected.
ost_name: имя ОСТ (например "CHTN").
stand_name: имя стенда (например "TEST4").
address: тег участка из конфигурации трубы (например "CHTN_linearParts_9").
suffix: суффикс сигнала (например "isLeakDetected").
namespace_index: номер пространства имён (по умолчанию NS=2).
"""
return f"ns={namespace_index};s={ost_name}{stand_name}:{address}.{suffix}"
def read_data_value_sync(
opc_url: str,
node_id: str,
timeout: float = OpcConst.CONNECT_TIMEOUT_S,
) -> ua.DataValue:
"""
Синхронное подключение и чтение DataValue тега. Используется в setup-части
(синхронный контекст pytest), чтобы понять, что OPC UA сервер жив и отдаёт данные.
"""
async def _read() -> ua.DataValue:
client = Client(url=opc_url, timeout=timeout)
try:
await client.connect()
return await client.get_node(node_id).read_data_value()
finally:
await client.disconnect()
return asyncio.run(_read())
class OpcUaClient:
"""Асинхронный OPC UA клиент с переподключением."""
def __init__(self, url: str, timeout: float = OpcConst.OPERATION_TIMEOUT_S) -> None:
self._url = url
self._timeout = timeout
self._client: Optional[Client] = None
async def connect(self, attempts: int = OpcConst.RECONNECT_ATTEMPTS) -> None:
"""Подключается к OPC UA серверу с ограниченным числом попыток."""
last_error: Optional[Exception] = None
for attempt in range(1, attempts + 1):
try:
self._client = Client(url=self._url, timeout=self._timeout)
await self._client.connect()
logger.info("[OPC] [OK] Подключено к %s", self._url)
return
except Exception as error:
last_error = error
logger.warning(
"[OPC] [WARNING] Подключение не удалось (попытка %s/%s): %s", attempt, attempts, error
)
if attempt < attempts:
await asyncio.sleep(OpcConst.RECONNECT_INTERVAL_S)
raise ConnectionError(f"Не удалось подключиться к OPC UA серверу {self._url}: {last_error}") from last_error
async def disconnect(self) -> None:
if self._client is not None:
await self._client.disconnect()
self._client = None
async def read_data_value(self, node_id: str) -> ua.DataValue:
"""Читает DataValue (Value + StatusCode + SourceTimestamp) тега."""
if self._client is None:
raise RuntimeError("OPC UA клиент не подключён")
return await self._client.get_node(node_id).read_data_value()
async def read_value(self, node_id: str) -> Any:
"""Читает только значение тега."""
if self._client is None:
raise RuntimeError("OPC UA клиент не подключён")
return await self._client.get_node(node_id).read_value()
constants/architecture_constants.py
class OpcUaConstants:
"""Константы для OPC UA клиента (чтение выходных сигналов СОУ)."""
# Номер пространства имён (namespace index) для тегов СОУ. На стенде всегда NS=2.
NAMESPACE_INDEX: int = 2
# Таймаут подключения и операций OPC UA (в секундах).
CONNECT_TIMEOUT_S: float = 10.0
OPERATION_TIMEOUT_S: float = 5.0
# Число попыток переподключения и пауза между ними.
RECONNECT_ATTEMPTS: int = 6
RECONNECT_INTERVAL_S: float = 5.0
# Пробный тег для setup-проверки доступности OPC UA (минимальный, не зависит от набора данных).
PROBE_ADDRESS: str = "CHTN_linearParts_9"
PROBE_SUFFIX: str = "isLeakDetected"
# Допуск при проверке времени изменения сигнала (SourceTimestamp), в секундах.
TIMESTAMP_DELTA_S: float = 60.0
енам
class OutputSignalType(Enum):
"""
Типы выходных сигналов СОУ, которые имитируются через api-gateway (обратный обмен СДКУ).
Значение каждого элемента — кортеж:
(signal_type, opc_suffix, section_type)
- signal_type: значение поля signalType в imitateOutputSignalRequest (контракт asyncapi).
- opc_suffix: суффикс OPC-тега, из которого собирается NodeId вида
ns=2;s={ost}{stand}:{address}.{suffix}.
- section_type: тип участка, к которому относится сигнал:
"linearParts" (ДУ, НПС-НПС) или "controlledSites" (КП-КП, сегменты ДУ).
"""
LEAK = ("leak", "isLeakDetected", "linearParts")
LEAK_COORDINATE = ("leakCoordinate", "leakCoordinate", "linearParts")
LEAK_VOLUME = ("leakVolume", "leakVolume", "linearParts")
LEAK_TIME = ("leakTime", "leakDetectedAt", "linearParts")
ACKNOWLEDGE = ("acknowledge", "isLeakAcknowledged", "linearParts")
MASK = ("mask", "isMasked", "linearParts")
MASK_REASON = ("maskReason", "maskReason", "linearParts")
PUMPING_STATUS = ("pumpingStatus", "pumpingStatus", "controlledSites")
LDS_STATUS = ("ldsStatus", "ldsStatus", "controlledSites")
FREE_FLOW = ("freeFlow", "hasFreeFlow", "controlledSites")
def __init__(self, signal_type: str, opc_suffix: str, section_type: str) -> None:
self.signal_type = signal_type
self.opc_suffix = opc_suffix
self.section_type = section_type
def __str__(self) -> str:
return f"{self.signal_type} -> .{self.opc_suffix} ({self.section_type})"
стенд сет мен
logger.info(f"[SETUP] [OK] OPC сервер {host}:{port} доступен")
def check_opc_server_availability(self) -> None:
"""
Проверяет доступность OPC UA сервера с runner'а: реальное подключение
(asyncua) и чтение пробного тега выходного сигнала.
"""
from clients.opc_ua_client import build_opc_node_id, read_data_value_sync
from constants.architecture_constants import OpcUaConstants as OpcConst
opc_url = os.environ.get(EnvKeyConstants.OPC_URL)
if not opc_url:
raise RuntimeError("[SETUP] [ERROR] OPC_URL не задан в переменных окружения")
probe_node_id = build_opc_node_id(
ost_name=self._ost_name,
stand_name=self._stand_name,
address=OpcConst.PROBE_ADDRESS,
suffix=OpcConst.PROBE_SUFFIX,
)
try:
data_value = read_data_value_sync(opc_url, probe_node_id)
except Exception as error:
raise RuntimeError(f"[SETUP] [ERROR] OPC UA сервер недоступен ({opc_url}): {error}") from error
status = data_value.StatusCode
if not status.is_good():
raise RuntimeError(f"[SETUP] [ERROR] OPC UA тег {probe_node_id} вернул плохой статус: {status}")
value = data_value.Value.Value if data_value.Value else None
snapshot = (
f"{probe_node_id}\n"
f"Value={value}\n"
f"StatusCode={status}\n"
f"SourceTimestamp={data_value.SourceTimestamp}"
)
logger.info("[SETUP] [OK] OPC UA сервер доступен. Слепок выходного сигнала:\n%s", snapshot)
allure.attach(
snapshot,
name="OPC UA выходной сигнал",
attachment_type=allure.attachment_type.TEXT,
)
test_config/datasets/__init__.py
"ALL_IS_REJECTED_CONFIGS",
"ALL_OUTPUT_SIGNAL_IMITATION_CONFIGS",
"ALL_IS_REJECTED_CONFIGS",
"ALL_OUTPUT_SIGNAL_IMITATION_CONFIGS",
test_config/datasets/output_signal_imitation.py
"""
Конфигурация тестового набора для имитации выходных сигналов СОУ (обратный обмен СДКУ).
Особенности набора:
- Имитация выходных сигналов (не входных, как imitate_sensor_signal).
- Проверка изменений выполняется напрямую через OPC UA (ns=2, string NodeId).
Пример:
- ДУ (НПС-НПС): CHTN_linearParts_9
- КП-КП (сегменты ДУ): CHTN_controlledSites_188
Запуск:
- Все наборы: pytest tests/test_output_signal_imitation.py
- Один набор: pytest tests/test_output_signal_imitation.py --suites=output_signal_imitation
"""
from constants.enums import TU, OutputSignalType
from constants.test_constants import BaseTN3Constants as TestConst
from test_config.models_for_tests import (
CaseMarkers,
OutputSignalImitationCase,
OutputSignalImitationConfig,
)
# ===== Константы набора =====
SUITE_NAME = "OutputSignalImitation_tn3"
SUITE_DATA_ID = 0
ARCHIVE_NAME = f"{SUITE_NAME}.tar.gz"
# Технологический участок (ЧТН)
TECHNOLOGICAL_UNIT = TU.TIKHORETSK_NOVOROSSIYSK_3
# ===== Адреса участков для имитации (примеры из конфигурации трубы) =====
LINEAR_PART_ADDRESS = "CHTN_linearParts_9"
CONTROLLED_SITE_ADDRESS = "CHTN_controlledSites_188"
# ===== Кейсы имитации выходных сигналов =====
# ДУ (НПС-НПС): утечка, координата, объём, время, квитирование, маскирование
LINEAR_PART_CASES = [
OutputSignalImitationCase(signal_type=OutputSignalType.LEAK, address=LINEAR_PART_ADDRESS, imitate_value="true"),
OutputSignalImitationCase(
signal_type=OutputSignalType.LEAK_COORDINATE, address=LINEAR_PART_ADDRESS, imitate_value="35000"
),
OutputSignalImitationCase(
signal_type=OutputSignalType.LEAK_VOLUME, address=LINEAR_PART_ADDRESS, imitate_value="100.5"
),
OutputSignalImitationCase(
signal_type=OutputSignalType.LEAK_TIME, address=LINEAR_PART_ADDRESS, imitate_value="2026-01-01T00:00:00Z"
),
OutputSignalImitationCase(
signal_type=OutputSignalType.ACKNOWLEDGE, address=LINEAR_PART_ADDRESS, imitate_value="true"
),
OutputSignalImitationCase(signal_type=OutputSignalType.MASK, address=LINEAR_PART_ADDRESS, imitate_value="true"),
OutputSignalImitationCase(
signal_type=OutputSignalType.MASK_REASON, address=LINEAR_PART_ADDRESS, imitate_value="test_reason"
),
]
# КП-КП (сегменты ДУ): режим МТ, режим СОУ, самотеки
CONTROLLED_SITE_CASES = [
OutputSignalImitationCase(
signal_type=OutputSignalType.PUMPING_STATUS, address=CONTROLLED_SITE_ADDRESS, imitate_value="1"
),
OutputSignalImitationCase(
signal_type=OutputSignalType.LDS_STATUS, address=CONTROLLED_SITE_ADDRESS, imitate_value="4"
),
OutputSignalImitationCase(
signal_type=OutputSignalType.FREE_FLOW, address=CONTROLLED_SITE_ADDRESS, imitate_value="true"
),
]
# ===== Конфигурация набора =====
OUTPUT_SIGNAL_IMITATION_CONFIG = OutputSignalImitationConfig(
ost_name=TestConst.CHTN_OST_NAME,
suite_name=SUITE_NAME,
suite_data_id=SUITE_DATA_ID,
archive_name=ARCHIVE_NAME,
technological_unit=TECHNOLOGICAL_UNIT,
cases=LINEAR_PART_CASES + CONTROLLED_SITE_CASES,
imitate_leak_test=CaseMarkers(test_case_id="0", offset=0),
imitate_leak_coordinate_test=CaseMarkers(test_case_id="0", offset=2),
imitate_leak_volume_test=CaseMarkers(test_case_id="0", offset=4),
imitate_leak_time_test=CaseMarkers(test_case_id="0", offset=6),
imitate_acknowledge_test=CaseMarkers(test_case_id="0", offset=8),
imitate_mask_test=CaseMarkers(test_case_id="0", offset=10),
imitate_mask_reason_test=CaseMarkers(test_case_id="0", offset=12),
imitate_pumping_status_test=CaseMarkers(test_case_id="0", offset=14),
imitate_lds_status_test=CaseMarkers(test_case_id="0", offset=16),
imitate_free_flow_test=CaseMarkers(test_case_id="0", offset=18),
)
модели тестов
actual_parsed_report: Any = None
@dataclass(frozen=True)
class OutputSignalImitationCase:
"""
Кейс имитации выходного сигнала СОУ (обратный обмен СДКУ).
- signal_type: OutputSignalType - связывает signalType контракта имитации,
OPC-суффикс и тип участка (linearParts / controlledSites).
- address: тег участка в OPC (например "CHTN_linearParts_9").
- imitate_value: имитируемое значение (строкой, как imitateInfo.value в контракте).
- object_id: идентификатор объекта для запроса имитации (заглушка до готовности бэка).
"""
signal_type: OutputSignalType
address: str
imitate_value: str
object_id: Optional[int] = None
@property
def opc_suffix(self) -> str:
return self.signal_type.opc_suffix
@property
def section_type(self) -> str:
return self.signal_type.section_type
@dataclass
class OutputSignalImitationConfig(BaseSuiteConfig):
"""
Конфигурация набора данных для имитации выходных сигналов СОУ.
"""
cases: list[OutputSignalImitationCase] = field(default_factory=list)
imitate_leak_test: Optional[CaseMarkers] = None
imitate_leak_coordinate_test: Optional[CaseMarkers] = None
imitate_leak_volume_test: Optional[CaseMarkers] = None
imitate_leak_time_test: Optional[CaseMarkers] = None
imitate_acknowledge_test: Optional[CaseMarkers] = None
imitate_mask_test: Optional[CaseMarkers] = None
imitate_mask_reason_test: Optional[CaseMarkers] = None
imitate_pumping_status_test: Optional[CaseMarkers] = None
imitate_lds_status_test: Optional[CaseMarkers] = None
imitate_free_flow_test: Optional[CaseMarkers] = None
test_scenarios/output_signal_imitation_scenarios.py
"""
Сценарий имитации выходных сигналов СОУ (обратный обмен СДКУ) с проверкой через OPC UA.
Порядок выполнения (чтобы не блочить сценарий):
1. Сначала читаем текущие значения всех тегов (снапшот "до").
2. Затем выполняем все действия имитации (заглушка до готовности api-gateway).
3. Затем читаем значения всех тегов (снапшот "после").
4. В конце — все проверки (мягкие, через SoftAssertions).
"""
import logging
import os
from datetime import datetime
from typing import Optional
import allure
from clients.opc_ua_client import OpcUaClient
from constants.architecture_constants import EnvKeyConstants
from test_config.models_for_tests import OutputSignalImitationCase, OutputSignalImitationConfig
from utils.helpers.asserts import SoftAssertions, StepCheck
from utils.helpers.opc_imitation_utils import (
build_case_node_id,
build_imitate_request,
coerce_expected_value,
format_opc_snapshot,
)
logger = logging.getLogger(__name__)
async def imitate_output_signal(
opc_client: OpcUaClient, cfg: OutputSignalImitationConfig, case: OutputSignalImitationCase
) -> None:
"""
Универсальный сценарий: имитирует один выходной сигнал и проверяет изменение
через OPC UA.
"""
stand_name = os.environ.get(EnvKeyConstants.STAND_NAME) or ""
node_id = build_case_node_id(cfg, case, stand_name)
# ===== 1. Слепок значений ноды до имитации =====
with allure.step("Чтение значения выходного сигнала до имитации (OPC UA)"):
before_dv = await opc_client.read_data_value(node_id)
allure.attach(
format_opc_snapshot(node_id, before_dv),
name=f"до: {case.address}.{case.opc_suffix}",
attachment_type=allure.attachment_type.TEXT,
)
# ===== 2. Действие имитации =====
with allure.step("Имитация выходного сигнала через api-gateway (заглушка)"):
request = build_imitate_request(case, cfg.tu_id)
allure.attach(
str(request),
name=f"imitateOutputSignalRequest: {case.address}.{case.opc_suffix}",
attachment_type=allure.attachment_type.JSON,
)
# TODO: реальный вызов имитации выходного сигнала через api-gateway.
logger.info(
"TODO: имитация выходного сигнала signalType=%s value=%s",
case.signal_type.signal_type,
case.imitate_value,
)
# ===== 3. Слепок значений ноды после имитации =====
with allure.step("Чтение значения выходного сигнала после имитации (OPC UA)"):
after_dv = await opc_client.read_data_value(node_id)
allure.attach(
format_opc_snapshot(node_id, after_dv),
name=f"после: {case.address}.{case.opc_suffix}",
attachment_type=allure.attachment_type.TEXT,
)
# ===== 4. Проверки =====
with allure.step("Проверки"):
with SoftAssertions() as soft_failures:
after_value = after_dv.Value.Value if after_dv and after_dv.Value else None
expected_value = coerce_expected_value(case.imitate_value, after_value)
StepCheck(
f"Проверка имитации {case.signal_type.signal_type} ({node_id})", "value", soft_failures
).actual(after_value).expected(expected_value).equal_to()
StepCheck(
f"Проверка качества сигнала {case.signal_type.signal_type} ({node_id})", "status", soft_failures
).actual(after_dv.StatusCode.is_good() if after_dv else False).expected(True).equal_to()
before_ts: Optional[datetime] = before_dv.SourceTimestamp if before_dv else None
after_ts: Optional[datetime] = after_dv.SourceTimestamp if after_dv else None
ts_changed = after_ts is not None and (before_ts is None or after_ts >= before_ts)
StepCheck(
f"Проверка времени изменения сигнала {case.signal_type.signal_type} ({node_id})",
"SourceTimestamp",
soft_failures,
).actual(ts_changed).expected(True).equal_to()
logger.info(
"Имитация %s: до=%s, после=%s",
node_id,
before_dv.Value.Value if before_dv and before_dv.Value else None,
after_value,
)
tests/test_output_signal_imitation.py
"""
Тесты имитации выходных сигналов СОУ (обратный обмен СДКУ).
Каждый тест имитирует ровно один выходной сигнал. Offset между тестами задан
с шагом 2 минуты, чтобы имитации не пересекались.
Запуск:
- Все наборы: pytest tests/test_output_signal_imitation.py
- Один набор: pytest tests/test_output_signal_imitation.py --suites=output_signal_imitation
"""
from typing import Any, List
import allure
import pytest
from clients.opc_ua_client import OpcUaClient
from constants.enums import OutputSignalType
from test_config.datasets import ALL_OUTPUT_SIGNAL_IMITATION_CONFIGS
from test_config.models_for_tests import OutputSignalImitationConfig
from test_scenarios import output_signal_imitation_scenarios as scenarios
from utils.helpers.opc_imitation_utils import find_case_by_signal_type
def _get_suite_markers(config: OutputSignalImitationConfig) -> List[pytest.MarkDecorator]:
"""Маркеры тестового набора для группировки и setup."""
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]:
"""Один параметр на каждый набор данных имитации выходных сигналов."""
return [
pytest.param(config, id=config.suite_name, marks=_get_suite_markers(config))
for config in ALL_OUTPUT_SIGNAL_IMITATION_CONFIGS
]
@pytest.mark.parametrize("config", _generate_suite_params())
class TestOutputSignalImitation:
"""Тесты имитации выходных сигналов СОУ (по одному сигналу на тест)."""
async def _imitate(
self, opc_client: OpcUaClient, config: OutputSignalImitationConfig, signal_type: OutputSignalType
) -> None:
case = find_case_by_signal_type(config.cases, signal_type)
allure.dynamic.tag("imitateOutputSignal")
allure.dynamic.tag("OPC_UA")
allure.dynamic.title(
f"[imitateOutputSignal] Имитация {case.signal_type.signal_type} ({case.address}.{case.opc_suffix})"
)
allure.dynamic.description(
f"Проверка имитации выходного сигнала {case.signal_type.signal_type} "
f"на наборе {config.suite_name}.\n"
"Проверки выполняются напрямую через OPC UA: значение, качество и время изменения сигнала."
)
await scenarios.imitate_output_signal(opc_client, config, case)
@pytest.mark.asyncio
async def test_imitate_leak(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.LEAK)
@pytest.mark.asyncio
async def test_imitate_leak_coordinate(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.LEAK_COORDINATE)
@pytest.mark.asyncio
async def test_imitate_leak_volume(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.LEAK_VOLUME)
@pytest.mark.asyncio
async def test_imitate_leak_time(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.LEAK_TIME)
@pytest.mark.asyncio
async def test_imitate_acknowledge(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.ACKNOWLEDGE)
@pytest.mark.asyncio
async def test_imitate_mask(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.MASK)
@pytest.mark.asyncio
async def test_imitate_mask_reason(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.MASK_REASON)
@pytest.mark.asyncio
async def test_imitate_pumping_status(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.PUMPING_STATUS)
@pytest.mark.asyncio
async def test_imitate_lds_status(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.LDS_STATUS)
@pytest.mark.asyncio
async def test_imitate_free_flow(self, opc_client: OpcUaClient, config: OutputSignalImitationConfig) -> None:
await self._imitate(opc_client, config, OutputSignalType.FREE_FLOW)
utils/helpers/opc_imitation_utils.py
"""
Хелперы для сценария имитации выходных сигналов СОУ (обратный обмен СДКУ).
"""
from typing import Any
from clients.opc_ua_client import build_opc_node_id
from constants.enums import OutputSignalType
from test_config.models_for_tests import OutputSignalImitationCase, OutputSignalImitationConfig
def build_imitate_request(case: OutputSignalImitationCase, tu_id: int) -> dict[str, Any]:
"""
Формирует тело запроса imitateOutputSignalRequest по контракту (asyncapi).
TODO: заменить на реальный вызов имитации через api-gateway (WebSocket invocation),
когда бэк реализует метод imitateOutputSignal.
"""
return {
'objectId': case.object_id,
'signalType': case.signal_type.signal_type,
'tuId': tu_id,
'imitateInfo': {'value': case.imitate_value},
}
def coerce_expected_value(expected: str, actual: Any) -> Any:
"""
Приводит строковое ожидаемое значение (imitateInfo.value) к типу фактического
значения OPC-тега, чтобы корректно сравнивать bool/int/float.
"""
if actual is None:
return expected
if isinstance(actual, bool):
return expected.strip().lower() in ("true", "1")
if isinstance(actual, int):
return int(float(expected))
if isinstance(actual, float):
return float(expected)
return expected
def build_case_node_id(cfg: OutputSignalImitationConfig, case: OutputSignalImitationCase, stand_name: str) -> str:
"""Собирает NodeId для кейса имитации выходного сигнала."""
return build_opc_node_id(cfg.ost_name, stand_name, case.address, case.opc_suffix)
def format_opc_snapshot(node_id: str, data_value: Any) -> str:
"""Текстовое описание значения/статуса/времени для allure-вложений и логов."""
value = data_value.Value.Value if data_value and data_value.Value else None
status = data_value.StatusCode if data_value else None
source_ts = data_value.SourceTimestamp if data_value else None
return f"{node_id}\nValue={value}\nStatusCode={status}\nSourceTimestamp={source_ts}"
def find_case_by_signal_type(
cases: list[OutputSignalImitationCase], signal_type: OutputSignalType
) -> OutputSignalImitationCase:
"""Находит кейс имитации в наборе данных по типу сигнала."""
for case in cases:
if case.signal_type == signal_type:
return case
raise ValueError(f"Не найден кейс имитации для типа сигнала {signal_type}")
конфтест
IS_REJECTED_SUITE_LEVEL_MAPPING = {
'test_rejection_report': 'rejection_report_test',
}
# Suite-level тесты имитации выходных сигналов (маркеры из OutputSignalImitationConfig - параметр config)
OUTPUT_SIGNAL_IMITATION_SUITE_LEVEL_MAPPING = {
'test_imitate_leak': 'imitate_leak_test',
'test_imitate_leak_coordinate': 'imitate_leak_coordinate_test',
'test_imitate_leak_volume': 'imitate_leak_volume_test',
'test_imitate_leak_time': 'imitate_leak_time_test',
'test_imitate_acknowledge': 'imitate_acknowledge_test',
'test_imitate_mask': 'imitate_mask_test',
'test_imitate_mask_reason': 'imitate_mask_reason_test',
'test_imitate_pumping_status': 'imitate_pumping_status_test',
'test_imitate_lds_status': 'imitate_lds_status_test',
'test_imitate_free_flow': 'imitate_free_flow_test',
}
# Мержим все вместе чтобы не переписывать логику коллектора айтемов (тестов)
SUITE_LEVEL_TEST_MAPPING = {
**SMOKE_SUITE_LEVEL_MAPPING,
**LDS_STATUS_SUITE_LEVEL_MAPPING,
**STATIONARY_STATUS_SUITE_LEVEL_MAPPING,
**IS_REJECTED_SUITE_LEVEL_MAPPING,
**OUTPUT_SIGNAL_IMITATION_SUITE_LEVEL_MAPPING,
}
attr_name = STATIONARY_STATUS_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
if test_name in OUTPUT_SIGNAL_IMITATION_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = OUTPUT_SIGNAL_IMITATION_SUITE_LEVEL_MAPPING[test_name]
cfg["stand_manager"] = stand_manager
try:
stand_manager.check_opc_server_availability()
@pytest_asyncio.fixture
async def opc_client(request):
"""
Фикстура OPC UA клиента для чтения выходных сигналов СОУ.
Подключается к серверу, указанному в OPC_URL (opc.tcp://host:port).
Ошибка подключения помечает тест как Failed (красный), а не Error (жёлтый).
"""
from clients.opc_ua_client import OpcUaClient
opc_url = os.environ.get(EnvKeyConstants.OPC_URL)
if not opc_url:
pytest.fail("OPC_URL не задан в переменных окружения")
client = OpcUaClient(opc_url)
try:
await client.connect()
except Exception as error:
pytest.fail(f"Не удалось подключиться к OPC UA серверу ({opc_url}): {error}")
try:
yield client
finally:
try:
await client.disconnect()
except Exception as error:
logger.warning("[OPC] [WARNING] Ошибка при отключении OPC UA клиента: %s", error)
def pytest_sessionfinish(session, exitstatus):
