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/dev/index.md b/docs/dev/index.md index 67d045a..7a527bb 100644 --- a/docs/dev/index.md +++ b/docs/dev/index.md @@ -8,4 +8,4 @@ DiracX projects: - [Reference](reference/index.md): technical details - [Explanations](explanations/index.md): why things are done this way -User documentation (`docs/user/`) will come once the library has a public API. +For code that uses interCEde, see the [user documentation](../user/index.md). diff --git a/docs/index.md b/docs/index.md new file mode 100644 index 0000000..20a8cb2 --- /dev/null +++ b/docs/index.md @@ -0,0 +1,13 @@ +# interCEde + +Unified interfaces to Computing Elements and batch systems for DIRAC / DiracX. + +A consumer submits and monitors jobs through one set of protocols, whatever the resource behind +them. interCEde is stateless: it never changes a job on its own, and the consumer stores the job +handles it gets back. + +- [User documentation](user/index.md): for code that uses interCEde, such as DiracX. Start with + the [API reference](user/reference/api.md): the protocols, data types and exceptions a consumer + may import. +- [Developer documentation](dev/index.md): for working on interCEde itself, including the + integration test stacks. diff --git a/docs/user/index.md b/docs/user/index.md new file mode 100644 index 0000000..d1b1fe2 --- /dev/null +++ b/docs/user/index.md @@ -0,0 +1,8 @@ +# User documentation + +For code that uses interCEde, such as DiracX. Organised following [Diátaxis](https://diataxis.fr/), +like the developer documentation: + +- [Reference](reference/index.md): the public API + +Tutorials, how-to guides and explanations will follow as the backends land. diff --git a/docs/user/reference/api.md b/docs/user/reference/api.md new file mode 100644 index 0000000..824c73e --- /dev/null +++ b/docs/user/reference/api.md @@ -0,0 +1,24 @@ +# 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 + +## Configuration + +::: intercede.config + +## Registry + +::: intercede.registry + +## Exceptions + +::: intercede.exceptions diff --git a/docs/user/reference/index.md b/docs/user/reference/index.md new file mode 100644 index 0000000..f85179f --- /dev/null +++ b/docs/user/reference/index.md @@ -0,0 +1,3 @@ +# Reference + +- [API reference](api.md): the protocols, data types and exceptions a consumer may import 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 8b418c1..aa21d24 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)", @@ -40,9 +42,6 @@ testing = [ "ty", ] -[tool.setuptools.packages.find] -where = ["src"] - [tool.pytest.ini_options] pythonpath = ["src"] testpaths = ["tests"] @@ -60,6 +59,10 @@ markers = [ "destructive_fetch: asserts one-shot/destructive output-retrieval semantics", ] +[tool.mypy] +strict = true +plugins = ["pydantic.mypy"] +files = ["src"] [tool.ruff] src = ["src", "tests"] @@ -122,6 +125,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 } @@ -134,6 +138,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..b92b224 100644 --- a/src/intercede/__init__.py +++ b/src/intercede/__init__.py @@ -5,10 +5,83 @@ from importlib.metadata import PackageNotFoundError from importlib.metadata import version as get_version +from intercede import registry +from intercede.config import BackendConfig +from intercede.exceptions import ( + AuthenticationError, + ConfigurationError, + InterCEdeError, + RegistryError, + 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__", + "registry", + # 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", + # configuration + "BackendConfig", + # exceptions + "AuthenticationError", + "ConfigurationError", + "InterCEdeError", + "RegistryError", + "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..a77faaf --- /dev/null +++ b/src/intercede/_backends/batch/__init__.py @@ -0,0 +1 @@ +"""The composed backend: a transport combined with a scheduler.""" diff --git a/src/intercede/_backends/batch/protocols.py b/src/intercede/_backends/batch/protocols.py new file mode 100644 index 0000000..5ea1bf1 --- /dev/null +++ b/src/intercede/_backends/batch/protocols.py @@ -0,0 +1,176 @@ +"""Internal collaborator protocols of the composed backend. + +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: + """Run one command on the resource and wait for it to finish. + + Args: + argv: The program and its arguments. Never passed through a shell. + + Returns: + The exit code and everything the command printed. A non-zero exit + code is returned, not raised: the scheduler decides what it means. + + Raises: + ResourceUnavailableError: The resource could not be reached. + AuthenticationError: The resource rejected the credentials. + + """ + ... + + async def put(self, local: Path, remote: str) -> None: + """Copy a file or directory from this host to the resource. + + Args: + local: The file or directory on this host. + remote: Its destination path on the resource. + + Raises: + ResourceUnavailableError: The copy failed. + + """ + ... + + async def get(self, remote: str, local: Path) -> None: + """Copy a file or directory from the resource to this host. + + Args: + remote: The file or directory on the resource. + local: Its destination path on this host. + + Raises: + ResourceUnavailableError: The copy failed. + + """ + ... + + +@runtime_checkable +class Scheduler(Protocol): + """How work is queued: builds commands locally and parses their output. + + A scheduler never runs anything itself; the backend runs its commands + through the transport. Its methods are pure, so they are tested without a + resource. + + Attributes: + stages_own_files: Whether the scheduler moves the job's files itself. + When False, the backend copies them with the transport first. + + """ + + stages_own_files: bool + + def submit_command( + self, spec: SubmissionSpec, copies: Sequence[CopyVars], workdir: str + ) -> Command: + """Build the command that queues every copy of ``spec``. + + Args: + spec: What every copy shares. + copies: One entry per copy, in copy order. + workdir: The directory on the resource where the files were staged. + + Returns: + The command to run through the transport. + + """ + ... + + def parse_submit(self, raw: str) -> Sequence[JobID]: + """Read the job IDs from the output of the submit command. + + Args: + raw: What the submit command printed. + + Returns: + One ID per copy, in copy order. + + Raises: + ValueError: ``raw`` does not hold one ID per copy. + + """ + ... + + def status_command(self, ids: Sequence[JobID]) -> Command: + """Build the command that reports the state of the given jobs. + + Args: + ids: The jobs to ask about. + + Returns: + The command to run through the transport. + + """ + ... + + def parse_status(self, raw: str, ids: Sequence[JobID]) -> SchedulerStatusMap: + """Read the jobs' states from the output of the status command. + + Args: + raw: What the status command printed. + ids: The jobs that were asked about. + + Returns: + A status for every ID in ``ids``; an ID missing from ``raw`` is + ``UNKNOWN``. + + """ + ... + + def kill_command(self, ids: Sequence[JobID]) -> Command: + """Build the command that cancels the given jobs. + + Args: + ids: The jobs to cancel. + + Returns: + The command to run through the transport. + + """ + ... diff --git a/src/intercede/_backends/config.py b/src/intercede/_backends/config.py new file mode 100644 index 0000000..5c57d33 --- /dev/null +++ b/src/intercede/_backends/config.py @@ -0,0 +1,204 @@ +"""What each backend needs from the public ``BackendConfig``. + +The consumer describes a backend with one flat model. ``resolve`` turns it into +the settings of the one backend it names, and is where per-kind rules live: a +field that does not apply to the chosen kind is rejected, not ignored. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import TYPE_CHECKING, Annotated, Any, Literal, Self + +from pydantic import Discriminator, PositiveInt, TypeAdapter, model_validator + +from intercede.models import _Model + +if TYPE_CHECKING: + from intercede.config import BackendConfig + +LIMIT_FIELDS = ("max_output_bytes", "max_output_files", "transfer_timeout_s") +SSH_FIELDS = ("host", "port", "user", "known_hosts", "jump_host", "connect_timeout_s") + + +class Limits(_Model): + """Bounds on output collection. Path containment is always on.""" + + max_output_bytes: PositiveInt = 10 * 2**30 + max_output_files: PositiveInt = 10_000 + transfer_timeout_s: PositiveInt = 3600 + + +class _BackendSettings(_Model): + queue: str | None = None + container_launcher: Sequence[str] | None = None + limits: Limits = Limits() + + @model_validator(mode="after") + def _launcher_names_the_image(self) -> Self: + launcher = self.container_launcher + if launcher is not None and not any("{image}" in arg for arg in launcher): + raise ValueError("container_launcher must contain an '{image}' placeholder") + return self + + +# -- monolithic backends ------------------------------------------------------ + + +class ARCSettings(_BackendSettings): + kind: Literal["arc"] + endpoint: str + # All of these together: a token for the channel, a proxy + # for delegation, or both. + auth: frozenset[Literal["token", "proxy"]] = frozenset({"token"}) + + @model_validator(mode="after") + def _auth_not_empty(self) -> Self: + if not self.auth: + raise ValueError("auth must name at least one of 'token', 'proxy'") + return self + + +class HTCondorCESettings(_BackendSettings): + kind: Literal["htcondor-ce"] + endpoint: str + # The token authenticates the channel; "proxy" adds an X.509 proxy to the + # submit description for sites that account on VOMS attributes. + auth: frozenset[Literal["token", "proxy"]] = frozenset({"token"}) + + @model_validator(mode="after") + def _token_required(self) -> Self: + if "token" not in self.auth: + raise ValueError( + "an HTCondor-CE authenticates with a token; auth must include 'token'" + ) + return self + + +# -- provider backends -------------------------------------------------------- + +_SECRET_NAMES = ("secret", "password", "token", "passphrase") +_SECRET_EXACT = ("key", "api_key", "access_key", "private_key") + + +class CloudSettings(_Model): + """A cloud provider that boots one VM per job. + + There is no queue and no filesystem to collect from afterwards, so the + queue, output limits and container launcher of the other families do not + apply. The provider's API keys come from the credential provider, never + from here. + """ + + kind: Literal["cloud"] + endpoint: str # the cloud's API endpoint, and its identity in job handles + driver: str # the driver library's name for this cloud, e.g. "openstack" + # Extra, non-secret arguments for the driver, e.g. {"ex_tenant_name": "dirac"}. + driver_options: Mapping[str, str] = {} + vm_image: str # an image ID, or "name:" + vm_flavor: str # a flavor ID, or "name:" + vm_networks: Sequence[str] = () + vm_ssh_key: str | None = None # the NAME of a key pair known to the cloud + vm_max_lifetime_s: PositiveInt = 14 * 24 * 3600 # older VMs are removed + + @model_validator(mode="after") + def _no_secret_driver_options(self) -> Self: + secret = sorted( + name + for name in self.driver_options + if name.lower() in _SECRET_EXACT + or any(word in name.lower() for word in _SECRET_NAMES) + ) + if secret: + raise ValueError( + f"driver_options must not hold secrets {secret}; " + "they come from the credential provider" + ) + return self + + +# -- composed backends -------------------------------------------------------- + + +class LocalTransportSettings(_Model): + kind: Literal["local"] + + +class SSHTransportSettings(_Model): + kind: Literal["ssh"] + host: str + port: PositiveInt = 22 + user: str + known_hosts: str # host key verification is not optional + auth: Literal["key", "password", "agent"] = "key" + jump_host: str | None = None + connect_timeout_s: PositiveInt = 30 + + @model_validator(mode="after") + def _known_hosts_not_empty(self) -> Self: + if not self.known_hosts.strip(): + raise ValueError("known_hosts must hold the expected host keys") + return self + + +class BatchSettings(_BackendSettings): + kind: Literal["batch"] + transport: Annotated[ + LocalTransportSettings | SSHTransportSettings, Discriminator("kind") + ] + scheduler: Literal["slurm", "htcondor", "direct"] + + +BackendSettings = Annotated[ + ARCSettings | HTCondorCESettings | CloudSettings | BatchSettings, + Discriminator("kind"), +] + + +class FakeSettings(_Model): + """The in-memory fake. Internal: no public configuration reaches it.""" + + kind: Literal["fake"] = "fake" + resource: str = "fake://local" + + +_SETTINGS: TypeAdapter[BackendSettings] = TypeAdapter(BackendSettings) + + +def resolve(config: BackendConfig) -> BackendSettings: + """Turn the consumer's flat configuration into the settings of one backend. + + Raises: + pydantic.ValidationError: A field is missing, invalid, or does not + apply to the chosen kind. + + """ + data: dict[str, Any] = config.model_dump(exclude_none=True) + if data["kind"] == "cloud": + # Nothing to reshape: every field is the cloud's own, and any limit is + # left where the consumer wrote it, to be refused by its own name. + return _SETTINGS.validate_python(data) + + limits = {} + for name in LIMIT_FIELDS: + if name in data: + limits[name] = data.pop(name) + if limits: + data["limits"] = limits + + if data["kind"] == "batch": + # The transport's own fields, and the transport kind, move into one + # nested object; anything left over is checked against the backend. + transport = {} + for name in SSH_FIELDS: + if name in data: + transport[name] = data.pop(name) + if "transport" in data: + transport["kind"] = data.pop("transport") + if "auth" in data: + auth = data.pop("auth") + # An SSH connection authenticates one way, chosen here. + transport["auth"] = next(iter(auth)) if len(auth) == 1 else sorted(auth) + data["transport"] = transport + + return _SETTINGS.validate_python(data) diff --git a/src/intercede/_backends/fake.py b/src/intercede/_backends/fake.py new file mode 100644 index 0000000..aa6e78d --- /dev/null +++ b/src/intercede/_backends/fake.py @@ -0,0 +1,375 @@ +"""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 Callable, Self + +from intercede._backends.config import FakeSettings +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: connecting happens on entry. + 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 + + @classmethod + def from_settings(cls, settings: FakeSettings) -> Self: + """Build the fake from its settings; how the registry constructs it.""" + return cls(resource=settings.resource) + + # -- 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) + for index, copy in enumerate(copy_vars): + if clashes := spec.clashes_with(copy): + raise SpecificationRejectedError( + f"copy {index}: secrets {sorted(clashes)} are named like inputs" + ) + + # 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], filename: 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( + filename=filename 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, before anything runs. + self._check_container_support(spec) + + if not self.resource_staging: + self._check_can_fetch(spec.inputs) + if spec.outputs is not None: # no outputs means nothing to upload + self._check_can_upload(spec.outputs.members) + + 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. +_backend: type[JobBackend] = FakeBackend +_output: type[OutputRetriever] = FakeBackend +_kill: type[Killable] = FakeBackend +_purge: type[Purgeable] = FakeBackend +_diag: type[Diagnosable] = FakeBackend +_built_by_registry: Callable[[FakeSettings], JobBackend] = FakeBackend.from_settings diff --git a/src/intercede/config.py b/src/intercede/config.py new file mode 100644 index 0000000..433cb4f --- /dev/null +++ b/src/intercede/config.py @@ -0,0 +1,142 @@ +"""Backend configuration: how a consumer says which backend it wants. + +One flat model describes every backend. ``kind`` picks the backend; a ``batch`` +backend also picks a ``transport`` and a ``scheduler``. Only the fields that +apply to that choice may be set, and the model is checked when it is built, so +a mistake is reported here rather than at the first submit. + +Configuration describes the resource and is never secret: keys, passwords, +tokens and proxies come from the credential provider at run time. +The kinds are a closed set: they list what interCEde ships. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Literal, Self + +from pydantic import PositiveInt, ValidationError, model_validator + +from intercede._backends.config import resolve +from intercede.models import _Model + +__all__ = ["BackendConfig"] + + +class BackendConfig(_Model): + """Which backend to use, and how to reach it. + + Examples, as TOML:: + + kind = "arc" + endpoint = "https://arc.example.org:443" + auth = ["token", "proxy"] + + kind = "batch" + transport = "ssh" + scheduler = "slurm" + host = "batch.example.org" + user = "dirac" + known_hosts = "batch.example.org ssh-ed25519 AAAA..." + + kind = "cloud" + endpoint = "https://cloud.example.org:5000" + driver = "openstack" + vm_image = "name:alma9-pilot" + vm_flavor = "name:m1.large" + + Which fields apply: + + ============================ ========================================== + ``kind`` fields + ============================ ========================================== + ``arc``, ``htcondor-ce`` ``endpoint``, ``auth`` (``token``, ``proxy``) + ``cloud`` ``endpoint``, ``driver``, ``vm_image``, + ``vm_flavor``, ``driver_options``, + ``vm_networks``, ``vm_ssh_key``, + ``vm_max_lifetime_s`` + ``batch`` + ``local`` ``scheduler`` + ``batch`` + ``ssh`` ``scheduler``, ``host``, ``user``, + ``known_hosts``, ``port``, ``jump_host``, + ``connect_timeout_s``, ``auth`` (one of + ``key``, ``password``, ``agent``) + all but ``cloud`` ``queue``, ``container_launcher``, + ``max_output_bytes``, ``max_output_files``, + ``transfer_timeout_s`` + ============================ ========================================== + """ + + # What to use. ``kind`` is also what ``JobHandle.backend`` records. + # Three families: grid CEs that schedule by themselves + # ("arc", "htcondor-ce"), a cloud provider that boots a VM per job + # ("cloud"), and a batch system reached through a transport ("batch"). + kind: Literal["arc", "htcondor-ce", "cloud", "batch"] + transport: Literal["local", "ssh"] | None = None + scheduler: Literal["slurm", "htcondor", "direct"] | None = None + + # Where it is. + # A CE or cloud API: "https://arc.example.org:443", "ce.example.org:9619" + endpoint: str | None = None + host: str | None = None # an SSH host + port: PositiveInt | None = None + user: str | None = None + known_hosts: str | None = None # expected host keys; verification is not optional + jump_host: str | None = None # "[user@]host[:port]" + connect_timeout_s: PositiveInt | None = None + + # Which VM a cloud boots. The cloud's API keys come from the credential + # provider, so driver_options holds only non-secret driver arguments. + driver: str | None = ( + None # the driver library's name for the cloud, e.g. "openstack" + ) + driver_options: Mapping[str, str] | None = None + vm_image: str | None = None # an image ID, or "name:" + vm_flavor: str | None = None # a flavor ID, or "name:" + vm_networks: Sequence[str] | None = None + vm_ssh_key: str | None = None # the NAME of a key pair known to the cloud + vm_max_lifetime_s: PositiveInt | None = None + + # How to authenticate: which credentials the provider must supply, never + # the credentials themselves. Defaults to a token for a CE, a key for SSH. + auth: frozenset[str] | None = None + + # How to run jobs there. + queue: str | None = None # used when the spec does not name one + # An argv template wrapping the executable in a container, e.g. + # ["apptainer", "exec", "--bind", "/cvmfs", "{image}"]. + container_launcher: Sequence[str] | None = None + + # Deployment policy for output collection; defaults apply when unset. + max_output_bytes: PositiveInt | None = None + max_output_files: PositiveInt | None = None + transfer_timeout_s: PositiveInt | None = None + + @model_validator(mode="after") + def _fits_the_backend(self) -> Self: + try: + resolve(self) + except ValidationError as exc: + raise ValueError(_describe(self, exc)) from None + return self + + +def _describe(config: BackendConfig, exc: ValidationError) -> str: + """Report problems by the flat field names the consumer wrote.""" + fields = set(BackendConfig.model_fields) + lines = [] + for error in exc.errors(): + named = [part for part in error["loc"] if part in fields] + prefix = f"{named[-1]}: " if named else "" + message = error["msg"].removeprefix("Value error, ") + if error["type"] == "extra_forbidden": + message = f"does not apply to {_choice(config)}" + elif error["type"] in ("missing", "union_tag_not_found"): + message = f"required for {_choice(config)}" + lines.append(prefix + message) + return "; ".join(lines) + + +def _choice(config: BackendConfig) -> str: + if config.kind != "batch": + return f"kind={config.kind!r}" + return f"kind='batch' with transport={config.transport!r}" diff --git a/src/intercede/exceptions.py b/src/intercede/exceptions.py new file mode 100644 index 0000000..6b99d35 --- /dev/null +++ b/src/intercede/exceptions.py @@ -0,0 +1,44 @@ +"""Exceptions for failures of a whole operation. + +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", + "RegistryError", + "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.""" + + +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.""" + + +class RegistryError(InterCEdeError): + """The registry cannot provide what a valid configuration asks for. + + Either this version of interCEde does not ship it, or its registry entry is + broken: it cannot be imported, or what it builds does not satisfy its protocol. + """ diff --git a/src/intercede/models.py b/src/intercede/models.py new file mode 100644 index 0000000..818692e --- /dev/null +++ b/src/intercede/models.py @@ -0,0 +1,280 @@ +"""Data types of the InterCEde contract: what a consumer submits and what comes back.""" + +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 urllib.parse import quote + +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. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + +def _check_in_workdir(name: str) -> str: + """Reject names that would leave the job's working directory.""" + 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.""" + + 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. + + Every input lands in the job's working directory under its ``name``, so two + inputs cannot share a name: the spec is refused when it is built. + """ + + 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 + + @model_validator(mode="after") + def _unique_input_names(self) -> Self: + names = [ref.name for ref in self.inputs] + if twice := sorted({name for name in names if names.count(name) > 1}): + raise ValueError(f"inputs {twice} are given more than once") + return self + + def clashes_with(self, copy: CopyVars) -> frozenset[str]: + """Return the names of ``copy``'s secrets that an input of this spec also uses. + + A secret and an input with the same name would be staged to the same + file in the working directory. A backend refuses such a copy at submit. + + Args: + copy: The per-copy values to check against this spec. + + Returns: + The clashing names; empty when there are none. + + """ + return frozenset(copy.secrets) & {ref.name for ref in self.inputs} + + +# -------------------------------------------------------------------------- +# 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 + + @property + def key(self) -> str: + return ":".join(quote(part) for part in (self.backend, self.resource, self.id)) + + def __str__(self) -> str: + return self.key + + def __hash__(self) -> int: + return hash(self.key) + + +class JobStatus(StrEnum): + """Common job states. + + UNKNOWN is fixed: a job the backend no longer knows is reported as UNKNOWN, + never dropped. 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.""" + + filename: str # the file that was read, e.g. "stdout" or "job.log" + content: str + + +class OpOutcome(_Model, Generic[T]): + """The outcome of an operation for one job. + + 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..008265f --- /dev/null +++ b/src/intercede/protocols.py @@ -0,0 +1,205 @@ +"""The capability protocols of the InterCEde contract. + +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. + + Local input files are read before this returns: the backend never + depends on caller-owned paths afterwards. + + Args: + spec: What every copy shares: executable, inputs, outputs, resources. + copies: A count for identical copies, or one ``CopyVars`` per copy + when each copy needs its own environment or secrets. + + Returns: + The handles of the submitted copies and the reasons for refused + ones, both keyed by copy index: ``handles[i]`` belongs to copy ``i``. + + Raises: + SpecificationRejectedError: The backend cannot honour ``spec``, or a + copy has a secret named like one of the spec's inputs. Nothing + is submitted. + ResourceUnavailableError: The resource could not be reached, or + nothing could be submitted. + AuthenticationError: The resource rejected the credentials. + + """ + ... + + +@runtime_checkable +class StatusReporter(Protocol): + """Reports the status of jobs.""" + + async def get_status(self, handles: Sequence[JobHandle]) -> StatusMap: + """Return the current status of each job. + + Args: + handles: The jobs to ask about, as returned by ``submit``. + + Returns: + One status per handle. A job the backend does not know is + ``UNKNOWN``, never left out. + + Raises: + ResourceUnavailableError: The resource could not be reached. + AuthenticationError: The resource rejected the credentials. + + """ + ... + + +@runtime_checkable +class JobBackend(Submitter, StatusReporter, Protocol): + """A complete backend: what the registry returns. + + Backends are asynchronous context managers: construction does no I/O, + connections are opened on entry, 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. +# Each operation reports per job; a failure of the whole operation raises. +# -------------------------------------------------------------------------- + + +@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 finished job's kept outputs into ``dest``. + + Collection is repeatable until the job is purged. + + Args: + handles: The jobs to collect. + dest: A directory owned by the consumer. InterCEde writes every file + under it and never deletes it. + + Returns: + Per job, a manifest of the files written, or the reason the job's + output could not be collected (for example, it has not finished). + + """ + ... + + +@runtime_checkable +class Killable(Protocol): + """Cancels jobs.""" + + async def kill( + self, handles: Sequence[JobHandle] + ) -> Mapping[JobHandle, OpOutcome[None]]: + """Cancel the jobs that are still active. + + Args: + handles: The jobs to cancel. + + Returns: + Per job, whether it was cancelled. A job that already finished is + left alone and reported as a success with a reason. + + """ + ... + + +@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 the jobs' outputs and remote state without collecting them. + + Args: + handles: The jobs to release. + + Returns: + Per job, whether it was released. Afterwards its output can no + longer be collected. + + """ + ... + + +@runtime_checkable +class Diagnosable(Protocol): + """Reads files from jobs before they complete.""" + + async def get_diagnostics( + self, handles: Sequence[JobHandle], filename: str | None = None + ) -> Mapping[JobHandle, OpOutcome[Diagnostics]]: + """Read one file from each job while it runs; repeatable and non-destructive. + + Args: + handles: The jobs to read from. + filename: A file in the job's working directory, for example + ``"job.log"``. ``None`` reads the job's standard output. + + Returns: + Per job, the file's name and content, or the reason it could not be + read. + + """ + ... diff --git a/src/intercede/py.typed b/src/intercede/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/intercede/registry.py b/src/intercede/registry.py new file mode 100644 index 0000000..51f858e --- /dev/null +++ b/src/intercede/registry.py @@ -0,0 +1,179 @@ +"""Turn a backend configuration into a backend. + +The registry is closed: the tables below list every backend, transport and +scheduler interCEde ships, and nothing outside the library can add to them. +Each entry names its class as ``"module:Class"`` and is imported only when a +configuration asks for it, so installing interCEde never imports every backend. + +Every registered class provides a ``from_settings`` classmethod, which is how +the registry builds it from the validated settings of ``_backends.config``. +What it builds is checked against the relevant protocol before it is returned, +so a broken entry fails here, with a clear error, and not at the first call. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from functools import cache +from importlib import import_module +from typing import Any, get_args + +from pydantic import ValidationError + +from intercede._backends.batch.protocols import Scheduler, Transport +from intercede._backends.config import ( + BackendSettings, + BatchSettings, + FakeSettings, + resolve, +) +from intercede.config import BackendConfig +from intercede.exceptions import ConfigurationError, RegistryError +from intercede.protocols import JobBackend + +__all__ = ["available", "backend"] + +# Name -> "module:Class". A name is listed only once its class exists; a valid +# configuration asking for anything else gets a RegistryError naming what is +# available. Monolithic backends ("arc", "htcondor-ce") and provider backends +# ("cloud") are built from their own settings alone. "batch" is the composed +# backend: it is built from one transport and one scheduler, so combinations +# never need classes of their own. +_BACKENDS: dict[str, str] = { + "fake": "intercede._backends.fake:FakeBackend", # internal, for the test suites +} +_TRANSPORTS: dict[str, str] = {} +_SCHEDULERS: dict[str, str] = {} + +_GROUPS: dict[str, dict[str, str]] = { + "backend": _BACKENDS, + "transport": _TRANSPORTS, + "scheduler": _SCHEDULERS, +} + + +def backend(config: BackendConfig | Mapping[str, Any]) -> JobBackend: + """Build the backend a configuration describes. + + The backend is returned unopened: use it as ``async with registry.backend(config) as ce``. + Construction does no I/O, so building a backend for an unreachable resource succeeds. + + Args: + config: A ``BackendConfig``, or the same data as a mapping, for example + a parsed TOML table. + + Returns: + The backend, not yet opened. + + Raises: + ConfigurationError: The configuration is invalid. + RegistryError: The configuration is valid, but this version of + interCEde does not provide what it asks for. + + """ + if not isinstance(config, BackendConfig): + try: + config = BackendConfig.model_validate(config) + except ValidationError as exc: + raise ConfigurationError(str(exc)) from exc + return _build(resolve(config)) + + +def available() -> dict[str, tuple[str, ...]]: + """List what a configuration can ask for in this version, per group. + + Returns: + The names per group, sorted, for tooling. For example + ``{"backend": ("arc", "batch"), "transport": ("ssh",), ...}``. + + """ + public = { + "backend": _choices(BackendConfig.model_fields["kind"].annotation), + "transport": _choices(BackendConfig.model_fields["transport"].annotation), + "scheduler": _choices(BackendConfig.model_fields["scheduler"].annotation), + } + result = {} + for group, table in _GROUPS.items(): + names = [] + for name in table: + if name in public[group]: + names.append(name) + result[group] = tuple(sorted(names)) + + return result + + +def _choices(annotation: Any) -> set[str]: + """Collect the string values of a Literal, also inside ``Literal[...] | None``.""" + values: set[str] = set() + for arg in get_args(annotation): + if isinstance(arg, str): + values.add(arg) + else: + values |= _choices(arg) + return values + + +def _build(settings: BackendSettings | FakeSettings) -> JobBackend: + """Build a backend from validated settings. Internal: the test suites call it directly.""" + if not isinstance(settings, BatchSettings): + built = _load("backend", settings.kind).from_settings(settings) + result: JobBackend = _checked("backend", settings.kind, built, JobBackend) + return result + + transport_kind = settings.transport.kind + transport: Transport = _checked( + "transport", + transport_kind, + _load("transport", transport_kind).from_settings(settings.transport), + Transport, + ) + scheduler: Scheduler = _checked( + "scheduler", + settings.scheduler, + _load("scheduler", settings.scheduler).from_settings(settings), + Scheduler, + ) + built = _load("backend", "batch").from_settings( + settings, transport=transport, scheduler=scheduler + ) + composed: JobBackend = _checked("backend", "batch", built, JobBackend) + return composed + + +@cache +def _load(group: str, name: str) -> Any: + """Import the class registered under ``name``, and only that one.""" + table = _GROUPS[group] + if name not in table: + offered = ", ".join(available()[group]) or "none yet" + raise RegistryError( + f"{group} {name!r} is not available in this version of interCEde (available: {offered})" + ) + + target = table[name] + module_name, _, attribute = target.partition(":") + try: + cls = getattr(import_module(module_name), attribute) + except ( + ImportError, + AttributeError, + ValueError, + ) as exc: # ValueError: empty module name + raise RegistryError( + f"{group} {name!r} is registered as {target!r}, which cannot be imported" + ) from exc + if not callable(getattr(cls, "from_settings", None)): + raise RegistryError( + f"{group} {name!r} ({target}) has no from_settings() to build it with" + ) + return cls + + +def _checked(group: str, name: str, built: Any, protocol: type) -> Any: + """Refuse an object that does not satisfy its protocol.""" + if not isinstance(built, protocol): + raise RegistryError( + f"{group} {name!r} built a {type(built).__name__}, which is not a {protocol.__name__}" + ) + return built diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/_registry_dummies.py b/tests/unit/_registry_dummies.py new file mode 100644 index 0000000..5dc02fe --- /dev/null +++ b/tests/unit/_registry_dummies.py @@ -0,0 +1,90 @@ +"""Stand-ins registered by test_registry.py. + +Imported only through the registry, which is how the tests show that lookup is +lazy. Not collected by pytest: the file name does not start with ``test_``. +""" + +from __future__ import annotations + +from typing import Any + + +class _Lifecycle: + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *exc_info: object) -> None: ... + + +class DummyTransport(_Lifecycle): + def __init__(self, settings: Any) -> None: + self.settings = settings + + @classmethod + def from_settings(cls, settings: Any) -> DummyTransport: + return cls(settings) + + async def run(self, argv: Any) -> Any: ... + + async def put(self, local: Any, remote: Any) -> None: ... + + async def get(self, remote: Any, local: Any) -> None: ... + + +class DummyScheduler: + stages_own_files = False + + def __init__(self, settings: Any) -> None: + self.settings = settings + + @classmethod + def from_settings(cls, settings: Any) -> DummyScheduler: + return cls(settings) + + def submit_command(self, spec: Any, copies: Any, workdir: Any) -> Any: ... + + def parse_submit(self, raw: Any) -> Any: ... + + def status_command(self, ids: Any) -> Any: ... + + def parse_status(self, raw: Any, ids: Any) -> Any: ... + + def kill_command(self, ids: Any) -> Any: ... + + +class DummyBatchBackend(_Lifecycle): + def __init__(self, settings: Any, transport: Any, scheduler: Any) -> None: + self.settings = settings + self.transport = transport + self.scheduler = scheduler + + @classmethod + def from_settings( + cls, settings: Any, *, transport: Any, scheduler: Any + ) -> DummyBatchBackend: + return cls(settings, transport, scheduler) + + async def submit(self, spec: Any, copies: Any = 1) -> Any: ... + + async def get_status(self, handles: Any) -> Any: ... + + +class DummyBackend(_Lifecycle): + """A single-class backend, standing in for the monolithic and provider families.""" + + def __init__(self, settings: Any) -> None: + self.settings = settings + + @classmethod + def from_settings(cls, settings: Any) -> DummyBackend: + return cls(settings) + + async def submit(self, spec: Any, copies: Any = 1) -> Any: ... + + async def get_status(self, handles: Any) -> Any: ... + + +class BuildsTheWrongThing: + @classmethod + def from_settings(cls, *args: Any, **kwargs: Any) -> object: + return object() diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py new file mode 100644 index 0000000..a0903f3 --- /dev/null +++ b/tests/unit/conftest.py @@ -0,0 +1,39 @@ +"""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. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Callable + +import pytest + +from intercede import JobBackend +from intercede._backends.config import FakeSettings +from intercede.registry import _build + +# Built through the registry, so the suite also covers the registry path. +BACKENDS: dict[str, Callable[[], JobBackend]] = { + "fake": lambda: _build(FakeSettings()), +} + + +@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_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..7fd45c4 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,150 @@ +"""Backend configuration: one flat model, checked against the backend it names.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from pydantic import ValidationError + +from intercede import BackendConfig +from intercede._backends.config import ( + ARCSettings, + BatchSettings, + HTCondorCESettings, + LocalTransportSettings, + SSHTransportSettings, + resolve, +) + +SSH: dict[str, Any] = { + "kind": "batch", + "transport": "ssh", + "scheduler": "slurm", + "host": "batch.example.org", + "user": "dirac", + "known_hosts": "batch.example.org ssh-ed25519 AAAAC3Nz...", +} + + +def rejected(**data: Any) -> str: + """Return the error message for configuration that must be refused.""" + with pytest.raises(ValidationError) as exc_info: + BackendConfig.model_validate(data) + return str(exc_info.value) + + +class TestResolve: + """The flat model becomes the settings of exactly one backend.""" + + def test_arc(self): + settings = resolve( + BackendConfig(kind="arc", endpoint="https://arc.example.org:443") + ) + assert isinstance(settings, ARCSettings) + assert settings.auth == {"token"} + + def test_htcondor_ce_with_proxy(self): + config = BackendConfig( + kind="htcondor-ce", endpoint="ce:9619", auth=frozenset({"token", "proxy"}) + ) + settings = resolve(config) + assert isinstance(settings, HTCondorCESettings) + assert settings.auth == {"token", "proxy"} + + def test_batch_over_ssh(self): + settings = resolve(BackendConfig.model_validate(SSH)) + assert isinstance(settings, BatchSettings) + assert settings.scheduler == "slurm" + assert isinstance(settings.transport, SSHTransportSettings) + assert settings.transport.host == "batch.example.org" + assert settings.transport.port == 22 + assert settings.transport.auth == "key" + + def test_batch_on_this_host(self): + settings = resolve( + BackendConfig(kind="batch", transport="local", scheduler="direct") + ) + assert isinstance(settings, BatchSettings) + assert isinstance(settings.transport, LocalTransportSettings) + + def test_ssh_auth_mode(self): + settings = resolve(BackendConfig.model_validate({**SSH, "auth": ["agent"]})) + assert isinstance(settings, BatchSettings) + assert isinstance(settings.transport, SSHTransportSettings) + assert settings.transport.auth == "agent" + + def test_limits(self): + settings = resolve( + BackendConfig(kind="arc", endpoint="https://a:443", max_output_files=500) + ) + assert settings.limits.max_output_files == 500 + assert settings.limits.transfer_timeout_s == 3600 + + +class TestRejected: + def test_unknown_kind(self): + assert "kind" in rejected(kind="lsf") + + def test_typo_is_not_silently_ignored(self): + assert "endpont" in rejected(kind="arc", endpoint="https://a:443", endpont="x") + + @pytest.mark.parametrize( + ("data", "field"), + [ + ({"kind": "arc", "endpoint": "https://a:443", "host": "h"}, "host"), + ( + {"kind": "arc", "endpoint": "https://a:443", "scheduler": "slurm"}, + "scheduler", + ), + ({**SSH, "endpoint": "https://a:443"}, "endpoint"), + ( + { + "kind": "batch", + "transport": "local", + "scheduler": "slurm", + "host": "h", + }, + "host", + ), + ], + ) + def test_fields_that_do_not_apply_are_refused( + self, data: dict[str, Any], field: str + ): + assert f"{field}: does not apply to" in rejected(**data) + + @pytest.mark.parametrize( + ("data", "field"), + [ + ({"kind": "arc"}, "endpoint"), + ({"kind": "batch", "scheduler": "slurm"}, "transport"), + ({"kind": "batch", "transport": "local"}, "scheduler"), + ({**SSH, "user": None}, "user"), + ({**SSH, "known_hosts": None}, "known_hosts"), + ], + ) + def test_missing_fields_are_named(self, data: dict[str, Any], field: str): + assert f"{field}: required for" in rejected(**data) + + def test_ssh_requires_host_key_verification(self): + assert "known_hosts" in rejected(**{**SSH, "known_hosts": " "}) + + @pytest.mark.parametrize("auth", [["token"], ["key", "password"]]) + def test_ssh_authenticates_one_supported_way(self, auth: list[str]): + assert "auth" in rejected(**{**SSH, "auth": auth}) + + def test_htcondor_ce_needs_a_token(self): + assert "token" in rejected( + kind="htcondor-ce", endpoint="ce:9619", auth=["proxy"] + ) + + def test_ce_auth_is_token_or_proxy(self): + assert "auth" in rejected(kind="arc", endpoint="https://a:443", auth=["key"]) + + def test_container_launcher_must_name_the_image(self): + base = {"kind": "batch", "transport": "local", "scheduler": "slurm"} + assert "{image}" in rejected(**base, container_launcher=["apptainer", "exec"]) + BackendConfig.model_validate( + {**base, "container_launcher": ["apptainer", "exec", "{image}"]} + ) diff --git a/tests/unit/test_conformance.py b/tests/unit/test_conformance.py new file mode 100644 index 0000000..5d6f51e --- /dev/null +++ b/tests/unit/test_conformance.py @@ -0,0 +1,140 @@ +"""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. 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, + SpecificationRejectedError, + 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) + + +async def test_a_secret_named_like_an_input_is_refused( + backend: JobBackend, tmp_path: Path +): + source = tmp_path / "pilot.cfg" + source.write_text("data") + spec = SubmissionSpec( + executable="/bin/true", inputs=[FileRef(name="pilot.cfg", source=str(source))] + ) + clashing = [CopyVars(), CopyVars(secrets={"pilot.cfg": b"s"})] + with pytest.raises(SpecificationRejectedError, match="pilot.cfg"): + await backend.submit(spec, clashing) diff --git a/tests/unit/test_fake_backend.py b/tests/unit/test_fake_backend.py new file mode 100644 index 0000000..5f8f452 --- /dev/null +++ b/tests/unit/test_fake_backend.py @@ -0,0 +1,130 @@ +"""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_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" + named = (await backend.get_diagnostics([handle], filename="job.log"))[handle] + assert outcome.value.filename == "stdout" # the default + assert named.value is not None and named.value.filename == "job.log" + + +@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..fe11e1b --- /dev/null +++ b/tests/unit/test_models.py @@ -0,0 +1,93 @@ +"""Validation rules of the contract data types.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from intercede import ( + CopyVars, + FileRef, + JobHandle, + JobOutput, + OpOutcome, + OutputMember, + Submission, + SubmissionSpec, +) + + +def handle(job_id: str = "1", **fields: object) -> JobHandle: + return JobHandle.model_validate( + {"backend": "fake", "resource": "r", "id": job_id, **fields} + ) + + +class TestJobHandle: + def test_a_stored_handle_round_trips(self): + stored = handle(routing={"schedd": "a"}) + assert JobHandle.model_validate_json(stored.model_dump_json()) == stored + + def test_a_newer_format_fails_loudly(self): + with pytest.raises(ValidationError): + handle(version=2) + with pytest.raises(ValidationError): + handle(extra="x") + + def test_the_same_id_on_two_resources_is_two_jobs(self): + assert len({handle(resource="a"), handle(resource="b")}) == 2 + + +@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) + + +def test_secret_content_never_shows(): + copy = CopyVars(secrets={"pilot.secret": b"TOPSECRET"}) + for shown in (repr(copy), str(copy), copy.model_dump_json()): + assert "TOPSECRET" not in shown + + +def test_a_copy_is_either_submitted_or_refused(): + with pytest.raises(ValidationError): + Submission(handles={0: handle()}, failures={0: "refused"}) + + +class TestOpOutcome: + def test_success_carries_a_value_unless_there_is_none(self): + assert OpOutcome[None](ok=True, reason="already finished").ok + with pytest.raises(ValidationError): + OpOutcome[JobOutput](ok=True) + + def test_failure_has_a_reason_and_no_value(self): + with pytest.raises(ValidationError): + OpOutcome[None](ok=False) + with pytest.raises(ValidationError): + OpOutcome[JobOutput](ok=False, reason="x", value=JobOutput()) + + +def test_an_input_name_is_given_once(): + # Two sources for one file in the working directory: which one would win? + twice = [FileRef(name="in", source="a"), FileRef(name="in", source="b")] + with pytest.raises(ValidationError, match="more than once"): + SubmissionSpec(executable="x", inputs=twice) + + +def test_secrets_that_clash_with_inputs_are_named(): + spec = SubmissionSpec(executable="x", inputs=[FileRef(name="in", source="a")]) + assert spec.clashes_with(CopyVars(secrets={"in": b"s", "other": b"s"})) == {"in"} + assert spec.clashes_with(CopyVars()) == frozenset() diff --git a/tests/unit/test_protocols.py b/tests/unit/test_protocols.py new file mode 100644 index 0000000..cb80a90 --- /dev/null +++ b/tests/unit/test_protocols.py @@ -0,0 +1,57 @@ +"""Which capabilities a backend has, as a consumer finds out with ``isinstance``.""" + +from __future__ import annotations + +import pytest + +import intercede +from intercede import ( + Diagnosable, + JobBackend, + Killable, + OutputRetriever, + Purgeable, +) +from intercede._backends.fake import FakeBackend + + +class Minimal: + """A backend with only the essential capabilities, like a cloud provider.""" + + async def submit(self, spec, copies=1): ... + + async def get_status(self, handles): ... + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc_info): ... + + +@pytest.mark.parametrize( + "capability", [JobBackend, OutputRetriever, Killable, Purgeable, Diagnosable] +) +def test_the_fake_has_every_capability(capability: type): + assert isinstance(FakeBackend(), capability) + + +def test_a_minimal_backend_has_only_the_essentials(): + assert isinstance(Minimal(), JobBackend) + assert not isinstance( + Minimal(), OutputRetriever | Killable | Purgeable | Diagnosable + ) + + +def test_a_backend_needs_a_lifecycle(): + class NoLifecycle: + async def submit(self, spec, copies=1): ... + + async def get_status(self, handles): ... + + assert not isinstance(NoLifecycle(), JobBackend) + + +def test_public_api_is_exported(): + for name in intercede.__all__: + assert hasattr(intercede, name), name + assert "Transport" not in intercede.__all__ # internal to the batch family diff --git a/tests/unit/test_registry.py b/tests/unit/test_registry.py new file mode 100644 index 0000000..3827216 --- /dev/null +++ b/tests/unit/test_registry.py @@ -0,0 +1,173 @@ +"""The registry: closed tables, lazy imports, and a protocol check on everything it builds.""" + +from __future__ import annotations + +import subprocess +import sys +from collections.abc import Iterator +from typing import Any + +import pytest + +from intercede import ( + BackendConfig, + ConfigurationError, + JobBackend, + RegistryError, + registry, +) +from intercede._backends.config import CloudSettings, FakeSettings +from intercede.registry import _build, _load + +DUMMIES = "tests.unit._registry_dummies" + +CLOUD: dict[str, Any] = { + "kind": "cloud", + "endpoint": "https://cloud.example.org:5000", + "driver": "openstack", + "vm_image": "name:alma9", + "vm_flavor": "name:m1.large", +} +SSH_SLURM: dict[str, Any] = { + "kind": "batch", + "transport": "ssh", + "scheduler": "slurm", + "host": "batch.example.org", + "user": "dirac", + "known_hosts": "batch.example.org ssh-ed25519 AAAAC3Nz...", +} +LOCAL_SLURM: dict[str, Any] = { + "kind": "batch", + "transport": "local", + "scheduler": "slurm", +} + + +@pytest.fixture(autouse=True) +def fresh_cache() -> Iterator[None]: + """Each test sees the tables as it left them, not classes cached by another test.""" + _load.cache_clear() + yield + _load.cache_clear() + + +@pytest.fixture +def composed(monkeypatch: pytest.MonkeyPatch) -> None: + """Register stand-ins for the composed backend, which has no real parts yet.""" + monkeypatch.setitem(registry._BACKENDS, "batch", f"{DUMMIES}:DummyBatchBackend") + monkeypatch.setitem(registry._TRANSPORTS, "ssh", f"{DUMMIES}:DummyTransport") + monkeypatch.setitem(registry._TRANSPORTS, "local", f"{DUMMIES}:DummyTransport") + monkeypatch.setitem(registry._SCHEDULERS, "slurm", f"{DUMMIES}:DummyScheduler") + + +class TestBuilds: + async def test_the_fake_for_the_test_suites(self): + async with _build(FakeSettings(resource="fake://test")) as backend: + assert isinstance(backend, JobBackend) + assert backend.resource == "fake://test" + + @pytest.mark.parametrize( + "config", + [CLOUD, BackendConfig.model_validate(CLOUD)], + ids=["mapping", "object"], + ) + def test_a_single_class_backend_gets_its_settings( + self, monkeypatch: pytest.MonkeyPatch, config: Any + ): + monkeypatch.setitem(registry._BACKENDS, "cloud", f"{DUMMIES}:DummyBackend") + built: Any = registry.backend(config) + assert type(built).__name__ == "DummyBackend" + assert isinstance(built.settings, CloudSettings) + + def test_a_batch_backend_is_composed_of_its_parts(self, composed: None): + built: Any = registry.backend({**SSH_SLURM, "queue": "short"}) + assert type(built).__name__ == "DummyBatchBackend" + # The transport gets only its own settings; the scheduler the batch settings. + assert built.transport.settings.kind == "ssh" + assert built.transport.settings.host == "batch.example.org" + assert built.scheduler.settings.queue == "short" + + +class TestRefused: + def test_invalid_configuration_keeps_the_details(self): + with pytest.raises(ConfigurationError) as exc_info: + registry.backend({"kind": "arc"}) + assert "endpoint: required for kind='arc'" in str(exc_info.value) + + @pytest.mark.parametrize( + "config", + [{"kind": "arc", "endpoint": "https://arc.example.org:443"}, CLOUD], + ids=["arc", "cloud"], + ) + def test_valid_but_not_shipped_yet(self, config: dict[str, Any]): + with pytest.raises( + RegistryError, match=f"backend '{config['kind']}' is not available" + ): + registry.backend(config) + + def test_composed_part_not_shipped_yet(self): + with pytest.raises(RegistryError, match="transport 'ssh' is not available"): + registry.backend(SSH_SLURM) + + def test_an_entry_that_cannot_be_imported( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", "tests.unit.does_not_exist:X") + with pytest.raises(RegistryError, match="cannot be imported") as exc_info: + registry.backend(SSH_SLURM) + assert isinstance(exc_info.value.__cause__, ImportError) + + @pytest.mark.parametrize( + ("table", "name", "protocol"), + [ + ("_TRANSPORTS", "local", "Transport"), + ("_SCHEDULERS", "slurm", "Scheduler"), + ("_BACKENDS", "batch", "JobBackend"), + ], + ) + def test_what_breaks_its_protocol_is_refused( + self, + composed: None, + monkeypatch: pytest.MonkeyPatch, + table: str, + name: str, + protocol: str, + ): + monkeypatch.setitem( + getattr(registry, table), name, f"{DUMMIES}:BuildsTheWrongThing" + ) + with pytest.raises(RegistryError, match=f"which is not a {protocol}"): + registry.backend(LOCAL_SLURM) + + +class TestAvailable: + def test_shipped_today(self): + assert registry.available() == {"backend": (), "transport": (), "scheduler": ()} + + def test_lists_only_what_a_configuration_can_ask_for( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._SCHEDULERS, "lsf", f"{DUMMIES}:DummyScheduler") + listing = registry.available() + assert listing["transport"] == ("local", "ssh") + assert "lsf" not in listing["scheduler"] + assert "fake" not in listing["backend"] + + +class TestLazyImports: + def test_importing_intercede_imports_no_backend(self): + code = ( + "import sys, intercede, intercede.registry;" + "loaded = [m for m in sys.modules if m.startswith(('intercede._backends.fake', 'tests.'))];" + "assert not loaded, loaded" + ) + subprocess.run([sys.executable, "-c", code], check=True) # noqa: S603 (constant input) + + def test_only_the_requested_entry_is_imported( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.delitem(sys.modules, DUMMIES, raising=False) + registry.available() + assert DUMMIES not in sys.modules # listing imports nothing + registry.backend(LOCAL_SLURM) + assert DUMMIES in sys.modules