bambuddy/backend/tests/unit/test_nozzle_print_time_1908.py

"""H2C per-nozzle print time (#1908).

Recent H2C firmware reports how long each nozzle has printed as ``p_t`` (seconds)
on its ``device.nozzle.info`` entry. It is carried as ``print_time_s`` through
the MQTT state, the WebSocket status push and the REST status, and stays None
where the firmware does not report it.
"""

import pytest

from backend.app.schemas.printer import NozzleRackSlot
from backend.app.services.bambu_mqtt import BambuMQTTClient, PrinterState, nozzle_print_time_s
from backend.app.services.printer_manager import printer_state_to_dict


def _nozzle(nozzle_id: int, **extra) -> dict:
    return {"id": nozzle_id, "type": "HS01", "diameter": 0.4, "tm": 350, "wear": 0, "stat": 184, **extra}


def _ingested(*nozzles: dict) -> BambuMQTTClient:
    client = BambuMQTTClient(ip_address="10.0.0.1", serial_number="H2C", access_code="c", model="H2C")
    client._process_message({"print": {"device": {"nozzle": {"info": list(nozzles)}}}})
    return client


class TestParse:
    @pytest.mark.parametrize(
        ("raw", "expected"),
        [
            (0, 0),
            (7265, 7265),
            (7265.9, 7265),
            (None, None),
            (-1, None),
            ("7265", None),
            (True, None),
        ],
    )
    def test_only_a_non_negative_number_is_a_print_time(self, raw, expected):
        assert nozzle_print_time_s(raw) == expected

    def test_each_nozzle_keeps_its_own_print_time(self):
        rack = _ingested(_nozzle(0, p_t=3600), _nozzle(16, p_t=90061), _nozzle(17)).state.nozzle_rack
        assert {n["id"]: n["print_time_s"] for n in rack} == {0: 3600, 16: 90061, 17: None}

    def test_firmware_without_p_t_leaves_it_unset(self):
        (nozzle,) = _ingested(_nozzle(0)).state.nozzle_rack
        assert nozzle["print_time_s"] is None


class TestSerialise:
    def test_the_websocket_push_carries_it(self):
        state = PrinterState(connected=True, state="IDLE")
        state.nozzle_rack = _ingested(_nozzle(16, p_t=7200)).state.nozzle_rack
        (slot,) = printer_state_to_dict(state)["nozzle_rack"]
        assert slot["print_time_s"] == 7200

    def test_the_push_and_the_rest_schema_have_the_same_fields(self):
        # The frontend shallow-merges pushes over the REST status, so a field
        # the push leaves out goes stale on the first push.
        state = PrinterState(connected=True, state="IDLE")
        state.nozzle_rack = _ingested(_nozzle(16, p_t=7200)).state.nozzle_rack
        (slot,) = printer_state_to_dict(state)["nozzle_rack"]
        assert set(slot) == set(NozzleRackSlot.model_fields)
        assert NozzleRackSlot(**slot).print_time_s == 7200


class TestRestStatus:
    @pytest.mark.asyncio
    async def test_the_status_route_carries_it(self, async_client, printer_factory):
        # What the page has before the first push arrives.
        from unittest.mock import MagicMock, patch

        printer = await printer_factory()
        state = _ingested(_nozzle(0, p_t=3600), _nozzle(16)).state
        state.connected = True
        state.state = "IDLE"

        with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
            mock_pm.get_status = MagicMock(return_value=state)
            mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False)
            response = await async_client.get(f"/api/v1/printers/{printer.id}/status")

        assert response.status_code == 200
        assert {n["id"]: n["print_time_s"] for n in response.json()["nozzle_rack"]} == {0: 3600, 16: None}