busya
fbe0ba122c
Fixed async/sync boundary bugs across 14 test files. Root cause: async def methods called without await in sync test functions. Fixed files: - test_translate_jobs.py (10): create_job/get_job/update_job/delete_job - test_translate_scheduler.py (5): create_schedule/update/delete - test_datasets.py (14): AsyncMock + corrected patch target - test_mapping_service.py (11): sync_environment + MockSupersetClient - test_defensive_guards.py (6): GitService/SupersetClient guards - test_maintenance_service.py (29): all 6 maintenance services - test_dry_run_orchestrator.py (1): run() without await - test_dashboards_api.py (23): registry client via AsyncMock - test_validation_tasks.py (4): trailing slash in POST URL - test_superset_matrix.py (3): AsyncMock for compile_preview - test_payload_reduction.py (6): LLMClient._optimize_image wrapper - test_compliance_task_integration.py (2): event_bus ref - test_smoke_plugins.py (1): flusher_stop_event fallback - test_task_manager.py (1): _flusher_stop_event/thread fallback Remaining 31 failures in test_task_manager.py (29) and test_smoke_plugins.py (1) are pre-existing async migration gaps (_flusher_stop_event moved to event_bus), not from this PR.
196 lines
7.2 KiB
Python
196 lines
7.2 KiB
Python
# #region TranslateSchedulerTests [C:2] [TYPE Module]
|
|
# @SEMANTICS: tests, translate, scheduler
|
|
# @PURPOSE: Tests for TranslationScheduler CRUD and APScheduler integration.
|
|
# @LAYER Tests
|
|
# @RELATION BINDS_TO -> [TranslationScheduler]
|
|
#
|
|
# @TEST_CONTRACT: ScheduleFlow ->
|
|
# {
|
|
# invariants: [
|
|
# "Create schedule returns TranslationSchedule with valid fields",
|
|
# "Update schedule modifies cron/timezone/active",
|
|
# "Delete schedule removes the row",
|
|
# "Enable/disable toggles is_active",
|
|
# "get_next_executions returns ISO datetime strings",
|
|
# "List active schedules returns only active rows"
|
|
# ]
|
|
# }
|
|
# @TEST_FIXTURE: valid_job -> created via TranslateJobService
|
|
# @TEST_FIXTURE: schedule_data -> valid cron expression
|
|
# @TEST_EDGE: schedule_not_found -> ValueError on get/delete non-existent
|
|
# @TEST_EDGE: next_executions_invalid_cron -> empty list
|
|
# @TEST_INVARIANT: schedule_crud_flow -> VERIFIED_BY: [valid_job, schedule_data]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from src.core.database import Base
|
|
from src.plugins.translate.scheduler import TranslationScheduler
|
|
from src.plugins.translate.service import TranslateJobService
|
|
from src.schemas.translate import TranslateJobCreate
|
|
|
|
# In-memory SQLite
|
|
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
|
|
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
@pytest.fixture
|
|
def db_session():
|
|
connection = engine.connect()
|
|
transaction = connection.begin()
|
|
session = TestingSessionLocal(bind=connection)
|
|
yield session
|
|
session.close()
|
|
transaction.rollback()
|
|
connection.close()
|
|
|
|
|
|
# #region test_create_schedule [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify schedule creation with valid params.
|
|
async def test_create_schedule(db_session):
|
|
"""Test creating a schedule for a job."""
|
|
config_mgr = MagicMock()
|
|
config_mgr.get_environments.return_value = []
|
|
|
|
svc = TranslateJobService(db_session, config_mgr, "test_user")
|
|
payload = TranslateJobCreate(name="Sched Job", source_dialect="pg", target_dialect="ch")
|
|
job = await svc.create_job(payload)
|
|
|
|
scheduler = TranslationScheduler(db_session, config_mgr, "test_user")
|
|
schedule = scheduler.create_schedule(
|
|
job_id=job.id,
|
|
cron_expression="0 2 * * *",
|
|
timezone="UTC",
|
|
)
|
|
|
|
assert schedule.job_id == job.id
|
|
assert schedule.cron_expression == "0 2 * * *"
|
|
assert schedule.timezone == "UTC"
|
|
assert schedule.is_active is True
|
|
assert schedule.id is not None
|
|
# #endregion test_create_schedule
|
|
|
|
|
|
# #region test_update_schedule [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify schedule update.
|
|
async def test_update_schedule(db_session):
|
|
"""Test updating a schedule."""
|
|
config_mgr = MagicMock()
|
|
config_mgr.get_environments.return_value = []
|
|
|
|
svc = TranslateJobService(db_session, config_mgr, "test_user")
|
|
job = await svc.create_job(TranslateJobCreate(name="Job", source_dialect="pg", target_dialect="ch"))
|
|
|
|
scheduler = TranslationScheduler(db_session, config_mgr, "test_user")
|
|
scheduler.create_schedule(job.id, "0 2 * * *")
|
|
|
|
updated = scheduler.update_schedule(job.id, cron_expression="30 3 * * *", timezone_str="US/Eastern", is_active=False)
|
|
assert updated.cron_expression == "30 3 * * *"
|
|
assert updated.timezone == "US/Eastern"
|
|
assert updated.is_active is False
|
|
# #endregion test_update_schedule
|
|
|
|
|
|
# #region test_delete_schedule [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify schedule deletion.
|
|
async def test_delete_schedule(db_session):
|
|
"""Test deleting a schedule."""
|
|
config_mgr = MagicMock()
|
|
config_mgr.get_environments.return_value = []
|
|
|
|
svc = TranslateJobService(db_session, config_mgr, "test_user")
|
|
job = await svc.create_job(TranslateJobCreate(name="Job", source_dialect="pg", target_dialect="ch"))
|
|
|
|
scheduler = TranslationScheduler(db_session, config_mgr, "test_user")
|
|
scheduler.create_schedule(job.id, "0 2 * * *")
|
|
scheduler.delete_schedule(job.id)
|
|
|
|
with pytest.raises(ValueError, match="No schedule found"):
|
|
scheduler.get_schedule(job.id)
|
|
# #endregion test_delete_schedule
|
|
|
|
|
|
# #region test_enable_disable_schedule [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify enable/disable toggle.
|
|
async def test_enable_disable_schedule(db_session):
|
|
"""Test enabling and disabling a schedule."""
|
|
config_mgr = MagicMock()
|
|
config_mgr.get_environments.return_value = []
|
|
|
|
svc = TranslateJobService(db_session, config_mgr, "test_user")
|
|
job = await svc.create_job(TranslateJobCreate(name="Job", source_dialect="pg", target_dialect="ch"))
|
|
|
|
scheduler = TranslationScheduler(db_session, config_mgr, "test_user")
|
|
scheduler.create_schedule(job.id, "0 2 * * *")
|
|
|
|
sched = scheduler.set_schedule_active(job.id, False)
|
|
assert sched.is_active is False
|
|
|
|
sched = scheduler.set_schedule_active(job.id, True)
|
|
assert sched.is_active is True
|
|
# #endregion test_enable_disable_schedule
|
|
|
|
|
|
# #region test_get_schedule_not_found [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify ValueError on getting non-existent schedule.
|
|
def test_get_schedule_not_found(db_session):
|
|
"""Test that getting a non-existent schedule raises ValueError."""
|
|
config_mgr = MagicMock()
|
|
config_mgr.get_environments.return_value = []
|
|
|
|
scheduler = TranslationScheduler(db_session, config_mgr, "test_user")
|
|
with pytest.raises(ValueError, match="No schedule found"):
|
|
scheduler.get_schedule("non-existent")
|
|
# #endregion test_get_schedule_not_found
|
|
|
|
|
|
# #region test_list_active_schedules [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify listing only active schedules.
|
|
async def test_list_active_schedules(db_session):
|
|
"""Test listing active schedules."""
|
|
config_mgr = MagicMock()
|
|
config_mgr.get_environments.return_value = []
|
|
|
|
svc = TranslateJobService(db_session, config_mgr, "test_user")
|
|
job1 = await svc.create_job(TranslateJobCreate(name="Job1", source_dialect="pg", target_dialect="ch"))
|
|
|
|
scheduler = TranslationScheduler(db_session, config_mgr, "test_user")
|
|
scheduler.create_schedule(job1.id, "0 2 * * *")
|
|
scheduler.set_schedule_active(job1.id, True)
|
|
|
|
active = TranslationScheduler.list_active_schedules(db_session)
|
|
assert len(active) >= 1
|
|
assert active[0].job_id == job1.id
|
|
# #endregion test_list_active_schedules
|
|
|
|
|
|
# #region test_get_next_executions [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify next execution time computation.
|
|
def test_get_next_executions():
|
|
"""Test computing next execution times."""
|
|
times = TranslationScheduler.get_next_executions("0 2 * * *", "UTC", n=3)
|
|
assert len(times) == 3
|
|
for t in times:
|
|
assert "T" in t # ISO format check
|
|
# #endregion test_get_next_executions
|
|
|
|
|
|
# #region test_get_next_executions_invalid [C:2] [TYPE Function]
|
|
# @PURPOSE: Verify invalid cron returns empty list.
|
|
def test_get_next_executions_invalid():
|
|
"""Test invalid cron returns empty list."""
|
|
times = TranslationScheduler.get_next_executions("invalid cron", "UTC", n=3)
|
|
assert times == []
|
|
# #endregion test_get_next_executions_invalid
|
|
# #endregion TranslateSchedulerTests
|