bambuddy/backend/tests/unit/test_completion_queue_item_match.py

"""A print completion must not close a queue row belonging to another print.

``on_print_complete`` finds its queue row by printer and ``status='printing'``
alone -- the MQTT payload carries no run identifier to match on -- so any
completion delivered for a printer closes whichever row happens to be printing.
That is fine while the only source of completions is the printer itself, and
wrong the moment one arrives from anywhere else: a live 14-hour print was closed
18 minutes in, and its plate 2 never dispatched, because a completion for an
unrelated subtask reached the same lookup.

These cover the guard that rules that out, and the deliberate decision to let
the unverifiable cases through rather than strand an item in ``printing``.
"""

import pytest

from backend.app.main import _completion_belongs_to_queue_item, _subtask_name_from_filename
from backend.app.models.archive import PrintArchive
from backend.app.models.print_queue import PrintQueueItem


class TestSubtaskNameFromFilename:
    """The dispatcher builds the subtask name off the archive file name, so
    stripping the extensions back off has to land on exactly what MQTT echoes."""

    @pytest.mark.parametrize(
        ("filename", "expected"),
        [
            ("AMS_Rack.gcode.3mf", "AMS_Rack"),
            ("AMS_Rack.3mf", "AMS_Rack"),
            ("plate.gcode", "plate"),
            # A dot in the model's own name is not an extension. Path.stem would
            # eat it and produce "My", which matches nothing.
            ("My.Model.3mf", "My.Model"),
            ("My.Model.gcode.3mf", "My.Model"),
            # Extensions are matched case-insensitively; the name is not.
            ("Cover.GCODE.3MF", "Cover"),
            # Only the file name matters -- archives store a path.
            ("archive/1/20260811_112435_AMS_Rack/AMS_Rack.gcode.3mf", "AMS_Rack"),
            # Nothing to strip.
            ("AMS_Rack", "AMS_Rack"),
        ],
    )
    def test_recovers_the_dispatched_subtask_name(self, filename, expected):
        assert _subtask_name_from_filename(filename) == expected


async def _seed(db, *, archive_filename: str | None) -> PrintQueueItem:
    """A printing queue item, optionally linked to an archive."""
    archive_id = None
    if archive_filename is not None:
        archive = PrintArchive(
            printer_id=1,
            filename=archive_filename,
            file_path=f"archive/1/{archive_filename}",
            file_size=1,
            status="printing",
        )
        db.add(archive)
        await db.flush()
        archive_id = archive.id

    item = PrintQueueItem(printer_id=1, status="printing", archive_id=archive_id)
    db.add(item)
    await db.flush()
    return item


@pytest.mark.asyncio
class TestCompletionBelongsToQueueItem:
    async def test_accepts_the_completion_for_its_own_print(self, db_session):
        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")

        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "AMS_Rack"}) is True

    async def test_rejects_a_completion_for_a_different_print(self, db_session):
        # The exact shape of the incident: the row was dispatched as AMS_Rack and
        # a completion for "Test" arrived on the same printer.
        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")

        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Test"}) is False

    async def test_matches_regardless_of_case(self, db_session):
        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")

        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "ams_rack"}) is True

    @pytest.mark.parametrize("subtask", [None, "", "   "])
    async def test_lets_an_unidentified_completion_through(self, db_session, subtask):
        # No subtask name to compare means unverifiable, not wrong. Refusing here
        # would leave the item printing forever and wedge the printer's queue,
        # which is the failure the indiscriminate lookup existed to avoid.
        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")

        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": subtask}) is True

    async def test_lets_an_archiveless_item_through(self, db_session):
        # Library-file dispatch links the archive after the fact; there is
        # nothing to compare against yet.
        item = await _seed(db_session, archive_filename=None)

        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Anything"}) is True

    async def test_lets_an_archive_without_a_filename_through(self, db_session):
        # `filename` is NOT NULL, but nothing stops it being empty.
        item = await _seed(db_session, archive_filename="")

        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Anything"}) is True

    async def test_lets_a_dangling_archive_reference_through(self, db_session):
        item = await _seed(db_session, archive_filename=None)
        item.archive_id = 999999

        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Anything"}) is True


class TestDisposableDatabaseGuard:
    """The suite must never be able to open a session against a real database.

    ``run_with_retry`` takes its session from ``backend.app.core.database``, not
    from the ``backend.app.main.async_session`` that most tests patch, so an
    unmocked completion path reaches the app's module-level engine. That engine
    is built from ``DATABASE_URL``; conftest redirects it to a throwaway SQLite
    file and asserts the redirect took, because the alternative is a suite that
    passes while having edited someone's live print history.
    """

    def test_the_app_engine_points_at_a_throwaway_sqlite_file(self):
        from backend.app.core.database import engine
        from backend.tests.conftest import _TEST_APP_DB_DIR

        assert engine.url.drivername.startswith("sqlite")
        assert str(engine.url.database).startswith(str(_TEST_APP_DB_DIR))

    def test_the_guard_rejects_a_real_database(self):
        from sqlalchemy.engine import make_url

        from backend.tests.conftest import _assert_disposable_database

        with pytest.raises(RuntimeError, match="Refusing to run tests"):
            _assert_disposable_database(
                make_url("postgresql+asyncpg://user:pw@192.168.0.2:5432/bambuddy"),
                "test",
            )

    def test_the_guard_rejects_another_sqlite_file(self):
        # A developer's own data/bambuddy.db is just as real as a server.
        from sqlalchemy.engine import make_url

        from backend.tests.conftest import _assert_disposable_database

        with pytest.raises(RuntimeError, match="Refusing to run tests"):
            _assert_disposable_database(make_url("sqlite+aiosqlite:///data/bambuddy.db"), "test")