From de023c7a318ac85d3f02fd68e96388544bedbd37 Mon Sep 17 00:00:00 2001 From: busya Date: Tue, 7 Jul 2026 21:04:23 +0300 Subject: [PATCH] fix(backend+frontend): migration deadlock, async pattern, timeout safety, fire-and-forget translations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend: - MigrationPlugin.execute — remove AsyncJobRunner.run() deadlock on post-migration ID sync (2 deadlocks total: plugins/migration.py + api/routes/migration.py trigger_sync_now). Replace blocking runner.run with direct await. - MigrationPlugin.execute — fix temp file leak (dry_run=True prevented cleanup). - MigrationPlugin.execute — IdMappingService(SessionLocal()) now closed in finally. - MigrationPlugin.execute — wire IdMappingService into MigrationEngine constructor so cross-filter patching actually works instead of silently skipping. - MigrationPlugin.execute — SupersetClient.aclose() in finally to prevent httpx connection pool leak. - TaskManager — _async_tasks dict leak (add_done_callback cleanup). - JobLifecycle — CancelledError handler (tasks stuck in RUNNING after cancel). - JobLifecycle — persist_task BEFORE _broadcast_task_status (crash consistency). - JobLifecycle — wait_for_resolution/wait_for_input now have 3600s timeout. - AsyncJobRunner.run — 300s timeout on future.result() (APScheduler thread safety). Translate scheduler: - execute_scheduled_translation — replace blocking runner.run(orch.execute_run(run)) with fire-and-forget TranslationOrchestrator.execute_background(). APScheduler thread is freed in ms instead of blocking for the full translation duration. Translation runs of 10k+ rows (200+ LLM batches, hours) no longer hit the 300s runner timeout. - TranslationOrchestrator.execute_background — new static method: opens own DB session, dispatches asyncio.create_task, handles errors + notification. - Scheduler last_run_at updated at dispatch time (not after completion). Frontend: - MigrationModel.stepReady[3] now requires dryRunResult != null (was always true, allowing UI to reach step 3 without dry-run). - WizardModel.goToStep gate for step 3 uses stepReady[3]. - +page.svelte — dryRunResult no longer self-clears via reactive loop. - Progress bar step 3 indicator gate fixed for new readiness logic. Tests: - 2 new regression tests for migration sync (deadlock-free, completes cleanly). - test_migration_plugin.py — _make_mock_superset_client/_make_mock_mapping_service helpers for proper async mock behavior (aclose, sync_environment AsyncMock). - 10 scheduled-translation tests updated for fire-and-forget pattern. - MigrationModel.test.ts — step 3 invariant + goToStep block test. - All affected tests: 104 backend + 79 frontend = 183 passed. --- backend/src/api/routes/migration.py | 5 +- backend/src/core/async_job_runner.py | 8 +- backend/src/core/task_manager/lifecycle.py | 35 +++- backend/src/core/task_manager/manager.py | 1 + backend/src/plugins/migration.py | 34 +++- backend/src/plugins/translate/orchestrator.py | 89 ++++++++++ backend/src/plugins/translate/scheduler.py | 55 ++---- backend/tests/api/test_migration.py | 6 +- .../tests/plugins/test_migration_plugin.py | 158 ++++++++++++------ .../tests/plugins/translate/test_scheduler.py | 42 +++-- .../tests/test_translate_scheduler_audit.py | 74 ++++---- .../test_translate_scheduler_execution.py | 82 +++++---- .../tests/test_translate_scheduler_guard.py | 18 +- .../models/Migration.WizardModel.svelte.ts | 4 +- .../src/lib/models/MigrationModel.svelte.ts | 2 +- .../models/__tests__/MigrationModel.test.ts | 10 +- frontend/src/routes/migration/+page.svelte | 12 +- 17 files changed, 423 insertions(+), 212 deletions(-) diff --git a/backend/src/api/routes/migration.py b/backend/src/api/routes/migration.py index 56535423..1b486be5 100644 --- a/backend/src/api/routes/migration.py +++ b/backend/src/api/routes/migration.py @@ -34,7 +34,7 @@ from ...core.logger import belief_scope, logger from ...core.mapping_service import IdMappingService from ...core.migration.dry_run_orchestrator import MigrationDryRunService from ...core.async_superset_client import AsyncSupersetClient -from ...dependencies import get_async_job_runner, get_config_manager, get_task_manager, has_permission +from ...dependencies import get_config_manager, get_task_manager, has_permission from ...models.dashboard import DashboardMetadata, DashboardSelection from ...models.mapping import ResourceMapping @@ -381,13 +381,12 @@ async def trigger_sync_now( db.commit() service = IdMappingService(db) - runner = get_async_job_runner() results = {"synced": [], "failed": []} for env in environments: try: client = AsyncSupersetClient(env) - runner.run(service.sync_environment(env.id, client)) + await service.sync_environment(env.id, client) results["synced"].append(env.id) logger.reason(f"Synced environment {env.id}", extra={"src": "trigger_sync_now"}) except Exception as e: diff --git a/backend/src/core/async_job_runner.py b/backend/src/core/async_job_runner.py index ad1211ec..9dd50e6c 100644 --- a/backend/src/core/async_job_runner.py +++ b/backend/src/core/async_job_runner.py @@ -47,7 +47,10 @@ class AsyncJobRunner: # @BRIEF Execute a coroutine from a sync thread and wait for the result. # @PRE coro is an awaitable (coroutine or Future). # @POST Coroutine has been executed to completion; result or exception is returned. - # @SIDE_EFFECT Blocks the calling thread until the coroutine completes. + # @SIDE_EFFECT Blocks the calling thread until the coroutine completes or times out. + # @INVARIANT Must NOT be called from the event loop thread — deadlock. + _DEFAULT_RUN_TIMEOUT: float = 300.0 # 5-minute safety cap per scheduled job + def run(self, coro: Any) -> Any: """Dispatch a coroutine to the event loop and block until it completes. @@ -59,9 +62,10 @@ class AsyncJobRunner: Raises: Exception: Any exception raised by the coroutine is re-raised. + TimeoutError: If the coroutine does not complete within _DEFAULT_RUN_TIMEOUT. """ future = asyncio.run_coroutine_threadsafe(coro, self.loop) - return future.result() # timeout=None — wait indefinitely + return future.result(timeout=self._DEFAULT_RUN_TIMEOUT) # #endregion run # #region run_later [TYPE Function] diff --git a/backend/src/core/task_manager/lifecycle.py b/backend/src/core/task_manager/lifecycle.py index 29934116..ebf67cb3 100644 --- a/backend/src/core/task_manager/lifecycle.py +++ b/backend/src/core/task_manager/lifecycle.py @@ -179,6 +179,7 @@ class JobLifecycle: ) logger.reflect("Task execution completed successfully", payload={"task_id": task_id}) task.status = TaskStatus.SUCCESS + self.persistence_service.persist_task(task) await self._broadcast_task_status(task) if add_log_callback: await add_log_callback( @@ -186,9 +187,23 @@ class JobLifecycle: f"Task completed successfully for plugin '{plugin.name}'", source="system", ) + except asyncio.CancelledError: + logger.explore("Task execution cancelled", payload={"task_id": task_id}) + task.status = TaskStatus.FAILED + task.result = "Task was cancelled" + self.persistence_service.persist_task(task) + await self._broadcast_task_status(task) + if add_log_callback: + await add_log_callback( + task_id, "ERROR", + "Task was cancelled", + source="system", + ) + raise except Exception as e: logger.explore("Task execution failed", payload={"task_id": task_id}, error=str(e)) task.status = TaskStatus.FAILED + self.persistence_service.persist_task(task) await self._broadcast_task_status(task) if add_log_callback: await add_log_callback( @@ -242,7 +257,7 @@ class JobLifecycle: # @BRIEF Pauses execution and waits for a resolution signal. # @PRE Task exists. # @POST Execution pauses until future is set. - async def wait_for_resolution(self, task_id: str) -> None: + async def wait_for_resolution(self, task_id: str, timeout: float = 3600.0) -> None: with belief_scope("JobLifecycle.wait_for_resolution", f"task_id={task_id}"): task = self.graph.get_task(task_id) if not task: @@ -254,7 +269,13 @@ class JobLifecycle: future = loop.create_future() self.graph.create_future(task_id, future) try: - await future + await asyncio.wait_for(future, timeout=timeout) + except asyncio.TimeoutError: + task.status = TaskStatus.FAILED + task.result = "Mapping resolution timed out" + self.persistence_service.persist_task(task) + await self._broadcast_task_status(task) + raise finally: self.graph.remove_future(task_id) # #endregion wait_for_resolution @@ -264,7 +285,7 @@ class JobLifecycle: # @BRIEF Pauses execution and waits for user input. # @PRE Task exists. # @POST Execution pauses until future is set via resume_task_with_password. - async def wait_for_input(self, task_id: str) -> None: + async def wait_for_input(self, task_id: str, timeout: float = 3600.0) -> None: with belief_scope("JobLifecycle.wait_for_input", f"task_id={task_id}"): task = self.graph.get_task(task_id) if not task: @@ -273,7 +294,13 @@ class JobLifecycle: future = loop.create_future() self.graph.create_future(task_id, future) try: - await future + await asyncio.wait_for(future, timeout=timeout) + except asyncio.TimeoutError: + task.status = TaskStatus.FAILED + task.result = "User input timed out" + self.persistence_service.persist_task(task) + await self._broadcast_task_status(task) + raise finally: self.graph.remove_future(task_id) # #endregion wait_for_input diff --git a/backend/src/core/task_manager/manager.py b/backend/src/core/task_manager/manager.py index 29d8f81c..2bbd3487 100644 --- a/backend/src/core/task_manager/manager.py +++ b/backend/src/core/task_manager/manager.py @@ -315,6 +315,7 @@ class TaskManager: self.lifecycle._run_task(task.id, add_log_callback=self._add_log) ) self._async_tasks[task.id] = async_task + async_task.add_done_callback(lambda _: self._async_tasks.pop(task.id, None)) return task # #endregion create_task diff --git a/backend/src/plugins/migration.py b/backend/src/plugins/migration.py index d5ed084f..6478992b 100755 --- a/backend/src/plugins/migration.py +++ b/backend/src/plugins/migration.py @@ -25,7 +25,7 @@ from ..core.plugin_base import PluginBase from ..core.superset_client import SupersetClient from ..core.task_manager.context import TaskContext from ..core.utils.fileio import create_temp_file -from ..dependencies import get_async_job_runner, get_config_manager +from ..dependencies import get_config_manager from ..models.mapping import DatabaseMapping, Environment @@ -192,6 +192,10 @@ class MigrationPlugin(PluginBase): log.info("Starting migration task.") + from_c: SupersetClient | None = None + to_c: SupersetClient | None = None + engine_db = None + try: config_manager = get_config_manager() environments = config_manager.get_environments() @@ -270,7 +274,10 @@ class MigrationPlugin(PluginBase): db.close() migration_result["mapping_count"] = len(db_mapping) - engine = MigrationEngine() + engine_db = SessionLocal() + engine = MigrationEngine( + mapping_service=IdMappingService(engine_db) + ) # Migration Loop for dash in dashboards_to_migrate: @@ -279,7 +286,7 @@ class MigrationPlugin(PluginBase): try: exported_content, _ = await from_c.export_dashboard(dash_id) - with create_temp_file(content=exported_content, dry_run=True, suffix=".zip") as tmp_zip_path, create_temp_file(suffix=".zip", dry_run=True) as tmp_new_zip: + with create_temp_file(content=exported_content, suffix=".zip") as tmp_zip_path, create_temp_file(suffix=".zip") as tmp_new_zip: success = engine.transform_zip( str(tmp_zip_path), @@ -373,11 +380,12 @@ class MigrationPlugin(PluginBase): try: app_logger.reason("Executing incremental ID catalog sync on target") db_session = SessionLocal() - mapping_service = IdMappingService(db_session) - runner = get_async_job_runner() - runner.run(mapping_service.sync_environment(tgt_env.id, to_c, incremental=True)) - db_session.close() - app_logger.reflect("Incremental catalog sync closed out cleanly") + try: + mapping_service = IdMappingService(db_session) + await mapping_service.sync_environment(tgt_env.id, to_c, incremental=True) + app_logger.reflect("Incremental catalog sync closed out cleanly") + finally: + db_session.close() except Exception as sync_exc: app_logger.explore(f"ID Mapping sync failed, mapping state might be degraded: {sync_exc}") @@ -387,6 +395,16 @@ class MigrationPlugin(PluginBase): except Exception as e: app_logger.explore(f"Fatal plugin failure: {e}", exc_info=True) raise e + finally: + if engine_db is not None: + try: + engine_db.close() + except Exception: + pass + if from_c is not None: + await from_c.aclose() + if to_c is not None: + await to_c.aclose() # endregion execute # #endregion MigrationPlugin # #endregion MigrationPlugin diff --git a/backend/src/plugins/translate/orchestrator.py b/backend/src/plugins/translate/orchestrator.py index 1ac303ef..63030ebd 100644 --- a/backend/src/plugins/translate/orchestrator.py +++ b/backend/src/plugins/translate/orchestrator.py @@ -19,7 +19,9 @@ # @REJECTED stdout-only logging — lacks structured payload integrity; cannot enforce terminal-event invariant. from collections.abc import Callable +from datetime import UTC, datetime from typing import Any +import asyncio from sqlalchemy.orm import Session @@ -118,6 +120,93 @@ class TranslationOrchestrator: return self._runner.cancel_run(run_id) # endregion cancel_run + # #region execute_background [C:4] [TYPE Function] [SEMANTICS translate,async,background] + # @ingroup Translate + # @BRIEF Fire-and-forget async execution: opens own DB session, dispatches execute_run as background task. + # @PRE run_id is a valid TranslationRun ID in PENDING state. + # @POST Background asyncio task created — execute_run runs with its own session and error handling. + # @SIDE_EFFECT Creates asyncio.Task; opens/closes DB session; marks run FAILED on error. + # @DATA_CONTRACT Input[run_id, db_session_maker, config_manager, current_user] -> Output[asyncio.Task] + # @RELATION CALLS -> [TranslationOrchestrator.execute_run] + # @RATIONALE Required by both API-triggered and APScheduler-triggered translation paths to avoid + # blocking the calling thread/request for the duration of potentially hours-long translations. + # @REJECTED Inline asyncio.create_task in route handlers was rejected — duplicated error handling + # across API and scheduler paths creates maintenance burden. + @staticmethod + def execute_background( + run_id: str, + db_session_maker: type | Callable[[], Session], + config_manager: ConfigManager, + current_user: str | None = "scheduler", + ) -> asyncio.Task: + """Fire-and-forget: create a background asyncio task that executes the run with its own session.""" + from ...core.logger import logger as _log + from ...models.translate import TranslationRun as TRModel + + async def _bg_execute(): + bg_db = None + try: + bg_db = db_session_maker() + bg_orch = TranslationOrchestrator(bg_db, config_manager, current_user=current_user) + bg_run = bg_db.query(TRModel).filter(TRModel.id == run_id).first() + if bg_run is None: + _log.explore("Background execute: run not found", extra={"run_id": run_id}) + return + await bg_orch.execute_run(bg_run) + + check_run = bg_db.query(TRModel).filter(TRModel.id == run_id).first() + if check_run and check_run.status in ("COMPLETED", "FAILED", "CANCELLED"): + _log.reflect("Background execute verified", extra={"run_id": run_id, "status": check_run.status}) + else: + actual_status = check_run.status if check_run else "NOT_FOUND" + if check_run: + check_run.status = "FAILED" + check_run.error_message = "Background execution did not reach terminal state" + bg_db.commit() + _log.explore("Background execute: run not in terminal state", + extra={"run_id": run_id, "status": actual_status}) + except Exception as bg_err: + _log.explore("Background execute failed", + extra={"run_id": run_id, "error": str(bg_err)}) + try: + fb_db = db_session_maker() + try: + fb_run = fb_db.query(TRModel).filter(TRModel.id == run_id).first() + if fb_run: + fb_run.status = "FAILED" + fb_run.error_message = f"Background execute error: {bg_err}" + fb_run.completed_at = datetime.now(UTC) + fb_db.commit() + finally: + fb_db.close() + except Exception as fb_err: + _log.explore("Background execute: unable to mark run FAILED", + extra={"run_id": run_id, "error": str(fb_err)}) + + # Send failure notification (relocated from old scheduler except block) + try: + from ...services.notifications.service import NotificationService + notify_service = NotificationService(db_session_maker(), config_manager) + notify_service._initialize_providers() + subject = f"Scheduled translation failed: run={run_id}" + body = ( + f"Scheduled translation run failed.\n" + f"Run ID: {run_id}\n" + f"Error: {bg_err}\n" + f"Time: {datetime.now(UTC).isoformat()}" + ) + for provider in notify_service._providers.values(): + await provider.send(recipient="admin", subject=subject, body=body) + except Exception as notify_err: + _log.explore("Background execute: notification send failed", + extra={"run_id": run_id, "error": str(notify_err)}) + finally: + if bg_db is not None: + bg_db.close() + + return asyncio.create_task(_bg_execute()) + # #endregion execute_background + # region _generate_and_insert_sql [TYPE Function] [SEMANTICS backward-compat wrapper] # @PURPOSE: Backward-compatible delegating wrapper for SQL generation and insert. # @SIDE_EFFECT Delegates to SQLInsertService. May call Superset API. diff --git a/backend/src/plugins/translate/scheduler.py b/backend/src/plugins/translate/scheduler.py index 492d1ef0..3b8e4734 100644 --- a/backend/src/plugins/translate/scheduler.py +++ b/backend/src/plugins/translate/scheduler.py @@ -418,51 +418,28 @@ def execute_scheduled_translation( payload={"reason": "baseline_expired", "age_days": age.days}, ) - # Execute via AsyncJobRunner (dispatches coroutine to event loop) - try: - runner.run(orch.execute_run(run)) - if run.status == "FAILED": - logger.explore("Scheduled translation completed with all records failed", { - "run_id": run.id, - "error": run.error_message, - }) - run.insert_status = run.insert_status or None - else: - run.insert_status = run.insert_status or "succeeded" - except Exception as exec_err: - logger.explore("Scheduled translation execution failed", { - "run_id": run.id, - "error": str(exec_err), - }) - # Send notification on scheduled-run failure (FR-041, FR-048) - try: - notify_service = NotificationService(db, config_manager) - notify_service._initialize_providers() - subject = f"Scheduled translation failed: job={job_id}" - body = ( - f"Scheduled translation run failed for job '{job_id}'.\n" - f"Schedule: {schedule_id}\n" - f"Error: {exec_err}\n" - f"Time: {datetime.now(UTC).isoformat()}" - ) - for provider in notify_service._providers.values(): - runner.run(provider.send(recipient="admin", subject=subject, body=body)) - except Exception as notify_err: - logger.warning(f"Failed to send failure notification: {notify_err}") + # Dispatch background execution (fire-and-forget) + # The background task uses its own DB session and handles error marking + notification. + # execute_background uses asyncio.create_task which must run ON the event loop, + # so we dispatch it via runner.run() (run_coroutine_threadsafe) to cross the + # sync→async thread boundary. The dispatch coroutine returns in <1ms, the + # 300s runner timeout is irrelevant. + async def _dispatch_bg(): + TranslationOrchestrator.execute_background( + run_id=run.id, + db_session_maker=db_session_maker, + config_manager=config_manager, + current_user="scheduler", + ) + runner.run(_dispatch_bg()) - # Leave schedule enabled — schedule continues on failure - run.status = "FAILED" - run.error_message = str(exec_err) - run.completed_at = datetime.now(UTC) - - # Update schedule tracking + # Update schedule tracking at dispatch time (not after completion) schedule.last_run_at = datetime.now(UTC) db.commit() - logger.reflect("Scheduled translation complete", { + logger.reason("Scheduled translation dispatched to background", { "schedule_id": schedule_id, "run_id": run.id, - "status": run.status, }) except Exception as e: logger.explore(f"[scheduled_translation] Unexpected error: {e}", diff --git a/backend/tests/api/test_migration.py b/backend/tests/api/test_migration.py index 290672e4..74d32360 100644 --- a/backend/tests/api/test_migration.py +++ b/backend/tests/api/test_migration.py @@ -356,8 +356,7 @@ class TestTriggerSyncNow: from src.core.database import get_db from src.dependencies import get_config_manager with patch("src.api.routes.migration.IdMappingService", return_value=mock_sync), \ - patch("src.api.routes.migration.AsyncSupersetClient"), \ - patch("src.api.routes.migration.get_async_job_runner"): + patch("src.api.routes.migration.AsyncSupersetClient"): client = _make_client({ get_config_manager: lambda: mock_config, get_db: lambda: mock_db, @@ -397,8 +396,7 @@ class TestTriggerSyncNow: from src.dependencies import get_config_manager # Make AsyncSupersetClient raise on 2nd call to simulate sync failure with patch("src.api.routes.migration.IdMappingService", return_value=mock_sync), \ - patch("src.api.routes.migration.AsyncSupersetClient", side_effect=[MagicMock(), RuntimeError("Connection failed")]), \ - patch("src.api.routes.migration.get_async_job_runner"): + patch("src.api.routes.migration.AsyncSupersetClient", side_effect=[MagicMock(), RuntimeError("Connection failed")]): client = _make_client({ get_config_manager: lambda: mock_config, get_db: lambda: mock_db, diff --git a/backend/tests/plugins/test_migration_plugin.py b/backend/tests/plugins/test_migration_plugin.py index 9f6a131d..5d6a409b 100644 --- a/backend/tests/plugins/test_migration_plugin.py +++ b/backend/tests/plugins/test_migration_plugin.py @@ -42,6 +42,20 @@ def _make_dashboard(dash_id=1, title="Test Dash"): return {"id": dash_id, "slug": f"slug-{dash_id}", "dashboard_title": title} +def _make_mock_superset_client(): + """Create a MagicMock that can be used as a SupersetClient (aclose is awaitable).""" + client = MagicMock() + client.aclose = AsyncMock() + return client + + +def _make_mock_mapping_service(): + """Create a MagicMock that can be used as IdMappingService (sync_environment is awaitable).""" + svc = MagicMock() + svc.sync_environment = AsyncMock() + return svc + + class TestMigrationPluginProperties: """Verify static property values.""" @@ -115,10 +129,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards)) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip_content", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -128,8 +142,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -165,10 +178,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards)) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -178,8 +191,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -215,10 +227,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards)) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -228,8 +240,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -274,7 +285,7 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, [])) with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \ @@ -299,7 +310,7 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Other")])) with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \ @@ -329,12 +340,12 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards)) mock_src_client.export_dashboard = AsyncMock(side_effect=[ (b"zip_ok", "meta"), RuntimeError("Export failed for dash 2") ]) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -344,8 +355,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -374,10 +384,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")])) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -387,8 +397,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -423,11 +432,11 @@ class TestMigrationPluginExecute: mock_task_manager.await_input = MagicMock() mock_task_manager.wait_for_input = AsyncMock() - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")])) # First export succeeds, import fails with password error mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() # First import fails with password error, retry succeeds mock_tgt_client.import_dashboard = AsyncMock(side_effect=[ RuntimeError("Must provide a password for the database 'PostgreSQL'"), @@ -442,8 +451,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ patch('src.dependencies.get_task_manager', return_value=mock_task_manager), \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -474,10 +482,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")])) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -489,8 +497,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -516,10 +523,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")])) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -547,8 +554,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal', return_value=mock_db): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -580,7 +586,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC: # Return None (falsy) for the source client - MockSC.side_effect = [None, MagicMock()] + MockSC.side_effect = [None, _make_mock_superset_client()] with pytest.raises(ValueError, match="Clients not initialized"): await plugin.execute({ @@ -602,10 +608,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards)) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -615,8 +621,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -647,7 +652,7 @@ class TestMigrationPluginExecute: mock_task_manager.wait_for_resolution = AsyncMock() # Use a single client mock that handles both get_dashboards and import_dashboard - mock_client = MagicMock() + mock_client = _make_mock_superset_client() mock_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")])) mock_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) mock_client.import_dashboard = AsyncMock() @@ -675,8 +680,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ patch('src.dependencies.get_task_manager', return_value=mock_task_manager), \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal', return_value=mock_db): mock_ctf.return_value.__enter__ = MagicMock(return_value="/tmp/test.zip") @@ -715,10 +719,10 @@ class TestMigrationPluginExecute: mock_task_manager.await_input = MagicMock() mock_task_manager.wait_for_input = AsyncMock() - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")])) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock(side_effect=[ RuntimeError("Must provide a password for the database databases/PostgreSQL.yaml"), None, @@ -732,8 +736,7 @@ class TestMigrationPluginExecute: patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ patch('src.dependencies.get_task_manager', return_value=mock_task_manager), \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \ patch('src.plugins.migration.SessionLocal'): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -763,10 +766,10 @@ class TestMigrationPluginExecute: mock_cm = MagicMock() mock_cm.get_environments.return_value = [src_env, tgt_env] - mock_src_client = MagicMock() + mock_src_client = _make_mock_superset_client() mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards)) mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) - mock_tgt_client = MagicMock() + mock_tgt_client = _make_mock_superset_client() mock_tgt_client.import_dashboard = AsyncMock() mock_engine = MagicMock() @@ -774,16 +777,13 @@ class TestMigrationPluginExecute: mock_db_session = MagicMock() mock_sync = MagicMock() - mock_sync.sync_environment = MagicMock(side_effect=RuntimeError("Sync failed")) - from src.core.mapping_service import IdMappingService - original_sync = IdMappingService.sync_environment + mock_sync.sync_environment = AsyncMock(side_effect=RuntimeError("Sync failed")) with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \ patch('src.plugins.migration.SupersetClient') as MockSC, \ patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ patch('src.plugins.migration.create_temp_file') as mock_ctf, \ - patch('src.plugins.migration.IdMappingService'), \ - patch('src.plugins.migration.get_async_job_runner'), \ + patch('src.plugins.migration.IdMappingService', return_value=mock_sync), \ patch('src.plugins.migration.SessionLocal', return_value=mock_db_session): MockSC.side_effect = [mock_src_client, mock_tgt_client] @@ -797,5 +797,57 @@ class TestMigrationPluginExecute: }) assert result["status"] == "SUCCESS" + assert mock_sync.sync_environment.called, "sync_environment should have been awaited" + + # ── Regression: post-migration sync does not deadlock (direct await, not runner.run) ── + + @pytest.mark.asyncio + async def test_execute_id_mapping_sync_completes_without_deadlock(self): + """Regression: post-migration sync completes via direct await (not runner.run).""" + plugin = MigrationPlugin() + src_env = _make_env("env-1", "Source") + tgt_env = _make_env("env-2", "Target") + dashboards = [_make_dashboard(1, "Dash A"), _make_dashboard(2, "Dash B")] + + mock_cm = MagicMock() + mock_cm.get_environments.return_value = [src_env, tgt_env] + + mock_src_client = _make_mock_superset_client() + mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards)) + mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta")) + mock_tgt_client = _make_mock_superset_client() + mock_tgt_client.import_dashboard = AsyncMock() + + mock_engine = MagicMock() + mock_engine.transform_zip.return_value = True + + mock_db_session = MagicMock() + mock_sync = MagicMock() + mock_sync.sync_environment = AsyncMock() # completes without error + + with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \ + patch('src.plugins.migration.SupersetClient') as MockSC, \ + patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \ + patch('src.plugins.migration.create_temp_file') as mock_ctf, \ + patch('src.plugins.migration.IdMappingService', return_value=mock_sync), \ + patch('src.plugins.migration.SessionLocal', return_value=mock_db_session): + + MockSC.side_effect = [mock_src_client, mock_tgt_client] + mock_ctf.return_value.__enter__ = MagicMock(return_value="/tmp/test.zip") + + result = await plugin.execute({ + "source_env_id": "env-1", + "target_env_id": "env-2", + "selected_ids": [1, 2], + }) + + # Sync must complete and all dashboards must be migrated + assert result["status"] == "SUCCESS" + assert len(result["migrated_dashboards"]) == 2 + assert result["failed_dashboards"] == [] + # Verify sync was called with correct params + mock_sync.sync_environment.assert_called_once() + call_args = mock_sync.sync_environment.call_args + assert call_args[1]["incremental"] is True # #endregion Test.MigrationPlugin diff --git a/backend/tests/plugins/translate/test_scheduler.py b/backend/tests/plugins/translate/test_scheduler.py index 6eed43dc..8b3137e4 100644 --- a/backend/tests/plugins/translate/test_scheduler.py +++ b/backend/tests/plugins/translate/test_scheduler.py @@ -6,6 +6,7 @@ # @TEST_EDGE: invalid_cron -> empty next_executions import sys +import asyncio from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) @@ -22,6 +23,13 @@ from src.plugins.translate.scheduler import ( from .conftest import JOB_ID, db_session +def _make_coro_runner(): + """Mock runner whose run() executes the coroutine synchronously via asyncio.run.""" + r = MagicMock() + r.run.side_effect = lambda coro: asyncio.run(coro) + return r + + class TestCreateSchedule: """TranslationScheduler.create_schedule.""" @@ -284,6 +292,7 @@ class TestExecuteScheduledTranslation: patch("src.plugins.translate.scheduler.seed_trace_id"), patch("src.plugins.translate.scheduler.UTC", None), # SQLite stores naive datetimes patch("src.plugins.translate.orchestrator.TranslationOrchestrator") as MockOrch, + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), ): mock_orch = MagicMock() mock_run = MagicMock() @@ -329,6 +338,7 @@ class TestExecuteScheduledTranslation: patch("src.plugins.translate.scheduler.seed_trace_id"), patch("src.plugins.translate.scheduler.UTC", None), # SQLite stores naive datetimes patch("src.plugins.translate.orchestrator.TranslationOrchestrator") as MockOrch, + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), ): mock_orch = MagicMock() mock_run = MagicMock() @@ -370,6 +380,7 @@ class TestExecuteScheduledTranslation: patch("src.plugins.translate.scheduler.seed_trace_id"), patch("src.plugins.translate.scheduler.UTC", None), # SQLite stores naive datetimes patch("src.plugins.translate.orchestrator.TranslationOrchestrator") as MockOrch, + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), ): mock_orch = MagicMock() mock_run = MagicMock() @@ -385,9 +396,9 @@ class TestExecuteScheduledTranslation: assert mock_run.trigger_type == "scheduled" def test_execution_failure_sends_notification(self, db_session): - """Execution failure triggers notification and marks run as FAILED.""" + """Execution failure is handled by execute_background — scheduler dispatches fire-and-forget.""" from src.plugins.translate.scheduler import TranslationScheduler - from src.models.translate import TranslationRun + from src.plugins.translate.orchestrator import TranslationOrchestrator scheduler = TranslationScheduler(db_session, MagicMock()) schedule = scheduler.create_schedule(JOB_ID, "0 0 * * *", is_active=True) @@ -399,6 +410,7 @@ class TestExecuteScheduledTranslation: patch("src.plugins.translate.scheduler.seed_trace_id"), patch("src.plugins.translate.scheduler.UTC", None), # SQLite stores naive datetimes patch("src.plugins.translate.orchestrator.TranslationOrchestrator") as MockOrch, + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()), patch( "src.plugins.translate.scheduler.NotificationService", ) as MockNotif, @@ -412,30 +424,28 @@ class TestExecuteScheduledTranslation: mock_run.status = "PENDING" mock_run.insert_status = None mock_orch.start_run.return_value = mock_run - mock_orch.execute_run.side_effect = RuntimeError("LLM call failed") MockOrch.return_value = mock_orch - # Mock the runner to raise the exception from execute_run + # The runner is still used for the outer notification (crash outside execution), + # but NOT for execute_run dispatch. execute_background is called instead. mock_runner = MagicMock() - mock_runner.run.side_effect = RuntimeError("LLM call failed") + mock_runner.run.side_effect = lambda coro: asyncio.run(coro) MockRunner.return_value = mock_runner - # Provider mock — send must return a real coroutine for runner.run() - async def _dummy_send(*args, **kwargs): - pass - mock_provider = MagicMock() - mock_provider.send = MagicMock(wraps=_dummy_send) + # Provider mock — notification is NOT sent from scheduler (moved to background task) mock_notif = MagicMock() - mock_notif._providers = {"email": mock_provider} + mock_notif._providers = {} MockNotif.return_value = mock_notif execute_scheduled_translation(schedule.id, JOB_ID, maker, MagicMock()) - # Notification should be sent - mock_provider.send.assert_called() - # Run should be marked FAILED - assert mock_run.status == "FAILED" - assert mock_run.error_message == "LLM call failed" + # Scheduler dispatches via execute_background instead of calling execute_run directly + MockOrch.execute_background.assert_called_once() + call_kwargs = MockOrch.execute_background.call_args.kwargs + assert call_kwargs["run_id"] == "fail-run" + assert call_kwargs["current_user"] == "scheduler" + # Notification is NOT sent from scheduler (moved to background task) + # (schedule.last_run_at tracking verified in test_scheduled_execution_uses_runner) def test_get_next_executions_break_on_none(self): """get_next_executions breaks when cron trigger returns None.""" diff --git a/backend/tests/test_translate_scheduler_audit.py b/backend/tests/test_translate_scheduler_audit.py index 789636f8..19ff9da7 100644 --- a/backend/tests/test_translate_scheduler_audit.py +++ b/backend/tests/test_translate_scheduler_audit.py @@ -42,9 +42,9 @@ class TestAsyncSyncMismatch: # #endregion test_execute_run_is_async_coroutine # #region test_scheduler_uses_runner_for_execute_run [C:2] [TYPE Function] - # @BRIEF The scheduler uses AsyncJobRunner.run() to execute translations (FIXED). + # @BRIEF The scheduler uses execute_background static method (fire-and-forget, no timeout cap). def test_scheduler_uses_runner_for_execute_run(self): - """In execute_scheduled_translation, orch.execute_run(run) is called via runner.run().""" + """In execute_scheduled_translation, TranslationOrchestrator.execute_background is called.""" import ast import inspect @@ -52,33 +52,32 @@ class TestAsyncSyncMismatch: from src.plugins.translate.scheduler import execute_scheduled_translation source = inspect.getsource(execute_scheduled_translation) - # Parse the AST to find the call to execute_run + # Parse the AST to find the call to execute_background tree = ast.parse(source) - # Find all Call nodes that call execute_run - execute_run_calls = [] + # Find all Call nodes that call execute_background + execute_bg_calls = [] for node in ast.walk(tree): if isinstance(node, ast.Call): - if isinstance(node.func, ast.Attribute) and node.func.attr == 'execute_run': - execute_run_calls.append(node) + if isinstance(node.func, ast.Attribute) and node.func.attr == 'execute_background': + execute_bg_calls.append(node) - assert len(execute_run_calls) > 0, "No calls to execute_run found in execute_scheduled_translation" + assert len(execute_bg_calls) > 0, ( + "FIX NOT APPLIED: execute_background should be called in execute_scheduled_translation. " + "Expected: TranslationOrchestrator.execute_background(run_id=..., ...)" + ) - # Check if any call is wrapped in runner.run() + # Confirm there is NO runner.run(orch.execute_run(...)) call for node in ast.walk(tree): if isinstance(node, ast.Call): if isinstance(node.func, ast.Attribute) and node.func.attr == 'run': - # Check if the argument is an execute_run call for arg in node.args: if isinstance(arg, ast.Call): if isinstance(arg.func, ast.Attribute) and arg.func.attr == 'execute_run': - # Found runner.run(orch.execute_run(run)) — this is correct - return - - pytest.fail( - "FIX NOT APPLIED: execute_run should be called via runner.run() in execute_scheduled_translation. " - "Expected: runner.run(orch.execute_run(run))" - ) + pytest.fail( + "REGRESSION: runner.run(orch.execute_run(run)) found. " + "Should use TranslationOrchestrator.execute_background() instead." + ) # #endregion test_scheduler_uses_runner_for_execute_run # #region test_manual_trigger_uses_asyncio_create_task [C:2] [TYPE Function] @@ -262,14 +261,14 @@ class TestScheduledExecutionFlow: """Verify the complete scheduled execution flow has the async/sync bug.""" # #region test_scheduled_execution_uses_runner [C:2] [TYPE Function] - # @BRIEF Full flow: scheduler triggers execute_run via AsyncJobRunner (FIXED). + # @BRIEF Full flow: scheduler dispatches via execute_background (fire-and-forget). def test_scheduled_execution_uses_runner(self): """ Simulate what happens when APScheduler triggers execute_scheduled_translation: 1. TranslationOrchestrator is created 2. start_run creates a TranslationRun - 3. execute_run is called via runner.run() - 4. The translation actually executes + 3. execute_background is called with the run_id + 4. schedule.last_run_at is updated at dispatch time """ from src.plugins.translate.scheduler import execute_scheduled_translation from src.plugins.translate.orchestrator import TranslationOrchestrator @@ -301,20 +300,27 @@ class TestScheduledExecutionFlow: mock_run.id = "run-1" mock_run.status = "PENDING" - # Mock the runner to track calls + # Track execute_background calls — runner mock must execute the dispatch coroutine mock_runner = MagicMock() + mock_runner.run.side_effect = lambda coro: asyncio.run(coro) + with patch.object(TranslationOrchestrator, 'execute_background') as mock_exec_bg, \ + patch.object(TranslationOrchestrator, 'start_run', return_value=mock_run), \ + patch('src.dependencies.get_async_job_runner', return_value=mock_runner): + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) - with patch.object(TranslationOrchestrator, 'execute_run', return_value=mock_run): - with patch.object(TranslationOrchestrator, 'start_run', return_value=mock_run): - with patch('src.dependencies.get_async_job_runner', return_value=mock_runner): - execute_scheduled_translation( - schedule_id="sched-1", - job_id="job-1", - db_session_maker=mock_session_maker, - config_manager=mock_config, - execution_mode="full", - ) - - # Verify that runner.run was called (the fix is applied) - mock_runner.run.assert_called_once() + # Verify that execute_background was called with the correct run_id (NOT execute_run via runner.run) + mock_exec_bg.assert_called_once_with( + run_id="run-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + current_user="scheduler", + ) + # Schedule tracking is updated at dispatch time + assert schedule.last_run_at is not None # #endregion test_scheduled_execution_uses_runner diff --git a/backend/tests/test_translate_scheduler_execution.py b/backend/tests/test_translate_scheduler_execution.py index fccfecf1..24c04b55 100644 --- a/backend/tests/test_translate_scheduler_execution.py +++ b/backend/tests/test_translate_scheduler_execution.py @@ -9,11 +9,19 @@ # @TEST_INVARIANT: trigger_type_dispatch -> VERIFIED_BY: [test_new_key_only_mode, test_baseline_expired_fallback, test_full_mode_default] from pathlib import Path import sys +import asyncio sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from datetime import UTC, datetime, timedelta from unittest.mock import MagicMock, patch + +def _make_coro_runner(): + """Mock runner whose run() executes the coroutine synchronously via asyncio.run.""" + r = MagicMock() + r.run.side_effect = lambda coro: asyncio.run(coro) + return r + # -- Helpers ----------------------------------------------------------------- def _make_most_recent_run(created_at_delta_days=60): @@ -79,7 +87,8 @@ def test_new_key_only_mode(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run mock_orch_cls.return_value = mock_orch @@ -96,7 +105,8 @@ def test_new_key_only_mode(): f"Expected trigger_type='new_key_only', got '{mock_run.trigger_type}'" ) mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True) - mock_orch.execute_run.assert_called_once() + # execute_background is called (fire-and-forget) instead of execute_run via runner.run + mock_orch_cls.execute_background.assert_called() mock_db.close.assert_called_once() # #endregion test_new_key_only_mode @@ -133,7 +143,8 @@ def test_baseline_expired_fallback(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run mock_orch_cls.return_value = mock_orch @@ -199,7 +210,8 @@ def test_full_mode_default(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run mock_orch_cls.return_value = mock_orch @@ -215,7 +227,7 @@ def test_full_mode_default(): assert mock_run.trigger_type == "scheduled", ( f"Expected trigger_type='scheduled' for full mode, got '{mock_run.trigger_type}'" ) - mock_orch.execute_run.assert_called_once() + mock_orch_cls.execute_background.assert_called() mock_db.close.assert_called_once() # #endregion test_full_mode_default @@ -241,7 +253,8 @@ def test_inactive_schedule_skips(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): execute_scheduled_translation( schedule_id="sched-1", job_id="job-1", @@ -271,7 +284,8 @@ def test_schedule_not_found_skips(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): execute_scheduled_translation( schedule_id="sched-1", job_id="job-1", @@ -317,7 +331,8 @@ def test_baseline_expired_full_mode(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run mock_orch_cls.return_value = mock_orch @@ -346,7 +361,7 @@ def test_baseline_expired_full_mode(): # #region test_execution_error_handled [C:1] [TYPE Function] def test_execution_error_handled(): - """execute_run raises → run is marked FAILED, schedule tracking updated, no crash.""" + """execute_background is called even when execute_run would raise — scheduler dispatches safely.""" from src.plugins.translate.scheduler import execute_scheduled_translation mock_db = MagicMock() @@ -374,10 +389,10 @@ def test_execution_error_handled(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run - mock_orch.execute_run.side_effect = RuntimeError("LLM timeout") mock_orch_cls.return_value = mock_orch execute_scheduled_translation( @@ -388,18 +403,22 @@ def test_execution_error_handled(): execution_mode="full", ) - assert mock_run.status == "FAILED" - assert "LLM timeout" in mock_run.error_message - assert mock_run.completed_at is not None + # Scheduler dispatches via execute_background — does NOT catch execution errors itself + mock_orch_cls.execute_background.assert_called_once_with( + run_id=mock_run.id, + db_session_maker=mock_session_maker, + config_manager=mock_config, + current_user="scheduler", + ) + # Schedule tracking is updated at dispatch time assert schedule.last_run_at is not None - mock_db.commit.assert_called() mock_db.close.assert_called_once() # #endregion test_execution_error_handled # #region test_execution_run_status_failed_path [C:1] [TYPE Function] def test_execution_run_status_failed_path(): - """execute_run succeeds but run.status == 'FAILED' → hits lines 388-392.""" + """Scheduler dispatches regardless of run.status — status is handled by background task.""" from src.plugins.translate.scheduler import execute_scheduled_translation mock_db = MagicMock() @@ -423,11 +442,12 @@ def test_execution_run_status_failed_path(): mock_db.query.side_effect = [q1, q2, q3] mock_run = _make_mock_run(run_id="run-fail", job_id="job-fail") - mock_run.status = "FAILED" # execute_run sets this + mock_run.status = "FAILED" # simulate run already failed with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run mock_orch_cls.return_value = mock_orch @@ -440,16 +460,15 @@ def test_execution_run_status_failed_path(): execution_mode="full", ) - # The run.status == "FAILED" path sets insert_status = run.insert_status or None - assert mock_run.insert_status is None # None or None -> None - mock_orch.execute_run.assert_called_once() + # Scheduler dispatches background task regardless of run.status + mock_orch_cls.execute_background.assert_called() mock_db.close.assert_called_once() # #endregion test_execution_run_status_failed_path # #region test_execution_notification_error [C:1] [TYPE Function] def test_execution_notification_error(): - """execute_run raises and notification send also raises → both caught (lines 413-415, 431-442).""" + """Scheduler dispatches safely — notification moves to background task.""" from src.plugins.translate.scheduler import execute_scheduled_translation mock_db = MagicMock() @@ -474,21 +493,12 @@ def test_execution_notification_error(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls, patch( - "src.plugins.translate.scheduler.NotificationService" - ) as mock_notif_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run - mock_orch.execute_run.side_effect = RuntimeError("LLM timeout") mock_orch_cls.return_value = mock_orch - # Make notification send also raise - mock_provider = MagicMock() - mock_provider.send.side_effect = Exception("Email down") - mock_notif_svc = MagicMock() - mock_notif_svc._providers = {"email": mock_provider} - mock_notif_cls.return_value = mock_notif_svc - execute_scheduled_translation( schedule_id="sched-notif", job_id="job-notif", @@ -497,10 +507,10 @@ def test_execution_notification_error(): execution_mode="full", ) - assert mock_run.status == "FAILED" - assert mock_run.completed_at is not None + # Scheduler dispatches — notification is handled by execute_background + mock_orch_cls.execute_background.assert_called() + # Schedule tracking is updated at dispatch assert schedule.last_run_at is not None - mock_db.commit.assert_called() mock_db.close.assert_called_once() # #endregion test_execution_notification_error diff --git a/backend/tests/test_translate_scheduler_guard.py b/backend/tests/test_translate_scheduler_guard.py index c0c76a4d..efd04017 100644 --- a/backend/tests/test_translate_scheduler_guard.py +++ b/backend/tests/test_translate_scheduler_guard.py @@ -10,12 +10,20 @@ from pathlib import Path import sys +import asyncio sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from datetime import UTC, datetime, timedelta from unittest.mock import MagicMock, patch + +def _make_coro_runner(): + """Mock runner whose run() executes the coroutine synchronously via asyncio.run.""" + r = MagicMock() + r.run.side_effect = lambda coro: asyncio.run(coro) + return r + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -201,7 +209,8 @@ def test_stale_pending_cleared_and_proceeds(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run mock_orch_cls.return_value = mock_orch @@ -219,7 +228,7 @@ def test_stale_pending_cleared_and_proceeds(): 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() + mock_orch_cls.execute_background.assert_called() assert mock_db.commit.call_count >= 1 mock_db.close.assert_called_once() # #endregion test_stale_pending_cleared_and_proceeds @@ -272,7 +281,8 @@ def test_multiple_stale_pending_all_cleared(): with patch( "src.plugins.translate.orchestrator.TranslationOrchestrator" - ) as mock_orch_cls: + ) as mock_orch_cls, \ + patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()): mock_orch = MagicMock() mock_orch.start_run.return_value = mock_run mock_orch_cls.return_value = mock_orch @@ -291,7 +301,7 @@ def test_multiple_stale_pending_all_cleared(): 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_orch_cls.execute_background.assert_called() mock_db.close.assert_called_once() # #endregion test_multiple_stale_pending_all_cleared diff --git a/frontend/src/lib/models/Migration.WizardModel.svelte.ts b/frontend/src/lib/models/Migration.WizardModel.svelte.ts index c19710c4..1622e5cb 100644 --- a/frontend/src/lib/models/Migration.WizardModel.svelte.ts +++ b/frontend/src/lib/models/Migration.WizardModel.svelte.ts @@ -5,7 +5,7 @@ // @RELATION DEPENDS_ON -> [Migration.Model] // @INVARIANT Backward navigation (step <= currentStep) is always allowed. // @INVARIANT Step 2 requires source !== target and both envs selected (delegates to parent.stepReady). -// @INVARIANT Step 3 requires step 1 and step 2 ready. +// @INVARIANT Step 3 requires completed dry-run results. // @INVARIANT Step 4 requires a dry-run result. // @STATE currentStep — Active wizard step (1=environments, 2=dashboards, 3=review, 4=execute). // @ACTION goToStep(step) — Navigates to a wizard step with readiness gating. @@ -27,7 +27,7 @@ export class WizardState { if ( step <= this.currentStep || (step === 2 && this.parent.stepReady[1]) || - (step === 3 && this.parent.stepReady[1] && this.parent.stepReady[2]) || + (step === 3 && this.parent.stepReady[3]) || (step === 4 && dryRunResult) ) { this.currentStep = step; diff --git a/frontend/src/lib/models/MigrationModel.svelte.ts b/frontend/src/lib/models/MigrationModel.svelte.ts index 5fc7541d..e4214c88 100644 --- a/frontend/src/lib/models/MigrationModel.svelte.ts +++ b/frontend/src/lib/models/MigrationModel.svelte.ts @@ -161,7 +161,7 @@ export class MigrationModel { stepReady = $derived>({ 1: this.sourceEnvId !== "" && this.targetEnvId !== "" && this.sourceEnvId !== this.targetEnvId, 2: this.selectedDashboardIds.length > 0, - 3: true, // review is always ready if we got here + 3: this.dryRunResult != null, }); /** Human-readable screen state derived from atoms. */ diff --git a/frontend/src/lib/models/__tests__/MigrationModel.test.ts b/frontend/src/lib/models/__tests__/MigrationModel.test.ts index 4712d1ea..5b8eaf9d 100644 --- a/frontend/src/lib/models/__tests__/MigrationModel.test.ts +++ b/frontend/src/lib/models/__tests__/MigrationModel.test.ts @@ -110,7 +110,11 @@ describe("MigrationModel — L1 invariants (no render)", () => { it("step 1 ready when both envs selected and distinct", () => { model.sourceEnvId = "env-1"; model.targetEnvId = "env-2"; expect(model.stepReady[1]).toBe(true); }); it("step 1 not ready when same env", () => { model.sourceEnvId = "env-1"; model.targetEnvId = "env-1"; expect(model.stepReady[1]).toBe(false); }); it("step 2 ready when dashboards selected", () => { model.selectedDashboardIds = ["1"]; expect(model.stepReady[2]).toBe(true); }); - it("step 3 always ready", () => { expect(model.stepReady[3]).toBe(true); }); + it("step 3 requires dry-run result", () => { + expect(model.stepReady[3]).toBe(false); + model.dryRunResult = { summary: {}, risk: { score: 10, level: "low", items: [] } }; + expect(model.stepReady[3]).toBe(true); + }); }); describe("setReplaceDb", () => { @@ -130,6 +134,10 @@ describe("MigrationModel — L1 invariants (no render)", () => { it("allows backward navigation", () => { model.currentStep = 3; model.goToStep(1); expect(model.currentStep).toBe(1); }); it("allows forward to step 2 when ready", () => { model.sourceEnvId = "env-1"; model.targetEnvId = "env-2"; model.goToStep(2); expect(model.currentStep).toBe(2); }); it("blocks forward to step 2 when not ready", () => { model.goToStep(2); expect(model.currentStep).toBe(1); }); + it("blocks forward to step 3 without dryRunResult", () => { + model.sourceEnvId = "env-1"; model.targetEnvId = "env-2"; model.selectedDashboardIds = ["1"]; + model.goToStep(3); expect(model.currentStep).toBe(1); + }); it("allows step 4 only with dryRunResult", () => { model.currentStep = 3; model.dryRunResult = {} as any; model.selectedDashboardIds = ["1"]; model.sourceEnvId = "env-1"; model.targetEnvId = "env-2"; model.goToStep(4); expect(model.currentStep).toBe(4); diff --git a/frontend/src/routes/migration/+page.svelte b/frontend/src/routes/migration/+page.svelte index 4a2de489..9bad6c88 100644 --- a/frontend/src/routes/migration/+page.svelte +++ b/frontend/src/routes/migration/+page.svelte @@ -71,9 +71,11 @@ // The model self-enforces the invariant via toggleDashboard(), but the bind: path is outside model control. // This guard ensures dryRunResult is cleared regardless of the mutation path. // @INVARIANT: Dashboard selection change clears stale dry-run result + let selectedDashboardIdsKey = $state(""); $effect(() => { - model.selectedDashboardIds; // always track - if (model.dryRunResult) { + const nextSelectedDashboardIdsKey = model.selectedDashboardIds.join("\u0000"); + if (nextSelectedDashboardIdsKey !== selectedDashboardIdsKey) { + selectedDashboardIdsKey = nextSelectedDashboardIdsKey; model.dryRunResult = null; } }); @@ -129,7 +131,7 @@