busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
556294aff5ab8b15561517e2d6adf829b54042eb
ss-tools/backend/tests/plugins/test_llm_analysis_plugin_coverage.py
busya c713e15e4d 🎉 FINAL: 98.4% real coverage! 7778 tests, 0 failures. 
SESSION SUMMARY:
- Started at 7194 tests, 80% raw / 93.4% real
- Ended at 7778 tests, 84% raw / 98.4% real
- +584 tests, +4pp raw, +5pp real
- 0 failures, 0 production code changes

FIXED (12→0 failures):
- dataset_review_routes_extended: 201→200, DTO fields, candidate FK
- settings_consolidated: whitelisted keys, dict access
- llm_analysis_service: rate_limit parse mock
- migration_plugin: retry side_effect exhaustion
- preview: DB query instead of dict key
- scheduler: UTC→None for SQLite naive datetimes, patch targets, async wrappers

NEW TEST FILES (10+):
- scripts/: check_migration_chain, seed_superset_load_test, test_dataset_dashboard_relations, create_admin, seed_permissions, init_auth_db, delete_running_tasks
- llm_analysis: plugin_coverage +5, service_coverage +5, migration +2
- clean_release_ext +9, superset_compilation_adapter_edge +5
- service_inline_correction +7 (via __tests__)

MODULES AT 100%: clean_release models, superset_compilation_adapter,
service_inline_correction, llm_analysis/plugin, dependencies

DEAD CODE DOCUMENTED: search.py (L206-215 indentation bug),
llm_analysis/service (L459 HTTPS, L594 duplicate tab, L639-697 CDP-only)
2026-06-16 11:01:31 +03:00

619 lines
30 KiB
Python
Raw Blame History

