ss-tools/backend/tests/test_translate_scheduler_guard.py

# #region TestTranslateSchedulerGuard [C:3] [TYPE Module] [SEMANTICS test, translate, scheduler, guard, concurrency, stale, pending]
# @BRIEF Verify execute_scheduled_translation concurrency guard and stale-PENDING clearance contracts.
# @RELATION BINDS_TO -> [execute_scheduled_translation]
# @TEST_EDGE: concurrent_run_skips — active RUNNING concurrent run → function skips and logs skipped_concurrent event
# @TEST_EDGE: recent_pending_run_not_stale — recent PENDING run (30 min old) → function skips, not stale
# @TEST_EDGE: stale_pending_cleared_and_proceeds — stale PENDING run (>1h old) → cleared as FAILED, execution proceeds
# @TEST_EDGE: multiple_stale_pending_all_cleared — multiple stale PENDING runs → all cleared as FAILED, execution proceeds
# @TEST_INVARIANT: concurrency_guard -> VERIFIED_BY: [test_concurrent_run_skips, test_recent_pending_run_not_stale]
# @TEST_INVARIANT: stale_pending_clearance -> VERIFIED_BY: [test_stale_pending_cleared_and_proceeds, test_multiple_stale_pending_all_cleared]

from pathlib import Path
import sys
import asyncio

sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from datetime import UTC, datetime, timedelta
from unittest.mock import MagicMock, patch


def _make_coro_runner():
    """Mock runner whose dispatch() executes the coroutine synchronously via asyncio.run."""
    r = MagicMock()
    r.dispatch.side_effect = lambda coro: asyncio.run(coro)
    return r

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _make_most_recent_run(created_at_delta_days=60):
    """Build a MagicMock TranslationRun with given age."""
    run = MagicMock()
    run.id = "prev-run-1"
    run.job_id = "job-1"
    run.status = "COMPLETED"
    run.insert_status = "succeeded"
    run.created_at = datetime.now(UTC) - timedelta(days=created_at_delta_days)
    return run


def _make_mock_run(run_id="run-1", job_id="job-1"):
    """Build a mock TranslationRun returned by start_run."""
    run = MagicMock()
    run.id = run_id
    run.job_id = job_id
    run.status = "PENDING"
    run.trigger_type = None
    run.total_records = 0
    run.successful_records = 0
    run.failed_records = 0
    run.skipped_records = 0
    run.insert_status = None
    run.superset_execution_id = None
    run.superset_execution_log = None
    run.error_message = None
    return run


# ---------------------------------------------------------------------------
# Tests — concurrency guard and stale PENDING clearance
# ---------------------------------------------------------------------------

# #region test_concurrent_run_skips [C:1] [TYPE Function]
def test_concurrent_run_skips():
    """Active RUNNING concurrent run → function skips and logs skipped_concurrent event."""
    from src.plugins.translate.scheduler import execute_scheduled_translation

    mock_db = MagicMock()
    mock_session_maker = MagicMock(return_value=mock_db)
    mock_config = MagicMock()

    schedule = MagicMock()
    schedule.id = "sched-1"
    schedule.job_id = "job-1"
    schedule.is_active = True
    schedule.execution_mode = "full"

    active_run = MagicMock()
    active_run.id = "active-run-1"
    active_run.job_id = "job-1"
    active_run.status = "RUNNING"

    q1 = MagicMock()
    q1.filter.return_value.first.return_value = schedule

    q2 = MagicMock()
    q2.filter.return_value.order_by.return_value.first.return_value = active_run

    mock_db.query.side_effect = [q1, q2]

    with patch(
        "src.plugins.translate.orchestrator.TranslationOrchestrator"
    ) as mock_orch_cls, patch(
        "src.plugins.translate.scheduler.TranslationEventLog"
    ) as mock_event_log_cls, \
         patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), \
         patch("src.core.database.SessionLocal", mock_session_maker), \
         patch("src.dependencies.get_config_manager", return_value=mock_config):
        mock_event_log = MagicMock()
        mock_event_log_cls.return_value = mock_event_log

        execute_scheduled_translation(
            schedule_id="sched-1",
            job_id="job-1",
            execution_mode="full",
        )

    mock_orch_cls.assert_not_called()
    mock_event_log.log_event.assert_called_once()
    event_call = mock_event_log.log_event.call_args
    assert event_call.kwargs["event_type"] == "RUN_STARTED"
    assert event_call.kwargs["payload"]["reason"] == "skipped_concurrent"
    mock_db.close.assert_called_once()
# #endregion test_concurrent_run_skips


# #region test_recent_pending_run_not_stale [C:1] [TYPE Function]
def test_recent_pending_run_not_stale():
    """Recent PENDING run (30 min old) → function skips, not stale."""
    from src.plugins.translate.scheduler import execute_scheduled_translation

    mock_db = MagicMock()
    mock_session_maker = MagicMock(return_value=mock_db)
    mock_config = MagicMock()

    schedule = MagicMock()
    schedule.id = "sched-1"
    schedule.job_id = "job-1"
    schedule.is_active = True
    schedule.execution_mode = "full"

    pending_run = MagicMock()
    pending_run.id = "pending-run-1"
    pending_run.job_id = "job-1"
    pending_run.status = "PENDING"
    pending_run.created_at = datetime.now(UTC) - timedelta(minutes=30)

    q1 = MagicMock()
    q1.filter.return_value.first.return_value = schedule

    q2 = MagicMock()
    q2.filter.return_value.order_by.return_value.first.return_value = pending_run

    mock_db.query.side_effect = [q1, q2]

    with patch(
        "src.plugins.translate.orchestrator.TranslationOrchestrator"
    ) as mock_orch_cls, patch(
        "src.plugins.translate.scheduler.TranslationEventLog"
    ) as mock_event_log_cls, \
         patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), \
         patch("src.core.database.SessionLocal", mock_session_maker), \
         patch("src.dependencies.get_config_manager", return_value=mock_config):
        mock_event_log = MagicMock()
        mock_event_log_cls.return_value = mock_event_log

        execute_scheduled_translation(
            schedule_id="sched-1",
            job_id="job-1",
            execution_mode="full",
        )

    mock_orch_cls.assert_not_called()
    mock_event_log.log_event.assert_called_once()
    event_call = mock_event_log.log_event.call_args
    assert event_call.kwargs["event_type"] == "RUN_STARTED"
    assert event_call.kwargs["payload"]["reason"] == "skipped_concurrent"
    mock_db.close.assert_called_once()
