KX-Bridge-Release/tests/test_kxgauge.py

"""KXGauge display integration (https://gitea.it-drui.de/viewit/kxgauge).

KXGauge is a small ESP32 round-face display with a GET-only HTTP API and no
push/websocket - the bridge has to actively poke it on printer state/temp
changes. Covers: the KXGaugeClient dedupe logic, settings GET/POST roundtrip
for the new fields, the mapping config loader, and the /api/kxgauge/test
connectivity check endpoint.
"""
import argparse
import tempfile
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import pytest_asyncio
from aiohttp.test_utils import TestClient, TestServer

from kobrax_moonraker_bridge import KobraXBridge, build_app
from kxgauge_client import KXGaugeClient


# ── KXGaugeClient (sync HTTP client, dedupe logic) ──────────────────────────

def test_set_emotion_sends_get_and_dedupes():
    kg = KXGaugeClient("http://192.168.1.99")
    with patch("requests.get") as mock_get:
        mock_get.return_value = MagicMock(raise_for_status=lambda: None)
        kg.set_emotion("happy")
        kg.set_emotion("happy")  # same emotion again - must not re-send
    assert mock_get.call_count == 1
    assert mock_get.call_args[0][0] == "http://192.168.1.99/emotion/happy"


def test_set_emotion_sends_again_after_change():
    kg = KXGaugeClient("http://192.168.1.99")
    with patch("requests.get") as mock_get:
        mock_get.return_value = MagicMock(raise_for_status=lambda: None)
        kg.set_emotion("happy")
        kg.set_emotion("scared")
    assert mock_get.call_count == 2
    assert mock_get.call_args[0][0] == "http://192.168.1.99/emotion/scared"


def test_set_heat_celsius_sends_peak_once_then_dedupes_within_tolerance():
    kg = KXGaugeClient("http://192.168.1.99", heat_peak=230)
    with patch("requests.get") as mock_get:
        mock_get.return_value = MagicMock(raise_for_status=lambda: None)
        kg.set_heat_celsius(200.0)
        kg.set_heat_celsius(200.5)  # within tolerance - no new /heat/celsius call
    urls = [c.args[0] for c in mock_get.call_args_list]
    assert "http://192.168.1.99/heat/peak/celsius/230" in urls
    assert urls.count("http://192.168.1.99/heat/celsius/200.0") == 1
    assert not any("/heat/celsius/200.5" in u for u in urls)


def test_set_heat_celsius_sends_again_beyond_tolerance():
    kg = KXGaugeClient("http://192.168.1.99")
    with patch("requests.get") as mock_get:
        mock_get.return_value = MagicMock(raise_for_status=lambda: None)
        kg.set_heat_celsius(200.0)
        kg.set_heat_celsius(210.0)
    heat_calls = [c.args[0] for c in mock_get.call_args_list if "/heat/celsius/" in c.args[0]]
    assert len(heat_calls) == 2


def test_client_swallows_connection_errors():
    kg = KXGaugeClient("http://192.168.1.99")
    with patch("requests.get", side_effect=OSError("unreachable")):
        kg.set_emotion("happy")  # must not raise
    assert kg._last_emotion is None  # failed send - state not updated


def test_health_check_true_on_200():
    kg = KXGaugeClient("http://192.168.1.99")
    with patch("requests.get") as mock_get:
        mock_get.return_value = MagicMock(raise_for_status=lambda: None)
        assert kg.health_check() is True


def test_health_check_false_on_error():
    kg = KXGaugeClient("http://192.168.1.99")
    with patch("requests.get", side_effect=OSError("unreachable")):
        assert kg.health_check() is False


# ── Bridge integration: settings GET/POST, test endpoint ───────────────────

