matthias.lotz
fcca3272ae
Device payload (e.g. lasercutter/session):
{session_id, session_minutes, session_seconds, session_start_time, freetime_s, ip}
- parsers/direct_session_parser.py: new parser returning message_type='session_complete'
- parsers/registry.py: register 'direct_session' with freetime_s parameter (default 0)
- mqtt_device.py: add ('direct_session', 'Direct Session (Lasercutter)') to _PARSER_SELECTION
- tests/unit/test_direct_session_parser.py: 24 tests (happy-path, missing fields, defaults, registry)
187 lines
8.2 KiB
Python
187 lines
8.2 KiB
Python
"""Unit tests for DirectSessionParser."""
|
|
|
|
import pytest
|
|
|
|
from parsers.direct_session_parser import DirectSessionParser
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Fixtures
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def parser() -> DirectSessionParser:
|
|
return DirectSessionParser()
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_payload() -> dict:
|
|
return {
|
|
"session_id": 0,
|
|
"session_minutes": 1,
|
|
"session_seconds": 34,
|
|
"session_start_time": "2026-03-11T14:12:51",
|
|
"freetime_s": 10,
|
|
"ip": "192.168.2.65",
|
|
}
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Happy-Path
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_parse_message_returns_dict_for_valid_payload(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result is not None
|
|
assert isinstance(result, dict)
|
|
|
|
|
|
def test_parse_message_type_is_session_complete(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result["message_type"] == "session_complete"
|
|
|
|
|
|
def test_parse_message_device_id_from_topic(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result["device_id"] == "lasercutter"
|
|
|
|
|
|
def test_parse_message_session_minutes(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result["session_minutes"] == 1
|
|
|
|
|
|
def test_parse_message_session_seconds(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result["session_seconds"] == 34
|
|
|
|
|
|
def test_parse_message_session_id(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result["session_id"] == 0
|
|
|
|
|
|
def test_parse_message_session_start_time(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result["session_start_time"] == "2026-03-11T14:12:51"
|
|
|
|
|
|
def test_parse_message_freetime_s(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert result["freetime_s"] == 10
|
|
|
|
|
|
def test_parse_message_ip_not_in_result(parser, valid_payload) -> None:
|
|
"""ip wird aus dem Payload verworfen."""
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert "ip" not in result
|
|
|
|
|
|
def test_parse_message_timestamp_present(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert "timestamp" in result
|
|
assert result["timestamp"] is not None
|
|
|
|
|
|
def test_parse_message_session_seconds_is_int(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert isinstance(result["session_seconds"], int)
|
|
|
|
|
|
def test_parse_message_session_minutes_is_int(parser, valid_payload) -> None:
|
|
result = parser.parse_message("lasercutter/session", valid_payload)
|
|
assert isinstance(result["session_minutes"], int)
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Fehlende Pflichtfelder
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_parse_message_missing_session_minutes_returns_none(parser) -> None:
|
|
payload = {"session_seconds": 34}
|
|
result = parser.parse_message("lasercutter/session", payload)
|
|
assert result is None
|
|
|
|
|
|
def test_parse_message_missing_session_seconds_returns_none(parser) -> None:
|
|
payload = {"session_minutes": 1}
|
|
result = parser.parse_message("lasercutter/session", payload)
|
|
assert result is None
|
|
|
|
|
|
def test_parse_message_empty_payload_returns_none(parser) -> None:
|
|
result = parser.parse_message("lasercutter/session", {})
|
|
assert result is None
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Optionale Felder / Defaults
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_parse_message_freetime_s_defaults_to_zero_when_absent(parser) -> None:
|
|
payload = {"session_minutes": 2, "session_seconds": 90}
|
|
result = parser.parse_message("lasercutter/session", payload)
|
|
assert result is not None
|
|
assert result["freetime_s"] == 0
|
|
|
|
|
|
def test_parse_message_session_id_none_when_absent(parser) -> None:
|
|
payload = {"session_minutes": 2, "session_seconds": 90}
|
|
result = parser.parse_message("lasercutter/session", payload)
|
|
assert result is not None
|
|
assert result["session_id"] is None
|
|
|
|
|
|
def test_parse_message_session_start_time_none_when_absent(parser) -> None:
|
|
payload = {"session_minutes": 2, "session_seconds": 90}
|
|
result = parser.parse_message("lasercutter/session", payload)
|
|
assert result is not None
|
|
assert result["session_start_time"] is None
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Multi-Level Topic
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_parse_message_device_id_from_multi_level_topic(parser, valid_payload) -> None:
|
|
result = parser.parse_message("workshop/lasercutter/session", valid_payload)
|
|
assert result["device_id"] == "workshop"
|
|
|
|
|
|
def test_parse_message_empty_topic_device_id_unknown(parser, valid_payload) -> None:
|
|
result = parser.parse_message("", valid_payload)
|
|
assert result["device_id"] == "unknown"
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Registry-Integration
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_direct_session_in_registry() -> None:
|
|
from parsers.registry import PARSER_REGISTRY
|
|
assert "direct_session" in PARSER_REGISTRY
|
|
|
|
|
|
def test_direct_session_registry_class_is_direct_session_parser() -> None:
|
|
from parsers.registry import PARSER_REGISTRY
|
|
assert PARSER_REGISTRY["direct_session"]["class"] is DirectSessionParser
|
|
|
|
|
|
def test_direct_session_registry_has_freetime_s_parameter() -> None:
|
|
from parsers.registry import get_schema
|
|
schema = get_schema()
|
|
params = {p["name"]: p for p in schema["direct_session"]["parameters"]}
|
|
assert "freetime_s" in params
|
|
assert params["freetime_s"]["default"] == 0
|
|
|
|
|
|
def test_direct_session_in_list_parser_types() -> None:
|
|
from parsers.registry import list_parser_types
|
|
assert "direct_session" in list_parser_types()
|