diff --git a/backend/src/api/routes/migration.py b/backend/src/api/routes/migration.py index 0bf36175..62ecb8f8 100644 --- a/backend/src/api/routes/migration.py +++ b/backend/src/api/routes/migration.py @@ -32,7 +32,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_config_manager, get_task_manager, has_permission +from ...dependencies import get_async_job_runner, get_config_manager, get_task_manager, has_permission from ...models.dashboard import DashboardMetadata, DashboardSelection from ...models.mapping import ResourceMapping @@ -377,12 +377,13 @@ 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) - service.sync_environment(env.id, client) + runner.run(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/api/routes/translate/_schedule_routes.py b/backend/src/api/routes/translate/_schedule_routes.py index 51991bea..c8dbd919 100644 --- a/backend/src/api/routes/translate/_schedule_routes.py +++ b/backend/src/api/routes/translate/_schedule_routes.py @@ -172,10 +172,10 @@ async def disable_schedule( logger.reason(f"disable_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"}) try: scheduler = TranslationScheduler(db, config_manager, current_user.username) - scheduler.set_schedule_active(job_id, is_active=False) + schedule = scheduler.set_schedule_active(job_id, is_active=False) from ....dependencies import get_scheduler_service sched_svc = get_scheduler_service() - sched_svc.remove_translation_job(schedule_id=job_id) + sched_svc.remove_translation_job(schedule_id=schedule.id) return {"status": "disabled", "job_id": job_id} except ValueError as e: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) @@ -199,10 +199,12 @@ async def delete_schedule( logger.reason(f"delete_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"}) try: scheduler = TranslationScheduler(db, config_manager, current_user.username) + schedule = scheduler.get_schedule(job_id) + schedule_id = schedule.id scheduler.delete_schedule(job_id) from ....dependencies import get_scheduler_service sched_svc = get_scheduler_service() - sched_svc.remove_translation_job(schedule_id=job_id) + sched_svc.remove_translation_job(schedule_id=schedule_id) return None except ValueError as e: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) diff --git a/backend/src/app.py b/backend/src/app.py index 40ed7e83..b10dc8d4 100755 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -61,7 +61,7 @@ from .core.database import AuthSessionLocal, init_db from .core.encryption_key import ensure_encryption_key from .core.logger import belief_scope, logger from .core.utils.network import NetworkError -from .dependencies import get_scheduler_service, get_task_manager +from .dependencies import get_async_job_runner, get_scheduler_service, get_task_manager from .models.auth import Role, User @@ -108,6 +108,8 @@ async def lifespan(app: FastAPI): _s.close() except Exception as _e: logger.warning(f"Failed to clean up stuck validation runs: {_e}") + logger.info("⏰ Initializing AsyncJobRunner...") + get_async_job_runner() # Initialize singleton with running event loop BEFORE scheduler starts logger.info("⏰ Starting scheduler...") scheduler = get_scheduler_service() scheduler.start() diff --git a/backend/src/core/async_job_runner.py b/backend/src/core/async_job_runner.py new file mode 100644 index 00000000..ad1211ec --- /dev/null +++ b/backend/src/core/async_job_runner.py @@ -0,0 +1,93 @@ +# #region Core.AsyncJobRunner [C:4] [TYPE Module] [SEMANTICS async,scheduler,bridge,event-loop] +# @defgroup Core Async/sync bridge for scheduled jobs. +# @BRIEF Centralized bridge between APScheduler (sync thread pool) and async coroutines. +# @LAYER Core +# @PRE Event loop is running (FastAPI lifespan). +# @POST All scheduled async jobs execute through this runner. +# @SIDE_EFFECT Submits coroutines to the event loop from sync threads. +# @RATIONALE APScheduler BackgroundScheduler runs callbacks in a sync thread pool. +# Async coroutines must be dispatched to the FastAPI event loop via +# asyncio.run_coroutine_threadsafe(). This runner centralizes that pattern. +# @REJECTED asyncio.run() — creates a new event loop each call; breaks when called +# from a context with an existing loop; incompatible with AsyncIOScheduler migration. +# @REJECTED Direct async def callbacks in BackgroundScheduler — coroutine is created +# but never awaited, causing silent no-ops (the root cause of scheduled translation failures). + +import asyncio +from typing import Any + + +# #region AsyncJobRunner [C:4] [TYPE Class] [SEMANTICS async,runner,scheduler] +# @ingroup Core +# @BRIEF Centralized async/sync bridge for dispatching coroutines from sync threads. +# @PRE Event loop is captured at construction time. +# @POST run() executes coroutines synchronously (blocking until complete). +# @POST run_later() schedules delayed coroutine execution. +# @SIDE_EFFECT Submits coroutines to the captured event loop. +# @DATA_CONTRACT Input[coroutine] -> Output[result] +class AsyncJobRunner: + """Bridge between APScheduler sync threads and the FastAPI async event loop. + + Usage: + runner = AsyncJobRunner(asyncio.get_running_loop()) + result = runner.run(some_async_function()) + """ + + # #region __init__ [TYPE Function] + # @BRIEF Capture the event loop for later coroutine dispatch. + # @PRE loop is a running asyncio event loop. + # @POST self.loop is set and ready for run()/run_later() calls. + def __init__(self, loop: asyncio.AbstractEventLoop) -> None: + self.loop = loop + # Track pending tasks from run_later() to prevent silent exception loss + self._pending_tasks: set[asyncio.Task] = set() + # #endregion __init__ + + # #region run [TYPE Function] + # @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. + def run(self, coro: Any) -> Any: + """Dispatch a coroutine to the event loop and block until it completes. + + Args: + coro: An awaitable (coroutine, Future, or Task). + + Returns: + The result of the coroutine. + + Raises: + Exception: Any exception raised by the coroutine is re-raised. + """ + future = asyncio.run_coroutine_threadsafe(coro, self.loop) + return future.result() # timeout=None — wait indefinitely + # #endregion run + + # #region run_later [TYPE Function] + # @BRIEF Schedule a coroutine to execute after a delay (for windowed validation). + # @PRE coro is an awaitable; delay_seconds >= 0. + # @POST Coroutine is scheduled on the event loop's call_later queue. + # @SIDE_EFFECT Non-blocking — returns immediately; coroutine executes later. + def run_later(self, coro: Any, delay_seconds: float) -> None: + """Schedule a coroutine to execute after a delay. + + Used for windowed validation: tasks are distributed across a time window + using call_later() instead of running them all at once. + + Args: + coro: An awaitable to execute after the delay. + delay_seconds: Seconds to wait before scheduling the coroutine. + """ + def _schedule() -> None: + task = asyncio.ensure_future(coro, loop=self.loop) + # Store task for exception tracking; remove when done + self._pending_tasks.add(task) + task.add_done_callback(self._pending_tasks.discard) + # Use call_soon_threadsafe to schedule from any thread + self.loop.call_soon_threadsafe( + lambda: self.loop.call_later(delay_seconds, _schedule) + ) + # #endregion run_later +# #endregion AsyncJobRunner +# #endregion Core.AsyncJobRunner diff --git a/backend/src/core/mapping_service.py b/backend/src/core/mapping_service.py index 45412bd9..29b13a84 100644 --- a/backend/src/core/mapping_service.py +++ b/backend/src/core/mapping_service.py @@ -1,21 +1,21 @@ -# #region IdMappingServiceModule [C:5] [TYPE Module] [SEMANTICS sqlalchemy, mapping, sync, schedule, superset, resource] +# #region IdMappingServiceModule [C:5] [TYPE Module] [SEMANTICS sqlalchemy, mapping, sync, superset, resource] # @defgroup Core Module group. # # @BRIEF Service for tracking and synchronizing Superset Resource IDs (UUID <-> Integer ID) # @LAYER Core # @RELATION DEPENDS_ON -> [MappingModels] # @RELATION DEPENDS_ON -> [LoggerModule] +# @RELATION DEPENDS_ON -> [AsyncJobRunner] # @PRE Database session is valid and Superset client factory returns authenticated clients for requested environments. # @POST Mapping synchronization and lookup APIs are available for environment-scoped UUID-to-integer resolution. -# @SIDE_EFFECT Reads/writes ResourceMapping rows, emits logs, and schedules periodic sync jobs. +# @SIDE_EFFECT Reads/writes ResourceMapping rows, emits logs. # @DATA_CONTRACT Input[environment_id, resource_type, uuid] -> Output[remote_integer_id|None] # @TEST_DATA: mock_superset_resources -> {'chart': [{'id': 42, 'uuid': '1234', 'slice_name': 'test'}], 'dataset': [{'id': 99, 'uuid': '5678', 'table_name': 'data'}]} # # @INVARIANT sync_environment must handle remote API failures gracefully. +# @REJECTED BackgroundScheduler — was never started; replaced by AsyncJobRunner for async/sync bridge. from datetime import UTC, datetime -from apscheduler.schedulers.background import BackgroundScheduler -from apscheduler.triggers.cron import CronTrigger from sqlalchemy.orm import Session from src.core.cot_logger import seed_trace_id @@ -27,11 +27,12 @@ from src.models.mapping import ResourceMapping, ResourceType # @defgroup Core Module group. # @BRIEF Service handling the cataloging and retrieval of remote Superset Integer IDs. # @PRE db_session is an active SQLAlchemy Session bound to mapping tables. -# @POST Service instance provides scheduler control and environment-scoped mapping synchronization APIs. +# @POST Service instance provides environment-scoped mapping synchronization APIs. # @RELATION DEPENDS_ON -> [MappingModels] # @RELATION DEPENDS_ON -> [LoggerModule] +# @RELATION DEPENDS_ON -> [AsyncJobRunner] # @INVARIANT self.db remains the authoritative session for all mapping operations. -# @SIDE_EFFECT Instantiates an in-process scheduler and performs database writes during sync cycles. +# @SIDE_EFFECT Performs database writes during sync cycles. # @DATA_CONTRACT Input[db_session] -> Output[IdMappingService] # # @TEST_CONTRACT IdMappingServiceModel -> @@ -53,47 +54,7 @@ class IdMappingService: # @PURPOSE: Initializes the mapping service. def __init__(self, db_session: Session): self.db = db_session - self.scheduler = BackgroundScheduler() - self._sync_job = None # #endregion __init__ - # #region start_scheduler [TYPE Function] -# @ingroup Core - # @PURPOSE: Starts the background scheduler with a given cron string. - # @PARAM cron_string (str) - Cron expression for the sync interval. - # @PARAM environments (List[str]) - List of environment IDs to sync. - # @PARAM superset_client_factory - Function to get a client for an environment. - def start_scheduler( - self, cron_string: str, environments: list[str], superset_client_factory - ): - with belief_scope("IdMappingService.start_scheduler"): - if self._sync_job: - self.scheduler.remove_job(self._sync_job.id) - logger.info( - "[IdMappingService.start_scheduler][Reflect] Removed existing sync job." - ) - def sync_all(): - seed_trace_id() - for env_id in environments: - client = superset_client_factory(env_id) - if client: - import asyncio - asyncio.run(self.sync_environment(env_id, client)) - self._sync_job = self.scheduler.add_job( - sync_all, - CronTrigger.from_crontab(cron_string), - id="id_mapping_sync_job", - replace_existing=True, - ) - if not self.scheduler.running: - self.scheduler.start() - logger.info( - f"[IdMappingService.start_scheduler][Coherence:OK] Started background scheduler with cron: {cron_string}" - ) - else: - logger.info( - f"[IdMappingService.start_scheduler][Coherence:OK] Updated background scheduler with cron: {cron_string}" - ) - # #endregion start_scheduler # #region sync_environment [TYPE Function] # @ingroup Core # @PURPOSE: Fully synchronizes mapping for a specific environment. diff --git a/backend/src/core/scheduler.py b/backend/src/core/scheduler.py index 0c96498a..64e5bdf6 100644 --- a/backend/src/core/scheduler.py +++ b/backend/src/core/scheduler.py @@ -3,12 +3,14 @@ # @BRIEF Manages scheduled tasks using APScheduler. # @LAYER Core # @RELATION DEPENDS_ON -> TaskManager +# @RELATION DEPENDS_ON -> [AsyncJobRunner] import asyncio from datetime import date, datetime, time, timedelta from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger +from .async_job_runner import AsyncJobRunner from .config_manager import ConfigManager from .cot_logger import seed_trace_id from .database import SessionLocal @@ -29,7 +31,11 @@ class SchedulerService: self.task_manager = task_manager self.config_manager = config_manager self.scheduler = BackgroundScheduler() - self.loop = asyncio.get_event_loop() + try: + self.loop = asyncio.get_running_loop() + except RuntimeError: + self.loop = asyncio.new_event_loop() + self.runner = AsyncJobRunner(self.loop) # #endregion __init__ # #region start [TYPE Function] # @ingroup Core @@ -40,7 +46,7 @@ class SchedulerService: with belief_scope("SchedulerService.start"): if not self.scheduler.running: self.scheduler.start() - logger.info("Scheduler started.") + logger.reason("Scheduler started") self.load_schedules() # #endregion start # #region stop [TYPE Function] @@ -52,7 +58,7 @@ class SchedulerService: with belief_scope("SchedulerService.stop"): if self.scheduler.running: self.scheduler.shutdown() - logger.info("Scheduler stopped.") + logger.reflect("Scheduler stopped") # #endregion stop # #region load_schedules [C:4] [TYPE Function] [SEMANTICS scheduler,backup,translation,apscheduler] # @ingroup Core @@ -218,7 +224,7 @@ class SchedulerService: def _trigger_backup(self, env_id: str): seed_trace_id() with belief_scope("SchedulerService._trigger_backup", f"env_id={env_id}"): - logger.info(f"Triggering scheduled backup for environment {env_id}") + logger.reason(f"Triggering scheduled backup for environment {env_id}") # Check if a backup is already running for this environment active_tasks = self.task_manager.get_tasks(limit=100) for task in active_tasks: @@ -227,17 +233,14 @@ class SchedulerService: and task.status in ["PENDING", "RUNNING"] and task.params.get("environment_id") == env_id ): - logger.warning( - f"Backup already running for environment {env_id}. Skipping scheduled run." - ) + logger.explore(f"Backup already running for environment {env_id}, skipping scheduled run", + error="Concurrent backup in progress") return - # Run the backup task - # We need to run this in the event loop since create_task is async - asyncio.run_coroutine_threadsafe( + # Run the backup task via AsyncJobRunner + self.runner.run( self.task_manager.create_task( "superset-backup", {"environment_id": env_id} - ), - self.loop, + ) ) # #endregion _trigger_backup # #region add_validation_job [C:3] [TYPE Function] [SEMANTICS scheduler,validation,cron,apscheduler] @@ -365,20 +368,25 @@ class SchedulerService: current_date=today, ) + now = datetime.now() for idx, dash_id in enumerate(dashboard_ids): sched_time = scheduled_times[idx] if idx < len(scheduled_times) else scheduled_times[-1] + delay = (sched_time - now).total_seconds() params: dict = { "dashboard_id": dash_id, "environment_id": policy.environment_id, "provider_id": policy.provider_id or "", "policy_id": policy_id, } - asyncio.run_coroutine_threadsafe( - self.task_manager.create_task( - plugin_id="llm_dashboard_validation", params=params - ), - self.loop, + coro = self.task_manager.create_task( + plugin_id="llm_dashboard_validation", params=params ) + if delay <= 0: + # Time already passed — run immediately + self.runner.run(coro) + else: + # Delayed execution within the window + self.runner.run_later(coro, delay_seconds=delay) logger.reason( f"Scheduled validation for dashboard {dash_id} at {sched_time.isoformat()}" ) diff --git a/backend/src/dependencies.py b/backend/src/dependencies.py index 6dca8db6..f4184050 100755 --- a/backend/src/dependencies.py +++ b/backend/src/dependencies.py @@ -23,6 +23,7 @@ from fastapi import Depends, HTTPException, Request, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError +from .core.async_job_runner import AsyncJobRunner from .core.auth.jwt import decode_token, is_token_blacklisted from .core.auth.repository import AuthRepository from .core.config_manager import ConfigManager @@ -58,6 +59,7 @@ config_manager: ConfigManager | None = None plugin_loader: PluginLoader | None = None task_manager: TaskManager | None = None scheduler_service: SchedulerService | None = None +async_job_runner: AsyncJobRunner | None = None resource_service: ResourceService | None = None @@ -155,6 +157,28 @@ def get_scheduler_service() -> SchedulerService: # #endregion get_scheduler_service +# #region get_async_job_runner [C:1] [TYPE Function] +# @BRIEF Dependency injector for AsyncJobRunner. +# @PRE Event loop is running (FastAPI lifespan). +# @POST Returns shared AsyncJobRunner instance. +def get_async_job_runner() -> AsyncJobRunner: + """Dependency injector for AsyncJobRunner.""" + import asyncio + global async_job_runner + if async_job_runner is None: + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + async_job_runner = AsyncJobRunner(loop) + logger.info("AsyncJobRunner initialized") + return async_job_runner + + +# #endregion get_async_job_runner + + # #region get_resource_service [C:1] [TYPE Function] # @BRIEF Dependency injector for ResourceService. # @PRE Global resource_service must be initialized. diff --git a/backend/src/plugins/llm_analysis/scheduler.py b/backend/src/plugins/llm_analysis/scheduler.py deleted file mode 100644 index 115257b7..00000000 --- a/backend/src/plugins/llm_analysis/scheduler.py +++ /dev/null @@ -1,59 +0,0 @@ -# #region LLMAnalysisScheduler [C:3] [TYPE Module] [SEMANTICS scheduler, llm, validation, task, cron] -# @defgroup LLMAnalysis Module group. -# @BRIEF Provides helper functions to schedule LLM-based validation tasks. -# @LAYER Domain -# @RELATION DEPENDS_ON -> [SchedulerService] - -from typing import Any - -from ...core.logger import belief_scope, logger -from ...dependencies import get_scheduler_service, get_task_manager - - -# #region schedule_dashboard_validation [TYPE Function] -# @ingroup LLMAnalysis -# @BRIEF Schedules a recurring dashboard validation task. -# @SIDE_EFFECT Adds a job to the scheduler service. -def schedule_dashboard_validation(dashboard_id: str, cron_expression: str, params: dict[str, Any]): - with belief_scope("schedule_dashboard_validation", f"dashboard_id={dashboard_id}"): - scheduler = get_scheduler_service() - task_manager = get_task_manager() - - job_id = f"llm_val_{dashboard_id}" - - async def job_func(): - await task_manager.create_task( - plugin_id="llm_dashboard_validation", - params={ - "dashboard_id": dashboard_id, - **params - } - ) - - scheduler.add_job( - job_func, - "cron", - id=job_id, - replace_existing=True, - **_parse_cron(cron_expression) - ) - logger.info(f"Scheduled validation for dashboard {dashboard_id} with cron {cron_expression}") -# #endregion schedule_dashboard_validation - -# #region _parse_cron [TYPE Function] -# @BRIEF Basic cron parser placeholder. -def _parse_cron(cron: str) -> dict[str, str]: - # Basic cron parser placeholder - parts = cron.split() - if len(parts) != 5: - return {} - return { - "minute": parts[0], - "hour": parts[1], - "day": parts[2], - "month": parts[3], - "day_of_week": parts[4] - } -# #endregion _parse_cron - -# #endregion LLMAnalysisScheduler diff --git a/backend/src/plugins/migration.py b/backend/src/plugins/migration.py index b6e2a773..d5ed084f 100755 --- a/backend/src/plugins/migration.py +++ b/backend/src/plugins/migration.py @@ -7,6 +7,7 @@ # @RELATION DEPENDS_ON -> MigrationEngine # @RELATION DEPENDS_ON -> IdMappingService # @RELATION DEPENDS_ON -> [TaskContext] +# @RELATION DEPENDS_ON -> [AsyncJobRunner] # @PRE Plugin loader can resolve infrastructure dependencies and execution requests provide validated migration context. # @POST Plugin metadata remains stable and migration execution preserves mapped-environment import guarantees. # @SIDE_EFFECT Reads config, opens database sessions, creates temporary artifacts, and triggers Superset export/import workflows. @@ -24,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_config_manager +from ..dependencies import get_async_job_runner, get_config_manager from ..models.mapping import DatabaseMapping, Environment @@ -224,7 +225,7 @@ class MigrationPlugin(PluginBase): if not from_c or not to_c: raise ValueError(f"Clients not initialized for environments: {from_env_name}, {to_env_name}") - _, all_dashboards = from_c.get_dashboards() + _, all_dashboards = await from_c.get_dashboards() # Selection Logic if selected_ids: @@ -277,7 +278,7 @@ class MigrationPlugin(PluginBase): app_logger.reason(f"Starting pipeline for dashboard '{title}'", extra={"dash_id": dash_id}) try: - exported_content, _ = from_c.export_dashboard(dash_id) + 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: success = engine.transform_zip( @@ -318,7 +319,7 @@ class MigrationPlugin(PluginBase): if success: app_logger.reason("Pushing transformed ZIP to target Superset") - to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug) + await to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug) migration_result["migrated_dashboards"].append({"id": dash_id, "title": title}) app_logger.reflect("Import successful", extra={"title": title}) else: @@ -355,7 +356,7 @@ class MigrationPlugin(PluginBase): if passwords: app_logger.reason(f"Retrying import for {title} with injected credentials") - to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug, passwords=passwords) + await to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug, passwords=passwords) migration_result["migrated_dashboards"].append({"id": dash_id, "title": title}) app_logger.reflect("Password injection unblocked import") if "passwords" in task.params: @@ -373,7 +374,8 @@ class MigrationPlugin(PluginBase): app_logger.reason("Executing incremental ID catalog sync on target") db_session = SessionLocal() mapping_service = IdMappingService(db_session) - mapping_service.sync_environment(tgt_env.id, to_c, incremental=True) + 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") except Exception as sync_exc: diff --git a/backend/src/plugins/translate/scheduler.py b/backend/src/plugins/translate/scheduler.py index 63da6925..138a5585 100644 --- a/backend/src/plugins/translate/scheduler.py +++ b/backend/src/plugins/translate/scheduler.py @@ -6,6 +6,7 @@ # @RELATION DEPENDS_ON -> [SchedulerService] # @RELATION DEPENDS_ON -> [TranslationOrchestrator] # @RELATION DEPENDS_ON -> [TranslationEventLog] +# @RELATION DEPENDS_ON -> [AsyncJobRunner] # @PRE Database session and SchedulerService are available. # @POST TranslationSchedule CRUD persisted; APScheduler jobs registered/updated/removed. # @SIDE_EFFECT Registers APScheduler jobs; runs translations on trigger; creates events. @@ -259,6 +260,8 @@ def execute_scheduled_translation( execution_mode: str = "full", ) -> None: """APScheduler job callback for scheduled translations.""" + from ...dependencies import get_async_job_runner + runner = get_async_job_runner() seed_trace_id() db: Session = db_session_maker() try: @@ -381,9 +384,9 @@ def execute_scheduled_translation( payload={"reason": "baseline_expired", "age_days": age.days}, ) - # Execute in same thread (APScheduler runs in background thread pool) + # Execute via AsyncJobRunner (dispatches coroutine to event loop) try: - orch.execute_run(run) + runner.run(orch.execute_run(run)) if run.status == "FAILED": logger.explore("Scheduled translation completed with all records failed", { "run_id": run.id, @@ -408,9 +411,8 @@ def execute_scheduled_translation( f"Error: {exec_err}\n" f"Time: {datetime.now(UTC).isoformat()}" ) - import asyncio for provider in notify_service._providers.values(): - asyncio.run(provider.send(recipient="admin", subject=subject, body=body)) + 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}") @@ -429,15 +431,15 @@ def execute_scheduled_translation( "status": run.status, }) except Exception as e: - logger.error(f"[scheduled_translation] Unexpected error: {e}") + logger.explore(f"[scheduled_translation] Unexpected error: {e}", + error="Scheduled translation crashed outside of main execution block") try: notify_service = NotificationService(db, config_manager) notify_service._initialize_providers() subject = f"Scheduled translation CRASHED: job={job_id}" body = f"Scheduled translation for job '{job_id}' crashed unexpectedly.\nError: {e}\nTime: {datetime.now(UTC).isoformat()}" - import asyncio for provider in notify_service._providers.values(): - asyncio.run(provider.send(recipient="admin", subject=subject, body=body)) + runner.run(provider.send(recipient="admin", subject=subject, body=body)) except Exception: pass finally: diff --git a/backend/tests/core/test_async_job_runner.py b/backend/tests/core/test_async_job_runner.py new file mode 100644 index 00000000..9fefcd8e --- /dev/null +++ b/backend/tests/core/test_async_job_runner.py @@ -0,0 +1,256 @@ +# #region Test.AsyncJobRunner [C:3] [TYPE Module] [SEMANTICS test,async,runner,scheduler] +# @BRIEF Unit tests for AsyncJobRunner — centralized async/sync bridge. +# @RELATION BINDS_TO -> [AsyncJobRunner] +# @TEST_CONTRACT: AsyncJobRunner.run -> result | executes coroutine and returns result +# @TEST_CONTRACT: AsyncJobRunner.run_later -> None | schedules delayed coroutine execution +# @TEST_EDGE: run_propagates_exception -> exception from coroutine is re-raised +# @TEST_EDGE: run_later_executes_after_delay -> coroutine runs after specified delay +# @TEST_EDGE: run_from_background_thread -> works correctly from non-main thread +# @TEST_INVARIANT: coroutine_must_be_awaited -> VERIFIED_BY: [test_run_executes_coroutine, test_run_propagates_exception] + +import asyncio +import sys +import threading +import time +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) + +from src.core.async_job_runner import AsyncJobRunner + + +# -- Helpers ----------------------------------------------------------------- + +async def async_return_value(value): + """Simple async function that returns a value.""" + return value + + +async def async_raise(exc_class, message=""): + """Simple async function that raises an exception.""" + raise exc_class(message) + + +async def async_delay(value, delay_seconds): + """Async function that sleeps then returns a value.""" + await asyncio.sleep(delay_seconds) + return value + + +# -- Fixtures ---------------------------------------------------------------- + +@pytest.fixture +def event_loop(): + """Create a new event loop for each test.""" + loop = asyncio.new_event_loop() + yield loop + loop.close() + + +@pytest.fixture +def runner(event_loop): + """Create an AsyncJobRunner with a new event loop.""" + return AsyncJobRunner(event_loop) + + +# -- Tests: run() ------------------------------------------------------------ + +# #region test_run_executes_coroutine [C:2] [TYPE Function] +# @BRIEF run() executes a coroutine and returns its result. +def test_run_executes_coroutine(runner, event_loop): + """run() dispatches coroutine to event loop and waits for result.""" + # Start the event loop in a background thread + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + result = runner.run(async_return_value(42)) + assert result == 42 + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_executes_coroutine + + +# #region test_run_returns_string [C:2] [TYPE Function] +# @BRIEF run() works with string return values. +def test_run_returns_string(runner, event_loop): + """run() correctly returns string results.""" + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + result = runner.run(async_return_value("hello")) + assert result == "hello" + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_returns_string + + +# #region test_run_returns_none [C:2] [TYPE Function] +# @BRIEF run() works when coroutine returns None. +def test_run_returns_none(runner, event_loop): + """run() correctly handles None return value.""" + async def async_none(): + return None + + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + result = runner.run(async_none()) + assert result is None + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_returns_none + + +# #region test_run_propagates_exception [C:2] [TYPE Function] +# @BRIEF run() re-raises exceptions from the coroutine. +def test_run_propagates_exception(runner, event_loop): + """run() re-raises exceptions from the coroutine in the calling thread.""" + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + with pytest.raises(ValueError, match="boom"): + runner.run(async_raise(ValueError, "boom")) + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_propagates_exception + + +# #region test_run_propagates_runtime_error [C:2] [TYPE Function] +# @BRIEF run() re-raises RuntimeError from the coroutine. +def test_run_propagates_runtime_error(runner, event_loop): + """run() re-raises RuntimeError from the coroutine.""" + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + with pytest.raises(RuntimeError, match="connection lost"): + runner.run(async_raise(RuntimeError, "connection lost")) + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_propagates_runtime_error + + +# -- Tests: run_later() ------------------------------------------------------ + +# #region test_run_later_executes_after_delay [C:2] [TYPE Function] +# @BRIEF run_later() schedules a coroutine to execute after a delay. +def test_run_later_executes_after_delay(runner, event_loop): + """run_later() dispatches coroutine after specified delay.""" + called = [] + + async def track(): + called.append(True) + + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + runner.run_later(track(), delay_seconds=0.1) + time.sleep(0.3) + assert len(called) == 1, f"Expected coroutine to execute once, got {len(called)} times" + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_later_executes_after_delay + + +# #region test_run_later_does_not_execute_immediately [C:2] [TYPE Function] +# @BRIEF run_later() does not execute coroutine before the delay. +def test_run_later_does_not_execute_immediately(runner, event_loop): + """run_later() does not execute coroutine before the delay expires.""" + called = [] + + async def track(): + called.append(True) + + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + runner.run_later(track(), delay_seconds=0.5) + time.sleep(0.1) # Check before delay expires + assert len(called) == 0, "Coroutine should not have executed yet" + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_later_does_not_execute_immediately + + +# -- Tests: from background thread ------------------------------------------ + +# #region test_run_from_background_thread [C:2] [TYPE Function] +# @BRIEF run() works correctly when called from a non-main thread. +def test_run_from_background_thread(event_loop): + """run() can be called from a background thread (simulating APScheduler).""" + runner = AsyncJobRunner(event_loop) + results = [] + errors = [] + + def background_work(): + try: + result = runner.run(async_return_value(99)) + results.append(result) + except Exception as e: + errors.append(e) + + # Start the event loop in a background thread + loop_thread = threading.Thread(target=event_loop.run_forever, daemon=True) + loop_thread.start() + + try: + # Call run() from another background thread (simulating APScheduler) + work_thread = threading.Thread(target=background_work) + work_thread.start() + work_thread.join(timeout=5) + + assert len(errors) == 0, f"Unexpected errors: {errors}" + assert results == [99], f"Expected [99], got {results}" + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + loop_thread.join(timeout=2) +# #endregion test_run_from_background_thread + + +# -- Tests: edge cases ------------------------------------------------------- + +# #region test_run_with_complex_return_value [C:2] [TYPE Function] +# @BRIEF run() works with complex return values (dict, list). +def test_run_with_complex_return_value(runner, event_loop): + """run() correctly returns complex objects.""" + async def async_dict(): + return {"key": "value", "nested": [1, 2, 3]} + + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + result = runner.run(async_dict()) + assert result == {"key": "value", "nested": [1, 2, 3]} + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_with_complex_return_value + + +# #region test_run_later_with_zero_delay [C:2] [TYPE Function] +# @BRIEF run_later() with zero delay executes immediately. +def test_run_later_with_zero_delay(runner, event_loop): + """run_later() with delay_seconds=0 executes immediately.""" + called = [] + + async def track(): + called.append(True) + + thread = threading.Thread(target=event_loop.run_forever, daemon=True) + thread.start() + try: + runner.run_later(track(), delay_seconds=0) + time.sleep(0.2) + assert len(called) == 1, "Coroutine should have executed immediately" + finally: + event_loop.call_soon_threadsafe(event_loop.stop) + thread.join(timeout=2) +# #endregion test_run_later_with_zero_delay diff --git a/backend/tests/core/test_async_job_runner_integration.py b/backend/tests/core/test_async_job_runner_integration.py new file mode 100644 index 00000000..625ce3f6 --- /dev/null +++ b/backend/tests/core/test_async_job_runner_integration.py @@ -0,0 +1,188 @@ +# #region Test.AsyncJobRunnerIntegration [C:3] [TYPE Module] [SEMANTICS test,async,runner,integration] +# @BRIEF Integration tests for AsyncJobRunner with real event loop and threading. +# @RELATION BINDS_TO -> [AsyncJobRunner] +# @TEST_CONTRACT: AsyncJobRunner.run_with_real_loop -> result | works with real asyncio event loop +# @TEST_CONTRACT: AsyncJobRunner.run_later_with_real_loop -> result | delayed execution works +# @TEST_EDGE: concurrent_runs -> multiple run() calls from different threads +# @TEST_EDGE: run_with_await -> coroutine with await works correctly +# @TEST_INVARIANT: thread_safety -> VERIFIED_BY: [test_concurrent_runs_from_multiple_threads] + +import asyncio +import sys +import threading +import time +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) + +from src.core.async_job_runner import AsyncJobRunner + + +# -- Helpers ----------------------------------------------------------------- + +async def async_db_operation(value, delay=0.01): + """Simulate an async DB operation.""" + await asyncio.sleep(delay) + return {"result": value, "status": "ok"} + + +async def async_chain_operation(value): + """Simulate chained async operations.""" + result1 = await async_db_operation(value) + result2 = await async_db_operation(result1["result"] * 2) + return result2 + + +# -- Tests: Integration with real event loop --------------------------------- + +# #region test_run_with_real_event_loop [C:2] [TYPE Function] +# @BRIEF run() works with a real asyncio event loop running in a background thread. +def test_run_with_real_event_loop(): + """run() dispatches coroutine to a real event loop and waits for result.""" + loop = asyncio.new_event_loop() + runner = AsyncJobRunner(loop) + + def run_loop(): + loop.run_forever() + + loop_thread = threading.Thread(target=run_loop, daemon=True) + loop_thread.start() + + try: + result = runner.run(async_db_operation("test")) + assert result == {"result": "test", "status": "ok"} + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join(timeout=2) + loop.close() +# #endregion test_run_with_real_event_loop + + +# #region test_run_later_with_real_event_loop [C:2] [TYPE Function] +# @BRIEF run_later() works with a real event loop — coroutine executes after delay. +def test_run_later_with_real_event_loop(): + """run_later() schedules coroutine on real event loop with correct timing.""" + loop = asyncio.new_event_loop() + runner = AsyncJobRunner(loop) + execution_times = [] + + async def track_time(): + execution_times.append(time.monotonic()) + + def run_loop(): + loop.run_forever() + + loop_thread = threading.Thread(target=run_loop, daemon=True) + loop_thread.start() + + try: + start = time.monotonic() + runner.run_later(track_time(), delay_seconds=0.2) + time.sleep(0.5) + assert len(execution_times) == 1 + elapsed = execution_times[0] - start + assert 0.15 < elapsed < 0.4, f"Expected ~0.2s delay, got {elapsed:.3f}s" + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join(timeout=2) + loop.close() +# #endregion test_run_later_with_real_event_loop + + +# #region test_concurrent_runs_from_multiple_threads [C:2] [TYPE Function] +# @BRIEF Multiple threads can call run() concurrently without deadlocks. +def test_concurrent_runs_from_multiple_threads(): + """run() is thread-safe — multiple threads can dispatch coroutines concurrently.""" + loop = asyncio.new_event_loop() + runner = AsyncJobRunner(loop) + results = {} + errors = [] + + def worker(thread_id): + try: + result = runner.run(async_db_operation(f"thread-{thread_id}")) + results[thread_id] = result + except Exception as e: + errors.append((thread_id, e)) + + def run_loop(): + loop.run_forever() + + loop_thread = threading.Thread(target=run_loop, daemon=True) + loop_thread.start() + + try: + threads = [] + for i in range(5): + t = threading.Thread(target=worker, args=(i,)) + threads.append(t) + t.start() + + for t in threads: + t.join(timeout=10) + + assert len(errors) == 0, f"Errors in threads: {errors}" + assert len(results) == 5 + for i in range(5): + assert results[i]["result"] == f"thread-{i}" + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join(timeout=2) + loop.close() +# #endregion test_concurrent_runs_from_multiple_threads + + +# #region test_run_with_chained_await [C:2] [TYPE Function] +# @BRIEF run() works with coroutines that await other coroutines. +def test_run_with_chained_await(): + """run() correctly handles coroutines that internally await other coroutines.""" + loop = asyncio.new_event_loop() + runner = AsyncJobRunner(loop) + + def run_loop(): + loop.run_forever() + + loop_thread = threading.Thread(target=run_loop, daemon=True) + loop_thread.start() + + try: + result = runner.run(async_chain_operation(5)) + assert result == {"result": 10, "status": "ok"} + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join(timeout=2) + loop.close() +# #endregion test_run_with_chained_await + + +# #region test_run_later_multiple_tasks [C:2] [TYPE Function] +# @BRIEF Multiple run_later() calls schedule multiple coroutines correctly. +def test_run_later_multiple_tasks(): + """Multiple run_later() calls all execute their coroutines.""" + loop = asyncio.new_event_loop() + runner = AsyncJobRunner(loop) + execution_order = [] + + async def track(name): + execution_order.append(name) + + def run_loop(): + loop.run_forever() + + loop_thread = threading.Thread(target=run_loop, daemon=True) + loop_thread.start() + + try: + runner.run_later(track("first"), delay_seconds=0.1) + runner.run_later(track("second"), delay_seconds=0.2) + runner.run_later(track("third"), delay_seconds=0.3) + time.sleep(0.5) + assert len(execution_order) == 3 + assert execution_order == ["first", "second", "third"] + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join(timeout=2) + loop.close() +# #endregion test_run_later_multiple_tasks diff --git a/backend/tests/core/test_mapping_service.py b/backend/tests/core/test_mapping_service.py index ac891976..ab898e7c 100644 --- a/backend/tests/core/test_mapping_service.py +++ b/backend/tests/core/test_mapping_service.py @@ -360,89 +360,6 @@ async def test_sync_environment_deletes_stale_mappings(db_session): # #endregion test_sync_environment_deletes_stale_mappings -# #region test_start_scheduler_basic [C:2] [TYPE Function] -# @RELATION BINDS_TO ->[TestMappingService] -@patch("src.core.mapping_service.seed_trace_id") -@patch("src.core.mapping_service.BackgroundScheduler") -def test_start_scheduler_basic(mock_scheduler_cls, mock_seed, db_session): - """start_scheduler adds job and starts scheduler.""" - mock_scheduler = MagicMock() - mock_scheduler_cls.return_value = mock_scheduler - mock_scheduler.running = False - - service = IdMappingService(db_session) - mock_factory = MagicMock(return_value=MagicMock()) - service.start_scheduler("0 */5 * * *", ["env1", "env2"], mock_factory) - - mock_scheduler.add_job.assert_called_once() - mock_scheduler.start.assert_called_once() - - -# #endregion test_start_scheduler_basic - - -# #region test_start_scheduler_sync_all_called [C:2] [TYPE Function] -# @RELATION BINDS_TO ->[TestMappingService] -def test_start_scheduler_sync_all_called(db_session): - """sync_all inner function executed via scheduler.add_job (lines 75-80).""" - mock_scheduler = MagicMock() - mock_scheduler.running = False - - captured_fn = [None] - - def _capture_job(fn, *args, **kwargs): - captured_fn[0] = fn - return MagicMock(id="sync-job") - - mock_scheduler.add_job.side_effect = _capture_job - - mock_factory = MagicMock(return_value=MagicMock()) - - with patch("src.core.mapping_service.seed_trace_id") as mock_seed, \ - patch("src.core.mapping_service.BackgroundScheduler", return_value=mock_scheduler): - - service = IdMappingService(db_session) - with patch.object(service, 'sync_environment', return_value=None) as mock_sync: - service.start_scheduler("0 */5 * * *", ["env1", "env2"], mock_factory) - # Call the captured sync_all - captured_fn[0]() - - mock_seed.assert_called_once() - assert mock_factory.call_count == 2 - mock_factory.assert_any_call("env1") - mock_factory.assert_any_call("env2") - assert mock_sync.call_count == 2 -# #endregion test_start_scheduler_sync_all_called - - -# #region test_start_scheduler_update_existing [C:2] [TYPE Function] -# @RELATION BINDS_TO ->[TestMappingService] -@patch("src.core.mapping_service.seed_trace_id") -@patch("src.core.mapping_service.BackgroundScheduler") -def test_start_scheduler_update_existing(mock_scheduler_cls, mock_seed, db_session): - """start_scheduler with existing job → updates, doesn't re-start scheduler.""" - mock_scheduler = MagicMock() - mock_scheduler_cls.return_value = mock_scheduler - mock_scheduler.running = False # Not running initially - - service = IdMappingService(db_session) - mock_factory = MagicMock(return_value=MagicMock()) - service.start_scheduler("0 */5 * * *", ["env1"], mock_factory) - - # First call should start the scheduler (was not running) - mock_scheduler.start.assert_called_once() - - # Second call with existing job — scheduler already running - mock_scheduler.running = True - service.start_scheduler("0 */10 * * *", ["env1"], mock_factory) - - mock_scheduler.remove_job.assert_called_once() # called because _sync_job exists - assert mock_scheduler.add_job.call_count == 2 - - -# #endregion test_start_scheduler_update_existing - - # #region test_get_remote_id_value_error [C:2] [TYPE Function] # @RELATION BINDS_TO ->[TestMappingService] def test_get_remote_id_value_error(db_session): diff --git a/backend/tests/plugins/test_llm_analysis_scheduler.py b/backend/tests/plugins/test_llm_analysis_scheduler.py deleted file mode 100644 index 66642b05..00000000 --- a/backend/tests/plugins/test_llm_analysis_scheduler.py +++ /dev/null @@ -1,136 +0,0 @@ -# #region Test.LLMAnalysisScheduler [C:3] [TYPE Module] [SEMANTICS test, llm, analysis, scheduler, cron] -# @BRIEF Tests for llm_analysis/scheduler.py — _parse_cron, schedule_dashboard_validation. -# @RELATION BINDS_TO -> [LLMAnalysisScheduler] -# @TEST_CONTRACT: _parse_cron -> dict | cron expression parsing -# @TEST_CONTRACT: schedule_dashboard_validation -> None | schedules a validation job - -import sys -from pathlib import Path -sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) - -import os -os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing") -os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") -os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:") - -from unittest.mock import AsyncMock, MagicMock, patch - -import pytest - -from src.plugins.llm_analysis.scheduler import _parse_cron, schedule_dashboard_validation - - -class TestParseCron: - """_parse_cron — basic cron expression parser.""" - - def test_standard_5_part_cron(self): - """Standard 5-part cron expression parsed correctly.""" - result = _parse_cron("0 8 * * 1-5") - assert result == { - "minute": "0", - "hour": "8", - "day": "*", - "month": "*", - "day_of_week": "1-5", - } - - def test_daily_at_midnight(self): - """Daily at midnight.""" - result = _parse_cron("0 0 * * *") - assert result["hour"] == "0" - assert result["minute"] == "0" - - def test_every_hour(self): - """Every hour at minute 0.""" - result = _parse_cron("0 * * * *") - assert result["minute"] == "0" - assert result["hour"] == "*" - - def test_invalid_parts_count(self): - """Less or more than 5 parts returns empty dict.""" - assert _parse_cron("0 8 * *") == {} - assert _parse_cron("0 8 * * 1-5 extra") == {} - - def test_empty_string(self): - """Empty string returns empty dict.""" - assert _parse_cron("") == {} - - def test_step_values(self): - """Step values like */15 parsed correctly.""" - result = _parse_cron("*/15 * * * *") - assert result["minute"] == "*/15" - - -class TestScheduleDashboardValidation: - """schedule_dashboard_validation — schedules recurring validation.""" - - def test_schedule_with_params(self): - """Scheduler and task manager called with correct args.""" - mock_scheduler = MagicMock() - mock_task_manager = MagicMock() - - with patch("src.plugins.llm_analysis.scheduler.get_scheduler_service", return_value=mock_scheduler), \ - patch("src.plugins.llm_analysis.scheduler.get_task_manager", return_value=mock_task_manager): - - schedule_dashboard_validation( - dashboard_id="dash-1", - cron_expression="0 6 * * *", - params={"threshold": 0.8}, - ) - - # Scheduler should have added a cron job - mock_scheduler.add_job.assert_called_once() - call_args = mock_scheduler.add_job.call_args - assert call_args[0][1] == "cron" # trigger type - assert call_args[1]["id"] == "llm_val_dash-1" - assert call_args[1]["replace_existing"] is True - assert call_args[1]["minute"] == "0" - assert call_args[1]["hour"] == "6" - - def test_schedule_creates_task_on_execution(self): - """The scheduled job function creates a task when executed.""" - mock_scheduler = MagicMock() - mock_task_manager = AsyncMock() - - with patch("src.plugins.llm_analysis.scheduler.get_scheduler_service", return_value=mock_scheduler), \ - patch("src.plugins.llm_analysis.scheduler.get_task_manager", return_value=mock_task_manager): - - schedule_dashboard_validation( - dashboard_id="dash-1", - cron_expression="0 6 * * *", - params={"threshold": 0.8}, - ) - - # Get the job function and call it - job_func = mock_scheduler.add_job.call_args[0][0] - import asyncio - asyncio.run(job_func()) - - mock_task_manager.create_task.assert_called_once_with( - plugin_id="llm_dashboard_validation", - params={"dashboard_id": "dash-1", "threshold": 0.8}, - ) - - def test_schedule_without_params(self): - """Schedule without extra params still works.""" - mock_scheduler = MagicMock() - mock_task_manager = AsyncMock() - - with patch("src.plugins.llm_analysis.scheduler.get_scheduler_service", return_value=mock_scheduler), \ - patch("src.plugins.llm_analysis.scheduler.get_task_manager", return_value=mock_task_manager): - - schedule_dashboard_validation( - dashboard_id="dash-2", - cron_expression="0 0 * * *", - params={}, - ) - - job_func = mock_scheduler.add_job.call_args[0][0] - import asyncio - asyncio.run(job_func()) - - mock_task_manager.create_task.assert_called_once_with( - plugin_id="llm_dashboard_validation", - params={"dashboard_id": "dash-2"}, - ) -# #endregion Test.LLMAnalysisScheduler diff --git a/backend/tests/plugins/translate/test_scheduler.py b/backend/tests/plugins/translate/test_scheduler.py index eab2a381..6eed43dc 100644 --- a/backend/tests/plugins/translate/test_scheduler.py +++ b/backend/tests/plugins/translate/test_scheduler.py @@ -402,6 +402,9 @@ class TestExecuteScheduledTranslation: patch( "src.plugins.translate.scheduler.NotificationService", ) as MockNotif, + patch( + "src.dependencies.get_async_job_runner", + ) as MockRunner, ): mock_orch = MagicMock() mock_run = MagicMock() @@ -412,7 +415,12 @@ class TestExecuteScheduledTranslation: mock_orch.execute_run.side_effect = RuntimeError("LLM call failed") MockOrch.return_value = mock_orch - # Provider mock — send must return a real coroutine for asyncio.run() + # Mock the runner to raise the exception from execute_run + mock_runner = MagicMock() + mock_runner.run.side_effect = RuntimeError("LLM call failed") + 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() diff --git a/backend/tests/test_core_scheduler.py b/backend/tests/test_core_scheduler.py index dc7eb2f8..4c40b22a 100644 --- a/backend/tests/test_core_scheduler.py +++ b/backend/tests/test_core_scheduler.py @@ -72,6 +72,7 @@ def test_load_schedules_reloads_translation_schedules(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() # Pre-register a translation job (simulates what happens before reload) svc.add_translation_job( @@ -126,6 +127,7 @@ def test_add_and_remove_translation_job(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() # Should NOT exist before add assert sched.get_job("translate_sched-x") is None @@ -166,6 +168,7 @@ def test_remove_nonexistent_translation_job_no_error(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() # Should not raise — silent ignore svc.remove_translation_job("nonexistent-sched") @@ -188,6 +191,7 @@ def test_add_validation_job(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() assert sched.get_job("validation_pol-1") is None @@ -221,6 +225,7 @@ def test_load_schedules_loads_validation_policies(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() # Pre-register a validation job (to test it survives reload) svc.add_validation_job( @@ -258,6 +263,7 @@ def test_load_schedules_loads_validation_policies(): def test_trigger_validation_creates_tasks(): """Verify _trigger_validation creates TaskManager tasks for each dashboard.""" from src.core.scheduler import SchedulerService + from src.core.async_job_runner import AsyncJobRunner sched = _start_scheduler() try: @@ -268,7 +274,9 @@ def test_trigger_validation_creates_tasks(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() svc.loop = asyncio.new_event_loop() + svc.runner = MagicMock(spec=AsyncJobRunner) # Mock the ValidationPolicy DB query mock_policy = MagicMock() @@ -283,21 +291,12 @@ def test_trigger_validation_creates_tasks(): 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): + with patch("src.core.scheduler.SessionLocal", return_value=mock_db): 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)}" + # Should have submitted 2 tasks via runner (one per dashboard) + assert svc.runner.run.call_count + svc.runner.run_later.call_count == 2, ( + f"Expected 2 task submissions via runner, got {svc.runner.run.call_count + svc.runner.run_later.call_count}" ) finally: _stop_scheduler(sched) @@ -318,6 +317,7 @@ def test_trigger_validation_skips_missing_policy(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() svc.loop = asyncio.new_event_loop() # Mock DB to return None (policy not found) @@ -349,6 +349,7 @@ def test_trigger_validation_skips_inactive_policy(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() svc.loop = asyncio.new_event_loop() mock_policy = MagicMock() @@ -383,6 +384,7 @@ def test_trigger_validation_skips_no_dashboards(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() svc.loop = asyncio.new_event_loop() mock_policy = MagicMock() @@ -416,6 +418,7 @@ def test_remove_validation_job(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() svc.add_validation_job("pol-remove", cron_expression="0 10 * * 1") assert sched.get_job("validation_pol-remove") is not None @@ -441,6 +444,7 @@ def test_remove_nonexistent_validation_job(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() # Should not raise svc.remove_validation_job("nonexistent-pol") @@ -463,6 +467,7 @@ def test_reload_validation_policy_adds_job(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() # Mock DB to return an active policy with schedule_days mock_policy = MagicMock() @@ -498,6 +503,7 @@ def test_reload_validation_policy_removes_job_when_policy_gone(): svc.task_manager = tm svc.config_manager = cfg svc.scheduler = sched + svc.runner = MagicMock() # Pre-register a job svc.add_validation_job("pol-gone", cron_expression="0 10 * * 1") diff --git a/backend/tests/test_translate_scheduler_audit.py b/backend/tests/test_translate_scheduler_audit.py new file mode 100644 index 00000000..789636f8 --- /dev/null +++ b/backend/tests/test_translate_scheduler_audit.py @@ -0,0 +1,320 @@ +# #region TestTranslateSchedulerAudit [C:3] [TYPE Module] [SEMANTICS test,translate,scheduler,audit,async-sync-mismatch,id-mismatch] +# @BRIEF Orthogonal audit of scheduled translation execution — async/sync mismatch and ID mismatch bugs. +# @RELATION BINDS_TO -> [execute_scheduled_translation] +# @RELATION BINDS_TO -> [TranslationScheduler] +# @RELATION BINDS_TO -> [SchedulerService] +# @RELATION BINDS_TO -> [AsyncJobRunner] +# @TEST_EDGE: async_coroutine_not_awaited — execute_run returns coroutine but never awaited +# @TEST_EDGE: disable_schedule_wrong_id — disable_schedule passes job_id instead of schedule_id +# @TEST_EDGE: delete_schedule_wrong_id — delete_schedule passes job_id instead of schedule_id +# @TEST_INVARIANT: execute_run_must_be_awaited -> VERIFIED_BY: [test_scheduler_uses_runner_for_execute_run] +# @TEST_INVARIANT: remove_job_uses_schedule_id -> VERIFIED_BY: [test_disable_schedule_uses_schedule_id, test_delete_schedule_uses_schedule_id] +import asyncio +import inspect +from pathlib import Path +from unittest.mock import MagicMock, patch, AsyncMock + +import pytest + +sys_path = str(Path(__file__).parent.parent / "src") +import sys +if sys_path not in sys.path: + sys.path.insert(0, sys_path) + + +# ============================================================================= +# BUG #1 FIX: async/sync mismatch — now uses AsyncJobRunner +# ============================================================================= + +class TestAsyncSyncMismatch: + """Verify that execute_run is async and uses AsyncJobRunner.""" + + # #region test_execute_run_is_async_coroutine [C:2] [TYPE Function] + # @BRIEF Confirm execute_run is an async method that returns a coroutine. + def test_execute_run_is_async_coroutine(self): + """execute_run is defined as async def — calling it without await returns a coroutine object.""" + from src.plugins.translate.orchestrator import TranslationOrchestrator + + # Verify the method is actually async + assert inspect.iscoroutinefunction(TranslationOrchestrator.execute_run), ( + "TranslationOrchestrator.execute_run must be an async method" + ) + # #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). + def test_scheduler_uses_runner_for_execute_run(self): + """In execute_scheduled_translation, orch.execute_run(run) is called via runner.run().""" + import ast + import inspect + + # Read the source code of execute_scheduled_translation + 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 + tree = ast.parse(source) + + # Find all Call nodes that call execute_run + execute_run_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) + + assert len(execute_run_calls) > 0, "No calls to execute_run found in execute_scheduled_translation" + + # Check if any call is wrapped in runner.run() + 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))" + ) + # #endregion test_scheduler_uses_runner_for_execute_run + + # #region test_manual_trigger_uses_asyncio_create_task [C:2] [TYPE Function] + # @BRIEF Manual trigger correctly uses asyncio.create_task for async execution. + def test_manual_trigger_uses_asyncio_create_task(self): + """The manual trigger route uses asyncio.create_task() which correctly handles async execution.""" + import ast + import inspect + from src.api.routes.translate._run_routes import run_translation + + source = inspect.getsource(run_translation) + tree = ast.parse(source) + + # Look for asyncio.create_task calls + has_create_task = False + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Attribute): + if node.func.attr == 'create_task': + has_create_task = True + elif isinstance(node.func, ast.Name): + if node.func.id == 'create_task': + has_create_task = True + + assert has_create_task, ( + "Manual trigger should use asyncio.create_task() for async execution" + ) + # #endregion test_manual_trigger_uses_asyncio_create_task + + +# ============================================================================= +# BUG #2: ID mismatch — disable_schedule and delete_schedule pass wrong ID +# ============================================================================= + +class TestScheduleIdMismatch: + """Verify that disable_schedule and delete_schedule use correct IDs.""" + + # #region test_add_translation_job_uses_schedule_id [C:2] [TYPE Function] + # @BRIEF add_translation_job constructs APScheduler job ID from schedule_id. + def test_add_translation_job_uses_schedule_id(self): + """add_translation_job creates job ID as 'translate_{schedule_id}'.""" + from src.core.scheduler import SchedulerService + + mock_task_manager = MagicMock() + mock_config = MagicMock() + + svc = SchedulerService(mock_task_manager, mock_config) + svc.scheduler = MagicMock() + + schedule_id = "sched-uuid-123" + job_id = "job-uuid-456" + + svc.add_translation_job( + schedule_id=schedule_id, + job_id=job_id, + cron_expression="*/30 * * * *", + timezone="UTC", + execution_mode="full" + ) + + # Verify the APScheduler job ID is based on schedule_id + call_args = svc.scheduler.add_job.call_args + aps_job_id = call_args.kwargs.get('id') or call_args.args[1] if len(call_args.args) > 1 else None + + assert aps_job_id == f"translate_{schedule_id}", ( + f"APScheduler job ID should be 'translate_{schedule_id}', got '{aps_job_id}'" + ) + # #endregion test_add_translation_job_uses_schedule_id + + # #region test_remove_translation_job_uses_schedule_id [C:2] [TYPE Function] + # @BRIEF remove_translation_job constructs APScheduler job ID from schedule_id. + def test_remove_translation_job_uses_schedule_id(self): + """remove_translation_job removes job with ID 'translate_{schedule_id}'.""" + from src.core.scheduler import SchedulerService + + mock_task_manager = MagicMock() + mock_config = MagicMock() + + svc = SchedulerService(mock_task_manager, mock_config) + svc.scheduler = MagicMock() + + schedule_id = "sched-uuid-123" + svc.remove_translation_job(schedule_id=schedule_id) + + # Verify it tries to remove the correct job ID + svc.scheduler.remove_job.assert_called_once_with(f"translate_{schedule_id}") + # #endregion test_remove_translation_job_uses_schedule_id + + # #region test_disable_schedule_uses_schedule_id [C:2] [TYPE Function] + # @BRIEF disable_schedule uses schedule.id (not job_id) for remove_translation_job (FIXED). + def test_disable_schedule_uses_schedule_id(self): + """FIXED: disable_schedule route uses schedule.id for remove_translation_job.""" + import ast + import inspect + from src.api.routes.translate._schedule_routes import disable_schedule + + source = inspect.getsource(disable_schedule) + tree = ast.parse(source) + + # Find calls to remove_translation_job + remove_calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Attribute) and node.func.attr == 'remove_translation_job': + remove_calls.append(node) + + assert len(remove_calls) > 0, "No calls to remove_translation_job found" + + # Check what argument is passed — should NOT be job_id directly + for call in remove_calls: + for kw in call.keywords: + if kw.arg == 'schedule_id': + # Should NOT be job_id (the parameter name) + if isinstance(kw.value, ast.Name) and kw.value.id == 'job_id': + pytest.fail( + "BUG STILL EXISTS: disable_schedule passes job_id to remove_translation_job. " + "Expected: schedule.id from set_schedule_active() return value." + ) + # #endregion test_disable_schedule_uses_schedule_id + + # #region test_delete_schedule_uses_schedule_id [C:2] [TYPE Function] + # @BRIEF delete_schedule uses schedule.id (not job_id) for remove_translation_job (FIXED). + def test_delete_schedule_uses_schedule_id(self): + """FIXED: delete_schedule route uses schedule.id for remove_translation_job.""" + import ast + import inspect + from src.api.routes.translate._schedule_routes import delete_schedule + + source = inspect.getsource(delete_schedule) + tree = ast.parse(source) + + # Find calls to remove_translation_job + remove_calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Attribute) and node.func.attr == 'remove_translation_job': + remove_calls.append(node) + + assert len(remove_calls) > 0, "No calls to remove_translation_job found" + + # Check what argument is passed — should NOT be job_id directly + for call in remove_calls: + for kw in call.keywords: + if kw.arg == 'schedule_id': + # Should NOT be job_id (the parameter name) + if isinstance(kw.value, ast.Name) and kw.value.id == 'job_id': + pytest.fail( + "BUG STILL EXISTS: delete_schedule passes job_id to remove_translation_job. " + "Expected: schedule_id from get_schedule() return value." + ) + # #endregion test_delete_schedule_uses_schedule_id + + # #region test_schedule_id_differs_from_job_id [C:2] [TYPE Function] + # @BRIEF TranslationSchedule.id and TranslationSchedule.job_id are different UUIDs. + def test_schedule_id_differs_from_job_id(self): + """TranslationSchedule has separate id and job_id fields — they are different UUIDs.""" + from src.models.translate import TranslationSchedule + + # Verify the model has both fields + assert hasattr(TranslationSchedule, 'id'), "TranslationSchedule should have 'id' field" + assert hasattr(TranslationSchedule, 'job_id'), "TranslationSchedule should have 'job_id' field" + + # The id is the primary key, job_id is a foreign key to TranslationJob + # They are different UUIDs + import uuid + schedule_id = str(uuid.uuid4()) + job_id = str(uuid.uuid4()) + + assert schedule_id != job_id, ( + "schedule_id and job_id are different UUIDs — " + "using job_id where schedule_id is expected will cause ID mismatch" + ) + # #endregion test_schedule_id_differs_from_job_id + + +# ============================================================================= +# Integration verification: end-to-end scheduled execution flow +# ============================================================================= + +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). + 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 + """ + from src.plugins.translate.scheduler import execute_scheduled_translation + from src.plugins.translate.orchestrator import TranslationOrchestrator + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + # Setup schedule + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = None # no concurrent + + q3 = MagicMock() + q3.filter.return_value.order_by.return_value.first.return_value = None # no recent run + + mock_db.query.side_effect = [q1, q2, q3] + + # Setup orchestrator mock + mock_run = MagicMock() + mock_run.id = "run-1" + mock_run.status = "PENDING" + + # Mock the runner to track calls + mock_runner = MagicMock() + + 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() + # #endregion test_scheduled_execution_uses_runner