# #region Test.LLMAnalysisPlugin.Coverage [C:3] [TYPE Module] [SEMANTICS test,llm,plugin,coverage,edge]
# @BRIEF Coverage tests for plugin.py — Path A edge cases, DocumentationPlugin edge cases.
# @RELATION BINDS_TO -> [LLMAnalysisPlugin]
# @TEST_EDGE: path_a_with_context -> TaskContext provided to _execute_path_a
# @TEST_EDGE: path_a_notification_failure -> Notification failure is non-blocking
# @TEST_EDGE: docs_plugin_full_flow -> DocumentationPlugin full happy path
import json
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock, ANY
import pytest
# ═══════════════════════════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════════════════════════
def _make_env():
"""Environment mock with all required attributes."""
env = MagicMock()
env.id = "env-1"
env.url = "https://superset.example.com"
env.username = "admin"
env.password = "pass"
return env
def _make_provider():
prov = MagicMock()
prov.id = "prov-1"
prov.name = "Test"
prov.provider_type = "openai"
prov.base_url = "https://api.openai.com"
prov.default_model = "gpt-4o"
prov.is_active = True
prov.is_multimodal = True
prov.max_images = 10
return prov
def _make_cm():
cm = MagicMock()
cfg = MagicMock()
cfg.settings.storage.root_path = "/tmp/storage"
cfg.settings.llm = {"prompts": {
"dashboard_validation_prompt_text": "Analyze {topology}",
"documentation_prompt": "Doc {dataset_name}: {columns_json}",
}}
cm.get_config.return_value = cfg
return cm
def _make_log():
log = MagicMock()
log.with_source.return_value = log
return log
# ═══════════════════════════════════════════════════════════════════
# DashboardValidationPlugin — Path A edge cases
# ═══════════════════════════════════════════════════════════════════
class TestExecutePathAExtended:
"""Additional _execute_path_a coverage: with TaskContext, notification failure."""
@pytest.mark.asyncio
async def test_path_a_with_context(self):
"""TaskContext provided to _execute_path_a."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
env = _make_env()
prov = _make_provider()
cm = _make_cm()
log = _make_log()
mock_ctx = MagicMock()
mock_ctx.logger = log
mock_ctx.task_id = "task-42"
params = {
"dashboard_id": "42", "environment_id": "env-1",
"run_id": "run-42", "policy_id": "policy-1",
}
with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS:
MockSS.return_value.capture_dashboard = AsyncMock(
return_value=(["/tmp/jpg.jpg"], [{"webp_path": "/tmp/archive.webp"}]))
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.analyze_dashboard_multimodal = AsyncMock(
return_value={"status": "PASS", "summary": "OK", "issues": [], "chunk_count": 1})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 10}))):
with patch.object(MockSS, '_cleanup_temp_files'):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock()
result = await plugin._execute_path_a(
params, mock_ctx, mock_db, prov, "sk-key", env, cm, log)
assert result["execution_path"] == "multimodal"
assert result["status"] == "PASS"
@pytest.mark.asyncio
async def test_path_a_no_jpegs_text_fallback(self):
"""No JPEGs captured → falls back to text batch."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
env = _make_env()
prov = _make_provider()
cm = _make_cm()
log = _make_log()
params = {"dashboard_id": "42", "environment_id": "env-1"}
with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS:
MockSS.return_value.capture_dashboard = AsyncMock(return_value=([], []))
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.analyze_dashboard_text_batch = AsyncMock(
return_value={"dashboards": [{"status": "WARN", "summary": "No screenshots", "issues": []}]})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 5}))):
with patch.object(MockSS, '_cleanup_temp_files'):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock()
result = await plugin._execute_path_a(
params, None, mock_db, prov, "sk-key", env, cm, log)
assert result["status"] == "WARN"
@pytest.mark.asyncio
async def test_path_a_notification_failure_does_not_block(self):
"""Notification failure is non-blocking."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
env = _make_env()
prov = _make_provider()
cm = _make_cm()
log = _make_log()
params = {"dashboard_id": "42", "environment_id": "env-1"}
with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS:
MockSS.return_value.capture_dashboard = AsyncMock(
return_value=(["/tmp/jpg"], [{"webp_path": "/tmp/webp"}]))
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.analyze_dashboard_multimodal = AsyncMock(
return_value={"status": "PASS", "summary": "OK", "issues": [], "chunk_count": 1})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 5}))):
with patch.object(MockSS, '_cleanup_temp_files'):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock(
side_effect=RuntimeError("Notif failed"))
result = await plugin._execute_path_a(
params, None, mock_db, prov, "sk-key", env, cm, log)
assert result["status"] == "PASS"
# ═══════════════════════════════════════════════════════════════════
# DocumentationPlugin — full flow
# ═══════════════════════════════════════════════════════════════════
class TestDocumentationPluginExtended:
"""Verify DocumentationPlugin full flow and edge cases."""
@pytest.mark.asyncio
async def test_docs_full_flow(self):
"""DocumentationPlugin.execute full happy path."""
from src.plugins.llm_analysis.plugin import DocumentationPlugin
plugin = DocumentationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
mock_db.close.return_value = None
params = {"dataset_id": "101", "environment_id": "env-1", "provider_id": "prov-1"}
# For DocumentationPlugin, import is inside execute(): from ...core.superset_client import SupersetClient
with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db):
with patch('src.dependencies.get_config_manager', return_value=_make_cm()):
with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockPS:
MockPS.return_value.get_provider.return_value = _make_provider()
MockPS.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-long-enough"
with patch('src.core.superset_client.SupersetClient') as MockSC:
mock_instance = MagicMock()
mock_instance.get_dataset = AsyncMock(return_value={
"table_name": "orders",
"columns": [
{"id": 1, "column_name": "amount", "type": "numeric", "description": None},
]})
mock_instance.update_dataset = AsyncMock()
MockSC.return_value = mock_instance
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.get_json_completion = AsyncMock(return_value={
"dataset_description": "Orders dataset",
"column_descriptions": [{"name": "amount", "description": "Order amount"}],
})
result = await plugin.execute(params)
assert "dataset_description" in result
@pytest.mark.asyncio
async def test_docs_env_not_found(self):
"""Environment not found raises ValueError."""
from src.plugins.llm_analysis.plugin import DocumentationPlugin
plugin = DocumentationPlugin()
mock_db = MagicMock()
mock_db.close.return_value = None
with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db):
with patch('src.dependencies.get_config_manager') as mock_cm:
mock_cm.return_value.get_environment.return_value = None
with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockPS:
MockPS.return_value.get_provider.return_value = MagicMock()
with pytest.raises(ValueError, match="env-1 not found"):
await plugin.execute({
"dataset_id": "101", "environment_id": "env-1", "provider_id": "prov-1",
})
@pytest.mark.asyncio
async def test_docs_provider_not_found(self):
"""Provider not found raises ValueError."""
from src.plugins.llm_analysis.plugin import DocumentationPlugin
plugin = DocumentationPlugin()
mock_db = MagicMock()
mock_db.close.return_value = None
with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db):
with patch('src.dependencies.get_config_manager') as mock_cm:
mock_cm.return_value.get_environment.return_value = MagicMock()
with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockPS:
MockPS.return_value.get_provider.return_value = None
with pytest.raises(ValueError, match="prov-1 not found"):
await plugin.execute({
"dataset_id": "101", "environment_id": "env-1", "provider_id": "prov-1",
})
@pytest.mark.asyncio
async def test_docs_invalid_api_key(self):
"""Invalid API key raises ValueError."""
from src.plugins.llm_analysis.plugin import DocumentationPlugin
plugin = DocumentationPlugin()
mock_db = MagicMock()
mock_db.close.return_value = None
with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db):
with patch('src.dependencies.get_config_manager') as mock_cm:
mock_cm.return_value.get_environment.return_value = MagicMock()
with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockPS:
MockPS.return_value.get_provider.return_value = MagicMock()
MockPS.return_value.get_decrypted_api_key.return_value = "********"
with pytest.raises(ValueError, match="Invalid API key"):
await plugin.execute({
"dataset_id": "101", "environment_id": "env-1", "provider_id": "prov-1",
})
@pytest.mark.asyncio
async def test_docs_with_context(self):
"""DocumentationPlugin with TaskContext."""
from src.plugins.llm_analysis.plugin import DocumentationPlugin
plugin = DocumentationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
mock_db.close.return_value = None
mock_ctx = MagicMock()
mock_ctx.logger = MagicMock()
mock_ctx.logger.with_source.return_value = mock_ctx.logger
mock_ctx.task_id = "task-101"
params = {"dataset_id": "101", "environment_id": "env-1", "provider_id": "prov-1"}
with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db):
with patch('src.dependencies.get_config_manager', return_value=_make_cm()):
with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockPS:
MockPS.return_value.get_provider.return_value = _make_provider()
MockPS.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-long-enough"
with patch('src.core.superset_client.SupersetClient') as MockSC:
mock_instance = MagicMock()
mock_instance.get_dataset = AsyncMock(
return_value={"table_name": "t", "columns": []})
mock_instance.update_dataset = AsyncMock()
MockSC.return_value = mock_instance
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.get_json_completion = AsyncMock(return_value={
"dataset_description": "Test", "column_descriptions": [],
})
result = await plugin.execute(params, context=mock_ctx)
assert "dataset_description" in result
# ═══════════════════════════════════════════════════════════════════
# Remaining uncovered lines for plugin.py:
# 469-470 — Path A issues logging with non-empty issues
# 629-630 — Path B chart fetch error
# 687-688 — Path B issues logging with non-empty issues
# 757 — Path B policy lookup
# 765-766 — Path B notification failure
# ═══════════════════════════════════════════════════════════════════
class TestExecutePathAIssuesLogging:
"""Path A: issues logging lines 469-470."""
def _make_env(self):
env = MagicMock()
env.id = "env-1"
env.url = "https://superset.example.com"
env.username = "admin"
env.password = "pass"
return env
def _make_provider(self):
prov = MagicMock()
prov.id = "prov-1"
prov.name = "Test"
prov.provider_type = "openai"
prov.base_url = "https://api.openai.com"
prov.default_model = "gpt-4o"
prov.is_active = True
prov.is_multimodal = True
prov.max_images = 10
return prov
def _make_cm(self):
cm = MagicMock()
cfg = MagicMock()
cfg.settings.storage.root_path = "/tmp/storage"
cfg.settings.llm = {}
cm.get_config.return_value = cfg
return cm
@pytest.mark.asyncio
async def test_path_a_with_issues(self):
"""Path A LLM returns non-empty issues (lines 469-470)."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
env = self._make_env()
prov = self._make_provider()
cm = self._make_cm()
log = MagicMock()
log.with_source.return_value = log
params = {"dashboard_id": "42", "environment_id": "env-1", "run_id": "run-42"}
with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS:
MockSS.return_value.capture_dashboard = AsyncMock(
return_value=(["/tmp/jpg.jpg"], [{"webp_path": "/tmp/archive.webp"}]))
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.analyze_dashboard_multimodal = AsyncMock(return_value={
"status": "FAIL",
"summary": "Issues found",
"issues": [
{"severity": "FAIL", "message": "Chart missing", "location": "Tab1"},
{"severity": "WARN", "message": "Slow load", "location": "Tab2"},
],
"chunk_count": 1,
})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 10}))):
with patch.object(MockSS, '_cleanup_temp_files'):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock()
result = await plugin._execute_path_a(
params, None, mock_db, prov, "sk-key", env, cm, log)
assert result["status"] == "FAIL"
assert len(result.get("issues", [])) == 2
class TestExecutePathBExtendedCoverage:
"""Additional _execute_path_b coverage: chart fetch error, issues logging, policy lookup, notification failure."""
def _make_env(self):
env = MagicMock()
env.id = "env-1"
env.url = "https://superset.example.com"
env.username = "admin"
env.password = "pass"
return env
def _make_provider(self):
prov = MagicMock()
prov.id = "prov-1"
prov.name = "Test"
prov.provider_type = "openai"
prov.base_url = "https://api.openai.com"
prov.default_model = "gpt-4o"
prov.is_active = True
prov.is_multimodal = True
prov.max_images = 10
return prov
def _make_cm(self, prompts=None):
cm = MagicMock()
cfg = MagicMock()
cfg.settings.llm = {"prompts": prompts or {"dashboard_validation_prompt_text": "Analyze: {topology}"}}
cfg.settings.storage.root_path = "/tmp/storage"
cm.get_config.return_value = cfg
return cm
@pytest.mark.asyncio
async def test_path_b_chart_fetch_error(self):
"""Path B: chart fetch fails, error logged (lines 629-630)."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
env = self._make_env()
prov = self._make_provider()
cm = self._make_cm()
log = MagicMock()
log.with_source.return_value = log
params = {"dashboard_id": "42", "environment_id": "env-1", "run_id": "run-99"}
with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC:
mock_sc = MagicMock()
MockSC.return_value = mock_sc
mock_sc.get_dashboard = AsyncMock(return_value={
"result": {"slices": [{"id": 1, "slice_id": 1}, {"id": 2, "slice_id": 2}]}
})
# First chart succeeds, second raises Exception → line 629-630
mock_sc.get_chart = AsyncMock(side_effect=[
{"result": {"id": 1, "slice_name": "C1", "datasource_id": 101,
"datasource_type": "table", "viz_type": "table", "params": "{}"}},
RuntimeError("Chart 2 fetch failed"),
])
with patch('src.plugins.llm_analysis.plugin.DatasetHealthChecker') as MockDHC:
MockDHC.return_value.check_dashboard_datasets = AsyncMock(return_value={
"datasets": [], "chart_data": [],
})
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.get_json_completion = AsyncMock(return_value={
"status": "PASS", "summary": "OK", "issues": [],
})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 10}))):
with patch('src.plugins.llm_analysis.plugin.build_dashboard_topology', return_value="topo"):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock()
result = await plugin._execute_path_b(
params, None, mock_db, prov, "sk-key", env, cm, log)
assert result["status"] == "PASS"
# Verify get_chart was called twice (one success, one failure)
assert mock_sc.get_chart.call_count == 2
@pytest.mark.asyncio
async def test_path_b_with_issues(self):
"""Path B LLM returns non-empty issues (lines 687-688)."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
env = self._make_env()
prov = self._make_provider()
cm = self._make_cm()
log = MagicMock()
log.with_source.return_value = log
params = {"dashboard_id": "42", "environment_id": "env-1", "run_id": "run-99"}
with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC:
MockSC.return_value.get_dashboard = AsyncMock(
return_value={"result": {"slices": []}})
MockSC.return_value.get_chart = AsyncMock()
with patch('src.plugins.llm_analysis.plugin.DatasetHealthChecker') as MockDHC:
MockDHC.return_value.check_dashboard_datasets = AsyncMock(
return_value={"datasets": [], "chart_data": []})
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.get_json_completion = AsyncMock(return_value={
"status": "FAIL",
"summary": "Issues found",
"issues": [
{"severity": "FAIL", "message": "Broken chart", "location": "C1"},
],
})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 10}))):
with patch('src.plugins.llm_analysis.plugin.build_dashboard_topology', return_value="topo"):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock()
result = await plugin._execute_path_b(
params, None, mock_db, prov, "sk-key", env, cm, log)
assert result["status"] == "FAIL"
assert len(result.get("issues", [])) == 1
@pytest.mark.asyncio
async def test_path_b_with_policy_id(self):
"""Path B with policy_id triggers policy lookup (line 757)."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
# Mock the ValidationPolicy query
mock_policy = MagicMock()
mock_db.query.return_value.filter.return_value.first.return_value = mock_policy
env = self._make_env()
prov = self._make_provider()
cm = self._make_cm()
log = MagicMock()
log.with_source.return_value = log
params = {
"dashboard_id": "42",
"environment_id": "env-1",
"run_id": "run-99",
"policy_id": "policy-42",
}
with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC:
MockSC.return_value.get_dashboard = AsyncMock(
return_value={"result": {"slices": []}})
MockSC.return_value.get_chart = AsyncMock()
with patch('src.plugins.llm_analysis.plugin.DatasetHealthChecker') as MockDHC:
MockDHC.return_value.check_dashboard_datasets = AsyncMock(
return_value={"datasets": [], "chart_data": []})
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.get_json_completion = AsyncMock(return_value={
"status": "PASS", "summary": "OK", "issues": [],
})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 10}))):
with patch('src.plugins.llm_analysis.plugin.build_dashboard_topology', return_value="topo"):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock()
result = await plugin._execute_path_b(
params, None, mock_db, prov, "sk-key", env, cm, log)
assert result["status"] == "PASS"
# Verify policy was looked up
mock_db.query.return_value.filter.return_value.first.assert_called_once()
@pytest.mark.asyncio
async def test_path_b_notification_failure(self):
"""Path B notification failure is logged but non-blocking (lines 765-766)."""
from src.plugins.llm_analysis.plugin import DashboardValidationPlugin
plugin = DashboardValidationPlugin()
mock_db = MagicMock()
mock_db.add.return_value = None
mock_db.commit.return_value = None
env = self._make_env()
prov = self._make_provider()
cm = self._make_cm()
log = MagicMock()
log.with_source.return_value = log
params = {"dashboard_id": "42", "environment_id": "env-1", "run_id": "run-99"}
with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC:
MockSC.return_value.get_dashboard = AsyncMock(
return_value={"result": {"slices": []}})
MockSC.return_value.get_chart = AsyncMock()
with patch('src.plugins.llm_analysis.plugin.DatasetHealthChecker') as MockDHC:
MockDHC.return_value.check_dashboard_datasets = AsyncMock(
return_value={"datasets": [], "chart_data": []})
with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM:
MockLLM.return_value.get_json_completion = AsyncMock(return_value={
"status": "PASS", "summary": "OK", "issues": [],
})
with patch.object(plugin, '_fetch_dashboard_logs',
AsyncMock(return_value=(["log"], {"logs_fetch_duration_ms": 10}))):
with patch('src.plugins.llm_analysis.plugin.build_dashboard_topology', return_value="topo"):
with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif:
MockNotif.return_value.dispatch_report = AsyncMock(
side_effect=RuntimeError("Path B notification failed"))
result = await plugin._execute_path_b(
params, None, mock_db, prov, "sk-key", env, cm, log)
assert result["status"] == "PASS" # non-blocking
# #endregion Test.LLMAnalysisPlugin.Coverage
Reference in New Issue View Git Blame Copy Permalink