mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-17 19:36:28 +02:00
maziggy
af7874546a
Projects made of many distinct files that each need N prints (e.g. 13
plates x 10 sets = 130 prints) only had aggregate progress. Finding out
"how many times have I printed plate_7?" meant reading the Activity
Timeline line by line, unusable at 130 events.
Projects now take an optional Copies per File target. Every printable
file in the project's linked folders shows an X / N badge with a mini
progress bar (gray not started, amber in progress, green done), and the
progress card gains a Complete Sets bar - the minimum per-file count,
i.e. how many finished assemblies can be shipped right now. Without the
target, printable files show a plain printed-count badge.
Counting matches the aggregate project stats: completed runs only,
served by a new /projects/{id}/file-progress endpoint. Runs attribute
to a file via a new library_file_id stamp on queue-dispatched archives,
falling back to content hash and then filename for historical rows.
Also fixed: files queued from a project-linked File Manager folder now
inherit that project, so their prints count toward project statistics -
previously only prints started from the project page were attributed.
Test-harness fix along the way: the test suite's get_db override never
committed, unlike production get_db, so endpoints relying on the
request-scoped commit silently lost their writes in tests. The override
now mirrors production commit/rollback semantics.
288 lines
11 KiB
Python
288 lines
11 KiB
Python
from contextlib import ExitStack
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
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
|
|
from backend.app.models.printer import Printer
|
|
from backend.app.services.print_scheduler import PrintScheduler
|
|
|
|
|
|
@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):
|
|
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.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,
|
|
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,
|
|
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()),
|
|
patch("backend.app.services.print_scheduler.spawn_background_task", MagicMock()),
|
|
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
|
|
|
|
|
|
@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
|