Files
maziggy 83142c726c fix(slice): report a finished slice once, not once per queued poll
setInterval does not await an async callback. Slicing a large project
blocks the backend for seconds, so poll ticks piled up behind one stalled
request, each holding a snapshot taken while the job was still active.
They resolved together, and every one of them ran the completion path —
one toast and two query invalidations each. A 20s stall against the 1.5s
interval produced 13 "Sliced X" toasts from a single slice.

Only one poll round is now in flight at a time, which also stops queueing
requests against a backend that is already saturated. Completion is
recorded once per job id, and a round still awaiting a response when the
effect tears down now returns instead of acting.
2026-07-30 17:14:07 +02:00

496 lines
23 KiB
Python

"""Tests for the per-slice 3MF input normalisation helpers."""
from __future__ import annotations
import json
import zipfile
from io import BytesIO
from backend.app.services.slicer_3mf_convert import (
count_plates_in_3mf,
extract_source_printer_model,
merge_plate_3mfs,
substitute_unused_plate_filaments,
)
def _make_3mf(entries: dict[str, bytes]) -> bytes:
buf = BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
for name, payload in entries.items():
zf.writestr(name, payload)
return buf.getvalue()
class TestExtractSourcePrinterModel:
def test_returns_canonical_short_code_for_x1c(self):
# Raw field is the long display name; we need the short code so
# is_dual_nozzle_model() matches against the model registry.
cfg = json.dumps({"printer_model": "Bambu Lab X1 Carbon", "other": "field"}).encode()
zip_bytes = _make_3mf({"Metadata/project_settings.config": cfg})
assert extract_source_printer_model(zip_bytes) == "X1C"
def test_returns_canonical_short_code_for_h2d(self):
cfg = json.dumps({"printer_model": "Bambu Lab H2D"}).encode()
zip_bytes = _make_3mf({"Metadata/project_settings.config": cfg})
assert extract_source_printer_model(zip_bytes) == "H2D"
def test_dual_nozzle_check_works_on_extracted_code(self):
"""The whole point of canonicalising in this helper: the result
must feed straight into is_dual_nozzle_model() without further
normalisation."""
from backend.app.utils.printer_models import is_dual_nozzle_model
h2d = _make_3mf({"Metadata/project_settings.config": json.dumps({"printer_model": "Bambu Lab H2D"}).encode()})
x1c = _make_3mf(
{"Metadata/project_settings.config": json.dumps({"printer_model": "Bambu Lab X1 Carbon"}).encode()}
)
assert is_dual_nozzle_model(extract_source_printer_model(h2d)) is True
assert is_dual_nozzle_model(extract_source_printer_model(x1c)) is False
def test_returns_none_when_field_missing(self):
cfg = json.dumps({"other": "field"}).encode()
zip_bytes = _make_3mf({"Metadata/project_settings.config": cfg})
assert extract_source_printer_model(zip_bytes) is None
def test_returns_none_when_field_empty(self):
cfg = json.dumps({"printer_model": ""}).encode()
zip_bytes = _make_3mf({"Metadata/project_settings.config": cfg})
assert extract_source_printer_model(zip_bytes) is None
def test_returns_none_when_no_embedded_config(self):
zip_bytes = _make_3mf({"Metadata/other.txt": b"hello"})
assert extract_source_printer_model(zip_bytes) is None
def test_returns_none_for_non_zip_bytes(self):
assert extract_source_printer_model(b"not a zip") is None
def test_returns_none_for_malformed_json(self):
zip_bytes = _make_3mf({"Metadata/project_settings.config": b"{not json"})
assert extract_source_printer_model(zip_bytes) is None
def test_returns_none_when_config_is_list_not_dict(self):
cfg = json.dumps(["not", "a", "dict"]).encode()
zip_bytes = _make_3mf({"Metadata/project_settings.config": cfg})
assert extract_source_printer_model(zip_bytes) is None
class TestCountPlatesIn3mf:
def test_counts_plater_id_entries(self):
xml = (
b'<?xml version="1.0"?>\n<config>\n'
b'<plate><metadata key="plater_id" value="1"/></plate>\n'
b'<plate><metadata key="plater_id" value="2"/></plate>\n'
b'<plate><metadata key="plater_id" value="3"/></plate>\n'
b"</config>\n"
)
zip_bytes = _make_3mf({"Metadata/model_settings.config": xml})
assert count_plates_in_3mf(zip_bytes) == 3
def test_returns_zero_for_no_model_settings(self):
zip_bytes = _make_3mf({"3D/3dmodel.model": b"<model/>"})
assert count_plates_in_3mf(zip_bytes) == 0
def test_returns_zero_for_non_zip(self):
assert count_plates_in_3mf(b"not a zip") == 0
def test_returns_zero_when_no_plate_ids(self):
zip_bytes = _make_3mf({"Metadata/model_settings.config": b"<config/>"})
assert count_plates_in_3mf(zip_bytes) == 0
class TestMergePlate3mfs:
"""Per-plate cross-class loop output → merged multi-plate 3MF. The
merge needs to: (1) carry forward the first plate's base metadata
(project_settings, model_settings, 3dmodel), (2) overlay each
plate's gcode + thumbnails, (3) re-assemble slice_info.config to
list every plate."""
@staticmethod
def _single_plate_3mf(plate_num: int, gcode_bytes: bytes, slice_info_block: str | None = None) -> bytes:
slice_info = (
'<?xml version="1.0" encoding="UTF-8"?>\n<config>\n'
'<header><header_item key="X-BBL-Client-Type" value="slicer"/></header>\n'
+ (slice_info_block or f'<plate><metadata key="index" value="{plate_num}"/></plate>')
+ "\n</config>\n"
).encode("utf-8")
return _make_3mf(
{
"3D/3dmodel.model": f"<model plate={plate_num}/>".encode(),
"Metadata/project_settings.config": b'{"printer_model": "Bambu Lab H2D"}',
"Metadata/model_settings.config": b"<config/>",
"Metadata/slice_info.config": slice_info,
f"Metadata/plate_{plate_num}.gcode": gcode_bytes,
f"Metadata/plate_{plate_num}.gcode.md5": b"d41d8cd98f00b204e9800998ecf8427e",
f"Metadata/plate_{plate_num}.json": b"{}",
f"Metadata/plate_{plate_num}.png": b"PLATE_PNG",
f"Metadata/plate_{plate_num}_small.png": b"SMALL",
f"Metadata/top_{plate_num}.png": b"TOP",
f"Metadata/pick_{plate_num}.png": b"PICK",
}
)
def test_empty_input_raises(self):
import pytest as _pytest
with _pytest.raises(ValueError):
merge_plate_3mfs([])
def test_single_plate_is_passthrough(self):
only = self._single_plate_3mf(1, b"GCODE-1")
assert merge_plate_3mfs([(1, only)]) == only
def test_overlays_per_plate_artifacts(self):
p1 = self._single_plate_3mf(1, b"GCODE-PLATE-1")
p2 = self._single_plate_3mf(2, b"GCODE-PLATE-2")
p3 = self._single_plate_3mf(3, b"GCODE-PLATE-3")
merged = merge_plate_3mfs([(1, p1), (2, p2), (3, p3)])
with zipfile.ZipFile(BytesIO(merged), "r") as zf:
assert zf.read("Metadata/plate_1.gcode") == b"GCODE-PLATE-1"
assert zf.read("Metadata/plate_2.gcode") == b"GCODE-PLATE-2"
assert zf.read("Metadata/plate_3.gcode") == b"GCODE-PLATE-3"
# Per-plate thumbnails and json overlaid too.
assert zf.read("Metadata/plate_2.png") == b"PLATE_PNG"
assert zf.read("Metadata/plate_3_small.png") == b"SMALL"
# Base 3MF's project_settings.config carried forward unchanged.
assert zf.read("Metadata/project_settings.config") == p1_project(p1)
def test_combined_slice_info_lists_every_plate(self):
p1 = self._single_plate_3mf(1, b"G1", slice_info_block='<plate><metadata key="index" value="1"/></plate>')
p2 = self._single_plate_3mf(2, b"G2", slice_info_block='<plate><metadata key="index" value="2"/></plate>')
merged = merge_plate_3mfs([(1, p1), (2, p2)])
with zipfile.ZipFile(BytesIO(merged), "r") as zf:
info = zf.read("Metadata/slice_info.config").decode("utf-8")
# Both plate blocks present.
assert 'value="1"' in info
assert 'value="2"' in info
# Two <plate> blocks total (we don't include the source's stale
# one from before slicing).
assert info.count("<plate>") == 2
def test_falls_back_to_source_thumbnails_when_sliced_outputs_lack_them(self):
"""BS CLI with --arrange generates fresh per-plate gcode but
doesn't always write a fresh ``plate_N.png``. The merger's
``source_3mf_bytes`` fallback should fill the gap from the
source 3MF's original per-plate render so the archive's per-
plate previews aren't blank."""
# Sliced outputs that lack plate_N.png entries entirely (only
# gcode + json + md5 — the thumbnail slot is empty).
def _no_thumb_3mf(plate_num: int) -> bytes:
return _make_3mf(
{
"3D/3dmodel.model": b"<model/>",
"Metadata/project_settings.config": b"{}",
"Metadata/model_settings.config": b"<config/>",
"Metadata/slice_info.config": (
'<?xml version="1.0"?>\n<config>'
f'<plate><metadata key="index" value="{plate_num}"/></plate>'
"</config>"
).encode(),
f"Metadata/plate_{plate_num}.gcode": f"G{plate_num}".encode(),
}
)
source = _make_3mf(
{
"3D/3dmodel.model": b"<model/>",
"Metadata/plate_1.png": b"SRC_PNG_1",
"Metadata/plate_1_small.png": b"SRC_SMALL_1",
"Metadata/plate_2.png": b"SRC_PNG_2",
"Metadata/plate_2_small.png": b"SRC_SMALL_2",
}
)
merged = merge_plate_3mfs(
[(1, _no_thumb_3mf(1)), (2, _no_thumb_3mf(2))],
source_3mf_bytes=source,
)
with zipfile.ZipFile(BytesIO(merged), "r") as zf:
assert zf.read("Metadata/plate_1.png") == b"SRC_PNG_1"
assert zf.read("Metadata/plate_1_small.png") == b"SRC_SMALL_1"
assert zf.read("Metadata/plate_2.png") == b"SRC_PNG_2"
assert zf.read("Metadata/plate_2_small.png") == b"SRC_SMALL_2"
def test_source_fallback_does_not_overwrite_fresh_sliced_thumbnails(self):
"""If a sliced output DID write its own ``plate_N.png`` (same-class
slice / older BS that renders even with arrange), keep it — the
sliced render reflects the actual H2D layout; the source fallback
only fills gaps."""
p1 = self._single_plate_3mf(1, b"G1") # has plate_1.png = PLATE_PNG
p2 = self._single_plate_3mf(2, b"G2") # has plate_2.png = PLATE_PNG
source = _make_3mf(
{
"Metadata/plate_1.png": b"SRC_PNG_1",
"Metadata/plate_2.png": b"SRC_PNG_2",
}
)
merged = merge_plate_3mfs([(1, p1), (2, p2)], source_3mf_bytes=source)
with zipfile.ZipFile(BytesIO(merged), "r") as zf:
# Sliced output's thumbnails win.
assert zf.read("Metadata/plate_1.png") == b"PLATE_PNG"
assert zf.read("Metadata/plate_2.png") == b"PLATE_PNG"
def test_unsorted_input_is_sorted_by_plate_number(self):
p1 = self._single_plate_3mf(1, b"G1")
p2 = self._single_plate_3mf(2, b"G2")
# Pass them out of order; the merger should still place plate 2's
# artifacts at plate_2.* and plate 1's at plate_1.*.
merged = merge_plate_3mfs([(2, p2), (1, p1)])
with zipfile.ZipFile(BytesIO(merged), "r") as zf:
assert zf.read("Metadata/plate_1.gcode") == b"G1"
assert zf.read("Metadata/plate_2.gcode") == b"G2"
def p1_project(zip_bytes: bytes) -> bytes:
"""Helper for the merge test — pulls plate-1's project_settings.config out
of a fixture so the test's assertion shows the actual reference value
rather than hard-coding the literal."""
with zipfile.ZipFile(BytesIO(zip_bytes), "r") as zf:
return zf.read("Metadata/project_settings.config")
class TestSubstituteUnusedPlateFilaments:
"""Unused-slot entries are overwritten with the plate's lowest *used*
slot so BambuStudio's validators don't trip on loaded filaments the
plate's G-code never actually touches — neither the filament-temp
spread nor the preset-vs-printer compatibility check."""
@staticmethod
def _model_settings_xml(per_plate_extruders: list[tuple[int, list[int]]]) -> bytes:
"""Build a minimal model_settings.config mapping each plate to a set
of extruder/slot numbers via per-object extruder metadata. Mirrors
the schema ``extract_plate_extruder_set_from_3mf`` parses:
- top-level ``<object id=N>`` with ``<metadata key="extruder" .../>``
- per-plate ``<plate>`` listing the object ids it contains.
``per_plate_extruders`` is a list of (plate_id, [extruder_ids]).
Object ids are auto-numbered globally so plates can reference them.
"""
objects = []
plates = []
oid = 1
for plate_id, exts in per_plate_extruders:
plate_obj_refs = []
for ext in exts:
objects.append(f'<object id="{oid}"><metadata key="extruder" value="{ext}"/></object>')
plate_obj_refs.append(
f'<model_instance><metadata key="object_id" value="{oid}"/>'
f'<metadata key="instance_id" value="0"/>'
f'<metadata key="identify_id" value="{oid}"/></model_instance>'
)
oid += 1
plates.append(
f'<plate><metadata key="plater_id" value="{plate_id}"/>' + "".join(plate_obj_refs) + "</plate>"
)
xml = (
'<?xml version="1.0" encoding="UTF-8"?>\n'
"<config>\n" + "\n".join(objects) + "\n" + "\n".join(plates) + "\n" + "</config>"
)
return xml.encode("utf-8")
def test_substitutes_unused_slot_with_slot_1(self):
# Plate 1 uses slot 1 only; slots 2 and 3 are loaded but unused.
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1])])})
items = ["pla_basic.json", "abs_loaded_but_unused.json", "abs_again_unused.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=1, items=items)
assert result == ["pla_basic.json", "pla_basic.json", "pla_basic.json"]
def test_no_substitution_when_all_used(self):
# Multi-colour plate where every slot is used: leave the user's selections alone.
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1, 2, 3])])})
items = ["pla_white.json", "pla_red.json", "pla_blue.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=1, items=items)
assert result == ["pla_white.json", "pla_red.json", "pla_blue.json"]
def test_no_op_when_plate_id_is_none(self):
items = ["a.json", "b.json", "c.json"]
result = substitute_unused_plate_filaments(b"any bytes", plate_id=None, items=items)
assert result == items
def test_no_op_when_single_filament(self):
result = substitute_unused_plate_filaments(b"any bytes", plate_id=1, items=["only.json"])
assert result == ["only.json"]
result = substitute_unused_plate_filaments(b"any bytes", plate_id=1, items=[])
assert result == []
def test_no_op_when_source_not_zip(self):
items = ["a.json", "b.json"]
result = substitute_unused_plate_filaments(b"not a zip", plate_id=1, items=items)
assert result == items
def test_no_op_when_no_model_settings(self):
# Empty parse result is treated as "every slot used" (fail-open default).
zip_bytes = _make_3mf({"3D/3dmodel.model": b"<model/>"})
items = ["a.json", "b.json", "c.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=1, items=items)
assert result == items
def test_support_material_slot_preserved(self):
# #1881 regression: object geometry references only slot 1 (PLA),
# but slot 2 (PVA) is configured as the support material in
# project_settings.config. Without the support-slot union, slot 2's
# user-picked PVA profile would be overwritten with slot 1's PLA
# and the print would come out single-material with PLA supports.
model_settings = self._model_settings_xml([(1, [1])])
project_settings = json.dumps(
{
"enable_support": "1",
"support_filament": "2",
"support_interface_filament": "2",
"filament_type": ["PLA", "PVA"],
}
).encode()
zip_bytes = _make_3mf(
{
"Metadata/model_settings.config": model_settings,
"Metadata/project_settings.config": project_settings,
}
)
items = ["pla.json", "pva_support.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=1, items=items)
assert result == ["pla.json", "pva_support.json"]
def test_support_disabled_still_substitutes_unused(self):
# When supports are off, slot 2 is genuinely unused — the temp-spread
# validator still needs the substitution to succeed.
model_settings = self._model_settings_xml([(1, [1])])
project_settings = json.dumps(
{
"enable_support": "0",
"support_filament": "2",
}
).encode()
zip_bytes = _make_3mf(
{
"Metadata/model_settings.config": model_settings,
"Metadata/project_settings.config": project_settings,
}
)
items = ["pla.json", "abs_never_used.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=1, items=items)
assert result == ["pla.json", "pla.json"]
# ---- #2628: the anchor is the lowest USED slot, not slot 1 ----------
def test_substitutes_from_first_used_slot_when_slot_1_is_unused(self):
"""michaelklos's report: plate 2 of a multi-plate project uses only
slot 2, while slot 1 carries an ``@Bambu Lab H2D`` preset baked into
the source 3MF. Anchoring on slot 1 made the substitution a no-op for
the one slot that needed it, and the CLI rejected the A1 slice with
"filament preset (slot 1) is not compatible with printer
Bambu Lab A1 0.4 nozzle"."""
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1]), (2, [2])])})
items = ["tpu_at_h2d.json", "pla_at_a1.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=2, items=items)
assert result == ["pla_at_a1.json", "pla_at_a1.json"]
def test_unused_slot_1_does_not_poison_the_other_unused_slots(self):
"""With more than one unused slot, the old anchor propagated slot 1's
own (foreign-printer) preset into every other unused slot — the exact
poisoning #1851 removed from the picker."""
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1]), (2, [3])])})
items = ["tpu_at_h2d.json", "abs.json", "pla_at_a1.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=2, items=items)
assert result == ["pla_at_a1.json", "pla_at_a1.json", "pla_at_a1.json"]
def test_anchor_is_the_lowest_used_slot_not_merely_a_used_one(self):
"""Deterministic pick so the same project always slices identically."""
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1]), (2, [2, 4])])})
items = ["slot1.json", "slot2.json", "slot3.json", "slot4.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=2, items=items)
assert result == ["slot2.json", "slot2.json", "slot2.json", "slot4.json"]
def test_no_op_when_every_used_slot_is_outside_the_submitted_list(self):
"""A plate referencing only slots beyond the picked list leaves nothing
trustworthy to copy from — keep the user's picks rather than invent a
substitution from a slot the plate doesn't use."""
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1]), (2, [5])])})
items = ["a.json", "b.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=2, items=items)
assert result == ["a.json", "b.json"]
def test_support_slot_can_be_the_anchor(self):
"""The support-filament union (#1881) feeds the same used-slot set, so
a plate whose only geometry slot is 2 with PVA supports in 3 anchors on
2 — never on the unused slot 1."""
model_settings = self._model_settings_xml([(1, [1]), (2, [2])])
project_settings = json.dumps(
{
"enable_support": "1",
"support_filament": "3",
"support_interface_filament": "3",
"filament_type": ["PLA", "PLA", "PVA"],
}
).encode()
zip_bytes = _make_3mf(
{
"Metadata/model_settings.config": model_settings,
"Metadata/project_settings.config": project_settings,
}
)
items = ["tpu_at_h2d.json", "pla.json", "pva.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=2, items=items)
assert result == ["pla.json", "pla.json", "pva.json"]
# ---- #2711: plate 0 is the slice-all sentinel, not a plate ----------
def test_no_op_for_the_slice_all_sentinel(self):
"""``plate=0`` means every plate, so every slot is used by something."""
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1]), (2, [2])])})
items = ["pla_white.json", "pla_red.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=0, items=items)
assert result == items
def test_slice_all_is_not_collapsed_onto_the_support_filament(self):
"""The guard that makes the plate-0 no-op load-bearing.
Geometry lookup for plate 0 matches nothing (plates are 1-indexed),
but the support-filament union reads project settings and has no
plate scope — so it survives as the *only* member of the used set.
Anchored on it, a slice-all would rewrite every colour in the
project to the support material and print the whole thing in PVA.
"""
model_settings = self._model_settings_xml([(1, [1]), (2, [2]), (3, [3])])
project_settings = json.dumps(
{
"enable_support": "1",
"support_filament": "4",
"support_interface_filament": "4",
"filament_type": ["PLA", "PLA", "PLA", "PVA"],
}
).encode()
zip_bytes = _make_3mf(
{
"Metadata/model_settings.config": model_settings,
"Metadata/project_settings.config": project_settings,
}
)
items = ["white.json", "red.json", "blue.json", "pva.json"]
result = substitute_unused_plate_filaments(zip_bytes, plate_id=0, items=items)
assert result == items, "slice-all collapsed the project onto the support filament"
def test_negative_plate_id_is_a_no_op(self):
"""Not reachable through the schema (``ge=0``), but the function is the
thing that must not guess — a caller resolving a plate wrongly should
get the user's picks back, not a rewrite anchored on nothing."""
zip_bytes = _make_3mf({"Metadata/model_settings.config": self._model_settings_xml([(1, [1])])})
items = ["a.json", "b.json"]
assert substitute_unused_plate_filaments(zip_bytes, plate_id=-1, items=items) == items