bambuddy/backend/tests/unit/test_internal_printer_jobs.py

"""The printer's own calibration runs leave no archive and send no notification.

Auto pressure-advance calibration -- the K-profile line the printer lays down
before a print when flow dynamics calibration is on -- reports over MQTT through
the same print-start event a real print uses, as the subtask name
``auto_pa_line_calib_mode`` with no ``/usr/`` path attached. The only guard
Bambuddy had tested ``filename.startswith("/usr/")``, so the calibration sailed
past it, found no 3MF anywhere on the printer (there is none to find), and left
a no-3MF archive named after itself in the user's history.

The same name is already known to the completion guard: #2829's capture of
queue item 649 has ``auto_pa_line_calib_mode`` arriving as the subtask name of a
completion that had to be refused against a running job.
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from backend.app.utils.print_jobs import is_internal_printer_job


class TestTheCalibrationIsRecognised:
    def test_the_pressure_advance_line_by_subtask_name(self):
        """How it actually arrives: a bare subtask name, no filename at all."""
        assert is_internal_printer_job("", "auto_pa_line_calib_mode")

    def test_the_pressure_advance_line_by_filename(self):
        """Both fields are tested, because which one carries it is not fixed."""
        assert is_internal_printer_job("auto_pa_line_calib_mode", None)

    def test_the_levelling_run_by_its_system_path(self):
        assert is_internal_printer_job("/usr/etc/print/auto_cali_for_user.gcode", "auto_cali_for_user")

    def test_the_levelling_run_by_name_alone(self):
        """The /usr/ path is not guaranteed, so the name is listed too."""
        assert is_internal_printer_job(None, "auto_cali_for_user")

    @pytest.mark.parametrize(
        "reported",
        [
            "auto_pa_line_calib_mode",
            "auto_pa_line_calib_mode.gcode",
            "auto_pa_line_calib_mode.3mf",
            "auto_pa_line_calib_mode.gcode.3mf",
            "AUTO_PA_LINE_CALIB_MODE",
            "/data/auto_pa_line_calib_mode.gcode.3mf",
        ],
    )
    def test_however_the_name_is_dressed_up(self, reported):
        """Path, suffix and case all vary between the fields and firmwares."""
        assert is_internal_printer_job(reported, None)

    def test_anything_under_usr_counts(self):
        """Nothing a user can print lives on the read-only system partition."""
        assert is_internal_printer_job("/usr/bin/firmware_test.gcode", "test")


class TestItLeavesRealPrintsAlone:
    """The failure that matters: swallowing somebody's actual print."""

    def test_an_ordinary_print(self):
        assert not is_internal_printer_job("Benchy.gcode.3mf", "Benchy")

    def test_nothing_reported_at_all(self):
        assert not is_internal_printer_job(None, None)
        assert not is_internal_printer_job("", "")

    @pytest.mark.parametrize(
        "reported",
        [
            "auto_pa_line_calib_mode_v2.3mf",
            "my_auto_pa_line_calib_mode.3mf",
            "auto_cali_for_user_test.gcode.3mf",
        ],
    )
    def test_a_users_file_that_merely_contains_the_name(self, reported):
        """Exact match after normalising, so no prefix or substring rule can
        eat a file somebody deliberately named after the calibration."""
        assert not is_internal_printer_job(reported, None)

    def test_a_calibration_cube(self):
        """The obvious false positive for any rule built on the word 'calib'."""
        assert not is_internal_printer_job("Calibration_Cube.gcode.3mf", "Calibration Cube")


def _mocked_print_start():
    """Patch set for driving on_print_start without a printer or database."""
    return (
        patch("backend.app.main.async_session"),
        patch("backend.app.main.notification_service"),
        patch("backend.app.main.smart_plug_manager"),
        patch("backend.app.main.ws_manager"),
        patch("backend.app.main.printer_manager"),
        patch("backend.app.main.mqtt_relay"),
    )


