busya
4205618ee6
Warnings fixed: - datetime.utcnow() → datetime.now(UTC) across 48+ files (src/ + tests/) - datetime.utcnow (callback ref) → lambda: datetime.now(UTC) in model fields (18 files) - Pydantic class Config → model_config = ConfigDict(...) (16 files) - Pydantic .dict() → .model_dump() (8 files) - ConfigDict(allow_population_by_field_name=True) → validate_by_name=True - SQLAlchemy declarative_base() import path updated - FastAPI on_event → lifespan context manager (app.py) - Import sorting (ruff I001) auto-fixed across all files - Fixed broken re-export chains that ruff F401 cleanup broke: _validate_bcp47: service.py now imports from dictionary_validation directly job_to_response: _job_routes.py and test imports from service_utils directly fetch_datasource_metadata: restored re-export in service.py - Added missing TranslateJobService import in _job_routes.py (was deleted by F401) - Added ConfigDict(protected_namespaces=()) for DashboardDatasetItem schema field - pytest.ini: replaced deprecated importmode with asyncio_mode All 440 tests pass with zero deprecation warnings.
299 lines
10 KiB
Python
299 lines
10 KiB
Python
# #region TestTranslateSchedulerGuard [C:3] [TYPE Module] [SEMANTICS test, translate, scheduler, guard, concurrency, stale, pending]
|
|
# @BRIEF Verify execute_scheduled_translation concurrency guard and stale-PENDING clearance contracts.
|
|
# @RELATION BINDS_TO -> [execute_scheduled_translation]
|
|
# @TEST_EDGE: concurrent_run_skips — active RUNNING concurrent run → function skips and logs skipped_concurrent event
|
|
# @TEST_EDGE: recent_pending_run_not_stale — recent PENDING run (30 min old) → function skips, not stale
|
|
# @TEST_EDGE: stale_pending_cleared_and_proceeds — stale PENDING run (>1h old) → cleared as FAILED, execution proceeds
|
|
# @TEST_EDGE: multiple_stale_pending_all_cleared — multiple stale PENDING runs → all cleared as FAILED, execution proceeds
|
|
# @TEST_INVARIANT: concurrency_guard -> VERIFIED_BY: [test_concurrent_run_skips, test_recent_pending_run_not_stale]
|
|
# @TEST_INVARIANT: stale_pending_clearance -> VERIFIED_BY: [test_stale_pending_cleared_and_proceeds, test_multiple_stale_pending_all_cleared]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_most_recent_run(created_at_delta_days=60):
|
|
"""Build a MagicMock TranslationRun with given age."""
|
|
run = MagicMock()
|
|
run.id = "prev-run-1"
|
|
run.job_id = "job-1"
|
|
run.status = "COMPLETED"
|
|
run.insert_status = "succeeded"
|
|
run.created_at = datetime.now(UTC) - timedelta(days=created_at_delta_days)
|
|
return run
|
|
|
|
|
|
def _make_mock_run(run_id="run-1", job_id="job-1"):
|
|
"""Build a mock TranslationRun returned by start_run."""
|
|
run = MagicMock()
|
|
run.id = run_id
|
|
run.job_id = job_id
|
|
run.status = "PENDING"
|
|
run.trigger_type = None
|
|
run.total_records = 0
|
|
run.successful_records = 0
|
|
run.failed_records = 0
|
|
run.skipped_records = 0
|
|
run.insert_status = None
|
|
run.superset_execution_id = None
|
|
run.superset_execution_log = None
|
|
run.error_message = None
|
|
return run
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — concurrency guard and stale PENDING clearance
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# #region test_concurrent_run_skips [C:1] [TYPE Function]
|
|
def test_concurrent_run_skips():
|
|
"""Active RUNNING concurrent run → function skips and logs skipped_concurrent event."""
|
|
from src.plugins.translate.scheduler import execute_scheduled_translation
|
|
|
|
mock_db = MagicMock()
|
|
mock_session_maker = MagicMock(return_value=mock_db)
|
|
mock_config = MagicMock()
|
|
|
|
schedule = MagicMock()
|
|
schedule.id = "sched-1"
|
|
schedule.job_id = "job-1"
|
|
schedule.is_active = True
|
|
schedule.execution_mode = "full"
|
|
|
|
active_run = MagicMock()
|
|
active_run.id = "active-run-1"
|
|
active_run.job_id = "job-1"
|
|
active_run.status = "RUNNING"
|
|
|
|
q1 = MagicMock()
|
|
q1.filter.return_value.first.return_value = schedule
|
|
|
|
q2 = MagicMock()
|
|
q2.filter.return_value.order_by.return_value.first.return_value = active_run
|
|
|
|
mock_db.query.side_effect = [q1, q2]
|
|
|
|
with patch(
|
|
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
|
) as mock_orch_cls, patch(
|
|
"src.plugins.translate.scheduler.TranslationEventLog"
|
|
) as mock_event_log_cls:
|
|
mock_event_log = MagicMock()
|
|
mock_event_log_cls.return_value = mock_event_log
|
|
|
|
execute_scheduled_translation(
|
|
schedule_id="sched-1",
|
|
job_id="job-1",
|
|
db_session_maker=mock_session_maker,
|
|
config_manager=mock_config,
|
|
execution_mode="full",
|
|
)
|
|
|
|
mock_orch_cls.assert_not_called()
|
|
mock_event_log.log_event.assert_called_once()
|
|
event_call = mock_event_log.log_event.call_args
|
|
assert event_call.kwargs["event_type"] == "RUN_STARTED"
|
|
assert event_call.kwargs["payload"]["reason"] == "skipped_concurrent"
|
|
mock_db.close.assert_called_once()
|
|
# #endregion test_concurrent_run_skips
|
|
|
|
|
|
# #region test_recent_pending_run_not_stale [C:1] [TYPE Function]
|
|
def test_recent_pending_run_not_stale():
|
|
"""Recent PENDING run (30 min old) → function skips, not stale."""
|
|
from src.plugins.translate.scheduler import execute_scheduled_translation
|
|
|
|
mock_db = MagicMock()
|
|
mock_session_maker = MagicMock(return_value=mock_db)
|
|
mock_config = MagicMock()
|
|
|
|
schedule = MagicMock()
|
|
schedule.id = "sched-1"
|
|
schedule.job_id = "job-1"
|
|
schedule.is_active = True
|
|
schedule.execution_mode = "full"
|
|
|
|
pending_run = MagicMock()
|
|
pending_run.id = "pending-run-1"
|
|
pending_run.job_id = "job-1"
|
|
pending_run.status = "PENDING"
|
|
pending_run.created_at = datetime.now(UTC) - timedelta(minutes=30)
|
|
|
|
q1 = MagicMock()
|
|
q1.filter.return_value.first.return_value = schedule
|
|
|
|
q2 = MagicMock()
|
|
q2.filter.return_value.order_by.return_value.first.return_value = pending_run
|
|
|
|
mock_db.query.side_effect = [q1, q2]
|
|
|
|
with patch(
|
|
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
|
) as mock_orch_cls, patch(
|
|
"src.plugins.translate.scheduler.TranslationEventLog"
|
|
) as mock_event_log_cls:
|
|
mock_event_log = MagicMock()
|
|
mock_event_log_cls.return_value = mock_event_log
|
|
|
|
execute_scheduled_translation(
|
|
schedule_id="sched-1",
|
|
job_id="job-1",
|
|
db_session_maker=mock_session_maker,
|
|
config_manager=mock_config,
|
|
execution_mode="full",
|
|
)
|
|
|
|
mock_orch_cls.assert_not_called()
|
|
mock_event_log.log_event.assert_called_once()
|
|
event_call = mock_event_log.log_event.call_args
|
|
assert event_call.kwargs["event_type"] == "RUN_STARTED"
|
|
assert event_call.kwargs["payload"]["reason"] == "skipped_concurrent"
|
|
mock_db.close.assert_called_once()
|
|
# #endregion test_recent_pending_run_not_stale
|
|
|
|
|
|
# #region test_stale_pending_cleared_and_proceeds [C:1] [TYPE Function]
|
|
def test_stale_pending_cleared_and_proceeds():
|
|
"""Stale PENDING run (>1h old) → cleared as FAILED, execution proceeds."""
|
|
from src.plugins.translate.scheduler import execute_scheduled_translation
|
|
|
|
mock_db = MagicMock()
|
|
mock_session_maker = MagicMock(return_value=mock_db)
|
|
mock_config = MagicMock()
|
|
|
|
schedule = MagicMock()
|
|
schedule.id = "sched-1"
|
|
schedule.job_id = "job-1"
|
|
schedule.is_active = True
|
|
schedule.execution_mode = "full"
|
|
|
|
stale_run = MagicMock()
|
|
stale_run.id = "stale-run-1"
|
|
stale_run.job_id = "job-1"
|
|
stale_run.status = "PENDING"
|
|
stale_run.created_at = datetime.now(UTC) - timedelta(hours=2)
|
|
|
|
most_recent = _make_most_recent_run(created_at_delta_days=30)
|
|
|
|
q1 = MagicMock()
|
|
q1.filter.return_value.first.return_value = schedule
|
|
|
|
q2 = MagicMock()
|
|
q2.filter.return_value.order_by.return_value.first.return_value = stale_run
|
|
|
|
q3 = MagicMock()
|
|
q3.filter.return_value.all.return_value = [stale_run]
|
|
|
|
q4 = MagicMock()
|
|
q4.filter.return_value.order_by.return_value.first.return_value = most_recent
|
|
|
|
mock_db.query.side_effect = [q1, q2, q3, q4]
|
|
|
|
mock_run = _make_mock_run()
|
|
|
|
with patch(
|
|
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
|
) as mock_orch_cls:
|
|
mock_orch = MagicMock()
|
|
mock_orch.start_run.return_value = mock_run
|
|
mock_orch_cls.return_value = mock_orch
|
|
|
|
execute_scheduled_translation(
|
|
schedule_id="sched-1",
|
|
job_id="job-1",
|
|
db_session_maker=mock_session_maker,
|
|
config_manager=mock_config,
|
|
execution_mode="full",
|
|
)
|
|
|
|
assert stale_run.status == "FAILED"
|
|
assert "Stale: concurrency deadlock" in stale_run.error_message
|
|
assert stale_run.completed_at is not None
|
|
|
|
mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True)
|
|
mock_orch.execute_run.assert_called_once()
|
|
assert mock_db.commit.call_count >= 1
|
|
mock_db.close.assert_called_once()
|
|
# #endregion test_stale_pending_cleared_and_proceeds
|
|
|
|
|
|
# #region test_multiple_stale_pending_all_cleared [C:1] [TYPE Function]
|
|
def test_multiple_stale_pending_all_cleared():
|
|
"""Multiple stale PENDING runs → all cleared as FAILED, execution proceeds."""
|
|
from src.plugins.translate.scheduler import execute_scheduled_translation
|
|
|
|
mock_db = MagicMock()
|
|
mock_session_maker = MagicMock(return_value=mock_db)
|
|
mock_config = MagicMock()
|
|
|
|
schedule = MagicMock()
|
|
schedule.id = "sched-1"
|
|
schedule.job_id = "job-1"
|
|
schedule.is_active = True
|
|
schedule.execution_mode = "full"
|
|
|
|
stale_run_a = MagicMock()
|
|
stale_run_a.id = "stale-run-a"
|
|
stale_run_a.job_id = "job-1"
|
|
stale_run_a.status = "PENDING"
|
|
stale_run_a.created_at = datetime.now(UTC) - timedelta(hours=5)
|
|
|
|
stale_run_b = MagicMock()
|
|
stale_run_b.id = "stale-run-b"
|
|
stale_run_b.job_id = "job-1"
|
|
stale_run_b.status = "PENDING"
|
|
stale_run_b.created_at = datetime.now(UTC) - timedelta(hours=3)
|
|
|
|
most_recent = _make_most_recent_run(created_at_delta_days=30)
|
|
|
|
q1 = MagicMock()
|
|
q1.filter.return_value.first.return_value = schedule
|
|
|
|
q2 = MagicMock()
|
|
q2.filter.return_value.order_by.return_value.first.return_value = stale_run_a
|
|
|
|
q3 = MagicMock()
|
|
q3.filter.return_value.all.return_value = [stale_run_a, stale_run_b]
|
|
|
|
q4 = MagicMock()
|
|
q4.filter.return_value.order_by.return_value.first.return_value = most_recent
|
|
|
|
mock_db.query.side_effect = [q1, q2, q3, q4]
|
|
|
|
mock_run = _make_mock_run()
|
|
|
|
with patch(
|
|
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
|
) as mock_orch_cls:
|
|
mock_orch = MagicMock()
|
|
mock_orch.start_run.return_value = mock_run
|
|
mock_orch_cls.return_value = mock_orch
|
|
|
|
execute_scheduled_translation(
|
|
schedule_id="sched-1",
|
|
job_id="job-1",
|
|
db_session_maker=mock_session_maker,
|
|
config_manager=mock_config,
|
|
execution_mode="full",
|
|
)
|
|
|
|
for sr in [stale_run_a, stale_run_b]:
|
|
assert sr.status == "FAILED"
|
|
assert "Stale: concurrency deadlock" in sr.error_message
|
|
assert sr.completed_at is not None
|
|
|
|
mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True)
|
|
mock_orch.execute_run.assert_called_once()
|
|
mock_db.close.assert_called_once()
|
|
# #endregion test_multiple_stale_pending_all_cleared
|
|
|
|
# #endregion TestTranslateSchedulerGuard
|