91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
|
"""Unit-Tests IPC-Protokoll (PLAN.md §6.2, ADR-0003)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from hms_protocol import (
|
||
|
|
Envelope,
|
||
|
|
IdempotencyRegistry,
|
||
|
|
MessageType,
|
||
|
|
decode_frame,
|
||
|
|
encode_frame,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_frame_roundtrip() -> None:
|
||
|
|
payload = {
|
||
|
|
"protocol_version": 1,
|
||
|
|
"message_id": "abc",
|
||
|
|
"type": "command",
|
||
|
|
"revision": 5,
|
||
|
|
"monotonic_timestamp_ns": 123,
|
||
|
|
"payload": {"key": "value", "nested": [1, 2, 3]},
|
||
|
|
}
|
||
|
|
frame = encode_frame(payload)
|
||
|
|
assert decode_frame(frame) == payload
|
||
|
|
|
||
|
|
|
||
|
|
def test_frame_length_prefix_is_4_byte_big_endian() -> None:
|
||
|
|
frame = encode_frame({"a": 1})
|
||
|
|
assert frame[:4] == len(frame[4:]).to_bytes(4, "big")
|
||
|
|
|
||
|
|
|
||
|
|
def test_oversized_payload_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
|
|
import hms_protocol.framing as framing
|
||
|
|
|
||
|
|
monkeypatch.setattr(framing, "MAX_PAYLOAD_SIZE", 16)
|
||
|
|
with pytest.raises(ValueError, match="too large"):
|
||
|
|
framing.encode_frame({"data": "x" * 64})
|
||
|
|
|
||
|
|
|
||
|
|
def test_truncated_frame_rejected() -> None:
|
||
|
|
frame = encode_frame({"a": 1})
|
||
|
|
with pytest.raises(ValueError, match="truncated"):
|
||
|
|
decode_frame(frame[:-2])
|
||
|
|
|
||
|
|
|
||
|
|
def test_declared_length_over_limit_rejected() -> None:
|
||
|
|
import struct
|
||
|
|
|
||
|
|
evil = struct.pack(">I", 2**31) + b"x" * 8
|
||
|
|
with pytest.raises(ValueError, match="exceeds limit"):
|
||
|
|
decode_frame(evil)
|
||
|
|
|
||
|
|
|
||
|
|
def test_envelope_defaults_and_validation() -> None:
|
||
|
|
env = Envelope(type=MessageType.COMMAND)
|
||
|
|
assert env.protocol_version == 1
|
||
|
|
assert env.revision == 0
|
||
|
|
assert env.message_id
|
||
|
|
assert env.payload == {}
|
||
|
|
|
||
|
|
|
||
|
|
def test_envelope_rejects_wrong_protocol_version() -> None:
|
||
|
|
with pytest.raises(ValueError, match="protocol_version"):
|
||
|
|
Envelope(type=MessageType.EVENT, protocol_version=2)
|
||
|
|
|
||
|
|
|
||
|
|
def test_idempotency_register_and_duplicate() -> None:
|
||
|
|
reg = IdempotencyRegistry()
|
||
|
|
assert reg.register("cmd-1") is True
|
||
|
|
assert reg.register("cmd-1") is False
|
||
|
|
assert reg.register("cmd-2") is True
|
||
|
|
assert len(reg) == 2
|
||
|
|
|
||
|
|
|
||
|
|
def test_idempotency_complete_returns_same_result() -> None:
|
||
|
|
reg = IdempotencyRegistry()
|
||
|
|
reg.register("cmd-1")
|
||
|
|
reg.complete("cmd-1", {"status": "ack"})
|
||
|
|
assert reg.result("cmd-1") == {"status": "ack"}
|
||
|
|
|
||
|
|
|
||
|
|
def test_idempotency_capacity_eviction_lru() -> None:
|
||
|
|
reg = IdempotencyRegistry(capacity=2)
|
||
|
|
reg.register("a")
|
||
|
|
reg.register("b")
|
||
|
|
reg.register("a") # a wird jüngst benutzt
|
||
|
|
reg.register("c") # verdrängt b (LRU)
|
||
|
|
assert reg.register("b") is True # b wurde verdrängt → neu
|
||
|
|
assert reg.register("c") is False # c existiert noch
|