# #endregion test_recent_pending_run_not_stale


# #region test_stale_pending_cleared_and_proceeds [C:1] [TYPE Function]
def test_stale_pending_cleared_and_proceeds():
    """Stale PENDING run (>1h old) → cleared as FAILED, execution proceeds."""
    from src.plugins.translate.scheduler import execute_scheduled_translation

    mock_db = MagicMock()
    mock_session_maker = MagicMock(return_value=mock_db)
    mock_config = MagicMock()

    schedule = MagicMock()
    schedule.id = "sched-1"
    schedule.job_id = "job-1"
    schedule.is_active = True
    schedule.execution_mode = "full"

    stale_run = MagicMock()
    stale_run.id = "stale-run-1"
    stale_run.job_id = "job-1"
    stale_run.status = "PENDING"
    stale_run.created_at = datetime.now(UTC) - timedelta(hours=2)

    most_recent = _make_most_recent_run(created_at_delta_days=30)

    q1 = MagicMock()
    q1.filter.return_value.first.return_value = schedule

    q2 = MagicMock()
    q2.filter.return_value.order_by.return_value.first.return_value = stale_run

    q3 = MagicMock()
    q3.filter.return_value.all.return_value = [stale_run]

    q4 = MagicMock()
    q4.filter.return_value.order_by.return_value.first.return_value = most_recent

    mock_db.query.side_effect = [q1, q2, q3, q4]

    mock_run = _make_mock_run()

    with patch(
        "src.plugins.translate.orchestrator.TranslationOrchestrator"
    ) as mock_orch_cls, \
         patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), \
         patch("src.core.database.SessionLocal", mock_session_maker), \
         patch("src.dependencies.get_config_manager", return_value=mock_config):
        mock_orch = MagicMock()
        mock_orch.start_run.return_value = mock_run
        mock_orch_cls.return_value = mock_orch

        execute_scheduled_translation(
            schedule_id="sched-1",
            job_id="job-1",
            execution_mode="full",
        )

    assert stale_run.status == "FAILED"
    assert "Stale: concurrency deadlock" in stale_run.error_message
    assert stale_run.completed_at is not None

    mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True)
    mock_orch_cls.execute_background.assert_called()
    assert mock_db.commit.call_count >= 1
    mock_db.close.assert_called_once()
# #endregion test_stale_pending_cleared_and_proceeds


# #region test_multiple_stale_pending_all_cleared [C:1] [TYPE Function]
def test_multiple_stale_pending_all_cleared():
    """Multiple stale PENDING runs → all cleared as FAILED, execution proceeds."""
    from src.plugins.translate.scheduler import execute_scheduled_translation

    mock_db = MagicMock()
    mock_session_maker = MagicMock(return_value=mock_db)
    mock_config = MagicMock()

    schedule = MagicMock()
    schedule.id = "sched-1"
    schedule.job_id = "job-1"
    schedule.is_active = True
    schedule.execution_mode = "full"

    stale_run_a = MagicMock()
    stale_run_a.id = "stale-run-a"
    stale_run_a.job_id = "job-1"
    stale_run_a.status = "PENDING"
    stale_run_a.created_at = datetime.now(UTC) - timedelta(hours=5)

    stale_run_b = MagicMock()
    stale_run_b.id = "stale-run-b"
    stale_run_b.job_id = "job-1"
    stale_run_b.status = "PENDING"
    stale_run_b.created_at = datetime.now(UTC) - timedelta(hours=3)

    most_recent = _make_most_recent_run(created_at_delta_days=30)

    q1 = MagicMock()
    q1.filter.return_value.first.return_value = schedule

    q2 = MagicMock()
    q2.filter.return_value.order_by.return_value.first.return_value = stale_run_a

    q3 = MagicMock()
    q3.filter.return_value.all.return_value = [stale_run_a, stale_run_b]

    q4 = MagicMock()
    q4.filter.return_value.order_by.return_value.first.return_value = most_recent

    mock_db.query.side_effect = [q1, q2, q3, q4]

    mock_run = _make_mock_run()

    with patch(
        "src.plugins.translate.orchestrator.TranslationOrchestrator"
    ) as mock_orch_cls, \
         patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), \
         patch("src.core.database.SessionLocal", mock_session_maker), \
         patch("src.dependencies.get_config_manager", return_value=mock_config):
        mock_orch = MagicMock()
        mock_orch.start_run.return_value = mock_run
        mock_orch_cls.return_value = mock_orch

        execute_scheduled_translation(
            schedule_id="sched-1",
            job_id="job-1",
            execution_mode="full",
        )

    for sr in [stale_run_a, stale_run_b]:
        assert sr.status == "FAILED"
        assert "Stale: concurrency deadlock" in sr.error_message
        assert sr.completed_at is not None

    mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True)
    mock_orch_cls.execute_background.assert_called()
    mock_db.close.assert_called_once()
# #endregion test_multiple_stale_pending_all_cleared

# #endregion TestTranslateSchedulerGuard