mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
8fd1f884dc
When a print reaches a terminal state Bambuddy holds the queue until
someone confirms the build plate is clear. That gate was visible only in
the Web UI: the printer's own MQTT push reports nothing beyond RUNNING,
PAUSE, FAILED, FINISH and IDLE, so an external automation could not tell
"finished" from "finished and still waiting for a human".
The per-printer status topic now carries an awaiting_plate_clear field,
and every transition is additionally published on a new retained topic,
bambuddy/printers/{serial}/plate_clear. Retained, and published from the
flag itself rather than from printer telemetry: a subscriber learns the
state of every printer the moment it connects, and the state stays
correct after Auto Off powers a printer down - telemetry stops there,
which would otherwise leave the status topic frozen at false.
Publishing is edge-triggered. The queue clears the gate on every
dispatch whether or not it was up, and no subscriber should see a
"plate cleared" for a plate that was never dirty. Persistence and the
WebSocket broadcast stay unconditional; they are idempotent and predate
this.
A matching Plate Clear Required notification event was added, off by
default on every provider because it fires after every print at the
same moment as the print-complete alert. Only the rising edge notifies.
Acknowledging still goes through POST /printers/{id}/clear-plate.
Two tests in test_printer_manager_status_broadcast.py asserted
_schedule_async.call_count == 2 for the setter. The new emission makes
it three on a transition, so they now assert that the persist and
broadcast coroutines are actually scheduled - which is the contract
Translated in all locales; wiki updated. Covered by backend and
frontend tests.
80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
"""Integration tests for the plate-clear-required notification (#2525).
|
|
|
|
The event is opt-in: it fires after every print, at the same moment as the
|
|
print-complete alert, so a provider only receives it when the toggle is
|
|
explicitly enabled.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from backend.app.services.notification_service import notification_service
|
|
|
|
|
|
class TestPlateClearNotificationDispatch:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_sends_to_a_provider_that_opted_in(self, notification_provider_factory, db_session):
|
|
await notification_provider_factory(name="Opted In", on_plate_clear_required=True)
|
|
|
|
send = AsyncMock()
|
|
with patch.object(notification_service, "_send_to_providers", send):
|
|
await notification_service.on_plate_clear_required(1, "Workshop", db_session)
|
|
|
|
assert send.await_count == 1
|
|
providers = send.await_args.args[0]
|
|
assert [p.name for p in providers] == ["Opted In"]
|
|
assert send.await_args.args[4] == "plate_clear_required"
|
|
# _build_message_from_template folds in app_name/timestamp; the caller's
|
|
# own variable is what matters here.
|
|
assert send.await_args.kwargs["variables"]["printer"] == "Workshop"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_silent_for_a_provider_that_did_not_opt_in(self, notification_provider_factory, db_session):
|
|
await notification_provider_factory(name="Default Off", on_plate_clear_required=False)
|
|
|
|
send = AsyncMock()
|
|
with patch.object(notification_service, "_send_to_providers", send):
|
|
await notification_service.on_plate_clear_required(1, "Workshop", db_session)
|
|
|
|
send.assert_not_awaited()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_skips_a_provider_scoped_to_a_different_printer(self, notification_provider_factory, db_session):
|
|
await notification_provider_factory(name="Other Printer", on_plate_clear_required=True, printer_id=99)
|
|
|
|
send = AsyncMock()
|
|
with patch.object(notification_service, "_send_to_providers", send):
|
|
await notification_service.on_plate_clear_required(1, "Workshop", db_session)
|
|
|
|
send.assert_not_awaited()
|
|
|
|
|
|
class TestPlateClearProviderField:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_defaults_to_off_on_create_and_round_trips_on_update(self, async_client: AsyncClient):
|
|
create = await async_client.post(
|
|
"/api/v1/notifications/",
|
|
json={
|
|
"name": "Plate Clear Test",
|
|
"provider_type": "ntfy",
|
|
"enabled": True,
|
|
"config": {"server": "https://ntfy.sh", "topic": "test-topic"},
|
|
},
|
|
)
|
|
assert create.status_code in (200, 201), create.text
|
|
provider_id = create.json()["id"]
|
|
assert create.json()["on_plate_clear_required"] is False
|
|
|
|
update = await async_client.patch(
|
|
f"/api/v1/notifications/{provider_id}",
|
|
json={"on_plate_clear_required": True},
|
|
)
|
|
assert update.status_code == 200, update.text
|
|
assert update.json()["on_plate_clear_required"] is True
|