mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
bf525661d3
both dialects rather than one. Postgres upgrades never got as far as the finance schema. database.py added on_billing_charge_failed with BOOLEAN DEFAULT 1. The 1 is a SQLite-ism; Postgres answers DatatypeMismatchError, and _safe_execute deliberately re-raises anything that is not an idempotency error, so run_migrations died there and rolled the whole transaction back. No finance tables, no columns, and the app does not start. Six lines above, the same change gets is_voided right with an is_sqlite() branch, so this was an oversight rather than a decision. Now branched the same way. This also explains the four test_security.py::TestBackupKeyFiles failures reporting "column print_archives.cost_center_id does not exist". That column's migration exists and works -- it simply never ran, because every startup aborted before committing. Reproduced against Postgres 16 by building a pre-billing schema from dev and upgrading over it: fails without this, completes with it, and re-running the migrations or starting from an empty database are both clean. test_billing_run_id_migration.py failed on any Postgres-configured checkout. It builds its own SQLite engine, but run_migrations branches on the global dialect rather than the connection in hand, so on a box whose DATABASE_URL points at Postgres it emitted md5(random()::text) and btrim() into SQLite. Given the same fixture test_ldap_migration.py already carries for exactly this reason. The suite now agrees across dialects -- 9190 passed either way, where it used to be 9184 on one and 9183 on the other. The kill switch could not tell a print Bambuddy started from one it merely watched. Authorization fell back to a print_archives row in status="printing" matched on subtask_id. But on_print_start archives every print it observes, including ones started from Bambu Studio or Handy, and stamps them with the same status and subtask_id -- the code says as much where it notes "a print Bambuddy didn't dispatch". So a foreign print became authorized the moment its 3MF finished downloading, and _active_prints was rehydrated from it, making that permanent. The switch fired only inside the download race, and never afterwards. Neither test caught it: one stubs the authorization call to False, the other stubs the query to return an archive, so the real lookup was never exercised against a foreign print. Authorization now requires a marker Bambuddy writes itself: billing_run_id, minted per dispatch in the scheduler, or created_by_id carried over from the queue item. Failing that, it looks for a queue row in status="printing" on that printer -- committed before the MQTT send, and the only durable trace a library-file dispatch leaves, since those have no archive at send time and the row created for them moments later carries neither marker. That row cannot be tied to a subtask_id, so it defers rather than authorizes. Deferring also closes a false positive the previous version shared: a restart in the window between the send and the download left no archive at all, and a Bambuddy print was stopped as unauthorized. Stopping a print is irreversible and declining to act costs a log line, so ambiguity resolves that way. Tests cover an unmarked archive not being authorization and not entering _active_prints, either marker alone authorizing and rehydrating the fast path without touching the queue, an unmarked archive with a live dispatch deferring, a dispatch not yet archived deferring, and nothing at all being unauthorized.
468 lines
17 KiB
Python
468 lines
17 KiB
Python
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from backend.app import main as main_module
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_kill_switch_state():
|
|
main_module._kill_switch_setting_cache = None
|
|
main_module._unauthorized_print_kill_sent.clear()
|
|
main_module._kill_switch_notification_tasks.clear()
|
|
main_module._expected_prints.clear()
|
|
main_module._active_prints.clear()
|
|
main_module._expected_print_registered_at.clear()
|
|
main_module._printer_reconciled_since_connect.clear()
|
|
yield
|
|
for task in main_module._kill_switch_notification_tasks.values():
|
|
if not task.done():
|
|
task.cancel()
|
|
main_module._unauthorized_print_kill_sent.clear()
|
|
main_module._kill_switch_notification_tasks.clear()
|
|
main_module._expected_prints.clear()
|
|
main_module._active_prints.clear()
|
|
main_module._expected_print_registered_at.clear()
|
|
main_module._printer_reconciled_since_connect.clear()
|
|
main_module._kill_switch_setting_cache = None
|
|
|
|
|
|
def test_gcode_3mf_status_filename_matches_registered_expected_print():
|
|
state = SimpleNamespace(
|
|
current_print=None,
|
|
subtask_name="",
|
|
gcode_file="foreign_job.gcode.3mf",
|
|
)
|
|
|
|
keys = main_module._build_status_print_keys(7, state)
|
|
|
|
assert (7, "foreign_job.gcode.3mf") in keys
|
|
assert (7, "foreign_job.gcode") in keys
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unauthorized_active_print_triggers_stop(monkeypatch):
|
|
stop_calls: list[int] = []
|
|
broadcast = AsyncMock()
|
|
provider_notification = AsyncMock(return_value=True)
|
|
|
|
async def fake_status(*args, **kwargs):
|
|
return None
|
|
|
|
async def kill_switch_enabled(_db):
|
|
return True
|
|
|
|
unauthorized = AsyncMock(return_value=False)
|
|
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
monkeypatch.setattr(
|
|
main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
|
|
)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
|
|
monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
|
|
monkeypatch.setattr(main_module.ws_manager, "broadcast", broadcast)
|
|
monkeypatch.setattr(main_module, "_is_bambuddy_authorized_print", unauthorized)
|
|
monkeypatch.setattr(main_module, "_send_kill_switch_provider_notification", provider_notification)
|
|
monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
|
|
|
|
state = SimpleNamespace(
|
|
connected=True,
|
|
state="RUNNING",
|
|
progress=0,
|
|
remaining_time=0,
|
|
layer_num=0,
|
|
temperatures={},
|
|
raw_data={},
|
|
stg_cur=0,
|
|
cooling_fan_speed=None,
|
|
big_fan1_speed=None,
|
|
big_fan2_speed=None,
|
|
chamber_light=False,
|
|
active_extruder=0,
|
|
tray_now=255,
|
|
door_open=False,
|
|
ams_filament_backup=False,
|
|
current_print=None,
|
|
subtask_name="foreign_job",
|
|
subtask_id="external-task-1",
|
|
gcode_file="foreign_job.gcode",
|
|
)
|
|
|
|
await main_module.on_printer_status_change(7, state)
|
|
await main_module.on_printer_status_change(7, state)
|
|
|
|
assert stop_calls == [7]
|
|
unauthorized.assert_awaited_once()
|
|
assert 7 in main_module._unauthorized_print_kill_sent
|
|
broadcast.assert_awaited_once_with(
|
|
{
|
|
"type": "kill_switch_triggered",
|
|
"printer_id": 7,
|
|
"printer_name": "Printer 7",
|
|
"filename": "foreign_job",
|
|
"reason": "unauthorized_print",
|
|
}
|
|
)
|
|
notification_task = main_module._kill_switch_notification_tasks[7]
|
|
assert await notification_task is True
|
|
provider_notification.assert_awaited_once_with(
|
|
7,
|
|
"Printer 7",
|
|
{
|
|
"status": "stopped",
|
|
"filename": "foreign_job.gcode",
|
|
"subtask_name": "foreign_job",
|
|
"progress": 0,
|
|
"reason": "unauthorized_print",
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_immediate_notification_allows_completion_retry():
|
|
task = main_module.spawn_background_task(_return_false(), name="test-kill-switch-notification-failure")
|
|
|
|
assert await main_module._kill_switch_notification_already_sent(task) is False
|
|
|
|
|
|
async def _return_false():
|
|
return False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bambuddy_authorized_print_is_not_stopped(monkeypatch):
|
|
monkeypatch.setitem(main_module._expected_prints, (7, "foreign_job"), 123)
|
|
|
|
stop_calls: list[int] = []
|
|
|
|
async def fake_status(*args, **kwargs):
|
|
return None
|
|
|
|
kill_switch_enabled = AsyncMock(return_value=True)
|
|
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
monkeypatch.setattr(
|
|
main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
|
|
)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
|
|
monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
|
|
monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
|
|
|
|
state = SimpleNamespace(
|
|
connected=True,
|
|
state="RUNNING",
|
|
progress=0,
|
|
remaining_time=0,
|
|
layer_num=0,
|
|
temperatures={},
|
|
raw_data={},
|
|
stg_cur=0,
|
|
cooling_fan_speed=None,
|
|
big_fan1_speed=None,
|
|
big_fan2_speed=None,
|
|
chamber_light=False,
|
|
active_extruder=0,
|
|
tray_now=255,
|
|
door_open=False,
|
|
ams_filament_backup=False,
|
|
current_print=None,
|
|
subtask_name="foreign_job",
|
|
gcode_file="foreign_job.gcode",
|
|
)
|
|
|
|
await main_module.on_printer_status_change(7, state)
|
|
|
|
assert stop_calls == []
|
|
assert 7 not in main_module._unauthorized_print_kill_sent
|
|
kill_switch_enabled.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kill_switch_setting_is_cached(monkeypatch):
|
|
kill_switch_enabled = AsyncMock(return_value=True)
|
|
|
|
class FakeSessionContext:
|
|
async def __aenter__(self):
|
|
return SimpleNamespace()
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
monkeypatch.setattr(main_module, "async_session", FakeSessionContext)
|
|
monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
|
|
|
|
assert await main_module._is_printer_kill_switch_enabled_cached() is True
|
|
assert await main_module._is_printer_kill_switch_enabled_cached() is True
|
|
kill_switch_enabled.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
|
|
stop_calls: list[int] = []
|
|
|
|
async def fake_status(*args, **kwargs):
|
|
return None
|
|
|
|
async def kill_switch_enabled(_db):
|
|
return True
|
|
|
|
async def unauthorized(*_args):
|
|
return False
|
|
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
monkeypatch.setattr(
|
|
main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
|
|
)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
|
|
monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
|
|
monkeypatch.setattr(main_module, "_is_bambuddy_authorized_print", unauthorized)
|
|
monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
|
|
|
|
active_state = SimpleNamespace(
|
|
connected=True,
|
|
state="RUNNING",
|
|
progress=0,
|
|
remaining_time=0,
|
|
layer_num=0,
|
|
temperatures={},
|
|
raw_data={},
|
|
stg_cur=0,
|
|
cooling_fan_speed=None,
|
|
big_fan1_speed=None,
|
|
big_fan2_speed=None,
|
|
chamber_light=False,
|
|
active_extruder=0,
|
|
tray_now=255,
|
|
door_open=False,
|
|
ams_filament_backup=False,
|
|
current_print=None,
|
|
subtask_name="foreign_job",
|
|
subtask_id="external-task-1",
|
|
gcode_file="foreign_job.gcode",
|
|
)
|
|
|
|
idle_state = SimpleNamespace(
|
|
connected=True,
|
|
state="IDLE",
|
|
progress=0,
|
|
remaining_time=0,
|
|
layer_num=0,
|
|
temperatures={},
|
|
raw_data={},
|
|
stg_cur=0,
|
|
cooling_fan_speed=None,
|
|
big_fan1_speed=None,
|
|
big_fan2_speed=None,
|
|
chamber_light=False,
|
|
active_extruder=0,
|
|
tray_now=255,
|
|
door_open=False,
|
|
ams_filament_backup=False,
|
|
current_print=None,
|
|
subtask_name="",
|
|
subtask_id=None,
|
|
gcode_file=None,
|
|
)
|
|
|
|
await main_module.on_printer_status_change(7, active_state)
|
|
assert stop_calls == [7]
|
|
assert 7 in main_module._unauthorized_print_kill_sent
|
|
|
|
await main_module.on_printer_status_change(7, idle_state)
|
|
|
|
assert 7 not in main_module._unauthorized_print_kill_sent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("printer_state", ["RUNNING", "PAUSE"])
|
|
async def test_persisted_print_is_authorized_after_restart(monkeypatch, printer_state):
|
|
# billing_run_id is the marker the scheduler stamps on its own dispatches;
|
|
# an archive without one proves only that Bambuddy watched the print.
|
|
archive = SimpleNamespace(
|
|
id=123,
|
|
filename="owned_job.gcode.3mf",
|
|
billing_run_id="d7c1f0b2-0000-4000-8000-000000000001",
|
|
created_by_id=None,
|
|
)
|
|
query_result = SimpleNamespace(scalar_one_or_none=lambda: archive)
|
|
db = SimpleNamespace(execute=AsyncMock(return_value=query_result))
|
|
|
|
class FakeSessionContext:
|
|
async def __aenter__(self):
|
|
return db
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
stop_calls: list[int] = []
|
|
|
|
async def fake_status(*args, **kwargs):
|
|
return None
|
|
|
|
async def kill_switch_enabled(_db):
|
|
return True
|
|
|
|
def discard_background_task(coro, **_kwargs):
|
|
coro.close()
|
|
|
|
monkeypatch.setattr(main_module, "async_session", FakeSessionContext)
|
|
monkeypatch.setattr(main_module, "spawn_background_task", discard_background_task)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
monkeypatch.setattr(
|
|
main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
|
|
)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
|
|
monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
|
|
monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
|
|
monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
|
|
|
|
state = SimpleNamespace(
|
|
connected=True,
|
|
state=printer_state,
|
|
progress=42,
|
|
remaining_time=600,
|
|
layer_num=50,
|
|
temperatures={},
|
|
raw_data={},
|
|
stg_cur=0,
|
|
cooling_fan_speed=None,
|
|
big_fan1_speed=None,
|
|
big_fan2_speed=None,
|
|
chamber_light=False,
|
|
active_extruder=0,
|
|
tray_now=255,
|
|
door_open=False,
|
|
ams_filament_backup=False,
|
|
current_print=None,
|
|
subtask_name="owned_job",
|
|
subtask_id="bambuddy-task-123",
|
|
gcode_file="owned_job.gcode.3mf",
|
|
)
|
|
|
|
await main_module.on_printer_status_change(7, state)
|
|
|
|
assert stop_calls == []
|
|
assert (7, "owned_job.gcode.3mf") in main_module._active_prints
|
|
assert main_module._active_prints[(7, "owned_job.gcode.3mf")] == 123
|
|
assert 7 not in main_module._unauthorized_print_kill_sent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kill_switch_defers_when_restart_identity_is_not_available(monkeypatch):
|
|
state = SimpleNamespace(
|
|
current_print=None,
|
|
subtask_name="owned_job",
|
|
subtask_id=None,
|
|
gcode_file="owned_job.gcode.3mf",
|
|
)
|
|
db = SimpleNamespace(execute=AsyncMock())
|
|
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
|
|
authorization = await main_module._is_bambuddy_authorized_print(7, state, db)
|
|
|
|
assert authorization is None
|
|
db.execute.assert_not_awaited()
|
|
|
|
|
|
def _authorization_db(archive, dispatched_queue_item_id=None):
|
|
"""Fake session answering the two lookups `_is_bambuddy_authorized_print` makes."""
|
|
|
|
query_result = SimpleNamespace(scalar_one_or_none=lambda: archive)
|
|
return SimpleNamespace(
|
|
execute=AsyncMock(return_value=query_result),
|
|
scalar=AsyncMock(return_value=dispatched_queue_item_id),
|
|
)
|
|
|
|
|
|
def _running_state(subtask_id="external-task-9"):
|
|
return SimpleNamespace(
|
|
current_print=None,
|
|
subtask_name="some_job",
|
|
subtask_id=subtask_id,
|
|
gcode_file="some_job.gcode.3mf",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_without_a_dispatch_marker_is_not_authorization(monkeypatch):
|
|
"""on_print_start archives prints started from Studio or Handy too.
|
|
|
|
Those rows carry the same status and subtask_id as Bambuddy's own, so treating
|
|
the row's existence as proof would switch the feature off a few seconds into
|
|
every foreign print — as soon as the 3MF finished downloading.
|
|
"""
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
observed_only = SimpleNamespace(
|
|
id=55,
|
|
filename="some_job.gcode.3mf",
|
|
billing_run_id=None,
|
|
created_by_id=None,
|
|
)
|
|
db = _authorization_db(observed_only, dispatched_queue_item_id=None)
|
|
|
|
assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is False
|
|
assert (9, "some_job.gcode.3mf") not in main_module._active_prints
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"marker",
|
|
[
|
|
{"billing_run_id": "9f0c2b6e-0000-4000-8000-00000000abcd", "created_by_id": None},
|
|
{"billing_run_id": None, "created_by_id": 4},
|
|
],
|
|
ids=["billing_run_id", "created_by_id"],
|
|
)
|
|
async def test_either_dispatch_marker_authorizes_after_a_restart(monkeypatch, marker):
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
archive = SimpleNamespace(id=77, filename="some_job.gcode.3mf", **marker)
|
|
db = _authorization_db(archive, dispatched_queue_item_id=None)
|
|
|
|
assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is True
|
|
assert main_module._active_prints[(9, "some_job.gcode.3mf")] == 77
|
|
# The fast path is rehydrated, so the queue is never consulted.
|
|
db.scalar.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_defers_while_bambuddy_has_a_job_running_on_that_printer(monkeypatch):
|
|
"""A library-file dispatch has no archive at send time, and the row created for
|
|
it moments later by on_print_start carries neither marker. The queue row is the
|
|
only durable trace, and it cannot be tied to a subtask_id — so it defers."""
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
unmarked = SimpleNamespace(id=56, filename="some_job.gcode.3mf", billing_run_id=None, created_by_id=None)
|
|
db = _authorization_db(unmarked, dispatched_queue_item_id=310)
|
|
|
|
assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is None
|
|
# Deferring must not authorize the print for every later frame.
|
|
assert (9, "some_job.gcode.3mf") not in main_module._active_prints
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_defers_when_the_dispatch_has_not_been_archived_yet(monkeypatch):
|
|
"""Restart during the window between the MQTT send and the 3MF download."""
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
db = _authorization_db(None, dispatched_queue_item_id=311)
|
|
|
|
assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_foreign_print_with_no_archive_and_no_dispatch_is_unauthorized(monkeypatch):
|
|
monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
|
|
db = _authorization_db(None, dispatched_queue_item_id=None)
|
|
|
|
assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is False
|