viewit
5a44d0abab
Frees up the printer's own limited storage automatically once a print finishes, while keeping the file safely in the bridge's own GCode store. Deliberately scoped to files uploaded through the bridge itself only (matched via the GCode store, same lookup the job-history feature already uses) - a print started directly from the printer or Anycubic Slicer has no backup anywhere else, so it's never touched regardless of the setting. Only triggers on a clean "finished" state, not on stopped/canceled prints, since the user may want to retry those. Off by default. The delete request is fire-and-forget, sent directly from _on_print() (which runs on the MQTT reader thread) rather than through the existing _wait_for_file_action() helper - that helper blocks waiting for a reply dispatched from that same thread, which would deadlock if called from within it.
158 lines
5.2 KiB
Python
158 lines
5.2 KiB
Python
"""Optional auto-delete of a printed file from the printer's own storage
|
|
after a successful print (Settings -> Print -> "Delete file from printer
|
|
after successful print").
|
|
|
|
Only applies to files that are also backed by the bridge's own GCode store
|
|
(otherwise the file would be gone with no copy left anywhere) and only on a
|
|
clean "finished" state - not on stoped/canceled prints, and never when the
|
|
setting is off (the default).
|
|
"""
|
|
import argparse
|
|
import tempfile
|
|
from unittest.mock import MagicMock
|
|
|
|
from kobrax_moonraker_bridge import GCodeStore, KobraXBridge
|
|
|
|
|
|
def _bridge(delete_after_print=1):
|
|
c = MagicMock()
|
|
c.callbacks = {}
|
|
c.connected = False
|
|
args = argparse.Namespace(
|
|
printer_ip="", mqtt_port=9883, username="", password="",
|
|
mode_id="20030", device_id="", host="127.0.0.1", port=7125,
|
|
data_dir=tempfile.mkdtemp(prefix="kxdelafterprint-"),
|
|
delete_printer_file_after_print=delete_after_print,
|
|
)
|
|
store = GCodeStore(args.data_dir)
|
|
b = KobraXBridge(c, args=args, store=store)
|
|
return b, c
|
|
|
|
|
|
def _seed_file(bridge, filename="test.gcode"):
|
|
file_id = "abc123"
|
|
bridge._store.save_file(file_id, filename, b"; gcode content")
|
|
return file_id
|
|
|
|
|
|
def _print_report(state, filename=None):
|
|
payload = {"state": state, "data": {}}
|
|
if filename is not None:
|
|
payload["data"]["filename"] = filename
|
|
return payload
|
|
|
|
|
|
def test_finished_print_deletes_printer_file_when_enabled_and_in_store():
|
|
b, c = _bridge(delete_after_print=1)
|
|
_seed_file(b, "test.gcode")
|
|
|
|
b._on_print(_print_report("printing", "test.gcode"))
|
|
assert b._current_job_id
|
|
assert b._current_job_filename == "test.gcode"
|
|
|
|
b._on_print(_print_report("finished"))
|
|
|
|
delete_calls = [
|
|
call for call in c.publish.call_args_list
|
|
if call.args[:2] == ("file", "deleteBatch")
|
|
]
|
|
assert len(delete_calls) == 1
|
|
payload = delete_calls[0].args[2]
|
|
assert payload == {"root": "local", "files": [{"path": "/", "filename": "test.gcode"}]}
|
|
|
|
|
|
def test_finished_print_no_delete_when_setting_disabled():
|
|
b, c = _bridge(delete_after_print=0)
|
|
_seed_file(b, "test.gcode")
|
|
|
|
b._on_print(_print_report("printing", "test.gcode"))
|
|
b._on_print(_print_report("finished"))
|
|
|
|
delete_calls = [
|
|
call for call in c.publish.call_args_list
|
|
if call.args[:2] == ("file", "deleteBatch")
|
|
]
|
|
assert delete_calls == []
|
|
|
|
|
|
def test_finished_print_no_delete_when_file_not_in_bridge_store():
|
|
"""Files started directly from the printer/Anycubic Slicer aren't in the
|
|
bridge's own GCode store - must never be deleted, since that would leave
|
|
no copy anywhere."""
|
|
b, c = _bridge(delete_after_print=1)
|
|
# No _seed_file() call - the file is not in the store.
|
|
|
|
b._on_print(_print_report("printing", "not_in_store.gcode"))
|
|
assert not b._current_job_id # no store match -> no job tracked either
|
|
|
|
b._on_print(_print_report("finished"))
|
|
|
|
delete_calls = [
|
|
call for call in c.publish.call_args_list
|
|
if call.args[:2] == ("file", "deleteBatch")
|
|
]
|
|
assert delete_calls == []
|
|
|
|
|
|
def test_canceled_print_does_not_delete_file():
|
|
"""Only a clean "finished" triggers the delete - a stopped/canceled
|
|
print keeps its file, since the user may want to retry it."""
|
|
b, c = _bridge(delete_after_print=1)
|
|
_seed_file(b, "test.gcode")
|
|
|
|
b._on_print(_print_report("printing", "test.gcode"))
|
|
b._on_print(_print_report("canceled"))
|
|
|
|
delete_calls = [
|
|
call for call in c.publish.call_args_list
|
|
if call.args[:2] == ("file", "deleteBatch")
|
|
]
|
|
assert delete_calls == []
|
|
assert b._current_job_filename == ""
|
|
|
|
|
|
def test_stoped_print_does_not_delete_file():
|
|
b, c = _bridge(delete_after_print=1)
|
|
_seed_file(b, "test.gcode")
|
|
|
|
b._on_print(_print_report("printing", "test.gcode"))
|
|
b._on_print(_print_report("stoped"))
|
|
|
|
delete_calls = [
|
|
call for call in c.publish.call_args_list
|
|
if call.args[:2] == ("file", "deleteBatch")
|
|
]
|
|
assert delete_calls == []
|
|
|
|
|
|
def test_current_job_filename_reset_after_finished():
|
|
"""Regression guard: _current_job_filename must not leak into the next
|
|
print's finished-handling if that next print isn't itself tracked."""
|
|
b, c = _bridge(delete_after_print=1)
|
|
_seed_file(b, "test.gcode")
|
|
|
|
b._on_print(_print_report("printing", "test.gcode"))
|
|
b._on_print(_print_report("finished"))
|
|
assert b._current_job_filename == ""
|
|
|
|
# A second "finished" with no new job in between must not re-trigger a delete.
|
|
c.publish.reset_mock()
|
|
b._on_print(_print_report("finished"))
|
|
delete_calls = [
|
|
call for call in c.publish.call_args_list
|
|
if call.args[:2] == ("file", "deleteBatch")
|
|
]
|
|
assert delete_calls == []
|
|
|
|
|
|
def test_delete_publish_failure_does_not_raise():
|
|
"""A broken MQTT send during the delete request must not propagate out
|
|
of _on_print() - it runs on the MQTT reader thread, and an unhandled
|
|
exception there would break processing of subsequent messages."""
|
|
b, c = _bridge(delete_after_print=1)
|
|
_seed_file(b, "test.gcode")
|
|
c.publish.side_effect = RuntimeError("send failed")
|
|
|
|
b._on_print(_print_report("printing", "test.gcode"))
|
|
b._on_print(_print_report("finished")) # must not raise
|