diff --git a/docs/user/reference/api.md b/docs/user/reference/api.md index a0dfb90..824c73e 100644 --- a/docs/user/reference/api.md +++ b/docs/user/reference/api.md @@ -11,6 +11,14 @@ Anything else, including every name starting with `_`, is internal. ::: intercede.models +## Configuration + +::: intercede.config + +## Registry + +::: intercede.registry + ## Exceptions ::: intercede.exceptions diff --git a/mkdocs.yml b/mkdocs.yml index e591f8c..33d5280 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -17,6 +17,12 @@ plugins: show_source: false members_order: source merge_init_into_class: true + inherited_members: true + # Field tables for the configuration models, read from the models + # themselves, so the documentation cannot drift from the code. + extensions: + - griffe_pydantic: + schema: false validation: links: diff --git a/pyproject.toml b/pyproject.toml index aa21d24..a86f715 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ Repository = "https://github.com/DIRACGrid/intercede" testing = [ "pytest", "pytest-asyncio", + "pytest-cov", "pytest-xdist", "pyyaml", "ty", @@ -59,6 +60,19 @@ markers = [ "destructive_fetch: asserts one-shot/destructive output-retrieval semantics", ] +[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"] @@ -100,6 +114,9 @@ ignore = [ "FLY002", ] +[tool.ruff.lint.pydocstyle] +convention = "google" + [tool.ruff.lint.isort] required-imports = ["from __future__ import annotations"] @@ -144,6 +161,7 @@ intercede = { path = ".", editable = true } mkdocs = "<2" mkdocs-material = "*" mkdocstrings = { version = "*", extras = ["python"] } +griffe-pydantic = "*" [tool.pixi.feature.docs.tasks] docs = { cmd = "mkdocs build --strict", description = "Build the documentation" } diff --git a/src/intercede/__init__.py b/src/intercede/__init__.py index 08dd317..50f1601 100644 --- a/src/intercede/__init__.py +++ b/src/intercede/__init__.py @@ -5,10 +5,21 @@ from importlib.metadata import PackageNotFoundError from importlib.metadata import version as get_version +from intercede import registry +from intercede.config import ( + ARCConfig, + BackendConfig, + BatchLocalConfig, + BatchSSHConfig, + CloudConfig, + HTCondorCEConfig, + validate_config, +) from intercede.exceptions import ( AuthenticationError, ConfigurationError, InterCEdeError, + RegistryError, ResourceUnavailableError, SpecificationRejectedError, ) @@ -47,6 +58,7 @@ __all__ = [ "__version__", + "registry", # protocols "Diagnosable", "JobBackend", @@ -71,10 +83,19 @@ "StatusMap", "Submission", "SubmissionSpec", + # configuration + "ARCConfig", + "BackendConfig", + "BatchLocalConfig", + "BatchSSHConfig", + "CloudConfig", + "HTCondorCEConfig", + "validate_config", # exceptions "AuthenticationError", "ConfigurationError", "InterCEdeError", + "RegistryError", "ResourceUnavailableError", "SpecificationRejectedError", ] diff --git a/src/intercede/_backends/config.py b/src/intercede/_backends/config.py new file mode 100644 index 0000000..f50a869 --- /dev/null +++ b/src/intercede/_backends/config.py @@ -0,0 +1,18 @@ +"""Settings of the in-memory fake, which no public configuration reaches. + +Every other backend is built from its public configuration model in +``intercede.config``. +""" + +from __future__ import annotations + +from typing import Literal + +from intercede.models import _Model + + +class FakeSettings(_Model): + """The in-memory fake, for the test suites.""" + + kind: Literal["fake"] = "fake" + resource: str = "fake://local" diff --git a/src/intercede/_backends/fake.py b/src/intercede/_backends/fake.py index be45be6..80b52e7 100644 --- a/src/intercede/_backends/fake.py +++ b/src/intercede/_backends/fake.py @@ -6,12 +6,13 @@ from __future__ import annotations -from collections.abc import Collection, Mapping, Sequence +from collections.abc import Callable, Collection, Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from types import TracebackType from typing import Self +from intercede._backends.config import FakeSettings from intercede.exceptions import ( InterCEdeError, ResourceUnavailableError, @@ -92,6 +93,18 @@ def __init__( 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. + + Args: + settings: The fake's settings: only the resource it reports. + + Returns: + A new fake, not yet opened. + """ + return cls(resource=settings.resource) + # -- lifecycle --------------------------------------------------------- async def __aenter__(self) -> Self: @@ -252,10 +265,8 @@ def _check_supported(self, spec: SubmissionSpec) -> None: if not self.resource_staging: self._check_can_fetch(spec.inputs) - if spec.outputs is not None: + if spec.outputs is not None: # no outputs means nothing to upload self._check_can_upload(spec.outputs.members) - else: - raise SpecificationRejectedError(f"tag: {spec.tag} contains no outputs") def _check_container_support(self, spec: SubmissionSpec) -> None: if spec.container is not None and not self.containers: @@ -368,3 +379,4 @@ def _write_job_files( _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..e9be28a --- /dev/null +++ b/src/intercede/config.py @@ -0,0 +1,351 @@ +"""Backend configuration: how a consumer says which backend it wants. + +There is one model per kind of backend, and ``BackendConfig`` is any of them, +chosen by ``kind`` (and, for ``batch``, by ``transport``). Each model has +exactly the fields of its backend, so a field that does not apply is refused +when the configuration is read, not ignored or discovered at the first submit. + +A consumer can build a model directly, or pass the same data as a mapping, such +as a parsed TOML table, to ``validate_config`` or ``registry.get_backend``:: + + kind = "batch" + transport = "ssh" + scheduler = "slurm" + host = "batch.example.org" + user = "dirac" + known_hosts = "batch.example.org ssh-ed25519 AAAA..." + +Configuration describes the resource and is never secret: keys, passwords, +tokens and proxies come from the credential provider at run time. Error +messages therefore never repeat the values they were given. The kinds are a +closed set: they list what interCEde ships. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Annotated, Any, Literal, Self + +from pydantic import ( + BaseModel, + ConfigDict, + Discriminator, + PositiveInt, + Tag, + TypeAdapter, + ValidationError, + model_validator, +) + +from intercede.exceptions import ConfigurationError + +__all__ = [ + "ARCConfig", + "BackendConfig", + "BatchLocalConfig", + "BatchSSHConfig", + "CloudConfig", + "HTCondorCEConfig", + "validate_config", +] + + +class _Config(BaseModel): + """Base of every configuration model: immutable, strict, and silent about values.""" + + model_config = ConfigDict( + frozen=True, + extra="forbid", + # A value given by mistake may be a secret: never repeat it in an error. + hide_input_in_errors=True, + use_attribute_docstrings=True, + ) + + +class _QueueConfig(_Config): + """What every backend that queues jobs on a resource shares.""" + + queue: str | None = None + """The queue a job goes to when its specification names none.""" + + container_launcher: Sequence[str] | None = None + """An argv template that runs the executable in a container image, with an + ``{image}`` placeholder, e.g. ``["apptainer", "exec", "--bind", "/cvmfs", "{image}"]``.""" + + max_output_bytes: PositiveInt = 10 * 2**30 + """Collection of a job's output stops beyond this many bytes.""" + + max_output_files: PositiveInt = 10_000 + """Collection of a job's output stops beyond this many files.""" + + transfer_timeout_s: PositiveInt = 3600 + """How long collecting one job's output may take, in seconds.""" + + @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 + + +# -- grid CEs that schedule by themselves ------------------------------------- + + +class ARCConfig(_QueueConfig): + """An ARC CE.""" + + kind: Literal["arc"] + """Selects this model.""" + + endpoint: str + """The CE's REST endpoint, e.g. ``"https://arc.example.org:443/arex"``.""" + + auth: frozenset[Literal["token", "proxy"]] = frozenset({"token"}) + """Which credentials the provider must supply, all of them together: a token + for the channel, a proxy for delegation, or both.""" + + @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 HTCondorCEConfig(_QueueConfig): + """An HTCondor-CE.""" + + kind: Literal["htcondor-ce"] + """Selects this model.""" + + endpoint: str + """The CE's collector, as ``"host"`` or ``"host:port"``, e.g. ``"ce.example.org:9619"``.""" + + auth: frozenset[Literal["token", "proxy"]] = frozenset({"token"}) + """Which credentials the provider must supply. The token authenticates the + channel; ``"proxy"`` adds an X.509 proxy for sites that account on VOMS attributes.""" + + @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 + + +# -- a cloud provider that boots one VM per job -------------------------------- + +_SECRET_NAMES = ("secret", "password", "token", "passphrase") +_SECRET_EXACT = ("key", "api_key", "access_key", "private_key") + + +class CloudConfig(_Config): + """A cloud provider that boots one VM per job. + + There is no queue and no filesystem to collect from afterwards, so the + queue, output limits and container launcher of the other kinds do not apply. + """ + + kind: Literal["cloud"] + """Selects this model.""" + + 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"``.""" + + driver_options: Mapping[str, str] = {} + """Extra, non-secret arguments for the driver, e.g. ``{"ex_tenant_name": "dirac"}``. + The cloud's API keys come from the credential provider, never from here.""" + + vm_image: str + """An image ID, or ``"name:"``.""" + + vm_flavor: str + """A flavor ID, or ``"name:"``.""" + + vm_networks: Sequence[str] = () + """The networks to attach the VM to.""" + + vm_ssh_key: str | None = None + """The name of a key pair known to the cloud, not the key itself.""" + + @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 + + +# -- a batch system reached through a transport -------------------------------- + + +class _BatchConfig(_QueueConfig): + kind: Literal["batch"] + """Selects a batch system; ``transport`` then selects the model.""" + + scheduler: Literal["slurm", "htcondor", "direct"] + """The batch system on the resource; ``"direct"`` runs jobs without one.""" + + +class BatchLocalConfig(_BatchConfig): + """A batch system on this host.""" + + transport: Literal["local"] + """Run the batch system's commands on this host.""" + + +class BatchSSHConfig(_BatchConfig): + """A batch system on a remote host, reached over SSH.""" + + transport: Literal["ssh"] + """Run the batch system's commands on ``host``, over SSH.""" + + host: str + """The host to log in to.""" + + port: PositiveInt = 22 + """The SSH port on ``host``.""" + + user: str + """The account to log in as.""" + + known_hosts: str + """The host keys to expect, in ``known_hosts`` format. Verification is not optional.""" + + auth: Literal["key", "password", "agent"] = "key" + """How to log in: a key or a password from the credential provider, or an SSH agent.""" + + jump_host: str | None = None + """A host to connect through, as ``"[user@]host[:port]"``.""" + + connect_timeout_s: PositiveInt = 30 + """How long connecting may take, in seconds.""" + + @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 + + +def _choice(value: Any) -> str | None: + """Return which model a configuration is for: its ``kind``, and for ``batch`` its transport.""" + if isinstance(value, Mapping): + kind, transport = value.get("kind"), value.get("transport") + else: + kind, transport = ( + getattr(value, "kind", None), + getattr(value, "transport", None), + ) + if not isinstance(kind, str): + return None + return f"batch+{transport}" if kind == "batch" else kind + + +_MODELS: dict[str, type[_Config]] = { + "arc": ARCConfig, + "htcondor-ce": HTCondorCEConfig, + "cloud": CloudConfig, + "batch+local": BatchLocalConfig, + "batch+ssh": BatchSSHConfig, +} + +BackendConfig = Annotated[ + Annotated[ARCConfig, Tag("arc")] + | Annotated[HTCondorCEConfig, Tag("htcondor-ce")] + | Annotated[CloudConfig, Tag("cloud")] + | Annotated[BatchLocalConfig, Tag("batch+local")] + | Annotated[BatchSSHConfig, Tag("batch+ssh")], + Discriminator(_choice), +] +"""The configuration of any backend: one of the models above, chosen by ``kind``.""" + +_ADAPTER: TypeAdapter[BackendConfig] = TypeAdapter(BackendConfig) + + +def validate_config(config: BackendConfig | Mapping[str, Any]) -> BackendConfig: + """Check a configuration and return the model of the backend it describes. + + A model is checked again, so one changed with ``model_copy``, which does not + validate, cannot reach a backend with fields that do not apply. + + Args: + config: A configuration model, or the same data as a mapping, for + example a parsed TOML table. + + Returns: + The model of the backend ``config`` describes. + + Raises: + ConfigurationError: The configuration is invalid. The message names + every problem by field, and never repeats a value it was given. + """ + # Only what was set: defaults of another kind's fields are not the user's mistake. + data = ( + config.model_dump(exclude_unset=True) + if isinstance(config, BaseModel) + else config + ) + try: + return _ADAPTER.validate_python(data) + except ValidationError as exc: + raise ConfigurationError(_describe(data, exc)) from None + + +def _describe(data: Any, exc: ValidationError) -> str: + """Report every problem by the field the consumer wrote, without its value. + + Args: + data: The configuration that was refused. + exc: What validating it raised. + + Returns: + One line per problem, joined with "; ", such as + "host: required for kind='batch' with transport='ssh'". + """ + choice = _choice(data) + model = _MODELS.get(choice or "") + where = ( + "kind='batch' with transport={!r}".format(choice.removeprefix("batch+")) + if choice and choice.startswith("batch+") + else f"kind={choice!r}" + ) + problems = [] + unknown = [] + for error in exc.errors(): + fields = [part for part in error["loc"] if isinstance(part, str)] + field = fields[-1] if fields and fields[-1] not in _MODELS else None + if error["type"] in ("union_tag_not_found", "union_tag_invalid"): + problems.append(_unknown_kind(data)) + elif error["type"] == "extra_forbidden" and field is not None: + unknown.append(field) + elif error["type"] == "missing": + problems.append(f"{field}: required for {where}") + else: + message = error["msg"].removeprefix("Value error, ") + problems.append(f"{field}: {message}" if field else message) + if unknown and model is not None: + expected = ", ".join(name for name in model.model_fields if name != "kind") + problems.insert( + 0, f"{', '.join(unknown)}: not for {where} (expected: {expected})" + ) + return "; ".join(problems) + + +def _unknown_kind(data: Any) -> str: + kinds = sorted({choice.split("+")[0] for choice in _MODELS}) + if isinstance(data, Mapping) and data.get("kind") == "batch": + return "transport: must be one of 'local', 'ssh'" + return f"kind: must be one of {', '.join(map(repr, kinds))}" diff --git a/src/intercede/exceptions.py b/src/intercede/exceptions.py index 839e051..6b99d35 100644 --- a/src/intercede/exceptions.py +++ b/src/intercede/exceptions.py @@ -10,6 +10,7 @@ "AuthenticationError", "ConfigurationError", "InterCEdeError", + "RegistryError", "ResourceUnavailableError", "SpecificationRejectedError", ] @@ -33,3 +34,11 @@ class ResourceUnavailableError(InterCEdeError): 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/registry.py b/src/intercede/registry.py new file mode 100644 index 0000000..b7329a8 --- /dev/null +++ b/src/intercede/registry.py @@ -0,0 +1,225 @@ +"""Turn a backend configuration into a backend. + +The registry is closed: the tables below list every backend, transport and +scheduler interCEde ships, and nothing outside the library can add to them. +Each entry names its class as ``"module:Class"`` and is imported only when a +configuration asks for it, so installing interCEde never imports every backend. + +Every registered class provides a ``from_settings`` classmethod, which is how +the registry builds it from a validated configuration model. +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 intercede._backends.batch.protocols import Scheduler, Transport +from intercede._backends.config import FakeSettings +from intercede.config import ( + _MODELS as _CONFIG_MODELS, +) +from intercede.config import ( + BackendConfig, + BatchLocalConfig, + BatchSSHConfig, + validate_config, +) +from intercede.exceptions import RegistryError +from intercede.protocols import JobBackend + +__all__ = ["available", "get_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 get_backend(config: BackendConfig | Mapping[str, Any]) -> JobBackend: + """Build the backend a configuration describes. + + The backend is returned unopened: use it as ``async with registry.get_backend(config) as ce``. + Construction does no I/O, so building a backend for an unreachable resource succeeds. + + Args: + config: A ``BackendConfig``, or the same data as a mapping, for example + a parsed TOML table. + + Returns: + The backend, not yet opened. + + Raises: + ConfigurationError: The configuration is invalid. + RegistryError: The configuration is valid, but this version of + interCEde does not provide what it asks for. + """ + return _build(validate_config(config)) + + +def available() -> dict[str, tuple[str, ...]]: + """List what a configuration can ask for in this version, per group. + + Returns: + The names per group, sorted, for tooling. For example + ``{"backend": ("arc", "batch"), "transport": ("ssh",), ...}``. + """ + batch = (BatchLocalConfig, BatchSSHConfig) + public = { + "backend": set().union( + *( + _choices(model.model_fields["kind"].annotation) + for model in _CONFIG_MODELS.values() + ) + ), + "transport": set().union( + *(_choices(model.model_fields["transport"].annotation) for model in batch) + ), + "scheduler": _choices(BatchLocalConfig.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``. + + Args: + annotation: A field's type annotation. + + Returns: + Every string the annotation allows; empty when it is not a Literal. + """ + 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(config: BackendConfig | FakeSettings) -> JobBackend: + """Build a backend from a validated configuration. Internal: the test suites call it directly. + + Args: + config: The validated configuration of one backend, or the fake's settings. + + Returns: + The backend, not yet opened. A ``batch`` backend is composed of the + transport and the scheduler its configuration names; both receive the + whole configuration. + + Raises: + RegistryError: Something the configuration names is not available, or + what was built does not satisfy its protocol. + """ + if not isinstance(config, BatchLocalConfig | BatchSSHConfig): + built = _load("backend", config.kind).from_settings(config) + result: JobBackend = _checked("backend", config.kind, built, JobBackend) + return result + + transport: Transport = _checked( + "transport", + config.transport, + _load("transport", config.transport).from_settings(config), + Transport, + ) + scheduler: Scheduler = _checked( + "scheduler", + config.scheduler, + _load("scheduler", config.scheduler).from_settings(config), + Scheduler, + ) + built = _load("backend", "batch").from_settings( + config, 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. + + Args: + group: ``"backend"``, ``"transport"`` or ``"scheduler"``. + name: The name the class is registered under, such as ``"ssh"``. + + Returns: + The registered class. It is cached, so each entry is imported once. + + Raises: + RegistryError: Nothing is registered under ``name``, or the entry + cannot be imported or has no ``from_settings``. + """ + 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. + + Args: + group: The registry group the object was built from, for the error. + name: The registered name it was built from, for the error. + built: What the registered class built. + protocol: The runtime-checkable protocol it must satisfy. + + Returns: + ``built`` itself, once checked. + + Raises: + RegistryError: ``built`` does not satisfy ``protocol``. + """ + if not isinstance(built, protocol): + raise RegistryError( + f"{group} {name!r} built a {type(built).__name__}, which is not a {protocol.__name__}" + ) + return built diff --git a/tests/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/_registry_dummies.py b/tests/unit/_registry_dummies.py new file mode 100644 index 0000000..5dc02fe --- /dev/null +++ b/tests/unit/_registry_dummies.py @@ -0,0 +1,90 @@ +"""Stand-ins registered by test_registry.py. + +Imported only through the registry, which is how the tests show that lookup is +lazy. Not collected by pytest: the file name does not start with ``test_``. +""" + +from __future__ import annotations + +from typing import Any + + +class _Lifecycle: + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *exc_info: object) -> None: ... + + +class DummyTransport(_Lifecycle): + def __init__(self, settings: Any) -> None: + self.settings = settings + + @classmethod + def from_settings(cls, settings: Any) -> DummyTransport: + return cls(settings) + + async def run(self, argv: Any) -> Any: ... + + async def put(self, local: Any, remote: Any) -> None: ... + + async def get(self, remote: Any, local: Any) -> None: ... + + +class DummyScheduler: + stages_own_files = False + + def __init__(self, settings: Any) -> None: + self.settings = settings + + @classmethod + def from_settings(cls, settings: Any) -> DummyScheduler: + return cls(settings) + + def submit_command(self, spec: Any, copies: Any, workdir: Any) -> Any: ... + + def parse_submit(self, raw: Any) -> Any: ... + + def status_command(self, ids: Any) -> Any: ... + + def parse_status(self, raw: Any, ids: Any) -> Any: ... + + def kill_command(self, ids: Any) -> Any: ... + + +class DummyBatchBackend(_Lifecycle): + def __init__(self, settings: Any, transport: Any, scheduler: Any) -> None: + self.settings = settings + self.transport = transport + self.scheduler = scheduler + + @classmethod + def from_settings( + cls, settings: Any, *, transport: Any, scheduler: Any + ) -> DummyBatchBackend: + return cls(settings, transport, scheduler) + + async def submit(self, spec: Any, copies: Any = 1) -> Any: ... + + async def get_status(self, handles: Any) -> Any: ... + + +class DummyBackend(_Lifecycle): + """A single-class backend, standing in for the monolithic and provider families.""" + + def __init__(self, settings: Any) -> None: + self.settings = settings + + @classmethod + def from_settings(cls, settings: Any) -> DummyBackend: + return cls(settings) + + async def submit(self, spec: Any, copies: Any = 1) -> Any: ... + + async def get_status(self, handles: Any) -> Any: ... + + +class BuildsTheWrongThing: + @classmethod + def from_settings(cls, *args: Any, **kwargs: Any) -> object: + return object() diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 321e841..a0903f3 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -12,10 +12,12 @@ import pytest from intercede import JobBackend -from intercede._backends.fake import FakeBackend +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": FakeBackend, + "fake": lambda: _build(FakeSettings()), } diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..02b1ffc --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,179 @@ +"""Backend configuration, through the public API: one model per kind of backend.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from intercede import ( + ARCConfig, + BatchLocalConfig, + BatchSSHConfig, + CloudConfig, + ConfigurationError, + HTCondorCEConfig, + validate_config, +) + +ARC: dict[str, Any] = {"kind": "arc", "endpoint": "https://arc.example.org:443"} +CE: dict[str, Any] = {"kind": "htcondor-ce", "endpoint": "ce.example.org:9619"} +CLOUD: dict[str, Any] = { + "kind": "cloud", + "endpoint": "https://cloud.example.org:5000", + "driver": "openstack", + "vm_image": "name:alma9-pilot", + "vm_flavor": "name:m1.large", +} +LOCAL: dict[str, Any] = {"kind": "batch", "transport": "local", "scheduler": "slurm"} +SSH: dict[str, Any] = { + "kind": "batch", + "transport": "ssh", + "scheduler": "slurm", + "host": "batch.example.org", + "user": "dirac", + "known_hosts": "batch.example.org ssh-ed25519 AAAAC3Nz...", +} + + +def rejected(**data: Any) -> str: + """Return the message for a configuration that must be refused.""" + with pytest.raises(ConfigurationError) as exc_info: + validate_config(data) + return str(exc_info.value) + + +@pytest.mark.parametrize( + ("data", "model"), + [ + (ARC, ARCConfig), + (CE, HTCondorCEConfig), + (CLOUD, CloudConfig), + (LOCAL, BatchLocalConfig), + (SSH, BatchSSHConfig), + ], + ids=["arc", "htcondor-ce", "cloud", "batch-local", "batch-ssh"], +) +def test_kind_and_transport_choose_the_model(data: dict[str, Any], model: type): + assert type(validate_config(data)) is model + + +def test_defaults_are_filled_in(): + ssh = validate_config(SSH) + assert isinstance(ssh, BatchSSHConfig) + assert (ssh.port, ssh.auth, ssh.connect_timeout_s) == (22, "key", 30) + assert ssh.transfer_timeout_s == 3600 + ce = validate_config(CE) + assert isinstance(ce, HTCondorCEConfig) and ce.auth == {"token"} + + +def test_a_model_passes_through_unchanged(): + model = HTCondorCEConfig(kind="htcondor-ce", endpoint="ce:9619", queue="short") + assert validate_config(model) == model + + +class TestRefused: + @pytest.mark.parametrize( + ("data", "message"), + [ + ( + {"kind": "lsf"}, + "kind: must be one of 'arc', 'batch', 'cloud', 'htcondor-ce'", + ), + ( + {**LOCAL, "transport": "pigeon"}, + "transport: must be one of 'local', 'ssh'", + ), + ({"kind": "batch", "scheduler": "slurm"}, "transport: must be one of"), + ], + ids=["kind", "transport", "no-transport"], + ) + def test_an_unknown_choice(self, data: dict[str, Any], message: str): + assert message in rejected(**data) + + @pytest.mark.parametrize( + ("data", "field", "where"), + [ + ({**ARC, "host": "h"}, "host", "kind='arc'"), + ({**CLOUD, "queue": "short"}, "queue", "kind='cloud'"), + ( + {**LOCAL, "known_hosts": "h k"}, + "known_hosts", + "kind='batch' with transport='local'", + ), + ( + {**SSH, "endpoint": "https://a"}, + "endpoint", + "kind='batch' with transport='ssh'", + ), + ], + ) + def test_a_field_of_another_kind( + self, data: dict[str, Any], field: str, where: str + ): + assert f"{field}: not for {where} (expected: " in rejected(**data) + + def test_every_missing_field_is_named_at_once(self): + message = rejected(kind="batch", transport="ssh", scheduler="slurm") + for field in ("host", "user", "known_hosts"): + assert f"{field}: required for kind='batch' with transport='ssh'" in message + + @pytest.mark.parametrize( + ("data", "message"), + [ + ({**ARC, "auth": []}, "auth must name at least one"), + ({**CE, "auth": ["proxy"]}, "must include 'token'"), + ({**ARC, "auth": ["key"]}, "auth"), + ({**SSH, "auth": "token"}, "auth"), + ( + {**SSH, "known_hosts": "\n\t "}, + "known_hosts must hold the expected host keys", + ), + ( + {**LOCAL, "container_launcher": ["apptainer", "exec"]}, + "'{image}' placeholder", + ), + ({**CLOUD, "driver_options": {"OS_PASSWORD": "x"}}, "credential provider"), + ({**LOCAL, "max_output_files": 0}, "max_output_files"), + ], + ids=[ + "arc-no-auth", + "htcondor-ce-no-token", + "ce-ssh-auth", + "ssh-ce-auth", + "no-host-keys", + "launcher-no-image", + "secret-driver-option", + "zero-limit", + ], + ) + def test_a_rule_of_the_kind(self, data: dict[str, Any], message: str): + assert message in rejected(**data) + + +class TestNeverRepeatsAValue: + """A value given by mistake may be a secret, so no error message repeats one.""" + + def test_an_unknown_field(self): + message = rejected(**SSH, password="toto4") # noqa: S106 (fake) + assert "toto4" not in message + assert "password: not for kind='batch' with transport='ssh'" in message + + def test_a_wrong_value(self): + assert "toto4" not in rejected(**SSH, port="toto4") + + def test_building_a_model_directly(self): + with pytest.raises(ValueError) as exc_info: + BatchSSHConfig(**SSH, password="toto4") # type: ignore[call-arg] # noqa: S106 + assert "toto4" not in str(exc_info.value) + + +def test_a_copied_model_is_checked_again(): + # model_copy does not validate: a field of the old transport must not slip through. + local = validate_config(SSH).model_copy(update={"transport": "local"}) + with pytest.raises(ConfigurationError) as exc_info: + validate_config(local) + assert ( + "host, user, known_hosts: not for kind='batch' with transport='local'" + in str(exc_info.value) + ) diff --git a/tests/unit/test_fake_backend.py b/tests/unit/test_fake_backend.py index 5f8f452..6a3ce61 100644 --- a/tests/unit/test_fake_backend.py +++ b/tests/unit/test_fake_backend.py @@ -1,8 +1,14 @@ -"""Behaviour of the fake that needs its test hooks: completion, partial failure, refusal.""" +"""The fake's test hooks, and the rules the conformance suite cannot reach without them. + +``test_conformance.py`` already runs the contract against the fake; what is here +needs a job to finish, fail or be refused on demand. +""" from __future__ import annotations +from collections.abc import Awaitable, Callable from pathlib import Path +from typing import Any import pytest @@ -10,6 +16,7 @@ ContainerSpec, CopyVars, FileRef, + JobHandle, JobStatus, OutputMember, OutputSpec, @@ -21,110 +28,165 @@ SPEC = SubmissionSpec(executable="/bin/true") - -async def test_partial_submission_keeps_copy_indexes(): - async with FakeBackend(refuse_copies={0, 2}) as backend: - copies = [CopyVars(environment={"STAMP": str(i)}) for i in range(4)] - sub = await backend.submit(SPEC, copies) - assert sorted(sub.handles) == [1, 3] - assert sorted(sub.failures) == [0, 2] - - -async def test_injected_failure_is_reported_per_job(): - async with FakeBackend(fail_jobs={"1"}) as backend: - sub = await backend.submit(SPEC, 2) - good, bad = sub.handles[0], sub.handles[1] - killed = await backend.kill([good, bad]) - assert killed[good].ok - assert not killed[bad].ok - - -async def test_whole_operation_failure_raises(): - async with FakeBackend() as backend: - backend.unavailable = True - with pytest.raises(ResourceUnavailableError): - await backend.submit(SPEC) - - -async def test_output_is_repeatable_and_lands_under_dest(tmp_path: Path): - async with FakeBackend() as backend: - handle = (await backend.submit(SPEC)).handles[0] - backend.finish(handle, stdout="hello", files={"result/out.txt": b"42"}) - for _ in range(2): +Operation = Callable[[FakeBackend, list[JobHandle], Path], Awaitable[Any]] +PER_JOB: dict[str, Operation] = { + "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), +} + + +class TestHooks: + async def test_refused_copies_keep_their_indexes(self): + async with FakeBackend(refuse_copies={0, 2}) as backend: + submission = await backend.submit(SPEC, 4) + assert sorted(submission.handles) == [1, 3] + assert sorted(submission.failures) == [0, 2] + + @pytest.mark.parametrize("operation", sorted(PER_JOB)) + async def test_an_injected_failure_is_per_job(self, 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 PER_JOB[operation](backend, [handle], tmp_path))[handle] + assert (await backend.get_status([handle]))[handle] == JobStatus.DONE + assert not outcome.ok and outcome.reason == "injected failure" + + async def test_an_unavailable_resource_fails_the_whole_operation(self): + async with FakeBackend() as backend: + backend.unavailable = True + with pytest.raises(ResourceUnavailableError): + await backend.submit(SPEC) + + @pytest.mark.parametrize( + ("options", "spec"), + [ + ( + {"resource_staging": False}, + SubmissionSpec( + executable="x", + inputs=[FileRef(name="in", source="https://example.org/in")], + ), + ), + ( + {"resource_staging": False}, + SubmissionSpec( + executable="x", + outputs=OutputSpec( + members=[ + OutputMember(pattern="kept"), + OutputMember(pattern="sent", destination="https://e.org/o"), + ] + ), + ), + ), + ( + {"containers": False}, + SubmissionSpec(executable="x", container=ContainerSpec(image="a")), + ), + ], + ids=["url-input", "url-output", "container"], + ) + async def test_what_a_resource_cannot_do_is_refused_before_anything_runs( + self, options: dict[str, bool], spec: SubmissionSpec + ): + async with FakeBackend(**options) as backend: + with pytest.raises(SpecificationRejectedError): + await backend.submit(spec, 3) + assert backend._jobs == {} + + +class TestSubmit: + async def test_a_copy_value_wins_over_the_spec(self): + spec = SubmissionSpec( + executable="x", environment={"SITE": "a", "STAMP": "spec"} + ) + copies = [ + CopyVars(environment={"STAMP": str(i)}, secrets={"s": b"%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.environment for job in jobs] == [ + {"SITE": "a", "STAMP": "0"}, + {"SITE": "a", "STAMP": "1"}, + ] + assert [job.secrets for job in jobs] == [{"s": b"0"}, {"s": b"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(ids) == 4 + + +class TestKill: + async def test_aborts_an_active_job(self): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, status=JobStatus.RUNNING) + assert (await backend.kill([handle]))[handle].ok + assert (await backend.get_status([handle]))[handle] == JobStatus.ABORTED + + async def test_leaves_a_finished_job_alone(self): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, status=JobStatus.FAILED) + outcome = (await backend.kill([handle]))[handle] + assert outcome.ok and outcome.reason == "already finished" + assert (await backend.get_status([handle]))[handle] == JobStatus.FAILED + + +class TestOutput: + async def test_collected_again_and_again(self, tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] + backend.finish(handle, stdout="hello", files={"deep/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["deep/out.txt"].read_bytes() == b"42" + + async def test_refused_while_active(self, tmp_path: Path): + async with FakeBackend() as backend: + handle = (await backend.submit(SPEC)).handles[0] 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" + assert not outcome.ok and outcome.reason == "job has not finished" + + @pytest.mark.parametrize("name", ["../../escaped", "/abs/escaped"]) + async def test_an_escaping_member_writes_nothing_at_all( + self, tmp_path: Path, name: str + ): + 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", name: b"x"}) + outcome = (await backend.get_output([handle], dest))[handle] + assert not outcome.ok + assert not dest.exists() # not even the safe members or stdout + + +async def test_a_purged_job_is_gone_for_every_operation(tmp_path: Path): 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("*")) + backend.finish(handle) + assert (await backend.purge([handle]))[handle].ok + assert (await backend.get_status([handle]))[handle] == JobStatus.UNKNOWN + for operation in PER_JOB.values(): + assert not (await operation(backend, [handle], tmp_path))[handle].ok -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 def test_diagnostics_read_a_running_job_by_name(): 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" + default = (await backend.get_diagnostics([handle]))[handle] named = (await backend.get_diagnostics([handle], filename="job.log"))[handle] - assert outcome.value.filename == "stdout" # the default - assert named.value is not None and named.value.filename == "job.log" - - -@pytest.mark.parametrize( - "spec", - [ - SubmissionSpec( - executable="x", inputs=[FileRef(name="in", source="https://example.org/in")] - ), - SubmissionSpec( - executable="x", - outputs=OutputSpec( - members=[OutputMember(pattern="out", destination="https://e.org/o")] - ), - ), - ], -) -async def test_spec_needing_resource_staging_is_rejected_at_submit( - spec: SubmissionSpec, -): - async with FakeBackend(resource_staging=False) as backend: - with pytest.raises(SpecificationRejectedError): - await backend.submit(spec) - - -async def test_spec_with_container_is_rejected_when_unsupported(): - async with FakeBackend(containers=False) as backend: - with pytest.raises(SpecificationRejectedError): - await backend.submit( - SubmissionSpec( - executable="x", container=ContainerSpec(image="docker://alpine") - ) - ) + assert default.value is not None and default.value.filename == "stdout" + assert default.value.content == "progress" + assert named.value is not None and named.value.filename == "job.log" diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index fe11e1b..1f36d34 100644 --- a/tests/unit/test_models.py +++ b/tests/unit/test_models.py @@ -10,6 +10,7 @@ FileRef, JobHandle, JobOutput, + JobStatus, OpOutcome, OutputMember, Submission, @@ -37,6 +38,13 @@ def test_a_newer_format_fails_loudly(self): def test_the_same_id_on_two_resources_is_two_jobs(self): assert len({handle(resource="a"), handle(resource="b")}) == 2 + def test_key_separators_inside_parts_are_escaped(self): + # Without escaping, ("a:b", "c") and ("a", "b:c") would share a key. + first = handle("c", resource="a:b") + second = handle("b:c", resource="a") + assert first.key != second.key + assert str(first) == first.key == "fake:a%3Ab:c" + @pytest.mark.parametrize("name", ["", ".", "/etc/passwd", "../up", "a/../../up"]) def test_names_must_stay_in_the_working_directory(name: str): @@ -62,6 +70,19 @@ def test_secret_content_never_shows(): assert "TOPSECRET" not in shown +def test_an_input_name_is_given_once(): + # Two sources for one file in the working directory: which one would win? + twice = [FileRef(name="in", source="a"), FileRef(name="in", source="b")] + with pytest.raises(ValidationError, match="more than once"): + SubmissionSpec(executable="x", inputs=twice) + + +def test_secrets_that_clash_with_inputs_are_named(): + spec = SubmissionSpec(executable="x", inputs=[FileRef(name="in", source="a")]) + assert spec.clashes_with(CopyVars(secrets={"in": b"s", "other": b"s"})) == {"in"} + assert spec.clashes_with(CopyVars()) == frozenset() + + def test_a_copy_is_either_submitted_or_refused(): with pytest.raises(ValidationError): Submission(handles={0: handle()}, failures={0: "refused"}) @@ -80,14 +101,14 @@ def test_failure_has_a_reason_and_no_value(self): OpOutcome[JobOutput](ok=False, reason="x", value=JobOutput()) -def test_an_input_name_is_given_once(): - # Two sources for one file in the working directory: which one would win? - twice = [FileRef(name="in", source="a"), FileRef(name="in", source="b")] - with pytest.raises(ValidationError, match="more than once"): - SubmissionSpec(executable="x", inputs=twice) - - -def test_secrets_that_clash_with_inputs_are_named(): - spec = SubmissionSpec(executable="x", inputs=[FileRef(name="in", source="a")]) - assert spec.clashes_with(CopyVars(secrets={"in": b"s", "other": b"s"})) == {"in"} - assert spec.clashes_with(CopyVars()) == frozenset() +def test_job_status_values_are_stable(): + """Stored statuses are compared by value: renaming one is a breaking change.""" + assert {status.value for status in JobStatus} == { + "Submitted", + "Waiting", + "Running", + "Done", + "Failed", + "Aborted", + "Unknown", + } diff --git a/tests/unit/test_registry.py b/tests/unit/test_registry.py new file mode 100644 index 0000000..ce50ac3 --- /dev/null +++ b/tests/unit/test_registry.py @@ -0,0 +1,174 @@ +"""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 ( + BatchSSHConfig, + CloudConfig, + ConfigurationError, + JobBackend, + RegistryError, + registry, +) +from intercede._backends.config import FakeSettings +from intercede.registry import _build, _load + +DUMMIES = "tests.unit._registry_dummies" + +CLOUD: dict[str, Any] = { + "kind": "cloud", + "endpoint": "https://cloud.example.org:5000", + "driver": "openstack", + "vm_image": "name:alma9", + "vm_flavor": "name:m1.large", +} +SSH_SLURM: dict[str, Any] = { + "kind": "batch", + "transport": "ssh", + "scheduler": "slurm", + "host": "batch.example.org", + "user": "dirac", + "known_hosts": "batch.example.org ssh-ed25519 AAAAC3Nz...", +} +LOCAL_SLURM: dict[str, Any] = { + "kind": "batch", + "transport": "local", + "scheduler": "slurm", +} + + +@pytest.fixture(autouse=True) +def fresh_cache() -> Iterator[None]: + """Each test sees the tables as it left them, not classes cached by another test.""" + _load.cache_clear() + yield + _load.cache_clear() + + +@pytest.fixture +def composed(monkeypatch: pytest.MonkeyPatch) -> None: + """Register stand-ins for the composed backend, which has no real parts yet.""" + monkeypatch.setitem(registry._BACKENDS, "batch", f"{DUMMIES}:DummyBatchBackend") + monkeypatch.setitem(registry._TRANSPORTS, "ssh", f"{DUMMIES}:DummyTransport") + monkeypatch.setitem(registry._TRANSPORTS, "local", f"{DUMMIES}:DummyTransport") + monkeypatch.setitem(registry._SCHEDULERS, "slurm", f"{DUMMIES}:DummyScheduler") + + +class TestBuilds: + async def test_the_fake_for_the_test_suites(self): + async with _build(FakeSettings(resource="fake://test")) as backend: + assert isinstance(backend, JobBackend) + assert backend.resource == "fake://test" + + @pytest.mark.parametrize( + "config", + [CLOUD, CloudConfig(**CLOUD)], + ids=["mapping", "object"], + ) + def test_a_single_class_backend_gets_its_settings( + self, monkeypatch: pytest.MonkeyPatch, config: Any + ): + monkeypatch.setitem(registry._BACKENDS, "cloud", f"{DUMMIES}:DummyBackend") + built: Any = registry.get_backend(config) + assert type(built).__name__ == "DummyBackend" + assert isinstance(built.settings, CloudConfig) + + def test_a_batch_backend_is_composed_of_its_parts(self, composed: None): + built: Any = registry.get_backend({**SSH_SLURM, "queue": "short"}) + assert type(built).__name__ == "DummyBatchBackend" + # Transport and scheduler each get the whole batch configuration. + assert isinstance(built.transport.settings, BatchSSHConfig) + assert built.transport.settings.host == "batch.example.org" + assert built.scheduler.settings.queue == "short" + + +class TestRefused: + def test_invalid_configuration_keeps_the_details(self): + with pytest.raises(ConfigurationError) as exc_info: + registry.get_backend({"kind": "arc"}) + assert "endpoint: required for kind='arc'" in str(exc_info.value) + + @pytest.mark.parametrize( + "config", + [{"kind": "arc", "endpoint": "https://arc.example.org:443"}, CLOUD], + ids=["arc", "cloud"], + ) + def test_valid_but_not_shipped_yet(self, config: dict[str, Any]): + with pytest.raises( + RegistryError, match=f"backend '{config['kind']}' is not available" + ): + registry.get_backend(config) + + def test_composed_part_not_shipped_yet(self): + with pytest.raises(RegistryError, match="transport 'ssh' is not available"): + registry.get_backend(SSH_SLURM) + + def test_an_entry_that_cannot_be_imported( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._TRANSPORTS, "ssh", "tests.unit.does_not_exist:X") + with pytest.raises(RegistryError, match="cannot be imported") as exc_info: + registry.get_backend(SSH_SLURM) + assert isinstance(exc_info.value.__cause__, ImportError) + + @pytest.mark.parametrize( + ("table", "name", "protocol"), + [ + ("_TRANSPORTS", "local", "Transport"), + ("_SCHEDULERS", "slurm", "Scheduler"), + ("_BACKENDS", "batch", "JobBackend"), + ], + ) + def test_what_breaks_its_protocol_is_refused( + self, + composed: None, + monkeypatch: pytest.MonkeyPatch, + table: str, + name: str, + protocol: str, + ): + monkeypatch.setitem( + getattr(registry, table), name, f"{DUMMIES}:BuildsTheWrongThing" + ) + with pytest.raises(RegistryError, match=f"which is not a {protocol}"): + registry.get_backend(LOCAL_SLURM) + + +class TestAvailable: + def test_shipped_today(self): + assert registry.available() == {"backend": (), "transport": (), "scheduler": ()} + + def test_lists_only_what_a_configuration_can_ask_for( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setitem(registry._SCHEDULERS, "lsf", f"{DUMMIES}:DummyScheduler") + listing = registry.available() + assert listing["transport"] == ("local", "ssh") + assert "lsf" not in listing["scheduler"] + assert "fake" not in listing["backend"] + + +class TestLazyImports: + def test_importing_intercede_imports_no_backend(self): + code = ( + "import sys, intercede, intercede.registry;" + "loaded = [m for m in sys.modules if m.startswith(('intercede._backends.fake', 'tests.'))];" + "assert not loaded, loaded" + ) + subprocess.run([sys.executable, "-c", code], check=True) # noqa: S603 (constant input) + + def test_only_the_requested_entry_is_imported( + self, composed: None, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.delitem(sys.modules, DUMMIES, raising=False) + registry.available() + assert DUMMIES not in sys.modules # listing imports nothing + registry.get_backend(LOCAL_SLURM) + assert DUMMIES in sys.modules