Joseph Doherty
50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
"""Tests for typed command error mapping."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from google.protobuf.json_format import ParseDict
|
|
|
|
from mxgateway.errors import ensure_mxaccess_success, ensure_protocol_success
|
|
from mxgateway import MxAccessError, MxGatewaySessionError
|
|
from mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
FIXTURE_ROOT = Path(__file__).resolve().parents[2] / "proto" / "fixtures" / "behavior"
|
|
|
|
|
|
def test_register_fixture_is_protocol_and_mxaccess_success() -> None:
|
|
reply = _load_reply("command-replies/register.ok.reply.json")
|
|
|
|
assert ensure_protocol_success("register", reply.protocol_status, reply) is reply
|
|
assert ensure_mxaccess_success("register", reply) is reply
|
|
|
|
|
|
def test_write_failure_fixture_preserves_raw_reply() -> None:
|
|
reply = _load_reply("command-replies/write.mxaccess-failure.reply.json")
|
|
|
|
assert ensure_protocol_success("write", reply.protocol_status, reply) is reply
|
|
with pytest.raises(MxAccessError) as captured:
|
|
ensure_mxaccess_success("write", reply)
|
|
|
|
assert captured.value.raw_reply is reply
|
|
assert captured.value.raw_reply.hresult == -2147220992
|
|
assert len(captured.value.raw_reply.statuses) == 2
|
|
|
|
|
|
def test_session_status_maps_to_session_error() -> None:
|
|
status = pb.ProtocolStatus(
|
|
code=pb.PROTOCOL_STATUS_CODE_SESSION_NOT_FOUND,
|
|
message="session missing",
|
|
)
|
|
|
|
with pytest.raises(MxGatewaySessionError) as captured:
|
|
ensure_protocol_success("invoke", status)
|
|
|
|
assert captured.value.protocol_status is status
|
|
|
|
|
|
def _load_reply(name: str) -> pb.MxCommandReply:
|
|
payload = json.loads((FIXTURE_ROOT / name).read_text(encoding="utf-8"))
|
|
return ParseDict(payload, pb.MxCommandReply())
|