31 lines
1.0 KiB
Python
31 lines
1.0 KiB
Python
"""Tests for generated protobuf and gRPC module importability."""
|
|
|
|
from mxgateway.generated import mxaccess_gateway_pb2
|
|
from mxgateway.generated import mxaccess_gateway_pb2_grpc
|
|
from mxgateway.generated import mxaccess_worker_pb2
|
|
|
|
|
|
def test_gateway_messages_import() -> None:
|
|
request = mxaccess_gateway_pb2.OpenSessionRequest(
|
|
client_session_name="pytest",
|
|
client_correlation_id="test-correlation",
|
|
)
|
|
|
|
assert request.client_session_name == "pytest"
|
|
assert hasattr(mxaccess_gateway_pb2_grpc, "MxAccessGatewayStub")
|
|
|
|
|
|
def test_worker_messages_import_gateway_types() -> None:
|
|
envelope = mxaccess_worker_pb2.WorkerEnvelope(
|
|
protocol_version=1,
|
|
session_id="test-session",
|
|
worker_command=mxaccess_worker_pb2.WorkerCommand(
|
|
command=mxaccess_gateway_pb2.MxCommand(
|
|
kind=mxaccess_gateway_pb2.MX_COMMAND_KIND_PING,
|
|
ping=mxaccess_gateway_pb2.PingCommand(message="hello"),
|
|
),
|
|
),
|
|
)
|
|
|
|
assert envelope.worker_command.command.ping.message == "hello"
|