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..824c73e --- /dev/null +++ b/docs/user/api-reference.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/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..cca1417 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)", @@ -35,13 +37,11 @@ Repository = "https://github.com/DIRACGrid/intercede" testing = [ "pytest", "pytest-asyncio", + "pytest-cov", "pytest-xdist", "ty", ] -[tool.setuptools.packages.find] -where = ["src"] - [tool.pytest.ini_options] pythonpath = ["src"] testpaths = ["tests"] @@ -51,6 +51,23 @@ markers = [ "xfail_backend: strict xfail a test on the named backend fixture(s)", ] +[tool.coverage.run] +branch = true +patch = ["subprocess"] # also measure the tests that start a fresh interpreter +source = ["intercede"] + +[tool.coverage.report] +show_missing = true +skip_covered = true +exclude_also = [ + "if TYPE_CHECKING:", + "^\\s*\\.\\.\\.$", # protocol method bodies +] + +[tool.mypy] +strict = true +plugins = ["pydantic.mypy"] +files = ["src"] [tool.ruff] src = ["src", "tests"] @@ -113,6 +130,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 +143,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..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/config.py b/src/intercede/_backends/config.py new file mode 100644 index 0000000..70fc756 --- /dev/null +++ b/src/intercede/_backends/config.py @@ -0,0 +1,204 @@ +"""What each backend needs from the public ``BackendConfig`` (IC-ADR-001 §5). + +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 (IC-ADR-001 §3). 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 (IC-ADR-003 §2): 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 (IC-ADR-001 §4). + + 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 (IC-ADR-003 §7). + """ + + 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 (IC-ADR-003 §6) + 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..89dfd5e --- /dev/null +++ b/src/intercede/_backends/fake.py @@ -0,0 +1,368 @@ +"""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 (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 + + @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) + + # 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: # 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 (IC-ADR-001 §4). +_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..c14676a --- /dev/null +++ b/src/intercede/config.py @@ -0,0 +1,142 @@ +"""Backend configuration: how a consumer says which backend it wants (IC-ADR-001 §5). + +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 (IC-ADR-003). +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 (IC-ADR-001 §4): 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}"] (IC-ADR-001 §2.4). + container_launcher: Sequence[str] | None = None + + # Deployment policy for output collection (IC-ADR-001 §3); 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..e9bcf59 --- /dev/null +++ b/src/intercede/exceptions.py @@ -0,0 +1,44 @@ +"""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", + "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 (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.""" + + +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..eb22377 --- /dev/null +++ b/src/intercede/models.py @@ -0,0 +1,256 @@ +"""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 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 (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 + + @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. + + 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/src/intercede/registry.py b/src/intercede/registry.py new file mode 100644 index 0000000..72dfdd3 --- /dev/null +++ b/src/intercede/registry.py @@ -0,0 +1,173 @@ +"""Turn a backend configuration into a backend (IC-ADR-001 §5). + +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. + + 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. + + For tooling, e.g. ``{"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 (IC-ADR-001 §5).""" + 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/test_smoke.py b/tests/test_smoke.py index 9773f64..5fd2a18 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -12,3 +12,19 @@ def test_package_imports(): def test_version_is_defined(): assert isinstance(intercede.version, str) assert intercede.version + + +def test_version_falls_back_when_not_installed(): + """Running from a source checkout, with no package metadata, still imports.""" + import subprocess + import sys + + code = ( + "import importlib.metadata as metadata\n" + "def missing(name):\n" + " raise metadata.PackageNotFoundError(name)\n" + "metadata.version = missing\n" + "import intercede\n" + "assert intercede.__version__ == intercede.version == 'unknown', intercede.version\n" + ) + subprocess.run([sys.executable, "-c", code], check=True) # noqa: S603 (constant input) 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..5adfd33 --- /dev/null +++ b/tests/unit/_registry_dummies.py @@ -0,0 +1,103 @@ +"""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. + + It records the settings it was built from and every call that reaches it, + so a test can show that calls on the returned object run this class's code. + """ + + def __init__(self, settings: Any) -> None: + self.settings = settings + self.calls: list[str] = [] + + @classmethod + def from_settings(cls, settings: Any) -> DummyBackend: + return cls(settings) + + async def submit(self, spec: Any, copies: Any = 1) -> Any: + self.calls.append("submit") + return f"submitted by {self.settings.kind}" + + async def get_status(self, handles: Any) -> Any: + self.calls.append("get_status") + return {} + + +class NoFromSettings(DummyTransport): + from_settings = None # type: ignore[assignment] + + +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..53e7109 --- /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 (IC-ADR-002). +""" + +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..e670fc1 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,572 @@ +"""Backend configuration: one flat model, checked against the backend it names.""" + +from __future__ import annotations + +import tomllib +from typing import Any + +import pytest +from pydantic import ValidationError + +from intercede import BackendConfig +from intercede._backends.config import ( + ARCSettings, + BatchSettings, + CloudSettings, + 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...", +} + + +CLOUD: dict[str, Any] = { + "kind": "cloud", + "endpoint": "https://cloud.example.org:5000", + "driver": "openstack", + "vm_image": "name:alma9-pilot", + "vm_flavor": "name:m1.large", +} + + +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 + + +def test_loads_from_toml(): + data = tomllib.loads( + """ + kind = "batch" + transport = "ssh" + scheduler = "slurm" + host = "batch.example.org" + user = "dirac" + known_hosts = "batch.example.org ssh-ed25519 AAAAC3Nz..." + queue = "short" + """ + ) + assert BackendConfig.model_validate(data).queue == "short" + + +def test_round_trips_through_json(): + config = BackendConfig.model_validate(SSH) + assert BackendConfig.model_validate_json(config.model_dump_json()) == config + + +class TestCloud: + """The provider family: a cloud boots one VM per job, with no queue to collect from.""" + + def test_resolves(self): + config = BackendConfig.model_validate( + { + **CLOUD, + "driver_options": {"ex_tenant_name": "dirac"}, + "vm_networks": ["public"], + } + ) + settings = resolve(config) + assert isinstance(settings, CloudSettings) + assert settings.driver_options == {"ex_tenant_name": "dirac"} + assert settings.vm_max_lifetime_s == 14 * 24 * 3600 + + @pytest.mark.parametrize("field", ["driver", "vm_image", "vm_flavor", "endpoint"]) + def test_required_fields(self, field: str): + assert f"{field}: required for kind='cloud'" in rejected( + **{**CLOUD, field: None} + ) + + @pytest.mark.parametrize( + ("field", "value"), + [ + ("queue", "short"), + ("max_output_files", 10), + ("container_launcher", ["apptainer", "exec", "{image}"]), + ("auth", ["token"]), + ("host", "h"), + ], + ) + def test_fields_of_other_families_do_not_apply(self, field: str, value: Any): + assert f"{field}: does not apply to kind='cloud'" in rejected( + **{**CLOUD, field: value} + ) + + @pytest.mark.parametrize( + "option", ["secret", "key", "api_key", "os_password", "auth_token"] + ) + def test_driver_options_hold_no_secrets(self, option: str): + assert "credential provider" in rejected( + **{**CLOUD, "driver_options": {option: "x"}} + ) + + def test_vm_fields_do_not_apply_to_a_ce(self): + assert "vm_image: does not apply" in rejected( + kind="arc", endpoint="https://a:443", vm_image="name:x" + ) + + +class TestRejected: + def test_unknown_kind(self): + assert "kind" in rejected(kind="lsf") + + def test_unknown_scheduler(self): + assert "scheduler" in rejected(**{**SSH, "scheduler": "pbs"}) + + 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_limits_must_be_positive(self): + assert "max_output_files" in rejected( + kind="arc", endpoint="https://a:443", max_output_files=0 + ) + + 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}"]} + ) + + @pytest.mark.parametrize("field", ["password", "private_key", "token", "proxy"]) + def test_there_is_nowhere_to_put_a_secret(self, field: str): + # Secrets come from the credential provider, never from configuration (IC-ADR-003). + assert field in rejected(**{**SSH, field: "x"}) + + +def test_configuration_is_immutable(): + config = BackendConfig(kind="arc", endpoint="https://a:443") + with pytest.raises(ValidationError): + config.endpoint = "https://b:443" # type: ignore[misc] + + +# -------------------------------------------------------------------------- +# Every field against every kind +# -------------------------------------------------------------------------- + +# The smallest valid configuration of each backend shape. +BASES: dict[str, dict[str, Any]] = { + "arc": {"kind": "arc", "endpoint": "https://arc.example.org:443"}, + "htcondor-ce": {"kind": "htcondor-ce", "endpoint": "ce.example.org:9619"}, + "cloud": CLOUD, + "batch-local": {"kind": "batch", "transport": "local", "scheduler": "slurm"}, + "batch-ssh": SSH, +} + +# Which flat fields each shape accepts (besides "kind"). +_SHARED = { + "queue", + "container_launcher", + "max_output_bytes", + "max_output_files", + "transfer_timeout_s", +} +APPLIES: dict[str, set[str]] = { + "arc": {"endpoint", "auth"} | _SHARED, + "htcondor-ce": {"endpoint", "auth"} | _SHARED, + "cloud": { + "endpoint", + "driver", + "driver_options", + "vm_image", + "vm_flavor", + "vm_networks", + "vm_ssh_key", + "vm_max_lifetime_s", + }, + "batch-local": {"transport", "scheduler"} | _SHARED, + "batch-ssh": { + "transport", + "scheduler", + "host", + "port", + "user", + "known_hosts", + "jump_host", + "connect_timeout_s", + "auth", + } + | _SHARED, +} + +# A valid value for every flat field. +VALUES: dict[str, Any] = { + "transport": "local", + "scheduler": "htcondor", + "endpoint": "https://other.example.org:443", + "host": "other.example.org", + "port": 2222, + "user": "pilot", + "known_hosts": "other.example.org ssh-ed25519 AAAAC3Nz...", + "jump_host": "gateway.example.org", + "connect_timeout_s": 5, + "driver": "ec2", + "driver_options": {"ex_tenant_name": "dirac"}, + "vm_image": "name:other", + "vm_flavor": "name:m1.small", + "vm_networks": ["public", "private"], + "vm_ssh_key": "pilot-keypair", + "vm_max_lifetime_s": 3600, + "auth": ["token"], + "queue": "long", + "container_launcher": ["apptainer", "exec", "{image}"], + "max_output_bytes": 1024, + "max_output_files": 7, + "transfer_timeout_s": 60, +} + + +def value_for(shape: str, field: str) -> Any: + """Return a value of ``field`` that is valid for ``shape``.""" + if field == "auth" and shape == "batch-ssh": + return ["password"] + if field == "transport" and shape == "batch-ssh": + return "ssh" + return VALUES[field] + + +def test_matrix_covers_every_flat_field(): + """Adding a field to BackendConfig without adding it here fails this test.""" + assert set(VALUES) == set(BackendConfig.model_fields) - {"kind"} + + +def test_matrix_covers_every_kind(): + kinds = {base["kind"] for base in BASES.values()} + assert kinds == {"arc", "htcondor-ce", "cloud", "batch"} + + +NOT_APPLYING = [ + (shape, field) + for shape in BASES + for field in sorted(VALUES) + if field not in APPLIES[shape] +] +APPLYING = [ + (shape, field) + for shape in BASES + for field in sorted(APPLIES[shape]) + if field not in BASES[shape] +] + + +@pytest.mark.parametrize(("shape", "field"), NOT_APPLYING) +def test_every_field_is_refused_where_it_does_not_apply(shape: str, field: str): + message = rejected(**{**BASES[shape], field: VALUES[field]}) + assert f"{field}: does not apply to kind=" in message + + +@pytest.mark.parametrize(("shape", "field"), APPLYING) +def test_every_optional_field_is_accepted_where_it_applies(shape: str, field: str): + config = BackendConfig.model_validate( + {**BASES[shape], field: value_for(shape, field)} + ) + assert resolve(config).kind == BASES[shape]["kind"] + + +@pytest.mark.parametrize(("shape", "field"), APPLYING) +def test_every_optional_field_reaches_the_settings(shape: str, field: str): + """A value the consumer sets is never lost on its way to the backend.""" + value = value_for(shape, field) + settings = resolve(BackendConfig.model_validate({**BASES[shape], field: value})) + if field in ("max_output_bytes", "max_output_files", "transfer_timeout_s"): + found = getattr(settings.limits, field) # type: ignore[union-attr] + elif shape == "batch-ssh" and field in ( + "host", + "port", + "user", + "known_hosts", + "jump_host", + "connect_timeout_s", + ): + found = getattr(settings.transport, field) # type: ignore[union-attr] + elif shape == "batch-ssh" and field == "auth": + found = [settings.transport.auth] # type: ignore[union-attr] + else: + found = getattr(settings, field) + if isinstance(value, list) and field != "auth": + assert list(found) == value + elif field == "auth" and shape != "batch-ssh": + assert found == set(value) + else: + assert found == value + + +REQUIRED = [ + ("arc", "endpoint"), + ("htcondor-ce", "endpoint"), + ("cloud", "endpoint"), + ("cloud", "driver"), + ("cloud", "vm_image"), + ("cloud", "vm_flavor"), + ("batch-local", "transport"), + ("batch-local", "scheduler"), + ("batch-ssh", "scheduler"), + ("batch-ssh", "host"), + ("batch-ssh", "user"), + ("batch-ssh", "known_hosts"), +] + + +@pytest.mark.parametrize(("shape", "field"), REQUIRED) +def test_every_required_field_is_named_when_missing(shape: str, field: str): + data = {name: value for name, value in BASES[shape].items() if name != field} + assert f"{field}: required for" in rejected(**data) + + +@pytest.mark.parametrize("shape", sorted(BASES)) +def test_every_shape_round_trips_through_json(shape: str): + config = BackendConfig.model_validate(BASES[shape]) + assert BackendConfig.model_validate_json(config.model_dump_json()) == config + + +# -------------------------------------------------------------------------- +# Defaults, values and messages +# -------------------------------------------------------------------------- + + +class TestDefaults: + def test_ce_authenticates_with_a_token(self): + for shape in ("arc", "htcondor-ce"): + settings = resolve(BackendConfig.model_validate(BASES[shape])) + assert settings.auth == {"token"} # type: ignore[union-attr] + + def test_ssh(self): + settings = resolve(BackendConfig.model_validate(SSH)) + assert isinstance(settings, BatchSettings) + transport = settings.transport + assert isinstance(transport, SSHTransportSettings) + assert (transport.port, transport.auth, transport.connect_timeout_s) == ( + 22, + "key", + 30, + ) + assert transport.jump_host is None + + def test_limits(self): + settings = resolve(BackendConfig.model_validate(BASES["arc"])) + assert settings.limits.max_output_bytes == 10 * 2**30 # type: ignore[union-attr] + assert settings.limits.max_output_files == 10_000 # type: ignore[union-attr] + assert settings.limits.transfer_timeout_s == 3600 # type: ignore[union-attr] + + def test_setting_one_limit_keeps_the_other_defaults(self): + config = BackendConfig.model_validate({**BASES["arc"], "transfer_timeout_s": 5}) + limits = resolve(config).limits # type: ignore[union-attr] + assert (limits.transfer_timeout_s, limits.max_output_files) == (5, 10_000) + + def test_cloud(self): + settings = resolve(BackendConfig.model_validate(CLOUD)) + assert isinstance(settings, CloudSettings) + assert settings.driver_options == {} + assert tuple(settings.vm_networks) == () + assert settings.vm_ssh_key is None + assert settings.vm_max_lifetime_s == 14 * 24 * 3600 + + def test_shared_fields(self): + settings = resolve(BackendConfig.model_validate(BASES["batch-local"])) + assert settings.queue is None # type: ignore[union-attr] + assert settings.container_launcher is None # type: ignore[union-attr] + + +class TestAuthValues: + @pytest.mark.parametrize("auth", [["token"], ["proxy"], ["token", "proxy"]]) + def test_arc_accepts_any_non_empty_combination(self, auth: list[str]): + settings = resolve(BackendConfig.model_validate({**BASES["arc"], "auth": auth})) + assert settings.auth == set(auth) # type: ignore[union-attr] + + def test_arc_needs_some_credential(self): + assert "auth must name at least one" in rejected(**BASES["arc"], auth=[]) + + @pytest.mark.parametrize("auth", [["token"], ["token", "proxy"]]) + def test_htcondor_ce_accepts_token_with_or_without_proxy(self, auth: list[str]): + config = BackendConfig.model_validate({**BASES["htcondor-ce"], "auth": auth}) + assert resolve(config).auth == set(auth) # type: ignore[union-attr] + + @pytest.mark.parametrize("auth", [[], ["proxy"]]) + def test_htcondor_ce_refuses_without_token(self, auth: list[str]): + assert "must include 'token'" in rejected(**BASES["htcondor-ce"], auth=auth) + + @pytest.mark.parametrize("auth", [["key"], ["password"], ["agent"]]) + def test_ssh_accepts_each_mode(self, auth: list[str]): + settings = resolve(BackendConfig.model_validate({**SSH, "auth": auth})) + assert settings.transport.auth == auth[0] # type: ignore[union-attr] + + @pytest.mark.parametrize("auth", [[], ["token"], ["proxy"], ["key", "agent"]]) + def test_ssh_refuses_anything_else(self, auth: list[str]): + assert "auth" in rejected(**SSH, auth=auth) + + @pytest.mark.parametrize("shape", ["arc", "htcondor-ce"]) + @pytest.mark.parametrize("auth", [["key"], ["password"], ["agent"], ["x509"]]) + def test_ce_refuses_ssh_and_unknown_modes(self, shape: str, auth: list[str]): + assert "auth" in rejected(**BASES[shape], auth=auth) + + +POSITIVE = [ + "port", + "connect_timeout_s", + "vm_max_lifetime_s", + "max_output_bytes", + "max_output_files", + "transfer_timeout_s", +] + + +@pytest.mark.parametrize("field", POSITIVE) +@pytest.mark.parametrize("value", [0, -1]) +def test_numbers_must_be_positive(field: str, value: int): + assert field in rejected(**{**SSH, field: value}) + + +@pytest.mark.parametrize("field", ["kind", "transport", "scheduler"]) +def test_choices_are_case_sensitive(field: str): + assert field in rejected(**{**SSH, field: SSH[field].upper()}) + + +def test_every_problem_is_reported_at_once(): + message = rejected(kind="batch", transport="ssh", scheduler="slurm") + assert "host: required" in message + assert "user: required" in message + assert "known_hosts: required" in message + assert message.count("; ") >= 2 + + +def test_message_has_no_pydantic_noise(): + message = rejected(**BASES["arc"], auth=[]) + assert "Value error, Value error" not in message + assert "tagged-union" not in message + + +@pytest.mark.parametrize( + "option", ["ex_tenant_name", "ex_force_auth_url", "region", "ex_keyname"] +) +def test_ordinary_driver_options_are_accepted(option: str): + config = BackendConfig.model_validate({**CLOUD, "driver_options": {option: "x"}}) + assert resolve(config).driver_options == {option: "x"} # type: ignore[union-attr] + + +@pytest.mark.parametrize( + "option", ["SECRET", "Password", "access_key", "private_key", "passphrase"] +) +def test_secret_driver_options_are_refused_in_any_case(option: str): + assert "credential provider" in rejected(**CLOUD, driver_options={option: "x"}) + + +def test_container_launcher_placeholder_may_be_inside_an_argument(): + config = BackendConfig.model_validate( + {**BASES["batch-local"], "container_launcher": ["run", "--image={image}"]} + ) + assert resolve(config).container_launcher == ["run", "--image={image}"] # type: ignore[union-attr] + + +def test_known_hosts_of_only_whitespace_is_refused(): + assert "known_hosts" in rejected(**{**SSH, "known_hosts": "\n\t "}) 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..9c2f958 --- /dev/null +++ b/tests/unit/test_fake_backend.py @@ -0,0 +1,524 @@ +"""Behaviour of the fake that needs its test hooks: completion, partial failure, refusal.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from pathlib import Path +from typing import Any + +import pytest + +from intercede import ( + ContainerSpec, + CopyVars, + FileRef, + InterCEdeError, + JobHandle, + JobStatus, + OutputMember, + OutputSpec, + ResourceUnavailableError, + SpecificationRejectedError, + SubmissionSpec, +) +from intercede._backends.config import FakeSettings +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") + ) + ) + + +# -------------------------------------------------------------------------- +# Every operation against every state a job can be in +# -------------------------------------------------------------------------- + +Operation = Callable[[FakeBackend, list[JobHandle], Path], Awaitable[Any]] + +OPERATIONS: dict[str, Operation] = { + "submit": lambda b, h, d: b.submit(SPEC), + "get_status": lambda b, h, d: b.get_status(h), + "get_output": lambda b, h, d: b.get_output(h, d), + "kill": lambda b, h, d: b.kill(h), + "purge": lambda b, h, d: b.purge(h), + "get_diagnostics": lambda b, h, d: b.get_diagnostics(h), +} +PER_JOB = ["get_output", "kill", "purge", "get_diagnostics"] + + +def foreign(handle: JobHandle, **change: str) -> JobHandle: + return handle.model_copy(update=change) + + +@pytest.mark.parametrize("operation", sorted(OPERATIONS)) +async def test_every_operation_refuses_when_closed(operation: str, tmp_path: Path): + backend = FakeBackend() + async with backend: + handles = list((await backend.submit(SPEC)).handles.values()) + with pytest.raises(InterCEdeError, match="closed"): + await OPERATIONS[operation](backend, handles, tmp_path) + + +@pytest.mark.parametrize("operation", sorted(OPERATIONS)) +async def test_every_operation_raises_when_unavailable(operation: str, tmp_path: Path): + async with FakeBackend() as backend: + handles = list((await backend.submit(SPEC)).handles.values()) + backend.unavailable = True + with pytest.raises(ResourceUnavailableError): + await OPERATIONS[operation](backend, handles, tmp_path) + + +@pytest.mark.parametrize("operation", PER_JOB) +@pytest.mark.parametrize( + ("make_handle", "reason"), + [ + (lambda h: foreign(h, id="999"), "unknown job"), + (lambda h: foreign(h, resource="fake://other"), "unknown job"), + (lambda h: foreign(h, backend="arc"), "unknown job"), + ], + ids=["unknown-id", "other-resource", "other-backend"], +) +async def test_every_per_job_operation_fails_for_jobs_it_does_not_hold( + operation: str, + make_handle: Callable[[JobHandle], JobHandle], + reason: str, + tmp_path: Path, +): + async with FakeBackend() as backend: + handle = make_handle((await backend.submit(SPEC)).handles[0]) + outcome = (await OPERATIONS[operation](backend, [handle], tmp_path))[handle] + assert not outcome.ok + assert outcome.reason == reason + + +@pytest.mark.parametrize("operation", PER_JOB) +async def test_every_per_job_operation_reports_injected_failures( + operation: str, tmp_path: Path +): + async with FakeBackend(fail_jobs={"0"}) as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle) + outcome = (await OPERATIONS[operation](backend, [handle], tmp_path))[handle] + assert not outcome.ok + assert outcome.reason == "injected failure" + + +@pytest.mark.parametrize("operation", PER_JOB) +async def test_every_per_job_operation_answers_for_an_empty_batch( + operation: str, tmp_path: Path +): + async with FakeBackend() as backend: + assert await OPERATIONS[operation](backend, [], tmp_path) == {} + + +async def test_status_of_an_empty_batch(): + async with FakeBackend() as backend: + assert await backend.get_status([]) == {} + + +async def test_injected_failure_does_not_hide_status(): + async with FakeBackend(fail_jobs={"0"}) as backend: + handle = (await backend.submit(SPEC)).handles[0] + assert (await backend.get_status([handle]))[handle] == JobStatus.SUBMITTED + + +class TestLifecycle: + async def test_reopening_after_close(self): + backend = FakeBackend() + async with backend: + pass + async with backend: + assert (await backend.submit(SPEC)).handles + + async def test_from_settings(self): + backend = FakeBackend.from_settings(FakeSettings(resource="fake://elsewhere")) + async with backend: + handle = (await backend.submit(SPEC)).handles[0] + assert handle.resource == "fake://elsewhere" + assert handle.backend == "fake" + + +class TestSubmit: + async def test_count(self): + async with FakeBackend() as backend: + submission = await backend.submit(SPEC, 3) + assert sorted(submission.handles) == [0, 1, 2] + + async def test_zero_copies(self): + async with FakeBackend() as backend: + submission = await backend.submit(SPEC, 0) + assert submission.handles == {} + assert submission.failures == {} + + async def test_every_copy_refused(self): + async with FakeBackend(refuse_copies={0, 1}) as backend: + submission = await backend.submit(SPEC, 2) + assert submission.handles == {} + assert sorted(submission.failures) == [0, 1] + + async def test_job_ids_are_unique_across_submissions(self): + async with FakeBackend() as backend: + first = await backend.submit(SPEC, 2) + second = await backend.submit(SPEC, 2) + ids = [h.id for s in (first, second) for h in s.handles.values()] + assert len(set(ids)) == 4 + + async def test_local_inputs_are_read_at_submit(self, tmp_path: Path): + source = tmp_path / "input.txt" + source.write_bytes(b"before") + spec = SubmissionSpec( + executable="x", + inputs=[ + FileRef(name="in", source=str(source)), + FileRef(name="remote", source="https://example.org/data"), + ], + ) + async with FakeBackend() as backend: + handle = (await backend.submit(spec)).handles[0] + source.write_bytes(b"after") # the caller changes its file afterwards + stored = backend._jobs[handle.id].inputs + assert stored == {"in": b"before"} # read once, URL left to the resource + + async def test_missing_local_input_fails_the_submission(self, tmp_path: Path): + spec = SubmissionSpec( + executable="x", + inputs=[FileRef(name="in", source=str(tmp_path / "missing"))], + ) + async with FakeBackend() as backend: + with pytest.raises(FileNotFoundError): + await backend.submit(spec) + + async def test_each_copy_keeps_its_own_secret_and_environment(self): + spec = SubmissionSpec( + executable="x", environment={"SITE": "a", "STAMP": "spec"} + ) + copies = [ + CopyVars( + environment={"STAMP": str(i)}, secrets={"pilot.secret": b"s%d" % i} + ) + for i in range(2) + ] + async with FakeBackend() as backend: + submission = await backend.submit(spec, copies) + jobs = [backend._jobs[submission.handles[i].id] for i in range(2)] + assert [job.secrets for job in jobs] == [ + {"pilot.secret": b"s0"}, + {"pilot.secret": b"s1"}, + ] + assert [job.environment for job in jobs] == [ + {"SITE": "a", "STAMP": "0"}, # the copy's value wins over the spec's + {"SITE": "a", "STAMP": "1"}, + ] + + +class TestRefusal: + @pytest.mark.parametrize( + "spec", + [ + SubmissionSpec(executable="x"), # no outputs at all: nothing to upload + SubmissionSpec(executable="x", outputs=OutputSpec()), + SubmissionSpec( + executable="x", + outputs=OutputSpec(members=[OutputMember(pattern="kept")]), + ), + SubmissionSpec( + executable="x", inputs=[FileRef(name="in", source="/dev/null")] + ), + ], + ids=["no-outputs", "empty-outputs", "kept-output", "local-input"], + ) + async def test_spec_without_resource_staging_is_accepted( + self, spec: SubmissionSpec + ): + async with FakeBackend(resource_staging=False) as backend: + assert (await backend.submit(spec)).handles + + async def test_one_url_output_among_kept_ones_is_refused(self): + spec = SubmissionSpec( + executable="x", + outputs=OutputSpec( + members=[ + OutputMember(pattern="kept"), + OutputMember(pattern="sent", destination="https://e.org/o"), + ] + ), + ) + async with FakeBackend(resource_staging=False) as backend: + with pytest.raises(SpecificationRejectedError, match="upload"): + await backend.submit(spec) + + async def test_refusal_submits_nothing(self): + spec = SubmissionSpec( + executable="x", container=ContainerSpec(image="docker://a") + ) + async with FakeBackend(containers=False) as backend: + with pytest.raises(SpecificationRejectedError): + await backend.submit(spec, 3) + assert backend._jobs == {} + + async def test_container_is_accepted_when_supported(self): + spec = SubmissionSpec( + executable="x", container=ContainerSpec(image="docker://a") + ) + async with FakeBackend() as backend: + assert (await backend.submit(spec)).handles + + async def test_resource_staging_is_accepted_when_supported(self): + spec = SubmissionSpec( + executable="x", + inputs=[FileRef(name="in", source="https://example.org/in")], + outputs=OutputSpec( + members=[OutputMember(pattern="out", destination="https://e.org/o")] + ), + ) + async with FakeBackend() as backend: + assert (await backend.submit(spec)).handles + + +class TestKill: + @pytest.mark.parametrize( + "status", [JobStatus.SUBMITTED, JobStatus.WAITING, JobStatus.RUNNING] + ) + async def test_aborts_every_active_state(self, status: JobStatus): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, status=status) + outcome = (await backend.kill([handle]))[handle] + assert outcome.ok and outcome.reason is None + assert (await backend.get_status([handle]))[handle] == JobStatus.ABORTED + + @pytest.mark.parametrize( + "status", [JobStatus.DONE, JobStatus.FAILED, JobStatus.ABORTED] + ) + async def test_leaves_a_finished_job_alone(self, status: JobStatus): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, status=status) + outcome = (await backend.kill([handle]))[handle] + assert outcome.ok + assert outcome.reason == "already finished" + assert (await backend.get_status([handle]))[handle] == status + + async def test_killing_twice(self): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + await backend.kill([handle]) + assert (await backend.kill([handle]))[handle].reason == "already finished" + + +class TestPurge: + async def test_purged_job_is_gone_for_every_operation(self, tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle) + assert (await backend.purge([handle]))[handle].ok + assert (await backend.get_status([handle]))[handle] == JobStatus.UNKNOWN + for operation in PER_JOB: + outcome = (await OPERATIONS[operation](backend, [handle], tmp_path))[ + handle + ] + assert not outcome.ok + + async def test_purge_keeps_other_jobs(self): + async with FakeBackend() as backend: + submission = await backend.submit(SPEC, 2) + await backend.purge([submission.handles[0]]) + status = await backend.get_status([submission.handles[1]]) + assert status[submission.handles[1]] == JobStatus.SUBMITTED + + +class TestOutput: + @pytest.mark.parametrize( + "status", [JobStatus.SUBMITTED, JobStatus.WAITING, JobStatus.RUNNING] + ) + async def test_refused_while_active(self, status: JobStatus, tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, status=status) + outcome = (await backend.get_output([handle], tmp_path))[handle] + assert not outcome.ok + assert outcome.reason == "job has not finished" + + @pytest.mark.parametrize( + "status", [JobStatus.DONE, JobStatus.FAILED, JobStatus.ABORTED] + ) + async def test_collected_once_finished(self, status: JobStatus, tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, status=status, stdout="log") + 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() == "log" + + async def test_manifest_lists_every_member(self, tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, files={"a.txt": b"a", "deep/er/b.bin": b"\x00\x01"}) + outcome = (await backend.get_output([handle], tmp_path))[handle] + assert outcome.value is not None + assert set(outcome.value.files) == {"a.txt", "deep/er/b.bin"} + assert outcome.value.files["deep/er/b.bin"].read_bytes() == b"\x00\x01" + assert ( + outcome.value.stderr is not None and outcome.value.stderr.read_text() == "" + ) + assert outcome.value.log is None + + async def test_jobs_land_in_separate_directories(self, tmp_path: Path): + async with FakeBackend() as backend: + submission = await backend.submit(SPEC, 2) + handles = list(submission.handles.values()) + for handle in handles: + backend.finish(handle, stdout=handle.id) + outcomes = await backend.get_output(handles, tmp_path) + paths = [outcomes[h].value.stdout for h in handles] # type: ignore[union-attr] + assert paths[0] != paths[1] + assert [p.read_text() for p in paths] == [h.id for h in handles] # type: ignore[union-attr] + + async def test_mixed_batch_reports_each_job(self, tmp_path: Path): + async with FakeBackend() as backend: + submission = await backend.submit(SPEC, 2) + done, running = submission.handles[0], submission.handles[1] + backend.finish(done) + outcomes = await backend.get_output([done, running], tmp_path) + assert outcomes[done].ok + assert not outcomes[running].ok + + async def test_escape_writes_nothing_at_all(self, tmp_path: Path): + dest = tmp_path / "dest" + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish( + handle, stdout="x", files={"ok.txt": b"x", "../../bad": b"x"} + ) + outcome = (await backend.get_output([handle], dest))[handle] + assert not outcome.ok + assert "../../bad" in (outcome.reason or "") + assert not dest.exists() # not even the safe members or stdout + + +class TestDiagnostics: + async def test_defaults_to_stdout(self): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + outcome = (await backend.get_diagnostics([handle]))[handle] + assert outcome.value is not None and outcome.value.name == "stdout" + + async def test_uses_the_requested_name(self): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + outcome = (await backend.get_diagnostics([handle], name="job.log"))[handle] + assert outcome.value is not None and outcome.value.name == "job.log" + + async def test_does_not_change_the_job(self): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + await backend.get_diagnostics([handle]) + assert (await backend.get_status([handle]))[handle] == JobStatus.SUBMITTED diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py new file mode 100644 index 0000000..5f1981c --- /dev/null +++ b/tests/unit/test_models.py @@ -0,0 +1,232 @@ +"""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()) + + +class TestJobHandleKey: + def test_key_joins_backend_resource_and_id(self): + assert build_default_handle("7").key == "fake:r:7" + + def test_str_is_the_key(self): + handle = build_default_handle("7") + assert str(handle) == handle.key + + def test_separators_inside_parts_are_escaped(self): + # Without escaping, ("a:b", "c") and ("a", "b:c") would share a key. + first = build_default_handle("c", resource="a:b") + second = build_default_handle("b:c", resource="a") + assert first.key != second.key + assert first.key.count(":") == 2 + + def test_equal_handles_hash_equal(self): + assert hash(build_default_handle("1")) == hash(build_default_handle("1")) + + @pytest.mark.parametrize("field", ["backend", "resource", "id"]) + def test_every_identity_field_changes_the_key(self, field: str): + assert ( + build_default_handle().key != build_default_handle(**{field: "other"}).key + ) + + def test_defaults(self): + handle = build_default_handle() + assert handle.version == 1 + assert handle.routing == {} + + +class TestOpOutcomeUnparametrised: + def test_success_without_a_type_needs_no_value(self): + assert OpOutcome(ok=True).ok + + def test_failure_without_a_type_still_needs_a_reason(self): + with pytest.raises(ValidationError): + OpOutcome(ok=False) + + +class TestDefaults: + def test_submission_spec(self): + spec = SubmissionSpec(executable="x") + assert spec.outputs is None + assert spec.container is None + assert (spec.stdout, spec.stderr, spec.tag) == (None, None, None) + assert tuple(spec.arguments) == () + assert spec.resources == Resources() + + def test_resources(self): + resources = Resources() + assert resources.cpus == 1 + assert (resources.memory_mb, resources.wall_time_s, resources.queue) == ( + None, + None, + None, + ) + + def test_job_output(self): + output = JobOutput() + assert (output.stdout, output.stderr, output.log) == (None, None, None) + assert output.files == {} + + def test_empty_submission_is_valid(self): + submission = Submission(handles={}) + assert submission.failures == {} + + +@pytest.mark.parametrize("value", [1, 2**31]) +def test_positive_resources_are_accepted(value: int): + assert Resources(cpus=value, memory_mb=value, wall_time_s=value).cpus == value + + +def test_job_status_values_are_stable(): + """Stored statuses are compared by value: renaming one is a breaking change.""" + from intercede import JobStatus + + assert {status.value for status in JobStatus} == { + "Submitted", + "Waiting", + "Running", + "Done", + "Failed", + "Aborted", + "Unknown", + } + + +@pytest.mark.parametrize("field", ["stdout", "stderr"]) +@pytest.mark.parametrize("name", ["job.out", "logs/job.out"]) +def test_stdout_and_stderr_names_are_accepted(field: str, name: str): + assert getattr(SubmissionSpec(executable="x", **{field: name}), field) == name 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__ diff --git a/tests/unit/test_registry.py b/tests/unit/test_registry.py new file mode 100644 index 0000000..6e18d0f --- /dev/null +++ b/tests/unit/test_registry.py @@ -0,0 +1,481 @@ +"""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, + SubmissionSpec, + registry, +) +from intercede._backends.config import ( + ARCSettings, + CloudSettings, + FakeSettings, + HTCondorCESettings, +) +from intercede._backends.fake import FakeBackend +from intercede.registry import _build, _load + +DUMMIES = "tests.unit._registry_dummies" + +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 TestResolution: + async def test_builds_the_fake_unopened(self): + built = _build(FakeSettings(resource="fake://test")) + assert isinstance(built, FakeBackend) + async with built as backend: + assert backend.resource == "fake://test" + + def test_composes_a_transport_and_a_scheduler(self, composed: None): + built: Any = registry.backend(SSH_SLURM) + assert type(built).__name__ == "DummyBatchBackend" + assert type(built.transport).__name__ == "DummyTransport" + assert built.transport.settings.host == "batch.example.org" + assert type(built.scheduler).__name__ == "DummyScheduler" + assert built.settings.kind == "batch" + + def test_one_backend_class_serves_every_combination(self, composed: None): + over_ssh = registry.backend(SSH_SLURM) + on_this_host = registry.backend(LOCAL_SLURM) + assert type(over_ssh) is type(on_this_host) + + def test_accepts_a_config_object(self, composed: None): + assert isinstance( + registry.backend(BackendConfig.model_validate(LOCAL_SLURM)), JobBackend + ) + + +class TestRefused: + def test_invalid_configuration(self): + with pytest.raises(ConfigurationError, match="kind"): + registry.backend({"kind": "lsf"}) + + @pytest.mark.parametrize( + "config", + [ + {"kind": "arc", "endpoint": "https://arc.example.org:443"}, + { + "kind": "cloud", + "endpoint": "https://cloud.example.org:5000", + "driver": "openstack", + "vm_image": "name:alma9", + "vm_flavor": "name:m1.large", + }, + ], + ) + 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) + + @pytest.mark.parametrize( + ("target", "message"), + [ + ("tests.unit.does_not_exist:Transport", "cannot be imported"), + (f"{DUMMIES}:Missing", "cannot be imported"), + (f"{DUMMIES}:NoFromSettings", "no from_settings"), + (f"{DUMMIES}:BuildsTheWrongThing", "which is not a Transport"), + ], + ) + def test_broken_entry_fails_at_resolution( + self, composed: None, monkeypatch: pytest.MonkeyPatch, target: str, message: str + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", target) + with pytest.raises(RegistryError, match=message): + registry.backend(SSH_SLURM) + + def test_backend_that_breaks_the_contract(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setitem( + registry._BACKENDS, "fake", f"{DUMMIES}:BuildsTheWrongThing" + ) + with pytest.raises(RegistryError, match="which is not a JobBackend"): + _build(FakeSettings()) + + +class TestAvailable: + def test_lists_only_what_a_configuration_can_ask_for(self): + listing = registry.available() + assert set(listing) == {"backend", "transport", "scheduler"} + assert "fake" not in listing["backend"] # internal, not a configurable kind + + def test_grows_with_the_tables(self, composed: None): + listing = registry.available() + assert listing["backend"] == ("batch",) + assert listing["transport"] == ("local", "ssh") + assert listing["scheduler"] == ("slurm",) + + +def test_importing_intercede_imports_no_backend(): + 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( + 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 + + +# -------------------------------------------------------------------------- +# Every logical path through the registry +# -------------------------------------------------------------------------- + + +class TestInput: + @pytest.mark.parametrize("config", [None, "arc", 42, ["kind", "arc"]]) + def test_non_mapping_is_a_configuration_error(self, config: Any): + with pytest.raises(ConfigurationError): + registry.backend(config) + + def test_configuration_error_keeps_the_validation_details(self): + with pytest.raises(ConfigurationError) as exc_info: + registry.backend({"kind": "arc"}) + assert "endpoint: required for kind='arc'" in str(exc_info.value) + assert exc_info.value.__cause__ is not None + + def test_errors_share_the_library_root(self): + from intercede import InterCEdeError + + assert issubclass(ConfigurationError, InterCEdeError) + assert issubclass(RegistryError, InterCEdeError) + + def test_mapping_and_config_object_build_the_same(self, composed: None): + from_mapping: Any = registry.backend(LOCAL_SLURM) + from_object: Any = registry.backend(BackendConfig.model_validate(LOCAL_SLURM)) + assert type(from_mapping) is type(from_object) + assert from_mapping.settings == from_object.settings + + +class TestComposition: + def test_transport_gets_only_its_own_settings(self, composed: None): + built: Any = registry.backend(SSH_SLURM) + assert built.transport.settings.kind == "ssh" + assert not hasattr(built.transport.settings, "scheduler") + + def test_scheduler_gets_the_batch_settings(self, composed: None): + built: Any = registry.backend({**LOCAL_SLURM, "queue": "short"}) + assert built.scheduler.settings.kind == "batch" + assert built.scheduler.settings.queue == "short" + + def test_each_backend_gets_fresh_parts(self, composed: None): + first: Any = registry.backend(LOCAL_SLURM) + second: Any = registry.backend(LOCAL_SLURM) + assert first is not second + assert first.transport is not second.transport + + def test_missing_scheduler_is_reported( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.delitem(registry._SCHEDULERS, "slurm") + with pytest.raises(RegistryError, match="scheduler 'slurm' is not available"): + registry.backend(LOCAL_SLURM) + + def test_missing_batch_backend_is_reported( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.delitem(registry._BACKENDS, "batch") + with pytest.raises(RegistryError, match="backend 'batch' is not available"): + registry.backend(LOCAL_SLURM) + + def test_scheduler_that_breaks_its_protocol( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem( + registry._SCHEDULERS, "slurm", f"{DUMMIES}:BuildsTheWrongThing" + ) + with pytest.raises(RegistryError, match="which is not a Scheduler"): + registry.backend(LOCAL_SLURM) + + def test_batch_backend_that_breaks_its_protocol( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem( + registry._BACKENDS, "batch", f"{DUMMIES}:BuildsTheWrongThing" + ) + with pytest.raises( + RegistryError, + match="backend 'batch' built a object, which is not a JobBackend", + ): + registry.backend(LOCAL_SLURM) + + def test_transport_is_checked_before_the_scheduler_is_loaded( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem( + registry._TRANSPORTS, "local", f"{DUMMIES}:BuildsTheWrongThing" + ) + monkeypatch.delitem(registry._SCHEDULERS, "slurm") + with pytest.raises(RegistryError, match="transport 'local'"): + registry.backend(LOCAL_SLURM) + + +class TestBrokenEntries: + @pytest.mark.parametrize( + "target", + [ + "tests.unit._registry_dummies", # no ":Class" part + "tests.unit._registry_dummies:", # empty class name + ":DummyTransport", # empty module name + "builtins:not_a_thing", + ], + ) + def test_malformed_target( + self, composed: None, monkeypatch: pytest.MonkeyPatch, target: str + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", target) + with pytest.raises(RegistryError, match="cannot be imported"): + registry.backend(SSH_SLURM) + + def test_class_without_from_settings( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", "builtins:object") + with pytest.raises(RegistryError, match="has no from_settings"): + registry.backend(SSH_SLURM) + + def test_import_failure_is_chained( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", "tests.unit.does_not_exist:X") + with pytest.raises(RegistryError) as exc_info: + registry.backend(SSH_SLURM) + assert isinstance(exc_info.value.__cause__, ImportError) + + def test_error_names_the_entry( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", "tests.unit.does_not_exist:X") + with pytest.raises(RegistryError, match="'tests.unit.does_not_exist:X'"): + registry.backend(SSH_SLURM) + + +class TestLoad: + def test_imports_each_entry_once( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + calls: list[str] = [] + real = registry.import_module + + def counting(name: str) -> Any: + calls.append(name) + return real(name) + + monkeypatch.setattr(registry, "import_module", counting) + for _ in range(3): + registry.backend(LOCAL_SLURM) + assert ( + calls.count(DUMMIES) == 3 + ) # transport, scheduler, batch: each loaded once + + def test_returns_the_registered_class(self, composed: None): + from tests.unit._registry_dummies import DummyTransport + + assert _load("transport", "ssh") is DummyTransport + + def test_unknown_group_entry_lists_none_yet(self): + with pytest.raises(RegistryError, match=r"\(available: none yet\)"): + _load("transport", "ssh") + + def test_unavailable_entry_lists_what_is(self, composed: None): + with pytest.raises(RegistryError, match=r"\(available: local, ssh\)"): + _load("transport", "carrier-pigeon") + + def test_failures_are_not_cached( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", "tests.unit.does_not_exist:X") + with pytest.raises(RegistryError): + _load("transport", "ssh") + monkeypatch.setitem(registry._TRANSPORTS, "ssh", f"{DUMMIES}:DummyTransport") + assert _load("transport", "ssh").__name__ == "DummyTransport" + + +class TestAvailableDetails: + def test_names_are_sorted(self, monkeypatch: pytest.MonkeyPatch): + for name in ("ssh", "local"): + monkeypatch.setitem(registry._TRANSPORTS, name, f"{DUMMIES}:DummyTransport") + assert registry.available()["transport"] == ("local", "ssh") + + def test_entries_no_configuration_can_ask_for_are_hidden( + self, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._SCHEDULERS, "lsf", f"{DUMMIES}:DummyScheduler") + assert "lsf" not in registry.available()["scheduler"] + + def test_imports_nothing(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", "tests.unit.does_not_exist:X") + assert registry.available()["transport"] == ("ssh",) # listed, never imported + + def test_shipped_tables_today(self): + assert registry.available() == {"backend": (), "transport": (), "scheduler": ()} + + +class TestChoices: + def test_literal(self): + from typing import Literal + + assert registry._choices(Literal["a", "b"]) == {"a", "b"} + + def test_optional_literal(self): + from typing import Literal, Optional + + assert registry._choices(Optional[Literal["a"]]) == {"a"} # noqa: UP045 + + def test_union_of_literals(self): + from typing import Literal + + assert registry._choices(Literal["a"] | Literal["b"] | None) == {"a", "b"} + + def test_non_literal(self): + assert registry._choices(str) == set() + assert registry._choices(int | None) == set() + + +async def test_backend_from_the_registry_is_usable(composed: None): + async with registry.backend(LOCAL_SLURM) as backend: + assert isinstance(backend, JobBackend) + + +async def test_fake_through_the_registry_runs_a_job(): + from intercede import SubmissionSpec + + async with _build(FakeSettings()) as backend: + submission = await backend.submit(SubmissionSpec(executable="/bin/true"), 2) + statuses = await backend.get_status(list(submission.handles.values())) + assert len(statuses) == 2 + + +# -------------------------------------------------------------------------- +# Every family resolves to its own class, and calls reach that class +# -------------------------------------------------------------------------- + +SINGLE_CLASS = { + "arc": ( + {"kind": "arc", "endpoint": "https://arc.example.org:443"}, + ARCSettings, + ), + "htcondor-ce": ( + {"kind": "htcondor-ce", "endpoint": "ce.example.org:9619"}, + HTCondorCESettings, + ), + "cloud": ( + { + "kind": "cloud", + "endpoint": "https://cloud.example.org:5000", + "driver": "openstack", + "vm_image": "name:alma9", + "vm_flavor": "name:m1.large", + }, + CloudSettings, + ), +} + + +@pytest.mark.parametrize("kind", sorted(SINGLE_CLASS)) +class TestSingleClassFamilies: + """Monolithic ("arc", "htcondor-ce") and provider ("cloud") backends.""" + + @pytest.fixture(autouse=True) + def registered(self, monkeypatch: pytest.MonkeyPatch, kind: str) -> None: + monkeypatch.setitem(registry._BACKENDS, kind, f"{DUMMIES}:DummyBackend") + + def test_built_from_its_own_settings(self, kind: str): + config, settings_type = SINGLE_CLASS[kind] + built: Any = registry.backend(config) + assert type(built).__name__ == "DummyBackend" + assert isinstance(built.settings, settings_type) + assert built.settings.kind == kind + + def test_needs_no_transport_or_scheduler(self, kind: str): + # Only the backend table is filled; the other two stay empty. + assert registry._TRANSPORTS == {} + assert registry._SCHEDULERS == {} + assert registry.backend(SINGLE_CLASS[kind][0]) is not None + + def test_is_listed_once_registered(self, kind: str): + assert kind in registry.available()["backend"] + + async def test_calls_reach_the_registered_class(self, kind: str): + built: Any = registry.backend(SINGLE_CLASS[kind][0]) + backend: JobBackend = built # what the consumer holds: only the protocol + async with backend as ce: + result = await ce.submit(SubmissionSpec(executable="x")) + await ce.get_status([]) + assert result == f"submitted by {kind}" + assert built.calls == ["submit", "get_status"] + + +async def test_composed_backend_calls_reach_the_registered_class(composed: None): + built: Any = registry.backend(LOCAL_SLURM) + backend: JobBackend = built + async with backend as ce: + await ce.get_status([]) + assert type(built).__name__ == "DummyBatchBackend" + + +async def test_each_kind_reaches_its_own_class(monkeypatch: pytest.MonkeyPatch): + """Two kinds registered to two classes: each configuration gets its own.""" + monkeypatch.setitem(registry._BACKENDS, "cloud", f"{DUMMIES}:DummyBackend") + cloud: Any = registry.backend(SINGLE_CLASS["cloud"][0]) + fake = _build(FakeSettings()) + assert type(cloud).__name__ == "DummyBackend" + assert isinstance(fake, FakeBackend) + async with cloud as ce: + await ce.submit(SubmissionSpec(executable="x")) + async with fake as ce: + await ce.submit(SubmissionSpec(executable="x")) + assert cloud.calls == ["submit"] # the fake's submit never touched the dummy