public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 03:01:21 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
bambuddy/backend/tests/unit/test_scheduler_cleanup_library.py
T
maziggy a4795c5ca3 Let API keys read and run slicer pipelines (#1425 follow-up) 
Every pipeline endpoint answered 403 for API keys whatever scopes the key
carried. PR A parked all three permissions on the admin denylist until the
run dispatch existed to decide about; it landed in PR C and the parking was
never revisited.

PIPELINES_READ now rides can_read_status. PIPELINES_RUN requires
can_queue AND can_manage_library together, so the allowlist gained tuple
values: a run slices into the library and then queues prints, and mapping
it to either flag alone would hand that flag the other one's authority.
The 403 names every flag the key is short of. PIPELINES_WRITE stays
admin-only -- a key can run the recipe, not rewrite it or clear the log.

Opening the run route also needed the cloud-owner fallback the direct
slice route makes: a pipeline can carry Bambu/Orca Cloud presets, and
resolving those reads a token off a user record that an API-keyed request
does not have. retry_failed forwards the new dependency explicitly,
since a direct call receives the Depends marker rather than None.
2026-08-15 10:38:51 +02:00

496 lines
20 KiB
Python
Raw Permalink Blame History

from contextlib import ExitStack
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
import backend.app.models # noqa: F401 - populate Base.metadata
import backend.app.services.print_scheduler as scheduler_module
from backend.app.core.database import Base
from backend.app.models.archive import PrintArchive
from backend.app.models.library import LibraryFile
from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
from backend.app.models.printer import Printer
from backend.app.services.print_scheduler import PrintScheduler
from backend.tests._fixtures.background_tasks import discarding_spawn_patch
@pytest.fixture
async def queue_factory(tmp_path):
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
session_maker = async_sessionmaker(engine, expire_on_commit=False)
case_counter = 0
async def make_case(*, cleanup=True, is_external=False, thumbnail_path=None, siblings=()):
nonlocal case_counter
case_counter += 1
base_dir = tmp_path / f"case-{case_counter}"
base_dir.mkdir()
source_path = base_dir / "library" / f"source-{case_counter}.3mf"
source_path.parent.mkdir()
source_path.write_bytes(b"library source")
thumbnail_actual_path = None
thumbnail_db_path = None
if thumbnail_path == "relative":
thumbnail_db_path = f"thumbs/preview-{case_counter}.png"
thumbnail_actual_path = base_dir / thumbnail_db_path
elif thumbnail_path == "absolute":
thumbnail_actual_path = tmp_path / f"absolute-preview-{case_counter}.png"
thumbnail_db_path = str(thumbnail_actual_path)
elif thumbnail_path is not None:
thumbnail_actual_path = Path(thumbnail_path)
thumbnail_db_path = str(thumbnail_path)
if thumbnail_actual_path:
thumbnail_actual_path.parent.mkdir(parents=True, exist_ok=True)
thumbnail_actual_path.write_bytes(b"thumbnail")
async with session_maker() as db:
printer = Printer(
name=f"Printer {case_counter}",
serial_number=f"SERIAL-{case_counter}",
ip_address="127.0.0.1",
access_code="access-code",
model="X1C",
)
library_file = LibraryFile(
filename=f"source-{case_counter}.3mf",
file_path=str(source_path),
file_type="3mf",
file_size=source_path.stat().st_size,
file_hash=None,
thumbnail_path=thumbnail_db_path,
file_metadata=None,
is_external=is_external,
)
db.add_all([printer, library_file])
await db.flush()
item = PrintQueueItem(
printer_id=printer.id,
library_file_id=library_file.id,
status="pending",
cleanup_library_after_dispatch=cleanup,
bed_levelling="on",
flow_cali="off",
vibration_cali=True,
layer_inspect=False,
timelapse=False,
use_ams=True,
nozzle_offset_cali="on",
)
db.add(item)
await db.flush()
# The other copies of a quantity>1 dispatch (#2819). Each entry is a
# dict of overrides: `status`, `own_archive` for a copy that already
# holds one, and `extra_variant` for a cross-model copy that keeps a
# candidate this cleanup does not consume.
sibling_ids = []
other_file = None
for spec in siblings:
sibling = PrintQueueItem(
printer_id=printer.id,
library_file_id=None if spec.get("variants") else library_file.id,
status=spec.get("status", "pending"),
cleanup_library_after_dispatch=cleanup,
)
if spec.get("own_archive"):
own = PrintArchive(
printer_id=printer.id,
filename="already-dispatched.3mf",
file_path="archives/already-dispatched.3mf",
file_size=1,
status="printing",
)
db.add(own)
await db.flush()
sibling.archive_id = own.id
db.add(sibling)
await db.flush()
if spec.get("variants"):
db.add(
PrintQueueVariant(
queue_item_id=sibling.id,
library_file_id=library_file.id,
target_model="X1C",
position=0,
)
)
if spec.get("extra_variant"):
if other_file is None:
other_path = base_dir / "library" / f"other-{case_counter}.3mf"
other_path.write_bytes(b"other source")
other_file = LibraryFile(
filename=f"other-{case_counter}.3mf",
file_path=str(other_path),
file_type="3mf",
file_size=other_path.stat().st_size,
)
db.add(other_file)
await db.flush()
db.add(
PrintQueueVariant(
queue_item_id=sibling.id,
library_file_id=other_file.id,
target_model="P1S",
position=1,
)
)
sibling_ids.append(sibling.id)
await db.commit()
return SimpleNamespace(
session_maker=session_maker,
base_dir=base_dir,
source_path=source_path,
thumbnail_path=thumbnail_actual_path,
printer_id=printer.id,
library_file_id=library_file.id,
queue_item_id=item.id,
sibling_ids=sibling_ids,
other_library_file_id=other_file.id if other_file is not None else None,
archive_path=None,
upload=AsyncMock(return_value=True),
start_print=MagicMock(return_value=True),
)
try:
yield make_case
finally:
await engine.dispose()
async def _dispatch_library_item(ctx, *, archive_failure=False, unlink_side_effect=None):
scheduler = PrintScheduler()
async def archive_print(
self,
*,
printer_id,
source_file,
original_filename,
created_by_id=None,
project_id=None,
cost_center_id=None,
plate_id=None,
library_file_id=None,
):
if archive_failure:
raise RuntimeError("archive copy failed")
archive_rel_path = Path("archives") / f"archive-{ctx.queue_item_id}.3mf"
ctx.archive_path = ctx.base_dir / archive_rel_path
ctx.archive_path.parent.mkdir(parents=True, exist_ok=True)
ctx.archive_path.write_bytes(Path(source_file).read_bytes())
archive = PrintArchive(
printer_id=printer_id,
filename=original_filename,
file_path=str(archive_rel_path),
file_size=ctx.archive_path.stat().st_size,
content_hash=None,
thumbnail_path=None,
timelapse_path=None,
print_time_seconds=120,
status="completed",
project_id=project_id,
library_file_id=library_file_id,
created_by_id=created_by_id,
)
self.db.add(archive)
await self.db.flush()
return archive
patches = [
patch.object(scheduler_module.settings, "base_dir", ctx.base_dir),
patch("backend.app.services.archive.ArchiveService.archive_print", new=archive_print),
patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=None)),
patch("backend.app.services.print_scheduler.printer_manager.start_print", ctx.start_print),
patch("backend.app.services.print_scheduler.printer_manager.set_awaiting_plate_clear", MagicMock()),
patch(
"backend.app.services.print_scheduler.get_ftp_retry_settings", AsyncMock(return_value=(False, 0, 0, 1.0))
),
patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True)),
patch("backend.app.services.print_scheduler.upload_file_async", ctx.upload),
patch("backend.app.services.print_scheduler.cache_3mf_download", MagicMock()),
discarding_spawn_patch(),
patch("backend.app.services.notification_service.notification_service.on_queue_job_started", AsyncMock()),
patch("backend.app.services.notification_service.notification_service.on_queue_job_failed", AsyncMock()),
patch("backend.app.services.mqtt_relay.mqtt_relay.on_queue_job_started", AsyncMock()),
patch.object(scheduler, "_propagate_owner_to_printer_manager", AsyncMock()),
patch.object(scheduler, "_power_off_if_needed", AsyncMock()),
]
if unlink_side_effect:
patches.append(patch.object(type(ctx.source_path), "unlink", unlink_side_effect))
with ExitStack() as stack:
for patcher in patches:
stack.enter_context(patcher)
async with ctx.session_maker() as db:
item = await db.get(PrintQueueItem, ctx.queue_item_id)
await scheduler._start_print(db, item)
async def _queue_snapshot(ctx):
async with ctx.session_maker() as db:
item = await db.get(PrintQueueItem, ctx.queue_item_id)
library_file = await db.get(LibraryFile, ctx.library_file_id)
archive = await db.get(PrintArchive, item.archive_id) if item.archive_id else None
return item, library_file, archive
@pytest.mark.asyncio
async def test_cleanup_unlinks_library_file_and_removes_db_row(queue_factory):
ctx = await queue_factory(cleanup=True)
await _dispatch_library_item(ctx)
item, library_file, archive = await _queue_snapshot(ctx)
assert item.status == "printing"
assert item.library_file_id is None
assert item.archive_id == archive.id
assert library_file is None
assert not ctx.source_path.exists()
@pytest.mark.asyncio
async def test_external_library_file_skips_cleanup(queue_factory):
ctx = await queue_factory(cleanup=True, is_external=True)
await _dispatch_library_item(ctx)
item, library_file, archive = await _queue_snapshot(ctx)
assert item.status == "printing"
assert item.library_file_id == ctx.library_file_id
assert item.archive_id == archive.id
assert library_file is not None
assert ctx.source_path.exists()
@pytest.mark.asyncio
async def test_archive_creation_failure_skips_cleanup_and_dispatch(queue_factory):
ctx = await queue_factory(cleanup=True, thumbnail_path="relative")
await _dispatch_library_item(ctx, archive_failure=True)
item, library_file, archive = await _queue_snapshot(ctx)
assert item.status == "failed"
assert item.error_message == "Failed to create archive from library file"
assert item.archive_id is None
assert archive is None
assert library_file is not None
assert ctx.source_path.exists()
assert ctx.thumbnail_path.exists()
ctx.upload.assert_not_awaited()
ctx.start_print.assert_not_called()
@pytest.mark.parametrize("thumbnail_path", ["absolute", "relative"])
@pytest.mark.asyncio
async def test_cleanup_resolves_absolute_and_relative_thumbnail_paths(queue_factory, thumbnail_path):
ctx = await queue_factory(cleanup=True, thumbnail_path=thumbnail_path)
await _dispatch_library_item(ctx)
item, library_file, archive = await _queue_snapshot(ctx)
assert item.status == "printing"
assert item.archive_id == archive.id
assert library_file is None
assert not ctx.source_path.exists()
assert not ctx.thumbnail_path.exists()
@pytest.mark.asyncio
async def test_archive_copy_survives_library_cleanup(queue_factory):
ctx = await queue_factory(cleanup=True)
await _dispatch_library_item(ctx)
assert not ctx.source_path.exists()
assert ctx.archive_path.exists()
assert ctx.archive_path.read_bytes() == b"library source"
uploaded_path = ctx.upload.await_args.args[2]
assert uploaded_path == ctx.archive_path
async def _sibling_snapshot(ctx):
async with ctx.session_maker() as db:
return [await db.get(PrintQueueItem, sid) for sid in ctx.sibling_ids]
async def _variant_files(ctx, sibling_id):
async with ctx.session_maker() as db:
rows = await db.execute(
select(PrintQueueVariant.library_file_id).where(PrintQueueVariant.queue_item_id == sibling_id)
)
return sorted(rows.scalars().all())
# ---------------------------------------------------------------------------
# Sibling copies of the same library row (#2819)
#
# `quantity > 1` on the printer-card upload-and-print flow puts the cleanup flag
# on every copy, and batch clones inherit `library_file_id`. Consuming the row
# for the first copy used to leave the others pointing at it, which failed with
# "Library file not found" on SQLite and deleted the rows outright on
# PostgreSQL, where the FK cascade is enforced.
#
# These run on SQLite, so they cover the orphan half directly. The cascade half
# was verified by hand against a real PostgreSQL 16, building this same fixture
# on both backends and comparing every row: without the fix the copies were gone
# after the delete -- including the finished ones a batch order counts its
# progress from, and a copy already printing from its own archive. With it, the
# two backends agree row for row. `print_archives.library_file_id` is SET NULL,
# so it is cleared by the same delete, which is why looking the archive up by
# the consumed library id -- the obvious alternative fix -- cannot work there.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_pending_copies_are_repointed_at_the_archive(queue_factory):
ctx = await queue_factory(cleanup=True, siblings=({}, {}))
await _dispatch_library_item(ctx)
item, library_file, archive = await _queue_snapshot(ctx)
assert library_file is None
for sibling in await _sibling_snapshot(ctx):
# Still queued -- the point is that they can now run, not that they run now.
assert sibling.status == "pending"
assert sibling.archive_id == archive.id
assert sibling.library_file_id is None
# Their file is already consumed; leaving this armed would delete
# whatever library row they were next given.
assert sibling.cleanup_library_after_dispatch is False
@pytest.mark.asyncio
async def test_copy_that_already_has_its_own_archive_keeps_it(queue_factory):
ctx = await queue_factory(cleanup=True, siblings=({"own_archive": True, "status": "printing"},))
await _dispatch_library_item(ctx)
_, _, archive = await _queue_snapshot(ctx)
(sibling,) = await _sibling_snapshot(ctx)
# It is mid-print from its own archive and does not need the library file.
# Re-pointing it would swap the file under a job already running.
assert sibling.archive_id != archive.id
assert sibling.status == "printing"
# Cleared all the same: on PostgreSQL a row still naming the file goes with
# it, and this one is a job that is currently printing.
assert sibling.library_file_id is None
@pytest.mark.parametrize("status", ["completed", "failed", "cancelled", "aborted"])
@pytest.mark.asyncio
async def test_finished_copies_keep_their_outcome_and_survive_the_delete(queue_factory, status):
ctx = await queue_factory(cleanup=True, siblings=({"status": status},))
await _dispatch_library_item(ctx)
(sibling,) = await _sibling_snapshot(ctx)
# A finished row is a record of what happened, not a spare part -- it keeps
# its outcome and is not handed the archive.
assert sibling.status == status
assert sibling.archive_id is None
# But the reference has to go: it is the only thing tying the row to the
# cascade that would otherwise delete it, and a batch order counts its
# progress from rows exactly like this one.
assert sibling.library_file_id is None
@pytest.mark.asyncio
async def test_skipped_copy_is_repointed_because_it_can_come_back(queue_factory):
ctx = await queue_factory(cleanup=True, siblings=({"status": "skipped"},))
await _dispatch_library_item(ctx)
_, _, archive = await _queue_snapshot(ctx)
(sibling,) = await _sibling_snapshot(ctx)
# Clearing the printer's previous-success gate puts skipped items back to
# pending, so this one is only waiting -- not finished.
assert sibling.status == "skipped"
assert sibling.archive_id == archive.id
assert sibling.library_file_id is None
@pytest.mark.asyncio
async def test_copies_are_untouched_when_the_dispatch_does_not_consume_the_file(queue_factory):
ctx = await queue_factory(cleanup=False, siblings=({},))
await _dispatch_library_item(ctx)
item, library_file, _ = await _queue_snapshot(ctx)
assert library_file is not None
(sibling,) = await _sibling_snapshot(ctx)
assert sibling.library_file_id == ctx.library_file_id
assert sibling.archive_id is None
@pytest.mark.asyncio
async def test_cross_model_copy_keeps_its_other_candidate_instead_of_the_archive(queue_factory):
ctx = await queue_factory(cleanup=True, siblings=({"variants": True, "extra_variant": True},))
await _dispatch_library_item(ctx)
(sibling,) = await _sibling_snapshot(ctx)
# It still has somewhere to go, and that candidate carries its own target
# model -- pointing it at this archive would print a file the matcher never
# chose.
assert sibling.archive_id is None
assert sibling.library_file_id is None
assert await _variant_files(ctx, sibling.id) == [ctx.other_library_file_id]
@pytest.mark.asyncio
async def test_copy_whose_only_candidate_was_consumed_is_repointed(queue_factory):
ctx = await queue_factory(cleanup=True, siblings=({"variants": True},))
await _dispatch_library_item(ctx)
_, _, archive = await _queue_snapshot(ctx)
(sibling,) = await _sibling_snapshot(ctx)
# Its one candidate is gone. Without the re-point the resolver would hold it
# pending forever with nothing left to dispatch.
assert sibling.archive_id == archive.id
assert sibling.library_file_id is None
assert await _variant_files(ctx, sibling.id) == []
@pytest.mark.asyncio
async def test_oserror_during_unlink_logs_orphan_path_and_does_not_crash_dispatch(queue_factory, caplog):
ctx = await queue_factory(cleanup=True, thumbnail_path="relative")
original_unlink = type(ctx.source_path).unlink
def unlink_with_source_failure(path, *args, **kwargs):
if Path(path) == ctx.source_path:
raise OSError("permission denied")
return original_unlink(path, *args, **kwargs)
with caplog.at_level("WARNING", logger="backend.app.services.print_scheduler"):
await _dispatch_library_item(ctx, unlink_side_effect=unlink_with_source_failure)
item, library_file, archive = await _queue_snapshot(ctx)
assert item.status == "printing"
assert item.archive_id == archive.id
assert item.library_file_id is None
assert library_file is None
assert ctx.source_path.exists()
assert not ctx.thumbnail_path.exists()
assert ctx.archive_path.exists()
assert "TRANSIENT_LIBRARY_FILE_ORPHAN" in caplog.text
assert str(ctx.source_path) in caplog.text
assert "permission denied" in caplog.text
Reference in New Issue View Git Blame Copy Permalink