mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-02 20:22:15 +02:00
400 lines
14 KiB
Python
400 lines
14 KiB
Python
"""Regression tests for the webhook printer-status / stop / cancel routes.
|
|
|
|
Pre-fix the routes treated ``printer_manager.get_status(...)``'s return value
|
|
as a dict and called ``.get(...)`` on it. The return is a ``PrinterState``
|
|
dataclass (``backend/app/services/bambu_mqtt.py``), so the call raised
|
|
``AttributeError`` and surfaced as a generic 500 for any printer that
|
|
actually had a status row. See #1584.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from backend.app.services.bambu_mqtt import HMSError, PrinterState
|
|
|
|
|
|
@pytest.fixture
|
|
async def api_key_data(async_client: AsyncClient, db_session):
|
|
"""API key with read_status + control_printer scopes — covers status,
|
|
stop, and cancel in a single fixture."""
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
api_key = APIKey(
|
|
name="webhook-status-test-key",
|
|
key_hash=key_hash,
|
|
key_prefix=key_prefix,
|
|
can_read_status=True,
|
|
can_control_printer=True,
|
|
enabled=True,
|
|
)
|
|
db_session.add(api_key)
|
|
await db_session.commit()
|
|
return full_key
|
|
|
|
|
|
@pytest.fixture
|
|
async def printer_row(db_session):
|
|
from backend.app.models.printer import Printer
|
|
|
|
printer = Printer(
|
|
name="StatusTest",
|
|
ip_address="192.168.1.44",
|
|
access_code="12345678",
|
|
serial_number="00M00A000000010",
|
|
model="P1S",
|
|
)
|
|
db_session.add(printer)
|
|
await db_session.commit()
|
|
return printer
|
|
|
|
|
|
class TestWebhookGetPrinterStatus:
|
|
"""``GET /api/v1/webhook/printer/{id}/status`` — the route reads the
|
|
dataclass via attribute access, not ``.get(...)``. Pre-fix the call
|
|
raised AttributeError → 500 for every printer with a status row.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_200_with_connected_dataclass_status(
|
|
self,
|
|
async_client: AsyncClient,
|
|
api_key_data,
|
|
printer_row,
|
|
):
|
|
"""A live PrinterState dataclass must yield a 200 with the
|
|
attributes mapped into the response — this is the exact regression
|
|
from #1584 where the dataclass crashed the ``.get(...)`` calls."""
|
|
state = PrinterState(
|
|
connected=True,
|
|
state="RUNNING",
|
|
current_print="bench.3mf",
|
|
progress=42.0,
|
|
remaining_time=1234,
|
|
)
|
|
with patch(
|
|
"backend.app.api.routes.webhook.printer_manager.get_status",
|
|
MagicMock(return_value=state),
|
|
):
|
|
resp = await async_client.get(
|
|
f"/api/v1/webhook/printer/{printer_row.id}/status",
|
|
headers={"X-API-Key": api_key_data},
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["id"] == printer_row.id
|
|
assert body["name"] == "StatusTest"
|
|
assert body["connected"] is True
|
|
assert body["state"] == "RUNNING"
|
|
assert body["current_print"] == "bench.3mf"
|
|
assert body["progress"] == 42.0
|
|
assert body["remaining_time"] == 1234
|
|
assert body["remaining_seconds"] == 1234 * 60
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_200_when_status_is_none(
|
|
self,
|
|
async_client: AsyncClient,
|
|
api_key_data,
|
|
printer_row,
|
|
):
|
|
"""A registered printer the manager hasn't seen yet returns None from
|
|
``get_status``; the response must still be 200 with sensible
|
|
defaults rather than 500."""
|
|
with patch(
|
|
"backend.app.api.routes.webhook.printer_manager.get_status",
|
|
MagicMock(return_value=None),
|
|
):
|
|
resp = await async_client.get(
|
|
f"/api/v1/webhook/printer/{printer_row.id}/status",
|
|
headers={"X-API-Key": api_key_data},
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["id"] == printer_row.id
|
|
assert body["connected"] is False
|
|
assert body["state"] is None
|
|
assert body["current_print"] is None
|
|
assert body["progress"] is None
|
|
assert body["remaining_time"] is None
|
|
assert body["serial_number"] == "00M00A000000010"
|
|
assert body["remaining_seconds"] is None
|
|
assert body["layer_num"] is None
|
|
assert body["total_layers"] is None
|
|
assert body["subtask_id"] is None
|
|
assert body["hms_errors"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_404_when_printer_does_not_exist(
|
|
self,
|
|
async_client: AsyncClient,
|
|
api_key_data,
|
|
):
|
|
resp = await async_client.get(
|
|
"/api/v1/webhook/printer/99999/status",
|
|
headers={"X-API-Key": api_key_data},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
async def _status(async_client: AsyncClient, key: str, printer_id: int, state: PrinterState | None):
|
|
with patch(
|
|
"backend.app.api.routes.webhook.printer_manager.get_status",
|
|
MagicMock(return_value=state),
|
|
):
|
|
return await async_client.get(
|
|
f"/api/v1/webhook/printer/{printer_id}/status",
|
|
headers={"X-API-Key": key},
|
|
)
|
|
|
|
|
|
class TestWebhookPrinterStatusFields:
|
|
"""What an external client such as a phone Live Activity needs from the
|
|
status route, all of it already on ``PrinterState`` (#2919)."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_running_print_reports_layers_job_and_seconds(
|
|
self, async_client: AsyncClient, api_key_data, printer_row
|
|
):
|
|
state = PrinterState(
|
|
connected=True,
|
|
state="RUNNING",
|
|
current_print="bench.3mf",
|
|
progress=62.0,
|
|
remaining_time=107,
|
|
layer_num=88,
|
|
total_layers=240,
|
|
subtask_id="512345678",
|
|
)
|
|
body = (await _status(async_client, api_key_data, printer_row.id, state)).json()
|
|
|
|
assert body["serial_number"] == "00M00A000000010"
|
|
assert body["layer_num"] == 88
|
|
assert body["total_layers"] == 240
|
|
assert body["subtask_id"] == "512345678"
|
|
# remaining_time stays in minutes for existing clients.
|
|
assert body["remaining_time"] == 107
|
|
assert body["remaining_seconds"] == 6420
|
|
assert body["hms_errors"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_paused_print_says_why(self, async_client: AsyncClient, api_key_data, printer_row):
|
|
"""A runout pause carries its HMS fault; a user pause carries none."""
|
|
runout = HMSError(
|
|
code="0x20008",
|
|
attr=0x07008000,
|
|
module=0x07,
|
|
severity=2,
|
|
description="Filament has run out.",
|
|
actions=["RESUME_PRINTING"],
|
|
job_id="512345678",
|
|
full_code="0700800000020008",
|
|
)
|
|
state = PrinterState(connected=True, state="PAUSE", subtask_id="512345678", hms_errors=[runout])
|
|
body = (await _status(async_client, api_key_data, printer_row.id, state)).json()
|
|
|
|
assert body["hms_errors"] == [
|
|
{
|
|
"code": "0x20008",
|
|
"attr": 0x07008000,
|
|
"module": 0x07,
|
|
"severity": 2,
|
|
"actions": ["RESUME_PRINTING"],
|
|
"job_id": "512345678",
|
|
"full_code": "0700800000020008",
|
|
"description": "Filament has run out.",
|
|
}
|
|
]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_hms_errors_match_the_printer_status_route(
|
|
self, async_client: AsyncClient, api_key_data, printer_row
|
|
):
|
|
"""One shape for a fault, whether the UI or an API key asks."""
|
|
from backend.app.api.routes.printers import hms_error_responses as used_by_printers_route
|
|
|
|
fault = HMSError(code="0x1", attr=0x0300_0100, module=0x03, severity=1, full_code="0300010000000001")
|
|
state = PrinterState(connected=True, state="FAILED", hms_errors=[fault])
|
|
body = (await _status(async_client, api_key_data, printer_row.id, state)).json()
|
|
|
|
assert body["hms_errors"] == [e.model_dump() for e in used_by_printers_route([fault])]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_numeric_subtask_id_is_returned_as_text(self, async_client: AsyncClient, api_key_data, printer_row):
|
|
"""Stored as the printer sent it; a number must not 500 the poll."""
|
|
state = PrinterState(connected=True, state="RUNNING", subtask_id=512345678)
|
|
resp = await _status(async_client, api_key_data, printer_row.id, state)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["subtask_id"] == "512345678"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize("raw", ["0", "", " ", 0])
|
|
async def test_a_job_without_an_id_reports_null(self, async_client: AsyncClient, api_key_data, printer_row, raw):
|
|
"""Bambu reports "0" or "" for local prints. Every such print would
|
|
share the same "id", so it must not be handed out as one."""
|
|
state = PrinterState(connected=True, state="RUNNING", subtask_id=raw)
|
|
resp = await _status(async_client, api_key_data, printer_row.id, state)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["subtask_id"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_idle_printer_reports_zero_not_null(self, async_client: AsyncClient, api_key_data, printer_row):
|
|
"""A connected idle printer has real zeros; null is kept for "no status yet"."""
|
|
state = PrinterState(connected=True, state="IDLE")
|
|
body = (await _status(async_client, api_key_data, printer_row.id, state)).json()
|
|
|
|
assert body["remaining_time"] == 0
|
|
assert body["remaining_seconds"] == 0
|
|
assert body["layer_num"] == 0
|
|
assert body["total_layers"] == 0
|
|
assert body["subtask_id"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_needs_the_read_status_scope(self, async_client: AsyncClient, db_session, printer_row):
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
db_session.add(
|
|
APIKey(
|
|
name="no-status",
|
|
key_hash=key_hash,
|
|
key_prefix=key_prefix,
|
|
can_read_status=False,
|
|
can_queue=True,
|
|
enabled=True,
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
resp = await _status(async_client, full_key, printer_row.id, PrinterState(connected=True))
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_key_for_another_printer_sees_nothing(self, async_client: AsyncClient, db_session, printer_row):
|
|
"""The serial and faults are only for keys allowed on this printer."""
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
db_session.add(
|
|
APIKey(
|
|
name="other-printer",
|
|
key_hash=key_hash,
|
|
key_prefix=key_prefix,
|
|
can_read_status=True,
|
|
printer_ids=[printer_row.id + 1000],
|
|
enabled=True,
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
resp = await _status(async_client, full_key, printer_row.id, PrinterState(connected=True))
|
|
assert resp.status_code == 403
|
|
assert "00M00A000000010" not in resp.text
|
|
|
|
|
|
class TestWebhookStopPrint:
|
|
"""``POST /api/v1/webhook/printer/{id}/stop`` — same dataclass-shape
|
|
fix applies to the connection / state precondition checks (#1584)."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_503_when_disconnected(
|
|
self,
|
|
async_client: AsyncClient,
|
|
api_key_data,
|
|
printer_row,
|
|
):
|
|
state = PrinterState(connected=False, state="unknown")
|
|
with patch(
|
|
"backend.app.api.routes.webhook.printer_manager.get_status",
|
|
MagicMock(return_value=state),
|
|
):
|
|
resp = await async_client.post(
|
|
f"/api/v1/webhook/printer/{printer_row.id}/stop",
|
|
headers={"X-API-Key": api_key_data},
|
|
)
|
|
# Pre-fix this would have 500'd on `status.get(...)`. Now it
|
|
# cleanly returns the documented 503.
|
|
assert resp.status_code == 503
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_409_when_not_running(
|
|
self,
|
|
async_client: AsyncClient,
|
|
api_key_data,
|
|
printer_row,
|
|
):
|
|
state = PrinterState(connected=True, state="FINISH")
|
|
with patch(
|
|
"backend.app.api.routes.webhook.printer_manager.get_status",
|
|
MagicMock(return_value=state),
|
|
):
|
|
resp = await async_client.post(
|
|
f"/api/v1/webhook/printer/{printer_row.id}/stop",
|
|
headers={"X-API-Key": api_key_data},
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
class TestWebhookCancelPrint:
|
|
"""``POST /api/v1/webhook/printer/{id}/cancel`` — same fix shape."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_503_when_disconnected(
|
|
self,
|
|
async_client: AsyncClient,
|
|
api_key_data,
|
|
printer_row,
|
|
):
|
|
state = PrinterState(connected=False, state="unknown")
|
|
with patch(
|
|
"backend.app.api.routes.webhook.printer_manager.get_status",
|
|
MagicMock(return_value=state),
|
|
):
|
|
resp = await async_client.post(
|
|
f"/api/v1/webhook/printer/{printer_row.id}/cancel",
|
|
headers={"X-API-Key": api_key_data},
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_409_when_not_running_or_paused(
|
|
self,
|
|
async_client: AsyncClient,
|
|
api_key_data,
|
|
printer_row,
|
|
):
|
|
state = PrinterState(connected=True, state="IDLE")
|
|
with patch(
|
|
"backend.app.api.routes.webhook.printer_manager.get_status",
|
|
MagicMock(return_value=state),
|
|
):
|
|
resp = await async_client.post(
|
|
f"/api/v1/webhook/printer/{printer_row.id}/cancel",
|
|
headers={"X-API-Key": api_key_data},
|
|
)
|
|
assert resp.status_code == 409
|