logger.info("[OPC] Подключено к %s", opc_url)
return await client.get_node(node_id).read_data_value(raise_on_bad_status=False)
self._client = None
async def read_data_value(self, node_id: str, raise_on_bad_status: bool = False) -> ua.DataValue:
"""
Читает DataValue (Value + StatusCode + SourceTimestamp) тега.
По умолчанию не бросает исключение при плохом статусе (например BadNoCommunication),
а возвращает DataValue со статусом — статус проверяется в сценарии.
"""
if self._client is None:
raise RuntimeError("OPC UA клиент не подключён")
return await self._client.get_node(node_id).read_data_value(raise_on_bad_status=raise_on_bad_status)
async def read_value(self, node_id: str) -> Any:
сцен
before_dv = await opc_client.read_data_value(node_id)
if not before_dv.StatusCode.is_good():
pytest.fail(
f"Нет данных по тегу {case.address}.{case.opc_suffix} (NodeId: {node_id}). "
f"Статус: {before_dv.StatusCode.name} ({before_dv.StatusCode.value})"
)
after_dv = await opc_client.read_data_value(node_id)
if not after_dv.StatusCode.is_good():
pytest.fail(
f"Нет данных по тегу {case.address}.{case.opc_suffix} (NodeId: {node_id}) после имитации. "
f"Статус: {after_dv.StatusCode.name} ({after_dv.StatusCode.value})"
)