bambuddy/backend/tests/unit/test_fallback_name_keeps_project_3126.py

"""Rejecting the wrong plate must not cost the archive its name (#3126).

The reporter's X2D was sent a print from Bambu Studio, which filed it on
internal eMMC (``"url": "brtc://emmc/..."``). FTPS cannot serve that, but the
bounded probe found a *same-named* file at the card's root -- an earlier slice
of the same project, plate 4, while the running print was plate 1. #1204's
guard caught the contradiction and refused the file, which is right: archiving
plate 4's thumbnail, filament and cost against this print is the swap #2957
removed.

What it then did with the name was not. The guard asks ``swap_plate_suffix``
for a corrected name and blanked ``subtask_name`` whenever it came back None --
but None covers two unrelated cases, and only one of them is a name that could
mislead. ``防雨防虫_模块化_排气口（50_75_80_100管可用）`` carries no
``- Plate N`` suffix at all, so it holds no stale plate number to be wrong
about; blanking it dropped the project name too and the row fell through to the
gcode_file path, titled ``plate_1``. #1204's own premise is consecutive plates
*of the same model*, so the project part of a lagging name is right either way.

Fixing that is only half of it, and the other half is why the title lives in its
own variable. The name is kept for *display*; every lookup still disowns it,
``_active_prints`` included. Key the row under a name the guard just watched
fetch the wrong plate and the cover endpoint -- which downloads that very name
for the running print's thumbnail -- hands the bytes to the recovery path, which
checks a candidate is a readable 3MF and never which plate it holds. The row
would be filled in with the file this branch had just deleted.

Pinned here: a name without a plate suffix survives the rejection intact as the
title, a name with a stale one still gets its number corrected, and the rejected
name keys nothing.
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from backend.app.main import (
    _active_prints,
    _expected_print_creators,
    _expected_print_registered_at,
    _expected_prints,
    _print_ams_mappings,
    _timelapse_baselines,
)

pytestmark = pytest.mark.unit

DISPATCH = "/data/Metadata/plate_1.gcode"
# The reporter's own subtask_name. Non-ASCII and parenthesised, with no plate
# suffix anywhere in it -- exactly the shape that used to be thrown away.
PROJECT = "防雨防虫_模块化_排气口（50_75_80_100管可用）"


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


def _printer():
    printer = MagicMock()
    printer.id = 1
    printer.auto_archive = True
    printer.external_camera_enabled = False
    printer.external_camera_url = None
    # Every unset MagicMock attribute is truthy, and leaving this one implicit
    # runs the plate-detection camera grab against a printer that is not there.
    printer.plate_detection_enabled = False
    printer.name = "X2D"
    printer.model = "X2D"
    printer.ip_address = "172.25.12.149"
    printer.access_code = "12345678"
    return printer


async def _run_print_start(subtask: str):
    """Drive on_print_start to the wrong-plate rejection and return the row.

    Every download succeeds and every 3MF peeks as plate 4, while the dispatch
    says plate 1 -- so the initial fetch is rejected and no re-download can
    satisfy the guard either, which is the reporter's sequence.
    """
    printer = _printer()

    def execute_router(stmt, *args, **kwargs):
        sql = str(stmt).lower()
        if "from printers" in sql or "from printer " in sql:
            return MagicMock(
                scalar_one_or_none=MagicMock(return_value=printer),
                scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[printer]))),
            )
        return MagicMock(
            scalar_one_or_none=MagicMock(return_value=None),
            scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
        )

    added: list = []
    session = AsyncMock()
    session.__aenter__ = AsyncMock(return_value=session)
    session.__aexit__ = AsyncMock()
    session.execute = AsyncMock(side_effect=execute_router)
    session.commit = AsyncMock()
    session.refresh = AsyncMock()
    session.add = MagicMock(side_effect=added.append)

    state = MagicMock(
        current_project_url=f"ftp://{subtask}.gcode.3mf",
        sdcard=True,
        sdcard_reported=True,
    )

    with (
        patch("backend.app.main.async_session") as session_maker,
        patch("backend.app.main.notification_service") as notif,
        patch("backend.app.main.smart_plug_manager") as plug,
        patch("backend.app.main.ws_manager") as ws,
        patch("backend.app.main.mqtt_relay") as relay,
        patch("backend.app.main.printer_manager") as pm,
        patch("backend.app.main.download_file_async", new=AsyncMock(return_value=True)),
        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"),
        # Plate 4 on the card, plate 1 on the printer -- the mismatch itself.
        patch("backend.app.main.peek_plate_index_in_3mf", return_value=4),
        # Imported inside the function, so patching it anywhere else lets the
        # directory walk open real sockets and the test hangs on connect.
        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),
        # Real, it would spawn a task that outlives the test by a minute.
        patch("backend.app.main._schedule_fallback_3mf_retry", new=MagicMock()),
    ):
        session_maker.return_value = session
        notif.on_print_start = AsyncMock()
        plug.on_print_start = AsyncMock()
        ws.send_print_start = AsyncMock()
        ws.send_archive_updated = AsyncMock()
        # Awaited between creating the fallback row and the rest of the
        # handler: a plain MagicMock raises, and the handler swallows it.
        ws.send_archive_created = AsyncMock()
        relay.on_print_start = AsyncMock()
        pm.get_status = MagicMock(return_value=state)
        pm.get_printer = MagicMock(return_value=MagicMock(serial_number="TEST3126"))

        from backend.app.main import on_print_start

        await on_print_start(1, {"filename": DISPATCH, "subtask_name": subtask})

    # Snapshot before the autouse fixture clears it in teardown.
    keys = {name for (_pid, name) in _active_prints}
    for row in added:
        extra = getattr(row, "extra_data", None)
        if isinstance(extra, dict) and extra.get("no_3mf_available"):
            return row, keys
    return None, keys


class TestANameWithNoPlateSuffixSurvives:
    @pytest.mark.asyncio
    async def test_the_reported_case_keeps_the_project_name(self):
        """The regression. This row used to be titled ``plate_1``."""
        row, _keys = await _run_print_start(PROJECT)

        assert row is not None
        assert row.print_name == PROJECT

    @pytest.mark.asyncio
    async def test_the_original_subtask_is_still_recorded(self):
        """Nothing reads this field today -- it is there for support, and a
        row that records the dispatch path under both its name and its
        subtask tells whoever reads the bundle nothing about the print."""
        row, _keys = await _run_print_start(PROJECT)

        assert row.extra_data["original_subtask"] == PROJECT

    @pytest.mark.asyncio
    async def test_an_ascii_name_too(self):
        """Nothing here is about the encoding -- any single-plate project name
        reaches the same branch."""
        row, _keys = await _run_print_start("Fan_Shroud")

        assert row.print_name == "Fan_Shroud"


class TestAStalePlateSuffixIsStillCorrected:
    """#1204's actual fix, which the change above must not undo."""

    @pytest.mark.asyncio
    async def test_the_spaced_form_gets_the_running_plate(self):
        row, _keys = await _run_print_start("Fan_Shroud - Plate 4")

        assert row.print_name == "Fan_Shroud - Plate 1"

    @pytest.mark.asyncio
    async def test_the_underscored_form_too(self):
        row, _keys = await _run_print_start("Fan_Shroud_plate_4")

        assert row.print_name == "Fan_Shroud_plate_1"


