public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-20 00:38:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Show Home Assistant sensors on the printer card (#1148, #448)

Browse Source 
Binds binary_sensor and reading-carrying sensor entities to a printer and
renders their state on its card, worded by Home Assistant's device_class.
Optional per-sensor alert condition drives a notification on the transition
into the alert state and an opt-in interlock that holds queued prints while
alerting -- a hold with a readable waiting_reason, never a failure, and only
ever on a sensor that was read successfully.

Sensors get their own table rather than a wider entity pattern on SmartPlug:
get_smart_plug_by_printer would otherwise hand the card's power button a door
contact to switch.

The hold is passed to the model matcher directly rather than merged into
busy_printers: _check_auto_drying reads that set as "is currently printing"
and would put an idle-but-held printer down the mid-print drying path.

The notification_providers migration spells its default FALSE, not 0 --
Postgres rejects an integer default for a boolean and _safe_execute swallows
the error.
This commit is contained in:
maziggy
2026-08-05 14:26:38 +02:00
parent 945d4ca6eb
commit cd004df817
45 changed files with 3782 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/integration/test_ha_sensors_api_1148.py
+317
View File
@@ -0,0 +1,317 @@
"""Integration tests for the printer-bound Home Assistant sensor API (#1148)."""
from unittest.mock import AsyncMock, patch
import pytest
from httpx import AsyncClient
from backend.app.services.ha_sensor_manager import SensorReading, ha_sensor_manager
DOOR = {
"name": "Enclosure Door",
"entity_id": "binary_sensor.enclosure_door",
"kind": "binary",
"device_class": "door",
"alert_state": "on",
}
TEMP = {
"name": "Enclosure Temp",
"entity_id": "sensor.enclosure_temp",
"kind": "numeric",
"device_class": "temperature",
"unit": "°C",
}
@pytest.fixture(autouse=True)
def _no_live_ha():
"""Creating or editing a sensor reads it once; keep that off the network."""
with patch.object(ha_sensor_manager, "refresh_one", AsyncMock()):
yield
@pytest.fixture(autouse=True)
def _clean_cache():
yield
ha_sensor_manager._readings.clear()
ha_sensor_manager._last_alerting.clear()
class TestCrud:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_bind_a_door_contact(self, async_client: AsyncClient, printer_factory):
printer = await printer_factory()
response = await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
assert response.status_code == 200
body = response.json()
assert body["entity_id"] == "binary_sensor.enclosure_door"
assert body["kind"] == "binary"
assert body["show_on_printer_card"] is True
# Display-only until the user opts in.
assert body["block_print"] is False
assert body["notify_on_alert"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_rejects_a_switch(self, async_client: AsyncClient, printer_factory):
"""Switches are smart plugs; this table is read-only sensors."""
printer = await printer_factory()
response = await async_client.post(
"/api/v1/ha-sensors/",
json={**DOOR, "printer_id": printer.id, "entity_id": "switch.printer_plug"},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_rejects_a_kind_that_contradicts_the_entity(self, async_client: AsyncClient, printer_factory):
printer = await printer_factory()
response = await async_client.post(
"/api/v1/ha-sensors/",
json={**TEMP, "printer_id": printer.id, "kind": "binary"},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_rejects_an_interlock_with_nothing_to_trigger_on(self, async_client: AsyncClient, printer_factory):
"""block_print without an alert condition would never fire — that reads
as a broken setting, not as a no-op."""
printer = await printer_factory()
response = await async_client.post(
"/api/v1/ha-sensors/",
json={**DOOR, "printer_id": printer.id, "alert_state": None, "block_print": True},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_rejects_a_duplicate_binding(self, async_client: AsyncClient, printer_factory):
printer = await printer_factory()
payload = {**DOOR, "printer_id": printer.id}
await async_client.post("/api/v1/ha-sensors/", json=payload)
response = await async_client.post("/api/v1/ha-sensors/", json=payload)
assert response.status_code == 400
assert "already bound" in response.json()["detail"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_rejects_an_unknown_printer(self, async_client: AsyncClient):
response = await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": 9999})
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_patch_revalidates_against_the_stored_row(self, async_client: AsyncClient, printer_factory):
"""The payload carries only block_print, so the coherence rule has to be
re-run against the merged row, not against the patch alone."""
printer = await printer_factory()
created = await async_client.post(
"/api/v1/ha-sensors/",
json={**DOOR, "printer_id": printer.id, "alert_state": None},
)
sensor_id = created.json()["id"]
response = await async_client.patch(f"/api/v1/ha-sensors/{sensor_id}", json={"block_print": True})
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_patch_accepts_a_coherent_change(self, async_client: AsyncClient, printer_factory):
printer = await printer_factory()
created = await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
sensor_id = created.json()["id"]
response = await async_client.patch(
f"/api/v1/ha-sensors/{sensor_id}",
json={"block_print": True, "notify_on_alert": True, "name": "Front Door"},
)
assert response.status_code == 200
assert response.json()["block_print"] is True
assert response.json()["name"] == "Front Door"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_drops_the_cached_reading(self, async_client: AsyncClient, printer_factory):
"""Otherwise a later sensor reusing the id inherits this one's state."""
printer = await printer_factory()
created = await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
sensor_id = created.json()["id"]
ha_sensor_manager._readings[sensor_id] = SensorReading("on", None, True, True)
response = await async_client.delete(f"/api/v1/ha-sensors/{sensor_id}")
assert response.status_code == 200
assert ha_sensor_manager.get_reading(sensor_id) is None
class TestReadings:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_serves_the_cached_reading(self, async_client: AsyncClient, printer_factory):
printer = await printer_factory()
created = await async_client.post(
"/api/v1/ha-sensors/",
json={**TEMP, "printer_id": printer.id, "alert_above": 35},
)
sensor_id = created.json()["id"]
ha_sensor_manager._readings[sensor_id] = SensorReading("41.2", 41.2, True, True)
response = await async_client.get(f"/api/v1/ha-sensors/by-printer/{printer.id}/readings")
assert response.status_code == 200
reading = response.json()[0]
assert reading["value"] == 41.2
assert reading["alerting"] is True
assert reading["unit"] == "°C"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_unpolled_sensor_reports_unreachable_not_missing(self, async_client: AsyncClient, printer_factory):
"""Right after a restart the card should still list the sensor, greyed
out — not drop it and reflow the layout."""
printer = await printer_factory()
await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
response = await async_client.get(f"/api/v1/ha-sensors/by-printer/{printer.id}/readings")
assert len(response.json()) == 1
assert response.json()[0]["reachable"] is False
assert response.json()[0]["alerting"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_hidden_sensors_stay_off_the_card(self, async_client: AsyncClient, printer_factory):
"""An interlock the user does not want cluttering the card still works."""
printer = await printer_factory()
await async_client.post(
"/api/v1/ha-sensors/",
json={**DOOR, "printer_id": printer.id, "show_on_printer_card": False},
)
response = await async_client.get(f"/api/v1/ha-sensors/by-printer/{printer.id}/readings")
assert response.json() == []
@pytest.mark.asyncio
@pytest.mark.integration
async def test_readings_follow_sort_order(self, async_client: AsyncClient, printer_factory):
printer = await printer_factory()
await async_client.post(
"/api/v1/ha-sensors/",
json={**TEMP, "printer_id": printer.id, "sort_order": 2},
)
await async_client.post(
"/api/v1/ha-sensors/",
json={**DOOR, "printer_id": printer.id, "sort_order": 1},
)
response = await async_client.get(f"/api/v1/ha-sensors/by-printer/{printer.id}/readings")
assert [r["name"] for r in response.json()] == ["Enclosure Door", "Enclosure Temp"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_other_printers_sensors_are_not_listed(self, async_client: AsyncClient, printer_factory):
one = await printer_factory()
two = await printer_factory(serial_number="OTHER123", name="Second")
await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": one.id})
response = await async_client.get(f"/api/v1/ha-sensors/by-printer/{two.id}/readings")
assert response.json() == []
class TestEntityPicker:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_explains_itself_when_ha_is_not_configured(self, async_client: AsyncClient):
response = await async_client.get("/api/v1/ha-sensors/entities")
assert response.status_code == 400
assert "Home Assistant not configured" in response.json()["detail"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_entities_is_not_parsed_as_a_sensor_id(self, async_client: AsyncClient):
"""Route ordering regression: /entities must not hit /{sensor_id}."""
response = await async_client.get("/api/v1/ha-sensors/entities")
assert response.status_code != 404
class TestCascadeAndUniqueness:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_patch_cannot_create_a_duplicate_binding(self, async_client: AsyncClient, printer_factory):
printer = await printer_factory()
await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
second = await async_client.post("/api/v1/ha-sensors/", json={**TEMP, "printer_id": printer.id})
response = await async_client.patch(
f"/api/v1/ha-sensors/{second.json()['id']}",
json={"entity_id": DOOR["entity_id"], "kind": "binary"},
)
assert response.status_code == 400
assert "already bound" in response.json()["detail"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_patch_to_the_same_entity_is_not_a_clash_with_itself(
self, async_client: AsyncClient, printer_factory
):
printer = await printer_factory()
created = await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
response = await async_client.patch(
f"/api/v1/ha-sensors/{created.json()['id']}",
json={"entity_id": DOOR["entity_id"], "name": "Front Door"},
)
assert response.status_code == 200
@pytest.mark.asyncio
@pytest.mark.integration
async def test_deleting_a_printer_takes_its_sensors(self, async_client: AsyncClient, printer_factory):
"""The relationship cascades, so no orphan row is left holding a
printer_id that no longer resolves."""
printer = await printer_factory()
await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
deleted = await async_client.delete(f"/api/v1/printers/{printer.id}")
assert deleted.status_code == 200
listed = await async_client.get("/api/v1/ha-sensors/")
assert listed.json() == []
class TestSaveSurvivesHomeAssistant:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_succeeds_even_if_the_first_read_blows_up(self, async_client: AsyncClient, printer_factory):
"""The row is committed before the read. Reporting a failure for work
that succeeded would send the user into a retry that 400s on the
duplicate they just created."""
printer = await printer_factory()
with patch.object(ha_sensor_manager, "refresh_one", AsyncMock(side_effect=RuntimeError("HA said no"))):
response = await async_client.post("/api/v1/ha-sensors/", json={**DOOR, "printer_id": printer.id})
assert response.status_code == 200
listed = await async_client.get(f"/api/v1/ha-sensors/?printer_id={printer.id}")
assert len(listed.json()) == 1
backend/tests/unit/test_ha_sensor_manager_1148.py
+281
View File
@@ -0,0 +1,281 @@
"""Unit tests for Home Assistant sensors bound to a printer (#1148, #448).
The alert rules decide three separate things — the pill colour on the card, a
notification, and whether the queue holds — so they are tested directly rather
than through any one of those consumers.
The recurring theme is that "we could not read it" must never be mistaken for
a reading. A door contact whose integration has dropped out reports
"unavailable", not "closed", and treating that as closed would let a print
start into an open enclosure; treating it as *open* would strand the queue.
Neither: it is not a reading at all.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from backend.app.services.ha_sensor_manager import (
HASensorManager,
SensorReading,
describe_state,
evaluate,
)
def _sensor(**overrides):
"""A sensor row as the poller sees it, without touching the DB."""
base = {
"id": 1,
"printer_id": 4,
"name": "Enclosure Door",
"entity_id": "binary_sensor.enclosure_door",
"kind": "binary",
"device_class": "door",
"unit": None,
"alert_state": "on",
"alert_above": None,
"alert_below": None,
"block_print": False,
"notify_on_alert": False,
"last_state": None,
}
base.update(overrides)
return SimpleNamespace(**base)
def _numeric(**overrides):
base = {
"entity_id": "sensor.enclosure_temp",
"kind": "numeric",
"device_class": "temperature",
"unit": "\u00b0C",
"alert_state": None,
"name": "Enclosure Temp",
}
base.update(overrides)
return _sensor(**base)
class TestEvaluateBinary:
def test_alerts_in_the_configured_state(self):
reading = evaluate(_sensor(), {"state": "on"})
assert reading == SensorReading(state="on", value=None, alerting=True, reachable=True)
def test_quiet_in_the_other_state(self):
assert evaluate(_sensor(), {"state": "off"}).alerting is False
def test_alert_state_off_inverts_the_rule(self):
"""A "fan running" contact alarms when it stops, not when it starts."""
sensor = _sensor(alert_state="off", name="Exhaust Fan")
assert evaluate(sensor, {"state": "off"}).alerting is True
assert evaluate(sensor, {"state": "on"}).alerting is False
def test_no_alert_state_never_alerts(self):
"""Display-only sensors are the default — they just show a state."""
sensor = _sensor(alert_state=None)
assert evaluate(sensor, {"state": "on"}).alerting is False
assert evaluate(sensor, {"state": "on"}).reachable is True
def test_state_is_normalised_to_lower_case(self):
"""Some integrations report "ON"; the alert rule stores "on"."""
assert evaluate(_sensor(), {"state": "ON"}).state == "on"
assert evaluate(_sensor(), {"state": "ON"}).alerting is True
class TestEvaluateNumeric:
def test_above_threshold_alerts(self):
assert evaluate(_numeric(alert_above=35), {"state": "41.2"}).alerting is True
def test_below_threshold_alerts(self):
assert evaluate(_numeric(alert_below=15), {"state": "12"}).alerting is True
def test_inside_the_band_is_quiet(self):
reading = evaluate(_numeric(alert_above=35, alert_below=15), {"state": "22.5"})
assert reading.alerting is False
assert reading.value == 22.5
def test_exactly_on_the_threshold_is_not_an_alert(self):
"""Strict comparison, so a 35 °C limit does not alarm at exactly 35."""
assert evaluate(_numeric(alert_above=35), {"state": "35"}).alerting is False
def test_a_sensor_that_stops_reporting_numbers_does_not_alert(self):
"""Reachable, but no value to compare — so no verdict either way."""
reading = evaluate(_numeric(alert_above=35), {"state": "calibrating"})
assert reading.reachable is True
assert reading.value is None
assert reading.alerting is False
class TestUnreadable:
@pytest.mark.parametrize("state", ["unavailable", "unknown", None])
def test_ha_non_states_are_not_readings(self, state):
reading = evaluate(_sensor(), {"state": state})
assert reading.reachable is False
assert reading.alerting is False
assert reading.state is None
def test_a_failed_fetch_is_not_a_reading(self):
"""fetch_states maps an entity it could not read to None."""
reading = evaluate(_sensor(), None)
assert reading == SensorReading(state=None, value=None, alerting=False, reachable=False)
class TestDescribeState:
def test_binary_uses_the_raw_state(self):
assert describe_state(_sensor(), evaluate(_sensor(), {"state": "on"})) == "on"
def test_numeric_carries_its_unit(self):
sensor = _numeric()
assert describe_state(sensor, evaluate(sensor, {"state": "41.20"})) == "41.2 °C"
def test_numeric_without_a_unit_is_bare(self):
sensor = _numeric(unit=None)
assert describe_state(sensor, evaluate(sensor, {"state": "7"})) == "7"
class TestBlockedPrinters:
"""The interlock only ever reports a positive, current finding."""
def _manager_with(self, sensors, readings):
manager = HASensorManager()
manager._readings = readings
db = AsyncMock()
db.execute.return_value = SimpleNamespace(scalars=lambda: SimpleNamespace(all=lambda: sensors))
return manager, db
@pytest.mark.asyncio
async def test_reports_an_alerting_blocking_sensor(self):
sensor = _sensor(block_print=True)
manager, db = self._manager_with([sensor], {1: SensorReading("on", None, True, True)})
assert await manager.blocked_printers(db) == {4: "Enclosure Door"}
@pytest.mark.asyncio
async def test_silent_when_not_alerting(self):
sensor = _sensor(block_print=True)
manager, db = self._manager_with([sensor], {1: SensorReading("off", None, False, True)})
assert await manager.blocked_printers(db) == {}
@pytest.mark.asyncio
async def test_silent_when_home_assistant_is_unreachable(self):
"""The queue must keep running when HA is down, not seize up."""
sensor = _sensor(block_print=True)
manager, db = self._manager_with([sensor], {1: SensorReading(None, None, False, False)})
assert await manager.blocked_printers(db) == {}
@pytest.mark.asyncio
async def test_silent_before_the_first_poll(self):
"""A cold cache is not evidence the door is open."""
sensor = _sensor(block_print=True)
manager, db = self._manager_with([sensor], {})
assert await manager.blocked_printers(db) == {}
@pytest.mark.asyncio
async def test_names_every_blocking_sensor_on_a_printer(self):
sensors = [
_sensor(id=1, block_print=True, name="Front Door"),
_sensor(id=2, block_print=True, name="Side Panel"),
]
manager, db = self._manager_with(
sensors,
{
1: SensorReading("on", None, True, True),
2: SensorReading("on", None, True, True),
},
)
assert await manager.blocked_printers(db) == {4: "Front Door, Side Panel"}
class TestNotificationEdge:
"""Alerts fire on the transition into the alert state, not while it lasts."""
async def _apply(self, manager, sensor, states, notify):
db = AsyncMock()
db.get.return_value = SimpleNamespace(name="X1C-1")
with patch("backend.app.services.notification_service.notification_service", notify):
await manager._apply(db, [sensor], states)
@pytest.mark.asyncio
async def test_fires_once_on_the_way_in(self):
manager = HASensorManager()
sensor = _sensor(notify_on_alert=True)
notify = AsyncMock()
# First poll seeds the cache; a door already open at startup has not
# just been opened.
await self._apply(manager, sensor, {sensor.entity_id: {"state": "off"}}, notify)
assert notify.on_ha_sensor_alert.await_count == 0
await self._apply(manager, sensor, {sensor.entity_id: {"state": "on"}}, notify)
assert notify.on_ha_sensor_alert.await_count == 1
# Still open on the next pass — no second alert.
await self._apply(manager, sensor, {sensor.entity_id: {"state": "on"}}, notify)
assert notify.on_ha_sensor_alert.await_count == 1
@pytest.mark.asyncio
async def test_silent_on_the_first_poll_after_a_restart(self):
"""Cold cache. Re-announcing every pre-existing alert on every restart
is how users learn to ignore the alert."""
manager = HASensorManager()
sensor = _sensor(notify_on_alert=True)
notify = AsyncMock()
await self._apply(manager, sensor, {sensor.entity_id: {"state": "on"}}, notify)
assert notify.on_ha_sensor_alert.await_count == 0
assert manager.get_reading(sensor.id).alerting is True
@pytest.mark.asyncio
async def test_silent_when_the_sensor_opts_out(self):
manager = HASensorManager()
sensor = _sensor(notify_on_alert=False)
notify = AsyncMock()
await self._apply(manager, sensor, {sensor.entity_id: {"state": "off"}}, notify)
await self._apply(manager, sensor, {sensor.entity_id: {"state": "on"}}, notify)
assert notify.on_ha_sensor_alert.await_count == 0
@pytest.mark.asyncio
async def test_re_arms_after_the_alert_clears(self):
manager = HASensorManager()
sensor = _sensor(notify_on_alert=True)
notify = AsyncMock()
for state in ("off", "on", "off", "on"):
await self._apply(manager, sensor, {sensor.entity_id: {"state": state}}, notify)
assert notify.on_ha_sensor_alert.await_count == 2
@pytest.mark.asyncio
async def test_a_dropout_does_not_count_as_the_alert_clearing(self):
"""on -> unavailable -> on is one continuous alert, not two.
Without this, a flaky Zigbee contact would notify on every reconnect.
"""
manager = HASensorManager()
sensor = _sensor(notify_on_alert=True)
notify = AsyncMock()
await self._apply(manager, sensor, {sensor.entity_id: {"state": "off"}}, notify)
await self._apply(manager, sensor, {sensor.entity_id: {"state": "on"}}, notify)
await self._apply(manager, sensor, {sensor.entity_id: None}, notify)
await self._apply(manager, sensor, {sensor.entity_id: {"state": "on"}}, notify)
assert notify.on_ha_sensor_alert.await_count == 1
backend/tests/unit/test_scheduler_ha_interlock_1148.py
+306
View File
@@ -0,0 +1,306 @@
"""The Home Assistant sensor interlock, seen from the scheduler (#1148).
The feature exists because the reporter wanted to know his enclosure was shut
before starting a print remotely. Displaying the door state only helps if he
looks; the interlock is what makes it act on its own.
It is a *hold*, never a failure: the item stays pending and dispatches by
itself once the door closes. And it holds only on a positive, freshly read
finding — a Home Assistant that is unreachable holds nothing, because a queue
that stops whenever an unrelated service goes down is worse than the problem.
"""
from contextlib import ExitStack
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
import backend.app.models # noqa: F401 - populate Base.metadata
from backend.app.core.database import Base
from backend.app.models.library import LibraryFile
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.printer import Printer
from backend.app.services.print_scheduler import PrintScheduler
@pytest.fixture
async def queue_db():
"""Two X1Cs, so a model-based job has somewhere else to go."""
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
session_maker = async_sessionmaker(engine, expire_on_commit=False)
async with session_maker() as db:
db.add_all(
[
Printer(
id=1,
name="X1C-1",
serial_number="X1C0001",
ip_address="10.0.0.1",
access_code="x",
model="X1C",
is_active=True,
),
Printer(
id=2,
name="X1C-2",
serial_number="X1C0002",
ip_address="10.0.0.2",
access_code="x",
model="X1C",
is_active=True,
),
]
)
await db.commit()
try:
yield SimpleNamespace(session_maker=session_maker)
finally:
await engine.dispose()
async def _add_item(ctx, *, printer_id=None, target_model=None):
async with ctx.session_maker() as db:
lib = LibraryFile(
filename="job.gcode.3mf",
file_path="/library/job.gcode.3mf",
file_size=10,
file_type="gcode.3mf",
file_metadata={"sliced_for_model": "X1C"},
)
db.add(lib)
await db.flush()
item = PrintQueueItem(
status="pending",
position=1,
printer_id=printer_id,
target_model=target_model,
library_file_id=lib.id,
)
db.add(item)
await db.commit()
return item.id
async def _run(ctx, scheduler, blocked, launched, finder=None, idle=True, drying=None):
patches = [
patch("backend.app.services.print_scheduler.async_session", ctx.session_maker),
patch("backend.app.core.database.async_session", ctx.session_maker),
patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=None)),
patch(
"backend.app.services.print_scheduler.ha_sensor_manager.blocked_printers",
AsyncMock(return_value=blocked),
),
patch(
"backend.app.services.notification_service.notification_service.on_queue_job_waiting",
AsyncMock(),
),
patch(
"backend.app.services.notification_service.notification_service.on_queue_job_assigned",
AsyncMock(),
),
patch.object(scheduler, "_is_printer_idle", MagicMock(return_value=idle)),
patch.object(scheduler, "_check_auto_drying", drying or AsyncMock()),
patch.object(scheduler, "_ensure_ams_mapping", AsyncMock(return_value=None)),
patch.object(scheduler, "_block_on_filament_deficit", AsyncMock(return_value=False)),
patch.object(scheduler, "_launch_uploads", launched),
]
if finder is not None:
patches.append(patch.object(scheduler, "_find_idle_printer_for_model", finder))
with ExitStack() as stack:
for p in patches:
stack.enter_context(p)
return await scheduler.check_queue()
async def _get_item(ctx, item_id):
async with ctx.session_maker() as db:
return (await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one()
class TestFixedPrinter:
@pytest.mark.asyncio
async def test_holds_the_item_and_says_why(self, queue_db):
item_id = await _add_item(queue_db, printer_id=1)
launched = MagicMock()
await _run(queue_db, PrintScheduler(), {1: "Enclosure Door"}, launched)
launched.assert_not_called()
item = await _get_item(queue_db, item_id)
assert item.status == "pending"
assert item.waiting_reason == "Waiting on Enclosure Door"
@pytest.mark.asyncio
async def test_holding_is_not_failing(self, queue_db):
"""The door being open is a thing the user fixes in five seconds. The
job must be there waiting when they do, not failed."""
item_id = await _add_item(queue_db, printer_id=1)
await _run(queue_db, PrintScheduler(), {1: "Enclosure Door"}, MagicMock())
item = await _get_item(queue_db, item_id)
assert item.status == "pending"
assert item.error_message is None
assert item.completed_at is None
@pytest.mark.asyncio
async def test_dispatches_once_the_hold_clears(self, queue_db):
item_id = await _add_item(queue_db, printer_id=1)
scheduler = PrintScheduler()
await _run(queue_db, scheduler, {1: "Enclosure Door"}, MagicMock())
launched = MagicMock()
await _run(queue_db, scheduler, {}, launched)
launched.assert_called_once()
assert launched.call_args[0][0] == [item_id]
@pytest.mark.asyncio
async def test_the_stale_reason_is_cleared_on_dispatch(self, queue_db):
"""Otherwise the queue shows "Waiting on Enclosure Door" against a job
that is already printing."""
item_id = await _add_item(queue_db, printer_id=1)
scheduler = PrintScheduler()
await _run(queue_db, scheduler, {1: "Enclosure Door"}, MagicMock())
await _run(queue_db, scheduler, {}, MagicMock())
assert (await _get_item(queue_db, item_id)).waiting_reason is None
@pytest.mark.asyncio
async def test_the_reason_clears_even_when_the_printer_is_still_busy(self, queue_db):
"""You shut the door, but the printer is midway through something else.
The hold has lifted and the queue must say so. Clearing the reason only
at dispatch would leave a shut door reading "Waiting on Enclosure Door"
for the rest of a ten-hour print.
"""
item_id = await _add_item(queue_db, printer_id=1)
scheduler = PrintScheduler()
await _run(queue_db, scheduler, {1: "Enclosure Door"}, MagicMock())
launched = MagicMock()
await _run(queue_db, scheduler, {}, launched, idle=False)
launched.assert_not_called()
assert (await _get_item(queue_db, item_id)).waiting_reason is None
@pytest.mark.asyncio
async def test_another_printers_sensor_does_not_hold_this_one(self, queue_db):
item_id = await _add_item(queue_db, printer_id=1)
launched = MagicMock()
await _run(queue_db, PrintScheduler(), {2: "Enclosure Door"}, launched)
launched.assert_called_once()
assert launched.call_args[0][0] == [item_id]
@pytest.mark.asyncio
async def test_nothing_blocked_dispatches_as_before(self, queue_db):
item_id = await _add_item(queue_db, printer_id=1)
launched = MagicMock()
await _run(queue_db, PrintScheduler(), {}, launched)
launched.assert_called_once()
assert launched.call_args[0][0] == [item_id]
@pytest.mark.asyncio
async def test_a_held_printer_is_not_reported_as_busy(self, queue_db):
"""A held printer is idle, not printing, and the rest of the scheduler
must keep seeing it that way.
busy_printers looks like the obvious place to put a hold, but
_check_auto_drying reads that set as "is currently printing" and would
route an idle-but-held printer down the mid-print drying path — capped
temperature, and past the queue-only gating. Auto-drying must see this
printer exactly as it did before the interlock existed.
"""
await _add_item(queue_db, printer_id=1)
scheduler = PrintScheduler()
drying = AsyncMock()
await _run(queue_db, scheduler, {1: "Enclosure Door"}, MagicMock(), drying=drying)
busy_printers = drying.await_args[0][2]
assert 1 not in busy_printers
@pytest.mark.asyncio
async def test_a_failing_interlock_lookup_never_stops_the_queue(self, queue_db):
"""If the check itself breaks, the answer is "no holds", not "no prints"."""
item_id = await _add_item(queue_db, printer_id=1)
launched = MagicMock()
broken = AsyncMock(side_effect=RuntimeError("HA sensor table is on fire"))
with patch(
"backend.app.services.print_scheduler.ha_sensor_manager.blocked_printers",
broken,
):
await _run(queue_db, PrintScheduler(), {}, launched)
launched.assert_called_once()
assert launched.call_args[0][0] == [item_id]
class TestModelBased:
def _finder(self):
"""Stand-in matcher that respects busy_printers, as the real one does.
That is the whole contract under test here: the interlock folds held
printers into busy_printers, so a matcher that honours the set honours
the interlock without knowing it exists.
"""
async def finder(db, model, busy, *args, **kwargs):
for printer_id in (1, 2):
if printer_id not in busy:
return printer_id, None
return None, "All printers busy"
return finder
@pytest.mark.asyncio
async def test_an_interlocked_printer_is_passed_over_for_its_sibling(self, queue_db):
"""The whole reason the hold is folded into busy_printers: an "Any X1C"
job should run on the printer whose door is shut, not queue behind the
one whose door is open."""
item_id = await _add_item(queue_db, target_model="X1C")
launched = MagicMock()
await _run(
queue_db,
PrintScheduler(),
{1: "Enclosure Door"},
launched,
finder=self._finder(),
)
launched.assert_called_once()
assert launched.call_args[0][0] == [item_id]
assert (await _get_item(queue_db, item_id)).printer_id == 2
@pytest.mark.asyncio
async def test_every_printer_held_leaves_the_job_waiting(self, queue_db):
item_id = await _add_item(queue_db, target_model="X1C")
launched = MagicMock()
await _run(
queue_db,
PrintScheduler(),
{1: "Enclosure Door", 2: "Enclosure Door"},
launched,
finder=self._finder(),
)
launched.assert_not_called()
item = await _get_item(queue_db, item_id)
assert item.status == "pending"
assert item.printer_id is None