From 715810ab90a3b388e7ff2f244f84034e0e3dcb43 Mon Sep 17 00:00:00 2001 From: alexander Date: Tue, 3 Nov 2026 11:04:36 +0100 Subject: [PATCH 1/4] test: give each backend its own folder in the unit tests Moves test_fake_backend.py to tests/unit/backends/fake/, ready for the HTCondor tests that follow. The tests themselves do not change. Part of #23. --- tests/unit/backends/__init__.py | 0 tests/unit/backends/fake/__init__.py | 0 tests/unit/{ => backends/fake}/test_fake_backend.py | 0 3 files changed, 0 insertions(+), 0 deletions(-) create mode 100644 tests/unit/backends/__init__.py create mode 100644 tests/unit/backends/fake/__init__.py rename tests/unit/{ => backends/fake}/test_fake_backend.py (100%) diff --git a/tests/unit/backends/__init__.py b/tests/unit/backends/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/backends/fake/__init__.py b/tests/unit/backends/fake/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_fake_backend.py b/tests/unit/backends/fake/test_fake_backend.py similarity index 100% rename from tests/unit/test_fake_backend.py rename to tests/unit/backends/fake/test_fake_backend.py From 082e2f65bf976b724617e6eb21f5a3dcf5fb0e59 Mon Sep 17 00:00:00 2001 From: alexander Date: Tue, 3 Nov 2026 11:04:36 +0100 Subject: [PATCH 2/4] feat: HTCondor submit descriptions in the shared core Renders a submit description from a plan: one process per copy, per-copy values indexed by ProcId, quoting, and refusing what a description cannot hold. It sets no policy on the job except keeping finished jobs until they are purged, so output can be collected again. Shared with the HTCondor scheduler to come (#21), so it knows nothing about any backend. Expected values were checked against HTCondor 25. Part of #23. --- src/intercede/_backends/_htcondor/__init__.py | 11 + src/intercede/_backends/_htcondor/submit.py | 228 ++++++++++++++++++ tests/unit/backends/_htcondor/__init__.py | 0 tests/unit/backends/_htcondor/test_core.py | 192 +++++++++++++++ 4 files changed, 431 insertions(+) create mode 100644 src/intercede/_backends/_htcondor/__init__.py create mode 100644 src/intercede/_backends/_htcondor/submit.py create mode 100644 tests/unit/backends/_htcondor/__init__.py create mode 100644 tests/unit/backends/_htcondor/test_core.py diff --git a/src/intercede/_backends/_htcondor/__init__.py b/src/intercede/_backends/_htcondor/__init__.py new file mode 100644 index 0000000..743a1fd --- /dev/null +++ b/src/intercede/_backends/_htcondor/__init__.py @@ -0,0 +1,11 @@ +"""What interCEde knows about HTCondor, shared by every backend that drives it. + +Plain functions and data only, with no dependency on any backend class: the +HTCondor-CE backend uses them directly, and the HTCondor scheduler of the +composed backend (issue #21) will use the same ones over its transport. They +share these functions, never a subtype. + +- ``submit``: a submit description built from a ``SubmissionSpec`` +- ``commands``: the argv of each HTCondor tool, and parsers for their output +- ``status``: the state mapping from job ads to ``JobStatus`` +""" diff --git a/src/intercede/_backends/_htcondor/submit.py b/src/intercede/_backends/_htcondor/submit.py new file mode 100644 index 0000000..bfd1474 --- /dev/null +++ b/src/intercede/_backends/_htcondor/submit.py @@ -0,0 +1,228 @@ +"""Submit descriptions: a ``SubmissionSpec`` written in HTCondor's submit language. + +One description submits every copy of a spec as one cluster, one process per +copy. What differs between copies (environment, secrets, working directory) is +expanded per process through macros indexed by ``$(ProcId)``, so the copies +stay one bulk submission. + +A description only says what to run and where its files go. It sets no +policy on the job's lifetime: no holds, no automatic removal, no retries. +Those are the consumer's decisions. The one exception is that a finished job +stays in the queue until it is purged, so its output can be collected again. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field + + +def quote_arguments(arguments: Sequence[str]) -> str: + """Quote command arguments using HTCondor's new syntax.""" + quoted_arguments = [_quote_item(argument) for argument in arguments] + + joined_arguments = " ".join(quoted_arguments) + return f'"{joined_arguments}"' + + +def quote_environment(environment: Mapping[str, str]) -> str: + """Quote environment variables using HTCondor's new syntax.""" + for name in environment: + _validate_environment_name(name) + + quoted_variables = [ + f"{name}={_quote_item(value)}" for name, value in sorted(environment.items()) + ] + + joined_variables = " ".join(quoted_variables) + return f'"{joined_variables}"' + + +def _validate_environment_name(name: str) -> None: + """Raise ValueError if the environment variable name is invalid.""" + is_valid = bool(name) and "=" not in name and name.replace("_", "").isalnum() + + if not is_valid: + raise ValueError(f"{name!r} is not a valid environment variable name") + + +def _quote_item(value: str) -> str: + """Escape and quote a single argument or environment variable value.""" + _refuse_newlines(value) + + escaped_value = value.replace('"', '""') + escaped_value = escaped_value.replace("'", "''") + escaped_value = _escape_macros(escaped_value) + + return f"'{escaped_value}'" + + +def _escape_macros(value: str) -> str: + # "$(...)" is macro syntax in a submit description; $(DOLLAR) is a literal "$". + return value.replace("$", "$(DOLLAR)") + + +def _refuse_newlines(value: str) -> None: + if "\n" in value or "\r" in value: + raise ValueError("a submit description value cannot contain a line break") + + +@dataclass(frozen=True) +class CopyPlan: + """What one process of the cluster gets on its own.""" + + initialdir: str # absolute; outputs come back here (SUBMIT_Iwd) + environment: Mapping[str, str] = field(default_factory=dict) + # Inputs of this copy only, relative to ``initialdir`` (e.g. "secrets/"). + inputs: Sequence[str] = () + + +@dataclass(frozen=True) +class SubmitPlan: + """Everything a submit description says, resolved to HTCondor's terms.""" + + executable: str + transfer_executable: bool + arguments: Sequence[str] + copies: Sequence[CopyPlan] + inputs: Sequence[str] = () # shared by every copy: absolute paths or URLs + stdout: str = "$(ClusterId).$(ProcId).out" + stderr: str = "$(ClusterId).$(ProcId).err" + log: str = "$(ClusterId).$(ProcId).log" + # None: every file the payload creates comes back; empty: none at all. + outputs: Sequence[str] | None = () + output_remaps: Mapping[str, str] = field(default_factory=dict) # name -> URL + cpus: int = 1 + memory_mb: int | None = None + wall_time_s: int | None = None + queue: str | None = None + proxy_file: str | None = None + + +def render(plan: SubmitPlan) -> str: + """Write ``plan`` as a submit description that queues one process per copy.""" + _validate_plan(plan) + + lines = _render_basics(plan) + lines.extend(_render_copies(plan)) + lines.extend(_render_file_transfer(plan)) + lines.extend(_render_resources(plan)) + lines.extend(_render_proxy(plan)) + lines.extend(_render_queue(plan)) + + return "\n".join(lines) + "\n" + + +def _join(items: Sequence[str]) -> str: + for item in items: + _refuse_newlines(item) + if "," in item: + raise ValueError(f"{item!r}: HTCondor file lists cannot hold a comma") + return ", ".join(_escape_macros(item) for item in items) + + +def _bool(value: bool) -> str: + return "true" if value else "false" + + +def _classad_string(value: str) -> str: + _refuse_newlines(value) + escaped = value.replace("\\", "\\\\").replace('"', '\\"') + return '"' + _escape_macros(escaped) + '"' + + +def _validate_plan(plan: SubmitPlan) -> None: + if not plan.copies: + raise ValueError("A submit description needs at least one copy") + + +def _render_basics(plan: SubmitPlan) -> list[str]: + lines = [ + "universe = vanilla", + f"executable = {_escape_macros(plan.executable)}", + f"transfer_executable = {_bool(plan.transfer_executable)}", + ] + + if plan.arguments: + lines.append(f"arguments = {quote_arguments(plan.arguments)}") + + return lines + + +def _render_copies(plan: SubmitPlan) -> list[str]: + lines = [] + + for index, copy in enumerate(plan.copies): + _refuse_newlines(copy.initialdir) + lines.extend( + [ + f"ic_dir_{index} = {_escape_macros(copy.initialdir)}", + f"ic_env_{index} = {quote_environment(copy.environment)}", + f"ic_in_{index} = {_join(copy.inputs)}", + ] + ) + + lines.extend( + ["initialdir = $(ic_dir_$(ProcId))", "environment = $(ic_env_$(ProcId))"] + ) + + return lines + + +def _render_file_transfer(plan: SubmitPlan) -> list[str]: + shared_inputs = _join(plan.inputs) + copy_inputs = "$(ic_in_$(ProcId))" + inputs = f"{shared_inputs}, {copy_inputs}" if shared_inputs else copy_inputs + + lines = [ + "should_transfer_files = YES", + "when_to_transfer_output = ON_EXIT_OR_EVICT", + f"transfer_input_files = {inputs}", + f"output = {plan.stdout}", + f"error = {plan.stderr}", + f"log = {plan.log}", + ] + + if plan.outputs is not None: + output_files = _join(plan.outputs) or '""' + lines.append(f"transfer_output_files = {output_files}") + + if plan.output_remaps: + remaps = "; ".join( + f"{name} = {url}" for name, url in sorted(plan.output_remaps.items()) + ) + lines.append(f'transfer_output_remaps = "{_escape_macros(remaps)}"') + + return lines + + +def _render_resources(plan: SubmitPlan) -> list[str]: + lines = [f"request_cpus = {plan.cpus}", f"+xcount = {plan.cpus}"] + + if plan.memory_mb is not None: + lines.extend( + [f"request_memory = {plan.memory_mb}", f"+maxMemory = {plan.memory_mb}"] + ) + + if plan.wall_time_s is not None: + lines.append(f"+BatchRuntime = {plan.wall_time_s}") + + if plan.queue is not None: + lines.append(f"+queue = {_classad_string(plan.queue)}") + + return lines + + +def _render_proxy(plan: SubmitPlan) -> list[str]: + if plan.proxy_file is None: + return [] + + return [f"x509userproxy = {_escape_macros(plan.proxy_file)}"] + + +def _render_queue(plan: SubmitPlan) -> list[str]: + return [ + # Keep completed jobs until purge() releases them + "leave_in_queue = (JobStatus == 4)", + f"queue {len(plan.copies)}", + ] diff --git a/tests/unit/backends/_htcondor/__init__.py b/tests/unit/backends/_htcondor/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/backends/_htcondor/test_core.py b/tests/unit/backends/_htcondor/test_core.py new file mode 100644 index 0000000..32afb19 --- /dev/null +++ b/tests/unit/backends/_htcondor/test_core.py @@ -0,0 +1,192 @@ +"""The shared HTCondor knowledge: submit descriptions, commands and the state mapping. + +Expected values were checked against HTCondor 25 (condor_submit -dry-run and a +live schedd): what is asserted here is what HTCondor actually does with them. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from intercede._backends._htcondor import submit +from intercede._backends._htcondor.submit import CopyPlan, SubmitPlan, render + +# -------------------------------------------------------------------------- +# Quoting +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("arguments", "quoted"), + [ + (["plain"], "\"'plain'\""), + (["two words"], "\"'two words'\""), + (["it's"], "\"'it''s'\""), + (['say "hi"'], '"\'say ""hi""\'"'), + (["cost $(HOME) $5"], "\"'cost $(DOLLAR)(HOME) $(DOLLAR)5'\""), + ([""], "\"''\""), + (["a", "b"], "\"'a' 'b'\""), + ], + ids=[ + "plain", + "space", + "single-quote", + "double-quote", + "dollar", + "empty", + "several", + ], +) +def test_quote_arguments(arguments: list[str], quoted: str): + assert submit.quote_arguments(arguments) == quoted + + +def test_quote_environment_sorts_and_quotes(): + quoted = submit.quote_environment({"STAMP": "a b", "X": "$(Y)", "A": "it's"}) + assert quoted == "\"A='it''s' STAMP='a b' X='$(DOLLAR)(Y)'\"" + + +@pytest.mark.parametrize("name", ["", "A=B", "WITH SPACE", "DASH-ED", "$X"]) +def test_environment_names_must_be_identifiers(name: str): + with pytest.raises(ValueError, match="not a valid environment variable name"): + submit.quote_environment({name: "x"}) + + +@pytest.mark.parametrize("value", ["line\nbreak", "carriage\rreturn"]) +def test_line_breaks_are_refused(value: str): + with pytest.raises(ValueError, match="line break"): + submit.quote_arguments([value]) + with pytest.raises(ValueError, match="line break"): + submit.quote_environment({"X": value}) + + +# -------------------------------------------------------------------------- +# Submit descriptions +# -------------------------------------------------------------------------- + + +def plan(**changes: Any) -> SubmitPlan: + values: dict[str, Any] = { + "executable": "/work/payload.sh", + "transfer_executable": True, + "arguments": ["a"], + "copies": [ + CopyPlan("/work/0", {"STAMP": "s0"}, ["secrets/"]), + CopyPlan("/work/1", {"STAMP": "s1"}), + ], + } + return SubmitPlan(**{**values, **changes}) + + +def lines(text: str) -> list[str]: + return text.splitlines() + + +class TestRender: + def test_one_process_per_copy(self): + assert lines(render(plan()))[-1] == "queue 2" + + def test_per_copy_values_are_indexed_by_proc_id(self): + text = render(plan()) + assert "ic_dir_0 = /work/0" in lines(text) + assert "ic_dir_1 = /work/1" in lines(text) + assert "ic_env_0 = \"STAMP='s0'\"" in lines(text) + assert "ic_in_0 = secrets/" in lines(text) + assert "ic_in_1 = " in lines(text) + assert "initialdir = $(ic_dir_$(ProcId))" in lines(text) + assert "environment = $(ic_env_$(ProcId))" in lines(text) + + def test_inputs_shared_and_per_copy(self): + text = render(plan(inputs=["/work/inputs/", "https://example.org/in.dat"])) + assert ( + "transfer_input_files = /work/inputs/, https://example.org/in.dat, $(ic_in_$(ProcId))" + in lines(text) + ) + + @pytest.mark.parametrize( + ("outputs", "expected"), + [ + ( + [], + 'transfer_output_files = ""', + ), # nothing but stdout, stderr and the log + (["result.txt", "dir/x"], "transfer_output_files = result.txt, dir/x"), + (None, None), # every file the payload creates + ], + ids=["none", "listed", "all"], + ) + def test_outputs(self, outputs: list[str] | None, expected: str | None): + text = render(plan(outputs=outputs)) + found = [ + line for line in lines(text) if line.startswith("transfer_output_files") + ] + assert found == ([expected] if expected else []) + + def test_output_remaps_upload_to_urls(self): + text = render( + plan( + outputs=["a", "b"], + output_remaps={"b": "https://s/b", "a": "https://s/a"}, + ) + ) + assert 'transfer_output_remaps = "a = https://s/a; b = https://s/b"' in lines( + text + ) + + def test_resources_for_a_schedd_and_for_a_ce_router(self): + text = render(plan(cpus=8, memory_mb=4000, wall_time_s=7200, queue="long")) + for line in [ + "request_cpus = 8", + "+xcount = 8", + "request_memory = 4000", + "+maxMemory = 4000", + "+BatchRuntime = 7200", + '+queue = "long"', + ]: + assert line in lines(text) + + def test_proxy(self): + assert "x509userproxy = /creds/proxy" in lines( + render(plan(proxy_file="/creds/proxy")) + ) + assert not any( + line.startswith("x509userproxy") for line in lines(render(plan())) + ) + + def test_finished_jobs_stay_until_purged(self): + # The one lifetime rule interCEde sets: output can be collected + # again until the consumer releases the job. + assert "leave_in_queue = (JobStatus == 4)" in lines(render(plan())) + + @pytest.mark.parametrize( + "command", + [ + "on_exit_hold", + "on_exit_remove", + "periodic_hold", + "periodic_release", + "periodic_remove", + "max_retries", + "requirements", + ], + ) + def test_sets_no_policy_on_the_job(self, command: str): + # Holding, releasing, removing or retrying jobs is the consumer's + # policy, never interCEde's. + assert not any(line.startswith(command) for line in lines(render(plan()))) + + def test_macros_in_paths_are_escaped(self): + assert "ic_dir_0 = /work/$(DOLLAR)0" in lines( + render(plan(copies=[CopyPlan("/work/$0")])) + ) + + @pytest.mark.parametrize("field", ["inputs", "outputs"]) + def test_file_lists_cannot_hold_a_comma(self, field: str): + with pytest.raises(ValueError, match="comma"): + render(plan(**{field: ["a,b"]})) + + def test_paths_cannot_hold_a_line_break(self): + with pytest.raises(ValueError, match="line break"): + render(plan(copies=[CopyPlan("/work/a\nqueue 100")])) From 8c5128a0961805b1ec9e3f4c34a758f335a38a01 Mon Sep 17 00:00:00 2001 From: alexander Date: Tue, 3 Nov 2026 11:04:36 +0100 Subject: [PATCH 3/4] feat: HTCondor commands and the state map in the shared core Builds the tools' command lines and reads what they print: condor_submit -terse, condor_rm, and the job ads of condor_q and condor_history, mapped to the common JobStatus. A finished job's result comes from its exit, a removed job keeps the result it had, and any hold that interCEde did not cause counts as aborted. Part of #23. --- src/intercede/_backends/_htcondor/commands.py | 117 ++++++++++++ src/intercede/_backends/_htcondor/status.py | 124 ++++++++++++ tests/unit/backends/_htcondor/test_core.py | 178 +++++++++++++++++- 3 files changed, 418 insertions(+), 1 deletion(-) create mode 100644 src/intercede/_backends/_htcondor/commands.py create mode 100644 src/intercede/_backends/_htcondor/status.py diff --git a/src/intercede/_backends/_htcondor/commands.py b/src/intercede/_backends/_htcondor/commands.py new file mode 100644 index 0000000..8c76276 --- /dev/null +++ b/src/intercede/_backends/_htcondor/commands.py @@ -0,0 +1,117 @@ +"""The argv of each HTCondor tool, and parsers for what they print. + +``remote`` is how a tool reaches its schedd: ``schedd_options(pool, name)`` +for a remote schedd, as an HTCondor-CE is reached, or ``()`` for the schedd +of the host the tool runs on, as the composed backend reaches it. +""" + +from __future__ import annotations + +import re +from collections.abc import Sequence + +from intercede.models import JobID + +# Ids per command line, so a status poll of thousands of jobs stays bounded. +CHUNK = 100 + + +def schedd_options(pool: str, name: str) -> list[str]: + """Reach the schedd ``name`` through the collector ``pool`` (host:port).""" + return ["-pool", pool, "-name", name] + + +def submit( + description: str, pool: str | None = None, name: str | None = None +) -> list[str]: + """Submit a description file and spool its inputs to the schedd. + + Spooling copies the inputs at submit time, so nothing on the submitting + host is needed afterwards. With ``pool`` and ``name`` the schedd is a + remote one (``-remote`` implies spooling); without, it is the local one. + """ + if (pool is None) != (name is None): + raise ValueError("a remote schedd needs both pool and name") + + command = ["condor_submit", "-terse"] + + if pool is None or name is None: + command.append("-spool") + else: + command.extend(["-pool", pool, "-remote", name]) + + command.append(description) + return command + + +def query( + ids: Sequence[JobID], attributes: Sequence[str], remote: Sequence[str] +) -> list[str]: + """Read the ads of jobs still in the queue.""" + return ["condor_q", *remote, *ids, "-attributes", ",".join(attributes), "-json"] + + +def history( + ids: Sequence[JobID], attributes: Sequence[str], remote: Sequence[str] +) -> list[str]: + """Read the ads of jobs that have left the queue.""" + return [ + "condor_history", + *remote, + *ids, + "-attributes", + ",".join(attributes), + "-json", + ] + + +def remove(ids: Sequence[JobID], remote: Sequence[str]) -> list[str]: + """Remove jobs: cancels active ones, releases finished ones.""" + return ["condor_rm", *remote, *ids] + + +def transfer_data(ids: Sequence[JobID], remote: Sequence[str]) -> list[str]: + """Fetch the spooled output of jobs into the directory they were submitted from.""" + return ["condor_transfer_data", *remote, *ids] + + +def chunks(ids: Sequence[JobID]) -> list[Sequence[JobID]]: + """Split ids into groups of at most ``CHUNK``.""" + return [ids[start : start + CHUNK] for start in range(0, len(ids), CHUNK)] + + +_TERSE = re.compile( + r"\s*(?P\d+)\.(?P\d+)" + r"\s*-\s*" + r"(?P\d+)\.(?P\d+)\s*" +) + + +def parse_submit(raw: str) -> list[JobID]: + """Parse job IDs from condor_submit -terse output, e.g. '12.0 - 12.3'.""" + lines = [line for line in raw.splitlines() if line.strip()] + + if len(lines) != 1: + raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}") + + match = _TERSE.fullmatch(lines[0]) + if match is None: + raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}") + + cluster = int(match["cluster"]) + first = int(match["first"]) + last_cluster = int(match["last_cluster"]) + last = int(match["last"]) + + if cluster != last_cluster or last < first: + raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}") + + return [JobID(f"{cluster}.{proc}") for proc in range(first, last + 1)] + + +_REMOVED = re.compile(r"^Job (\d+\.\d+) marked for removal", re.MULTILINE) + + +def parse_remove(raw: str) -> set[JobID]: + """Return the ids ``condor_rm`` reports as removed; any other id was not.""" + return {JobID(job) for job in _REMOVED.findall(raw)} diff --git a/src/intercede/_backends/_htcondor/status.py b/src/intercede/_backends/_htcondor/status.py new file mode 100644 index 0000000..40d8aca --- /dev/null +++ b/src/intercede/_backends/_htcondor/status.py @@ -0,0 +1,124 @@ +"""From HTCondor job ads to ``JobStatus``. + +Reading only: the state is taken from what HTCondor records about a job, never +from a policy interCEde sets on it. A finished job reads as FAILED when its +payload exited non-zero or was killed by a signal, from ``ExitCode`` and +``ExitBySignal``. + +Ported from DIRAC's ``BatchSystems/Condor.py`` (``STATES_MAP`` and +``getCondorStatus``), with differences: DIRAC puts a payload that fails on +hold to recognise it, where the exit code is read here instead; the states +DIRAC leaves unmapped (6, 7) are mapped; and a removed job reports what it did +before it was removed, so a purged job keeps its result. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from typing import Any + +from intercede.models import JobID, JobStatus + +# The ad attributes every status query asks for. +STATE_ATTRIBUTES: Sequence[str] = ( + "ClusterId", + "ProcId", + "JobStatus", + "LastJobStatus", + "ExitCode", + "ExitBySignal", + "HoldReasonCode", + "HoldReason", +) + +# HTCondor's JobStatus codes. +_IDLE, _RUNNING, _REMOVED, _COMPLETED, _HELD, _TRANSFERRING, _SUSPENDED = range(1, 8) + +_STATES: Mapping[int, JobStatus] = { + _IDLE: JobStatus.WAITING, + _RUNNING: JobStatus.RUNNING, + _TRANSFERRING: JobStatus.RUNNING, # transferring output after running + _SUSPENDED: JobStatus.WAITING, +} + +_HOLD_SPOOLING = 16 # HoldReasonCode: input files are still being spooled + + +def job_id(ad: Mapping[str, Any]) -> JobID: + """Return the ``cluster.proc`` id of a job ad.""" + return JobID(f"{ad['ClusterId']}.{ad['ProcId']}") + + +def job_status(ad: Mapping[str, Any]) -> JobStatus: + """Map one job ad to the common status.""" + code = ad.get("JobStatus") + if code == _COMPLETED: + return _result(ad) + if code == _REMOVED: + # purge() removes finished jobs to release them, after which the + # history only says "removed": the state it left tells the result. + return ( + _result(ad) if ad.get("LastJobStatus") == _COMPLETED else JobStatus.ABORTED + ) + if code == _HELD: + # Held by the site or by HTCondor, not by interCEde: the job is stuck, + # except while its inputs are still being spooled. + return ( + JobStatus.WAITING + if ad.get("HoldReasonCode") == _HOLD_SPOOLING + else JobStatus.ABORTED + ) + return ( + _STATES.get(code, JobStatus.UNKNOWN) + if isinstance(code, int) + else JobStatus.UNKNOWN + ) + + +def is_removed(ad: Mapping[str, Any]) -> bool: + """Tell whether a job was removed, even while it is still in the queue. + + ``condor_rm`` only marks a job for removal: the schedd drops it from the + queue a moment later. In between it is already released. + + Args: + ad: A job ad with ``JobStatus``. + + Returns: + True when the job has been removed. + + """ + return ad.get("JobStatus") == _REMOVED + + +def _result(ad: Mapping[str, Any]) -> JobStatus: + """Tell DONE from FAILED for a job whose payload has finished.""" + if ad.get("ExitBySignal") is True: + return JobStatus.FAILED + exit_code = ad.get("ExitCode") + return ( + JobStatus.FAILED + if isinstance(exit_code, int) and exit_code != 0 + else JobStatus.DONE + ) + + +def hold_reason(ad: Mapping[str, Any]) -> str | None: + """Return why a held job is held, if it is.""" + reason = ad.get("HoldReason") + return reason if ad.get("JobStatus") == _HELD and isinstance(reason, str) else None + + +def parse_ads(raw: str) -> list[dict[str, Any]]: + """Parse ``condor_q -json`` or ``condor_history -json`` output. + + ``condor_q`` prints nothing at all when no job matches, and + ``condor_history`` prints an empty list. + """ + if not raw.strip(): + return [] + ads = json.loads(raw) + if not isinstance(ads, list) or not all(isinstance(ad, dict) for ad in ads): + raise ValueError("expected a JSON list of job ads") + return ads diff --git a/tests/unit/backends/_htcondor/test_core.py b/tests/unit/backends/_htcondor/test_core.py index 32afb19..76cc9ec 100644 --- a/tests/unit/backends/_htcondor/test_core.py +++ b/tests/unit/backends/_htcondor/test_core.py @@ -10,7 +10,8 @@ import pytest -from intercede._backends._htcondor import submit +from intercede import JobStatus +from intercede._backends._htcondor import commands, status, submit from intercede._backends._htcondor.submit import CopyPlan, SubmitPlan, render # -------------------------------------------------------------------------- @@ -190,3 +191,178 @@ def test_file_lists_cannot_hold_a_comma(self, field: str): def test_paths_cannot_hold_a_line_break(self): with pytest.raises(ValueError, match="line break"): render(plan(copies=[CopyPlan("/work/a\nqueue 100")])) + + +# -------------------------------------------------------------------------- +# Commands +# -------------------------------------------------------------------------- + + +class TestCommands: + def test_submit_to_the_local_schedd_spools(self): + assert commands.submit("/w/job.sub") == [ + "condor_submit", + "-terse", + "-spool", + "/w/job.sub", + ] + + @pytest.mark.parametrize( + ("count", "sizes"), + [(100, [100]), (101, [100, 1]), (250, [100, 100, 50])], + ) + def test_chunks(self, count: int, sizes: list[int]): + ids = [status.JobID(f"1.{i}") for i in range(count)] + chunked = commands.chunks(ids) + assert [len(chunk) for chunk in chunked] == sizes + assert [i for chunk in chunked for i in chunk] == ids + + +class TestParseSubmit: + @pytest.mark.parametrize( + ("raw", "ids"), + [ + ("1.0 - 1.1\n", ["1.0", "1.1"]), + (" 42.0 - 42.0 ", ["42.0"]), + ("\n7.3 - 7.5\n\n", ["7.3", "7.4", "7.5"]), + ], + ) + def test_terse_output(self, raw: str, ids: list[str]): + assert commands.parse_submit(raw) == ids + + @pytest.mark.parametrize( + "raw", + [ + "", + "Submitting job(s).\n1 job(s) submitted to cluster 1.", + "1.0 - 2.0", + "1.3 - 1.1", + "1.0 - 1.1\n2.0 - 2.1", + "x", + ], + ids=["empty", "verbose", "two-clusters", "backwards", "two-lines", "garbage"], + ) + def test_anything_else_is_refused(self, raw: str): + with pytest.raises(ValueError, match="unexpected condor_submit output"): + commands.parse_submit(raw) + + +def test_parse_remove(): + raw = "Job 1.0 marked for removal\nJob 1.2 marked for removal\nCouldn't find/remove all jobs matching 1.1\n" + assert commands.parse_remove(raw) == {"1.0", "1.2"} + assert commands.parse_remove("") == set() + + +# -------------------------------------------------------------------------- +# State mapping +# -------------------------------------------------------------------------- + + +def ad(**fields: Any) -> dict[str, Any]: + return {"ClusterId": 3, "ProcId": 1, **fields} + + +class TestJobStatus: + @pytest.mark.parametrize( + ("code", "expected"), + [ + (1, JobStatus.WAITING), + (2, JobStatus.RUNNING), + (4, JobStatus.DONE), + (6, JobStatus.RUNNING), + (7, JobStatus.WAITING), + ], + ids=["idle", "running", "completed", "transferring-output", "suspended"], + ) + def test_plain_states(self, code: int, expected: JobStatus): + assert status.job_status(ad(JobStatus=code)) == expected + + @pytest.mark.parametrize( + "code", + [8, None], + ids=["unexpected", "missing"], + ) + def test_unknown_codes(self, code: Any): + assert status.job_status(ad(JobStatus=code)) == JobStatus.UNKNOWN + + @pytest.mark.parametrize( + ("fields", "expected"), + [ + ({"ExitCode": 0, "ExitBySignal": False}, JobStatus.DONE), + ({"ExitCode": 3, "ExitBySignal": False}, JobStatus.FAILED), + ({"ExitBySignal": True, "ExitSignal": 9}, JobStatus.FAILED), + ({}, JobStatus.DONE), + ], + ids=["exit-0", "exit-3", "signal", "no-exit-info"], + ) + def test_finished_payload_result_comes_from_its_exit( + self, fields: dict[str, Any], expected: JobStatus + ): + # Read from what HTCondor records; no hold is put on the job to tell. + assert status.job_status(ad(JobStatus=4, **fields)) == expected + + def test_spooling_is_waiting(self): + assert ( + status.job_status(ad(JobStatus=5, HoldReasonCode=16)) == JobStatus.WAITING + ) + + @pytest.mark.parametrize( + "fields", + [{"HoldReasonCode": 1}, {"HoldReasonCode": 3, "HoldReasonSubCode": 55}, {}], + ids=["by-user", "by-site-policy", "no-reason"], + ) + def test_any_other_hold_is_aborted(self, fields: dict[str, Any]): + assert status.job_status(ad(JobStatus=5, **fields)) == JobStatus.ABORTED + + @pytest.mark.parametrize( + ("fields", "expected"), + [ + ({"LastJobStatus": 4, "ExitCode": 0}, JobStatus.DONE), + ({"LastJobStatus": 4}, JobStatus.DONE), + ({"LastJobStatus": 4, "ExitCode": 2}, JobStatus.FAILED), + ({"LastJobStatus": 4, "ExitBySignal": True}, JobStatus.FAILED), + ({"LastJobStatus": 5, "ExitCode": 3}, JobStatus.ABORTED), + ({"LastJobStatus": 2}, JobStatus.ABORTED), + ({"LastJobStatus": 1}, JobStatus.ABORTED), + ({}, JobStatus.ABORTED), + ], + ids=[ + "purged-done", + "purged-done-no-code", + "purged-failed", + "purged-signal", + "removed-while-held", + "killed-running", + "killed-idle", + "bare", + ], + ) + def test_removed_jobs_keep_their_result( + self, fields: dict[str, Any], expected: JobStatus + ): + assert status.job_status(ad(JobStatus=3, **fields)) == expected + + +class TestParseAds: + def test_condor_q_prints_nothing_for_no_match(self): + assert status.parse_ads("") == [] + assert status.parse_ads("\n \n") == [] + + def test_condor_history_prints_an_empty_list(self): + assert status.parse_ads("[\n]\n") == [] + + @pytest.mark.parametrize("raw", ["{}", "[1, 2]", '"text"', "not json"]) + def test_anything_else_is_refused(self, raw: str): + with pytest.raises(ValueError): + status.parse_ads(raw) + + def test_every_status_query_asks_for_what_the_mapping_reads(self): + assert { + "ClusterId", + "ProcId", + "JobStatus", + "LastJobStatus", + "ExitCode", + "ExitBySignal", + "HoldReasonCode", + } <= set(status.STATE_ATTRIBUTES) From be3a762ac05cb63b2573e3a50feec8168e89e73c Mon Sep 17 00:00:00 2001 From: alexander Date: Wed, 4 Nov 2026 14:45:21 +0100 Subject: [PATCH 4/4] fix: no X.509 proxy in HTCondor submit descriptions interCEde no longer supports proxies, so a description has nothing to say about one. --- src/intercede/_backends/_htcondor/submit.py | 9 --------- tests/unit/backends/_htcondor/test_core.py | 8 -------- 2 files changed, 17 deletions(-) diff --git a/src/intercede/_backends/_htcondor/submit.py b/src/intercede/_backends/_htcondor/submit.py index bfd1474..0fa5a0f 100644 --- a/src/intercede/_backends/_htcondor/submit.py +++ b/src/intercede/_backends/_htcondor/submit.py @@ -96,7 +96,6 @@ class SubmitPlan: memory_mb: int | None = None wall_time_s: int | None = None queue: str | None = None - proxy_file: str | None = None def render(plan: SubmitPlan) -> str: @@ -107,7 +106,6 @@ def render(plan: SubmitPlan) -> str: lines.extend(_render_copies(plan)) lines.extend(_render_file_transfer(plan)) lines.extend(_render_resources(plan)) - lines.extend(_render_proxy(plan)) lines.extend(_render_queue(plan)) return "\n".join(lines) + "\n" @@ -213,13 +211,6 @@ def _render_resources(plan: SubmitPlan) -> list[str]: return lines -def _render_proxy(plan: SubmitPlan) -> list[str]: - if plan.proxy_file is None: - return [] - - return [f"x509userproxy = {_escape_macros(plan.proxy_file)}"] - - def _render_queue(plan: SubmitPlan) -> list[str]: return [ # Keep completed jobs until purge() releases them diff --git a/tests/unit/backends/_htcondor/test_core.py b/tests/unit/backends/_htcondor/test_core.py index 76cc9ec..185d2e6 100644 --- a/tests/unit/backends/_htcondor/test_core.py +++ b/tests/unit/backends/_htcondor/test_core.py @@ -148,14 +148,6 @@ def test_resources_for_a_schedd_and_for_a_ce_router(self): ]: assert line in lines(text) - def test_proxy(self): - assert "x509userproxy = /creds/proxy" in lines( - render(plan(proxy_file="/creds/proxy")) - ) - assert not any( - line.startswith("x509userproxy") for line in lines(render(plan())) - ) - def test_finished_jobs_stay_until_purged(self): # The one lifetime rule interCEde sets: output can be collected # again until the consumer releases the job.