Skyvern/tests/unit_tests/test_update_status_guard.py
Andrew Neilson 114b0ecee5
Some checks are pending
Run tests and pre-commit / Run tests and pre-commit hooks (push) Waiting to run
Run tests and pre-commit / Frontend Lint and Build (push) Waiting to run
Publish Fern Docs / run (push) Waiting to run
fix: prevent final→non-final status overwrite race in update_status (#4928)
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Shuchang Zheng <wintonzheng0325@gmail.com>
2026-03-02 03:12:49 +00:00

53 lines
1.6 KiB
Python

"""Tests for the update_status finalization guard in default_persistent_sessions_manager."""
from __future__ import annotations
import sys
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock
import pytest
# Stub the AWS module to avoid import-time boto session creation.
sys.modules.setdefault("skyvern.forge.sdk.api.aws", MagicMock())
from skyvern.forge.sdk.schemas.persistent_browser_sessions import ( # noqa: E402
PersistentBrowserSession,
PersistentBrowserSessionStatus,
)
from skyvern.webeye.default_persistent_sessions_manager import update_status # noqa: E402
SESSION_ID = "sess_1"
ORG_ID = "org_1"
NOW = datetime.now(timezone.utc)
def _make_session(status: str) -> PersistentBrowserSession:
return PersistentBrowserSession(
persistent_browser_session_id=SESSION_ID,
organization_id=ORG_ID,
status=status,
created_at=NOW,
modified_at=NOW,
)
@pytest.mark.parametrize(
"desired_status",
[
pytest.param(PersistentBrowserSessionStatus.running, id="non-final-to-final"),
pytest.param(PersistentBrowserSessionStatus.failed, id="final-to-final"),
],
)
@pytest.mark.asyncio
async def test_rejects_update_when_already_final(desired_status: str):
"""A finalized session must not accept any status update."""
db = AsyncMock()
db.get_persistent_browser_session.return_value = _make_session(
PersistentBrowserSessionStatus.completed,
)
result = await update_status(db, SESSION_ID, ORG_ID, desired_status)
assert result is None
db.update_persistent_browser_session.assert_not_called()