mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
164382b38b
Manual flow dynamics has two shapes and each reports under its own name with no auto_ prefix. pa_pattern_calib_mode was already filtered; pa_line_calib_mode was not, so it still swept FTP for a 3MF that cannot exist and wrote a no-3MF archive named after the calibration.
253 lines
12 KiB
Python
253 lines
12 KiB
Python
"""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.
|
|
|
|
The manual flow-dynamics run reaches Bambuddy the same way under two further
|
|
names -- ``pa_line_calib_mode`` and ``pa_pattern_calib_mode``, the line and the
|
|
pattern shape, both without the ``auto_`` prefix -- so the automatic entry never
|
|
covered either of them.
|
|
"""
|
|
|
|
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_manual_pressure_advance_pattern_by_subtask_name(self):
|
|
"""The same calibration run by hand rather than before a print. It
|
|
prints a pattern where the automatic one prints a line, and carries its
|
|
own name with no ``auto_`` prefix -- so the automatic entry never
|
|
covered it and it left the same no-3MF archive."""
|
|
assert is_internal_printer_job("", "pa_pattern_calib_mode")
|
|
|
|
def test_the_manual_pressure_advance_pattern_by_filename(self):
|
|
assert is_internal_printer_job("pa_pattern_calib_mode", None)
|
|
|
|
def test_the_manual_pressure_advance_line_by_subtask_name(self):
|
|
"""Manual flow dynamics has a line shape as well as a pattern one, and
|
|
it reports as ``pa_line_calib_mode`` -- the automatic name without its
|
|
``auto_`` prefix, which is close enough to the automatic entry to look
|
|
covered and is not."""
|
|
assert is_internal_printer_job("", "pa_line_calib_mode")
|
|
|
|
def test_the_manual_pressure_advance_line_by_filename(self):
|
|
assert is_internal_printer_job("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",
|
|
"pa_pattern_calib_mode",
|
|
"pa_pattern_calib_mode.gcode.3mf",
|
|
"PA_Pattern_Calib_Mode",
|
|
"/data/pa_pattern_calib_mode.gcode",
|
|
"pa_line_calib_mode",
|
|
"pa_line_calib_mode.gcode.3mf",
|
|
"PA_Line_Calib_Mode",
|
|
"/data/pa_line_calib_mode.gcode",
|
|
],
|
|
)
|
|
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",
|
|
"pa_pattern_calib_mode_v2.3mf",
|
|
"my_pa_pattern_calib_mode.3mf",
|
|
"pa_line_calib_mode_v2.3mf",
|
|
"my_pa_line_calib_mode.3mf",
|
|
# The stem the four calibration names share, which is why the set
|
|
# holds literals instead of a ``pa_`` rule.
|
|
"pa_bracket.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"
|