def _make_bridge(pid="1", **arg_overrides):
    c = MagicMock()
    c.callbacks = {}
    c.connected = False
    args = argparse.Namespace(
        printer_ip="192.168.1.50", mqtt_port=9883, username="", password="",
        mode_id="20030", device_id="", host="127.0.0.1", port=7125,
        data_dir=tempfile.mkdtemp(prefix="kxgauge-"),
        kxgauge_url="", kxgauge_enabled=0, kxgauge_heat_peak=250,
    )
    for k, v in arg_overrides.items():
        setattr(args, k, v)
    all_bridges = {}
    bridge = KobraXBridge(c, args=args, printer_id=pid, all_bridges=all_bridges)
    all_bridges[pid] = bridge
    return bridge


@pytest_asyncio.fixture
async def kxgauge_client():
    bridge = _make_bridge(kxgauge_url="http://192.168.1.77", kxgauge_enabled=1, kxgauge_heat_peak=230)
    # _find_config_path() otherwise resolves to the real repo config/config.ini
    # (env_loader.find_config_path() walks from the script location, ignoring
    # args.data_dir) - a settings POST in a test must never touch that file.
    import pathlib
    sandbox_cfg = pathlib.Path(tempfile.mkdtemp(prefix="kxgauge-cfg-")) / "config.ini"
    bridge._find_config_path = lambda: sandbox_cfg
    bridge._restart_bridge = lambda: None  # a settings POST schedules a restart - don't kill the test process
    app = build_app(bridge)
    async with TestClient(TestServer(app)) as c:
        yield c, bridge


def _fake_get_response(status=200, json_body=None):
    resp = MagicMock()
    resp.status = status
    resp.json = AsyncMock(return_value=json_body or {})
    ctx = MagicMock()
    ctx.__aenter__ = AsyncMock(return_value=resp)
    ctx.__aexit__ = AsyncMock(return_value=False)
    return ctx


@pytest.mark.asyncio
async def test_bridge_creates_kxgauge_client_when_enabled():
    bridge = _make_bridge(kxgauge_url="http://192.168.1.77", kxgauge_enabled=1)
    assert bridge._kxgauge is not None
    assert bridge._kxgauge.base_url == "http://192.168.1.77"


@pytest.mark.asyncio
async def test_bridge_no_kxgauge_client_when_disabled():
    bridge = _make_bridge(kxgauge_url="http://192.168.1.77", kxgauge_enabled=0)
    assert bridge._kxgauge is None


@pytest.mark.asyncio
async def test_bridge_no_kxgauge_client_when_no_url():
    bridge = _make_bridge(kxgauge_enabled=1, kxgauge_url="")
    assert bridge._kxgauge is None


@pytest.mark.asyncio
async def test_settings_roundtrip_persists_kxgauge_fields(kxgauge_client):
    c, bridge = kxgauge_client
    resp = await c.get("/api/settings")
    data = await resp.json()
    assert data["kxgauge_url"] == "http://192.168.1.77"
    assert data["kxgauge_enabled"] == 1
    assert data["kxgauge_heat_peak"] == 230
    assert isinstance(data["kxgauge_mapping"], dict)
    assert data["kxgauge_mapping"]["printing"] == "happy"


@pytest.mark.asyncio
async def test_kxgauge_mapping_default_when_no_config_section():
    bridge = _make_bridge()
    assert bridge._kxgauge_mapping["free"] == "neutral"
    assert bridge._kxgauge_mapping["error"] == "scared"


@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_success(kxgauge_client):
    c, bridge = kxgauge_client
    with patch("aiohttp.ClientSession.get", return_value=_fake_get_response(200, {"emotion": "HAPPY"})) as mock_get:
        resp = await c.post("/api/kxgauge/test", json={})
        data = await resp.json()
    assert resp.status == 200
    assert data["result"] == "ok"
    assert data["emotion"] == "HAPPY"
    assert mock_get.call_args[0][0] == "http://192.168.1.77/status"


