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).
114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
"""Integrationstests Art-Net-Receiver über echten Loopback-UDP-Socket (§16.1).
|
|
|
|
Kein Mock: echter Datagramm-Versand auf 127.0.0.1. Async-Tests laufen
|
|
über asyncio.run(), damit keine pytest-asyncio-Versionsbindung entsteht.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from hms_artnet import ArtNetReceiver, build_dmx, build_poll, parse_poll_reply
|
|
|
|
|
|
def test_receiver_emits_updates_and_telemetry() -> None:
|
|
asyncio.run(_recv_updates_impl())
|
|
|
|
|
|
async def _recv_updates_impl() -> None:
|
|
receiver = ArtNetReceiver(universes={0}, bind_host="127.0.0.1", port=0)
|
|
await receiver.start()
|
|
sock = receiver._transport.get_extra_info("socket")
|
|
actual_port = sock.getsockname()[1]
|
|
|
|
received: list = []
|
|
receiver.on_dmx(received.append)
|
|
|
|
receiver._transport.sendto(
|
|
build_dmx(0, b"\x11\x22\x33\x44"), ("127.0.0.1", actual_port)
|
|
)
|
|
await asyncio.sleep(0.2)
|
|
|
|
assert len(received) >= 1
|
|
assert received[-1].universe == 0
|
|
assert received[-1].data == b"\x11\x22\x33\x44"
|
|
tel = receiver.telemetry()
|
|
assert tel[0].packets >= 1
|
|
assert tel[0].last_sender_ip == "127.0.0.1"
|
|
|
|
# nicht abonniertes Universe ignorieren
|
|
n_before = tel[0].packets
|
|
receiver._transport.sendto(
|
|
build_dmx(9, b"\x00\x00"), ("127.0.0.1", actual_port)
|
|
)
|
|
await asyncio.sleep(0.15)
|
|
assert receiver.telemetry()[0].packets == n_before
|
|
|
|
await receiver.stop()
|
|
|
|
|
|
def test_receiver_answers_poll_as_media_server() -> None:
|
|
asyncio.run(_poll_reply_impl())
|
|
|
|
|
|
async def _poll_reply_impl() -> None:
|
|
receiver = ArtNetReceiver(
|
|
universes={0},
|
|
bind_host="127.0.0.1",
|
|
port=0,
|
|
short_name="HMS Test",
|
|
long_name="HMS MediaEngine Test Node",
|
|
)
|
|
await receiver.start()
|
|
sock = receiver._transport.get_extra_info("socket")
|
|
actual_port = sock.getsockname()[1]
|
|
|
|
reply_ready = asyncio.Event()
|
|
replies: list = []
|
|
|
|
class _Client(asyncio.DatagramProtocol):
|
|
def datagram_received(self, data, addr) -> None:
|
|
replies.append(data)
|
|
reply_ready.set()
|
|
|
|
loop = asyncio.get_running_loop()
|
|
client_transport, _ = await loop.create_datagram_endpoint(
|
|
_Client, local_addr=("127.0.0.1", 0)
|
|
)
|
|
client_transport.sendto(build_poll(), ("127.0.0.1", actual_port))
|
|
await asyncio.wait_for(reply_ready.wait(), timeout=2.0)
|
|
|
|
info = parse_poll_reply(replies[0])
|
|
assert info is not None
|
|
assert info.style == 0x02 # StMedia (§3.4)
|
|
assert info.short_name == "HMS Test"
|
|
|
|
client_transport.close()
|
|
await receiver.stop()
|
|
|
|
|
|
def test_receiver_allowlist_blocks_foreign_sender() -> None:
|
|
asyncio.run(_allowlist_impl())
|
|
|
|
|
|
async def _allowlist_impl() -> None:
|
|
receiver = ArtNetReceiver(
|
|
universes={0},
|
|
bind_host="127.0.0.1",
|
|
port=0,
|
|
sender_allowlist={"192.168.50.10"}, # nur dieser Sender erlaubt
|
|
)
|
|
await receiver.start()
|
|
sock = receiver._transport.get_extra_info("socket")
|
|
actual_port = sock.getsockname()[1]
|
|
|
|
received: list = []
|
|
receiver.on_dmx(received.append)
|
|
|
|
receiver._transport.sendto(
|
|
build_dmx(0, b"\x00\x00"), ("127.0.0.1", actual_port)
|
|
)
|
|
await asyncio.sleep(0.2)
|
|
assert received == [] # Absender 127.0.0.1 steht nicht in der Allowlist
|
|
await receiver.stop()
|