mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
3dc681d9f1
slice_and_persist always wrote to get_library_files_dir() while giving the new row the source folder's id, so slicing a file on a NAS mount produced a .gcode.3mf that showed up in the right folder in the UI and never reached the share -- invisible from the web UI, which is why it did not reproduce. Resolve the destination from the target folder like uploads (#1112) and moves already do, set is_external and store the absolute path. Collisions uniquify to "Model (2).gcode.3mf": a 409 would throw away minutes of CPU on a routine re-slice, and overwriting a file on someone's NAS is worse. An external folder that cannot take the file (read-only, unreachable, not writable) falls back to managed storage rather than discarding the slice, and reports why on SliceResponse.external_write_fallback -- surfaced as a warning toast. Silent fallback is what made this bug invisible.
232 lines
9.4 KiB
Python
232 lines
9.4 KiB
Python
"""Slicing a file on an external mount writes the result to that mount (#2810).
|
|
|
|
Uploads learned to respect external folders in #1112 and moves in its
|
|
follow-up; ``slice_and_persist`` was the last write path that still sent
|
|
everything to managed storage. It kept giving the new row the external
|
|
folder's ``folder_id``, so the sliced file appeared in the right folder in the
|
|
File Manager while the share it was supposed to land on stayed empty -- which
|
|
is why the bug could not be reproduced from the web UI at all.
|
|
|
|
The fallback cases matter as much as the happy path. A slice costs minutes of
|
|
CPU, so an unwritable mount must not throw the bytes away; it stores them in
|
|
the managed library and *says so*, because filing the output somewhere the user
|
|
is not looking with no signal is the failure this issue was made of.
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.api.routes.library import (
|
|
_resolve_slice_destination,
|
|
_unique_external_name,
|
|
slice_and_persist,
|
|
)
|
|
from backend.app.models.library import LibraryFile, LibraryFolder
|
|
from backend.app.schemas.slicer import SliceRequest
|
|
from backend.app.services.slicer_api import SliceResult
|
|
|
|
|
|
def _external_folder(path: Path, *, readonly: bool = False) -> LibraryFolder:
|
|
return LibraryFolder(
|
|
name="NAS",
|
|
parent_id=None,
|
|
is_external=True,
|
|
external_path=str(path),
|
|
external_readonly=readonly,
|
|
)
|
|
|
|
|
|
class TestResolveSliceDestination:
|
|
def test_managed_folder_keeps_the_uuid_name(self, tmp_path):
|
|
folder = LibraryFolder(name="Models", parent_id=None, is_external=False)
|
|
|
|
path, is_external, fallback = _resolve_slice_destination(folder, "Bidoof.gcode.3mf")
|
|
|
|
assert is_external is False
|
|
assert fallback is None
|
|
# Managed storage is content-addressed by uuid: the display name lives
|
|
# on the DB row, so two files of the same name can coexist.
|
|
assert path.name.endswith(".gcode.3mf")
|
|
assert path.name != "Bidoof.gcode.3mf"
|
|
|
|
def test_no_folder_at_all_is_managed(self):
|
|
path, is_external, fallback = _resolve_slice_destination(None, "Bidoof.gcode.3mf")
|
|
|
|
assert is_external is False
|
|
assert fallback is None
|
|
assert path.name.endswith(".gcode.3mf")
|
|
|
|
def test_writable_external_folder_gets_the_real_filename(self, tmp_path):
|
|
mount = tmp_path / "share"
|
|
mount.mkdir()
|
|
|
|
path, is_external, fallback = _resolve_slice_destination(_external_folder(mount), "Bidoof.gcode.3mf")
|
|
|
|
assert is_external is True
|
|
assert fallback is None
|
|
# The point of the whole fix: next to the source, under a name a human
|
|
# can find on the share.
|
|
assert path == mount / "Bidoof.gcode.3mf"
|
|
|
|
def test_read_only_mount_falls_back_instead_of_failing(self, tmp_path):
|
|
mount = tmp_path / "share"
|
|
mount.mkdir()
|
|
|
|
path, is_external, fallback = _resolve_slice_destination(
|
|
_external_folder(mount, readonly=True), "Bidoof.gcode.3mf"
|
|
)
|
|
|
|
assert is_external is False
|
|
assert fallback == "external_readonly"
|
|
assert path.parent != mount
|
|
|
|
def test_vanished_mount_falls_back(self, tmp_path):
|
|
missing = tmp_path / "unplugged-nas" # deliberately not created
|
|
|
|
_path, is_external, fallback = _resolve_slice_destination(_external_folder(missing), "Bidoof.gcode.3mf")
|
|
|
|
assert is_external is False
|
|
assert fallback == "external_unreachable"
|
|
|
|
def test_folder_with_no_path_configured_falls_back(self):
|
|
folder = LibraryFolder(name="NAS", parent_id=None, is_external=True, external_path=None)
|
|
|
|
_path, is_external, fallback = _resolve_slice_destination(folder, "Bidoof.gcode.3mf")
|
|
|
|
assert is_external is False
|
|
assert fallback == "external_no_path"
|
|
|
|
@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores the write bit")
|
|
def test_unwritable_mount_falls_back(self, tmp_path):
|
|
mount = tmp_path / "share"
|
|
mount.mkdir()
|
|
mount.chmod(0o500)
|
|
try:
|
|
_path, is_external, fallback = _resolve_slice_destination(_external_folder(mount), "Bidoof.gcode.3mf")
|
|
finally:
|
|
mount.chmod(0o700)
|
|
|
|
assert is_external is False
|
|
assert fallback == "external_not_writable"
|
|
|
|
def test_a_name_that_escapes_the_mount_lands_in_managed_storage(self, tmp_path):
|
|
mount = tmp_path / "share"
|
|
mount.mkdir()
|
|
|
|
path, is_external, fallback = _resolve_slice_destination(_external_folder(mount), "../escaped.gcode.3mf")
|
|
|
|
# Never write outside the configured mount, whatever the name claims.
|
|
assert is_external is False
|
|
assert fallback == "external_invalid_name"
|
|
assert path.parent.resolve() != tmp_path.resolve()
|
|
|
|
|
|
class TestUniqueExternalName:
|
|
def test_free_name_is_used_as_is(self, tmp_path):
|
|
assert _unique_external_name(tmp_path, "Bidoof.gcode.3mf") == "Bidoof.gcode.3mf"
|
|
|
|
def test_collision_suffixes_before_the_compound_extension(self, tmp_path):
|
|
(tmp_path / "Bidoof.gcode.3mf").write_bytes(b"first slice")
|
|
|
|
# Not "Bidoof.gcode (2).3mf" -- the whole ".gcode.3mf" is the extension
|
|
# and splitting it would produce a name the printer path won't accept.
|
|
assert _unique_external_name(tmp_path, "Bidoof.gcode.3mf") == "Bidoof (2).gcode.3mf"
|
|
|
|
def test_it_keeps_counting_past_the_first_collision(self, tmp_path):
|
|
(tmp_path / "Bidoof.gcode.3mf").write_bytes(b"first")
|
|
(tmp_path / "Bidoof (2).gcode.3mf").write_bytes(b"second")
|
|
|
|
assert _unique_external_name(tmp_path, "Bidoof.gcode.3mf") == "Bidoof (3).gcode.3mf"
|
|
|
|
def test_re_slicing_never_overwrites_what_is_already_on_the_share(self, tmp_path):
|
|
(tmp_path / "Bidoof.gcode.3mf").write_bytes(b"do not lose me")
|
|
|
|
chosen = _unique_external_name(tmp_path, "Bidoof.gcode.3mf")
|
|
|
|
assert (tmp_path / chosen).exists() is False
|
|
assert (tmp_path / "Bidoof.gcode.3mf").read_bytes() == b"do not lose me"
|
|
|
|
|
|
class TestSliceAndPersistWritesToTheMount:
|
|
"""End to end through ``slice_and_persist`` with the slicer stubbed out."""
|
|
|
|
@staticmethod
|
|
def _patched_slicer(content: bytes = b"PK\x03\x04 not-a-real-3mf"):
|
|
return patch(
|
|
"backend.app.api.routes.library._run_slicer_with_fallback",
|
|
AsyncMock(return_value=(SliceResult(content, 3600, 12.5, 4200.0), False)),
|
|
)
|
|
|
|
async def _slice_into(self, db_session, folder: LibraryFolder):
|
|
db_session.add(folder)
|
|
await db_session.commit()
|
|
await db_session.refresh(folder)
|
|
|
|
with self._patched_slicer():
|
|
response = await slice_and_persist(
|
|
db_session,
|
|
model_bytes=b"source model",
|
|
model_filename="Bidoof.3mf",
|
|
folder_id=folder.id,
|
|
extra_metadata=None,
|
|
request=SliceRequest(printer_preset_id=1, process_preset_id=2, filament_preset_id=3),
|
|
current_user_id=None,
|
|
)
|
|
file_row = await db_session.get(LibraryFile, response.library_file_id)
|
|
return response, file_row
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_bytes_land_on_the_share(self, db_session, tmp_path):
|
|
mount = tmp_path / "share"
|
|
mount.mkdir()
|
|
|
|
response, file_row = await self._slice_into(db_session, _external_folder(mount))
|
|
|
|
assert (mount / "Bidoof.gcode.3mf").exists()
|
|
assert response.external_write_fallback is None
|
|
# The row has to agree with the disk, or the next move/scan/delete
|
|
# works on a path that isn't there.
|
|
assert file_row.is_external is True
|
|
assert file_row.file_path == str(mount / "Bidoof.gcode.3mf")
|
|
assert file_row.filename == "Bidoof.gcode.3mf"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_row_records_the_suffixed_name_on_a_collision(self, db_session, tmp_path):
|
|
mount = tmp_path / "share"
|
|
mount.mkdir()
|
|
(mount / "Bidoof.gcode.3mf").write_bytes(b"an earlier slice")
|
|
|
|
_response, file_row = await self._slice_into(db_session, _external_folder(mount))
|
|
|
|
assert file_row.filename == "Bidoof (2).gcode.3mf"
|
|
assert file_row.file_path == str(mount / "Bidoof (2).gcode.3mf")
|
|
assert (mount / "Bidoof.gcode.3mf").read_bytes() == b"an earlier slice"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_managed_folder_is_unaffected(self, db_session, tmp_path):
|
|
folder = LibraryFolder(name="Models", parent_id=None, is_external=False)
|
|
|
|
response, file_row = await self._slice_into(db_session, folder)
|
|
|
|
assert response.external_write_fallback is None
|
|
assert file_row.is_external is False
|
|
# Managed rows stay relative to base_dir so the install stays portable.
|
|
assert not Path(file_row.file_path).is_absolute()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_read_only_mount_still_yields_a_usable_file_and_says_why(self, db_session, tmp_path):
|
|
mount = tmp_path / "share"
|
|
mount.mkdir()
|
|
|
|
response, file_row = await self._slice_into(db_session, _external_folder(mount, readonly=True))
|
|
|
|
# Minutes of slicing must not be discarded because the mount is
|
|
# read-only -- but the user has to learn where the file went.
|
|
assert response.external_write_fallback == "external_readonly"
|
|
assert file_row.is_external is False
|
|
assert (file_row.file_metadata or {}).get("external_write_fallback") == "external_readonly"
|
|
assert list(mount.iterdir()) == []
|