bambuddy/backend/tests/integration/test_user_delete_in_flight_work.py

"""Work a deleted user started must not be credited to whoever gets their id.

A running print, a slice, a pipeline run and an open WebSocket all keep the
user id they started with, in memory, for minutes or hours. Written after the
delete, that id is a dangling reference, and on SQLite, which gives a freed id
to the next user created, it is someone else: they are credited with the print
and billed for it, own the sliced file, see the slice job and receive the
deleted user's notifications.
"""

from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest
from httpx import AsyncClient
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from backend.app import main
from backend.app.api.routes.library import slice_and_persist, slice_and_persist_as_archive
from backend.app.api.routes.pipeline_runs import _make_orchestration_callable
from backend.app.core.printer_scope import ALL_PRINTERS, PrinterScope
from backend.app.core.websocket import ws_manager
from backend.app.models.archive import PrintArchive
from backend.app.models.library import LibraryFile
from backend.app.models.pipeline_run import PipelineJob, PipelineRun
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.printer import Printer
from backend.app.models.slicer_pipeline import SlicerPipeline
from backend.app.models.user import User
from backend.app.schemas.slicer import SliceRequest, SliceResponse
from backend.app.services.printer_manager import printer_manager
from backend.app.services.slice_dispatch import SliceJob, slice_dispatch
from backend.app.services.slicer_api import SliceResult

pytestmark = [pytest.mark.integration, pytest.mark.asyncio]


async def _user(db, name: str) -> int:
    user = User(username=name, password_hash="x", role="user")
    db.add(user)
    await db.commit()
    return user.id


async def _replace_user(db, user_id: int) -> None:
    """Delete the user and create the next one, which gets the same id on SQLite."""
    await db.execute(delete(User).where(User.id == user_id))
    await db.commit()
    assert await _user(db, "bob") == user_id


def _slice_job(owner_id: int) -> SliceJob:
    job = SliceJob(id=987_654, kind="library_file", source_id=1, source_name="a.3mf", owner_id=owner_id)
    slice_dispatch._jobs[job.id] = job
    return job


def _slicer_that_runs_while(during):
    async def slice_(*args, **kwargs):
        await during()
        return SliceResult(b"PK\x03\x04 not-a-real-3mf", 3600, 12.5, 4200.0), False

    return patch("backend.app.api.routes.library._run_slicer_with_fallback", AsyncMock(side_effect=slice_))


async def test_deleting_a_user_forgets_what_the_app_holds_for_them(async_client: AsyncClient, db_session):
    alice = await _user(db_session, "alice")
    printer_manager.set_current_print_user(4242, alice, "alice")
    main._expected_print_creators[(4242, "a.3mf")] = alice
    job = _slice_job(alice)
    socket = SimpleNamespace(
        state=SimpleNamespace(
            bambuddy_principal_user_id=alice,
            bambuddy_scope_principal=("alice", None),
            bambuddy_printer_scope=None,
        ),
        close=AsyncMock(),
        send_text=AsyncMock(),
    )
    ws_manager.active_connections.append(socket)
    try:
        response = await async_client.delete(f"/api/v1/users/{alice}")
        assert response.status_code == 204, response.text

        # The print ends later; nobody is credited with it.
        assert printer_manager.get_current_print_user(4242) is None
        # Nor is the print-start email sent to whoever is given the id.
        assert (4242, "a.3mf") not in main._expected_print_creators
        # The slice job is no longer theirs, so the next user given the id can't poll it.
        assert job.owner_id is None
        # Their open dashboard is closed, not left to receive the next user's messages.
        socket.close.assert_awaited_once_with(code=4401)
        assert socket not in ws_manager.active_connections
    finally:
        printer_manager.clear_current_print_user(4242)
        main._expected_print_creators.pop((4242, "a.3mf"), None)
        slice_dispatch._jobs.pop(job.id, None)
        if socket in ws_manager.active_connections:
            ws_manager.active_connections.remove(socket)