@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_uses_url_from_body_over_configured():
    bridge = _make_bridge()  # no configured kxgauge_url
    app = build_app(bridge)
    async with TestClient(TestServer(app)) as c:
        with patch("aiohttp.ClientSession.get", return_value=_fake_get_response(200, {"emotion": "NEUTRAL"})) as mock_get:
            resp = await c.post("/api/kxgauge/test", json={"url": "http://192.168.1.88"})
            data = await resp.json()
        assert resp.status == 200
        assert mock_get.call_args[0][0] == "http://192.168.1.88/status"


@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_no_url_configured_error():
    bridge = _make_bridge()
    app = build_app(bridge)
    async with TestClient(TestServer(app)) as c:
        resp = await c.post("/api/kxgauge/test", json={})
        assert resp.status == 400


@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_unreachable_returns_502(kxgauge_client):
    c, bridge = kxgauge_client
    with patch("aiohttp.ClientSession.get", side_effect=OSError("connection refused")):
        resp = await c.post("/api/kxgauge/test", json={})
    assert resp.status == 502


@pytest.mark.asyncio
async def test_settings_post_saves_custom_mapping(kxgauge_client):
    c, bridge = kxgauge_client
    resp = await c.post("/api/settings", json={
        "printer_ip": "192.168.1.50", "mqtt_port": 9883,
        "kxgauge_url": "http://192.168.1.77", "kxgauge_enabled": 1,
        "kxgauge_heat_peak": 240,
        "kxgauge_mapping": {"printing": "love", "error": "furious"},
    })
    assert resp.status == 200
    assert bridge._kxgauge_mapping["printing"] == "love"
    assert bridge._kxgauge_mapping["error"] == "furious"


@pytest.mark.asyncio
async def test_settings_post_rejects_invalid_emotion(kxgauge_client):
    c, bridge = kxgauge_client
    before = dict(bridge._kxgauge_mapping)
    resp = await c.post("/api/settings", json={
        "printer_ip": "192.168.1.50", "mqtt_port": 9883,
        "kxgauge_mapping": {"printing": "not_a_real_emotion"},
    })
    assert resp.status == 200
    # invalid value must be rejected, mapping keeps its previous value
    assert bridge._kxgauge_mapping["printing"] == before["printing"]


# ── MQTT callback hooks fire the mapped emotion / heat value ───────────────

@pytest.mark.asyncio
async def test_on_temp_pushes_heat_to_kxgauge(kxgauge_client):
    c, bridge = kxgauge_client
    bridge._kxgauge.set_heat_celsius = MagicMock()
    bridge._on_temp({"data": {"curr_nozzle_temp": 205, "target_nozzle_temp": 210,
                               "curr_hotbed_temp": 60, "target_hotbed_temp": 60}})
    bridge._kxgauge.set_heat_celsius.assert_called_once_with(205.0)


@pytest.mark.asyncio
async def test_on_print_pushes_mapped_emotion_to_kxgauge(kxgauge_client):
    c, bridge = kxgauge_client
    bridge._kxgauge.set_emotion = MagicMock()
    bridge._on_print({"state": "printing", "data": {}})
    bridge._kxgauge.set_emotion.assert_called_once_with("happy")


@pytest.mark.asyncio
async def test_on_print_unmapped_state_does_not_call_kxgauge(kxgauge_client):
    c, bridge = kxgauge_client
    bridge._kxgauge_mapping = {}  # no entries at all
    bridge._kxgauge.set_emotion = MagicMock()
    bridge._on_print({"state": "printing", "data": {}})
    bridge._kxgauge.set_emotion.assert_not_called()


@pytest.mark.asyncio
async def test_no_kxgauge_configured_callbacks_are_noop():
    bridge = _make_bridge()  # kxgauge disabled
    assert bridge._kxgauge is None
    # must not raise even though _kxgauge is None
    bridge._on_temp({"data": {"curr_nozzle_temp": 200, "target_nozzle_temp": 200,
                               "curr_hotbed_temp": 60, "target_hotbed_temp": 60}})
    bridge._on_print({"state": "printing", "data": {}})