Joseph Doherty
d6b2f24c3f
One cross-client conformance pass; also closes first-cycle CLI-08. CLI-37: an MxStatusProxy entry is a failure iff `category != MX_STATUS_CATEGORY_OK`. The proto contract has always said so — `success` is the raw 16-bit COM member carried verbatim for diagnostics, not a boolean — but four clients branched on `success` alone and .NET required both, so the same gateway reply produced opposite verdicts per language. An absent entry stays success; a present entry with an UNSPECIFIED category is a failure, because the worker always maps a category and an unmapped one is not proven OK. CLI-38: a reply fails on HRESULT iff `hresult` is present and negative, so positive COM success codes such as S_FALSE (1) pass. .NET/Go/Java used `!= 0`, which errored on a parity-preserving S_FALSE that Python and Rust accepted. This makes the existing ClientLibrariesDesign.md claim true rather than rewriting the doc to describe the divergence. Four shared fixtures pin both rules cross-client, and each language suite also carries a table test for the two edges a fixture cannot express (absent entry, UNSPECIFIED category). A Java test fake that built a status with a bare `setSuccess(1)` and no category is fixed — under the category rule that reply was never a success.
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""Tests for typed command error mapping."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from google.protobuf.json_format import ParseDict
|
|
|
|
from zb_mom_ww_mxgateway.errors import ensure_mxaccess_success, ensure_protocol_success
|
|
from zb_mom_ww_mxgateway import MxAccessError, MxGatewaySessionError
|
|
from zb_mom_ww_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
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("fixture", "expect_failure"),
|
|
[
|
|
("command-replies/register.ok.reply.json", False),
|
|
("command-replies/write.status-category-error-success-set.reply.json", True),
|
|
("command-replies/write.status-category-ok-success-zero.reply.json", False),
|
|
("command-replies/write.hresult-s-false.reply.json", False),
|
|
("command-replies/write.hresult-e-fail.reply.json", True),
|
|
],
|
|
)
|
|
def test_reply_validation_fixtures_branch_on_category_and_negative_hresult(
|
|
fixture: str,
|
|
expect_failure: bool,
|
|
) -> None:
|
|
reply = _load_reply(fixture)
|
|
|
|
if expect_failure:
|
|
with pytest.raises(MxAccessError):
|
|
ensure_mxaccess_success("write", reply)
|
|
else:
|
|
assert ensure_mxaccess_success("write", reply) is reply
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("category", "success", "expect_failure"),
|
|
[
|
|
(pb.MX_STATUS_CATEGORY_OK, 0, False),
|
|
(pb.MX_STATUS_CATEGORY_OK, 1, False),
|
|
(pb.MX_STATUS_CATEGORY_COMMUNICATION_ERROR, 1, True),
|
|
(pb.MX_STATUS_CATEGORY_UNSPECIFIED, 1, True),
|
|
],
|
|
)
|
|
def test_status_entry_verdict_ignores_the_raw_success_member(
|
|
category: int,
|
|
success: int,
|
|
expect_failure: bool,
|
|
) -> None:
|
|
reply = pb.MxCommandReply(
|
|
protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
|
|
statuses=[pb.MxStatusProxy(success=success, category=category)],
|
|
)
|
|
|
|
if expect_failure:
|
|
with pytest.raises(MxAccessError):
|
|
ensure_mxaccess_success("write", reply)
|
|
else:
|
|
assert ensure_mxaccess_success("write", reply) is reply
|
|
|
|
|
|
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())
|