bambuddy/backend/tests/integration/test_print_start_name_match_ap2.py

"""A print matched to a 'printing' archive by name only is a new print when
the subtask ids show it.

An archive can be left in 'printing' when its completion was missed. Started
again within two hours, the same file used to be folded into that archive:
no new history entry, no timelapse, no cost estimate. Two different subtask
ids are two prints, unless a queue job of this printer is running that
archive: a Bambuddy reprint mints a fresh id for the same row (#1807).
"""

from contextlib import ExitStack
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from backend.app.main import (
    _active_prints,
    _expected_print_creators,
    _expected_print_registered_at,
    _expected_prints,
    _print_ams_mappings,
)
from backend.app.models.archive import PrintArchive
from backend.app.models.print_queue import PrintQueueItem

DISPATCH = "/data/Metadata/plate_1.gcode"
SUBTASK = "Bracket_plate_1"


@pytest.fixture(autouse=True)
def _clear_print_state():
    dicts = (
        _expected_prints,
        _expected_print_registered_at,
        _expected_print_creators,
        _print_ams_mappings,
        _active_prints,
    )
    for d in dicts:
        d.clear()
    yield
    for d in dicts:
        d.clear()


async def _start(test_engine, printer_id: int, subtask_id: str, last_dispatch: str | None = None) -> None:
    session_maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    state = MagicMock(
        current_project_url=f"ftp://{SUBTASK}.gcode.3mf",
        sdcard=True,
        sdcard_reported=True,
        subtask_id=subtask_id,
        progress=0.0,
    )
    patches = [
        patch("backend.app.main.async_session", session_maker),
        patch("backend.app.core.database.async_session", session_maker),
        patch("backend.app.main.download_file_async", new=AsyncMock(return_value=False)),
        patch("backend.app.main.download_file_try_paths_async", new=AsyncMock(return_value=None)),
        patch("backend.app.main.get_cached_3mf", return_value=None),
        patch("backend.app.main.cache_3mf_download"),
        patch("backend.app.main.peek_plate_index_in_3mf", return_value=None),
        patch("backend.app.services.bambu_ftp.list_files_async", new=AsyncMock(return_value=[])),
        patch("backend.app.main.ftps_handshake_blocked", return_value=False),
        patch("backend.app.main.get_ftp_retry_settings", new=AsyncMock(return_value=(False, 3, 2.0, 30))),
        patch("backend.app.main._record_energy_start", new_callable=AsyncMock),
        patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock),
        patch("backend.app.main._maybe_start_layer_timelapse"),
        patch("backend.app.main._capture_timelapse_baseline_at_start", new_callable=AsyncMock),
        patch("backend.app.main._schedule_fallback_3mf_retry"),
        patch("backend.app.main._store_spoolman_print_data", new_callable=AsyncMock),
        patch("backend.app.main._load_objects_from_archive"),
        patch("backend.app.services.usage_tracker.on_print_start", new_callable=AsyncMock),
    ]
    with ExitStack() as stack:
        for p in patches:
            stack.enter_context(p)
        notif = stack.enter_context(patch("backend.app.main.notification_service"))
        plug = stack.enter_context(patch("backend.app.main.smart_plug_manager"))
        ws = stack.enter_context(patch("backend.app.main.ws_manager"))
        relay = stack.enter_context(patch("backend.app.main.mqtt_relay"))
        pm = stack.enter_context(patch("backend.app.main.printer_manager"))
        notif.on_print_start = AsyncMock()
        plug.on_print_start = AsyncMock()
        ws.send_print_start = AsyncMock()
        ws.send_archive_created = AsyncMock()
        ws.send_archive_updated = AsyncMock()
        relay.on_print_start = AsyncMock()
        relay.on_archive_created = AsyncMock()
        pm.get_status = MagicMock(return_value=state)
        pm.get_printer = MagicMock(return_value=MagicMock(serial_number="TESTAP2"))
        pm.get_client = MagicMock(
            return_value=MagicMock(last_dispatch_subtask_id=last_dispatch) if last_dispatch else None
        )

        from backend.app.main import on_print_start

        await on_print_start(
            printer_id, {"filename": DISPATCH, "subtask_name": SUBTASK, "raw_data": {"subtask_id": subtask_id}}
        )


async def _stuck_archive(db_session, printer_id: int, subtask_id: str | None) -> int:
    archive = PrintArchive(
        printer_id=printer_id,
        filename=f"{SUBTASK}.gcode.3mf",
        file_path=f"archives/test/{SUBTASK}.gcode.3mf",
        file_size=2048,
        print_name=SUBTASK,
        status="printing",
        subtask_id=subtask_id,
        started_at=datetime.now(timezone.utc) - timedelta(minutes=30),
    )
    db_session.add(archive)
    await db_session.commit()
    return archive.id


async def _archives(db_session, printer_id: int) -> list[PrintArchive]:
    db_session.expire_all()
    result = await db_session.execute(
        select(PrintArchive).where(PrintArchive.printer_id == printer_id).order_by(PrintArchive.id)
    )
    return list(result.scalars().all())


