mirror of
https://github.com/rommapp/romm.git
synced 2026-10-12 02:21:58 +00:00
fix(tasks): code-review pass on the task catalog split
Loads every model before a job resolves its task, so the mappers configure in a work horse that imported a single task module. Moves the httpx2 import into RemoteFilePullTask.run, builds the push-pull job meta from its spec, gates the scheduled scan on its spec, and widens the loaded-modules test to cover task modules and httpx2. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
258b87de0d
commit
e62c452ea9
@@ -6,7 +6,6 @@ from typing import Annotated
|
||||
from fastapi import BackgroundTasks, Depends, HTTPException, Request, status
|
||||
from pydantic import Field
|
||||
|
||||
from config import TASK_TIMEOUT
|
||||
from decorators.auth import protected_route
|
||||
from endpoints.play_sessions import PlaySessionEntry
|
||||
from endpoints.responses.base import PAGE_QUERY, BaseModel, PageParams
|
||||
@@ -42,6 +41,7 @@ from models.assets import Save
|
||||
from models.deleted_asset import DeletedAsset
|
||||
from models.device import SyncMode
|
||||
from models.sync_session import SyncSessionStatus
|
||||
from tasks.registry import SYNC_PUSH_PULL_SPEC
|
||||
from utils.auth import token_device_id
|
||||
from utils.datetime import to_utc
|
||||
from utils.router import APIRouter
|
||||
@@ -567,12 +567,8 @@ def trigger_push_pull(
|
||||
device_id=device.id,
|
||||
session_id=sync_session.id,
|
||||
force=True,
|
||||
job_timeout=TASK_TIMEOUT,
|
||||
meta={
|
||||
"task_key": "sync_push_pull",
|
||||
"task_name": "Push-Pull Sync",
|
||||
"task_type": "sync",
|
||||
},
|
||||
job_timeout=SYNC_PUSH_PULL_SPEC.timeout,
|
||||
meta=SYNC_PUSH_PULL_SPEC.job_meta("sync_push_pull"),
|
||||
)
|
||||
|
||||
log.info(f"Enqueued push-pull sync for device {device.id}")
|
||||
|
||||
@@ -1,6 +1,3 @@
|
||||
from config import (
|
||||
ENABLE_SCHEDULED_RESCAN,
|
||||
)
|
||||
from endpoints.sockets.scan import ScanStats, scan_platforms
|
||||
from handler.scan_handler import ScanType, get_enabled_metadata_sources
|
||||
from logger.logger import log
|
||||
@@ -15,7 +12,7 @@ class ScanLibraryTask(PeriodicTask):
|
||||
async def run(self) -> dict[str, str]:
|
||||
scan_stats = ScanStats()
|
||||
|
||||
if not ENABLE_SCHEDULED_RESCAN:
|
||||
if not self.spec.enabled:
|
||||
log.info("Scheduled library scan not enabled, skipping...")
|
||||
return scan_stats.to_dict()
|
||||
|
||||
|
||||
@@ -5,7 +5,6 @@ from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Any, ClassVar
|
||||
|
||||
import httpx2
|
||||
from rq import get_current_job
|
||||
from rq.exceptions import AbandonedJobError
|
||||
from rq.job import Job
|
||||
@@ -40,8 +39,12 @@ async def run_task_by_name(
|
||||
Whatever the task returns.
|
||||
"""
|
||||
# Imported here because the task modules the registry resolves import this one.
|
||||
from models import load_all_models
|
||||
from tasks.registry import get_task
|
||||
|
||||
# A task module may load a model whose relationships name models it doesn't
|
||||
# import, and the mappers configure only once every one of them is loaded.
|
||||
load_all_models()
|
||||
task = get_task(name)
|
||||
if task is None:
|
||||
raise TaskNotFoundException(name)
|
||||
@@ -228,6 +231,8 @@ class RemoteFilePullTask(PeriodicTask, ABC):
|
||||
|
||||
# Imported here because the HTTP client stack would otherwise load in
|
||||
# every process that reads the task catalog, the cron scheduler included.
|
||||
import httpx2
|
||||
|
||||
from utils.context import ctx_httpx_client
|
||||
|
||||
httpx_client = ctx_httpx_client.get()
|
||||
|
||||
@@ -12,7 +12,7 @@ from endpoints import auth as auth_endpoints
|
||||
from endpoints import permissions as permission_endpoints
|
||||
from handler.auth.base_handler import auth_handler
|
||||
from models.user import User
|
||||
from tasks.tasks import Task, TaskType
|
||||
from tasks.tasks import TaskSpec, TaskType
|
||||
|
||||
|
||||
def _auth(token: str) -> dict[str, str]:
|
||||
@@ -155,7 +155,7 @@ class TestUsers:
|
||||
def test_a_manual_task_run_is_recorded(
|
||||
_enqueue, _worker, client: TestClient, access_token: str
|
||||
):
|
||||
task = Mock(spec=Task)
|
||||
task = Mock(spec=TaskSpec)
|
||||
task.title = "Cleanup"
|
||||
task.task_type = TaskType.CLEANUP
|
||||
task.can_run_manually = True
|
||||
|
||||
@@ -8,7 +8,7 @@ from rq.exceptions import DeserializationError, NoSuchJobError
|
||||
from handler.redis_handler import low_prio_queue, redis_client
|
||||
from tasks.manual.cleanup_missing_firmware import CleanupMissingFirmwareStats
|
||||
from tasks.manual.cleanup_missing_roms import CleanupMissingRomsStats
|
||||
from tasks.tasks import Task, TaskType
|
||||
from tasks.tasks import TaskSpec, TaskType
|
||||
|
||||
|
||||
def _job_with_meta(meta: dict[str, Any]) -> Mock:
|
||||
@@ -35,7 +35,7 @@ def task_worker_listening():
|
||||
@pytest.fixture
|
||||
def mock_task():
|
||||
"""Create a mock task for testing"""
|
||||
task = Mock(spec=Task)
|
||||
task = Mock(spec=TaskSpec)
|
||||
task.title = "Test Task"
|
||||
task.description = "A test task for unit testing"
|
||||
task.task_type = TaskType.CLEANUP
|
||||
@@ -44,14 +44,13 @@ def mock_task():
|
||||
task.can_run_manually = True
|
||||
task.cron_string = "0 0 * * *"
|
||||
task.timeout = 300
|
||||
task.run = Mock()
|
||||
return task
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_disabled_task():
|
||||
"""Create a mock disabled task for testing"""
|
||||
task = Mock(spec=Task)
|
||||
task = Mock(spec=TaskSpec)
|
||||
task.title = "Disabled Task"
|
||||
task.description = "A disabled task for testing"
|
||||
task.task_type = TaskType.CLEANUP
|
||||
@@ -60,14 +59,13 @@ def mock_disabled_task():
|
||||
task.can_run_manually = False
|
||||
task.cron_string = None
|
||||
task.timeout = 300
|
||||
task.run = Mock()
|
||||
return task
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_non_manual_task():
|
||||
"""Create a mock task that cannot be run manually"""
|
||||
task = Mock(spec=Task)
|
||||
task = Mock(spec=TaskSpec)
|
||||
task.title = "Non-Manual Task"
|
||||
task.description = "A task that cannot be run manually"
|
||||
task.task_type = TaskType.CLEANUP
|
||||
@@ -76,7 +74,6 @@ def mock_non_manual_task():
|
||||
task.can_run_manually = False
|
||||
task.cron_string = "0 0 * * *"
|
||||
task.timeout = 300
|
||||
task.run = Mock()
|
||||
return task
|
||||
|
||||
|
||||
@@ -109,7 +106,7 @@ class TestListTasks:
|
||||
"endpoints.tasks.MANUAL_TASKS",
|
||||
{
|
||||
"test_manual": Mock(
|
||||
spec=Task,
|
||||
spec=TaskSpec,
|
||||
task_type=TaskType.CLEANUP,
|
||||
title="Manual Task",
|
||||
description="Manual task",
|
||||
@@ -125,7 +122,7 @@ class TestListTasks:
|
||||
"endpoints.tasks.VISIBLE_SCHEDULED_TASKS",
|
||||
{
|
||||
"test_scheduled": Mock(
|
||||
spec=Task,
|
||||
spec=TaskSpec,
|
||||
task_type=TaskType.UPDATE,
|
||||
title="Scheduled Task",
|
||||
description="Scheduled task",
|
||||
@@ -312,7 +309,7 @@ class TestRunSingleTask:
|
||||
"endpoints.tasks.RUNNABLE_TASKS",
|
||||
{
|
||||
"disabled_task": Mock(
|
||||
spec=Task,
|
||||
spec=TaskSpec,
|
||||
task_type=TaskType.CLEANUP,
|
||||
title="Disabled Task",
|
||||
description="Disabled Description",
|
||||
@@ -320,7 +317,6 @@ class TestRunSingleTask:
|
||||
manual_run=True,
|
||||
can_run_manually=False,
|
||||
timeout=300,
|
||||
run=Mock(),
|
||||
),
|
||||
},
|
||||
)
|
||||
@@ -340,7 +336,7 @@ class TestRunSingleTask:
|
||||
"endpoints.tasks.RUNNABLE_TASKS",
|
||||
{
|
||||
"non_manual_task": Mock(
|
||||
spec=Task,
|
||||
spec=TaskSpec,
|
||||
task_type=TaskType.CLEANUP,
|
||||
title="Non-Manual Task",
|
||||
description="Non-Manual Description",
|
||||
@@ -348,7 +344,6 @@ class TestRunSingleTask:
|
||||
manual_run=False,
|
||||
can_run_manually=False,
|
||||
timeout=300,
|
||||
run=Mock(),
|
||||
),
|
||||
},
|
||||
)
|
||||
@@ -693,7 +688,7 @@ class TestTaskInfoBuilding:
|
||||
"endpoints.tasks.MANUAL_TASKS",
|
||||
{
|
||||
"test_task": Mock(
|
||||
spec=Task,
|
||||
spec=TaskSpec,
|
||||
title="Test Task",
|
||||
description="Test Description",
|
||||
enabled=True,
|
||||
@@ -732,7 +727,7 @@ class TestIntegration:
|
||||
"endpoints.tasks.RUNNABLE_TASKS",
|
||||
{
|
||||
"workflow_task": Mock(
|
||||
spec=Task,
|
||||
spec=TaskSpec,
|
||||
task_type=TaskType.CLEANUP,
|
||||
title="Workflow Task",
|
||||
description="Workflow Description",
|
||||
@@ -740,7 +735,6 @@ class TestIntegration:
|
||||
manual_run=True,
|
||||
can_run_manually=True,
|
||||
timeout=300,
|
||||
run=Mock(),
|
||||
),
|
||||
},
|
||||
):
|
||||
@@ -769,7 +763,7 @@ class TestRunSingleTaskArgumentHandling:
|
||||
"endpoints.tasks.RUNNABLE_TASKS",
|
||||
{
|
||||
"allowed_task": Mock(
|
||||
spec=Task,
|
||||
spec=TaskSpec,
|
||||
task_type=TaskType.CLEANUP,
|
||||
title="Allowed Task",
|
||||
description="Allowed",
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from dataclasses import replace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
@@ -34,7 +35,7 @@ class TestScanLibraryTask:
|
||||
"""Test run when scheduled rescan is enabled"""
|
||||
for name in ("meta_ra_handler", "meta_launchbox_handler"):
|
||||
mocker.patch.object(providers[name], "is_enabled", return_value=True)
|
||||
mocker.patch("tasks.scheduled.scan_library.ENABLE_SCHEDULED_RESCAN", True)
|
||||
task.spec = replace(task.spec, enabled=True)
|
||||
|
||||
scan_result = MagicMock()
|
||||
mock_scan_platforms = mocker.patch(
|
||||
@@ -55,7 +56,7 @@ class TestScanLibraryTask:
|
||||
|
||||
async def test_run_disabled(self, task, mocker):
|
||||
"""Test run when scheduled rescan is disabled"""
|
||||
mocker.patch("tasks.scheduled.scan_library.ENABLE_SCHEDULED_RESCAN", False)
|
||||
task.spec = replace(task.spec, enabled=False)
|
||||
mock_scan_platforms = mocker.patch(
|
||||
"tasks.scheduled.scan_library.scan_platforms"
|
||||
)
|
||||
|
||||
@@ -40,8 +40,16 @@ def test_task_func_resolves_in_a_fresh_interpreter(func):
|
||||
|
||||
|
||||
# Loaded only by the jobs that need them, so the long-lived scheduler and
|
||||
# worker parents stay small.
|
||||
APP_STACK_MODULES = ("handler.database", "handler.auth.base_handler", "fastapi")
|
||||
# worker parents stay small. Each name covers its submodules too.
|
||||
JOB_ONLY_MODULES = (
|
||||
"tasks.scheduled",
|
||||
"tasks.manual",
|
||||
"tasks.sync_push_pull_task",
|
||||
"handler.database",
|
||||
"handler.auth.base_handler",
|
||||
"fastapi",
|
||||
"httpx2",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("module", ["tasks.cron_config", "handler.rq_worker"])
|
||||
@@ -52,9 +60,11 @@ def test_the_scheduler_and_worker_load_no_task_code(module):
|
||||
"-c",
|
||||
"import importlib, sys\n"
|
||||
"importlib.import_module(sys.argv[1])\n"
|
||||
"print('LOADED:' + ','.join(m for m in sys.argv[2:] if m in sys.modules))\n",
|
||||
"watched = sys.argv[2:]\n"
|
||||
"print('LOADED:' + ','.join(m for m in sys.modules\n"
|
||||
" if any(m == w or m.startswith(w + '.') for w in watched)))\n",
|
||||
module,
|
||||
*APP_STACK_MODULES,
|
||||
*JOB_ONLY_MODULES,
|
||||
],
|
||||
cwd=BACKEND_ROOT,
|
||||
env={**os.environ, "PYTHONPATH": str(BACKEND_ROOT)},
|
||||
|
||||
Reference in New Issue
Block a user