bambuddy/backend/tests/unit/test_print_control.py

"""One way to stop a print on a user's behalf.

The printer card, the queue and the two webhook routes each used to send the
stop and set the "stopped by the user" mark themselves, and one of them forgot
the mark: the print then ended as failed, with a "print failed" notification
and, on some printers, a mislabelled error such as a layer shift. They now all
call print_control.stop_print_by_user, and only it (and the automatic kill
switch, which is not a user stop) may send a stop command.
"""

import ast
from pathlib import Path
from unittest.mock import patch

import pytest

from backend.app.services.print_control import stop_print_by_user

BACKEND_APP = Path(__file__).resolve().parents[2] / "app"


@pytest.fixture
def marks():
    from backend.app.main import _user_stopped_printers

    _user_stopped_printers.clear()
    yield _user_stopped_printers
    _user_stopped_printers.clear()


def test_a_sent_stop_marks_the_printer(marks):
    with patch("backend.app.services.print_control.printer_manager.stop_print", return_value=True) as send:
        assert stop_print_by_user(4) is True
    send.assert_called_once_with(4)
    assert 4 in marks


def test_an_unsent_stop_leaves_no_mark(marks):
    with patch("backend.app.services.print_control.printer_manager.stop_print", return_value=False):
        assert stop_print_by_user(4) is False
    assert 4 not in marks


def test_the_queue_can_mark_an_unsent_stop(marks):
    """An offline printer that later reports the job as failed must still read
    it as cancelled by the user (the queue closes the job at once)."""
    with patch("backend.app.services.print_control.printer_manager.stop_print", return_value=False):
        assert stop_print_by_user(4, mark_when_unsent=True) is False
    assert 4 in marks


def test_a_failing_send_raises_and_marks_only_when_asked(marks):
    with patch("backend.app.services.print_control.printer_manager.stop_print", side_effect=OSError("mqtt down")):
        with pytest.raises(OSError):
            stop_print_by_user(4)
        assert 4 not in marks
        with pytest.raises(OSError):
            stop_print_by_user(4, mark_when_unsent=True)
    assert 4 in marks


@pytest.mark.parametrize(
    ("state", "marked"),
    [
        ("RUNNING", True),
        ("PAUSE", True),
        ("PREPARE", True),
        ("IDLE", False),
        ("FINISH", False),
        # Before the first report after a (re)connect, and after a power-off.
        ("unknown", True),
        ("", True),
        (None, True),
    ],
)
def test_a_stop_on_an_idle_printer_leaves_no_mark(marks, state, marked):
    """USP/I4: the mark is cleared only by the next print's start, which is
    missed when a new MQTT client first sees that print mid-way; its later
    failure was then reported as the user's cancel. A printer in an unknown
    state is still marked."""
    from types import SimpleNamespace

    status = SimpleNamespace(state=state) if state is not None else None
    with (
        patch("backend.app.services.print_control.printer_manager.stop_print", return_value=True) as send,
        patch("backend.app.services.print_control.printer_manager.get_status", return_value=status),
    ):
        assert stop_print_by_user(4) is True
    send.assert_called_once_with(4)
    assert (4 in marks) is marked


def test_the_queue_marks_an_idle_printer_too(marks):
    """The queue stops a job it knows it started; a printer that reads idle
    there may simply not have reported yet."""
    from types import SimpleNamespace

    with (
        patch("backend.app.services.print_control.printer_manager.stop_print", return_value=True),
        patch(
            "backend.app.services.print_control.printer_manager.get_status", return_value=SimpleNamespace(state="IDLE")
        ),
    ):
        stop_print_by_user(4, mark_when_unsent=True)
    assert 4 in marks


# Files allowed to send a stop command themselves: the shared function, the
# printer manager and MQTT client that implement it, and main.py's kill switch,
# which stops unauthorised prints automatically (not a user stop, no mark).
ALLOWED = {
    "services/print_control.py",
    "services/printer_manager.py",
    "services/bambu_mqtt.py",
    "main.py",
}


def test_no_other_code_sends_a_stop():
    """Catches a new ``.stop_print(`` call. It cannot see a stop sent as a raw
    MQTT payload (the HMS dialog's "Stop Printing" in bambu_mqtt.py, whose
    route marks the stop itself); such a path must call mark_stopped_by_user."""
    offenders = []
    for path in sorted(BACKEND_APP.rglob("*.py")):
        rel = path.relative_to(BACKEND_APP).as_posix()
        if rel in ALLOWED:
            continue
        for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))):
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == "stop_print":
                offenders.append(f"{rel}:{node.lineno}")
    assert not offenders, (
        "Stop a print through services.print_control.stop_print_by_user, which also sets the "
        f"stopped-by-user mark: {offenders}"
    )