37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc
|
|
|
|
|
|
def test_read_tag_request_roundtrip():
|
|
req = opc_gateway_pb2.ReadTagRequest(server_id='s1', addr='ns=2;s=Tag1', data_type='Float')
|
|
|
|
data = req.SerializeToString()
|
|
parsed = opc_gateway_pb2.ReadTagRequest()
|
|
parsed.ParseFromString(data)
|
|
|
|
assert parsed.server_id == 's1'
|
|
assert parsed.addr == 'ns=2;s=Tag1'
|
|
assert parsed.data_type == 'Float'
|
|
|
|
|
|
def test_read_tag_response_roundtrip():
|
|
resp = opc_gateway_pb2.ReadTagResponse(
|
|
value='42.0',
|
|
opc_status_code='Good',
|
|
source_timestamp='2026-07-14T00:00:00Z',
|
|
server_timestamp='2026-07-14T00:00:01Z',
|
|
persisted=True,
|
|
)
|
|
|
|
data = resp.SerializeToString()
|
|
parsed = opc_gateway_pb2.ReadTagResponse()
|
|
parsed.ParseFromString(data)
|
|
|
|
assert parsed.value == '42.0'
|
|
assert parsed.opc_status_code == 'Good'
|
|
assert parsed.persisted is True
|
|
|
|
|
|
def test_grpc_stub_module_loads():
|
|
assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayStub')
|
|
assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayServicer')
|