bambuddy/backend/tests/unit/test_previous_success_per_printer.py

"""'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()