ss-tools/backend/tests/plugins/test_llm_analysis_scheduler.py

# #region Test.LLMAnalysisScheduler [C:3] [TYPE Module] [SEMANTICS test, llm, analysis, scheduler, cron]
# @BRIEF Tests for llm_analysis/scheduler.py — _parse_cron, schedule_dashboard_validation.
# @RELATION BINDS_TO -> [LLMAnalysisScheduler]
# @TEST_CONTRACT: _parse_cron -> dict | cron expression parsing
# @TEST_CONTRACT: schedule_dashboard_validation -> None | schedules a validation job

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

import os
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:")

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from src.plugins.llm_analysis.scheduler import _parse_cron, schedule_dashboard_validation


class TestParseCron:
    """_parse_cron — basic cron expression parser."""

    def test_standard_5_part_cron(self):
        """Standard 5-part cron expression parsed correctly."""
        result = _parse_cron("0 8 * * 1-5")
        assert result == {
            "minute": "0",
            "hour": "8",
            "day": "*",
            "month": "*",
            "day_of_week": "1-5",
        }

    def test_daily_at_midnight(self):
        """Daily at midnight."""
        result = _parse_cron("0 0 * * *")
        assert result["hour"] == "0"
        assert result["minute"] == "0"

    def test_every_hour(self):
        """Every hour at minute 0."""
        result = _parse_cron("0 * * * *")
        assert result["minute"] == "0"
        assert result["hour"] == "*"

    def test_invalid_parts_count(self):
        """Less or more than 5 parts returns empty dict."""
        assert _parse_cron("0 8 * *") == {}
        assert _parse_cron("0 8 * * 1-5 extra") == {}

    def test_empty_string(self):
        """Empty string returns empty dict."""
        assert _parse_cron("") == {}

    def test_step_values(self):
        """Step values like */15 parsed correctly."""
        result = _parse_cron("*/15 * * * *")
        assert result["minute"] == "*/15"


class TestScheduleDashboardValidation:
    """schedule_dashboard_validation — schedules recurring validation."""

    def test_schedule_with_params(self):
        """Scheduler and task manager called with correct args."""
        mock_scheduler = MagicMock()
        mock_task_manager = MagicMock()

        with patch("src.plugins.llm_analysis.scheduler.get_scheduler_service", return_value=mock_scheduler), \
             patch("src.plugins.llm_analysis.scheduler.get_task_manager", return_value=mock_task_manager):

            schedule_dashboard_validation(
                dashboard_id="dash-1",
                cron_expression="0 6 * * *",
                params={"threshold": 0.8},
            )

        # Scheduler should have added a cron job
        mock_scheduler.add_job.assert_called_once()
        call_args = mock_scheduler.add_job.call_args
        assert call_args[0][1] == "cron"  # trigger type
        assert call_args[1]["id"] == "llm_val_dash-1"
        assert call_args[1]["replace_existing"] is True
        assert call_args[1]["minute"] == "0"
        assert call_args[1]["hour"] == "6"

    def test_schedule_creates_task_on_execution(self):
        """The scheduled job function creates a task when executed."""
        mock_scheduler = MagicMock()
        mock_task_manager = AsyncMock()

        with patch("src.plugins.llm_analysis.scheduler.get_scheduler_service", return_value=mock_scheduler), \
             patch("src.plugins.llm_analysis.scheduler.get_task_manager", return_value=mock_task_manager):

            schedule_dashboard_validation(
                dashboard_id="dash-1",
                cron_expression="0 6 * * *",
                params={"threshold": 0.8},
            )

        # Get the job function and call it
        job_func = mock_scheduler.add_job.call_args[0][0]
        import asyncio
        asyncio.run(job_func())

        mock_task_manager.create_task.assert_called_once_with(
            plugin_id="llm_dashboard_validation",
            params={"dashboard_id": "dash-1", "threshold": 0.8},
        )

    def test_schedule_without_params(self):
        """Schedule without extra params still works."""
        mock_scheduler = MagicMock()
        mock_task_manager = AsyncMock()

        with patch("src.plugins.llm_analysis.scheduler.get_scheduler_service", return_value=mock_scheduler), \
             patch("src.plugins.llm_analysis.scheduler.get_task_manager", return_value=mock_task_manager):

            schedule_dashboard_validation(
                dashboard_id="dash-2",
                cron_expression="0 0 * * *",
                params={},
            )

        job_func = mock_scheduler.add_job.call_args[0][0]
        import asyncio
        asyncio.run(job_func())

        mock_task_manager.create_task.assert_called_once_with(
            plugin_id="llm_dashboard_validation",
            params={"dashboard_id": "dash-2"},
        )
# #endregion Test.LLMAnalysisScheduler