@pytest.mark.asyncio
@pytest.mark.integration
async def test_another_subtask_id_is_a_new_print(test_engine, db_session, printer_factory):
    printer = await printer_factory()
    stuck_id = await _stuck_archive(db_session, printer.id, "111")
    _active_prints[(printer.id, f"{SUBTASK}.gcode.3mf")] = stuck_id
    _active_prints[(printer.id, "old/plate_1.gcode")] = stuck_id

    await _start(test_engine, printer.id, "222")

    rows = await _archives(db_session, printer.id)
    stuck = next(a for a in rows if a.id == stuck_id)
    assert (stuck.status, stuck.failure_reason) == ("cancelled", "noStatusUpdate")
    new = [a for a in rows if a.id != stuck_id]
    assert len(new) == 1 and new[0].subtask_id == "222"
    assert stuck_id not in _active_prints.values()


@pytest.mark.asyncio
@pytest.mark.integration
async def test_a_queue_reprint_of_the_archive_resumes_it(test_engine, db_session, printer_factory):
    printer = await printer_factory()
    stuck_id = await _stuck_archive(db_session, printer.id, "111")
    db_session.add(PrintQueueItem(printer_id=printer.id, archive_id=stuck_id, status="printing"))
    await db_session.commit()

    await _start(test_engine, printer.id, "222")

    rows = await _archives(db_session, printer.id)
    assert [(a.id, a.status, a.subtask_id) for a in rows] == [(stuck_id, "printing", "222")]


@pytest.mark.asyncio
@pytest.mark.integration
async def test_an_archive_without_a_subtask_id_is_resumed(test_engine, db_session, printer_factory):
    printer = await printer_factory()
    stuck_id = await _stuck_archive(db_session, printer.id, None)

    await _start(test_engine, printer.id, "222")

    rows = await _archives(db_session, printer.id)
    assert [(a.id, a.status, a.subtask_id) for a in rows] == [(stuck_id, "printing", "222")]


@pytest.mark.asyncio
@pytest.mark.integration
async def test_a_cancelled_archive_with_the_same_subtask_id_is_not_reopened(test_engine, db_session, printer_factory):
    """Completion already charged and logged it; reopening would do it twice."""
    printer = await printer_factory()
    old_id = await _stuck_archive(db_session, printer.id, "333")
    old = await db_session.get(PrintArchive, old_id)
    old.status, old.failure_reason = "cancelled", "noStatusUpdate"
    await db_session.commit()

    await _start(test_engine, printer.id, "333")

    rows = await _archives(db_session, printer.id)
    assert next(a for a in rows if a.id == old_id).status == "cancelled"
    assert [a.subtask_id for a in rows if a.id != old_id] == ["333"]
    assert old_id not in _active_prints.values()


async def _queue_item(db_session, printer_id: int, archive_id: int, status: str = "printing") -> None:
    db_session.add(PrintQueueItem(printer_id=printer_id, archive_id=archive_id, status=status))
    await db_session.commit()


@pytest.mark.asyncio
@pytest.mark.integration
async def test_a_queue_job_left_printing_does_not_take_another_print(test_engine, db_session, printer_factory):
    """The archive's queue job stayed 'printing' with it, but Bambuddy's last
    dispatch to this printer was another job id: this print is not that run."""
    printer = await printer_factory()
    stuck_id = await _stuck_archive(db_session, printer.id, "111")
    await _queue_item(db_session, printer.id, stuck_id)

    await _start(test_engine, printer.id, "222", last_dispatch="111")

    rows = await _archives(db_session, printer.id)
    assert next(a for a in rows if a.id == stuck_id).status == "cancelled"
    assert [a.subtask_id for a in rows if a.id != stuck_id] == ["222"]


@pytest.mark.asyncio
@pytest.mark.integration
async def test_the_dispatched_job_id_resumes_without_a_queue_lookup(test_engine, db_session, printer_factory):
    printer = await printer_factory()
    stuck_id = await _stuck_archive(db_session, printer.id, "111")
    await _queue_item(db_session, printer.id, stuck_id)

    await _start(test_engine, printer.id, "222", last_dispatch="222")

    rows = await _archives(db_session, printer.id)
    assert [(a.id, a.status, a.subtask_id) for a in rows] == [(stuck_id, "printing", "222")]


@pytest.mark.asyncio
@pytest.mark.integration
@pytest.mark.parametrize("elsewhere", ["other_printer", "finished"])
async def test_only_this_printers_running_queue_job_exempts(test_engine, db_session, printer_factory, elsewhere):
    printer = await printer_factory()
    stuck_id = await _stuck_archive(db_session, printer.id, "111")
    if elsewhere == "other_printer":
        other = await printer_factory()
        await _queue_item(db_session, other.id, stuck_id)
    else:
        await _queue_item(db_session, printer.id, stuck_id, status="completed")

    await _start(test_engine, printer.id, "222")

    rows = await _archives(db_session, printer.id)
    assert next(a for a in rows if a.id == stuck_id).status == "cancelled"