vinez/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-08-11 00:30:12 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions
bambuddy/backend/tests/integration/test_library_slice_api.py
maziggy e95c42c021 Add auto-orient and auto-arrange to server-side slicing (#2548) 
Both are per-slice checkboxes, off by default, forwarded as the sidecar's
orient / arrange form fields. An unticked box is sent by omission: the
sidecar treats any present value as truthy, so a literal "false" would
have arranged every slice.

Arrange unions with the #1493 cross-class decision rather than replacing
it, and the per-plate slice-all loop is now keyed on the arrange flag
itself — the project-wide collapse belongs to --arrange, not to the
cross-class case. The loop also covers the embedded-settings path, whose
crash-retry is suppressed there since a single --slice 0 retry would
return one consolidated plate.
2026-08-03 14:10:02 +02:00

2356 lines
101 KiB
Python
Raw Permalink Blame History

"""Integration tests for the slice-via-API flow.
Routes under test:
- POST /library/files/{id}/slice (returns 202 + job_id; bg task does the work)
- POST /archives/{id}/slice (same shape; result lands in archives table)
- GET /slice-jobs/{id} (poll for terminal state)
The synchronous validation paths (404 missing source, 400 wrong file type)
are tested directly. The bg-task paths poll until the job finishes and then
assert on the captured state.
"""
from __future__ import annotations
import asyncio
import io
import json
import zipfile
from collections.abc import Callable
import httpx
import pytest
from httpx import AsyncClient
from backend.app.api.routes.library import _slicer_rejection_message
from backend.app.core.config import settings as app_settings
from backend.app.models.library import LibraryFile
from backend.app.models.local_preset import LocalPreset
from backend.app.models.settings import Settings as SettingsModel
from backend.app.services import slicer_api as slicer_api_module
from backend.app.services.slice_dispatch import slice_dispatch
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_3mf_with_settings(settings_payload: dict | None = None) -> bytes:
"""Build a tiny in-memory 3MF zip with all the embedded-config files
that real-world Bambu Studio / OrcaSlicer 3MFs ship with.
The strip-before-forwarding helper has to remove ALL of these (not
just `project_settings.config`) — leftover entries reference printer
/ filament IDs from the original slice and trip the CLI's input
validation when a different `--load-settings` triplet is supplied.
"""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/project_settings.config",
json.dumps(settings_payload or {"prime_tower_brim_width": "-1"}),
)
zf.writestr("Metadata/model_settings.config", "<config><object id='1'/></config>")
zf.writestr(
"Metadata/slice_info.config",
"<config><plate><metadata key='filament' value='GFL00'/></plate></config>",
)
zf.writestr("Metadata/cut_information.xml", "<cut><part id='1'/></cut>")
return buf.getvalue()
def _install_mock_sidecar(handler: Callable[[httpx.Request], httpx.Response]) -> httpx.AsyncClient:
"""Pin a MockTransport-backed httpx client onto the slicer_api singleton
so per-request `SlicerApiService` instances reuse it instead of opening
a real connection."""
client = httpx.AsyncClient(transport=httpx.MockTransport(handler), timeout=10.0)
slicer_api_module.set_shared_http_client(client)
return client
def _is_slice_post(request: httpx.Request) -> bool:
"""True for the slice call itself, false for the progress polls beside it.
Since #2730 a slice is supervised by a 1 Hz poll of
``GET /slice/progress/{id}``, which shares this mock transport. Tests that
count *slice attempts* — primary vs embedded-settings fallback — have to
exclude those, or the count becomes a measure of how long the test took.
"""
return request.method == "POST" and request.url.path.endswith("/slice")
async def _wait_for_job(client: AsyncClient, job_id: int, timeout: float = 5.0) -> dict:
"""Poll `/api/v1/slice-jobs/{id}` until the job hits a terminal state.
The dispatcher runs work as an asyncio task on the same event loop, so
poll-with-sleep here is enough — a few yields and the task finishes.
"""
deadline = asyncio.get_event_loop().time() + timeout
while asyncio.get_event_loop().time() < deadline:
r = await client.get(f"/api/v1/slice-jobs/{job_id}")
if r.status_code != 200:
raise AssertionError(f"slice-jobs poll failed: {r.status_code} {r.text}")
body = r.json()
if body["status"] in ("completed", "failed"):
return body
await asyncio.sleep(0.05)
raise AssertionError(f"slice job {job_id} did not finish in {timeout}s")
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
async def slice_test_setup(db_session, tmp_path, monkeypatch):
"""Source LibraryFile + 3 LocalPresets + preferred_slicer=orcaslicer.
``base_dir`` is patched via ``monkeypatch`` rather than assigned and
restored by hand. ``app_settings`` is a process-wide singleton, and the
hand-rolled version only restored after the ``yield`` — so anything raising
during setup (a commit, a refresh) left ``base_dir`` pointing at a
``tmp_path`` that pytest then deleted, and every later test in that xdist
worker which reads it failed. That was the cause of intermittent failures
in ``TestLibraryPathHelpers`` and ``TestArchivePlatesDesignOverrides``,
which share nothing with this module but land in the same worker.
"""
storage_dir = tmp_path / "library" / "files"
storage_dir.mkdir(parents=True, exist_ok=True)
src_path = storage_dir / "Cube.stl"
src_path.write_bytes(b"solid Cube\nendsolid\n")
monkeypatch.setattr(app_settings, "base_dir", tmp_path)
src_file = LibraryFile(
filename="Cube.stl",
file_path=str(src_path.relative_to(tmp_path)),
file_type="stl",
file_size=src_path.stat().st_size,
)
db_session.add(src_file)
presets = {}
for kind in ("printer", "process", "filament"):
p = LocalPreset(
name=f"Test {kind}",
preset_type=kind,
source="orcaslicer",
setting=json.dumps({"name": f"Test {kind}", "type": kind}),
)
db_session.add(p)
presets[kind] = p
db_session.add(SettingsModel(key="preferred_slicer", value="orcaslicer"))
await db_session.commit()
for p in presets.values():
await db_session.refresh(p)
await db_session.refresh(src_file)
yield {
"src_file_id": src_file.id,
"printer_id": presets["printer"].id,
"process_id": presets["process"].id,
"filament_id": presets["filament"].id,
"tmp_path": tmp_path,
}
slicer_api_module.set_shared_http_client(None)
# ---------------------------------------------------------------------------
# POST /library/files/{id}/slice — synchronous validation paths
# ---------------------------------------------------------------------------
class TestSliceValidation:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_returns_404_when_source_missing(self, async_client: AsyncClient, slice_test_setup):
_install_mock_sidecar(lambda r: httpx.Response(200, content=b""))
response = await async_client.post(
"/api/v1/library/files/999999/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_returns_400_for_wrong_file_type(self, async_client: AsyncClient, db_session, slice_test_setup):
gcode_path = slice_test_setup["tmp_path"] / "library" / "files" / "out.gcode"
gcode_path.write_bytes(b"; gcode\n")
gfile = LibraryFile(
filename="out.gcode",
file_path=str(gcode_path.relative_to(slice_test_setup["tmp_path"])),
file_type="gcode",
file_size=10,
)
db_session.add(gfile)
await db_session.commit()
await db_session.refresh(gfile)
_install_mock_sidecar(lambda r: httpx.Response(200, content=b""))
response = await async_client.post(
f"/api/v1/library/files/{gfile.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 400
assert "STL, 3MF, or STEP" in response.json()["detail"]
# ---------------------------------------------------------------------------
# POST /library/files/{id}/slice — async dispatch + bg job
# ---------------------------------------------------------------------------
class TestSliceLibraryFile:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_happy_path_returns_202_then_job_completes_with_library_file(
self, async_client: AsyncClient, slice_test_setup
):
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["url"] = str(request.url)
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(), # #2671: real zip; validation rejects non-3MF bodies
headers={
"x-print-time-seconds": "656",
"x-filament-used-g": "0.94",
"x-filament-used-mm": "302.5",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202, response.text
body = response.json()
assert body["status"] == "pending"
assert body["status_url"].startswith("/api/v1/slice-jobs/")
final = await _wait_for_job(async_client, body["job_id"])
assert final["status"] == "completed", final
assert final["result"]["library_file_id"] != slice_test_setup["src_file_id"]
assert final["result"]["print_time_seconds"] == 656
assert captured["url"].endswith("/slice")
@pytest.mark.asyncio
@pytest.mark.integration
async def test_bed_type_override_patches_process_profile(self, async_client: AsyncClient, slice_test_setup):
"""#1337: when SliceRequest.bed_type is set, the process JSON sent to
the sidecar must carry curr_bed_type with that exact value. Without
the patch, slicing high-temp filaments on a "Cool Plate" process
preset fails inside the slicer CLI with "does not support filament 1"
and the user has no way to switch plates from the SliceModal."""
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["body"] = bytes(request.content)
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(), # #2671: real zip; validation rejects non-3MF bodies
headers={
"x-print-time-seconds": "10",
"x-filament-used-g": "0.1",
"x-filament-used-mm": "1.0",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
"bed_type": "Textured PEI Plate",
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
# The presetProfile part of the multipart upload now carries the
# override. Searching the raw body avoids parsing the multipart by
# hand — the substring is unique enough since we control the JSON
# being patched.
assert b'"curr_bed_type": "Textured PEI Plate"' in captured["body"], (
"bed_type override must appear in the process JSON sent to the sidecar"
)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_bed_type_omitted_leaves_process_profile_untouched(self, async_client: AsyncClient, slice_test_setup):
"""Companion to the override test: the patch must NOT fire when the
client omits bed_type, so the process preset's own curr_bed_type
(or absence thereof) is forwarded to the sidecar unchanged."""
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["body"] = bytes(request.content)
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(), # #2671: real zip; validation rejects non-3MF bodies
headers={
"x-print-time-seconds": "10",
"x-filament-used-g": "0.1",
"x-filament-used-mm": "1.0",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert b"curr_bed_type" not in captured["body"], (
"bed_type must stay out of the process JSON when no override is set"
)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_auto_orient_and_arrange_reach_the_sidecar(self, async_client: AsyncClient, slice_test_setup):
"""#2548: the two layout passes are per-slice options, so ticking
them in the SliceModal has to come out the other end as the
sidecar's ``orient`` / ``arrange`` form fields. Before this the
flags existed on the wire but only #1493's cross-class detector
could set arrange, and nothing at all could set orient."""
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["body"] = bytes(request.content)
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(),
headers={
"x-print-time-seconds": "10",
"x-filament-used-g": "0.1",
"x-filament-used-mm": "1.0",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
"auto_orient": True,
"auto_arrange": True,
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert b'name="orient"' in captured["body"]
assert b'name="arrange"' in captured["body"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_layout_flags_absent_by_default(self, async_client: AsyncClient, slice_test_setup):
"""Companion to the above. Both default to off, and off is expressed
by omitting the field — the sidecar reads any present value as
truthy, so a "false" on the wire would auto-arrange every slice."""
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["body"] = bytes(request.content)
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(),
headers={
"x-print-time-seconds": "10",
"x-filament-used-g": "0.1",
"x-filament-used-mm": "1.0",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert b'name="orient"' not in captured["body"]
assert b'name="arrange"' not in captured["body"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_invalid_preset_id_surfaces_as_failed_job_with_status_400(
self, async_client: AsyncClient, slice_test_setup
):
_install_mock_sidecar(lambda r: httpx.Response(200, content=b""))
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
# Swap printer/filament — both exist but wrong preset_type.
"printer_preset_id": slice_test_setup["filament_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["printer_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "failed"
assert final["error_status"] == 400
assert "preset_type" in (final["error_detail"] or "")
@pytest.mark.asyncio
@pytest.mark.integration
async def test_unknown_preferred_slicer_fails_with_400(
self, async_client: AsyncClient, db_session, slice_test_setup
):
await db_session.execute(
SettingsModel.__table__.update().where(SettingsModel.key == "preferred_slicer").values(value="prusaslicer")
)
await db_session.commit()
_install_mock_sidecar(lambda r: httpx.Response(200, content=b""))
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "failed"
assert final["error_status"] == 400
assert "preferred_slicer" in (final["error_detail"] or "")
@pytest.mark.asyncio
@pytest.mark.integration
async def test_sidecar_unreachable_fails_with_502(self, async_client: AsyncClient, slice_test_setup):
def handler(_: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("connection refused")
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "failed"
assert final["error_status"] == 502
assert "unreachable" in (final["error_detail"] or "").lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_3mf_falls_back_to_embedded_settings_on_cli_failure(
self, async_client: AsyncClient, db_session, slice_test_setup
):
# When the slicer CLI fails on the --load-settings path (segfault
# on complex H2D models), Bambuddy retries with no profile triplet
# so the CLI uses the file's embedded settings.
src_3mf_path = slice_test_setup["tmp_path"] / "library" / "files" / "complex.3mf"
src_3mf_path.write_bytes(_make_3mf_with_settings({"prime_tower_brim_width": "-1"}))
threemf = LibraryFile(
filename="complex.3mf",
file_path=str(src_3mf_path.relative_to(slice_test_setup["tmp_path"])),
file_type="3mf",
file_size=src_3mf_path.stat().st_size,
)
db_session.add(threemf)
await db_session.commit()
await db_session.refresh(threemf)
call_count = {"n": 0}
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
call_count["n"] += 1
# First call: profile triplet present → simulate CLI 5xx
if call_count["n"] == 1:
return httpx.Response(
status_code=500,
json={"message": "Failed to slice the model"},
)
# Retry: no profile triplet → succeed with embedded settings
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(), # #2671: real zip; validation rejects non-3MF bodies
headers={
"x-print-time-seconds": "100",
"x-filament-used-g": "1.0",
"x-filament-used-mm": "100",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{threemf.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert final["result"]["used_embedded_settings"] is True
assert call_count["n"] == 2 # primary + fallback retry
@pytest.mark.asyncio
@pytest.mark.integration
async def test_stl_does_not_fall_back_on_cli_failure(self, async_client: AsyncClient, slice_test_setup):
# STL has no embedded settings — the CLI 5xx is terminal.
call_count = {"n": 0}
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
call_count["n"] += 1
return httpx.Response(
status_code=500,
json={"message": "Failed to slice the model"},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "failed"
assert final["error_status"] == 502
assert call_count["n"] == 1 # No retry for STL
@pytest.mark.asyncio
@pytest.mark.integration
async def test_3mf_input_forwarded_unmodified_to_sidecar(
self, async_client: AsyncClient, db_session, slice_test_setup
):
# 3MF input must be forwarded to the sidecar verbatim — every
# Metadata/*.config the source carries (project_settings,
# model_settings, slice_info, cut_information) is needed by the
# CLI to find plate definitions and baseline config; an earlier
# version of this code stripped them and caused the CLI to
# silently exit immediately after "Initializing StaticPrintConfigs"
# for every 3MF slice. --load-settings overrides the specific
# fields the user changed; the rest comes from the embedded data.
src_3mf_path = slice_test_setup["tmp_path"] / "library" / "files" / "real.3mf"
src_3mf_path.write_bytes(_make_3mf_with_settings({"prime_tower_brim_width": "-1"}))
threemf = LibraryFile(
filename="real.3mf",
file_path=str(src_3mf_path.relative_to(slice_test_setup["tmp_path"])),
file_type="3mf",
file_size=src_3mf_path.stat().st_size,
)
db_session.add(threemf)
await db_session.commit()
await db_session.refresh(threemf)
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["body"] = request.content
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(), # #2671: real zip; validation rejects non-3MF bodies
headers={
"x-print-time-seconds": "1",
"x-filament-used-g": "0",
"x-filament-used-mm": "0",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{threemf.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
# Recover the embedded zip from the multipart body and assert ALL
# the source's Metadata/*.config files are still present — the
# opposite of the previous (broken) "strip everything" test.
body = captured["body"]
pk = body.find(b"PK\x03\x04")
assert pk >= 0, "3MF body not found in multipart payload"
with zipfile.ZipFile(io.BytesIO(body[pk:]), "r") as zin:
names = set(zin.namelist())
assert "Metadata/project_settings.config" in names
assert "Metadata/model_settings.config" in names
assert "Metadata/slice_info.config" in names
assert "Metadata/cut_information.xml" in names
assert "3D/3dmodel.model" in names
@pytest.mark.asyncio
@pytest.mark.integration
async def test_use_embedded_settings_skips_profile_triplet(
self, async_client: AsyncClient, db_session, slice_test_setup
):
# "Slice as designed" (#2611): with use_embedded_settings the 3MF is
# sliced on its own project_settings.config — no --load-settings — so
# the sidecar request carries ONLY the model file, never the
# printer/process/filament profile parts. Succeeds on the first call
# (no crash-fallback), and the result is flagged used_embedded_settings.
src_3mf_path = slice_test_setup["tmp_path"] / "library" / "files" / "designed.3mf"
src_3mf_path.write_bytes(_make_3mf_with_settings({"wall_loops": "5"}))
threemf = LibraryFile(
filename="designed.3mf",
file_path=str(src_3mf_path.relative_to(slice_test_setup["tmp_path"])),
file_type="3mf",
file_size=src_3mf_path.stat().st_size,
)
db_session.add(threemf)
await db_session.commit()
await db_session.refresh(threemf)
captured: dict = {}
call_count = {"n": 0}
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
call_count["n"] += 1
captured["body"] = request.content
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(), # #2671: real zip; validation rejects non-3MF bodies
headers={
"x-print-time-seconds": "100",
"x-filament-used-g": "1.0",
"x-filament-used-mm": "100",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{threemf.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
"use_embedded_settings": True,
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert final["result"]["used_embedded_settings"] is True
assert call_count["n"] == 1 # embedded path taken directly, no fallback retry
# The multipart body must NOT carry any profile part — that is the
# whole point of the mode. Their presence would mean --load-settings
# ran and overrode the designer's embedded settings.
body = captured["body"]
assert b"printerProfile" not in body
assert b"presetProfile" not in body
assert b"filamentProfile" not in body
@pytest.mark.asyncio
@pytest.mark.integration
async def test_use_embedded_settings_ignored_for_stl(self, async_client: AsyncClient, slice_test_setup):
# An STL has no embedded project settings to honour, so the flag is a
# no-op: the normal profile path runs and the triplet is forwarded.
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["body"] = request.content
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(), # #2671: real zip; validation rejects non-3MF bodies
headers={
"x-print-time-seconds": "1",
"x-filament-used-g": "0",
"x-filament-used-mm": "0",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{slice_test_setup['src_file_id']}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
"use_embedded_settings": True,
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert final["result"]["used_embedded_settings"] is False
assert b"printerProfile" in captured["body"] # profile path still ran
# ---------------------------------------------------------------------------
# GET /slice-jobs/{id}
# ---------------------------------------------------------------------------
class TestSliceJobs:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_unknown_job_returns_404(self, async_client: AsyncClient):
# Sweep dispatcher state so a fresh ID is unknown.
slice_dispatch._jobs.clear()
r = await async_client.get("/api/v1/slice-jobs/999999")
assert r.status_code == 404
# ---------------------------------------------------------------------------
# POST /archives/{id}/slice — re-sliced archive reflects the target printer
# ---------------------------------------------------------------------------
def _make_sliced_3mf(printer_model_id: str, bed_type: str | None = None) -> bytes:
"""A minimal sliced-output 3MF that embeds a printer_model_id in
slice_info.config, the way a real Bambu Studio / OrcaSlicer export does.
ThreeMFParser reads this into metadata['sliced_for_model']. When
``bed_type`` is set, also embed ``curr_bed_type`` so the parser surfaces
``metadata['bed_type']`` — needed for the bed-type lift assertion in
TestSliceArchiveReslicedBedType."""
extra_meta = f"<metadata key='curr_bed_type' value='{bed_type}'/>" if bed_type else ""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/slice_info.config",
(
"<config><plate>"
f"<metadata key='printer_model_id' value='{printer_model_id}'/>"
f"{extra_meta}"
"</plate></config>"
),
)
return buf.getvalue()
class TestCrossClassSliceAllLoop:
"""#1493: when the user picks "Slice all plates" on a cross-class source
(X1C → H2D), Bambuddy must NOT send a single ``--slice 0 --arrange 1``
call — that consolidates every plate's objects onto one bed via BS's
project-wide arrange. Instead it loops per plate (``plate=N, arrange=true``)
and merges the N single-plate outputs into one multi-plate 3MF locally.
This test mocks the sidecar to assert (a) N calls happen, one per plate,
each with arrange=true, and (b) the resulting archive's stored 3MF
contains plate_1..plate_N.gcode entries."""
@staticmethod
def _make_multi_plate_x1c_source(plate_count: int = 3) -> bytes:
"""Source 3MF: X1C-stamped, N plates declared via model_settings."""
plate_blocks = "\n".join(
f'<plate><metadata key="plater_id" value="{i}"/></plate>' for i in range(1, plate_count + 1)
)
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/project_settings.config",
json.dumps({"printer_model": "Bambu Lab X1 Carbon"}),
)
zf.writestr(
"Metadata/model_settings.config",
f"<?xml version='1.0'?>\n<config>\n{plate_blocks}\n</config>\n",
)
return buf.getvalue()
@staticmethod
def _make_single_plate_sliced_output(plate_num: int) -> bytes:
"""Mock per-plate output: looks like what BS CLI returns for
--slice N. Carries an H2D project_settings (target), a one-line
slice_info <plate> block, and a per-plate gcode + thumbnail."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/project_settings.config",
json.dumps({"printer_model": "Bambu Lab H2D"}),
)
zf.writestr("Metadata/model_settings.config", "<config/>")
zf.writestr(
"Metadata/slice_info.config",
f"<config><plate><metadata key='index' value='{plate_num}'/>"
f"<metadata key='printer_model_id' value='O1D'/></plate></config>",
)
zf.writestr(f"Metadata/plate_{plate_num}.gcode", f"G{plate_num}".encode())
zf.writestr(f"Metadata/plate_{plate_num}.gcode.md5", b"deadbeef")
zf.writestr(f"Metadata/plate_{plate_num}.json", b"{}")
zf.writestr(f"Metadata/plate_{plate_num}.png", f"P{plate_num}".encode())
return buf.getvalue()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_loops_per_plate_when_cross_class_with_plate_zero(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "mewtwo.3mf"
src_3mf.write_bytes(self._make_multi_plate_x1c_source(plate_count=3))
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="mewtwo.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
# H2D target preset — the cross-class detector reads the
# ``printer_model`` field off the resolved JSON.
h2d = LocalPreset(
name="# Bambu Lab H2D 0.4 nozzle",
preset_type="printer",
source="orcaslicer",
setting=json.dumps({"name": "Bambu Lab H2D 0.4 nozzle", "printer_model": "Bambu Lab H2D"}),
)
db_session.add(h2d)
await db_session.commit()
await db_session.refresh(h2d)
# Mock sidecar: capture every request and respond with that
# plate's single-plate output. We expect one request per plate
# in the source (3 here).
captured_requests: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
# Multipart bodies aren't trivially parseable here; pull
# the plate field by string search since the helper sends
# ``name="plate"`` immediately followed by the value.
body = request.content
plate = None
marker = b'name="plate"\r\n\r\n'
idx = body.find(marker)
if idx != -1:
# Find the next CRLF after the value start.
start = idx + len(marker)
end = body.find(b"\r\n", start)
try:
plate = int(body[start:end].decode("utf-8"))
except (UnicodeDecodeError, ValueError):
plate = None
arrange_in_body = b'name="arrange"' in body
captured_requests.append({"plate": plate, "arrange": arrange_in_body})
return httpx.Response(
status_code=200,
content=self._make_single_plate_sliced_output(plate or 1),
headers={
"x-print-time-seconds": "600",
"x-filament-used-g": "5.0",
"x-filament-used-mm": "1600.0",
},
)
_install_mock_sidecar(handler)
# plate=0 + cross-class triplet → backend should enter the
# per-plate loop, slice each of the 3 plates with arrange=True,
# and merge into one archive.
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset": {"source": "local", "id": str(h2d.id)},
"process_preset": {"source": "local", "id": str(slice_test_setup["process_id"])},
"filament_presets": [{"source": "local", "id": str(slice_test_setup["filament_id"])}],
"plate": 0,
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"], timeout=15.0)
assert final["status"] == "completed", final
# Exactly one sidecar call per plate, in plate order. The
# ``--arrange 1`` flag travels with every per-plate sub-slice
# (it's what fixes the cross-class boundary error).
plates_called = [c["plate"] for c in captured_requests]
arrange_used = [c["arrange"] for c in captured_requests]
assert plates_called == [1, 2, 3], plates_called
assert all(arrange_used), arrange_used
# The merged archive has plate_1..plate_3.gcode inside its one
# output 3MF (single Bambuddy archive, three plates).
new_archive = await db_session.get(PrintArchive, final["result"]["archive_id"])
archive_path = tmp_path / new_archive.file_path
with zipfile.ZipFile(archive_path, "r") as zf:
entries = set(zf.namelist())
assert "Metadata/plate_1.gcode" in entries
assert "Metadata/plate_2.gcode" in entries
assert "Metadata/plate_3.gcode" in entries
# Per-plate-result totals are summed onto the merged archive.
assert new_archive.print_time_seconds == 600 * 3
assert new_archive.filament_used_grams == pytest.approx(5.0 * 3)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_user_requested_arrange_also_loops_per_plate(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""#2548 inherits #1493's hazard. The per-plate loop exists because
``--arrange`` is project-wide: a single ``--slice 0 --arrange 1``
collapses every plate's objects onto one bed. That is a property of
the flag, not of the cross-class detour that first needed it — so a
user ticking auto-arrange over "all plates" on a SAME-class source
must take the same loop. Keying the loop on the cross-class decision
alone would send one call and silently return a one-plate result.
"""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src_same_class"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "tray.3mf"
src_3mf.write_bytes(self._make_multi_plate_x1c_source(plate_count=2))
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="tray.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
# X1C target: same nozzle class as the X1C source, so #1493's
# detector stays off and only the user's flag is in play.
x1c = LocalPreset(
name="# Bambu Lab X1 Carbon 0.4 nozzle",
preset_type="printer",
source="orcaslicer",
setting=json.dumps({"name": "Bambu Lab X1 Carbon 0.4 nozzle", "printer_model": "Bambu Lab X1 Carbon"}),
)
db_session.add(x1c)
await db_session.commit()
await db_session.refresh(x1c)
captured_requests: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
body = request.content
plate = None
marker = b'name="plate"\r\n\r\n'
idx = body.find(marker)
if idx != -1:
start = idx + len(marker)
end = body.find(b"\r\n", start)
try:
plate = int(body[start:end].decode("utf-8"))
except (UnicodeDecodeError, ValueError):
plate = None
captured_requests.append(
{
"plate": plate,
"arrange": b'name="arrange"' in body,
"orient": b'name="orient"' in body,
}
)
return httpx.Response(
status_code=200,
content=self._make_single_plate_sliced_output(plate or 1),
headers={
"x-print-time-seconds": "300",
"x-filament-used-g": "2.0",
"x-filament-used-mm": "800.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset": {"source": "local", "id": str(x1c.id)},
"process_preset": {"source": "local", "id": str(slice_test_setup["process_id"])},
"filament_presets": [{"source": "local", "id": str(slice_test_setup["filament_id"])}],
"plate": 0,
"auto_arrange": True,
"auto_orient": True,
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"], timeout=15.0)
assert final["status"] == "completed", final
assert [c["plate"] for c in captured_requests] == [1, 2]
assert all(c["arrange"] for c in captured_requests)
# Orient rides along on every sub-slice too — it is per-object, so
# dropping it on the loop path would quietly ignore the user's tick.
assert all(c["orient"] for c in captured_requests)
new_archive = await db_session.get(PrintArchive, final["result"]["archive_id"])
with zipfile.ZipFile(tmp_path / new_archive.file_path, "r") as zf:
entries = set(zf.namelist())
assert "Metadata/plate_1.gcode" in entries
assert "Metadata/plate_2.gcode" in entries
@pytest.mark.asyncio
@pytest.mark.integration
async def test_embedded_settings_slice_all_with_arrange_still_loops(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
""" "Slice as designed" must not skip the loop. The project-wide
collapse comes from ``--arrange``; where the print config came from
has no bearing on it. Taking the single-call embedded branch here
would return one consolidated plate for a job the user asked to
slice as N — and the per-plate calls must still omit the profile
triplet, or "as designed" would silently stop meaning that.
"""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src_embedded"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "kit.3mf"
src_3mf.write_bytes(self._make_multi_plate_x1c_source(plate_count=2))
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="kit.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
x1c = LocalPreset(
name="# Bambu Lab X1 Carbon 0.4 nozzle",
preset_type="printer",
source="orcaslicer",
setting=json.dumps({"name": "Bambu Lab X1 Carbon 0.4 nozzle", "printer_model": "Bambu Lab X1 Carbon"}),
)
db_session.add(x1c)
await db_session.commit()
await db_session.refresh(x1c)
captured_requests: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
body = request.content
plate = None
marker = b'name="plate"\r\n\r\n'
idx = body.find(marker)
if idx != -1:
start = idx + len(marker)
end = body.find(b"\r\n", start)
try:
plate = int(body[start:end].decode("utf-8"))
except (UnicodeDecodeError, ValueError):
plate = None
captured_requests.append(
{
"plate": plate,
"arrange": b'name="arrange"' in body,
"has_profiles": b'name="printerProfile"' in body,
}
)
return httpx.Response(
status_code=200,
content=self._make_single_plate_sliced_output(plate or 1),
headers={
"x-print-time-seconds": "300",
"x-filament-used-g": "2.0",
"x-filament-used-mm": "800.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset": {"source": "local", "id": str(x1c.id)},
"process_preset": {"source": "local", "id": str(slice_test_setup["process_id"])},
"filament_presets": [{"source": "local", "id": str(slice_test_setup["filament_id"])}],
"plate": 0,
"use_embedded_settings": True,
"auto_arrange": True,
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"], timeout=15.0)
assert final["status"] == "completed", final
assert [c["plate"] for c in captured_requests] == [1, 2]
assert all(c["arrange"] for c in captured_requests)
# No --load-settings on any sub-call: the file's own settings drive
# each plate, which is what "slice as designed" promises.
assert not any(c["has_profiles"] for c in captured_requests)
assert final["result"]["used_embedded_settings"] is True
new_archive = await db_session.get(PrintArchive, final["result"]["archive_id"])
with zipfile.ZipFile(tmp_path / new_archive.file_path, "r") as zf:
entries = set(zf.namelist())
assert "Metadata/plate_1.gcode" in entries
assert "Metadata/plate_2.gcode" in entries
@pytest.mark.asyncio
@pytest.mark.integration
async def test_cross_class_arrange_survives_user_leaving_the_box_unticked(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""The user's per-slice choice is a union with #1493's decision, not
a replacement for it. Arrange is what keeps a class-crossing slice
from landing in the target's dead zone or segfaulting ZFiller — so
the default-false ``auto_arrange`` must not be able to turn it off.
"""
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src_cross_single"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "clip.3mf"
src_3mf.write_bytes(self._make_multi_plate_x1c_source(plate_count=1))
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="clip.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
h2d = LocalPreset(
name="# Bambu Lab H2D 0.4 nozzle",
preset_type="printer",
source="orcaslicer",
setting=json.dumps({"name": "Bambu Lab H2D 0.4 nozzle", "printer_model": "Bambu Lab H2D"}),
)
db_session.add(h2d)
await db_session.commit()
await db_session.refresh(h2d)
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
captured["body"] = bytes(request.content)
return httpx.Response(
status_code=200,
content=self._make_single_plate_sliced_output(1),
headers={
"x-print-time-seconds": "300",
"x-filament-used-g": "2.0",
"x-filament-used-mm": "800.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset": {"source": "local", "id": str(h2d.id)},
"process_preset": {"source": "local", "id": str(slice_test_setup["process_id"])},
"filament_presets": [{"source": "local", "id": str(slice_test_setup["filament_id"])}],
"plate": 1,
"auto_arrange": False,
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"], timeout=15.0)
assert final["status"] == "completed", final
assert b'name="arrange"' in captured["body"], "cross-class arrange must survive an explicit auto_arrange=false"
class TestSliceArchiveResliceModel:
"""Re-slicing an archive for a different printer must stamp the new
archive with the printer it was sliced FOR, not the source's printer."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_reslice_uses_target_model_not_source_model(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
# archive_dir is a static path off the real data dir; point it under
# base_dir (= tmp_path) so the new archive's file resolves there.
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
# Source archive: a 3MF that was sliced for an X1C.
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(_make_3mf_with_settings())
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
source_id = source.id
# The slicer returns a 3MF whose embedded printer_model_id is O1D (H2D).
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=200,
content=_make_sliced_3mf("O1D"),
headers={
"x-print-time-seconds": "600",
"x-filament-used-g": "5.0",
"x-filament-used-mm": "1600.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source_id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new_id = final["result"]["archive_id"]
assert new_id != source_id
new_archive = await db_session.get(PrintArchive, new_id)
# The fix: the re-sliced archive reflects H2D — the printer it was
# sliced for — instead of inheriting X1C from the source archive.
assert new_archive.sliced_for_model == "H2D"
# Source archive is untouched.
source_reloaded = await db_session.get(PrintArchive, source_id)
assert source_reloaded.sliced_for_model == "X1C"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_cross_model_reslice_drops_source_printer_id(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""A cross-model re-slice (source's X1C → target's H2D) must not carry
over ``source.printer_id``. The archive card and reprint modal both
read ``printer_id`` first and only fall back to ``sliced_for_model``
when it's None, so leaving the inherited id makes the H2D-sliced card
display the source's X1C printer name (the "Workshop H2C" bug)."""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(_make_3mf_with_settings())
source_printer = await printer_factory()
source = await archive_factory(
source_printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
source_id = source.id
source_printer_id = source_printer.id
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=200,
content=_make_sliced_3mf("O1D"), # H2D
headers={
"x-print-time-seconds": "600",
"x-filament-used-g": "5.0",
"x-filament-used-mm": "1600.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source_id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new_archive = await db_session.get(PrintArchive, final["result"]["archive_id"])
assert new_archive.sliced_for_model == "H2D"
# Card / reprint modal will now fall back to the sliced_for_model
# badge instead of showing the source printer's name.
assert new_archive.printer_id is None
# Source untouched: still bound to its original printer.
source_reloaded = await db_session.get(PrintArchive, source_id)
assert source_reloaded.printer_id == source_printer_id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_same_model_reslice_preserves_source_printer_id(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""Same-model re-slice (X1C → X1C, e.g. just swapped a process preset)
keeps ``printer_id`` so the reprint modal pre-selects the original
printer. Only cross-model re-slices null it out."""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(_make_3mf_with_settings())
source_printer = await printer_factory()
source = await archive_factory(
source_printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=200,
content=_make_sliced_3mf("C11"), # X1C — same model as source
headers={
"x-print-time-seconds": "600",
"x-filament-used-g": "5.0",
"x-filament-used-mm": "1600.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new_archive = await db_session.get(PrintArchive, final["result"]["archive_id"])
assert new_archive.sliced_for_model == "X1C"
# Same-model: keep the source's printer assignment so reprint pre-selects it.
assert new_archive.printer_id == source_printer.id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_reslice_with_unknown_source_model_preserves_printer_id(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""When ``source.sliced_for_model`` is None (older archive that
predates that column being populated), the backend can't tell whether
this is a cross-model re-slice. Fail open and preserve ``printer_id``
rather than spuriously nulling it — current pre-fix behaviour, kept
as a deliberate edge case."""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(_make_3mf_with_settings())
source_printer = await printer_factory()
source = await archive_factory(
source_printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model=None,
with_run=False,
)
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=200,
content=_make_sliced_3mf("O1D"),
headers={
"x-print-time-seconds": "600",
"x-filament-used-g": "5.0",
"x-filament-used-mm": "1600.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new_archive = await db_session.get(PrintArchive, final["result"]["archive_id"])
# Insufficient info to decide cross-model → preserve printer_id.
assert new_archive.printer_id == source_printer.id
class TestSliceArchiveReslicedThumbnail:
"""#1493 follow-up: the re-sliced archive's cover image preference order is
source's per-plate render > sliced output's per-plate render >
Auxiliaries marketing thumbnail. BS CLI rarely writes a fresh
``Metadata/plate_N.png`` on the sliced output, so the source's render
of the same plate (closer to what's actually printing) wins over the
project-wide marketing image."""
@staticmethod
def _make_source_with_plate_png(plate_png_bytes: bytes) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr("Metadata/plate_1.png", plate_png_bytes)
# Project-wide marketing image — the unwanted fallback target.
zf.writestr("Auxiliaries/.thumbnails/thumbnail_middle.png", b"COVER_ART")
return buf.getvalue()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_uses_source_plate_png_when_sliced_output_lacks_one(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""Sliced output has no per-plate PNG (typical of BS CLI output
with --arrange). The source's plate_1.png must win over the
sliced output's Auxiliaries fallback."""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
# Source has its own plate_1.png AND a project-wide cover.
source_plate_marker = b"SOURCE_PLATE_RENDER"
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(self._make_source_with_plate_png(source_plate_marker))
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
# Mock slicer returns a 3MF with NO Metadata/plate_1.png — only
# the Auxiliaries cover, mimicking BS CLI output with --arrange.
def handler(request: httpx.Request) -> httpx.Response:
sliced_buf = io.BytesIO()
with zipfile.ZipFile(sliced_buf, "w") as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr("Metadata/slice_info.config", "<config/>")
zf.writestr("Auxiliaries/.thumbnails/thumbnail_middle.png", b"SLICED_COVER_ART")
return httpx.Response(
status_code=200,
content=sliced_buf.getvalue(),
headers={"x-print-time-seconds": "60", "x-filament-used-g": "1", "x-filament-used-mm": "100"},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new = await db_session.get(PrintArchive, final["result"]["archive_id"])
assert new.thumbnail_path is not None
thumb_full = tmp_path / new.thumbnail_path
assert thumb_full.read_bytes() == source_plate_marker, (
"Re-sliced archive's thumbnail should be the source's per-plate render, not the Auxiliaries cover art."
)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_falls_back_to_auxiliaries_when_source_lacks_plate_png(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""When the source has no per-plate render (unsliced library upload),
the Auxiliaries marketing image from the sliced output is the
next-best preview — better than no card thumbnail at all."""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
# Source has no Metadata/plate_1.png at all.
bare_buf = io.BytesIO()
with zipfile.ZipFile(bare_buf, "w") as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "bare.3mf"
src_3mf.write_bytes(bare_buf.getvalue())
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="bare.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
def handler(request: httpx.Request) -> httpx.Response:
sliced_buf = io.BytesIO()
with zipfile.ZipFile(sliced_buf, "w") as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr("Metadata/slice_info.config", "<config/>")
zf.writestr("Auxiliaries/.thumbnails/thumbnail_middle.png", b"COVER_ART_FALLBACK")
return httpx.Response(
status_code=200,
content=sliced_buf.getvalue(),
headers={"x-print-time-seconds": "60", "x-filament-used-g": "1", "x-filament-used-mm": "100"},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new = await db_session.get(PrintArchive, final["result"]["archive_id"])
assert new.thumbnail_path is not None
thumb_full = tmp_path / new.thumbnail_path
assert thumb_full.read_bytes() == b"COVER_ART_FALLBACK"
class TestSliceArchiveReslicedBedType:
"""#1493 follow-up: the re-sliced archive's ``bed_type`` column must be
set from the produced 3MF's ``curr_bed_type`` so the frontend's archive
card shows the right build-plate badge (the card reads the column, not
extra_data, so the value was previously invisible after a re-slice)."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_bed_type_lifted_from_sliced_output(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(_make_3mf_with_settings())
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
bed_type="Cool Plate",
with_run=False,
)
# Mock slicer: produced 3MF declares a different plate type than
# the source archive's ``Cool Plate``. The new column must reflect
# the slicer's value (the user picked a different plate in the
# SliceModal) instead of inheriting the source's.
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=200,
content=_make_sliced_3mf("O1D", bed_type="Textured PEI Plate"),
headers={
"x-print-time-seconds": "600",
"x-filament-used-g": "5.0",
"x-filament-used-mm": "1600.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new = await db_session.get(PrintArchive, final["result"]["archive_id"])
assert new.bed_type == "Textured PEI Plate"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_bed_type_falls_back_to_source_when_missing_from_output(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""An older sidecar or sparse slice profile may produce a 3MF without
``curr_bed_type``. The source archive's ``bed_type`` is the right
default in that case — better than leaving the badge blank."""
from backend.app.models.archive import PrintArchive
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(_make_3mf_with_settings())
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
bed_type="Cool Plate",
with_run=False,
)
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=200,
# No bed_type embedded — simulates a sidecar that drops it.
content=_make_sliced_3mf("O1D"),
headers={
"x-print-time-seconds": "600",
"x-filament-used-g": "5.0",
"x-filament-used-mm": "1600.0",
},
)
_install_mock_sidecar(handler)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert resp.status_code == 202, resp.text
final = await _wait_for_job(async_client, resp.json()["job_id"])
assert final["status"] == "completed", final
new = await db_session.get(PrintArchive, final["result"]["archive_id"])
assert new.bed_type == "Cool Plate"
# ---------------------------------------------------------------------------
# Slicer content rejections surface instead of silently falling back
# ---------------------------------------------------------------------------
class TestSlicerRejectionMessage:
"""_slicer_rejection_message distinguishes a real slicer content rejection
(surface it to the user) from a CLI crash (fall back to embedded)."""
def test_extracts_bed_boundary_reason(self):
text = (
"Slicer CLI failed (500): Slicing failed with error from slicer: "
"Some objects are located over the boundary of the heated bed.: "
"Slicer process failed (exit code 204)\nstdout: trace ..."
)
assert _slicer_rejection_message(text) == "Some objects are located over the boundary of the heated bed."
def test_extracts_filament_temp_reason(self):
text = (
"Slicer CLI failed (500): Slicing failed with error from slicer: "
"The temperature difference of the filaments used is too large.: "
"Slicer process failed (exit code 194)"
)
assert _slicer_rejection_message(text) == "The temperature difference of the filaments used is too large."
def test_generic_cli_failure_is_not_a_rejection(self):
# The #1201 CLI-crash signature carries no slicer error_string, so it
# must still fall through to the embedded-settings fallback.
assert _slicer_rejection_message("Slicer CLI failed (500): Failed to slice the model") is None
def test_empty_or_unrelated_text(self):
assert _slicer_rejection_message("") is None
assert _slicer_rejection_message("Slicer sidecar unreachable: connection reset") is None
def test_replaces_input_preset_invalid_placeholder_with_cli_error_line(self):
# #1851: the CLI emits its catch-all "input preset file is invalid"
# placeholder for every -5 exit, including real preset-vs-printer
# compatibility rejections. The actual diagnostic only appears in the
# stdout `[error] run NNNN:` line; the function must prefer that.
text = (
"Slicer CLI failed (500): Slicing failed with error from slicer: "
"The input preset file is invalid and can not be parsed.: "
"Slicer process failed (exit code 251)\n"
"stdout: [2026-06-29 04:12:11.952784] [trace] Initializing StaticPrintConfigs\n"
"[2026-06-29 04:12:12.175810] [error] run 3008: filament preset "
"Generic PLA @BBL H2C (slot 1) is not compatible with printer "
"Bambu Lab A1 0.4 nozzle.\n"
"run found error, return -5, exit..."
)
assert (
_slicer_rejection_message(text) == "filament preset Generic PLA @BBL H2C (slot 1) is not compatible with "
"printer Bambu Lab A1 0.4 nozzle."
)
def test_keeps_meaningful_reason_even_when_cli_error_line_present(self):
# When the headline error_string is already a useful reason (here:
# the bed-boundary rejection), don't override it with a generic
# `[error]` line that may just be the same message restated. Avoids
# double-text duplication in the user-facing detail.
text = (
"Slicer CLI failed (500): Slicing failed with error from slicer: "
"Some objects are located over the boundary of the heated bed.: "
"Slicer process failed (exit code 204)\n"
"stdout: [error] some unrelated stdout chatter"
)
assert _slicer_rejection_message(text) == "Some objects are located over the boundary of the heated bed."
def test_cli_error_line_without_run_prefix(self):
# The CLI sometimes logs `[error] <msg>` without the `run NNNN:`
# prefix (different code paths). The regex must still pick it up.
text = (
"Slicer CLI failed (500): Slicing failed with error from slicer: "
"The input preset file is invalid and can not be parsed.: "
"Slicer process failed (exit code 251)\n"
"stdout: [2026-06-29 12:00:00.000000] [error] Configuration parse failed: missing key 'printer_settings_id'"
)
assert _slicer_rejection_message(text) == "Configuration parse failed: missing key 'printer_settings_id'"
class TestSliceSlicerRejection:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_3mf_surfaces_slicer_rejection_instead_of_falling_back(
self, async_client: AsyncClient, db_session, slice_test_setup
):
"""A real slicer content rejection (e.g. re-slicing for a printer with
a smaller bed) must surface as a 400 — not silently fall back to the
source 3MF's embedded settings, which would re-slice for the original
printer and hide the problem."""
src_3mf_path = slice_test_setup["tmp_path"] / "library" / "files" / "toobig.3mf"
src_3mf_path.write_bytes(_make_3mf_with_settings())
threemf = LibraryFile(
filename="toobig.3mf",
file_path=str(src_3mf_path.relative_to(slice_test_setup["tmp_path"])),
file_type="3mf",
file_size=src_3mf_path.stat().st_size,
)
db_session.add(threemf)
await db_session.commit()
await db_session.refresh(threemf)
call_count = {"n": 0}
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
call_count["n"] += 1
return httpx.Response(
status_code=500,
json={
"message": (
"Slicing failed with error from slicer: Some objects are "
"located over the boundary of the heated bed."
),
"details": "Slicer process failed (exit code 204)",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{threemf.id}/slice",
json={
"printer_preset_id": slice_test_setup["printer_id"],
"process_preset_id": slice_test_setup["process_id"],
"filament_preset_id": slice_test_setup["filament_id"],
},
)
assert response.status_code == 202
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "failed", final
assert final["error_status"] == 400
assert "boundary of the heated bed" in (final["error_detail"] or "")
# The slicer rejection must NOT trigger the embedded-settings retry.
assert call_count["n"] == 1
# ---------------------------------------------------------------------------
# Nozzle-class re-slice guard — single-nozzle <-> dual-nozzle (H2D) is blocked
# ---------------------------------------------------------------------------
from fastapi import HTTPException # noqa: E402
from backend.app.api.routes.library import ( # noqa: E402
_canonical_printer_model,
guard_nozzle_class_reslice,
)
class TestCanonicalPrinterModel:
"""_canonical_printer_model strips the '# ' clone prefix and the
' 0.4 nozzle' variant suffix so preset names resolve to a model code."""
def test_strips_nozzle_suffix(self):
assert _canonical_printer_model("Bambu Lab H2D 0.4 nozzle") == "H2D"
def test_strips_clone_prefix_and_suffix(self):
assert _canonical_printer_model("# Bambu Lab X1 Carbon 0.4 nozzle") == "X1C"
def test_bare_model_and_empty(self):
assert _canonical_printer_model("Bambu Lab H2D") == "H2D"
assert _canonical_printer_model(None) is None
assert _canonical_printer_model("") is None
class TestNozzleClassGuard:
"""guard_nozzle_class_reslice is now a no-op (#1493). Cross-class re-slicing
is handled by the two-pass conversion in _run_slicer_with_fallback — so the
guard never blocks. The function is kept (and these tests with it) so
external forks / pinned versions that call it still link, and so a future
regression that re-introduces a raise inside the helper gets caught here."""
@staticmethod
def _request() -> object:
return type("_Req", (), {})()
@pytest.mark.asyncio
async def test_single_to_dual_is_allowed(self, monkeypatch):
"""Cross-class re-slice: handled by the two-pass converter, so the
guard does NOT raise."""
import backend.app.api.routes.library as lib
async def _target(_db, _user, _request):
return "H2D"
monkeypatch.setattr(lib, "_resolve_target_printer_model", _target)
# No raise — the converter handles this case now.
await guard_nozzle_class_reslice(None, None, self._request(), "X1C")
@pytest.mark.asyncio
async def test_dual_to_single_is_allowed(self, monkeypatch):
import backend.app.api.routes.library as lib
async def _target(_db, _user, _request):
return "X1C"
monkeypatch.setattr(lib, "_resolve_target_printer_model", _target)
await guard_nozzle_class_reslice(None, None, self._request(), "H2D")
@pytest.mark.asyncio
async def test_same_nozzle_class_is_allowed(self, monkeypatch):
import backend.app.api.routes.library as lib
async def _target(_db, _user, _request):
return "P1S"
monkeypatch.setattr(lib, "_resolve_target_printer_model", _target)
await guard_nozzle_class_reslice(None, None, self._request(), "X1C")
@pytest.mark.asyncio
async def test_no_source_model_is_a_noop(self, monkeypatch):
import backend.app.api.routes.library as lib
async def _target(_db, _user, _request):
return "H2D"
monkeypatch.setattr(lib, "_resolve_target_printer_model", _target)
await guard_nozzle_class_reslice(None, None, self._request(), None)
@pytest.mark.asyncio
async def test_null_request_is_a_noop(self):
await guard_nozzle_class_reslice(None, None, None, "X1C")
@pytest.mark.asyncio
@pytest.mark.integration
async def test_archive_reslice_x1c_to_h2d_preset_path_is_not_400(
self, async_client: AsyncClient, db_session, slice_test_setup, printer_factory, archive_factory, monkeypatch
):
"""End to end: the preset-driven archive re-slice from X1C to H2D no
longer gets a synchronous 400 from the guard. It may still fail
downstream (no sidecar in test env), but it must not be rejected by
the nozzle-class guard's old "isn't supported yet" message."""
tmp_path = slice_test_setup["tmp_path"]
monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
src_dir = tmp_path / "archives" / "src"
src_dir.mkdir(parents=True, exist_ok=True)
src_3mf = src_dir / "cube.3mf"
src_3mf.write_bytes(_make_3mf_with_settings())
printer = await printer_factory()
source = await archive_factory(
printer.id,
filename="cube.3mf",
file_path=str(src_3mf.relative_to(tmp_path)),
sliced_for_model="X1C",
with_run=False,
)
h2d = LocalPreset(
name="# Bambu Lab H2D 0.4 nozzle",
preset_type="printer",
source="orcaslicer",
setting=json.dumps({"name": "Bambu Lab H2D 0.4 nozzle", "printer_model": "Bambu Lab H2D"}),
)
db_session.add(h2d)
await db_session.commit()
await db_session.refresh(h2d)
resp = await async_client.post(
f"/api/v1/archives/{source.id}/slice",
json={
"printer_preset": {"source": "local", "id": str(h2d.id)},
"process_preset": {"source": "local", "id": str(slice_test_setup["process_id"])},
"filament_presets": [{"source": "local", "id": str(slice_test_setup["filament_id"])}],
},
)
if resp.status_code == 400:
detail = resp.json().get("detail", "")
assert "isn't supported" not in detail, f"guard still firing on preset path: {detail!r}"
class TestUnusedSlotSubstitutionOnSinglePlateSource:
"""#2711: a single-plate 3MF must still get its unused slots substituted.
The SliceModal omits ``plate`` entirely for single-plate and STL sources —
it skips the plate picker, so ``selectedPlate`` stays null and the field
never reaches the body. The schema documents an absent plate as "plate 1",
but the substitution used to read it as "unknown plate" and skip, so every
single-plate project reached the CLI with the dropdown values of slots the
plate never paints with.
In the reported case that was a MakerWorld project declaring four filaments
while plate 1 paints with one, the other three carrying presets baked into
the source for a different printer. The CLI rejected the whole slice with
"filament preset ... (slot 1) is not compatible with printer ...", and the
modal disables unused rows so there was no way to correct it by hand.
"""
@staticmethod
def _single_plate_using_only_slot_3() -> bytes:
"""One plate, one object, painted with slot 3 — slots 1, 2 and 4 are
declared by the project but unused. Mirrors the reported file."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/project_settings.config",
json.dumps({"filament_type": ["PLA", "PLA", "PLA", "TPU"]}),
)
zf.writestr(
"Metadata/model_settings.config",
"<?xml version='1.0'?>\n<config>"
'<object id="1"><metadata key="extruder" value="3"/></object>'
'<plate><metadata key="plater_id" value="1"/>'
'<model_instance><metadata key="object_id" value="1"/>'
'<metadata key="instance_id" value="0"/></model_instance>'
"</plate></config>",
)
return buf.getvalue()
@staticmethod
def _filament_names_sent(body: bytes) -> list[str]:
"""Pull the ``name`` of each ``filamentProfile`` part, in slot order.
``slice_model`` sends one repeated ``filamentProfile`` part per slot as
``filament_N.json``; the parts stay in submission order, so a plain
scan preserves the slot mapping.
"""
names: list[str] = []
marker = b'name="filamentProfile"; filename="filament_'
pos = body.find(marker)
while pos != -1:
start = body.find(b"{", pos)
end = body.find(b"\r\n", start)
names.append(json.loads(body[start:end].decode("utf-8"))["name"])
pos = body.find(marker, end)
return names
@pytest.mark.asyncio
@pytest.mark.integration
async def test_unused_slots_are_substituted_when_the_body_omits_plate(
self, async_client: AsyncClient, db_session, slice_test_setup
):
tmp_path = slice_test_setup["tmp_path"]
src = tmp_path / "library" / "files" / "train.3mf"
src.write_bytes(self._single_plate_using_only_slot_3())
threemf = LibraryFile(
filename="train.3mf",
file_path=str(src.relative_to(tmp_path)),
file_type="3mf",
file_size=src.stat().st_size,
)
db_session.add(threemf)
# Four distinguishable filament presets, one per project slot. Only
# slot 3's is compatible with the target in the reported scenario.
slots = []
for i in range(1, 5):
p = LocalPreset(
name=f"slot{i}",
preset_type="filament",
source="orcaslicer",
setting=json.dumps({"name": f"slot{i}", "type": "filament"}),
)
db_session.add(p)
slots.append(p)
await db_session.commit()
await db_session.refresh(threemf)
for p in slots:
await db_session.refresh(p)
captured: list[list[str]] = []
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
captured.append(self._filament_names_sent(request.content))
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(),
headers={
"x-print-time-seconds": "100",
"x-filament-used-g": "1.0",
"x-filament-used-mm": "100",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{threemf.id}/slice",
json={
"printer_preset": {"source": "local", "id": str(slice_test_setup["printer_id"])},
"process_preset": {"source": "local", "id": str(slice_test_setup["process_id"])},
"filament_presets": [{"source": "local", "id": str(p.id)} for p in slots],
# No "plate" — exactly what the modal sends for a single-plate
# source. This is the whole point of the test.
},
)
assert response.status_code == 202, response.text
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert captured, "sidecar was never called"
# Every slot carries slot 3's profile: the array length stays intact
# (the source's per-slot references depend on it) while nothing the
# plate doesn't print with can fail the CLI's validators.
assert captured[0] == ["slot3", "slot3", "slot3", "slot3"], captured[0]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_slice_all_keeps_every_slot(self, async_client: AsyncClient, db_session, slice_test_setup):
"""``plate=0`` is the all-plates sentinel, so nothing is unused.
It reaches the same call site, and plate ids are 1-indexed — the
geometry lookup for plate 0 matches nothing. Without an explicit
exclusion the project's support-filament slot would be the only
member of the used set and would be copied over every colour.
"""
tmp_path = slice_test_setup["tmp_path"]
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/project_settings.config",
json.dumps(
{
"enable_support": "1",
"support_filament": "4",
"support_interface_filament": "4",
"filament_type": ["PLA", "PLA", "PLA", "PVA"],
}
),
)
zf.writestr(
"Metadata/model_settings.config",
"<?xml version='1.0'?>\n<config>"
'<object id="1"><metadata key="extruder" value="1"/></object>'
'<object id="2"><metadata key="extruder" value="2"/></object>'
'<plate><metadata key="plater_id" value="1"/>'
'<model_instance><metadata key="object_id" value="1"/></model_instance></plate>'
'<plate><metadata key="plater_id" value="2"/>'
'<model_instance><metadata key="object_id" value="2"/></model_instance></plate>'
"</config>",
)
src = tmp_path / "library" / "files" / "multi.3mf"
src.write_bytes(buf.getvalue())
threemf = LibraryFile(
filename="multi.3mf",
file_path=str(src.relative_to(tmp_path)),
file_type="3mf",
file_size=src.stat().st_size,
)
db_session.add(threemf)
slots = []
for i in range(1, 5):
p = LocalPreset(
name=f"slot{i}",
preset_type="filament",
source="orcaslicer",
setting=json.dumps({"name": f"slot{i}", "type": "filament"}),
)
db_session.add(p)
slots.append(p)
await db_session.commit()
await db_session.refresh(threemf)
for p in slots:
await db_session.refresh(p)
captured: list[list[str]] = []
def handler(request: httpx.Request) -> httpx.Response:
if not _is_slice_post(request):
return httpx.Response(404)
captured.append(self._filament_names_sent(request.content))
return httpx.Response(
status_code=200,
content=_make_3mf_with_settings(),
headers={
"x-print-time-seconds": "100",
"x-filament-used-g": "1.0",
"x-filament-used-mm": "100",
},
)
_install_mock_sidecar(handler)
response = await async_client.post(
f"/api/v1/library/files/{threemf.id}/slice",
json={
"printer_preset": {"source": "local", "id": str(slice_test_setup["printer_id"])},
"process_preset": {"source": "local", "id": str(slice_test_setup["process_id"])},
"filament_presets": [{"source": "local", "id": str(p.id)} for p in slots],
"plate": 0,
},
)
assert response.status_code == 202, response.text
final = await _wait_for_job(async_client, response.json()["job_id"])
assert final["status"] == "completed", final
assert captured, "sidecar was never called"
assert captured[0] == ["slot1", "slot2", "slot3", "slot4"], captured[0]
class TestFilamentRequirementsFullSlots:
"""#2712: what the slice modal is handed must be positional.
The modal's filament list maps index 0 to slot 1, and the backend forwards
it in that order as ``filament_1.json``..``filament_N.json``. A MakerWorld
source that ships slice_info and paints with slot 4 alone therefore has to
present four rows — a one-row list binds the user's pick to slot 1, and
slot 4 slices with whatever the source had baked in. Picking PETG produced
a PLA print, and the print dialog then correctly refused to match PETG.
Print-time AMS matching shares this endpoint and needs the opposite: only
the slots the plate consumes, so it doesn't demand spools for slots the
G-code never touches. Hence the opt-in flag rather than a shape change.
"""
@staticmethod
def _sliced_source_using_only_slot_4() -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/project_settings.config",
json.dumps(
{
"filament_type": ["PLA", "PLA", "PLA", "PLA"],
"filament_colour": ["#38CC0A", "#161616", "#898989", "#898989"],
}
),
)
# MakerWorld ships slice_info without plate G-code, which is what
# sends this file down the "already sliced" branch.
zf.writestr(
"Metadata/slice_info.config",
"<?xml version='1.0'?>\n<config><plate>"
"<metadata key='index' value='1'/>"
"<filament id='4' tray_info_idx='GFL99' type='PLA' color='#898989'"
" used_m='35.51' used_g='105.92'/>"
"</plate></config>",
)
return buf.getvalue()
async def _make_file(self, db_session, tmp_path) -> int:
src = tmp_path / "library" / "files" / "tunnel.3mf"
src.write_bytes(self._sliced_source_using_only_slot_4())
lib = LibraryFile(
filename="tunnel.3mf",
file_path=str(src.relative_to(tmp_path)),
file_type="3mf",
file_size=src.stat().st_size,
)
db_session.add(lib)
await db_session.commit()
await db_session.refresh(lib)
return lib.id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_full_slots_returns_one_row_per_project_slot(
self, async_client: AsyncClient, db_session, slice_test_setup
):
file_id = await self._make_file(db_session, slice_test_setup["tmp_path"])
r = await async_client.get(f"/api/v1/library/files/{file_id}/filament-requirements?plate_id=1&full_slots=true")
assert r.status_code == 200, r.text
filaments = r.json()["filaments"]
assert [f["slot_id"] for f in filaments] == [1, 2, 3, 4]
# Only slot 4 prints, so only its row is selectable in the modal.
assert [f["used_in_plate"] for f in filaments] == [False, False, False, True]
# The used row keeps what the slice actually reported.
assert filaments[3]["used_grams"] == 105.9
assert filaments[3]["tray_info_idx"] == "GFL99"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_the_print_path_still_gets_only_the_used_slot(
self, async_client: AsyncClient, db_session, slice_test_setup
):
"""Without the flag the response must be byte-for-byte what it was.
PrintModal drives AMS matching off this; widening it would ask the
user to load three spools the print never touches.
"""
file_id = await self._make_file(db_session, slice_test_setup["tmp_path"])
r = await async_client.get(f"/api/v1/library/files/{file_id}/filament-requirements?plate_id=1")
assert r.status_code == 200, r.text
filaments = r.json()["filaments"]
assert [f["slot_id"] for f in filaments] == [4]
assert filaments[0]["used_in_plate"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_unsliced_sources_are_unaffected_by_the_flag(
self, async_client: AsyncClient, db_session, slice_test_setup
):
"""Those already returned the full project list; the flag must not
double-handle them or change what the modal has been getting."""
tmp_path = slice_test_setup["tmp_path"]
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("3D/3dmodel.model", "<model/>")
zf.writestr(
"Metadata/project_settings.config",
json.dumps({"filament_type": ["PLA", "PETG"], "filament_colour": ["#000000", "#FFFFFF"]}),
)
src = tmp_path / "library" / "files" / "raw.3mf"
src.write_bytes(buf.getvalue())
lib = LibraryFile(
filename="raw.3mf",
file_path=str(src.relative_to(tmp_path)),
file_type="3mf",
file_size=src.stat().st_size,
)
db_session.add(lib)
await db_session.commit()
await db_session.refresh(lib)
with_flag = await async_client.get(
f"/api/v1/library/files/{lib.id}/filament-requirements?plate_id=1&full_slots=true"
)
without = await async_client.get(f"/api/v1/library/files/{lib.id}/filament-requirements?plate_id=1")
assert with_flag.status_code == 200 and without.status_code == 200
assert with_flag.json()["filaments"] == without.json()["filaments"]
assert [f["slot_id"] for f in with_flag.json()["filaments"]] == [1, 2]
Reference in a new issue View git blame Copy permalink