model-platform/backend/tests/test_storage_upload_status.py

"""Unit tests for ``upload_bytes_to_session`` failure-path status persistence.

P0-5 / B1: the route handler wraps every request in ``session_scope``
(``backend.dependencies.database_session``), which rolls back on
exception. A naive ``upload.upload_status = "failed"; raise HTTPException(...)``
loses the status flip and leaves the row stuck in ``created``/``uploading``
forever. The fix is ``_mark_upload_failed_and_raise`` which opens a
*separate* session from ``request.app.state.session_factory`` and commits
the status change on it before raising. The outer rollback is then a
no-op on the caller's session, AND the named-lock connection (acquired
inside the helper's caller for the size/hash/storage failure branches)
stays pristine so ``release_named_lock`` runs on the same connection that
ran ``GET_LOCK`` — closing the lock-leak window Codex flagged.

These tests exercise the helper directly + the expired-status branch of
``upload_bytes_to_session`` with a mocked session.
"""
from __future__ import annotations

from datetime import datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi import HTTPException

from backend.services.storage import (
    _mark_upload_failed_and_raise,
    upload_bytes_to_session,
)


def _make_upload(upload_id: str = "01UPL0000000000000000000A") -> SimpleNamespace:
    """Minimal UploadSessions row with the columns the helper touches."""
    return SimpleNamespace(
        upload_id=upload_id,
        upload_status="uploading",
        expires_at=None,
        expected_size_bytes=None,
        expected_hash=None,
        bucket_name="workspace",
        object_key="ws/user/file.bin",
        object_key_hash=b"\x00" * 32,
    )


class _AsyncContextManager:
    """Async context manager that yields ``session`` on enter.

    Used to mock the result of ``session_factory()`` without pulling in
    a real SQLAlchemy engine.
    """

    def __init__(self, session: object) -> None:
        self._session = session

    async def __aenter__(self) -> object:
        return self._session

    async def __aexit__(self, *_args: object) -> None:
        return None


async def test_mark_upload_failed_and_raise_persists_status() -> None:
    """The helper must open a fresh session, mark 'failed', commit, then
    raise — surviving both the surrounding ``session_scope`` rollback and
    the named-lock connection-pool leak the original implementation opened
    up by committing on the lock-bound session.
    """
    upload = _make_upload()
    fresh_session = SimpleNamespace(
        scalar=AsyncMock(return_value=upload),
        commit=AsyncMock(),
    )
    session_factory = MagicMock(return_value=_AsyncContextManager(fresh_session))
    request = SimpleNamespace(
        app=SimpleNamespace(state=SimpleNamespace(session_factory=session_factory)),
    )

    with pytest.raises(HTTPException) as excinfo:
        await _mark_upload_failed_and_raise(
            request, upload.upload_id, 409, "size mismatch",
        )

    assert excinfo.value.status_code == 409
    assert excinfo.value.detail == "size mismatch"
    assert upload.upload_status == "failed"
    # The fresh session — not the caller's — must have committed.
    fresh_session.scalar.assert_awaited_once()
    fresh_session.commit.assert_awaited_once()
    session_factory.assert_called_once()


async def test_mark_upload_failed_and_raise_raises_even_if_upload_missing() -> None:
    """If the row has been hard-deleted between the caller's lookup and the
    helper's separate-session write, the helper still raises HTTPException
    with the requested status/detail — it just skips the commit. The caller
    still gets the same error contract.
    """
    fresh_session = SimpleNamespace(
        scalar=AsyncMock(return_value=None),
        commit=AsyncMock(),
    )
    session_factory = MagicMock(return_value=_AsyncContextManager(fresh_session))
    request = SimpleNamespace(
        app=SimpleNamespace(state=SimpleNamespace(session_factory=session_factory)),
    )

    with pytest.raises(HTTPException) as excinfo:
        await _mark_upload_failed_and_raise(
            request, "01UPL0000000000000000000A", 503, "boom",
        )

    assert excinfo.value.status_code == 503
    assert excinfo.value.detail == "boom"
    # No row to update, so commit must NOT have been called.
    fresh_session.commit.assert_not_awaited()


async def test_upload_bytes_to_session_expired_commits_status() -> None:
    """The early ``if upload.expires_at < now`` branch must also
    commit the expired status, otherwise a client retrying after the
    expiry window would see ``created`` again and re-upload.

    The expired branch is OUTSIDE the named-lock critical section, so it
    can still commit on the caller's session safely — no separate-session
    detour is needed.
    """
    expired_upload = _make_upload()
    expired_upload.upload_status = "created"
    # expires_at < utcnow_naive() triggers the expired branch.
    expired_upload.expires_at = datetime.utcnow() - timedelta(minutes=1)

    session = SimpleNamespace(
        scalar=AsyncMock(return_value=expired_upload),
        commit=AsyncMock(),
        get=AsyncMock(return_value=None),
    )
    request = SimpleNamespace(
        app=SimpleNamespace(state=SimpleNamespace(object_stores={})),
    )

    with pytest.raises(HTTPException) as excinfo:
        await upload_bytes_to_session("01UPL0000000000000000000A", session, request)

    assert excinfo.value.status_code == 409
    assert "expired" in excinfo.value.detail
    assert expired_upload.upload_status == "expired"
    session.commit.assert_awaited()