async def test_a_slice_finishing_after_its_user_was_deleted_belongs_to_nobody(db_session):
    alice = await _user(db_session, "alice")
    job = _slice_job(alice)
    try:

        async def alice_is_deleted_and_bob_created():
            slice_dispatch.forget_owner(alice)  # what deleting her does
            await _replace_user(db_session, alice)

        with _slicer_that_runs_while(alice_is_deleted_and_bob_created):
            response = await slice_and_persist(
                db_session,
                model_bytes=b"model",
                model_filename="a.3mf",
                folder_id=None,
                extra_metadata=None,
                request=SliceRequest(printer_preset_id=1, process_preset_id=2, filament_preset_id=3),
                current_user_id=alice,
                job_id=job.id,
            )
    finally:
        slice_dispatch._jobs.pop(job.id, None)

    sliced = await db_session.get(LibraryFile, response.library_file_id)
    assert sliced.created_by_id is None


async def test_an_archive_slice_finishing_after_its_user_was_deleted_belongs_to_nobody(
    db_session, monkeypatch, tmp_path
):
    from backend.app.core.config import settings

    monkeypatch.setattr(settings, "base_dir", tmp_path)
    monkeypatch.setattr(settings, "archive_dir", tmp_path / "archive")
    alice = await _user(db_session, "alice")
    source = PrintArchive(filename="a.3mf", file_path="archive/a.3mf", file_size=1, status="completed")
    db_session.add(source)
    await db_session.commit()
    job = _slice_job(alice)
    try:

        async def alice_is_deleted_and_bob_created():
            slice_dispatch.forget_owner(alice)
            await _replace_user(db_session, alice)

        with _slicer_that_runs_while(alice_is_deleted_and_bob_created):
            response = await slice_and_persist_as_archive(
                db_session,
                model_bytes=b"model",
                model_filename="a.3mf",
                request=SliceRequest(printer_preset_id=1, process_preset_id=2, filament_preset_id=3),
                source_archive=source,
                current_user_id=alice,
                job_id=job.id,
            )
    finally:
        slice_dispatch._jobs.pop(job.id, None)

    sliced = await db_session.get(PrintArchive, response.archive_id)
    assert sliced.created_by_id is None


async def _run_pipeline_while_its_creator_is_replaced(
    db_session, test_engine, tmp_path, *, specific_printer: bool, printer_scope=ALL_PRINTERS
) -> list[PrintQueueItem]:
    alice = await _user(db_session, "alice")
    printer = Printer(name="P", serial_number="SPIPE1", ip_address="192.0.2.9", access_code="x", model="X1C")
    source = LibraryFile(filename="a.3mf", file_path="library/a.3mf", file_type="3mf", file_size=1)
    pipeline = SlicerPipeline(
        name="p",
        printer_preset_source="local",
        printer_preset_id="1",
        process_preset_source="local",
        process_preset_id="2",
        filament_presets_json='[{"source": "local", "id": "3"}]',
    )
    db_session.add_all([printer, source, pipeline])
    await db_session.commit()
    if specific_printer:
        pipeline.target_kind, pipeline.target_printer_id = "specific_printer", printer.id
    else:
        pipeline.target_kind, pipeline.target_model_class = "printer_class", "X1C"
    run = PipelineRun(pipeline_id=pipeline.id, source_library_file_id=source.id, copies=1, created_by=alice)
    db_session.add(run)
    await db_session.commit()
    db_session.add(PipelineJob(pipeline_run_id=run.id, copy_index=0))
    await db_session.commit()
    model = tmp_path / "a.3mf"
    model.write_bytes(b"model")

    async def slice_and_persist_while_alice_is_replaced(session, **kwargs):
        await _replace_user(db_session, alice)
        sliced = LibraryFile(filename="a.gcode.3mf", file_path="library/a.gcode.3mf", file_type="3mf", file_size=1)
        session.add(sliced)
        await session.flush()
        return SliceResponse(
            library_file_id=sliced.id, name="a", print_time_seconds=1, filament_used_g=1.0, filament_used_mm=1.0
        )

    orchestrate = _make_orchestration_callable(
        run_id=run.id,
        pipeline_id=pipeline.id,
        src_kind="library_file",
        src_id=source.id,
        src_filename="a.3mf",
        src_path=Path(model),
        creator_user_id=alice,
        copies=1,
        printer_scope=printer_scope,
    )
    with (
        patch(
            "backend.app.api.routes.pipeline_runs.async_session",
            async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False),
        ),
        patch("backend.app.api.routes.library.slice_and_persist", new=slice_and_persist_while_alice_is_replaced),
        patch("backend.app.api.routes.pipeline_runs._publish_run_event", new=AsyncMock()),
    ):
        await orchestrate(1)

    statement = select(PrintQueueItem).execution_options(populate_existing=True)
    return list((await db_session.execute(statement)).scalars().all())


