ss-tools/backend/tests/services/test_validation_service_triggers.py

# #region Test.ValidationTaskService.Triggers [C:3] [TYPE Module] [SEMANTICS test,validation,trigger,run,detail]
# @BRIEF Tests for validation_service.py — trigger runs, list all runs, run detail.
# @RELATION BINDS_TO -> [ValidationService]

from pathlib import Path
import sys

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

from unittest.mock import AsyncMock, MagicMock, patch
from datetime import UTC, datetime
import pytest


def make_policy(**overrides):
    policy = MagicMock()
    policy.id = "pol-1"
    policy.name = "Test Task"
    policy.description = None
    policy.environment_id = "env-1"
    policy.is_active = True
    policy.dashboard_ids = []
    policy.provider_id = "prov-1"
    policy.schedule_days = None
    policy.window_start = None
    policy.window_end = None
    policy.notify_owners = None
    policy.custom_channels = None
    policy.alert_condition = None
    policy.created_at = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC)
    policy.updated_at = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC)
    policy.sources = []
    policy.screenshot_enabled = True
    policy.logs_enabled = True
    policy.execute_chart_data = False
    policy.llm_batch_size = 1
    policy.policy_dashboard_concurrency_limit = 3
    policy.prompt_template = None
    for k, v in overrides.items():
        setattr(policy, k, v)
    return policy


def make_valid_query_mock(policies=None):
    m = MagicMock()
    policies = policies or []
    first_result = policies[0] if policies else None
    m.configure_mock(**{"count.return_value": len(policies)})
    m.configure_mock(**{"order_by.return_value.offset.return_value.limit.return_value.all.return_value": policies})
    m.configure_mock(**{"filter.return_value.first.return_value": first_result})
    m.configure_mock(**{"filter.return_value.count.return_value": len(policies)})
    m.configure_mock(**{"filter.return_value.order_by.return_value.first.return_value": None})
    m.configure_mock(**{"filter.return_value.order_by.return_value.offset.return_value.limit.return_value.all.return_value": policies})
    return m


class TestTriggerRun:
    """trigger_run — create run and spawn task."""

    @pytest.mark.asyncio
    async def test_not_found(self):
        from src.services.validation_service import ValidationTaskService
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = None
        svc = ValidationTaskService(db)
        with pytest.raises(ValueError, match="not found"):
            await svc.trigger_run("nonexistent", AsyncMock())

    @pytest.mark.asyncio
    async def test_existing_running_run(self):
        from src.services.validation_service import ValidationTaskService
        policy = make_policy(dashboard_ids=["42", "99"])
        db = MagicMock()
        mock_q = MagicMock()
        mock_q.filter.return_value.first.side_effect = [policy, MagicMock()]
        db.query.return_value = mock_q
        svc = ValidationTaskService(db)
        with pytest.raises(ValueError, match="already has a running run"):
            await svc.trigger_run("pol-1", AsyncMock())

    @pytest.mark.asyncio
    async def test_worker_limit_reached(self):
        from src.services.validation_service import ValidationTaskService
        policy = make_policy(dashboard_ids=["42"], screenshot_enabled=True)
        db = MagicMock()
        call_log = {"count": 0}

        def query_fn(model):
            if model.__name__ == "ValidationPolicy":
                m = MagicMock()
                m.filter.return_value.first.return_value = policy
                return m
            elif model.__name__ == "ValidationRun":
                m = MagicMock()
                def run_filter(*a, **kw):
                    call_log["count"] += 1
                    fm = MagicMock()
                    if call_log["count"] == 1:
                        fm.first.return_value = None
                    else:
                        fm.count.return_value = 5
                    return fm
                m.filter = run_filter
                return m
            else:
                m = MagicMock()
                m.filter.return_value.all.return_value = []
                return m

        db.query = query_fn
        cm = MagicMock()
        cm.configure_mock(**{"config.settings.global_validation_worker_limit": 3})
        svc = ValidationTaskService(db, config_manager=cm)
        with pytest.raises(ValueError, match="worker limit"):
            await svc.trigger_run("pol-1", AsyncMock())

    @pytest.mark.asyncio
    async def test_no_dashboard_ids_raises(self):
        from src.services.validation_service import ValidationTaskService
        policy = make_policy(dashboard_ids=[], screenshot_enabled=True)
        db = MagicMock()

        def query_fn(model):
            m = MagicMock()
            if model.__name__ == "ValidationPolicy":
                m.filter.return_value.first.return_value = policy
            elif model.__name__ == "ValidationRun":
                m.filter.return_value.first.return_value = None
                m.filter.return_value.count.return_value = 0
            elif model.__name__ == "ValidationSource":
                m.filter.return_value.all.return_value = []
            return m

        db.query = query_fn
        cm = MagicMock()
        cm.configure_mock(**{"config.settings.global_validation_worker_limit": 5})
        svc = ValidationTaskService(db, config_manager=cm)
        with patch("src.services.validation_service.LLMProviderService") as mock_cls:
            mock_svc = MagicMock()
            provider = MagicMock()
            provider.is_active = True
            provider.is_multimodal = True
            mock_svc.get_provider.return_value = provider
            mock_cls.return_value = mock_svc
            with pytest.raises(ValueError, match="no dashboard IDs"):
                await svc.trigger_run("pol-1", AsyncMock())

    @pytest.mark.asyncio
    async def test_successful_trigger(self):
        from src.services.validation_service import ValidationTaskService
        policy = make_policy(
            dashboard_ids=["42", "99"],
            screenshot_enabled=True, logs_enabled=True,
            execute_chart_data=False, llm_batch_size=1,
            policy_dashboard_concurrency_limit=3, prompt_template=None,
        )
        db = MagicMock()

        def query_fn(model):
            m = MagicMock()
            if model.__name__ == "ValidationSource":
                m.filter.return_value.all.return_value = []
            elif model.__name__ == "ValidationRun":
                m.filter.return_value.first.return_value = None
                m.filter.return_value.count.return_value = 0
            elif model.__name__ == "ValidationPolicy":
                m.filter.return_value.first.return_value = policy
            return m

        db.query = query_fn
        cm = MagicMock()
        cm.configure_mock(**{"config.settings.global_validation_worker_limit": 5})
        svc = ValidationTaskService(db, config_manager=cm, username="tester")

        with patch("src.services.validation_service.LLMProviderService") as mock_cls:
            mock_svc = MagicMock()
            provider = MagicMock()
            provider.is_active = True
            provider.is_multimodal = True
            mock_svc.get_provider.return_value = provider
            mock_cls.return_value = mock_svc

            task_manager = AsyncMock()
            task = MagicMock()
            task.id = "spawned-task-1"
            task_manager.create_task.return_value = task
            result = await svc.trigger_run("pol-1", task_manager)

        assert result["task_id"] == "spawned-task-1"
        assert "run_id" in result


