From 03475b18f46b0fc66707e0911e540e69fb0663b4 Mon Sep 17 00:00:00 2001 From: mschwoerer <82171591+mschwoer@users.noreply.github.com> Date: Fri, 4 Sep 2026 10:24:52 +0200 Subject: [PATCH 1/2] dispatch jobs by runner name: QuantingEnv.runner, factory takes a Runner Task 5 of tasks/todo.md. Exported env var _JOB_ENGINE becomes _RUNNER. Transitional for this commit only: processor_impl feeds runner=settings.job_engine, valid because every in-repo runner is named after its engine; T6 replaces it with settings.runner. Co-Authored-By: Claude Fable 5.1 --- airflow_src/dags/impl/processor_impl.py | 12 +++--- airflow_src/plugins/common/quanting_env.py | 2 +- airflow_src/plugins/jobs/job_handler.py | 27 ++++++------ airflow_src/plugins/sensors/ssh_sensor.py | 6 +-- airflow_src/tests/conftest.py | 2 +- .../tests/dags/impl/test_processor_impl.py | 8 ++-- .../tests/plugins/jobs/test_job_handler.py | 43 ++++++++++++++++--- .../tests/plugins/sensors/test_ssh_sensor.py | 10 ++--- tasks/todo.md | 12 +++--- 9 files changed, 74 insertions(+), 48 deletions(-) diff --git a/airflow_src/dags/impl/processor_impl.py b/airflow_src/dags/impl/processor_impl.py index d12b129d..3b3f5ba5 100644 --- a/airflow_src/dags/impl/processor_impl.py +++ b/airflow_src/dags/impl/processor_impl.py @@ -225,7 +225,8 @@ def _create_quanting_env( settings_version=settings.version, relative_raw_file_path=str(relative_raw_file_path), config_params=substituted_params, - job_engine=settings.job_engine, + # transitional until Settings.runner exists: every in-repo runner is named after its engine + runner=settings.job_engine, year_month_folder=get_created_at_year_month(raw_file), ) @@ -284,6 +285,7 @@ def _prepare_custom_command(settings: Settings, substituted_params: str) -> str: "settings_name", "year_month_folder", "slurm_mem", + "runner", ) # composed of a yaml base path (admin configuration) and fields checked above, e.g. @@ -295,7 +297,6 @@ def _prepare_custom_command(settings: Settings, substituted_params: str) -> str: "custom_command", "config_params", "slurm_time", # contains ":", validated in the webapp - "job_engine", # selectbox value, dispatch only, not exported to the job ) @@ -391,10 +392,7 @@ def submit_job( output_path.mkdir(parents=True, exist_ok=True) - job_id = start_job( - quanting_env, - engine=quanting_env.job_engine, - ) + job_id = start_job(quanting_env, runner_name=quanting_env.runner) # TODO: race condition here, e.g. for file in ERROR status update_raw_file(quanting_env.raw_file_id, new_status=RawFileStatus.QUANTING) @@ -468,7 +466,7 @@ def check_job_result(*, quanting_env_dict: dict, job_id: str, ti: TaskInstance) """ quanting_env = QuantingEnv.from_dict(quanting_env_dict) - job_status, time_elapsed = get_job_result(job_id, engine=quanting_env.job_engine) + job_status, time_elapsed = get_job_result(job_id, runner_name=quanting_env.runner) logging.info(f"Job {job_id} exited with status {job_status}.") diff --git a/airflow_src/plugins/common/quanting_env.py b/airflow_src/plugins/common/quanting_env.py index f89bd36e..8e186719 100644 --- a/airflow_src/plugins/common/quanting_env.py +++ b/airflow_src/plugins/common/quanting_env.py @@ -42,7 +42,7 @@ class QuantingEnv(BaseModel): relative_raw_file_path: str = Field(alias="_RELATIVE_RAW_FILE_PATH") config_params: str = Field(alias="_CONFIG_PARAMS") - job_engine: str = Field(alias="_JOB_ENGINE") + runner: str = Field(alias="_RUNNER") year_month_folder: str = Field(alias="_YEAR_MONTH_FOLDER") def to_dict(self) -> dict: diff --git a/airflow_src/plugins/jobs/job_handler.py b/airflow_src/plugins/jobs/job_handler.py index 26f7d76a..daddb64c 100644 --- a/airflow_src/plugins/jobs/job_handler.py +++ b/airflow_src/plugins/jobs/job_handler.py @@ -12,10 +12,12 @@ from shared.keys import EnvVars, JobEngines from shared.path_views import CLUSTER_VIEW, DOCKER_HOST_VIEW, Locations +from shared.runners import Runner, get_runner -def _get_job_handler(engine: str) -> "JobHandler": - """Factory function to get the appropriate job handler for the given engine.""" +def _get_job_handler(runner: Runner) -> "JobHandler": + """Factory function to get the job handler for the engine of the given runner.""" + engine = runner.engine if engine == JobEngines.SLURM: from jobs.slurm_ssh_job_handler import SlurmSSHJobHandler @@ -81,28 +83,25 @@ def get_job_result(self, job_id: str) -> tuple[str, int]: """ -def start_job( - quanting_env: QuantingEnv, - engine: str, -) -> str: - """Start a job using the given job engine. +def start_job(quanting_env: QuantingEnv, runner_name: str) -> str: + """Start a job on the given runner. Delegates to JobHandler.start_job(), see docs there. """ - handler = _get_job_handler(engine) + handler = _get_job_handler(get_runner(runner_name)) return handler.start_job(quanting_env) -def get_job_status(job_id: str, engine: str) -> str: - """Get the job status using the given job engine. +def get_job_status(job_id: str, runner_name: str) -> str: + """Get the job status from the given runner. Delegates to JobHandler.get_job_status(), see docs there. """ - handler = _get_job_handler(engine) + handler = _get_job_handler(get_runner(runner_name)) return handler.get_job_status(job_id) -def get_job_result(job_id: str, engine: str) -> tuple[str, int]: - """Get the job status and time elapsed using the given job engine.""" - handler = _get_job_handler(engine) +def get_job_result(job_id: str, runner_name: str) -> tuple[str, int]: + """Get the job status and time elapsed from the given runner.""" + handler = _get_job_handler(get_runner(runner_name)) return handler.get_job_result(job_id) diff --git a/airflow_src/plugins/sensors/ssh_sensor.py b/airflow_src/plugins/sensors/ssh_sensor.py index ef3bdaf9..72285250 100644 --- a/airflow_src/plugins/sensors/ssh_sensor.py +++ b/airflow_src/plugins/sensors/ssh_sensor.py @@ -38,7 +38,7 @@ def __init__( self.xcom_source_task_id: str = xcom_source_task_id self.quanting_env_source_task_id: str = quanting_env_source_task_id self._job_id: str | None = None - self._engine: str | None = None + self._runner_name: str | None = None def pre_execute(self, context: dict[str, Any]) -> None: """Persist the job id and job engine from XCom.""" @@ -55,13 +55,13 @@ def pre_execute(self, context: dict[str, Any]) -> None: task_ids=self.quanting_env_source_task_id, map_indexes=ti.map_index, ) - self._engine = QuantingEnv.from_dict(quanting_env_dict).job_engine + self._runner_name = QuantingEnv.from_dict(quanting_env_dict).runner def poke(self, context: dict[str, Any]) -> bool: """Check the output of the ssh command.""" del context # unused - job_status = get_job_status(self._job_id, self._engine) + job_status = get_job_status(self._job_id, self._runner_name) logging.info(f"job_status: '{job_status}'") return job_status not in self.states diff --git a/airflow_src/tests/conftest.py b/airflow_src/tests/conftest.py index 5ef362a1..1a6e13aa 100644 --- a/airflow_src/tests/conftest.py +++ b/airflow_src/tests/conftest.py @@ -32,7 +32,7 @@ "settings_version": 1, "relative_raw_file_path": "instrument1/1970_01/test_file.raw", "config_params": "", - "job_engine": "slurm", + "runner": "slurm", "year_month_folder": "1970_01", } diff --git a/airflow_src/tests/dags/impl/test_processor_impl.py b/airflow_src/tests/dags/impl/test_processor_impl.py index ba69e660..730449a5 100644 --- a/airflow_src/tests/dags/impl/test_processor_impl.py +++ b/airflow_src/tests/dags/impl/test_processor_impl.py @@ -105,7 +105,7 @@ def test_create_quanting_env( "PROJECT_ID": "some_project_id", "SETTINGS_NAME": "test_settings", "SETTINGS_VERSION": 1, - "_JOB_ENGINE": "slurm", + "_RUNNER": "slurm", "_YEAR_MONTH_FOLDER": "1970_01", "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw", "_CONFIG_PARAMS": "", @@ -188,7 +188,7 @@ def test_create_quanting_env_custom_software( "PROJECT_ID": "some_project_id", "SETTINGS_NAME": "test_custom_settings", "SETTINGS_VERSION": 1, - "_JOB_ENGINE": "slurm", + "_RUNNER": "slurm", "_YEAR_MONTH_FOLDER": "1970_01", "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw", "_CONFIG_PARAMS": expected_config_params, @@ -387,7 +387,7 @@ def test_check_content_allows_image_name_in_software_field( ) -> None: """Test that a docker image name in the software field is accepted.""" quanting_env = make_quanting_env( - software="alphakraken-msqc", job_engine=JobEngines.DOCKER + software="alphakraken-msqc", runner=JobEngines.DOCKER ) errors = _check_content(quanting_env, MagicMock(config_params=None)) @@ -562,7 +562,7 @@ def test_submit_job_executes_ssh_command_and_stores_job_id( assert output_dir.exists() mock_start_job.assert_called_once_with( quanting_env, - engine="slurm", + runner_name="slurm", ) mock_get_raw_file_by_id.assert_called_once_with("test_file.raw") mock_update.assert_called_once_with( diff --git a/airflow_src/tests/plugins/jobs/test_job_handler.py b/airflow_src/tests/plugins/jobs/test_job_handler.py index 2e25ac64..2d92ac07 100644 --- a/airflow_src/tests/plugins/jobs/test_job_handler.py +++ b/airflow_src/tests/plugins/jobs/test_job_handler.py @@ -2,18 +2,21 @@ import importlib.util import sys +from collections.abc import Callable from pathlib import Path from unittest.mock import MagicMock, patch import pytest from airflow.exceptions import AirflowFailException +from common.quanting_env import QuantingEnv from jobs._experimental.file_based_job_handler import FileBasedJobHandler -from jobs.job_handler import _get_job_handler +from jobs.job_handler import _get_job_handler, start_job from jobs.slurm_ssh_job_handler import SlurmSSHJobHandler from airflow_src.tests.helpers import yaml_locations from shared.keys import JobEngines from shared.path_views import DOCKER_HOST_VIEW +from shared.runners import Runner, get_runner # `docker` is an optional dependency, cf. requirements_docker_job_engine.txt HAS_DOCKER = importlib.util.find_spec("docker") is not None @@ -24,14 +27,18 @@ @yaml_locations(slurm=str(SLURM_BASE_DIR)) def test_get_job_handler_routes_engine_to_handler() -> None: """Test that the factory returns the handler matching the requested engine.""" - assert isinstance(_get_job_handler(JobEngines.SLURM), SlurmSSHJobHandler) - assert isinstance(_get_job_handler(JobEngines.FILE_BASED), FileBasedJobHandler) + assert isinstance( + _get_job_handler(get_runner(JobEngines.SLURM)), SlurmSSHJobHandler + ) + assert isinstance( + _get_job_handler(get_runner(JobEngines.FILE_BASED)), FileBasedJobHandler + ) @yaml_locations(slurm=str(SLURM_BASE_DIR)) def test_get_job_handler_injects_slurm_base_dir() -> None: """Test that the factory reads the base dir and hands it to the Slurm handler.""" - handler = _get_job_handler(JobEngines.SLURM) + handler = _get_job_handler(get_runner(JobEngines.SLURM)) assert handler._cluster_base_dir == SLURM_BASE_DIR @@ -42,7 +49,7 @@ def test_get_job_handler_injects_docker_host_view( mock_from_env: MagicMock, # noqa: ARG001 ) -> None: """Test that the factory hands the docker host view to the docker handler.""" - handler = _get_job_handler(JobEngines.DOCKER) + handler = _get_job_handler(get_runner(JobEngines.DOCKER)) assert handler._docker_host_view is DOCKER_HOST_VIEW @@ -57,7 +64,7 @@ def test_get_job_handler_docker_without_mounts_env( patch.object(DOCKER_HOST_VIEW, "_locations", {}), pytest.raises(AirflowFailException, match="MOUNTS_PATH"), ): - _get_job_handler(JobEngines.DOCKER) + _get_job_handler(get_runner(JobEngines.DOCKER)) def test_get_job_handler_docker_without_optional_dependency() -> None: @@ -67,4 +74,26 @@ def test_get_job_handler_docker_without_optional_dependency() -> None: patch.dict(sys.modules, {"docker": None, "jobs.docker_job_handler": None}), pytest.raises(AirflowFailException, match="requirements_docker_job_engine"), ): - _get_job_handler(JobEngines.DOCKER) + _get_job_handler(get_runner(JobEngines.DOCKER)) + + +def test_get_job_handler_rejects_unknown_engine() -> None: + """Test that a runner with an engine the factory has no branch for is rejected.""" + runner = Runner( + name="k8s", + engine="kubernetes", + os="linux", + view=MagicMock(), + ssh_connection_id_prefix=None, + ) + + with pytest.raises(ValueError, match="kubernetes"): + _get_job_handler(runner) + + +def test_start_job_rejects_undeclared_runner( + make_quanting_env: Callable[..., QuantingEnv], +) -> None: + """Test that a runner name not in the yaml fails listing the declared runners.""" + with pytest.raises(KeyError, match=r"'nope'.*\['slurm', 'docker', 'file_based'\]"): + start_job(make_quanting_env(), runner_name="nope") diff --git a/airflow_src/tests/plugins/sensors/test_ssh_sensor.py b/airflow_src/tests/plugins/sensors/test_ssh_sensor.py index 10112308..29eae80d 100644 --- a/airflow_src/tests/plugins/sensors/test_ssh_sensor.py +++ b/airflow_src/tests/plugins/sensors/test_ssh_sensor.py @@ -11,7 +11,7 @@ JOB_ID_SOURCE_TASK_ID = "processing.submit_job" QUANTING_ENV_SOURCE_TASK_ID = "processing.prepare_job" -ENGINE = "file_based" +RUNNER_NAME = "file_based" @patch("plugins.sensors.ssh_sensor.get_job_status") @@ -25,7 +25,7 @@ def test_poke_executes_ssh_command_and_checks_returned_state( mock_ti.map_index = 0 mock_ti.xcom_pull.side_effect = [ "12345", - make_quanting_env(job_engine=ENGINE).to_dict(), + make_quanting_env(runner=RUNNER_NAME).to_dict(), ] mock_get_job_status.return_value = JobStates.RUNNING context = {"ti": mock_ti} @@ -49,7 +49,7 @@ def test_poke_executes_ssh_command_and_checks_returned_state( ] ) assert not operator.poke(context) - mock_get_job_status.assert_called_once_with("12345", ENGINE) + mock_get_job_status.assert_called_once_with("12345", RUNNER_NAME) @pytest.mark.parametrize( @@ -79,7 +79,7 @@ def test_poke_returns_true_when_state_not_in_running_states( mock_ti.map_index = 2 mock_ti.xcom_pull.side_effect = [ "12345", - make_quanting_env(job_engine=ENGINE).to_dict(), + make_quanting_env(runner=RUNNER_NAME).to_dict(), ] mock_get_job_status.return_value = job_status context = {"ti": mock_ti} @@ -103,4 +103,4 @@ def test_poke_returns_true_when_state_not_in_running_states( ] ) assert operator.poke(context) is expected_poke - mock_get_job_status.assert_called_once_with("12345", ENGINE) + mock_get_job_status.assert_called_once_with("12345", RUNNER_NAME) diff --git a/tasks/todo.md b/tasks/todo.md index 3fd86e2d..050eda4d 100644 --- a/tasks/todo.md +++ b/tasks/todo.md @@ -115,14 +115,14 @@ message). `runner` joins the strict list of `_check_content`. Slurm branch still `CLUSTER_VIEW` for its base dir (T8). Spec 2.3, 2.5 factory signature, 7.5 partial, 9.5. **Acceptance criteria:** -- [ ] `QuantingEnv.to_dict()` differs from before only in `_JOB_ENGINE` -> `_RUNNER` (existing full-dict assertion in `test_create_quanting_env`, spec 7.3). -- [ ] Factory per engine with a `Runner`; unknown engine raises `ValueError`. -- [ ] `start_job(env, runner_name="nope")` raises `KeyError` listing declared runners (9.5). -- [ ] `grep -rn job_engine airflow_src --include='*.py'` hits only the transitional line. +- [x] `QuantingEnv.to_dict()` differs from before only in `_JOB_ENGINE` -> `_RUNNER` (existing full-dict assertion in `test_create_quanting_env`, spec 7.3). +- [x] Factory per engine with a `Runner`; unknown engine raises `ValueError`. +- [x] `start_job(env, runner_name="nope")` raises `KeyError` listing declared runners (9.5). +- [x] `grep -rn job_engine airflow_src --include='*.py'` hits only the transitional line. **Verification:** -- [ ] `pytest airflow_src/tests/plugins/jobs/test_job_handler.py airflow_src/tests/plugins/sensors airflow_src/tests/dags/impl/test_processor_impl.py` -- [ ] `pytest airflow_src` +- [x] `pytest airflow_src/tests/plugins/jobs/test_job_handler.py airflow_src/tests/plugins/sensors airflow_src/tests/dags/impl/test_processor_impl.py` +- [x] `pytest airflow_src` **Dependencies:** T3, T4 **Files:** `airflow_src/plugins/common/quanting_env.py`, `airflow_src/plugins/jobs/job_handler.py`, `airflow_src/plugins/sensors/ssh_sensor.py`, `airflow_src/dags/impl/processor_impl.py`, tests: `conftest.py`, `test_job_handler.py`, `test_ssh_sensor.py`, `test_processor_impl.py` From 7549add3f0a73e2cf40e2e1f21a93a67ef1f6b13 Mon Sep 17 00:00:00 2001 From: mschwoerer <82171591+mschwoer@users.noreply.github.com> Date: Fri, 4 Sep 2026 11:27:39 +0200 Subject: [PATCH 2/2] rename QuantingEnv.runner to runner_name, alias _RUNNER_NAME Co-Authored-By: Claude Fable 5.1 --- SPEC.md | 6 +++--- airflow_src/dags/impl/processor_impl.py | 10 ++++++---- airflow_src/plugins/common/quanting_env.py | 2 +- airflow_src/plugins/sensors/ssh_sensor.py | 2 +- airflow_src/tests/conftest.py | 2 +- .../tests/dags/impl/test_processor_impl.py | 6 +++--- .../tests/plugins/sensors/test_ssh_sensor.py | 4 ++-- tasks/todo.md | 16 ++++++++-------- 8 files changed, 25 insertions(+), 23 deletions(-) diff --git a/SPEC.md b/SPEC.md index bf5bdc4f..feafa4e1 100644 --- a/SPEC.md +++ b/SPEC.md @@ -182,7 +182,7 @@ def get_runner(name: str) -> Runner: - `Settings.runner = StringField(required=True, max_length=64)`, no default. `job_engine` is deleted. `create_settings(runner=...)`. -- `QuantingEnv.job_engine` -> `runner: str = Field(alias="_RUNNER")`. Underscore prefix, so it +- `QuantingEnv.job_engine` -> `runner_name: str = Field(alias="_RUNNER_NAME")`. Underscore prefix, so it is never exported to a job. ### 2.4 prepare_job @@ -195,7 +195,7 @@ resolved absolute strings change flavour. ### 2.5 Handler factory and SSH - `_get_job_handler(runner: Runner)`. `start_job/get_job_status/get_job_result(..., runner_name)` - look the runner up. `ssh_sensor.py:58` reads `.runner`. + look the runner up. `ssh_sensor.py:58` reads `.runner_name`. - Slurm: `SlurmSSHJobHandler(runner.view.resolve(Locations.SLURM), runner.ssh_connection_id_prefix)`. The factory raises `AirflowFailException` naming the runner if the prefix is `None`. - Docker: unchanged, `DockerJobHandler(DOCKER_HOST_VIEW)`. The host view is a property of the @@ -392,7 +392,7 @@ pytest, tests next to the existing ones (`shared/tests`, `airflow_src/tests`, `w substituted `_CONFIG_PARAMS`, and `_check_content` returns no errors. `_check_content` still rejects `..`, `;`, `$` in relative paths, file names, `software`, `config_params`. - 7.3 Regression: for a `slurm` runner whose `view` equals the former `absolute_path` - values, `QuantingEnv.to_dict()` is byte-identical to before, except `_JOB_ENGINE` -> `_RUNNER`. + values, `QuantingEnv.to_dict()` is byte-identical to before, except `_JOB_ENGINE` -> `_RUNNER_NAME`. `get_backup_base_path` yields the same string as before for the same yaml values. - 7.4 `test_utils.py`: two prefixes select disjoint connection sets. - 7.5 `test_job_handler.py`: factory per engine with a `Runner`; unknown engine raises; a diff --git a/airflow_src/dags/impl/processor_impl.py b/airflow_src/dags/impl/processor_impl.py index 3b3f5ba5..60e4b927 100644 --- a/airflow_src/dags/impl/processor_impl.py +++ b/airflow_src/dags/impl/processor_impl.py @@ -226,7 +226,7 @@ def _create_quanting_env( relative_raw_file_path=str(relative_raw_file_path), config_params=substituted_params, # transitional until Settings.runner exists: every in-repo runner is named after its engine - runner=settings.job_engine, + runner_name=settings.job_engine, year_month_folder=get_created_at_year_month(raw_file), ) @@ -285,7 +285,7 @@ def _prepare_custom_command(settings: Settings, substituted_params: str) -> str: "settings_name", "year_month_folder", "slurm_mem", - "runner", + "runner_name", ) # composed of a yaml base path (admin configuration) and fields checked above, e.g. @@ -392,7 +392,7 @@ def submit_job( output_path.mkdir(parents=True, exist_ok=True) - job_id = start_job(quanting_env, runner_name=quanting_env.runner) + job_id = start_job(quanting_env, runner_name=quanting_env.runner_name) # TODO: race condition here, e.g. for file in ERROR status update_raw_file(quanting_env.raw_file_id, new_status=RawFileStatus.QUANTING) @@ -466,7 +466,9 @@ def check_job_result(*, quanting_env_dict: dict, job_id: str, ti: TaskInstance) """ quanting_env = QuantingEnv.from_dict(quanting_env_dict) - job_status, time_elapsed = get_job_result(job_id, runner_name=quanting_env.runner) + job_status, time_elapsed = get_job_result( + job_id, runner_name=quanting_env.runner_name + ) logging.info(f"Job {job_id} exited with status {job_status}.") diff --git a/airflow_src/plugins/common/quanting_env.py b/airflow_src/plugins/common/quanting_env.py index 8e186719..b4008923 100644 --- a/airflow_src/plugins/common/quanting_env.py +++ b/airflow_src/plugins/common/quanting_env.py @@ -42,7 +42,7 @@ class QuantingEnv(BaseModel): relative_raw_file_path: str = Field(alias="_RELATIVE_RAW_FILE_PATH") config_params: str = Field(alias="_CONFIG_PARAMS") - runner: str = Field(alias="_RUNNER") + runner_name: str = Field(alias="_RUNNER_NAME") year_month_folder: str = Field(alias="_YEAR_MONTH_FOLDER") def to_dict(self) -> dict: diff --git a/airflow_src/plugins/sensors/ssh_sensor.py b/airflow_src/plugins/sensors/ssh_sensor.py index 72285250..cbec50b0 100644 --- a/airflow_src/plugins/sensors/ssh_sensor.py +++ b/airflow_src/plugins/sensors/ssh_sensor.py @@ -55,7 +55,7 @@ def pre_execute(self, context: dict[str, Any]) -> None: task_ids=self.quanting_env_source_task_id, map_indexes=ti.map_index, ) - self._runner_name = QuantingEnv.from_dict(quanting_env_dict).runner + self._runner_name = QuantingEnv.from_dict(quanting_env_dict).runner_name def poke(self, context: dict[str, Any]) -> bool: """Check the output of the ssh command.""" diff --git a/airflow_src/tests/conftest.py b/airflow_src/tests/conftest.py index 1a6e13aa..82ccfd36 100644 --- a/airflow_src/tests/conftest.py +++ b/airflow_src/tests/conftest.py @@ -32,7 +32,7 @@ "settings_version": 1, "relative_raw_file_path": "instrument1/1970_01/test_file.raw", "config_params": "", - "runner": "slurm", + "runner_name": "slurm", "year_month_folder": "1970_01", } diff --git a/airflow_src/tests/dags/impl/test_processor_impl.py b/airflow_src/tests/dags/impl/test_processor_impl.py index 730449a5..fdc3284d 100644 --- a/airflow_src/tests/dags/impl/test_processor_impl.py +++ b/airflow_src/tests/dags/impl/test_processor_impl.py @@ -105,7 +105,7 @@ def test_create_quanting_env( "PROJECT_ID": "some_project_id", "SETTINGS_NAME": "test_settings", "SETTINGS_VERSION": 1, - "_RUNNER": "slurm", + "_RUNNER_NAME": "slurm", "_YEAR_MONTH_FOLDER": "1970_01", "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw", "_CONFIG_PARAMS": "", @@ -188,7 +188,7 @@ def test_create_quanting_env_custom_software( "PROJECT_ID": "some_project_id", "SETTINGS_NAME": "test_custom_settings", "SETTINGS_VERSION": 1, - "_RUNNER": "slurm", + "_RUNNER_NAME": "slurm", "_YEAR_MONTH_FOLDER": "1970_01", "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw", "_CONFIG_PARAMS": expected_config_params, @@ -387,7 +387,7 @@ def test_check_content_allows_image_name_in_software_field( ) -> None: """Test that a docker image name in the software field is accepted.""" quanting_env = make_quanting_env( - software="alphakraken-msqc", runner=JobEngines.DOCKER + software="alphakraken-msqc", runner_name=JobEngines.DOCKER ) errors = _check_content(quanting_env, MagicMock(config_params=None)) diff --git a/airflow_src/tests/plugins/sensors/test_ssh_sensor.py b/airflow_src/tests/plugins/sensors/test_ssh_sensor.py index 29eae80d..54c36e70 100644 --- a/airflow_src/tests/plugins/sensors/test_ssh_sensor.py +++ b/airflow_src/tests/plugins/sensors/test_ssh_sensor.py @@ -25,7 +25,7 @@ def test_poke_executes_ssh_command_and_checks_returned_state( mock_ti.map_index = 0 mock_ti.xcom_pull.side_effect = [ "12345", - make_quanting_env(runner=RUNNER_NAME).to_dict(), + make_quanting_env(runner_name=RUNNER_NAME).to_dict(), ] mock_get_job_status.return_value = JobStates.RUNNING context = {"ti": mock_ti} @@ -79,7 +79,7 @@ def test_poke_returns_true_when_state_not_in_running_states( mock_ti.map_index = 2 mock_ti.xcom_pull.side_effect = [ "12345", - make_quanting_env(runner=RUNNER_NAME).to_dict(), + make_quanting_env(runner_name=RUNNER_NAME).to_dict(), ] mock_get_job_status.return_value = job_status context = {"ti": mock_ti} diff --git a/tasks/todo.md b/tasks/todo.md index 050eda4d..b4f05796 100644 --- a/tasks/todo.md +++ b/tasks/todo.md @@ -104,18 +104,18 @@ yet. Consistency test: every in-repo yaml declares `runners`, engine and os know ## Phase 2: Jobs are dispatched by runner -### Task 5: `QuantingEnv.runner`, factory takes a `Runner` +### Task 5: `QuantingEnv.runner_name`, factory takes a `Runner` -**Description:** `QuantingEnv.job_engine` -> `runner: str = Field(alias="_RUNNER")`. +**Description:** `QuantingEnv.job_engine` -> `runner_name: str = Field(alias="_RUNNER_NAME")`. `_get_job_handler(runner: Runner)` dispatches on `runner.engine`; `start_job/get_job_status/ -get_job_result(..., runner_name)` call `get_runner`. `ssh_sensor.py` reads `.runner`. -`processor_impl` passes `quanting_env.runner` to the job functions and, for this commit only, -sets `runner=settings.job_engine` (transitional line, removed in T6, named in the commit +get_job_result(..., runner_name)` call `get_runner`. `ssh_sensor.py` reads `.runner_name`. +`processor_impl` passes `quanting_env.runner_name` to the job functions and, for this commit only, +sets `runner_name=settings.job_engine` (transitional line, removed in T6, named in the commit message). `runner` joins the strict list of `_check_content`. Slurm branch still resolves `CLUSTER_VIEW` for its base dir (T8). Spec 2.3, 2.5 factory signature, 7.5 partial, 9.5. **Acceptance criteria:** -- [x] `QuantingEnv.to_dict()` differs from before only in `_JOB_ENGINE` -> `_RUNNER` (existing full-dict assertion in `test_create_quanting_env`, spec 7.3). +- [x] `QuantingEnv.to_dict()` differs from before only in `_JOB_ENGINE` -> `_RUNNER_NAME` (existing full-dict assertion in `test_create_quanting_env`, spec 7.3). - [x] Factory per engine with a `Runner`; unknown engine raises `ValueError`. - [x] `start_job(env, runner_name="nope")` raises `KeyError` listing declared runners (9.5). - [x] `grep -rn job_engine airflow_src --include='*.py'` hits only the transitional line. @@ -131,7 +131,7 @@ message). `runner` joins the strict list of `_check_content`. Slurm branch still ### Task 6: `Settings.runner` replaces `job_engine`; webapp selectbox **Description:** `Settings.runner = StringField(required=True, max_length=64)`, `job_engine` -deleted; `create_settings(runner=...)`. The T5 transitional line becomes `runner=settings.runner`. +deleted; `create_settings(runner=...)`. The T5 transitional line becomes `runner_name=settings.runner`. Webapp (spec 2.7): options `list(RUNNERS)`, default first declared, `SHOW_RUNNER_SELECT`, prefill key `runner`, docker-only-custom check via `RUNNERS[runner].engine`, help texts at 314 and 415 say "runner". Empty `RUNNERS` (spec 1.2.4): no settings form, a notice instead; covers the @@ -172,7 +172,7 @@ precondition. Spec 2.8, 9.7. ### Checkpoint 2: Runner flows - [ ] Three suites green, pre-commit clean; `grep -n "job_engine" airflow_src/dags/impl/processor_impl.py` empty. -- [ ] Local stack: create a settings entry in the webapp with runner `slurm`, trigger a quanting DAG with `debug_no_cluster_ssh=true`, the `prepare_job` XCom shows `_RUNNER: slurm` and the same paths as before. +- [ ] Local stack: create a settings entry in the webapp with runner `slurm`, trigger a quanting DAG with `debug_no_cluster_ssh=true`, the `prepare_job` XCom shows `_RUNNER_NAME: slurm` and the same paths as before. - [ ] Migration `--dry-run` runs against a sandbox DB copy; note the distinct names (open question 1). - [ ] Human review before Phase 3.