|
|
@@ -0,0 +1,234 @@
|
|
|
+"""'Only if the previous print succeeded' is about the printer a job lands on."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+from datetime import datetime, timezone
|
|
|
+from pathlib import Path
|
|
|
+from types import SimpleNamespace
|
|
|
+from unittest.mock import AsyncMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
+
|
|
|
+import backend.app.models # noqa: F401
|
|
|
+from backend.app.core.database import Base
|
|
|
+from backend.app.models.archive import PrintArchive
|
|
|
+from backend.app.models.print_queue import PrintQueueItem
|
|
|
+from backend.app.models.printer import Printer
|
|
|
+from backend.app.services.print_scheduler import MODEL_GATE_WAIT_REASON, PrintScheduler
|
|
|
+from backend.tests.unit.test_scheduler_busy_reasons_3018 import _scheduler
|
|
|
+
|
|
|
+pytestmark = pytest.mark.asyncio
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+async def farm(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)
|
|
|
+ base_dir = tmp_path / "farm"
|
|
|
+ (base_dir / "archives").mkdir(parents=True)
|
|
|
+ (base_dir / "archives" / "job.3mf").write_bytes(b"archive payload")
|
|
|
+ async with session_maker() as db:
|
|
|
+ printers = [
|
|
|
+ Printer(name=name, serial_number=name, ip_address=f"10.0.0.{i}", access_code="code", model="X1C")
|
|
|
+ for i, name in enumerate(("A", "B"), start=1)
|
|
|
+ ]
|
|
|
+ db.add_all(printers)
|
|
|
+ await db.flush()
|
|
|
+ archive = PrintArchive(
|
|
|
+ printer_id=printers[0].id,
|
|
|
+ filename="job.3mf",
|
|
|
+ file_path="archives/job.3mf",
|
|
|
+ file_size=15,
|
|
|
+ print_time_seconds=120,
|
|
|
+ status="completed",
|
|
|
+ sliced_for_model="X1C",
|
|
|
+ )
|
|
|
+ db.add(archive)
|
|
|
+ await db.flush()
|
|
|
+ ids = SimpleNamespace(a=printers[0].id, b=printers[1].id, archive=archive.id)
|
|
|
+ await db.commit()
|
|
|
+ try:
|
|
|
+ yield SimpleNamespace(session_maker=session_maker, base_dir=base_dir, ids=ids)
|
|
|
+ finally:
|
|
|
+ await engine.dispose()
|
|
|
+
|
|
|
+
|
|
|
+async def _last_print(farm, printer_id: int, status: str) -> None:
|
|
|
+ async with farm.session_maker() as db:
|
|
|
+ db.add(
|
|
|
+ PrintQueueItem(
|
|
|
+ printer_id=printer_id,
|
|
|
+ archive_id=farm.ids.archive,
|
|
|
+ status=status,
|
|
|
+ position=0,
|
|
|
+ completed_at=datetime(2026, 10, 1, tzinfo=timezone.utc),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ await db.commit()
|
|
|
+
|
|
|
+
|
|
|
+async def _model_job(farm) -> int:
|
|
|
+ async with farm.session_maker() as db:
|
|
|
+ item = PrintQueueItem(
|
|
|
+ target_model="X1C",
|
|
|
+ archive_id=farm.ids.archive,
|
|
|
+ status="pending",
|
|
|
+ position=1,
|
|
|
+ require_previous_success=True,
|
|
|
+ )
|
|
|
+ db.add(item)
|
|
|
+ await db.commit()
|
|
|
+ return item.id
|
|
|
+
|
|
|
+
|
|
|
+async def _run(farm, item_id: int, waiting=None) -> PrintQueueItem:
|
|
|
+ async def matcher(db, model, exclude, *args, **kwargs):
|
|
|
+ for printer_id in (farm.ids.a, farm.ids.b):
|
|
|
+ if printer_id not in exclude:
|
|
|
+ return printer_id, None
|
|
|
+ return None, "All matching printers are busy"
|
|
|
+
|
|
|
+ async with _scheduler(farm, idle=True) as scheduler:
|
|
|
+ with (
|
|
|
+ patch.object(scheduler, "_find_idle_printer_for_model", matcher),
|
|
|
+ patch(
|
|
|
+ "backend.app.services.notification_service.notification_service.on_queue_job_waiting",
|
|
|
+ waiting or AsyncMock(),
|
|
|
+ ),
|
|
|
+ ):
|
|
|
+ await scheduler.check_queue()
|
|
|
+ async with farm.session_maker() as db:
|
|
|
+ return await db.get(PrintQueueItem, item_id)
|
|
|
+
|
|
|
+
|
|
|
+async def test_a_printer_whose_last_print_failed_is_passed_over(farm):
|
|
|
+ await _last_print(farm, farm.ids.a, "failed")
|
|
|
+ item_id = await _model_job(farm)
|
|
|
+
|
|
|
+ item = await _run(farm, item_id)
|
|
|
+
|
|
|
+ assert item.status != "skipped"
|
|
|
+ assert item.printer_id == farm.ids.b
|
|
|
+
|
|
|
+
|
|
|
+async def test_the_job_waits_when_every_printer_failed_last(farm):
|
|
|
+ await _last_print(farm, farm.ids.a, "failed")
|
|
|
+ await _last_print(farm, farm.ids.b, "aborted")
|
|
|
+ item_id = await _model_job(farm)
|
|
|
+
|
|
|
+ item = await _run(farm, item_id)
|
|
|
+
|
|
|
+ assert item.status == "pending"
|
|
|
+ assert item.printer_id is None
|
|
|
+ assert item.waiting_reason == MODEL_GATE_WAIT_REASON
|
|
|
+
|
|
|
+
|
|
|
+async def test_the_check_reads_the_printer_it_is_given(db_session, printer_factory):
|
|
|
+ a, b = await printer_factory(), await printer_factory()
|
|
|
+ db_session.add(PrintQueueItem(printer_id=a.id, status="failed", completed_at=datetime.now(timezone.utc)))
|
|
|
+ job = PrintQueueItem(target_model="X1C", status="pending", require_previous_success=True)
|
|
|
+ db_session.add(job)
|
|
|
+ await db_session.commit()
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+
|
|
|
+ assert await scheduler._check_previous_success(db_session, job, printer_id=a.id) is False
|
|
|
+ assert await scheduler._check_previous_success(db_session, job, printer_id=b.id) is True
|
|
|
+
|
|
|
+
|
|
|
+async def test_a_row_ended_without_a_time_is_not_the_latest(db_session, printer_factory):
|
|
|
+ printer = await printer_factory()
|
|
|
+ db_session.add(PrintQueueItem(printer_id=printer.id, status="failed", completed_at=datetime.now(timezone.utc)))
|
|
|
+ await db_session.flush()
|
|
|
+ db_session.add(PrintQueueItem(printer_id=printer.id, status="cancelled", completed_at=None))
|
|
|
+ job = PrintQueueItem(printer_id=printer.id, status="pending", require_previous_success=True)
|
|
|
+ db_session.add(job)
|
|
|
+ await db_session.commit()
|
|
|
+
|
|
|
+ assert await PrintScheduler()._check_previous_success(db_session, job) is False
|
|
|
+
|
|
|
+
|
|
|
+async def test_cancelling_an_order_stamps_when_its_jobs_ended(
|
|
|
+ async_client, db_session, printer_factory, archive_factory
|
|
|
+):
|
|
|
+ from backend.tests.integration.test_print_batch_orders import _create_order, _queue_item
|
|
|
+
|
|
|
+ printer = await printer_factory()
|
|
|
+ archive = await archive_factory(printer.id)
|
|
|
+ order = await _create_order(async_client, archive.id, [{"plate_id": 1, "quantity_target": 2}])
|
|
|
+ item = await _queue_item(async_client, printer.id, archive.id, order["id"], plate_id=1)
|
|
|
+
|
|
|
+ response = await async_client.delete(f"/api/v1/queue/batches/{order['id']}")
|
|
|
+
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ db_session.expire_all()
|
|
|
+ row = await db_session.get(PrintQueueItem, item["id"])
|
|
|
+ assert row.status == "cancelled"
|
|
|
+ assert row.completed_at is not None
|
|
|
+
|
|
|
+
|
|
|
+async def test_a_job_that_waited_for_a_busy_printer_says_when_it_now_needs_you(farm):
|
|
|
+ await _last_print(farm, farm.ids.a, "failed")
|
|
|
+ await _last_print(farm, farm.ids.b, "failed")
|
|
|
+ item_id = await _model_job(farm)
|
|
|
+ async with farm.session_maker() as db:
|
|
|
+ (await db.get(PrintQueueItem, item_id)).waiting_reason = "Busy: A, B"
|
|
|
+ await db.commit()
|
|
|
+ waiting = AsyncMock()
|
|
|
+
|
|
|
+ await _run(farm, item_id, waiting)
|
|
|
+
|
|
|
+ waiting.assert_awaited_once()
|
|
|
+ assert waiting.await_args.kwargs["waiting_reason"] == MODEL_GATE_WAIT_REASON
|
|
|
+
|
|
|
+
|
|
|
+async def test_resuming_a_model_clears_the_gate_on_its_printers(async_client, db_session, printer_factory):
|
|
|
+ x1c_a = await printer_factory(model="X1C")
|
|
|
+ x1c_b = await printer_factory(model="X1C")
|
|
|
+ p1s = await printer_factory(model="P1S")
|
|
|
+ ids = {"a": x1c_a.id, "b": x1c_b.id, "p": p1s.id}
|
|
|
+ for printer_id in ids.values():
|
|
|
+ db_session.add(PrintQueueItem(printer_id=printer_id, status="failed", completed_at=datetime.now(timezone.utc)))
|
|
|
+ await db_session.commit()
|
|
|
+
|
|
|
+ response = await async_client.post("/api/v1/queue/model/X1C/resume")
|
|
|
+
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ assert response.json() == {"acknowledged": 2, "printers": 2}
|
|
|
+ db_session.expire_all()
|
|
|
+ from sqlalchemy import select
|
|
|
+
|
|
|
+ rows = (await db_session.execute(select(PrintQueueItem))).scalars().all()
|
|
|
+ acknowledged = {row.printer_id: row.gate_acknowledged for row in rows}
|
|
|
+ assert acknowledged == {ids["a"]: True, ids["b"]: True, ids["p"]: False}
|
|
|
+
|
|
|
+
|
|
|
+async def test_resuming_a_model_keeps_to_the_callers_printers(async_client, db_session, printer_factory):
|
|
|
+ from sqlalchemy import select
|
|
|
+
|
|
|
+ from backend.tests.integration.test_printer_scope_1727 import _admin_token, _auth, _group, _user
|
|
|
+
|
|
|
+ ours, theirs = await printer_factory(model="X1C"), await printer_factory(model="X1C")
|
|
|
+ ours_id, theirs_id = ours.id, theirs.id
|
|
|
+ for printer_id in (ours_id, theirs_id):
|
|
|
+ db_session.add(PrintQueueItem(printer_id=printer_id, status="failed", completed_at=datetime.now(timezone.utc)))
|
|
|
+ await db_session.commit()
|
|
|
+ token = await _admin_token(async_client)
|
|
|
+ perms = await _group(async_client, token, "perms", permissions=["printers:read", "queue:read", "queue:update_all"])
|
|
|
+ team = await _group(async_client, token, "team", printer_ids=[ours_id])
|
|
|
+ member = _auth((await _user(async_client, token, "member", [perms, team]))[0])
|
|
|
+
|
|
|
+ response = await async_client.post("/api/v1/queue/model/X1C/resume", headers=member)
|
|
|
+
|
|
|
+ assert response.json() == {"acknowledged": 1, "printers": 1}
|
|
|
+ db_session.expire_all()
|
|
|
+ rows = (await db_session.execute(select(PrintQueueItem))).scalars().all()
|
|
|
+ assert {row.printer_id: row.gate_acknowledged for row in rows} == {ours_id: True, theirs_id: False}
|
|
|
+
|
|
|
+
|
|
|
+def test_the_queue_page_knows_the_wait_reason():
|
|
|
+ """The queue page offers Resume for jobs waiting with exactly this reason."""
|
|
|
+ page = Path(__file__).resolve().parents[3] / "frontend" / "src" / "pages" / "QueuePage.tsx"
|
|
|
+ assert f"'{MODEL_GATE_WAIT_REASON}'" in page.read_text()
|