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

Merge remote-tracking branch 'origin/main' into 0.2.4.2

Browse Source
This commit is contained in:
maziggy
2026-05-19 14:26:00 +02:00
parent 162db57923 7d3c01293b
commit 58ac4b3b81
5 changed files with 124 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/updates.py
+9
View File
@@ -619,12 +619,21 @@ async def _perform_update(target_ref: str):
# locally resolvable for the reset below. `--tags` is required —
# plain `git fetch origin` doesn't bring tags by default, so a
# release tag would not be resolvable.
#
# `--force` lets a moved tag on the remote overwrite the local copy.
# Without it, any tag that was re-tagged upstream (e.g. v0.2.1 being
# re-pointed after a hotfix re-tag) makes `git fetch --tags` return
# a non-zero exit even though every other ref fetched cleanly —
# which we'd then surface as "Failed to fetch updates" to the user.
# The in-app updater's contract is "sync me to the remote"; force-
# overwriting a stale local tag matches that intent.
process = await asyncio.create_subprocess_exec(
git_path,
*git_config,
"fetch",
"--prune",
"--tags",
"--force",
"origin",
cwd=str(base_dir),
stdout=asyncio.subprocess.PIPE,
backend/app/main.py
+31 -40
View File
@@ -627,6 +627,31 @@ def _get_start_ams_mapping(data: dict, archive_id: int | None) -> list[int] | No
return stored_ams_mapping
def _maybe_start_layer_timelapse(printer, printer_id: int, archive_id: int) -> bool:
"""Start a layer-timelapse session for *archive_id* when the printer has
an external camera configured. Returns True if a session was started.
Three call sites in on_print_start (expected-archive promotion, fallback
archive creation, fresh-archive creation) used to inline this same
if-block; the inline copies kept drifting (#1353 fixed only one of them
on the first pass). Centralising the conditional + call here makes the
contract testable in isolation and keeps the three sites locked in step.
"""
if not (printer.external_camera_enabled and printer.external_camera_url):
return False
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
archive_id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logging.getLogger(__name__).info("Started layer timelapse for printer %s, archive %s", printer_id, archive_id)
return True
def _format_hms_error_summary(hms_errors: list[dict]) -> str | None:
"""Build a human-readable failure reason from MQTT hms_errors for PrintQueueItem.error_message.
@@ -2033,22 +2058,10 @@ async def on_print_start(printer_id: int, data: dict):
_active_prints[(printer_id, f"{subtask_name}.3mf")] = archive.id
# Start timelapse session if external camera is enabled (#1353).
# The two new-archive paths below also call start_session, but
# queue / VP-dispatched prints land here in the expected-archive
# branch and used to skip it entirely — so the timelapse session
# never started, no frames were captured, and the post-print
# stitch silently returned None.
if printer.external_camera_enabled and printer.external_camera_url:
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
archive.id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logger.info("Started layer timelapse for printer %s, expected archive %s", printer_id, archive.id)
# Queue / VP-dispatched prints land here in the expected-archive
# branch and used to skip start_session entirely — frames were
# never captured and the post-print stitch silently returned None.
_maybe_start_layer_timelapse(printer, printer_id, archive.id)
# Inject ams_mapping into usage tracker session — the session was created
# before expected-print promotion, so it may have ams_mapping=None when
@@ -2562,18 +2575,7 @@ async def on_print_start(printer_id: int, data: dict):
logger.info("Created fallback archive %s for %s (no 3MF available)", fallback_archive.id, print_name)
# Start timelapse session if external camera is enabled
if printer.external_camera_enabled and printer.external_camera_url:
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
fallback_archive.id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logger.info("Started layer timelapse for printer %s, archive %s", printer_id, fallback_archive.id)
_maybe_start_layer_timelapse(printer, printer_id, fallback_archive.id)
# Track as active print
_active_prints[(printer_id, fallback_archive.filename)] = fallback_archive.id
@@ -2652,18 +2654,7 @@ async def on_print_start(printer_id: int, data: dict):
logger.info("Created archive %s for %s", archive.id, downloaded_filename)
# Start timelapse session if external camera is enabled
if printer.external_camera_enabled and printer.external_camera_url:
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
archive.id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logger.info("Started layer timelapse for printer %s, archive %s", printer_id, archive.id)
_maybe_start_layer_timelapse(printer, printer_id, archive.id)
# Record starting energy from smart plug if available (#941: persisted column)
await _record_energy_start(archive, printer_id, db, context="auto-archive")
backend/tests/integration/test_updates_api.py
+10
View File
@@ -443,6 +443,16 @@ class TestUpdatesAPI:
"for tag-based updates) are resolvable for the subsequent reset. "
f"Captured fetch call: {fetch_calls[0]['args']}"
)
# Fetch must include --force so a re-pointed tag on the remote
# (common after re-tagging a release post-release-notes edit) doesn't
# surface as "Failed to fetch updates" to the user just because their
# local copy of the moved tag would be clobbered. The relevant target
# ref is fetched fine; we only want git's tag-clobber to be silent.
assert "--force" in fetch_calls[0]["args"], (
"Fetch must use --force so re-pointed tags on the remote don't "
"fail the whole fetch (the rest of the refs update cleanly). "
f"Captured fetch call: {fetch_calls[0]['args']}"
)
@pytest.mark.asyncio
async def test_apply_update_passes_discovered_release_to_perform_update(self, async_client: AsyncClient):
backend/tests/unit/test_layer_timelapse_expected_archive.py
+72 -186
View File
@@ -12,212 +12,98 @@ The expected-archive branch — where reprints and queue dispatch land —
updated the existing archive's status to "printing" but never started a
timelapse session.
Fix: start_session is now called in the expected-archive branch too, guarded
by the same `external_camera_enabled and external_camera_url` check that
the other two paths use.
Fix: the three start_session call sites in on_print_start were unified
behind `_maybe_start_layer_timelapse(printer, printer_id, archive_id)`,
which gates on the same `external_camera_enabled and external_camera_url`
check. Testing the helper directly (instead of driving the whole
on_print_start flow) keeps this regression locked in without dragging in
unrelated side effects (plate detection, DB queries, MQTT relay, etc.).
"""
from unittest.mock import AsyncMock, MagicMock, patch
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from backend.app.main import (
_active_prints,
_expected_print_creators,
_expected_print_registered_at,
_expected_prints,
_print_ams_mappings,
register_expected_print,
)
from backend.app.main import _maybe_start_layer_timelapse
@pytest.fixture(autouse=True)
def _clear_dicts():
"""Clear module-level tracking dicts before and after each test."""
_expected_prints.clear()
_expected_print_registered_at.clear()
_expected_print_creators.clear()
_print_ams_mappings.clear()
_active_prints.clear()
yield
_expected_prints.clear()
_expected_print_registered_at.clear()
_expected_print_creators.clear()
_print_ams_mappings.clear()
_active_prints.clear()
def _build_mocks(*, external_camera_enabled: bool, external_camera_url: str | None):
"""Construct the mock matrix needed to drive on_print_start through the
expected-archive branch. Returns a dict of mock contexts that the test
enters via contextlib.ExitStack.
The session.execute mock returns the printer for the first call (printer
lookup) and the archive row for the second call (expected-archive
re-fetch). The archive row carries a unique filename so the
expected-print key lookup succeeds.
def _make_printer(*, external_camera_enabled: bool, external_camera_url: str | None):
"""Construct a minimal printer-shaped object with exactly the attributes
the helper reads. SimpleNamespace is used over MagicMock so attribute
access raises AttributeError on anything unexpected — keeps the test
honest about which fields the helper actually depends on.
"""
mock_printer = MagicMock()
mock_printer.id = 1
mock_printer.auto_archive = True
mock_printer.external_camera_enabled = external_camera_enabled
mock_printer.external_camera_url = external_camera_url
mock_printer.external_camera_type = "snapshot"
mock_printer.external_camera_snapshot_url = external_camera_url
mock_printer.name = "TestA1"
mock_archive = MagicMock()
mock_archive.id = 42
mock_archive.filename = "Universal_Spirit_level_Holder.3mf"
mock_archive.subtask_id = None
mock_archive.print_time_seconds = None
mock_archive.created_by_id = None
mock_archive.printer_id = 1
mock_archive.print_name = "Universal Spirit Level Holder"
mock_archive.status = "pending"
mock_archive.file_path = "/tmp/fake.3mf"
return mock_printer, mock_archive
@pytest.mark.asyncio
async def test_expected_archive_path_starts_timelapse_when_external_camera_enabled():
"""Queue/VP-dispatched prints land in the expected-archive branch and must
start the timelapse session there (the #1353 root cause)."""
mock_printer, mock_archive = _build_mocks(
external_camera_enabled=True, external_camera_url="http://camera.local:5000/snapshot.jpg"
return SimpleNamespace(
external_camera_enabled=external_camera_enabled,
external_camera_url=external_camera_url,
external_camera_type="snapshot",
external_camera_snapshot_url=external_camera_url,
)
# Register the expected print so the dispatch flow finds an archive_id.
register_expected_print(1, "Universal_Spirit_level_Holder.3mf", archive_id=42, ams_mapping=[1])
# on_print_start fires many db.execute() calls (settings lookups,
# usage tracker, plate detection, etc) before reaching the expected-
# archive branch. Route on SQL text so each query gets a sensible
# response regardless of order, rather than queuing N mocks.
def execute_router(stmt, *args, **kwargs):
sql = str(stmt).lower()
if "from printers" in sql or "from printer " in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_printer),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_printer]))),
)
if "from print_archives" in sql or "from print_archive" in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_archive),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_archive]))),
)
# Settings, spool assignments, anything else — return empty.
return MagicMock(
scalar_one_or_none=MagicMock(return_value=None),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
)
def test_starts_timelapse_when_external_camera_enabled():
"""Queue/VP-dispatched prints land in the expected-archive branch and must
start the timelapse session there (the #1353 root cause). The helper is
called from all three on_print_start paths (expected-archive promotion,
fallback archive, fresh archive) so testing it once covers all three."""
printer = _make_printer(
external_camera_enabled=True,
external_camera_url="http://camera.local:5000/snapshot.jpg",
)
mock_session = AsyncMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock()
mock_session.execute = AsyncMock(side_effect=execute_router)
mock_session.commit = AsyncMock()
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
started = _maybe_start_layer_timelapse(printer, printer_id=1, archive_id=42)
with (
patch("backend.app.main.async_session") as mock_session_maker,
patch("backend.app.main.notification_service") as mock_notif,
patch("backend.app.main.smart_plug_manager") as mock_plug,
patch("backend.app.main.ws_manager") as mock_ws,
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.mqtt_relay") as mock_relay,
patch("backend.app.main._record_energy_start", new_callable=AsyncMock),
patch("backend.app.main._load_objects_from_archive"),
patch("backend.app.main._store_spoolman_print_data", new_callable=AsyncMock),
patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock),
# The actual subject under test: assert start_session is called.
patch("backend.app.services.layer_timelapse.start_session") as mock_start_session,
):
mock_session_maker.return_value = mock_session
mock_notif.on_print_start = AsyncMock()
mock_plug.on_print_start = AsyncMock()
mock_ws.send_print_start = AsyncMock()
mock_ws.send_archive_updated = AsyncMock()
mock_relay.on_print_start = AsyncMock()
mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
from backend.app.main import on_print_start
await on_print_start(
1,
{
"filename": "Universal_Spirit_level_Holder.3mf",
"subtask_name": "Universal_Spirit_level_Holder",
},
)
mock_start_session.assert_called_once()
# Verify it was called with the archive_id from the expected-print
# registration, not a fresh one — that's the contract.
call_args = mock_start_session.call_args
assert call_args.args[0] == 1, "printer_id must match"
assert call_args.args[1] == 42, "archive_id must come from the expected-print registration"
assert call_args.args[2] == "http://camera.local:5000/snapshot.jpg"
assert call_args.args[3] == "snapshot"
assert started is True
mock_start_session.assert_called_once_with(
1,
42,
"http://camera.local:5000/snapshot.jpg",
"snapshot",
snapshot_url="http://camera.local:5000/snapshot.jpg",
)
@pytest.mark.asyncio
async def test_expected_archive_path_skips_timelapse_when_external_camera_disabled():
def test_skips_timelapse_when_external_camera_disabled():
"""The same guard that the new-archive paths use must hold here: no
external camera → no timelapse session. Otherwise we'd try to capture
from a None URL and crash the print-start flow."""
mock_printer, mock_archive = _build_mocks(external_camera_enabled=False, external_camera_url=None)
printer = _make_printer(external_camera_enabled=False, external_camera_url=None)
mock_archive.filename = "test.3mf"
mock_archive.id = 99
register_expected_print(1, "test.3mf", archive_id=99, ams_mapping=None)
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
started = _maybe_start_layer_timelapse(printer, printer_id=1, archive_id=99)
def execute_router(stmt, *args, **kwargs):
sql = str(stmt).lower()
if "from printers" in sql or "from printer " in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_printer),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_printer]))),
)
if "from print_archives" in sql or "from print_archive" in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_archive),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_archive]))),
)
return MagicMock(
scalar_one_or_none=MagicMock(return_value=None),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
)
assert started is False
mock_start_session.assert_not_called()
mock_session = AsyncMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock()
mock_session.execute = AsyncMock(side_effect=execute_router)
mock_session.commit = AsyncMock()
with (
patch("backend.app.main.async_session") as mock_session_maker,
patch("backend.app.main.notification_service") as mock_notif,
patch("backend.app.main.smart_plug_manager") as mock_plug,
patch("backend.app.main.ws_manager") as mock_ws,
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.mqtt_relay") as mock_relay,
patch("backend.app.main._record_energy_start", new_callable=AsyncMock),
patch("backend.app.main._load_objects_from_archive"),
patch("backend.app.main._store_spoolman_print_data", new_callable=AsyncMock),
patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock),
patch("backend.app.services.layer_timelapse.start_session") as mock_start_session,
):
mock_session_maker.return_value = mock_session
mock_notif.on_print_start = AsyncMock()
mock_plug.on_print_start = AsyncMock()
mock_ws.send_print_start = AsyncMock()
mock_ws.send_archive_updated = AsyncMock()
mock_relay.on_print_start = AsyncMock()
mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
def test_skips_timelapse_when_url_missing_even_if_flag_set():
"""If the toggle is on but the URL field is empty (legacy / half-configured
install), the guard must still hold — calling start_session with an empty
URL would crash downstream when the capture thread tries to fetch frames."""
printer = _make_printer(external_camera_enabled=True, external_camera_url=None)
from backend.app.main import on_print_start
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
started = _maybe_start_layer_timelapse(printer, printer_id=1, archive_id=7)
await on_print_start(1, {"filename": "test.3mf", "subtask_name": "test"})
assert started is False
mock_start_session.assert_not_called()
mock_start_session.assert_not_called()
def test_camera_type_defaults_to_mjpeg_when_unset():
"""external_camera_type defaults to 'mjpeg' in start_session when the
printer column is None — pre-existing contract preserved by the helper."""
printer = SimpleNamespace(
external_camera_enabled=True,
external_camera_url="http://cam/feed",
external_camera_type=None,
external_camera_snapshot_url=None,
)
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
_maybe_start_layer_timelapse(printer, printer_id=2, archive_id=11)
assert mock_start_session.called
call_kwargs = mock_start_session.call_args.kwargs
call_args = mock_start_session.call_args.args
assert call_args[3] == "mjpeg"
assert call_kwargs["snapshot_url"] is None
requirements.txt
+2
View File
@@ -11,6 +11,8 @@ greenlet>=3.0.0
# Pydantic
pydantic>=2.0.0
pydantic-settings>=2.0.0
# Transitive of pydantic-settings, floor-pinned to patch CVE-2026-28684 (dotenv 1.2.1)
python-dotenv>=1.2.2
# Bambu Lab Printer Communication
paho-mqtt>=2.0.0