class TestListAllRuns:
    """list_all_runs — cross-task filtering."""

    def make_record(self, **overrides):
        from src.models.llm import ValidationRecord as VR
        r = MagicMock(spec=VR)
        r.id = "rec-1"
        r.run_id = "run-1"
        r.policy_id = "pol-1"
        r.dashboard_id = "42"
        r.dashboard_title = "Sales"
        r.environment_id = "env-1"
        r.status = "PASS"
        r.summary = "All good"
        r.issues = []
        r.raw_response = None
        r.screenshot_path = None
        r.screenshot_paths = None
        r.execution_path = None
        r.dataset_health = None
        r.chart_data_results = []
        r.tab_screenshots = []
        r.logs_sent_to_llm = []
        r.token_usage = None
        r.timings = None
        r.timestamp = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC)
        for k, v in overrides.items():
            setattr(r, k, v)
        return r

    def test_no_filters(self):
        from src.services.validation_service import ValidationTaskService
        record = self.make_record()
        db = MagicMock()
        m = MagicMock()
        m.configure_mock(**{"count.return_value": 1})
        m.configure_mock(**{"order_by.return_value.offset.return_value.limit.return_value.all.return_value": [record]})
        m.configure_mock(**{"filter.return_value.count.return_value": 1})
        m.configure_mock(**{"filter.return_value.order_by.return_value.offset.return_value.limit.return_value.all.return_value": [record]})
        db.query.return_value = m
        svc = ValidationTaskService(db)
        total, records = svc.list_all_runs()
        assert total == 1
        assert len(records) == 1
        assert records[0]["status"] == "PASS"

    def test_filter_by_status(self):
        from src.services.validation_service import ValidationTaskService
        db = MagicMock()
        svc = ValidationTaskService(db)
        svc.list_all_runs(status="FAIL")
        db.query.return_value.filter.assert_called()

    def test_filter_by_environment(self):
        from src.services.validation_service import ValidationTaskService
        db = MagicMock()
        svc = ValidationTaskService(db)
        svc.list_all_runs(environment_id="env-1")

    def test_filter_by_dashboard_id(self):
        from src.services.validation_service import ValidationTaskService
        db = MagicMock()
        svc = ValidationTaskService(db)
        svc.list_all_runs(dashboard_id="42")


class TestGetRunDetail:
    """get_run_detail — run + records."""

    def test_not_found(self):
        from src.services.validation_service import ValidationTaskService
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = None
        svc = ValidationTaskService(db)
        with pytest.raises(ValueError, match="not found"):
            svc.get_run_detail("nonexistent")

    def test_success(self):
        from src.services.validation_service import ValidationTaskService
        run = MagicMock()
        run.id = "run-1"
        run.policy_id = "pol-1"
        run.task_id = "task-1"
        run.status = "completed"
        run.trigger = "manual"
        run.started_at = None
        run.finished_at = None
        run.dashboard_count = 0
        run.pass_count = 0
        run.warn_count = 0
        run.fail_count = 0
        run.unknown_count = 0
        run.created_at = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC)

        from src.models.llm import ValidationRecord as VR
        record = MagicMock(spec=VR)
        record.id = "rec-1"
        record.run_id = "run-1"
        record.policy_id = "pol-1"
        record.dashboard_id = "42"
        record.dashboard_title = "Sales"
        record.environment_id = "env-1"
        record.status = "PASS"
        record.summary = "OK"
        record.issues = []
        record.raw_response = None
        record.screenshot_path = None
        record.screenshot_paths = None
        record.execution_path = None
        record.dataset_health = None
        record.chart_data_results = []
        record.tab_screenshots = []
        record.logs_sent_to_llm = []
        record.token_usage = None
        record.timings = None
        record.timestamp = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC)

        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = run
        db.query.return_value.filter.return_value.order_by.return_value.all.return_value = [record]
        svc = ValidationTaskService(db)
        result = svc.get_run_detail("run-1")
        assert result["run"]["status"] == "completed"
        assert len(result["records"]) == 1
        assert result["records"][0]["status"] == "PASS"
# #endregion Test.ValidationTaskService.Triggers