class TestPrintStartSkipsTheCalibration:
    @pytest.mark.asyncio
    async def test_no_archive_and_no_notification(self, capture_logs):
        sess, notif, plug, ws, pm, relay = _mocked_print_start()
        with sess as mock_session_maker, notif as mock_notif, plug as mock_plug, ws as mock_ws, pm as mock_pm, relay:
            mock_notif.on_print_start = AsyncMock()
            mock_plug.on_print_start = AsyncMock()
            mock_ws.send_print_start = AsyncMock()
            mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))

            mock_printer = MagicMock()
            mock_printer.auto_archive = True
            mock_printer.id = 1

            mock_session = AsyncMock()
            mock_session.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session.__aexit__ = AsyncMock()
            mock_session.execute = AsyncMock(
                return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_printer))
            )
            mock_session_maker.return_value = mock_session

            with patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock) as mock_notify:
                from backend.app.main import on_print_start

                # No filename: exactly what the printer reports for this run,
                # and the reason the old /usr/ prefix test never fired.
                await on_print_start(1, {"filename": "", "subtask_name": "auto_pa_line_calib_mode"})

                mock_notify.assert_not_called()

        skipped = [r for r in capture_logs.records if "internal printer job" in str(r.message)]
        assert skipped, "Should log that the calibration run was skipped"


class TestPrintCompleteStaysQuiet:
    @pytest.mark.asyncio
    async def test_no_orphan_notification_when_the_calibration_finishes(self):
        """With no archive to close, the completion would otherwise fall into
        the no-archive notification path -- which attributes an unmatched
        completion to any queue item this printer finished in the last five
        minutes. For a calibration running alongside a real print that means
        telling its owner their print is done, early and twice.
        """
        with (
            patch("backend.app.main.async_session") as mock_session_maker,
            patch("backend.app.main.ws_manager") as mock_ws,
            patch("backend.app.main.printer_manager") as mock_pm,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.spawn_background_task") as mock_spawn,
            patch("backend.app.main.clear_3mf_cache"),
        ):
            mock_ws.send_print_complete = AsyncMock()
            mock_relay.on_print_complete = AsyncMock()
            mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
            mock_pm.get_current_print_user = MagicMock(return_value=None)
            mock_pm.clear_current_print_user = MagicMock()
            mock_pm.set_awaiting_plate_clear = MagicMock()

            mock_session = AsyncMock()
            mock_session.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session.__aexit__ = AsyncMock()
            mock_session.execute = AsyncMock(
                return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=None), scalars=MagicMock())
            )
            mock_session_maker.return_value = mock_session

            from backend.app.main import on_print_complete

            await on_print_complete(
                1,
                {"filename": "", "subtask_name": "auto_pa_line_calib_mode", "status": "completed"},
            )

            spawned = [c for c in mock_spawn.call_args_list if "notify-no-archive" in str(c)]
            assert not spawned, "No completion notification should be spawned for a calibration run"

    @pytest.mark.asyncio
    async def test_a_real_orphan_print_still_notifies(self):
        """The no-archive path exists for prints started outside Bambuddy. The
        guard must not take those down with it.
        """
        with (
            patch("backend.app.main.async_session") as mock_session_maker,
            patch("backend.app.main.ws_manager") as mock_ws,
            patch("backend.app.main.printer_manager") as mock_pm,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.spawn_background_task") as mock_spawn,
            patch("backend.app.main.clear_3mf_cache"),
        ):
            mock_ws.send_print_complete = AsyncMock()
            mock_relay.on_print_complete = AsyncMock()
            mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
            mock_pm.get_current_print_user = MagicMock(return_value=None)
            mock_pm.clear_current_print_user = MagicMock()
            mock_pm.set_awaiting_plate_clear = MagicMock()

            mock_session = AsyncMock()
            mock_session.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session.__aexit__ = AsyncMock()
            mock_session.execute = AsyncMock(
                return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=None), scalars=MagicMock())
            )
            mock_session_maker.return_value = mock_session

            from backend.app.main import on_print_complete

            await on_print_complete(
                1,
                {"filename": "", "subtask_name": "Benchy", "status": "completed"},
            )

            spawned = [c for c in mock_spawn.call_args_list if "notify-no-archive" in str(c)]
            assert spawned, "An unmatched real print must still notify"