HMS MediaEngine Agent
362e089be0
Der Nutzer hat recht: Der Ordner war voller Entwicklungs-Muell. Jetzt ist sauber getrennt: ROOT (was der Nutzer sieht und braucht): - run.py = das Programm - hms_app/ = der Anwendungscode - HMS MediaEngine.app = macOS Doppelklick-Starter - HMS-Start.vbs = Windows Doppelklick-Starter - HMS-Install.vbs = Windows Erst-Installation - HMS-Mac-Install.command = macOS Homebrew-Installation - HMS-Portable-Install.command = macOS Portable-Installation (16GB-Fix) - installer_gui.py = grafischer Installer - launcher.pyw + launcher_core.py = interne Start-Logik - LIESMICH.txt = 10-Zeilen-Kurzanleitung - .gitignore _entwicklung/ (alles andere, NICHT benoetigt): - packages/ apps/ native/ plugins/ tools/ schemas/ tests/ docs/ build/ fixture_profiles/ - PLAN.md STATUS.md ERRORS.md TEST_REPORT.md CHANGELOG.md README.md - pyproject.toml uv.lock setup_*.sh/ps1 make_mac_app.py Diese Trennung gilt ab sofort fuer alle Commits. Der Nutzer kann _entwicklung/ loeschen wenn er Platz braucht - die App laeuft ohne. Verifiziert: App startet nach Aufraeumen unveraendert (Health 200).
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
|