diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index a17603a..4e49af3 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -39,3 +39,4 @@ repos: hooks: - id: mypy exclude: tests + additional_dependencies: ["pydantic>=2.0.0"] diff --git a/docs/index.md b/docs/index.md new file mode 100644 index 0000000..6bfb507 --- /dev/null +++ b/docs/index.md @@ -0,0 +1,6 @@ +# interCEde + +Unified interfaces to Computing Elements and batch systems for DIRAC / DiracX. + +The contract is described in IC-ADR-001. See the [API reference](user/api-reference.md) +for the protocols, data types and exceptions a consumer may import. diff --git a/docs/user/api-reference.md b/docs/user/api-reference.md new file mode 100644 index 0000000..a0dfb90 --- /dev/null +++ b/docs/user/api-reference.md @@ -0,0 +1,16 @@ +# API reference + +Everything listed here is importable from `intercede` and follows SemVer. +Anything else, including every name starting with `_`, is internal. + +## Protocols + +::: intercede.protocols + +## Data types + +::: intercede.models + +## Exceptions + +::: intercede.exceptions diff --git a/mkdocs.yml b/mkdocs.yml new file mode 100644 index 0000000..e591f8c --- /dev/null +++ b/mkdocs.yml @@ -0,0 +1,25 @@ +site_name: interCEde +site_description: Unified interfaces to Computing Elements and batch systems for DIRAC / DiracX +repo_url: https://github.com/DIRACGrid/intercede + +theme: + name: material + logo: assets/intercede-icon.svg + favicon: assets/intercede-icon.svg + +plugins: + - search + - mkdocstrings: + handlers: + python: + paths: [src] + options: + show_source: false + members_order: source + merge_init_into_class: true + +validation: + links: + # The developer docs link to repository files outside docs/ (e.g. renovate.json), + # which resolve on GitHub but not in the built site. + not_found: info diff --git a/pyproject.toml b/pyproject.toml index 6e6f077..6e8b14f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,9 @@ requires-python = ">=3.11" keywords = [] license = { text = "GPL-3.0-only" } dynamic = ["version"] -dependencies = [] +dependencies = [ + "pydantic>=2" +] classifiers = [ "Intended Audience :: Science/Research", "License :: OSI Approved :: GNU General Public License v3 (GPLv3)", @@ -39,9 +41,6 @@ testing = [ "ty", ] -[tool.setuptools.packages.find] -where = ["src"] - [tool.pytest.ini_options] pythonpath = ["src"] testpaths = ["tests"] @@ -51,6 +50,10 @@ markers = [ "xfail_backend: strict xfail a test on the named backend fixture(s)", ] +[tool.mypy] +strict = true +plugins = ["pydantic.mypy"] +files = ["src"] [tool.ruff] src = ["src", "tests"] @@ -113,6 +116,7 @@ py311 = { features = ["testing", "py311"] } py312 = { features = ["testing", "py312"] } py313 = { features = ["testing", "py313"] } py314 = { features = ["testing", "py314"] } +docs = { features = ["docs"], solve-group = "default" } [tool.pixi.feature.testing.pypi-dependencies] intercede = { path = ".", editable = true } @@ -125,6 +129,16 @@ pre-commit = "*" pre-commit = { cmd = "pre-commit", description = "Run pre-commit" } +# Features for building the documentation +[tool.pixi.feature.docs.pypi-dependencies] +intercede = { path = ".", editable = true } +mkdocs = "<2" +mkdocs-material = "*" +mkdocstrings = { version = "*", extras = ["python"] } +[tool.pixi.feature.docs.tasks] +docs = { cmd = "mkdocs build --strict", description = "Build the documentation" } + + [tool.pixi.tasks] unit-test = { cmd = "pytest -v", description = "Run the unit tests" } diff --git a/src/intercede/__init__.py b/src/intercede/__init__.py index 79016b3..08dd317 100644 --- a/src/intercede/__init__.py +++ b/src/intercede/__init__.py @@ -5,10 +5,76 @@ from importlib.metadata import PackageNotFoundError from importlib.metadata import version as get_version +from intercede.exceptions import ( + AuthenticationError, + ConfigurationError, + InterCEdeError, + ResourceUnavailableError, + SpecificationRejectedError, +) +from intercede.models import ( + ContainerSpec, + CopyVars, + Diagnostics, + FileRef, + JobHandle, + JobID, + JobOutput, + JobStatus, + OpOutcome, + OutputMember, + OutputSpec, + Resources, + StatusMap, + Submission, + SubmissionSpec, +) +from intercede.protocols import ( + Diagnosable, + JobBackend, + Killable, + OutputRetriever, + Purgeable, + StatusReporter, + Submitter, +) + try: __version__ = get_version(__name__) - version = __version__ except PackageNotFoundError: - version = "Unknown" + __version__ = "unknown" +version = __version__ -__all__ = ["__version__"] +__all__ = [ + "__version__", + # protocols + "Diagnosable", + "JobBackend", + "Killable", + "OutputRetriever", + "Purgeable", + "StatusReporter", + "Submitter", + # data types + "ContainerSpec", + "CopyVars", + "Diagnostics", + "FileRef", + "JobHandle", + "JobID", + "JobOutput", + "JobStatus", + "OpOutcome", + "OutputMember", + "OutputSpec", + "Resources", + "StatusMap", + "Submission", + "SubmissionSpec", + # exceptions + "AuthenticationError", + "ConfigurationError", + "InterCEdeError", + "ResourceUnavailableError", + "SpecificationRejectedError", +] diff --git a/src/intercede/_backends/__init__.py b/src/intercede/_backends/__init__.py new file mode 100644 index 0000000..2dc6f82 --- /dev/null +++ b/src/intercede/_backends/__init__.py @@ -0,0 +1 @@ +"""Internal backend implementations. Nothing here is public API.""" diff --git a/src/intercede/_backends/batch/__init__.py b/src/intercede/_backends/batch/__init__.py new file mode 100644 index 0000000..b689b6d --- /dev/null +++ b/src/intercede/_backends/batch/__init__.py @@ -0,0 +1 @@ +"""The composed backend: a transport combined with a scheduler (IC-ADR-001 §4).""" diff --git a/src/intercede/_backends/batch/protocols.py b/src/intercede/_backends/batch/protocols.py new file mode 100644 index 0000000..bd85d09 --- /dev/null +++ b/src/intercede/_backends/batch/protocols.py @@ -0,0 +1,78 @@ +"""Internal collaborator protocols of the composed backend (IC-ADR-001 §4). + +Shapes only: transports and schedulers are implemented in later issues. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from pathlib import Path +from types import TracebackType +from typing import Protocol, Self, runtime_checkable + +from pydantic import BaseModel, ConfigDict + +from intercede.models import CopyVars, JobID, JobStatus, SubmissionSpec + +Command = Sequence[str] # an argv: only argv and staged files ever cross the transport +# One resource, so its own IDs are unique: no need for handles here. +SchedulerStatusMap = Mapping[JobID, JobStatus] + + +class CommandResult(BaseModel): + """The outcome of running one command on the resource.""" + + model_config = ConfigDict(frozen=True) + + returncode: int + stdout: str + stderr: str + + +@runtime_checkable +class Transport(Protocol): + """How InterCEde reaches a resource: local shell, SSH, ... + + Transports hold connections, so they share the backend lifecycle: cheap + construction with no I/O, and an idempotent close. + """ + + async def __aenter__(self) -> Self: ... + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: ... + + async def run(self, argv: Command) -> CommandResult: ... + + async def put(self, local: Path, remote: str) -> None: ... + + async def get(self, remote: str, local: Path) -> None: ... + + +@runtime_checkable +class Scheduler(Protocol): + """How work is queued: builds commands locally and parses their output.""" + + stages_own_files: bool # does the scheduler move the sandbox itself? + + def submit_command( + self, spec: SubmissionSpec, copies: Sequence[CopyVars], workdir: str + ) -> Command: + """Build the command that queues every copy; ``workdir`` is where the sandbox was staged.""" + ... + + def parse_submit(self, raw: str) -> Sequence[JobID]: + """Read the IDs from the submit command's output, one per copy, in copy order.""" + ... + + def status_command(self, ids: Sequence[JobID]) -> Command: ... + + def parse_status(self, raw: str, ids: Sequence[JobID]) -> SchedulerStatusMap: + """Return a status for every id asked about; ids missing from ``raw`` are ``UNKNOWN``.""" + ... + + def kill_command(self, ids: Sequence[JobID]) -> Command: ... diff --git a/src/intercede/_backends/fake.py b/src/intercede/_backends/fake.py new file mode 100644 index 0000000..3ba2506 --- /dev/null +++ b/src/intercede/_backends/fake.py @@ -0,0 +1,363 @@ +"""An in-memory backend that implements the whole contract. + +It proves the contract is easy to implement and to fake, and it is what the +unit conformance suite runs against. Nothing leaves the process. +""" + +from __future__ import annotations + +from collections.abc import Collection, Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from types import TracebackType +from typing import Self + +from intercede.exceptions import ( + InterCEdeError, + ResourceUnavailableError, + SpecificationRejectedError, +) +from intercede.models import ( + CopyVars, + Diagnostics, + FileRef, + JobHandle, + JobID, + JobOutput, + JobStatus, + OpOutcome, + OutputMember, + StatusMap, + Submission, + SubmissionSpec, +) +from intercede.protocols import ( + Diagnosable, + JobBackend, + Killable, + OutputRetriever, + Purgeable, +) + +BACKEND = "fake" +_ACTIVE = (JobStatus.SUBMITTED, JobStatus.WAITING, JobStatus.RUNNING) + + +def _is_url(source: str) -> bool: + return "://" in source + + +@dataclass +class _FakeJob: + environment: Mapping[str, str] + inputs: dict[str, bytes] # contents, read at submission (see submit) + secrets: dict[str, bytes] + status: JobStatus = JobStatus.SUBMITTED + stdout: str = "" + stderr: str = "" + files: dict[str, bytes] = field(default_factory=dict) + + +class FakeBackend: + """In-memory backend with every capability, and hooks for tests. + + Args: + resource: Identity of the fake resource, stored in every handle. + refuse_copies: Copy indexes that ``submit`` refuses, to test partial submission. + fail_jobs: Job IDs for which later operations fail, to test per-job failures. + resource_staging: Whether the resource moves URL inputs and outputs itself; + when False, a spec that needs it is rejected at submit. + containers: Whether the resource can run a container image; when False, a + spec that asks for one is rejected at submit. + + """ + + def __init__( + self, + resource: str = "fake://local", + *, + refuse_copies: Collection[int] = (), + fail_jobs: Collection[str] = (), + resource_staging: bool = True, + containers: bool = True, + ) -> None: + # Construction is cheap and does no I/O (IC-ADR-001 §3). + self.resource = resource + self.refuse_copies = set(refuse_copies) + self.fail_jobs = set(fail_jobs) + self.resource_staging = resource_staging + self.containers = containers + self.unavailable = False # set to True to make whole operations fail + self._jobs: dict[JobID, _FakeJob] = {} + self._next_id = 0 + self._closed = False + + # -- lifecycle --------------------------------------------------------- + + async def __aenter__(self) -> Self: + self._closed = False + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + self._closed = True # idempotent: closing twice is harmless + + def _check_usable(self) -> None: + if self._closed: + raise InterCEdeError("backend is closed") + if self.unavailable: + raise ResourceUnavailableError(f"{self.resource} is unavailable") + + # -- essential capabilities ------------------------------------------- + + async def submit( + self, spec: SubmissionSpec, copies: int | Sequence[CopyVars] = 1 + ) -> Submission: + self._check_usable() + self._check_supported(spec) + copy_vars = [CopyVars()] * copies if isinstance(copies, int) else list(copies) + + # File ownership: read local inputs now, so the backend never depends on + # caller-owned paths after submit() returns. + inputs = self._read_local_inputs(spec.inputs) + + handles: dict[int, JobHandle] = {} + failures: dict[int, str] = {} + for index, copy in enumerate(copy_vars): + if index in self.refuse_copies: + failures[index] = "refused by the fake backend" + continue + job_id = JobID(str(self._next_id)) + self._next_id += 1 + self._jobs[job_id] = _FakeJob( + environment={**spec.environment, **copy.environment}, + inputs=dict(inputs), + secrets={ + name: secret.get_secret_value() + for name, secret in copy.secrets.items() + }, + ) + handles[index] = JobHandle( + backend=BACKEND, resource=self.resource, id=job_id + ) + return Submission(handles=handles, failures=failures) + + async def get_status(self, handles: Sequence[JobHandle]) -> StatusMap: + self._check_usable() + + results: dict[JobHandle, JobStatus] = {} + for handle in handles: + job = self._find(handle) + if job is not None: + results[handle] = job.status + else: + results[handle] = JobStatus.UNKNOWN + return results + + # -- optional capabilities -------------------------------------------- + + async def get_output( + self, handles: Sequence[JobHandle], dest: Path + ) -> Mapping[JobHandle, OpOutcome[JobOutput]]: + self._check_usable() + results: dict[JobHandle, OpOutcome[JobOutput]] = {} + for handle in handles: + job, error = self._lookup(handle) + if job is None: + results[handle] = OpOutcome[JobOutput](ok=False, reason=error) + elif job.status in _ACTIVE: + results[handle] = OpOutcome[JobOutput]( + ok=False, reason="job has not finished" + ) + else: + results[handle] = self._write_output(handle.id, job, dest) + return results + + async def kill( + self, handles: Sequence[JobHandle] + ) -> Mapping[JobHandle, OpOutcome[None]]: + self._check_usable() + results: dict[JobHandle, OpOutcome[None]] = {} + for handle in handles: + job, error = self._lookup(handle) + if job is None: + results[handle] = OpOutcome[None](ok=False, reason=error) + elif job.status in _ACTIVE: + job.status = JobStatus.ABORTED + results[handle] = OpOutcome[None](ok=True) + else: + results[handle] = OpOutcome[None](ok=True, reason="already finished") + return results + + async def purge( + self, handles: Sequence[JobHandle] + ) -> Mapping[JobHandle, OpOutcome[None]]: + self._check_usable() + results: dict[JobHandle, OpOutcome[None]] = {} + for handle in handles: + job, error = self._lookup(handle) + if job is None: + results[handle] = OpOutcome[None](ok=False, reason=error) + else: + del self._jobs[handle.id] + results[handle] = OpOutcome[None](ok=True) + return results + + async def get_diagnostics( + self, handles: Sequence[JobHandle], name: str | None = None + ) -> Mapping[JobHandle, OpOutcome[Diagnostics]]: + self._check_usable() + results: dict[JobHandle, OpOutcome[Diagnostics]] = {} + for handle in handles: + job, error = self._lookup(handle) + if job is None: + results[handle] = OpOutcome[Diagnostics](ok=False, reason=error) + else: + diagnostics = Diagnostics(name=name or "stdout", content=job.stdout) + results[handle] = OpOutcome[Diagnostics](ok=True, value=diagnostics) + return results + + # -- test hooks -------------------------------------------------------- + + def finish( + self, + handle: JobHandle, + *, + status: JobStatus = JobStatus.DONE, + stdout: str = "", + files: Mapping[str, bytes] | None = None, + ) -> None: + """Mark a job as finished, with the given output. Test hook, not part of the contract.""" + job = self._jobs[handle.id] + job.status = status + job.stdout = stdout + job.files = dict(files or {}) + + # -- helpers ----------------------------------------------------------- + + def _check_supported(self, spec: SubmissionSpec) -> None: + # Refuse at submit what the resource cannot honour (IC-ADR-001 §2.3, §2.4). + self._check_container_support(spec) + + if not self.resource_staging: + self._check_can_fetch(spec.inputs) + if spec.outputs is not None: + self._check_can_upload(spec.outputs.members) + else: + raise SpecificationRejectedError(f"tag: {spec.tag} contains no outputs") + + def _check_container_support(self, spec: SubmissionSpec) -> None: + if spec.container is not None and not self.containers: + raise SpecificationRejectedError( + f"{self.resource} cannot run container images" + ) + + def _check_can_fetch(self, inputs: Sequence[FileRef]) -> None: + for ref in inputs: + if _is_url(ref.source): + raise SpecificationRejectedError( + f"{self.resource} cannot fetch inputs from URLs" + ) + + def _check_can_upload(self, members: Sequence[OutputMember]) -> None: + for member in members: + if member.destination is not None: + raise SpecificationRejectedError( + f"{self.resource} cannot upload outputs to URLs" + ) + + def _find(self, handle: JobHandle) -> _FakeJob | None: + if handle.backend != BACKEND or handle.resource != self.resource: + return None # a handle for another backend is never ours + return self._jobs.get(handle.id) + + def _lookup(self, handle: JobHandle) -> tuple[_FakeJob | None, str | None]: + if handle.id in self.fail_jobs: + return None, "injected failure" + job = self._find(handle) + if job is not None: + return (job, None) + else: + return (None, "unknown job") + + @staticmethod + def _read_local_inputs(inputs: Sequence[FileRef]) -> dict[str, bytes]: + contents: dict[str, bytes] = {} + for ref in inputs: + if not _is_url(ref.source): + contents[ref.name] = Path(ref.source).read_bytes() + return contents + + @staticmethod + def _write_output( + job_id: JobID, + job: _FakeJob, + dest: Path, + ) -> OpOutcome[JobOutput]: + job_dir = (dest / job_id).resolve() + files_dir = job_dir / "files" + + targets, escaping = FakeBackend._build_output_targets(files_dir, job.files) + + if escaping: + return OpOutcome[JobOutput]( + ok=False, + reason=f"members escape the output directory: {escaping}", + ) + + job_dir.mkdir(parents=True, exist_ok=True) + + stdout = job_dir / "stdout" + stderr = job_dir / "stderr" + + stdout.write_text(job.stdout) + stderr.write_text(job.stderr) + + FakeBackend._write_job_files(job.files, targets) + + return OpOutcome[JobOutput]( + ok=True, + value=JobOutput( + stdout=stdout, + stderr=stderr, + files=targets, + ), + ) + + @staticmethod + def _build_output_targets( + files_dir: Path, files: Mapping[str, bytes] + ) -> tuple[dict[str, Path], list[str]]: + targets = {} + escaping = [] + + for name in files: + path = (files_dir / name).resolve() + + if path.is_relative_to(files_dir): + targets[name] = path + else: + escaping.append(name) + + return targets, sorted(escaping) + + @staticmethod + def _write_job_files( + files: Mapping[str, bytes], targets: Mapping[str, Path] + ) -> None: + for name, path in targets.items(): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(files[name]) + + +# The type checker proves conformance: these assignments fail mypy if the fake +# ever drifts from the contract (IC-ADR-001 §4). +_backend: type[JobBackend] = FakeBackend +_output: type[OutputRetriever] = FakeBackend +_kill: type[Killable] = FakeBackend +_purge: type[Purgeable] = FakeBackend +_diag: type[Diagnosable] = FakeBackend diff --git a/src/intercede/exceptions.py b/src/intercede/exceptions.py new file mode 100644 index 0000000..717a82b --- /dev/null +++ b/src/intercede/exceptions.py @@ -0,0 +1,35 @@ +"""Exceptions for failures of a whole operation (IC-ADR-001 §3). + +Anything that can fail for some jobs and succeed for others is reported in the +returned map instead, never raised. +""" + +from __future__ import annotations + +__all__ = [ + "AuthenticationError", + "ConfigurationError", + "InterCEdeError", + "ResourceUnavailableError", + "SpecificationRejectedError", +] + + +class InterCEdeError(Exception): + """Root of every exception InterCEde raises.""" + + +class ConfigurationError(InterCEdeError): + """A backend configuration is invalid.""" + + +class AuthenticationError(InterCEdeError): + """The resource rejected the credentials (IC-ADR-003 §3).""" + + +class ResourceUnavailableError(InterCEdeError): + """The resource could not be reached, or failed the whole operation.""" + + +class SpecificationRejectedError(InterCEdeError): + """The backend cannot honour this specification, so no copy was submitted.""" diff --git a/src/intercede/models.py b/src/intercede/models.py new file mode 100644 index 0000000..ccc7e7e --- /dev/null +++ b/src/intercede/models.py @@ -0,0 +1,257 @@ +"""Data types of the InterCEde contract (IC-ADR-001 §2).""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from enum import StrEnum +from pathlib import Path, PurePosixPath +from typing import Annotated, Generic, Literal, NewType, Self, TypeVar + +from pydantic import ( + AfterValidator, + BaseModel, + ConfigDict, + PositiveInt, + SecretBytes, + model_validator, +) + +__all__ = [ + "ContainerSpec", + "CopyVars", + "Diagnostics", + "FileRef", + "JobHandle", + "JobID", + "JobOutput", + "JobStatus", + "OpOutcome", + "OutputMember", + "OutputSpec", + "Resources", + "StatusMap", + "Submission", + "SubmissionSpec", +] + +T = TypeVar("T") + + +class _Model(BaseModel): + """Base for every contract type: immutable, and strict about unknown fields. + + Values arrive from configuration and from stored handles, so they are + validated rather than trusted (IC-ADR-001 §2). + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + +def _check_in_workdir(name: str) -> str: + """Reject names that would leave the job's working directory (IC-ADR-001 §2, §3).""" + path = PurePosixPath(name) + if ( + not name + or path.is_absolute() + or ".." in path.parts + or path == PurePosixPath(".") + ): + raise ValueError( + f"{name!r} must be a relative path inside the working directory" + ) + return name + + +WorkdirName = Annotated[str, AfterValidator(_check_in_workdir)] + + +# -------------------------------------------------------------------------- +# What you submit +# -------------------------------------------------------------------------- + + +class Resources(_Model): + """What a job asks of the resource.""" + + cpus: PositiveInt = 1 # resources must never be 0 or negative, and will be rejected + memory_mb: PositiveInt | None = None + wall_time_s: PositiveInt | None = None + queue: str | None = None + + +class FileRef(_Model): + """An input file: its name in the working directory, and where it comes from.""" + + name: WorkdirName + source: str # a local path (staged by InterCEde) or a URL the resource fetches + + +class ContainerSpec(_Model): + """The image the resource runs the executable in. InterCEde never pulls, reads or builds it.""" + + image: str # "docker://...", a .sif path or URL, "oras://..." + + +class OutputMember(_Model): + """An output file, and who moves it.""" + + pattern: WorkdirName # a name or a glob in the working directory + destination: str | None = ( + None # None: kept for get_output(); URL: uploaded by the resource + ) + + +class OutputSpec(_Model): + """Which outputs a job produces.""" + + members: Sequence[OutputMember] = () + include_unlisted: bool = False # never picks up stdout, stderr or the scheduler log + + +class CopyVars(_Model): + """What differs between the copies of one specification (IC-ADR-001 §2.1).""" + + environment: Mapping[str, str] = {} # identifiers; readable by the site + # File name -> content, staged at 0600 and never put in the job description. + # SecretBytes keeps the content out of repr(), logs and tracebacks. + secrets: Mapping[WorkdirName, SecretBytes] = {} + + +class SubmissionSpec(_Model): + """What every copy of a submission shares.""" + + executable: str + arguments: Sequence[str] = () + inputs: Sequence[FileRef] = () + outputs: OutputSpec | None = None + resources: Resources = Resources() + environment: Mapping[str, str] = {} + container: ContainerSpec | None = None + stdout: WorkdirName | None = None # the backend picks a unique name when None + stderr: WorkdirName | None = None + tag: str | None = None # consumer token for correlation and debugging only + + +# -------------------------------------------------------------------------- +# What comes back +# -------------------------------------------------------------------------- + +JobID = NewType( + "JobID", str +) # the resource's own job ID, unique only within that resource + + +class JobHandle(_Model): + """The durable, serialisable identity of one submitted job. + + The consumer stores it and passes it back into later calls. It carries + everything the backend needs to find the job again, and nothing secret. + Handles are hashable, and every per-job result map is keyed by them: a + ``JobID`` alone is not unique across resources. + + Unknown fields are rejected, so a handle written by a newer format fails + loudly instead of losing its routing; ``version`` is what a newer reader + branches on. + """ + + version: Literal[1] = 1 # format of the handle itself, for stored handles + backend: str # the configuration kind, e.g. "arc" or "batch" + resource: str # endpoint or host; same string as CredentialRequirements.resource + id: JobID + routing: Mapping[str, str] = {} # backend-specific, opaque to consumers + + def __hash__(self) -> int: + # The default hash of a frozen model fails on the routing dict. + return hash( + ( + self.version, + self.backend, + self.resource, + self.id, + frozenset(self.routing.items()), + ) + ) + + +class JobStatus(StrEnum): + """Common job states. + + Only UNKNOWN is fixed by the ADR. The other values are a proposal, aligned + with DIRAC's pilot states; confirm before release. + """ + + SUBMITTED = "Submitted" + WAITING = "Waiting" + RUNNING = "Running" + DONE = "Done" + FAILED = "Failed" + ABORTED = "Aborted" + UNKNOWN = "Unknown" # the backend no longer knows the job; never dropped + + +StatusMap = Mapping[JobHandle, JobStatus] + + +class Submission(_Model): + """The outcome of submit(). + + Both maps are keyed by copy index, so ``handles[i]`` belongs to ``copies[i]`` + even when other copies were refused. Every copy appears in exactly one map. + """ + + handles: Mapping[int, JobHandle] + failures: Mapping[int, str] = {} # copy index -> reason, for refused copies + + @model_validator(mode="after") + def _disjoint(self) -> Self: + if both := self.handles.keys() & self.failures.keys(): + raise ValueError(f"copies {sorted(both)} are both submitted and refused") + return self + + +class JobOutput(_Model): + """A manifest of collected output: paths, never file contents.""" + + stdout: Path | None = None + stderr: Path | None = None + log: Path | None = None # the scheduler or CE log, never a payload file + files: Mapping[str, Path] = {} # payload members only + + +class Diagnostics(_Model): + """A file read from a job before it completes. + + The fields are a proposal; the ADR does not fix them. + """ + + name: str + content: str + + +class OpOutcome(_Model, Generic[T]): + """The outcome of an operation for one job (IC-ADR-001 §3). + + A failed outcome carries a ``reason`` and no value. A successful one carries + a value, except for operations that return nothing (``OpOutcome[None]``); + it may still give a ``reason``, such as "already finished". + """ + + ok: bool + value: T | None = None + reason: str | None = None + + @model_validator(mode="after") + def _consistent(self) -> Self: + if not self.ok: + if self.reason is None: + raise ValueError("a failed outcome needs a reason") + if self.value is not None: + raise ValueError("a failed outcome cannot carry a value") + elif self.value is None and self._expects_value(): + raise ValueError("a successful outcome needs a value") + return self + + @classmethod + def _expects_value(cls) -> bool: + args = cls.__pydantic_generic_metadata__["args"] + return bool(args) and args[0] not in (None, type(None)) diff --git a/src/intercede/protocols.py b/src/intercede/protocols.py new file mode 100644 index 0000000..e104b88 --- /dev/null +++ b/src/intercede/protocols.py @@ -0,0 +1,140 @@ +"""The capability protocols of the InterCEde contract (IC-ADR-001 §3). + +A backend qualifies by having the right methods; it never inherits from these +classes. Every protocol is runtime-checkable, so consumers can narrow on a +capability with ``isinstance``. Every operation is asynchronous, works on many +jobs at once, and returns one entry per handle it was given. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from pathlib import Path +from types import TracebackType +from typing import Protocol, Self, runtime_checkable + +from intercede.models import ( + CopyVars, + Diagnostics, + JobHandle, + JobOutput, + OpOutcome, + StatusMap, + Submission, + SubmissionSpec, +) + +__all__ = [ + "Diagnosable", + "JobBackend", + "Killable", + "OutputRetriever", + "Purgeable", + "StatusReporter", + "Submitter", +] + + +# -------------------------------------------------------------------------- +# Essential capabilities: every usable backend has these. +# -------------------------------------------------------------------------- + + +@runtime_checkable +class Submitter(Protocol): + """Submits one specification as one or more copies.""" + + async def submit( + self, spec: SubmissionSpec, copies: int | Sequence[CopyVars] = 1 + ) -> Submission: + """Submit ``spec`` once per copy. + + Pass a count for identical copies, or one ``CopyVars`` per copy when each + copy needs its own identity or secret. ``Submission.handles[i]`` belongs + to copy ``i``; refused copies are reported in ``Submission.failures``. + + A spec the backend cannot honour raises ``SpecificationRejectedError`` + and submits nothing. Local input files are read before this returns: + the backend never depends on caller-owned paths afterwards. + """ + ... + + +@runtime_checkable +class StatusReporter(Protocol): + """Reports the status of jobs.""" + + async def get_status(self, handles: Sequence[JobHandle]) -> StatusMap: + """Return a status per handle. Jobs the backend does not know are ``UNKNOWN``, never dropped.""" + ... + + +@runtime_checkable +class JobBackend(Submitter, StatusReporter, Protocol): + """A complete backend: what the registry returns. + + Backends are asynchronous context managers (IC-ADR-001 §3): construction + does no I/O, and closing is idempotent. + """ + + async def __aenter__(self) -> Self: ... + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: ... + + +# -------------------------------------------------------------------------- +# Optional capabilities: a backend implements only those it really has. +# -------------------------------------------------------------------------- + + +@runtime_checkable +class OutputRetriever(Protocol): + """Collects the outputs a job left at the resource.""" + + async def get_output( + self, handles: Sequence[JobHandle], dest: Path + ) -> Mapping[JobHandle, OpOutcome[JobOutput]]: + """Write each job's kept outputs into ``dest`` and return a manifest per job. + + ``dest`` belongs to the consumer: InterCEde writes into it and never + deletes it. Every member lands under ``dest``. Collection is repeatable. + """ + ... + + +@runtime_checkable +class Killable(Protocol): + """Cancels jobs.""" + + async def kill( + self, handles: Sequence[JobHandle] + ) -> Mapping[JobHandle, OpOutcome[None]]: + """Cancel the jobs; the outcome is reported per job.""" + ... + + +@runtime_checkable +class Purgeable(Protocol): + """Releases a job's remaining state at the resource.""" + + async def purge( + self, handles: Sequence[JobHandle] + ) -> Mapping[JobHandle, OpOutcome[None]]: + """Release outputs and remote state without collecting them.""" + ... + + +@runtime_checkable +class Diagnosable(Protocol): + """Reads files from jobs before they complete.""" + + async def get_diagnostics( + self, handles: Sequence[JobHandle], name: str | None = None + ) -> Mapping[JobHandle, OpOutcome[Diagnostics]]: + """Read a file from each job; repeatable and non-destructive.""" + ... diff --git a/src/intercede/py.typed b/src/intercede/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py new file mode 100644 index 0000000..30d4d94 --- /dev/null +++ b/tests/unit/conftest.py @@ -0,0 +1,37 @@ +"""Fixtures for the unit suite. + +``backend`` is parametrised over every backend the conformance suite runs +against. Only the in-memory fake exists for now; real backends join the list +in the integration suite (IC-ADR-002). +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Callable + +import pytest + +from intercede import JobBackend +from intercede._backends.fake import FakeBackend + +BACKENDS: dict[str, Callable[[], JobBackend]] = { + "fake": FakeBackend, +} + + +@pytest.fixture(params=sorted(BACKENDS)) +async def backend(request: pytest.FixtureRequest) -> AsyncIterator[JobBackend]: + name = request.param + + backend_only_marker = request.node.get_closest_marker("backend_only") + if backend_only_marker is not None and name not in backend_only_marker.args: + pytest.skip(f"only runs on {', '.join(backend_only_marker.args)}") + + xfail_marker = request.node.get_closest_marker("xfail_backend") + if xfail_marker is not None and name in xfail_marker.args: + request.applymarker( + pytest.mark.xfail(strict=True, reason=f"known failure on {name}") + ) + + async with BACKENDS[name]() as instance: + yield instance diff --git a/tests/unit/test_conformance.py b/tests/unit/test_conformance.py new file mode 100644 index 0000000..0a4e0b8 --- /dev/null +++ b/tests/unit/test_conformance.py @@ -0,0 +1,126 @@ +"""Unit conformance suite: what every backend must do, written against the contract only. + +Tests here use nothing but the public API, so the same suite runs against real +backends in the integration stacks (IC-ADR-002). Behaviour that needs a test +hook (finishing a job, injecting failures) lives in ``test_fake_backend.py``. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from intercede import ( + CopyVars, + FileRef, + InterCEdeError, + JobBackend, + JobHandle, + JobID, + JobStatus, + Killable, + OutputRetriever, + Purgeable, + SubmissionSpec, +) + +SPEC = SubmissionSpec(executable="/bin/true", tag="conformance") + + +def stranger(backend_handle: JobHandle) -> JobHandle: + """Return a handle with the same job ID on another resource.""" + return backend_handle.model_copy(update={"resource": "elsewhere://"}) + + +async def test_submit_returns_one_handle_per_copy(backend: JobBackend): + sub = await backend.submit(SPEC, 3) + assert sorted(sub.handles) == [0, 1, 2] + assert not sub.failures + assert len(set(sub.handles.values())) == 3 + + +async def test_submit_with_copy_vars(backend: JobBackend): + copies = [ + CopyVars(environment={"STAMP": str(i)}, secrets={"pilot.secret": b"s%d" % i}) + for i in range(2) + ] + sub = await backend.submit(SPEC, copies) + assert sorted(sub.handles) == [0, 1] + + +async def test_handles_survive_serialisation(backend: JobBackend): + sub = await backend.submit(SPEC) + stored = JobHandle.model_validate_json(sub.handles[0].model_dump_json()) + status = await backend.get_status([stored]) + assert status[stored] != JobStatus.UNKNOWN + + +async def test_submit_does_not_retain_caller_paths(backend: JobBackend, tmp_path: Path): + source = tmp_path / "input.txt" + source.write_text("data") + sub = await backend.submit( + SubmissionSpec( + executable="/bin/true", inputs=[FileRef(name="in", source=str(source))] + ) + ) + source.unlink() # the caller may delete its files as soon as submit returns + status = await backend.get_status(list(sub.handles.values())) + assert all(status != JobStatus.UNKNOWN for status in status.values()) + + +async def test_status_answers_every_handle(backend: JobBackend): + sub = await backend.submit(SPEC, 2) + handles = list(sub.handles.values()) + unknown = handles[0].model_copy(update={"id": JobID("does-not-exist")}) + status = await backend.get_status([*handles, unknown]) + assert set(status) == {*handles, unknown} + assert status[unknown] == JobStatus.UNKNOWN + + +async def test_status_does_not_confuse_jobs_across_resources(backend: JobBackend): + sub = await backend.submit(SPEC) + ours = sub.handles[0] + theirs = stranger(ours) + status = await backend.get_status([ours, theirs]) + assert status[ours] != JobStatus.UNKNOWN + assert status[theirs] == JobStatus.UNKNOWN + + +async def test_kill_reports_per_job(backend: JobBackend): + if not isinstance(backend, Killable): + pytest.skip("backend cannot kill") + sub = await backend.submit(SPEC) + ours = sub.handles[0] + theirs = stranger(ours) + outcome = await backend.kill([ours, theirs]) + assert outcome[ours].ok + assert not outcome[theirs].ok + assert outcome[theirs].reason + + +async def test_purge_forgets_the_job(backend: JobBackend): + if not isinstance(backend, Purgeable): + pytest.skip("backend cannot purge") + sub = await backend.submit(SPEC) + handle = sub.handles[0] + assert (await backend.purge([handle]))[handle].ok + assert (await backend.get_status([handle]))[handle] == JobStatus.UNKNOWN + + +async def test_output_of_an_unknown_job_fails_per_job( + backend: JobBackend, tmp_path: Path +): + if not isinstance(backend, OutputRetriever): + pytest.skip("backend cannot retrieve output") + sub = await backend.submit(SPEC) + theirs = stranger(sub.handles[0]) + outcome = await backend.get_output([theirs], tmp_path) + assert not outcome[theirs].ok + + +async def test_close_is_idempotent(backend: JobBackend): + await backend.__aexit__(None, None, None) + await backend.__aexit__(None, None, None) + with pytest.raises(InterCEdeError): + await backend.submit(SPEC) diff --git a/tests/unit/test_fake_backend.py b/tests/unit/test_fake_backend.py new file mode 100644 index 0000000..3ada5f0 --- /dev/null +++ b/tests/unit/test_fake_backend.py @@ -0,0 +1,131 @@ +"""Behaviour of the fake that needs its test hooks: completion, partial failure, refusal.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from intercede import ( + ContainerSpec, + CopyVars, + FileRef, + JobStatus, + OutputMember, + OutputSpec, + ResourceUnavailableError, + SpecificationRejectedError, + SubmissionSpec, +) +from intercede._backends.fake import FakeBackend + +SPEC = SubmissionSpec(executable="/bin/true") + + +async def test_construction_does_no_io(): + FakeBackend(resource="unreachable://") # nothing to connect to, and nothing raised + + +async def test_partial_submission_keeps_copy_indexes(): + async with FakeBackend(refuse_copies={0, 2}) as backend: + copies = [CopyVars(environment={"STAMP": str(i)}) for i in range(4)] + sub = await backend.submit(SPEC, copies) + assert sorted(sub.handles) == [1, 3] + assert sorted(sub.failures) == [0, 2] + + +async def test_injected_failure_is_reported_per_job(): + async with FakeBackend(fail_jobs={"1"}) as backend: + sub = await backend.submit(SPEC, 2) + good, bad = sub.handles[0], sub.handles[1] + killed = await backend.kill([good, bad]) + assert killed[good].ok + assert not killed[bad].ok + + +async def test_whole_operation_failure_raises(): + async with FakeBackend() as backend: + backend.unavailable = True + with pytest.raises(ResourceUnavailableError): + await backend.submit(SPEC) + + +async def test_output_is_repeatable_and_lands_under_dest(tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, stdout="hello", files={"result/out.txt": b"42"}) + for _ in range(2): + outcome = (await backend.get_output([handle], tmp_path))[handle] + assert outcome.ok and outcome.value is not None + assert outcome.value.stdout is not None + assert outcome.value.stdout.read_text() == "hello" + assert outcome.value.files["result/out.txt"].read_bytes() == b"42" + assert outcome.value.files["result/out.txt"].is_relative_to( + tmp_path.resolve() + ) + + +async def test_output_of_a_running_job_fails_per_job(tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + assert not (await backend.get_output([handle], tmp_path))[handle].ok + + +@pytest.mark.parametrize("name", ["../../escaped", "/abs/escaped"]) +async def test_output_members_cannot_escape_dest(tmp_path: Path, name: str): + dest = tmp_path / "dest" + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, files={name: b"x"}) + outcome = (await backend.get_output([handle], dest))[handle] + assert not outcome.ok + assert not any(p.name == "escaped" for p in tmp_path.rglob("*")) + + +async def test_kill_stops_active_jobs(): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + await backend.kill([handle]) + assert (await backend.get_status([handle]))[handle] == JobStatus.ABORTED + + +async def test_diagnostics_are_repeatable(): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, status=JobStatus.RUNNING, stdout="progress") + for _ in range(2): + outcome = (await backend.get_diagnostics([handle]))[handle] + assert outcome.value is not None + assert outcome.value.content == "progress" + + +@pytest.mark.parametrize( + "spec", + [ + SubmissionSpec( + executable="x", inputs=[FileRef(name="in", source="https://example.org/in")] + ), + SubmissionSpec( + executable="x", + outputs=OutputSpec( + members=[OutputMember(pattern="out", destination="https://e.org/o")] + ), + ), + ], +) +async def test_spec_needing_resource_staging_is_rejected_at_submit( + spec: SubmissionSpec, +): + async with FakeBackend(resource_staging=False) as backend: + with pytest.raises(SpecificationRejectedError): + await backend.submit(spec) + + +async def test_spec_with_container_is_rejected_when_unsupported(): + async with FakeBackend(containers=False) as backend: + with pytest.raises(SpecificationRejectedError): + await backend.submit( + SubmissionSpec( + executable="x", container=ContainerSpec(image="docker://alpine") + ) + ) diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py new file mode 100644 index 0000000..7bcd413 --- /dev/null +++ b/tests/unit/test_models.py @@ -0,0 +1,139 @@ +"""Validation rules of the contract data types.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from intercede import ( + CopyVars, + Diagnostics, + FileRef, + JobHandle, + JobID, + JobOutput, + OpOutcome, + OutputMember, + Resources, + Submission, + SubmissionSpec, +) + + +def build_default_handle(job_id: str = "1", **kwargs: object) -> JobHandle: + return JobHandle.model_validate( + {"backend": "fake", "resource": "r", "id": job_id, **kwargs} + ) + + +class TestJobHandle: + def test_round_trips_through_json(self): + handle = build_default_handle(routing={"schedd": "a"}) + assert JobHandle.model_validate_json(handle.model_dump_json()) == handle + + def test_is_hashable_with_routing(self): + handle = build_default_handle(routing={"schedd": "a"}) + assert {handle: 1}[build_default_handle(routing={"schedd": "a"})] == 1 + + def test_same_id_on_different_resources_are_different_keys(self): + handle_a = build_default_handle(resource="cluster-a") + handle_b = build_default_handle(resource="cluster-b") + assert len({handle_a: 1, handle_b: 2}) == 2 + + def test_routing_is_part_of_identity(self): + assert build_default_handle(routing={"schedd": "a"}) != build_default_handle( + routing={"schedd": "b"} + ) + assert ( + len( + { + build_default_handle(routing={"schedd": "a"}), + build_default_handle(routing={"schedd": "b"}), + } + ) + == 2 + ) + + def test_rejects_unknown_fields_and_versions(self): + with pytest.raises(ValidationError): + build_default_handle(extra="x") + with pytest.raises(ValidationError): + build_default_handle(version=2) + + def test_is_immutable(self): + with pytest.raises(ValidationError): + build_default_handle().id = JobID("2") # type: ignore[misc] + + +class TestCopyVars: + def test_secret_content_never_appears_in_repr(self): + copy = CopyVars(secrets={"pilot.secret": b"TOPSECRET"}) + assert "TOPSECRET" not in repr(copy) + assert "TOPSECRET" not in str(copy) + assert "TOPSECRET" not in copy.model_dump_json() + + def test_secret_content_is_readable_on_purpose(self): + copy = CopyVars(secrets={"pilot.secret": b"TOPSECRET"}) + assert copy.secrets["pilot.secret"].get_secret_value() == b"TOPSECRET" + + +@pytest.mark.parametrize("name", ["", ".", "/etc/passwd", "../up", "a/../../up"]) +def test_names_must_stay_in_the_working_directory(name: str): + with pytest.raises(ValidationError): + FileRef(name=name, source="local/input") + with pytest.raises(ValidationError): + OutputMember(pattern=name) + with pytest.raises(ValidationError): + CopyVars(secrets={name: b"x"}) + with pytest.raises(ValidationError): + SubmissionSpec(executable="x", stdout=name) + + +@pytest.mark.parametrize("name", ["input.txt", "sub/dir/file", "*.root"]) +def test_relative_names_are_accepted(name: str): + FileRef(name=name, source="local/input") + OutputMember(pattern=name) + + +@pytest.mark.parametrize("field", ["cpus", "memory_mb", "wall_time_s"]) +def test_resources_must_be_positive(field: str): + with pytest.raises(ValidationError): + Resources.model_validate({field: 0}) + + +class TestSubmission: + def test_handles_are_keyed_by_copy_index(self): + sub = Submission( + handles={1: build_default_handle("a")}, failures={0: "refused"} + ) + assert sub.handles[1].id == "a" + + def test_a_copy_cannot_be_both_submitted_and_refused(self): + with pytest.raises(ValidationError): + Submission(handles={0: build_default_handle()}, failures={0: "refused"}) + + +class TestOpOutcome: + def test_success_carries_a_value(self): + assert OpOutcome[Diagnostics]( + ok=True, value=Diagnostics(name="n", content="c") + ).ok + + def test_success_without_a_value_is_only_valid_for_none(self): + assert OpOutcome[None](ok=True).ok + with pytest.raises(ValidationError): + OpOutcome[JobOutput](ok=True) + + def test_success_may_give_a_reason(self): + assert ( + OpOutcome[None](ok=True, reason="already finished").reason + == "already finished" + ) + + def test_failure_needs_a_reason(self): + with pytest.raises(ValidationError): + OpOutcome[None](ok=False) + + def test_failure_cannot_carry_a_value(self): + with pytest.raises(ValidationError): + OpOutcome[JobOutput](ok=False, reason="x", value=JobOutput()) diff --git a/tests/unit/test_protocols.py b/tests/unit/test_protocols.py new file mode 100644 index 0000000..a2bfdfb --- /dev/null +++ b/tests/unit/test_protocols.py @@ -0,0 +1,109 @@ +"""``isinstance`` narrowing against the runtime-checkable protocols.""" + +from __future__ import annotations + +import pytest + +import intercede +from intercede import ( + Diagnosable, + JobBackend, + Killable, + OutputRetriever, + Purgeable, + StatusReporter, + Submitter, +) +from intercede._backends.batch.protocols import Scheduler, Transport +from intercede._backends.fake import FakeBackend + +ALL = ( + Submitter, + StatusReporter, + JobBackend, + OutputRetriever, + Killable, + Purgeable, + Diagnosable, +) + + +class SubmitOnly: + async def submit(self, spec, copies=1): ... + + +class StatusOnly: + async def get_status(self, handles): ... + + +class Minimal(SubmitOnly, StatusOnly): + """A backend with only the essential capabilities, like a cloud provider.""" + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc_info): ... + + +class FakeTransport: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc_info): ... + + async def run(self, argv): ... + + async def put(self, local, remote): ... + + async def get(self, remote, local): ... + + +class FakeScheduler: + stages_own_files = False + + def submit_command(self, spec, copies, workdir): ... + + def parse_submit(self, raw): ... + + def status_command(self, ids): ... + + def parse_status(self, raw, ids): ... + + def kill_command(self, ids): ... + + +@pytest.mark.parametrize("protocol", ALL) +def test_fake_backend_has_every_capability(protocol: type): + assert isinstance(FakeBackend(), protocol) + + +def test_minimal_backend_has_only_the_essentials(): + backend = Minimal() + assert isinstance(backend, JobBackend) + assert not isinstance(backend, OutputRetriever | Killable | Purgeable | Diagnosable) + + +def test_narrow_protocols_stand_alone(): + assert isinstance(SubmitOnly(), Submitter) + assert not isinstance(SubmitOnly(), StatusReporter | JobBackend) + assert isinstance(StatusOnly(), StatusReporter) + assert not isinstance(StatusOnly(), Submitter | JobBackend) + + +def test_a_backend_without_lifecycle_is_not_a_job_backend(): + class NoLifecycle(SubmitOnly, StatusOnly): ... + + assert not isinstance(NoLifecycle(), JobBackend) + + +def test_collaborator_protocols(): + assert isinstance(FakeTransport(), Transport) + assert isinstance(FakeScheduler(), Scheduler) + assert not isinstance(object(), Transport | Scheduler) + + +def test_public_api_is_exported(): + for name in intercede.__all__: + assert hasattr(intercede, name), name + assert "Transport" not in intercede.__all__ + assert "Scheduler" not in intercede.__all__