busya
31680a1bc9
- Auto-expiry: expired events auto-end on GET /events via task manager - Height: _estimate_markdown_height adapted to unit=8px scale with padding detection - CRITICAL-1: removed dead import remove_chart_from_position (QA finding) - CRITICAL-2: added chart alive check in ensure_banner_chart (QA finding) - CRITICAL-3: fixed test assertions update_markdown_chart → update_banner_on_dashboard - @POST contract: fixed return range [2,12] → [19,200] - Tests: 8 new tests (auto-expiry + layout height)
519 lines
18 KiB
Python
519 lines
18 KiB
Python
# #region TestCoreScheduler [C:3] [TYPE Module] [SEMANTICS test, scheduler, apscheduler, load, reload]
|
|
# @BRIEF Verify SchedulerService load_schedules and translation job add/remove contracts.
|
|
# @RELATION BINDS_TO -> [SchedulerService]
|
|
# @TEST_EDGE: load_schedules_reloads — jobs survive remove_all_jobs when re-registered from DB
|
|
# @TEST_EDGE: add_remove_translation_job — job correctly registered and removed from APScheduler
|
|
# @TEST_INVARIANT: translation_jobs_persist_across_reload -> VERIFIED_BY: [test_load_schedules_reloads_translation_schedules]
|
|
|
|
import asyncio
|
|
from datetime import time
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers — fresh scheduler & mock config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_mock_config_manager(backup_env_id="env1", backup_cron="0 3 * * *"):
|
|
"""Build a mock ConfigManager returning a single backup-enabled environment."""
|
|
mgr = MagicMock()
|
|
mock_env = MagicMock()
|
|
mock_env.id = backup_env_id
|
|
mock_env.backup_schedule = MagicMock()
|
|
mock_env.backup_schedule.enabled = True
|
|
mock_env.backup_schedule.cron_expression = backup_cron
|
|
mock_config = MagicMock()
|
|
mock_config.environments = [mock_env]
|
|
mgr.get_config.return_value = mock_config
|
|
return mgr
|
|
|
|
|
|
def _make_mock_task_manager():
|
|
"""Build a mock TaskManager returning no active tasks."""
|
|
tm = MagicMock()
|
|
tm.get_tasks.return_value = []
|
|
return tm
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixture helpers — start/stop a real BackgroundScheduler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _start_scheduler():
|
|
"""Create and start a BackgroundScheduler. Returns the scheduler instance."""
|
|
sched = BackgroundScheduler()
|
|
sched.start()
|
|
return sched
|
|
|
|
|
|
def _stop_scheduler(sched):
|
|
"""Shutdown a BackgroundScheduler if running."""
|
|
if sched.running:
|
|
sched.shutdown(wait=False)
|
|
|
|
|
|
# #region test_load_schedules_reloads_translation_schedules [C:1] [TYPE Function]
|
|
def test_load_schedules_reloads_translation_schedules():
|
|
"""Verify load_schedules re-registers translation jobs after remove_all_jobs."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
# Pre-register a translation job (simulates what happens before reload)
|
|
svc.add_translation_job(
|
|
schedule_id="sched-1",
|
|
job_id="job-1",
|
|
cron_expression="0 2 * * *",
|
|
)
|
|
|
|
# Verify it exists BEFORE reload
|
|
job_before = sched.get_job("translate_sched-1")
|
|
assert job_before is not None, "Translation job should exist before reload"
|
|
|
|
# Mock the DB query that load_schedules uses internally
|
|
mock_schedule = MagicMock()
|
|
mock_schedule.id = "sched-1"
|
|
mock_schedule.job_id = "job-1"
|
|
mock_schedule.cron_expression = "0 2 * * *"
|
|
mock_schedule.timezone = "UTC"
|
|
mock_schedule.execution_mode = "full"
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.all.return_value = [mock_schedule]
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db):
|
|
svc.load_schedules()
|
|
|
|
# After reload, the translation job should STILL exist
|
|
job_after = sched.get_job("translate_sched-1")
|
|
assert job_after is not None, (
|
|
"Translation job should survive load_schedules (re-registered from DB)"
|
|
)
|
|
|
|
# Backup job should also exist (added from config)
|
|
backup_job = sched.get_job("backup_env1")
|
|
assert backup_job is not None, "Backup job should be registered by load_schedules"
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_load_schedules_reloads_translation_schedules
|
|
|
|
|
|
# #region test_add_and_remove_translation_job [C:1] [TYPE Function]
|
|
def test_add_and_remove_translation_job():
|
|
"""Verify add_translation_job registers and remove_translation_job removes."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
# Should NOT exist before add
|
|
assert sched.get_job("translate_sched-x") is None
|
|
|
|
svc.add_translation_job(
|
|
schedule_id="sched-x",
|
|
job_id="job-x",
|
|
cron_expression="*/5 * * * *",
|
|
)
|
|
|
|
# Should exist after add
|
|
job = sched.get_job("translate_sched-x")
|
|
assert job is not None, "Translation job should be registered after add"
|
|
assert job.id == "translate_sched-x"
|
|
|
|
svc.remove_translation_job("sched-x")
|
|
|
|
# Should NOT exist after remove
|
|
assert sched.get_job("translate_sched-x") is None, (
|
|
"Translation job should be removed after remove_translation_job"
|
|
)
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_add_and_remove_translation_job
|
|
|
|
|
|
# #region test_remove_nonexistent_translation_job_no_error [C:1] [TYPE Function]
|
|
def test_remove_nonexistent_translation_job_no_error():
|
|
"""Verify remove_translation_job handles non-existent job silently."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
# Should not raise — silent ignore
|
|
svc.remove_translation_job("nonexistent-sched")
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_remove_nonexistent_translation_job_no_error
|
|
|
|
|
|
# #region test_add_validation_job [C:1] [TYPE Function]
|
|
def test_add_validation_job():
|
|
"""Verify add_validation_job registers an APScheduler job."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
assert sched.get_job("validation_pol-1") is None
|
|
|
|
svc.add_validation_job(
|
|
policy_id="pol-1",
|
|
cron_expression="30 10 * * 0,2,4", # 10:30 Mon, Wed, Fri
|
|
app_timezone="Europe/Moscow",
|
|
)
|
|
|
|
job = sched.get_job("validation_pol-1")
|
|
assert job is not None, "Validation job should be registered after add"
|
|
assert job.id == "validation_pol-1"
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_add_validation_job
|
|
|
|
|
|
# #region test_load_schedules_loads_validation_policies [C:1] [TYPE Function]
|
|
def test_load_schedules_loads_validation_policies():
|
|
"""Verify load_schedules re-registers validation jobs from DB."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
# Add app_timezone to config so cron gets a timezone
|
|
cfg.get_config.return_value.settings.app_timezone = "Europe/Moscow"
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
# Pre-register a validation job (to test it survives reload)
|
|
svc.add_validation_job(
|
|
policy_id="pol-1",
|
|
cron_expression="0 10 * * 1,3,5",
|
|
)
|
|
assert sched.get_job("validation_pol-1") is not None
|
|
|
|
# Mock DB to return an active ValidationPolicy with schedule_days
|
|
mock_policy = MagicMock()
|
|
mock_policy.id = "pol-1"
|
|
mock_policy.schedule_days = [0, 2, 4] # Mon, Wed, Fri
|
|
mock_policy.window_start = time(10, 30)
|
|
mock_policy.window_end = time(11, 0)
|
|
|
|
mock_db = MagicMock()
|
|
# First query is for TranslationSchedule (empty), second is ValidationPolicy
|
|
mock_db.query.side_effect = [
|
|
MagicMock(filter=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))),
|
|
MagicMock(filter=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_policy])))),
|
|
]
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db):
|
|
svc.load_schedules()
|
|
|
|
# Validation job should still exist after reload
|
|
job = sched.get_job("validation_pol-1")
|
|
assert job is not None, "Validation job should survive load_schedules"
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_load_schedules_loads_validation_policies
|
|
|
|
|
|
# #region test_trigger_validation_creates_tasks [C:1] [TYPE Function]
|
|
def test_trigger_validation_creates_tasks():
|
|
"""Verify _trigger_validation creates TaskManager tasks for each dashboard."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
svc.loop = asyncio.new_event_loop()
|
|
|
|
# Mock the ValidationPolicy DB query
|
|
mock_policy = MagicMock()
|
|
mock_policy.id = "pol-1"
|
|
mock_policy.is_active = True
|
|
mock_policy.dashboard_ids = ["dash-1", "dash-2"]
|
|
mock_policy.environment_id = "ss-dev"
|
|
mock_policy.provider_id = "prov-1"
|
|
mock_policy.window_start = time(10, 0)
|
|
mock_policy.window_end = time(10, 30)
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_policy
|
|
|
|
# Patch asyncio.run_coroutine_threadsafe to call the function synchronously
|
|
# so we can verify the mock was called
|
|
captured_calls = []
|
|
|
|
def fake_run_coro(_coro_or_fn, _loop):
|
|
captured_calls.append(1)
|
|
return MagicMock()
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db), \
|
|
patch("src.core.scheduler.asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
|
|
svc._trigger_validation("pol-1")
|
|
|
|
# Should have submitted 2 coroutines via run_coroutine_threadsafe (one per dashboard)
|
|
assert len(captured_calls) == 2, (
|
|
f"Expected 2 task submissions via run_coroutine_threadsafe, got {len(captured_calls)}"
|
|
)
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_trigger_validation_creates_tasks
|
|
|
|
|
|
# #region test_trigger_validation_skips_missing_policy [C:1] [TYPE Function]
|
|
def test_trigger_validation_skips_missing_policy():
|
|
"""Verify _trigger_validation does not fail when policy is missing."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
svc.loop = asyncio.new_event_loop()
|
|
|
|
# Mock DB to return None (policy not found)
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db):
|
|
# Should not raise
|
|
svc._trigger_validation("nonexistent-policy")
|
|
|
|
# create_task should not have been called
|
|
assert tm.create_task.call_count == 0
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_trigger_validation_skips_missing_policy
|
|
|
|
|
|
# #region test_trigger_validation_skips_inactive_policy [C:1] [TYPE Function]
|
|
def test_trigger_validation_skips_inactive_policy():
|
|
"""Verify _trigger_validation skips an inactive policy."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
svc.loop = asyncio.new_event_loop()
|
|
|
|
mock_policy = MagicMock()
|
|
mock_policy.id = "pol-inactive"
|
|
mock_policy.is_active = False
|
|
mock_policy.dashboard_ids = ["dash-1"]
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_policy
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db):
|
|
svc._trigger_validation("pol-inactive")
|
|
|
|
# create_task should not have been called for inactive policy
|
|
assert tm.create_task.call_count == 0
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_trigger_validation_skips_inactive_policy
|
|
|
|
|
|
# #region test_trigger_validation_skips_no_dashboards [C:1] [TYPE Function]
|
|
def test_trigger_validation_skips_no_dashboards():
|
|
"""Verify _trigger_validation skips a policy with no dashboards."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
svc.loop = asyncio.new_event_loop()
|
|
|
|
mock_policy = MagicMock()
|
|
mock_policy.id = "pol-empty"
|
|
mock_policy.is_active = True
|
|
mock_policy.dashboard_ids = []
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_policy
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db):
|
|
svc._trigger_validation("pol-empty")
|
|
|
|
assert tm.create_task.call_count == 0
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_trigger_validation_skips_no_dashboards
|
|
|
|
|
|
# #region test_remove_validation_job [C:1] [TYPE Function]
|
|
def test_remove_validation_job():
|
|
"""Verify remove_validation_job removes an existing job."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
svc.add_validation_job("pol-remove", cron_expression="0 10 * * 1")
|
|
assert sched.get_job("validation_pol-remove") is not None
|
|
|
|
svc.remove_validation_job("pol-remove")
|
|
assert sched.get_job("validation_pol-remove") is None
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_remove_validation_job
|
|
|
|
|
|
# #region test_remove_nonexistent_validation_job [C:1] [TYPE Function]
|
|
def test_remove_nonexistent_validation_job():
|
|
"""Verify remove_validation_job handles non-existent job silently."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
# Should not raise
|
|
svc.remove_validation_job("nonexistent-pol")
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_remove_nonexistent_validation_job
|
|
|
|
|
|
# #region test_reload_validation_policy_adds_job [C:1] [TYPE Function]
|
|
def test_reload_validation_policy_adds_job():
|
|
"""Verify reload_validation_policy registers a job for an active policy with schedule."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
# Mock DB to return an active policy with schedule_days
|
|
mock_policy = MagicMock()
|
|
mock_policy.id = "pol-reload"
|
|
mock_policy.is_active = True
|
|
mock_policy.schedule_days = [0, 2, 4]
|
|
mock_policy.window_start = time(10, 30)
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_policy
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db):
|
|
svc.reload_validation_policy("pol-reload")
|
|
|
|
job = sched.get_job("validation_pol-reload")
|
|
assert job is not None, "Job should be registered after reload"
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_reload_validation_policy_adds_job
|
|
|
|
|
|
# #region test_reload_validation_policy_removes_job_when_policy_gone [C:1] [TYPE Function]
|
|
def test_reload_validation_policy_removes_job_when_policy_gone():
|
|
"""Verify reload_validation_policy removes job when policy is missing or inactive."""
|
|
from src.core.scheduler import SchedulerService
|
|
|
|
sched = _start_scheduler()
|
|
try:
|
|
cfg = _make_mock_config_manager()
|
|
tm = _make_mock_task_manager()
|
|
|
|
svc = SchedulerService.__new__(SchedulerService)
|
|
svc.task_manager = tm
|
|
svc.config_manager = cfg
|
|
svc.scheduler = sched
|
|
|
|
# Pre-register a job
|
|
svc.add_validation_job("pol-gone", cron_expression="0 10 * * 1")
|
|
assert sched.get_job("validation_pol-gone") is not None
|
|
|
|
# Mock DB to return None (policy not found)
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
|
|
with patch("src.core.scheduler.SessionLocal", return_value=mock_db):
|
|
svc.reload_validation_policy("pol-gone")
|
|
|
|
# Job should have been removed
|
|
assert sched.get_job("validation_pol-gone") is None
|
|
finally:
|
|
_stop_scheduler(sched)
|
|
# #endregion test_reload_validation_policy_removes_job_when_policy_gone
|
|
# #endregion TestCoreScheduler
|