async def test_a_pipeline_run_finishing_after_its_user_was_deleted_queues_ownerless_copies(
    db_session, test_engine, tmp_path
):
    copies = await _run_pipeline_while_its_creator_is_replaced(db_session, test_engine, tmp_path, specific_printer=True)

    assert [copy.created_by_id for copy in copies] == [None]


async def test_any_model_copies_of_a_deleted_restricted_user_wait_for_an_admin(db_session, test_engine, tmp_path):
    """As delete_user does with such jobs that already exist: without an owner the
    scheduler would run them on any printer of the model, not only on theirs."""
    copies = await _run_pipeline_while_its_creator_is_replaced(
        db_session, test_engine, tmp_path, specific_printer=False, printer_scope=PrinterScope(frozenset({1}))
    )

    assert [(copy.printer_id, copy.created_by_id, copy.manual_start) for copy in copies] == [(None, None, True)]


async def test_an_ownerless_run_is_not_announced_to_everyone(db_session):
    """broadcast_to_user(None) goes to every connection; with auth on, a run whose
    user was deleted is nobody's to watch (the runs page polls)."""
    from backend.app.api.routes.pipeline_runs import _publish_run_event
    from backend.app.api.routes.settings import set_setting

    run = PipelineRun(copies=1, created_by=None)
    db_session.add(run)
    await db_session.commit()

    with patch("backend.app.api.routes.pipeline_runs.ws_manager.broadcast_to_user", new=AsyncMock()) as sent:
        await _publish_run_event(db_session, run)
        assert sent.await_count == 1  # auth off: one user, everything is theirs
        await set_setting(db_session, "auth_enabled", "true")
        await db_session.commit()
        await _publish_run_event(db_session, run)
        assert sent.await_count == 1


async def test_a_finished_print_is_credited_to_its_user_unless_they_were_deleted(db_session):
    from backend.app.main import _print_user_id_if_still_there

    alice = await _user(db_session, "alice")
    # Renamed while the print ran: still theirs.
    assert await _print_user_id_if_still_there(db_session, {"user_id": alice, "username": "old name"}) == alice
    await db_session.execute(delete(User).where(User.id == alice))
    await db_session.commit()
    assert await _print_user_id_if_still_there(db_session, {"user_id": alice, "username": "alice"}) is None


async def test_a_queue_print_whose_user_was_deleted_during_the_upload_is_nobodys(db_session, test_engine, monkeypatch):
    """The scheduler loaded the user and the item before the upload; after it,
    neither may bring the deleted user back (and with them, whoever gets the id)."""
    from backend.app.services import print_scheduler as scheduler_module
    from backend.app.services.print_scheduler import PrintScheduler

    alice = await _user(db_session, "alice")
    printer = Printer(name="P", serial_number="SUPL1", ip_address="192.0.2.10", access_code="x")
    db_session.add(printer)
    await db_session.commit()
    item = PrintQueueItem(printer_id=printer.id, position=1, created_by_id=alice)
    db_session.add(item)
    await db_session.commit()
    assert await db_session.get(User, alice) is not None  # held by the dispatch, as _start_print does

    other = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    async with other() as admin_db:  # meanwhile, through the API
        await admin_db.execute(delete(User).where(User.id == alice))
        await admin_db.commit()
        admin_db.add(User(username="bob", password_hash="x", role="user"))
        await admin_db.commit()

    credited: list = []
    monkeypatch.setattr(scheduler_module.printer_manager, "set_current_print_user", lambda *args: credited.append(args))
    await PrintScheduler()._propagate_owner_to_printer_manager(db_session, item)

    assert credited == []