mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
7a42e0a7e5
With a switch fitted, an AMS is not wired to a nozzle any more. It is plumbed into one of the switch's two inlets and reaches both nozzles through it, so every unit reports its extruder as "not fixed" (0xE) and ams_extruder_map comes back empty on these machines. The printer card had nothing to fall back on but the AMS unit number, so AMS-A was badged R and AMS-B was badged L purely because their unit ids are 0 and 1, a third unit got no badge at all, and every one of those labels was wrong. The SpoolBuddy assign modal had the same fallback in a worse form, mapping anything that was not extruder 1 to R. The binding turned out to need no new telemetry. BambuStudio reads it out of bits 24-27 of the same AMS info string we already parse for the AMS type and the extruder id -- 0 is In-B, 1 is In-A -- and it is only meaningful when a switch is installed, because without one 0xE really does mean an uninitialised unit and those bits carry nothing. That gates the read, which in turn forced the switch block to be parsed before the AMS block: _handle_ams_data runs early in _process_message and _update_state only much later, so the binding was lost on every frame that carried both. _parse_fila_switch is split out and called first, and left in _update_state as well so that stays a complete absorb step. The badge keeps L and R rather than A and B, because the lettering is familiar and matches the physical layout. It is a different colour from the plain nozzle badge, and its tooltip names the inlet in full, since the letter is the inlet's position and not a claim about which nozzle that AMS feeds -- the switch can route either inlet to either outlet. An AMS still reporting a real extruder id keeps its ordinary badge, which BambuStudio also treats as authoritative over any switch binding, and a switch that has been fitted but not yet set up on the printer shows nothing rather than a guess. The print dialog's slot dropdown gets the same label. It replaces a left/right hint that never once rendered: ftsExtruderForSlot compared snow-encoded in[] values against global tray ids and could not match. Decoding it correctly would not have saved it -- the firmware reports which slot sits in each inlet and which nozzle each outlet feeds, but never which inlet is currently paired with which outlet, so no per-slot nozzle can be derived. That function is gone rather than fixed. The dialog also points out when every filament a print needs sits behind one inlet. Bambu's own guidance is that this is legal but slow: a change between two filaments on the same inlet retracts the outgoing spool all the way back to its AMS before the next can be fed up the shared tube, where a change across the two inlets only retracts as far as the switch. All on one inlet means every change in the job takes the slow path, and moving a single spool fixes it. So it advises, it does not block. Both views update live. Two things were stopping that. fila_switch and ams_switch_inlet were absent from printer_state_to_dict, and the frontend shallow-merges each WebSocket push over its cached status, so a field the push omits keeps whatever the last full fetch left behind. And the broadcast dedup key had no term for either, so "Join IN-B" on the printer screen moved nothing: the binding is not in the tray component of that key, and it is not in the AMS change-hash either, which covers tray fields only and must stay that way because it drives Spoolman sync. Assigning an AMS to an inlet remains printer-side. BambuStudio can read the binding and has no command to write it -- its switch class is parse and getters only, and the recommended-arrangement popup draws and publishes nothing -- so there is no wire format for us to copy. Adding the two fields to PrinterState broke four test modules whose SimpleNamespace stubs predate them. The stubs are fixed rather than the production reads made defensive: the real dataclass always carries both, and a getattr in the dedup key would silently stop tracking the field if it were ever renamed.
489 lines
18 KiB
Python
489 lines
18 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,
|
|
# Real PrinterState always carries these; the status-broadcast dedup
|
|
# key reads them so a Filament Track Switch rebind reaches the card.
|
|
fila_switch=None,
|
|
ams_switch_inlet={},
|
|
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,
|
|
# Real PrinterState always carries these; the status-broadcast dedup
|
|
# key reads them so a Filament Track Switch rebind reaches the card.
|
|
fila_switch=None,
|
|
ams_switch_inlet={},
|
|
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,
|
|
# Real PrinterState always carries these; the status-broadcast dedup
|
|
# key reads them so a Filament Track Switch rebind reaches the card.
|
|
fila_switch=None,
|
|
ams_switch_inlet={},
|
|
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,
|
|
# Real PrinterState always carries these; the status-broadcast dedup
|
|
# key reads them so a Filament Track Switch rebind reaches the card.
|
|
fila_switch=None,
|
|
ams_switch_inlet={},
|
|
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,
|
|
# Real PrinterState always carries these; the status-broadcast dedup
|
|
# key reads them so a Filament Track Switch rebind reaches the card.
|
|
fila_switch=None,
|
|
ams_switch_inlet={},
|
|
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
|