class TestTheDisownedNameStillFindsNoFiles:
    """The other half, and the reason the name is kept in its own variable.

    A name the guard just watched fetch another plate's 3MF must not key
    ``_active_prints``. The cover endpoint downloads that same name for the
    running print's thumbnail and offers the bytes to
    ``try_recover_fallback_archive``, which matches on those keys and hands
    whatever it gets to ``_recover_fallback_archive`` -- and that checks a
    candidate is a readable 3MF, never which plate it holds. Key the row under
    the rejected name and the cover endpoint fills it in with the exact file
    this branch just deleted, which is #2957's swap coming back in through a
    different door.
    """

    @pytest.mark.asyncio
    async def test_the_rejected_name_is_not_registered(self):
        row, keys = await _run_print_start(PROJECT)

        assert row.print_name == PROJECT, "the title is the whole point of the fix"
        assert PROJECT not in keys
        assert f"{PROJECT}.3mf" not in keys

    @pytest.mark.asyncio
    async def test_the_dispatch_path_still_is(self):
        """Disowning the subtask name must not leave the archive unfindable at
        print completion -- the gcode_file key is what matches it there."""
        _row, keys = await _run_print_start(PROJECT)

        assert DISPATCH in keys

    @pytest.mark.asyncio
    async def test_a_corrected_name_is_registered(self):
        """#1204's case is different: the swapped name points at the plate that
        really is running, so a file found under it is the right file."""
        _row, keys = await _run_print_start("Fan_Shroud - Plate 4")

        assert "Fan_Shroud - Plate 1" in keys
        assert "Fan_Shroud - Plate 4" not in keys