314 lines
12 KiB
Python
314 lines
12 KiB
Python
# #region Test.MaintenanceBannerPlugin [C:3] [TYPE Module] [SEMANTICS test, maintenance, banner, plugin, coverage]
|
|
# @BRIEF Unit tests for MaintenanceBannerPlugin — properties, get_schema, execute operations, and edge cases.
|
|
# @RELATION BINDS_TO -> [MaintenanceBannerPlugin]
|
|
# @TEST_EDGE: unknown_operation -> Returns failed status
|
|
# @TEST_EDGE: start_without_event_id -> Returns failed status
|
|
# @TEST_EDGE: end_without_event_id -> Returns failed status
|
|
# @TEST_EDGE: environment_not_found -> Returns failed status
|
|
# @TEST_EDGE: db_rollback_on_exception -> Rollback called
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from src.plugins.maintenance_banner import MaintenanceBannerPlugin
|
|
|
|
|
|
# ── Helpers ──
|
|
|
|
|
|
def _make_env(id_val="env-1", name="Test Env"):
|
|
env = MagicMock()
|
|
env.id = id_val
|
|
env.name = name
|
|
return env
|
|
|
|
|
|
def _make_task_context():
|
|
ctx = MagicMock()
|
|
ctx.logger = MagicMock()
|
|
return ctx
|
|
|
|
|
|
class TestMaintenanceBannerPluginProperties:
|
|
"""Verify static property values."""
|
|
|
|
def test_id(self):
|
|
plugin = MaintenanceBannerPlugin()
|
|
assert plugin.id == "maintenance_banner_apply"
|
|
|
|
def test_name(self):
|
|
plugin = MaintenanceBannerPlugin()
|
|
assert plugin.name == "Maintenance Banner"
|
|
|
|
def test_description(self):
|
|
plugin = MaintenanceBannerPlugin()
|
|
assert plugin.description == "Applies or removes maintenance banners on Superset dashboards."
|
|
|
|
def test_version(self):
|
|
plugin = MaintenanceBannerPlugin()
|
|
assert plugin.version == "1.0.0"
|
|
|
|
def test_ui_route(self):
|
|
plugin = MaintenanceBannerPlugin()
|
|
assert plugin.ui_route == ""
|
|
|
|
|
|
class TestMaintenanceBannerPluginGetSchema:
|
|
"""Verify get_schema returns correct structure."""
|
|
|
|
def test_get_schema(self):
|
|
plugin = MaintenanceBannerPlugin()
|
|
schema = plugin.get_schema()
|
|
|
|
assert schema["type"] == "object"
|
|
assert "operation" in schema["properties"]
|
|
assert schema["properties"]["operation"]["enum"] == ["start", "end", "end_all"]
|
|
assert "operation" in schema["required"]
|
|
assert "event_id" in schema["properties"]
|
|
assert "environment_id" in schema["properties"]
|
|
|
|
|
|
class TestMaintenanceBannerPluginExecute:
|
|
"""Verify MaintenanceBannerPlugin.execute with different operations."""
|
|
|
|
# ── Unknown operation ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_unknown_operation(self):
|
|
"""Unknown operation returns failed status."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.filter.return_value.first.return_value = None # no MaintenanceSettings
|
|
mock_db.query.return_value = mock_query
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm):
|
|
result = await plugin.execute({"operation": "bogus", "environment_id": "env-1"})
|
|
|
|
assert result["status"] == "failed"
|
|
assert "Unknown operation" in result["error"]
|
|
|
|
# ── start without event_id ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_start_without_event_id(self):
|
|
"""start operation without event_id returns failed."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm):
|
|
result = await plugin.execute({"operation": "start", "environment_id": "env-1"})
|
|
|
|
assert result["status"] == "failed"
|
|
assert "event_id required" in result["error"]
|
|
|
|
# ── end without event_id ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_end_without_event_id(self):
|
|
"""end operation without event_id returns failed."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm):
|
|
result = await plugin.execute({"operation": "end", "environment_id": "env-1"})
|
|
|
|
assert result["status"] == "failed"
|
|
assert "event_id required" in result["error"]
|
|
|
|
# ── Environment not found ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_environment_not_found(self):
|
|
"""Non-matching environment_id returns failed."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm):
|
|
result = await plugin.execute({
|
|
"operation": "start",
|
|
"event_id": "evt-1",
|
|
"environment_id": "nonexistent",
|
|
})
|
|
|
|
assert result["status"] == "failed"
|
|
assert "not found" in result["error"]
|
|
|
|
# ── environment_id resolved from MaintenanceSettings ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_resolve_env_from_settings(self):
|
|
"""environment_id resolved from MaintenanceSettings when not in params."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-default", "DefaultEnv")
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_settings = MagicMock()
|
|
mock_settings.target_environment_id = "env-default"
|
|
mock_query.filter.return_value.first.return_value = mock_settings
|
|
mock_db.query.return_value = mock_query
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm), \
|
|
patch('src.plugins.maintenance_banner.start_maintenance', AsyncMock(return_value={"status": "ok"})):
|
|
result = await plugin.execute({
|
|
"operation": "start",
|
|
"event_id": "evt-1",
|
|
})
|
|
|
|
assert result["status"] == "ok"
|
|
|
|
# ── environment_id resolved from first env when no settings ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_resolve_env_from_first_env(self):
|
|
"""environment_id falls back to first env when no settings match."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-first", "FirstEnv")
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.filter.return_value.first.return_value = None # no MaintenanceSettings
|
|
mock_db.query.return_value = mock_query
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm), \
|
|
patch('src.plugins.maintenance_banner.start_maintenance', AsyncMock(return_value={"status": "ok"})):
|
|
result = await plugin.execute({
|
|
"operation": "start",
|
|
"event_id": "evt-1",
|
|
})
|
|
|
|
assert result["status"] == "ok"
|
|
|
|
# ── start operation calls start_maintenance ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_start_success(self):
|
|
"""start operation delegates to start_maintenance."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
mock_start = AsyncMock(return_value={"status": "ok", "message": "maintenance started"})
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm), \
|
|
patch('src.plugins.maintenance_banner.start_maintenance', mock_start):
|
|
result = await plugin.execute({
|
|
"operation": "start",
|
|
"event_id": "evt-1",
|
|
"environment_id": "env-1",
|
|
})
|
|
|
|
assert result["status"] == "ok"
|
|
mock_start.assert_awaited_once()
|
|
mock_db.commit.assert_called_once()
|
|
|
|
# ── end operation calls end_maintenance ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_end_success(self):
|
|
"""end operation delegates to end_maintenance."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
mock_end = AsyncMock(return_value={"status": "ok", "message": "maintenance ended"})
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm), \
|
|
patch('src.plugins.maintenance_banner.end_maintenance', mock_end):
|
|
result = await plugin.execute({
|
|
"operation": "end",
|
|
"event_id": "evt-1",
|
|
"environment_id": "env-1",
|
|
})
|
|
|
|
assert result["status"] == "ok"
|
|
mock_end.assert_awaited_once()
|
|
|
|
# ── end_all operation calls end_all_maintenance ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_end_all_success(self):
|
|
"""end_all operation delegates to end_all_maintenance."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
mock_end_all = AsyncMock(return_value={"status": "ok", "message": "all maintenance ended"})
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm), \
|
|
patch('src.plugins.maintenance_banner.end_all_maintenance', mock_end_all):
|
|
result = await plugin.execute({
|
|
"operation": "end_all",
|
|
"environment_id": "env-1",
|
|
})
|
|
|
|
assert result["status"] == "ok"
|
|
mock_end_all.assert_awaited_once()
|
|
mock_db.commit.assert_called_once()
|
|
|
|
# ── Exception causes rollback ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_exception_causes_rollback(self):
|
|
"""Exception during operation triggers rollback and returns failed."""
|
|
plugin = MaintenanceBannerPlugin()
|
|
env = _make_env("env-1", "Test")
|
|
mock_db = MagicMock()
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_environments.return_value = [env]
|
|
|
|
with patch('src.plugins.maintenance_banner.SessionLocal', return_value=mock_db), \
|
|
patch('src.plugins.maintenance_banner.get_config_manager', return_value=mock_cm), \
|
|
patch('src.plugins.maintenance_banner.start_maintenance',
|
|
AsyncMock(side_effect=RuntimeError("DB error"))):
|
|
result = await plugin.execute({
|
|
"operation": "start",
|
|
"event_id": "evt-1",
|
|
"environment_id": "env-1",
|
|
})
|
|
|
|
assert result["status"] == "failed"
|
|
assert "DB error" in result["error"]
|
|
mock_db.rollback.assert_called_once()
|
|
mock_db.close.assert_called_once()
|