diff --git a/packages/timenet/src/timenet/_fs.py b/packages/timenet/src/timenet/_fs.py index ec216e113..97c7bf03e 100644 --- a/packages/timenet/src/timenet/_fs.py +++ b/packages/timenet/src/timenet/_fs.py @@ -1,5 +1,7 @@ """Replace a directory with a complete staging directory.""" +from collections.abc import Generator +from contextlib import contextmanager from pathlib import Path import shutil import uuid @@ -37,3 +39,28 @@ def publish_directory(staging: Path, final: Path) -> None: if old is not None: # The new directory is published, so failing to delete the old one must not fail the publish. shutil.rmtree(old, ignore_errors=True) + + +@contextmanager +def staged_directory(final: Path) -> Generator[Path]: + """Yield an empty staging directory beside ``final``, and publish it as ``final`` if the block succeeds. + + The old ``final`` stays in place until the new directory is complete. The staging directory is + removed in every case. + + Args: + final: The destination directory. + + Yields: + The staging directory. + """ + final.parent.mkdir(parents=True, exist_ok=True) + # A hard kill can leave a staging directory behind. Do not sweep other .tmp-* directories: + # a concurrent write of the same release may still use one. + staging = final.with_name(f"{final.name}.tmp-{uuid.uuid4().hex}") + staging.mkdir() + try: + yield staging + publish_directory(staging, final) + finally: + shutil.rmtree(staging, ignore_errors=True) diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index bc5ca76b5..64d529a0e 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -308,20 +308,16 @@ def load( return reader.read(load=LoadPlan.ALL_EAGER) def _reject_unhosted_access(self, dataset_id: str, version: str | None) -> None: - """Raise for a non-open dataset read from a registry that does not host its data. - - Credentialed and restricted datasets cannot be redistributed, so TimeNet never serves their - bytes. Such a dataset is build-your-own: a local registry holds only what the user built, so a - local read is fine, but a hosted registry can only point the user at where to obtain access. + """Check dataset access against the registry's requirements. Args: - dataset_id: The bare dataset id. - version: The resolved version, or ``None`` for the latest. + dataset_id: The dataset ID without a version suffix. + version: The version string, or ``None`` for the latest. Raises: - TimeNetAccessError: If the dataset is not open and the registry is not a local one. + TimeNetAccessError: If the dataset or its lineage lacks open access and the registry requires it. """ - if isinstance(self._registry, LocalRegistry): + if not self._registry.requires_open_access: return try: manifest = self._registry.get_manifest(dataset_id, version) @@ -394,20 +390,17 @@ def _missing_hint(*, auto_build: bool, fetch_from: BaseRegistry | None) -> str | def _prepare( registry: LocalRegistry, dataset_id: str, version: str | None, *, auto_build: bool, fetch_from: BaseRegistry | None ) -> DatasetVersion: - """Prepare a local version and its payload sources in the order that :meth:`TimeNet.open` names. + """Prepare a local release and its payload sources. - If a payload source is still missing after the copy from ``fetch_from``, this function builds it - with its own connector. That connector is the backend whose card declares the exact version of - the source. The version reads files of the source, so the build gets the pins of the version. - The build publishes the source only if it reproduces the pinned files. + Try local copies, then ``fetch_from``, then a build if ``auto_build`` is true. + A built payload source must match the files that the release pins. Returns: - The version with each payload source attached. + The release with each payload source attached. Raises: - TimeNetMissingParentError: If a payload source is absent and this function cannot copy or - build it. - TimeNetInvalidCardError: If the card of a connector cannot be read. + TimeNetMissingParentError: If a payload source is missing and cannot be copied or built. + TimeNetInvalidCardError: If a connector's dataset card cannot be read. """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain and exact_builder try: manifest = registry.get_manifest(dataset_id, version) @@ -415,7 +408,7 @@ def _prepare( manifest = _obtain(registry, dataset_id, version, miss, auto_build=auto_build, fetch_from=fetch_from) reference = DatasetRef(dataset_id=dataset_id, version=manifest.metadata.dataset_version) # A registry that is not local serves only releases with access declared ``open``. - check = None if fetch_from is None or isinstance(fetch_from, LocalRegistry) else _require_open + check = _require_open if fetch_from is not None and fetch_from.requires_open_access else None built: set[DatasetRef] = set() while True: try: diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index 2b38af2ab..0cdc59012 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -198,52 +198,29 @@ def plan_parents( build_parents: bool = False, force_build: bool = False, ) -> list[ParentStep]: - """Plan how to get each parent release that a build of ``dataset_id`` needs. - - For each declared parent, a build needs the parent and the payload sources of that parent. It - does not need the payload sources of those payload sources. The plan visits the declared parents - of a release only if the plan builds that release. So if the output already holds the requested - release, the plan visits its parents only if ``force_build`` is set. If ``force_build`` is not - set, the plan visits only the payload sources of that release. - - If the output does not hold a release, the plan fetches it from ``fetch_from``. If ``fetch_from`` - does not hold it either, the plan builds it with the installed connector that declares that - exact version. If a local output holds a damaged copy, the plan fetches the release again and - does not build it. A registry that is not local holds only releases whose cards declare - ``access: open``. So for such an output, each release that the plan visits must declare - ``access: open``. If ``fetch_from`` is not local, each release that the plan fetches from it - must declare ``access: open`` too. - - The plan works in two phases. First, a walk from the requested release finds the graph of the - releases that the build needs. Then :class:`graphlib.TopologicalSorter` orders the graph and - finds cycles, and the plan makes its steps in that order. + """Plan how to fetch or build the releases a dataset needs. + + Prefer complete copies, then ``fetch_from``, then builds if ``build_parents`` is true. + An existing release needs only its payload sources unless ``force_build`` is set. Args: dataset_id: The dataset to build. - output: The registry that the build writes to. - fetch_from: The registry to fetch missing parents from. If it is ``None``, the plan fetches - nothing. + output: The registry to build into. + fetch_from: The registry to fetch missing or damaged parents from, or ``None``. build_parents: Build missing parents with their installed connectors. - force_build: Build the requested release again, even if the output holds it, and plan its - parents. + force_build: Plan the requested release's parents even if the output already holds the release. Returns: - The steps to fetch or build parents. Each step comes after the steps of the releases that it - depends on. + Fetch or build steps, ordered with dependencies first. Raises: - LookupError: If no installed connector declares a card for ``dataset_id``. - TimeNetInvalidCardError: If the card of an installed connector cannot be read. - TimeFValidationError: If the parents form a cycle, if an installed connector declares another - version than a parent pins, or if ``fetch_from`` is set and the output is not local. - TimeNetMissingParentError: If a parent or a payload source is missing and ``build_parents`` - is not set. + LookupError: If no installed connector declares a dataset card for ``dataset_id``. + TimeNetInvalidCardError: If an installed connector's dataset card cannot be read. + TimeFValidationError: If parents cycle, a connector builds the wrong version, or fetching uses non-local output. + TimeNetMissingParentError: If a parent or payload source is missing and ``build_parents`` is false. TimeNetDatasetNotFoundError: If a parent is missing and no installed connector builds it. - TimeNetFileError: If a local output holds a damaged copy of a parent, and the plan cannot - fetch the parent again. - TimeNetAccessError: If the output is not local and the plan visits a release whose card, or - a card in its lineage, does not declare ``access: open``. Also if ``fetch_from`` is not - local and the plan would fetch such a release from it. + TimeNetFileError: If a local parent copy has a missing or resized file and cannot be fetched again. + TimeNetAccessError: If a release or its lineage lacks open access where a registry requires it. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers of the walk if fetch_from is not None and not isinstance(output, LocalRegistry): # A registry that is not local accepts a child only if it already holds the parents. Also, a @@ -259,7 +236,7 @@ def plan_parents( # If the output already holds the requested release, a reader needs only its payload sources to # open it. Its declared parents matter only to a rebuild. held = None if force_build else _manifest(output, root) - if held is None and not isinstance(output, LocalRegistry): + if held is None and output.requires_open_access: require_open_access(card) searched = str(output) if fetch_from is None else f"{output} or {fetch_from}" graph = _Graph(output, fetch_from, build_parents=build_parents, searched=searched) @@ -346,21 +323,16 @@ def _locate( ) -> tuple[Manifest | None, Literal["fetch"] | None]: # The damage is raised from a variable, the access error by require_open_access. # ruff: ignore[docstring-extraneous-exception] - """Find a release in the output, or else in ``fetch_from``. + """Find a release in ``output`` or ``fetch_from``. - The output must hold an intact copy. In a local output, a copy with a missing file or a file of - another size is damaged. + A local copy is intact when every file has its recorded size. Returns: - The manifest and ``None`` if the output holds an intact copy. The manifest and ``"fetch"`` if - the plan fetches the release. ``None`` and ``None`` if the plan builds it. + ``(manifest, None)`` for an output copy, ``(manifest, "fetch")`` for a fetch, or ``(None, None)`` if absent. Raises: - TimeNetFileError: If a local output holds a copy with a missing file or a file of another - size, and the plan cannot fetch the release again. - TimeNetAccessError: If the card of the release, or a card in its lineage, does not declare - ``access: open``, and the output that holds the release, or the ``fetch_from`` registry - to fetch it from, is not local. + TimeNetFileError: If the local copy has a missing or resized file and ``fetch_from`` lacks the release. + TimeNetAccessError: If the release or its lineage lacks open access where its registry requires it. """ manifest = _manifest(output, reference) damage = None @@ -375,14 +347,14 @@ def _locate( except TimeNetFileError as error: manifest, damage = None, error if manifest is not None: - if not isinstance(output, LocalRegistry): + if output.requires_open_access: require_open_access(manifest.metadata, manifest.lineage) return manifest, None fetched = None if fetch_from is None else _manifest(fetch_from, reference) if fetched is not None: - # A registry that is not local holds only releases whose cards declare access: open. So the - # plan refuses another release before any download. - if not isinstance(fetch_from, LocalRegistry): + # A registry that requires open access serves only releases whose cards declare access: open. + # So the plan refuses another release before any download. + if fetch_from is not None and fetch_from.requires_open_access: require_open_access(fetched.metadata, fetched.lineage) return fetched, "fetch" if damage is not None: @@ -399,19 +371,19 @@ def _card_to_build( # ruff: ignore[too-many-arguments] pinned: bool, searched: str, ) -> DatasetMetadata: - """Return the card to build a release that the output does not hold and the plan cannot fetch. + """Find a dataset card for building a missing release. Returns: - The card of the installed connector that builds exactly ``reference``. + The installed connector's card for the exact release in ``reference``. Raises: - TimeNetMissingParentError: If ``build_parents`` is not set. - TimeNetAccessError: If the output is not local and the card does not declare ``access: open``. + TimeNetMissingParentError: If ``build_parents`` is false. + TimeNetAccessError: If the card lacks open access and the output requires it. """ # ruff: ignore[docstring-extraneous-exception] - the access error is raised by require_open_access if not build_parents: - raise TimeNetMissingParentError(reference, needed_by, pinned, searched) + raise TimeNetMissingParentError(reference, needed_by, searched, payload_source=pinned) card = _exact_card(reference) - if not isinstance(output, LocalRegistry): + if output.requires_open_access: require_open_access(card) return card diff --git a/packages/timenet/src/timenet/engine/engine.py b/packages/timenet/src/timenet/engine/engine.py index d592bf0d4..f34c8d339 100644 --- a/packages/timenet/src/timenet/engine/engine.py +++ b/packages/timenet/src/timenet/engine/engine.py @@ -113,26 +113,26 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] force: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> str: - """Run a connector and publish its dataset to a registry. + """Build and publish a dataset to a writable registry. - If the version already exists, return it unless ``force`` is set. Remote and S3 registries - require ``access: open`` on the dataset card and its lineage. + Reuse an existing version unless ``force`` is set. + The dataset card and lineage must declare open access if the registry requires it. Args: connector: The connector to build. - registry: The destination registry. - cache_dir: Download directory, or ``None`` for ``/``. - keep_cache: Keep the default cache after publishing (a supplied ``cache_dir`` is always kept). - progress_cb: Optional callback for write progress. - force: Republish even if the version already exists. - pinned_by: Releases that already read this release's files, with the pins that the new files must match. + registry: The writable registry to publish to. + cache_dir: The download directory, or ``None`` for ``/``. + keep_cache: Keep the default cache after publishing. A supplied ``cache_dir`` is always kept. + progress_cb: Callback for writer progress events, or ``None``. + force: Publish again even if the version already exists. + pinned_by: Releases that read this release's files, mapped to file records the build must match. Returns: - The published version string. + The existing or newly published version string. Raises: - TimeFValidationError: If declared parents lack a custom ``compose``, or ``compose`` changes the dataset card. - TimeNetAccessError: If a nonlocal publish includes a dataset or ancestor without open access. + TimeFValidationError: If a connector declares parents without overriding ``compose``, or changes its card. + TimeNetAccessError: If the card, dataset, or lineage lacks open access and the registry requires it. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_composes, _compose, and require_open_access values_backend = _check_values_backend(connector) metadata = connector.metadata() @@ -140,7 +140,7 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] version = str(metadata.dataset_version) if not force and registry.exists(dataset_id, version): return version - if not isinstance(registry, LocalRegistry): + if registry.requires_open_access: require_open_access(metadata) _check_composes(connector, metadata) @@ -149,7 +149,7 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] with open_parents(metadata, registry) as context: dataset = _compose(connector, metadata, cache, context) - if not isinstance(registry, LocalRegistry): + if registry.requires_open_access: require_open_access(dataset.metadata, dataset.lineage.values()) dataset.derive_schema() registry.put_dataset( diff --git a/packages/timenet/src/timenet/errors.py b/packages/timenet/src/timenet/errors.py index a1ee95f82..f87b91f26 100644 --- a/packages/timenet/src/timenet/errors.py +++ b/packages/timenet/src/timenet/errors.py @@ -12,7 +12,8 @@ sentence. """ -from typing import TYPE_CHECKING +from functools import partial +from typing import TYPE_CHECKING, Literal if TYPE_CHECKING: @@ -33,31 +34,75 @@ class TimeNetDatasetNotFoundError(TimeNetError): class TimeNetMissingParentError(TimeNetDatasetNotFoundError): - """A release that another release needs is absent: a declared parent of a build, or a payload source. + """A required parent release or payload source is missing. Attributes: missing: The absent release. needed_by: The release that needs it. - payload_source: Whether ``needed_by`` reads files of ``missing`` (a payload source), rather than - declaring it as a parent to build from. - searched: Where the release was looked for. - hint: How the caller can obtain the release, added by the front end, or ``None``. + payload_source: ``True`` if ``needed_by`` reads files of ``missing``, ``False`` for a declared build parent. + searched: The search location. + hint: Advice on obtaining the release, or ``None``. """ - def __init__(self, missing: "DatasetRef", needed_by: "DatasetRef", payload_source: bool, searched: str) -> None: - super().__init__(missing, needed_by, payload_source, searched) + def __init__(self, missing: "DatasetRef", needed_by: "DatasetRef", searched: str, *, payload_source: bool) -> None: + super().__init__(missing, needed_by, searched, payload_source) self.missing = missing self.needed_by = needed_by self.payload_source = payload_source self.searched = searched self.hint: str | None = None + def __reduce__(self) -> tuple[object, ...]: + # Pickle passes the arguments by position, so the keyword-only flag travels in a partial. + rebuild = partial(type(self), payload_source=self.payload_source) + return rebuild, (self.missing, self.needed_by, self.searched), self.__dict__ + def __str__(self) -> str: role = "payload source" if self.payload_source else "parent release" message = f"{role} {self.missing} of {self.needed_by} is not in {self.searched}." return message if self.hint is None else f"{message} {self.hint}" +class TimeNetVersionConflictError(TimeNetError): + """A registry already holds a release with other files or another card than the copy. + + A changed release needs a new version. + + Attributes: + release: The release that differs. + registry: The destination registry, as text. + source: The source registry, as text. + needed_by: The requested release that needs this payload source, or ``None`` if ``release`` is the request. + difference: ``"files"`` if the file records differ, else ``"card"`` for the rest of the manifest. + hint: Advice to append to the error message, or ``None``. + """ + + def __init__( + self, + release: "DatasetRef", + registry: str, + source: str, + needed_by: "DatasetRef | None" = None, + difference: Literal["files", "card"] = "files", + ) -> None: + super().__init__(release, registry, source, needed_by, difference) + self.release = release + self.registry = registry + self.source = source + self.needed_by = needed_by + self.difference = difference + self.hint: str | None = None + + def __str__(self) -> str: + held = str(self.release) if self.needed_by is None else f"{self.release}, a payload source of {self.needed_by}," + other = "other files" if self.difference == "files" else "another card" + message = ( + f"{self.registry} already holds {held} with {other} than {self.source}. " + "A published version cannot change, so a changed release needs a new version." + ) + return message if self.hint is None else f"{message} {self.hint}" + + class TimeNetAccessError(TimeNetError): """A dataset needs credentialed or restricted access, so only the user's own registry may hold its data. diff --git a/packages/timenet/src/timenet/manifest/__init__.py b/packages/timenet/src/timenet/manifest/__init__.py index 480bdf788..9f8984f9a 100644 --- a/packages/timenet/src/timenet/manifest/__init__.py +++ b/packages/timenet/src/timenet/manifest/__init__.py @@ -2,7 +2,7 @@ from timenet.manifest.counts import ManifestCounts from timenet.manifest.files import FilePart, ManifestFiles -from timenet.manifest.manifest import BuildEnvironment, Manifest, parse_manifest, read_manifest +from timenet.manifest.manifest import BuildEnvironment, Manifest, parse_manifest, read_manifest, same_release from timenet.manifest.payload_source import PayloadSource @@ -15,4 +15,5 @@ "PayloadSource", "parse_manifest", "read_manifest", + "same_release", ] diff --git a/packages/timenet/src/timenet/manifest/manifest.py b/packages/timenet/src/timenet/manifest/manifest.py index f93815e71..fb1137ff9 100644 --- a/packages/timenet/src/timenet/manifest/manifest.py +++ b/packages/timenet/src/timenet/manifest/manifest.py @@ -106,6 +106,19 @@ def to_json(self) -> str: return validated.model_dump_json(indent=2, by_alias=True) +def same_release(a: Manifest, b: Manifest) -> bool: + """Compare two manifests, ignoring their build environments. + + Args: + a: One manifest. + b: The other manifest. + + Returns: + Whether the manifests are equal apart from ``build_env``. + """ + return a.model_copy(update={"build_env": {}}) == b.model_copy(update={"build_env": {}}) + + def parse_manifest(text: str | bytes, where: str) -> Manifest: """Parse the JSON text of a stored manifest. diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py new file mode 100644 index 000000000..04a399754 --- /dev/null +++ b/packages/timenet/src/timenet/progress.py @@ -0,0 +1,44 @@ +"""Progress events for copying and building releases. + +Each release starts with a :class:`StepStarted` event. +A :class:`StepFinished` event follows only if the copy or build succeeds. +""" + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Literal + +from timenet.types.references import DatasetRef + + +@dataclass(frozen=True) +class StepStarted: + """The start of a copy or build for one release. + + Attributes: + release: The release to copy or build. + action: The operation, either ``"copy"`` or ``"build"``. + """ + + release: DatasetRef + action: Literal["copy", "build"] + + +@dataclass(frozen=True) +class StepFinished: + """The successful completion of a copy or build for one release. + + Attributes: + release: The copied or built release. + action: The completed operation, either ``"copy"`` or ``"build"``. + """ + + release: DatasetRef + action: Literal["copy", "build"] + + +ProgressEvent = StepStarted | StepFinished +"""An event that marks the start or successful end of a copy or build.""" + +ProgressCallback = Callable[[ProgressEvent], None] +"""A callback that receives a progress event and returns nothing.""" diff --git a/packages/timenet/src/timenet/registry/__init__.py b/packages/timenet/src/timenet/registry/__init__.py index eaa0a5dfc..bc13fea69 100644 --- a/packages/timenet/src/timenet/registry/__init__.py +++ b/packages/timenet/src/timenet/registry/__init__.py @@ -1,5 +1,6 @@ """Dataset registries: serve compiled TimeF versions and search over metadata.""" +from timenet.registry._copy import copy_release from timenet.registry.base import BaseRegistry from timenet.registry.dependencies import open_local_version from timenet.registry.factory import ( @@ -24,6 +25,7 @@ "RemoteRegistry", "S3Registry", "WritableRegistry", + "copy_release", "default_registry_path", "local_registry_path", "open_local_version", diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py new file mode 100644 index 000000000..b9ec678e9 --- /dev/null +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -0,0 +1,225 @@ +"""Copy releases and their payload sources between registries. + +A payload source is a release whose data files another release reads. +Manifests must match for existing releases, except for their build environments. +""" + +from pathlib import Path +import tempfile + +from timenet.errors import TimeNetMissingParentError, TimeNetVersionConflictError +from timenet.manifest import Manifest, same_release +from timenet.progress import ProgressCallback, StepFinished, StepStarted +from timenet.registry._paths import release_dir +from timenet.registry.base import BaseRegistry +from timenet.registry.dependencies import find_payload_sources, validate_manifest_reference +from timenet.registry.local import LocalRegistry +from timenet.registry.writable import WritableRegistry +from timenet.types.access import require_open_access +from timenet.types.references import DatasetRef + + +def copy_release( + reference: DatasetRef, + source: BaseRegistry, + dest: WritableRegistry, + *, + with_payload_sources: bool = True, + progress: ProgressCallback | None = None, +) -> None: + """Copy a release and its payload sources to another registry. + + Copy payload sources first, reuse complete releases in ``dest``, and fetch damaged local copies again. + All checks run before the first copy, so a failed check leaves ``dest`` unchanged. + + Args: + reference: The exact release to copy. + source: The registry to copy from. + dest: The registry to copy to. + with_payload_sources: Also copy the payload sources that the manifest of the release lists. + progress: Callback for ``StepStarted`` and ``StepFinished`` events for each copied or repaired release. + + Raises: + TimeNetVersionConflictError: If ``dest`` holds the release or a payload source with other files or another card. + TimeNetAccessError: If a release lacks open access where the source or destination requires it. + TimeNetDatasetNotFoundError: If ``source`` does not hold the release. + TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` lacks a complete copy. + TimeFFormatError: If a manifest names the wrong release or pinned files differ. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers + manifests = _manifests(reference, source, dest, with_payload_sources=with_payload_sources) + # Check every release before anything is copied, so that a refusal leaves dest unchanged. A payload + # source that only dest holds is complete there, so it is never pending. + pending = [ref for ref, manifest in manifests.items() if _needs_copy(reference, ref, manifest, source, dest)] + _check_access(manifests, set(pending), source, dest) + if not pending: + return + # Only a copy from a non-local source to a non-local dest downloads into this directory, each + # pending release on its own. + with tempfile.TemporaryDirectory(prefix="timenet-copy-", ignore_cleanup_errors=True) as staging: + for ref in pending: + if progress is not None: + progress(StepStarted(ref, "copy")) + _copy_one(ref, manifests[ref], source, dest, Path(staging)) + if progress is not None: + progress(StepFinished(ref, "copy")) + + +def _copy_one( + reference: DatasetRef, manifest: Manifest, source: BaseRegistry, dest: WritableRegistry, staging: Path +) -> None: + """Copy one release without its payload sources to ``dest``, or repair its local copy. + + Args: + reference: The release to copy. + manifest: The release's manifest from ``source``. + source: The registry to copy from. + dest: The registry to copy to. + staging: Temporary download root when both registries are remote. + """ + # A local registry is a folder that the copy reads or writes directly. + if isinstance(dest, LocalRegistry): + source.download_release(reference, release_dir(dest.local_root, reference), manifest=manifest) + return + if isinstance(source, LocalRegistry): + directory = release_dir(source.root, reference) + else: + # put_release() needs only the release directory, and the walk checked the pins. So the payload + # sources, which dest may hold already, are not downloaded. The download checks each file by size + # and checksum. + directory = release_dir(staging, reference) + source.download_release(reference, directory, manifest=manifest) + dest.put_release(directory) + + +def _check_access( + manifests: dict[DatasetRef, Manifest], fetched: set[DatasetRef], source: BaseRegistry, dest: WritableRegistry +) -> None: + """Check release access against the source and destination requirements. + + Check all releases against ``dest`` and fetched releases against ``source``. + + Args: + manifests: Release references mapped to their manifests. + fetched: The releases that the copy fetches from ``source``. + source: The registry to copy from. + dest: The registry to copy to. + + Raises: + TimeNetAccessError: If a release or its lineage lacks open access where a registry requires it. + """ # ruff: ignore[docstring-extraneous-exception] - raised by require_open_access + for reference, manifest in manifests.items(): + dest_needs_open = dest.requires_open_access + source_needs_open = reference in fetched and source.requires_open_access + if dest_needs_open or source_needs_open: + require_open_access(manifest.metadata, manifest.lineage) + + +def _manifests( + reference: DatasetRef, source: BaseRegistry, dest: WritableRegistry, *, with_payload_sources: bool +) -> dict[DatasetRef, Manifest]: + """Collect the manifests of a release and of the payload sources that its manifest lists. + + Use a complete copy in ``dest`` when ``source`` does not serve a payload source. + + Args: + reference: The requested release. + source: The registry to copy from. + dest: The registry to copy to. + with_payload_sources: Also collect the payload sources of the release. + + Returns: + Manifests keyed by release, with the payload sources before the release. + + Raises: + TimeNetAccessError: If ``source`` lists a payload source without serving it and ``dest`` lacks a complete copy. + TimeNetMissingParentError: If neither ``source`` nor ``dest`` holds a complete payload source. + """ # ruff: ignore[docstring-extraneous-exception] - raised by require_open_access + manifest = source.get_manifest(reference.dataset_id, str(reference.version)) + validate_manifest_reference(manifest, reference) + if not with_payload_sources: + return {reference: manifest} + sources = find_payload_sources( + reference, manifest, source, find=source.find_manifest, held=lambda ref: _complete_copy(ref, dest) + ) + for missing in sources.missing: + listed = source.find_manifest(missing) + if listed is not None: + # The source lists the payload source, but does not serve its files. + require_open_access(listed.metadata, listed.lineage) + raise TimeNetMissingParentError(missing, reference, str(source), payload_source=True) + return {**sources.fetched, **sources.held, reference: manifest} + + +def _complete_copy(reference: DatasetRef, dest: WritableRegistry) -> Manifest | None: + """Return the manifest of a complete copy in a registry. + + Args: + reference: The release. + dest: The registry to look in. + + Returns: + The manifest, or ``None`` if ``dest`` lacks the release or a local file is missing or resized. + """ + held = dest.find_manifest(reference) + return held if held is not None and _complete(reference, held, dest) else None + + +def _needs_copy( + requested: DatasetRef, + reference: DatasetRef, + manifest: Manifest, + source: BaseRegistry, + dest: WritableRegistry, +) -> bool: + """Check whether a release needs copying or local repair. + + Args: + requested: The requested release, for the error message. + reference: The release to check. + manifest: The release's manifest from ``source``, or from ``dest`` if only ``dest`` holds it. + source: The source registry, for the error message. + dest: The registry to copy to. + + Returns: + ``True`` if ``dest`` lacks the release or its local copy has a missing or resized file. + + Raises: + TimeNetVersionConflictError: If ``dest`` holds the same version with other files or another card. + """ + held = dest.find_manifest(reference) + if held is None: + return True + files_differ = _file_records(held) != _file_records(manifest) + if not files_differ and same_release(held, manifest): + return not _complete(reference, manifest, dest) + needed_by = None if reference == requested else requested + raise TimeNetVersionConflictError(reference, str(dest), str(source), needed_by, "files" if files_differ else "card") + + +def _complete(reference: DatasetRef, manifest: Manifest, dest: WritableRegistry) -> bool: + """Check if a stored release is complete. + + A local copy is complete when every file has its recorded size. + Treat releases in non-local registries as complete. + + Args: + reference: The release to check. + manifest: The manifest that records the files in ``dest``. + dest: The registry that holds the release. + + Returns: + ``True`` for a complete local copy or any release in a non-local registry. + """ + return not isinstance(dest, LocalRegistry) or dest.damage(reference, manifest) is None + + +def _file_records(manifest: Manifest) -> frozenset[tuple[str, int, str]]: + """Collect each file's path, size, and checksum from a manifest. + + Args: + manifest: The manifest to read. + + Returns: + A frozen set of ``(path, size, checksum)`` entries. + """ + return frozenset((part.path, part.size, part.checksum) for part in manifest.files.all_files()) diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index 18f953ad8..afbea2b7e 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -11,13 +11,14 @@ from pathlib import Path import shutil from typing import BinaryIO, TypeVar -import uuid -from timenet.errors import TimeFValidationError, TimeNetFileError +from timenet._fs import staged_directory +from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError, TimeNetFileError from timenet.format.checksums import file_checksum +from timenet.format.constants import MANIFEST_FILE from timenet.manifest import FilePart, Manifest from timenet.registry._paths import safe_version_path, split_release_dir -from timenet.registry.dependencies import prepare_local_copy +from timenet.registry.dependencies import prepare_local_copy, validate_manifest_reference from timenet.registry.shared import complete_copy from timenet.registry.version import DatasetVersion from timenet.types import DatasetMetadata, Domain, License, Task @@ -32,7 +33,7 @@ class BaseRegistry(ABC): - """A source of TimeF datasets: their manifests, files, and searchable metadata.""" + """A registry of dataset manifests, files, and searchable metadata.""" def __str__(self) -> str: return type(self).__name__ @@ -42,6 +43,11 @@ def __str__(self) -> str: def local_root(self) -> Path: """The local registry or cache directory that holds prepared releases.""" + @property + def requires_open_access(self) -> bool: + """Whether stored or served releases must declare open access.""" + return True + @abstractmethod def list_datasets(self) -> list[DatasetMetadata]: """Return the metadata of every dataset, using the latest version. The result is sorted by dataset id. @@ -65,6 +71,25 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest: TimeNetDatasetNotFoundError: If the dataset id or version is unknown. """ + def find_manifest(self, release: DatasetRef) -> Manifest | None: + """Return the manifest of a release, or ``None`` if this registry does not hold it. + + Args: + release: The exact release. + + Returns: + The manifest, or ``None``. + + Raises: + TimeFFormatError: If the stored manifest names another release. + """ # ruff: ignore[docstring-extraneous-exception] - raised by validate_manifest_reference + try: + manifest = self.get_manifest(release.dataset_id, str(release.version)) + except TimeNetDatasetNotFoundError: + return None + validate_manifest_reference(manifest, release) + return manifest + @abstractmethod def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: """Open one file of a dataset version for binary reading. @@ -143,60 +168,43 @@ def download_version( # ruff: ignore[too-many-arguments] progress_cb=progress_cb, ) - def _download_release( # ruff: ignore[too-many-arguments] + def download_release( self, - dataset_id: str, - version: str, - dest_dir: str | Path, + reference: DatasetRef, + directory: str | Path, *, force: bool = False, manifest: Manifest | None = None, progress_cb: ProgressCallback | None = None, ) -> None: - """Download a version's files into ``dest_dir``, swapping the directory in atomically. + """Download one release's files to a local directory. - Fetches every file through :meth:`open_file` into a sibling ``.tmp-*`` staging - directory, with ``manifest.json`` last as the commit marker. It then renames the staging - directory over ``dest_dir`` in one step. An interrupted download never leaves a half-written - copy in place. Subclasses can override this method with a faster path, such as concurrent - streaming from a remote service. + Reuse a complete copy unless ``force`` is set. + Replace ``directory`` only after every downloaded file matches its manifest record. Args: - dataset_id: The dataset id. - version: The version string. - dest_dir: The target ``<...>//`` directory. - force: Re-download even if a copy already exists. - manifest: The parsed manifest, passed to avoid re-fetching it. Fetched if ``None``. - progress_cb: Called with each staged file's byte count, for a progress display. + reference: The exact release. + directory: The local release directory, ``//``. + force: Download again even if a complete copy exists. + manifest: The release's manifest, or ``None`` to fetch it from this registry. + progress_cb: Callback that receives each downloaded file's byte count, or ``None``. Raises: - TimeNetFileError: If a downloaded file's size or checksum differs from the manifest. + TimeNetFileError: If a downloaded file's size or checksum differs from its manifest record. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_staged_file + dataset_id, version = reference.dataset_id, str(reference.version) if manifest is None: manifest = self.get_manifest(dataset_id, version) - target = Path(dest_dir) + target = Path(directory) if complete_copy(target, manifest) and not force: return # already downloaded and current - parent = target.parent - parent.mkdir(parents=True, exist_ok=True) - # Each download stages into a unique dir and removes it in the finally below. A hard kill - # (SIGKILL/power loss) can leave one behind, but do NOT sweep sibling .tmp-* dirs here: - # a concurrent download of the same version has a live staging dir with the same prefix, and - # sweeping it would break that download mid-write. A rare orphaned dir is the lesser evil. - staging = parent / f"{target.name}.tmp-{uuid.uuid4().hex}" - release = DatasetRef.parse(f"{dataset_id}@{version}") - try: + with staged_directory(target) as staging: for part in manifest.files.all_files(): self._stage_file(dataset_id, version, part.path, staging) - self._check_staged_file(staging / part.path, part, release) + self._check_staged_file(staging / part.path, part, reference) if progress_cb is not None: progress_cb(part.size) - (staging / "manifest.json").write_text(manifest.to_json()) - if target.exists(): - shutil.rmtree(target) - staging.replace(target) - finally: - shutil.rmtree(staging, ignore_errors=True) + (staging / MANIFEST_FILE).write_text(manifest.to_json()) def _check_staged_file(self, path: Path, part: FilePart, release: DatasetRef) -> None: """Compare one downloaded file with its manifest record, first by size, then by checksum. diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index 8836975e7..29b533bd4 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -9,7 +9,7 @@ """ from collections.abc import Callable, Iterable -from dataclasses import replace +from dataclasses import dataclass, field, replace from pathlib import Path from typing import TYPE_CHECKING @@ -27,6 +27,7 @@ from timenet.registry._paths import release_dir from timenet.registry.shared import complete_copy from timenet.registry.version import DatasetVersion +from timenet.types.access import Access, effective_access from timenet.types.reference_keys import SourceKey from timenet.types.references import DatasetRef @@ -135,7 +136,7 @@ def attach_payload_sources( try: source = open_source(pinned.release) except TimeNetDatasetNotFoundError as error: - raise TimeNetMissingParentError(pinned.release, release, True, searched) from error + raise TimeNetMissingParentError(pinned.release, release, searched, payload_source=True) from error validate_manifest_reference(source.manifest, pinned.release) check_pins(release, pinned, source.manifest) check_file_sizes(source, pinned.files, owner=f"payload source {pinned.release} of {release}") @@ -143,6 +144,74 @@ def attach_payload_sources( return replace(version, payload_sources=sources) +@dataclass(frozen=True) +class CopySources: + """The payload sources of a release, by where a copy of the release finds them.""" + + fetched: dict[DatasetRef, Manifest] = field(default_factory=dict) + """The payload sources that the registry serves, with their manifests there. The copy brings them.""" + held: dict[DatasetRef, Manifest] = field(default_factory=dict) + """The other payload sources that the destination holds as complete copies, with their manifests.""" + missing: dict[DatasetRef, PayloadSource] = field(default_factory=dict) + """The payload sources that neither holds, with the pins of the release on each.""" + + +def find_payload_sources( + release: DatasetRef, + manifest: Manifest, + registry: "BaseRegistry", + *, + find: Callable[[DatasetRef], Manifest | None], + held: Callable[[DatasetRef], Manifest | None], +) -> CopySources: + """Find each payload source of a release in the registry to copy from, else in the destination. + + A release needs exactly the payload sources that its manifest lists, never their own payload sources. + + Args: + release: The release to copy. + manifest: The manifest of ``release``. + registry: The registry to copy from. + find: Return the manifest of a release in ``registry``, or ``None`` if it lacks the release. + held: Return the manifest of a complete copy in the destination, or ``None``. + + Returns: + The payload sources, by where the copy finds them. + + Raises: + TimeFFormatError: If a payload source that the copy finds differs from the files that ``release`` pins. + """ # ruff: ignore[docstring-extraneous-exception] - raised by check_pins + sources = CopySources() + for pinned in manifest.payload_sources.values(): + source = pinned.release + found = find(source) + if found is not None and _serves(registry, found): + sources.fetched[source] = found + elif (found := held(source)) is not None: + sources.held[source] = found + else: + sources.missing[source] = pinned + continue + check_pins(release, pinned, found) + return sources + + +def _serves(registry: "BaseRegistry", manifest: Manifest) -> bool: + """Check whether a registry serves the files of a release that it holds. + + A registry that requires open access serves a release only if its card and each card in its lineage are open. + + Args: + registry: The registry that holds the release. + manifest: The manifest of the release in ``registry``. + + Returns: + ``True`` if the registry serves the files. + """ + open_access = effective_access(manifest.metadata, manifest.lineage) is Access.OPEN + return open_access or not registry.requires_open_access + + def check_pins(release: DatasetRef, pinned: PayloadSource, manifest: Manifest) -> None: """Check that a payload source still records each pinned file as the release pins it. @@ -362,24 +431,19 @@ def _download_dataset_version( # ruff: ignore[too-many-arguments] progress_cb: Callable[[int], None] | None, before_fetch: Callable[[Manifest], object] | None = None, ) -> Manifest: - """Fetch one complete release, checking its identity before downloading its files. + """Check a release's identity and download its files. + + Run ``before_fetch`` after checking the manifest and before downloading files. Returns: - The manifest after the registry has verified and stored the downloaded files. + The manifest of the downloaded and verified release. """ if manifest is None: manifest = registry.get_manifest(reference.dataset_id, str(reference.version)) validate_manifest_reference(manifest, reference) if before_fetch is not None: before_fetch(manifest) - registry._download_release( - reference.dataset_id, - str(reference.version), - directory, - force=True, - manifest=manifest, - progress_cb=progress_cb, - ) + registry.download_release(reference, directory, force=True, manifest=manifest, progress_cb=progress_cb) return manifest diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 81b3a2eaf..0376d8ab0 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -4,15 +4,16 @@ from pathlib import Path import shutil from typing import BinaryIO -import uuid -from timenet._fs import publish_directory +import pyarrow.fs as pafs + +from timenet._fs import staged_directory from timenet.dataset import TimeFDataset -from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError +from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError, TimeNetFileError from timenet.format.constants import MANIFEST_FILE from timenet.manifest import Manifest, PayloadSource, read_manifest from timenet.registry._paths import release_dir -from timenet.registry.dependencies import open_local_release +from timenet.registry.dependencies import check_file_sizes, open_local_release from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry from timenet.types import DatasetMetadata, Version @@ -48,6 +49,15 @@ def local_root(self) -> Path: """The directory that holds complete local releases.""" return self._root + @property + def requires_open_access(self) -> bool: + """Whether local releases must declare open access. + + Returns: + ``False``, since a local registry accepts any access level. + """ + return False + def list_datasets(self) -> list[DatasetMetadata]: """Return the latest-version metadata of every dataset, sorted by id. @@ -204,19 +214,32 @@ def put_release(self, from_directory: Path) -> DatasetRef: """ # ruff: ignore[docstring-extraneous-exception] - raised by _read_release and publish_directory manifest = self._read_release(from_directory) reference = DatasetRef(dataset_id=manifest.dataset_id, version=manifest.metadata.dataset_version) - final = release_dir(self._root, reference) - final.parent.mkdir(parents=True, exist_ok=True) - staging = final.with_name(f"{final.name}.tmp-{uuid.uuid4().hex}") - try: + with staged_directory(release_dir(self._root, reference)) as staging: for relpath in (*sorted({part.path for part in manifest.files.all_files()}), MANIFEST_FILE): target = staging / relpath target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(from_directory / relpath, target) - publish_directory(staging, final) - finally: - shutil.rmtree(staging, ignore_errors=True) return reference + def damage(self, release: DatasetRef, manifest: Manifest | None = None) -> TimeNetFileError | None: + """Return the first missing or resized file of a stored release, or ``None`` for a complete copy. + + Args: + release: The exact release. + manifest: Its manifest, or ``None`` to read the stored one. + + Returns: + The error that names the file, or ``None``. + """ + if manifest is None: + manifest = self.get_manifest(release.dataset_id, str(release.version)) + version = DatasetVersion(manifest, pafs.LocalFileSystem(), str(release_dir(self._root, release))) + try: + check_file_sizes(version, manifest.files.all_files(), owner=str(release)) + except TimeNetFileError as error: + return error + return None + def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: """Open a committed version as a local, random-access handle. diff --git a/packages/timenet/src/timenet/registry/remote/_download.py b/packages/timenet/src/timenet/registry/remote/_download.py index 2006933d8..093a7089b 100644 --- a/packages/timenet/src/timenet/registry/remote/_download.py +++ b/packages/timenet/src/timenet/registry/remote/_download.py @@ -11,13 +11,12 @@ from collections.abc import Coroutine import hashlib from pathlib import Path -import shutil import threading from typing import Any -import uuid import httpx +from timenet._fs import staged_directory from timenet.errors import TimeNetFileError from timenet.manifest import Manifest from timenet.manifest.files import FilePart @@ -52,13 +51,13 @@ def download_version_files( # ruff: ignore[too-many-arguments] Args: http: The registry HTTP client. - manifest: The version's parsed manifest (its file list drives the download). + manifest: The release's manifest. dataset_id: The ``org/name`` id. version: The version string. dest_dir: The target ``<...>//`` directory. force: Re-download even if ``dest_dir/manifest.json`` already exists. max_concurrency: Maximum concurrent file downloads. - progress_cb: Called with each chunk's byte count as it is written, for a progress display. + progress_cb: Callback that receives each downloaded chunk's byte count, or ``None``. Returns: ``dest_dir``. @@ -67,21 +66,9 @@ def download_version_files( # ruff: ignore[too-many-arguments] if complete_copy(dest_dir, manifest) and not force: return dest_dir files = list(manifest.files.all_files()) - parent = dest_dir.parent - parent.mkdir(parents=True, exist_ok=True) - # Each download stages into a unique dir and removes it in the finally below. A hard kill - # (SIGKILL/power loss) can leave one behind, but do NOT sweep sibling .tmp-* dirs here: a - # concurrent download of the same version has a live staging dir with the same prefix, and sweeping - # it would break that download mid-write. A rare orphaned dir is the lesser evil. - staging = parent / f"{dest_dir.name}.tmp-{uuid.uuid4().hex}" - try: + with staged_directory(dest_dir) as staging: _run(_download_all(http, dataset_id, version, files, staging, max_concurrency, progress_cb)) (staging / "manifest.json").write_text(manifest.to_json()) # already parsed; no extra round trip - if dest_dir.exists(): - shutil.rmtree(dest_dir) - staging.replace(dest_dir) - finally: - shutil.rmtree(staging, ignore_errors=True) return dest_dir diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index 35a2740a4..258363079 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -130,30 +130,29 @@ def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: filesystem = RegistryFileSystem(self._base_url, files, token=self._http.token, transport=self._http.transport) return DatasetVersion(manifest, pafs.PyFileSystem(pafs.FSSpecHandler(filesystem)), root) - def _download_release( # ruff: ignore[too-many-arguments] + def download_release( self, - dataset_id: str, - version: str, - dest_dir: str | Path, + reference: DatasetRef, + directory: str | Path, *, force: bool = False, manifest: Manifest | None = None, progress_cb: ProgressCallback | None = None, ) -> None: - """Download a version's files into ``dest_dir`` (used by ``TimeNet.download`` for remotes). + """Download one release's files to a local directory. Args: - dataset_id: The ``org/name`` id. - version: The version string. - dest_dir: The target ``<...>//`` directory. - force: Re-download even if a copy already exists. - manifest: The already-parsed manifest, passed to avoid re-fetching it; fetched if ``None``. - progress_cb: Called with each chunk's byte count as it is written, for a progress display. + reference: The exact release. + directory: The local release directory, ``//``. + force: Download again even if a complete copy exists. + manifest: The release's manifest, or ``None`` to fetch it from this registry. + progress_cb: Callback that receives each downloaded chunk's byte count, or ``None``. """ + dataset_id, version = reference.dataset_id, str(reference.version) if manifest is None: manifest = self.get_manifest(dataset_id, version) download_version_files( - self._http, manifest, dataset_id, version, Path(dest_dir), force=force, progress_cb=progress_cb + self._http, manifest, dataset_id, version, Path(directory), force=force, progress_cb=progress_cb ) def put_release(self, from_directory: Path) -> DatasetRef: diff --git a/packages/timenet/src/timenet/registry/writable.py b/packages/timenet/src/timenet/registry/writable.py index 953ca548c..a4d434b6f 100644 --- a/packages/timenet/src/timenet/registry/writable.py +++ b/packages/timenet/src/timenet/registry/writable.py @@ -13,7 +13,6 @@ import pyarrow.fs as pafs from timenet.dataset import TimeFDataset -from timenet.errors import TimeNetDatasetNotFoundError from timenet.format.constants import MANIFEST_FILE from timenet.manifest import Manifest, PayloadSource, read_manifest from timenet.registry.base import BaseRegistry @@ -91,7 +90,7 @@ def _read_release(directory: Path) -> Manifest: return manifest def exists(self, dataset_id: str, version: str) -> bool: - """Return whether a committed version already exists in this registry. + """Check whether this registry holds a dataset version. Args: dataset_id: The dataset id. @@ -100,8 +99,4 @@ def exists(self, dataset_id: str, version: str) -> bool: Returns: ``True`` if the version has a committed manifest, else ``False``. """ - try: - self.get_manifest(dataset_id, version) - except TimeNetDatasetNotFoundError: - return False - return True + return self.find_manifest(DatasetRef.parse(f"{dataset_id}@{version}")) is not None diff --git a/packages/timenet/tests/registry/test_copy.py b/packages/timenet/tests/registry/test_copy.py new file mode 100644 index 000000000..a82858fe6 --- /dev/null +++ b/packages/timenet/tests/registry/test_copy.py @@ -0,0 +1,467 @@ +"""``copy_release()`` copies a release with its payload sources and never changes a published version. + +The ``source`` registry holds a parent from ``make_dataset()`` and a child that keeps one of its +Records, so the child reads payloads that the parent stores: the parent is a payload source of the +child. The ``chain_source`` registry adds one level: a grandchild reads a middle release, which reads +a grandparent. +""" + +from pathlib import Path +import shutil +import tempfile + +from _fake_registry import build_service_fake +import numpy as np +import pytest + +from timenet.dataset import Record, Signal, Source, TimeFDataset +from timenet.errors import ( + TimeFFormatError, + TimeNetAccessError, + TimeNetMissingParentError, + TimeNetVersionConflictError, +) +from timenet.progress import ProgressEvent, StepFinished, StepStarted +from timenet.reader import TimeFReader +from timenet.registry import LocalRegistry, RemoteRegistry, WritableRegistry, copy_release, open_local_version +from timenet.registry.shared import complete_copy +from timenet.testing import make_dataset +from timenet.types import Access, Annotation, DatasetMetadata, DatasetSchema, License, TimeSeriesSpec, Version, ureg +from timenet.types.references import DatasetRef +from timenet.writer import TimeFWriter + + +parent_ref = DatasetRef.parse("timenet/hello-world@1.0.0") +child_ref = DatasetRef.parse("test/child@1.0.0") +parent = (parent_ref.dataset_id, str(parent_ref.version)) +child = (child_ref.dataset_id, str(child_ref.version)) + +grandparent_ref = DatasetRef.parse("test/grandparent@1.0.0") +middle_ref = DatasetRef.parse("test/middle@1.0.0") +grandchild_ref = DatasetRef.parse("test/grandchild@1.0.0") +chain = [(ref.dataset_id, str(ref.version)) for ref in (grandparent_ref, middle_ref, grandchild_ref)] + +_SPEC = TimeSeriesSpec(spec_type="env", name="env", unit_value=ureg.dimensionless) +_CREDENTIALED = {"access": Access.CREDENTIALED, "access_url": "https://example.org/access"} +_OPEN = {"access": Access.OPEN, "access_url": None} + + +def _derive(loaded: TimeFDataset, dataset_id: str, records, **card) -> TimeFDataset: + """Return a release under ``dataset_id`` that keeps read Records and their ids, as an edit does. + + ``card`` sets other fields of the card of the release. + """ + return TimeFDataset.from_parts( + metadata=loaded.metadata.model_copy(update={"dataset_id": dataset_id, **card}), + records=records, + tasks=(), + schema=DatasetSchema(), + last_ids=loaded.last_ids, + ) + + +def _write(root: Path, dataset: TimeFDataset) -> Path: + dataset.derive_schema() + with TimeFWriter(root, dataset) as writer: + writer.write() + return root / dataset.metadata.dataset_id / str(dataset.metadata.dataset_version) + + +def _write_parent_and_child(root: Path, *, credentialed: DatasetRef | None = None) -> None: + """Write the parent and a child that reads payloads of the parent under ``root``. + + With ``credentialed``, the card of that release declares credentialed access, and the other card + declares open access. + """ + dataset = make_dataset() + if credentialed == parent_ref: + dataset._metadata = dataset.metadata.model_copy(update=_CREDENTIALED) + parent_dir = _write(root, dataset) + with TimeFReader(open_local_version(parent_dir)) as reader: + loaded = reader.read() + card = _CREDENTIALED if credentialed == child_ref else _OPEN + _write(root, _derive(loaded, child_ref.dataset_id, [loaded.records[0]], **card)) + + +def _release(dataset_id: str, records: list[Record]) -> TimeFDataset: + dataset = TimeFDataset( + metadata=DatasetMetadata( + dataset_id=dataset_id, + dataset_version=Version(1, 0, 0), + name=dataset_id, + description="d", + license=License.MIT, + ) + ) + for record in records: + dataset.add_record(record=record) + return dataset + + +def _record(name: str) -> Record: + """Return a named Record whose Signal values are held in memory.""" + signal = Signal.from_irregular( + np.arange(3, dtype=np.float32), time_offsets_us=[0, 1_000, 2_000], spec=_SPEC, name=name + ) + return Record(sources=(Source(name="env", signals=(signal,)),), metadata={"name": name}) + + +def _write_chain(root: Path, *, grandchild_reads_grandparent: bool) -> None: + """Write a grandparent, a middle release that reads it, and a grandchild that reads the middle release. + + The middle release keeps the grandparent's Record ``"a"`` and stores its own Record ``"b"``. The + grandchild keeps ``"b"``, so it reads only the middle release. With ``grandchild_reads_grandparent``, + it also keeps ``"a"``, so it reads the grandparent too. + """ + grandparent_dir = _write(root, _release(grandparent_ref.dataset_id, [_record("a")])) + with TimeFReader(open_local_version(grandparent_dir)) as reader: + loaded = reader.read() + middle = _derive(loaded, middle_ref.dataset_id, loaded.records) + middle.add_record(record=_record("b")) + middle_dir = _write(root, middle) + with TimeFReader(open_local_version(middle_dir)) as reader: + loaded = reader.read() + kept = [record for record in loaded.records if grandchild_reads_grandparent or record.metadata["name"] == "b"] + _write(root, _derive(loaded, grandchild_ref.dataset_id, kept)) + + +def _payload_sources(registry: WritableRegistry, reference: DatasetRef) -> list[DatasetRef]: + manifest = registry.get_manifest(reference.dataset_id, str(reference.version)) + return [pinned.release for pinned in manifest.payload_sources.values()] + + +def _other_build(dataset_id: str) -> TimeFDataset: + """Return a release of ``dataset_id`` at version 1.0.0 whose files differ from the source's copy.""" + dataset = make_dataset() + dataset._metadata = dataset.metadata.model_copy(update={"dataset_id": dataset_id}) + dataset.records[0].annotate(Annotation(key="build", value="another")) + return dataset + + +def _remote(tmp_path: Path, name: str, *, token: str | None = None) -> tuple[RemoteRegistry, Path]: + """Return a remote registry over the service fake, and the directory that the fake serves.""" + root = tmp_path / name + root.mkdir(exist_ok=True) + transport, _ = build_service_fake(root, token=token) + registry = RemoteRegistry( + f"http://{name}.local", token=token, cache_dir=tmp_path / f"{name}-cache", transport=transport + ) + return registry, root + + +@pytest.fixture +def source(tmp_path) -> LocalRegistry: + root = tmp_path / "source" + _write_parent_and_child(root) + return LocalRegistry(root) + + +@pytest.fixture +def chain_source(tmp_path) -> LocalRegistry: + root = tmp_path / "chain" + _write_chain(root, grandchild_reads_grandparent=False) + registry = LocalRegistry(root) + assert _payload_sources(registry, grandchild_ref) == [middle_ref] + assert _payload_sources(registry, middle_ref) == [grandparent_ref] + return registry + + +@pytest.fixture(params=["local", "remote"]) +def dest(request, tmp_path) -> WritableRegistry: + if request.param == "local": + return LocalRegistry(tmp_path / "dest") + registry, _ = _remote(tmp_path, "dest", token="tok_rw") + return registry + + +def test_copy_release_copies_a_release_and_its_payload_sources(source, dest): + copy_release(child_ref, source, dest) + assert dest.exists(*child) and dest.exists(*parent) + assert dest.get_manifest(*child) == source.get_manifest(*child) + assert dest.get_manifest(*parent) == source.get_manifest(*parent) + + +def test_copy_release_does_nothing_for_the_same_release(source, dest, monkeypatch): + copy_release(child_ref, source, dest) + + def unexpected(*args, **kwargs): + pytest.fail("a copy of the same release must not download or upload it again") + + monkeypatch.setattr(dest, "put_release", unexpected) + monkeypatch.setattr(source, "download_release", unexpected) + events: list[ProgressEvent] = [] + copy_release(child_ref, source, dest, progress=events.append) # no error, no second upload + assert events == [] + + +def _rewrite_child_manifest(dest: LocalRegistry, **update) -> Path: + """Rewrite the manifest of the child in ``dest`` with ``update``; return the manifest file.""" + path = dest.root / child_ref.dataset_id / str(child_ref.version) / "manifest.json" + path.write_text(dest.get_manifest(*child).model_copy(update=update).to_json()) + return path + + +def test_copy_release_refuses_a_release_with_the_same_files_and_another_card(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, source, dest) + card = dest.get_manifest(*child).metadata.model_copy(update={"description": "Another description"}) + path = _rewrite_child_manifest(dest, metadata=card) + before = path.read_bytes() + # A fresh registry: the rewrite can keep the mtime of the first write, so a cached manifest could stay. + with pytest.raises(TimeNetVersionConflictError, match="another card than") as raised: + copy_release(child_ref, source, LocalRegistry(dest.root)) + assert raised.value.difference == "card" + assert path.read_bytes() == before + + +def test_copy_release_ignores_the_build_environment(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, source, dest) + _rewrite_child_manifest(dest, build_env={"python": "0.0.0"}) + events: list[ProgressEvent] = [] + copy_release(child_ref, source, LocalRegistry(dest.root), progress=events.append) + assert events == [] + + +def test_copy_release_refuses_another_build_of_the_version(source, dest): + # the destination holds child_ref built from other records + dest.put_dataset(_other_build(child_ref.dataset_id)) + before = dest.get_manifest(*child) + with pytest.raises(TimeNetVersionConflictError, match="new version") as raised: + copy_release(child_ref, source, dest) + assert (raised.value.release, raised.value.needed_by) == (child_ref, None) + assert (raised.value.registry, raised.value.source) == (str(dest), str(source)) + assert raised.value.difference == "files" # the card differs too, but the files come first + assert dest.get_manifest(*child) == before + assert not dest.exists(*parent) # nothing is copied, not even the payload source + + +def test_copy_release_refuses_a_payload_source_with_other_files(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + # the destination holds parent_ref with other files + dest.put_dataset(_other_build(parent_ref.dataset_id)) + before = dest.get_manifest(*parent) + with pytest.raises(TimeNetVersionConflictError, match=str(parent_ref)) as raised: + copy_release(child_ref, source, dest) + assert (raised.value.release, raised.value.needed_by) == (parent_ref, child_ref) + assert not dest.exists(*child) + assert dest.get_manifest(*parent) == before + + +def _without_parent(source: LocalRegistry) -> LocalRegistry: + """Remove the parent from ``source``, and return a new registry over its directory. + + The new registry does not keep a cached manifest of the parent. + """ + shutil.rmtree(source.root / parent_ref.dataset_id) + return LocalRegistry(source.root) + + +def test_copy_release_reads_a_payload_source_that_only_dest_holds(source, dest): + copy_release(parent_ref, source, dest) + before = dest.get_manifest(*parent) + events: list[ProgressEvent] = [] + copy_release(child_ref, _without_parent(source), dest, progress=events.append) + assert dest.get_manifest(*child) == source.get_manifest(*child) + assert dest.get_manifest(*parent) == before + assert events == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")] + + +def test_copy_release_refuses_a_payload_source_that_only_dest_holds_with_other_files(source, dest): + # the destination holds parent_ref with other payload files than the child pins + dest.put_dataset(_release(parent_ref.dataset_id, [_record("other")])) + before = dest.get_manifest(*parent) + with pytest.raises(TimeFFormatError, match=f"payload source {parent_ref} differs"): + copy_release(child_ref, _without_parent(source), dest) + assert not dest.exists(*child) + assert dest.get_manifest(*parent) == before + + +def test_copy_release_checks_the_access_of_a_payload_source_that_only_dest_holds(tmp_path): + root = tmp_path / "source" + _write_parent_and_child(root, credentialed=parent_ref) + dest, dest_root = _remote(tmp_path, "dest", token="tok_rw") + # the service holds the parent, whose card declares credentialed access + shutil.copytree(root / parent_ref.dataset_id, dest_root / parent_ref.dataset_id) + with pytest.raises(TimeNetAccessError) as raised: + copy_release(child_ref, _without_parent(LocalRegistry(root)), dest) + assert raised.value.dataset == parent_ref + assert not dest.exists(*child) + + +def test_copy_release_refuses_a_payload_source_that_no_registry_holds(source, dest): + with pytest.raises(TimeNetMissingParentError) as raised: + copy_release(child_ref, _without_parent(source), dest) + assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) + assert not dest.exists(*child) + + +def test_copy_release_refuses_a_payload_source_that_only_dest_holds_damaged(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(parent_ref, source, dest) + parent_dir = dest.root / parent_ref.dataset_id / str(parent_ref.version) + (parent_dir / dest.get_manifest(*parent).files.control[0].path).unlink() + with pytest.raises(TimeNetMissingParentError) as raised: + copy_release(child_ref, _without_parent(source), dest) + assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) + assert not dest.exists(*child) + + +def test_copy_release_repairs_a_damaged_local_copy(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, source, dest) + child_dir = dest.root / child_ref.dataset_id / str(child_ref.version) + manifest = dest.get_manifest(*child) + (child_dir / manifest.files.control[0].path).unlink() + assert not complete_copy(child_dir, manifest) + events: list[ProgressEvent] = [] + copy_release(child_ref, source, dest, progress=events.append) + assert complete_copy(child_dir, manifest) + assert events == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")] + + +def test_copy_release_reports_each_copied_release(source, tmp_path): + events: list[ProgressEvent] = [] + copy_release(child_ref, source, LocalRegistry(tmp_path / "dest"), progress=events.append) + assert [(e.action, e.release) for e in events if isinstance(e, StepStarted)] == [ + ("copy", parent_ref), + ("copy", child_ref), + ] + assert events == [ + StepStarted(parent_ref, "copy"), + StepFinished(parent_ref, "copy"), + StepStarted(child_ref, "copy"), + StepFinished(child_ref, "copy"), + ] + + +def test_copy_release_copies_only_the_payload_sources_that_the_release_lists(chain_source, dest): + # The grandchild reads the middle release only. The source lacks the grandparent, which the middle reads. + shutil.rmtree(chain_source.root / grandparent_ref.dataset_id) + events: list[ProgressEvent] = [] + copy_release(grandchild_ref, LocalRegistry(chain_source.root), dest, progress=events.append) + assert [e.release for e in events if isinstance(e, StepStarted)] == [middle_ref, grandchild_ref] + assert not dest.exists(*chain[0]) + + +def test_copy_release_without_payload_sources_copies_the_release_alone(chain_source, dest): + copy_release(middle_ref, chain_source, dest, with_payload_sources=False) + assert dest.get_manifest(*chain[1]) == chain_source.get_manifest(*chain[1]) + assert not dest.exists(*chain[0]) + + +def test_copy_release_copies_a_shared_payload_source_once(tmp_path): + root = tmp_path / "chain" + _write_chain(root, grandchild_reads_grandparent=True) + shared_source = LocalRegistry(root) + assert set(_payload_sources(shared_source, grandchild_ref)) == {grandparent_ref, middle_ref} + events: list[ProgressEvent] = [] + copy_release(grandchild_ref, shared_source, LocalRegistry(tmp_path / "dest"), progress=events.append) + assert [e.release for e in events if isinstance(e, StepStarted)] == [grandparent_ref, middle_ref, grandchild_ref] + + +def test_copy_release_between_remote_registries_removes_its_download(tmp_path, monkeypatch): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + downloads = tmp_path / "tmp" + downloads.mkdir() + monkeypatch.setattr(tempfile, "tempdir", str(downloads)) + copy_release(child_ref, remote_source, dest) + assert dest.get_manifest(*child) == remote_source.get_manifest(*child) + assert dest.get_manifest(*parent) == remote_source.get_manifest(*parent) + assert list(downloads.iterdir()) == [] + + +def test_copy_release_between_remote_registries_downloads_only_the_releases_that_dest_lacks(tmp_path, monkeypatch): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + copy_release(parent_ref, remote_source, dest) + downloads: list[DatasetRef] = [] + download = remote_source.download_release + + def counting(reference, *args, **kwargs): + downloads.append(reference) + return download(reference, *args, **kwargs) + + monkeypatch.setattr(remote_source, "download_release", counting) + copy_release(child_ref, remote_source, dest) + assert downloads == [child_ref] # dest already holds the parent + assert dest.get_manifest(*child) == remote_source.get_manifest(*child) + + +def test_copy_release_removes_its_download_after_an_error(tmp_path, monkeypatch): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + downloads = tmp_path / "tmp" + downloads.mkdir() + monkeypatch.setattr(tempfile, "tempdir", str(downloads)) + + def refuse(directory): + raise RuntimeError("upload failed") + + monkeypatch.setattr(dest, "put_release", refuse) + with pytest.raises(RuntimeError, match="upload failed"): + copy_release(child_ref, remote_source, dest) + assert list(downloads.iterdir()) == [] + + +@pytest.mark.parametrize("remote", [False, True], ids=["local", "remote"]) +def test_a_forced_download_keeps_the_old_copy_if_the_final_rename_fails(remote, tmp_path, monkeypatch): + if remote: + registry, root = _remote(tmp_path, "source") + else: + root = tmp_path / "source" + registry = LocalRegistry(root) + _write_parent_and_child(root) + manifest = registry.get_manifest(*parent) + directory = tmp_path / "copy" / parent_ref.dataset_id / str(parent_ref.version) + registry.download_release(parent_ref, directory, manifest=manifest) + rename = Path.replace + + def replace(self, target): + if ".tmp-" in self.name: + raise OSError("rename failed") + return rename(self, target) + + monkeypatch.setattr(Path, "replace", replace) + with pytest.raises(OSError, match="rename failed"): + registry.download_release(parent_ref, directory, force=True, manifest=manifest) + assert complete_copy(directory, manifest) + + +@pytest.mark.parametrize("credentialed", [child_ref, parent_ref], ids=["release", "payload-source"]) +def test_copy_release_refuses_a_release_that_is_not_open_for_a_registry_that_is_not_local(credentialed, tmp_path): + root = tmp_path / "source" + _write_parent_and_child(root, credentialed=credentialed) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + with pytest.raises(TimeNetAccessError) as raised: + copy_release(child_ref, LocalRegistry(root), dest) + assert raised.value.dataset == credentialed + assert not dest.exists(*child) + assert not dest.exists(*parent) + + +@pytest.mark.parametrize("credentialed", [child_ref, parent_ref], ids=["release", "payload-source"]) +def test_copy_release_copies_a_release_that_is_not_open_between_local_registries(credentialed, tmp_path): + root = tmp_path / "source" + _write_parent_and_child(root, credentialed=credentialed) + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, LocalRegistry(root), dest) + assert dest.get_manifest(*child).metadata.access is ( + Access.CREDENTIALED if credentialed == child_ref else Access.OPEN + ) + assert dest.exists(*parent) + + +def test_copy_release_refuses_to_fetch_a_release_that_is_not_open_from_a_registry_that_is_not_local(tmp_path): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root, credentialed=child_ref) + dest = LocalRegistry(tmp_path / "dest") + with pytest.raises(TimeNetAccessError) as raised: + copy_release(child_ref, remote_source, dest) + assert raised.value.dataset == child_ref + assert not dest.exists(*child) + assert not dest.exists(*parent) # the open parent is not copied either: every check runs first diff --git a/packages/timenet/tests/registry/test_registry.py b/packages/timenet/tests/registry/test_registry.py index 00cf716f8..01e1438f4 100644 --- a/packages/timenet/tests/registry/test_registry.py +++ b/packages/timenet/tests/registry/test_registry.py @@ -152,6 +152,27 @@ def test_get_manifest_names_the_path_of_a_cut_manifest(registry_root): LocalRegistry(registry_root).get_manifest("timenet/hello-world", "1.0.0") +def test_find_manifest_answers_none_for_an_absent_release_and_refuses_a_misplaced_one(registry_root): + registry = LocalRegistry(registry_root) + assert registry.find_manifest(DatasetRef.parse("timenet/hello-world@9.0.0")) is None + misplaced = registry_root / "timenet/hello-world/2.0.0" + misplaced.mkdir() + (misplaced / "manifest.json").write_bytes((registry_root / "timenet/hello-world/1.0.0/manifest.json").read_bytes()) + with pytest.raises(TimeFFormatError, match="declares a different dataset ID or version"): + registry.find_manifest(DatasetRef.parse("timenet/hello-world@2.0.0")) + + +def test_damage_names_the_missing_file_of_a_stored_release(registry_root): + registry = LocalRegistry(registry_root) + release = DatasetRef.parse("timenet/hello-world@1.0.0") + assert registry.damage(release) is None + part = registry.get_manifest(release.dataset_id, "1.0.0").files.control[0] + (registry_root / "timenet/hello-world/1.0.0" / part.path).unlink() + damage = registry.damage(release) + assert damage is not None + assert (damage.release, damage.path) == (release, part.path) + + def test_latest_version_ignores_staging_dirs(registry_root): # A crashed build can leave a `.tmp-` sibling (briefly holding a manifest.json). stale = registry_root / "demo/ecg" / "2.0.0.tmp-deadbeef" diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index 3ea6740b2..0ad9fc02f 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -636,6 +636,8 @@ def _hosted(client, metadata, lineage=()): """Serve one manifest from a registry that is not local, so the build-your-own gate applies.""" class _Hosted: + requires_open_access = True + def get_manifest(self, dataset_id, version=None): return SimpleNamespace(metadata=metadata, lineage=lineage) diff --git a/packages/timenet/tests/test_engine.py b/packages/timenet/tests/test_engine.py index 2fb1f96bd..9efecaa79 100644 --- a/packages/timenet/tests/test_engine.py +++ b/packages/timenet/tests/test_engine.py @@ -106,6 +106,8 @@ class _ZarrConnector(_DemoConnector): def test_publish_pipeline_passes_the_requested_values_backend_to_the_registry(tmp_path): class _Registry: + requires_open_access = True + def __init__(self) -> None: self.values_backend: str | None = None @@ -484,6 +486,8 @@ def test_publish_pipeline_refuses_a_card_that_is_not_open_before_downloading(tmp class _HostedRegistry: """A registry that is not local. It serves the parents from a local registry and refuses writes.""" + requires_open_access = True + def __init__(self, root: Path) -> None: self.local = LocalRegistry(root) diff --git a/packages/timenet/tests/test_errors.py b/packages/timenet/tests/test_errors.py index 1df76869e..075de37d1 100644 --- a/packages/timenet/tests/test_errors.py +++ b/packages/timenet/tests/test_errors.py @@ -15,6 +15,7 @@ TimeNetFileError, TimeNetMissingParentError, TimeNetRegistryError, + TimeNetVersionConflictError, ) from timenet.types import Access from timenet.types.access import AccessRequirement @@ -66,12 +67,12 @@ def test_catchable_as_base(): ("error", "facts", "message"), [ ( - TimeNetMissingParentError(_A, _B, True, "here"), + TimeNetMissingParentError(_A, _B, "here", payload_source=True), {"missing": _A, "needed_by": _B, "payload_source": True}, "payload source org/a@1.0.0 of org/b@2.0.0 is not in here.", ), ( - TimeNetMissingParentError(_A, _B, False, "here"), + TimeNetMissingParentError(_A, _B, "here", payload_source=False), {"payload_source": False}, "parent release org/a@1.0.0 of org/b@2.0.0 is not in here.", ), @@ -113,6 +114,24 @@ def test_catchable_as_base(): {"requirements": ()}, "org/a@1.0.0 is credentialed.", ), + ( + TimeNetVersionConflictError(_A, "registry here", "registry there"), + { + "release": _A, + "registry": "registry here", + "source": "registry there", + "needed_by": None, + "difference": "files", + }, + "registry here already holds org/a@1.0.0 with other files than registry there. " + "A published version cannot change, so a changed release needs a new version.", + ), + ( + TimeNetVersionConflictError(_A, "registry here", "registry there", _B, "card"), + {"release": _A, "needed_by": _B, "difference": "card"}, + "registry here already holds org/a@1.0.0, a payload source of org/b@2.0.0, with another card than " + "registry there. A published version cannot change, so a changed release needs a new version.", + ), ], ) def test_an_error_with_facts_ends_with_the_hint_of_its_front_end(error, facts, message): diff --git a/packages/timenet/tests/test_progress.py b/packages/timenet/tests/test_progress.py new file mode 100644 index 000000000..2b89ae30d --- /dev/null +++ b/packages/timenet/tests/test_progress.py @@ -0,0 +1,12 @@ +from dataclasses import FrozenInstanceError + +import pytest + +from timenet.progress import StepStarted +from timenet.types.references import DatasetRef + + +def test_progress_events_are_frozen(): + event = StepStarted(DatasetRef.parse("a/b@1.0.0"), "copy") + with pytest.raises(FrozenInstanceError): + event.action = "build" # ty: ignore[invalid-assignment] diff --git a/zensical.toml b/zensical.toml index 283c61573..a10b4762d 100644 --- a/zensical.toml +++ b/zensical.toml @@ -22,7 +22,7 @@ nav = [ { "Datasets" = [{ "Use datasets" = [{ "The client (SDK)" = "client.md" }, { "Registries" = "registry.md" }, { "Load into your stack" = "usage.md" }] }, { "Build datasets" = [{ "Connectors" = "connectors.md" }, { "Build & publish" = "build.md" }] }, { "Catalog" = [{ "Datasets" = "catalog/datasets.md" }, { "Benchmarks" = "catalog/benchmarks.md" }, { "Licensing" = "catalog/licensing.md" }] }] }, { "CLI" = [{ "Overview" = "cli/index.md" }, { "timenet (consumer)" = "cli/timenet.md" }, { "timenet-build (producer)" = "cli/build.md" }] }, # >>> api-nav: generated by scripts/gen_api_docs.py — do not edit >>> - { "API Reference" = [{ "Overview" = "api/index.md" }, { "timenet.builders" = "api/builders.md" }, { "timenet.client" = "api/client.md" }, { "timenet.config" = "api/config.md" }, { "timenet.connectors" = "api/connectors.md" }, { "timenet.dataset" = "api/dataset.md" }, { "timenet.engine" = "api/engine.md" }, { "timenet.errors" = "api/errors.md" }, { "timenet.format" = "api/format.md" }, { "timenet.json" = "api/json.md" }, { "timenet.manifest" = "api/manifest.md" }, { "timenet.provenance" = "api/provenance.md" }, { "timenet.reader" = "api/reader.md" }, { "timenet.registry" = "api/registry.md" }, { "timenet.storage" = "api/storage.md" }, { "timenet.types" = "api/types.md" }, { "timenet.values_backends" = "api/values_backends.md" }, { "timenet.writer" = "api/writer.md" }, { "timenet_connectors.api" = "api/api.md" }, { "timenet_connectors.bases" = "api/bases.md" }, { "timenet_connectors.discovery" = "api/discovery.md" }, { "timenet_connectors.download" = "api/download.md" }, { "timenet_connectors.sources" = "api/sources.md" }, { "timenet_connectors.splitting" = "api/splitting.md" }, { "timenet_connectors.time_axes" = "api/time_axes.md" }] }, + { "API Reference" = [{ "Overview" = "api/index.md" }, { "timenet.builders" = "api/builders.md" }, { "timenet.client" = "api/client.md" }, { "timenet.config" = "api/config.md" }, { "timenet.connectors" = "api/connectors.md" }, { "timenet.dataset" = "api/dataset.md" }, { "timenet.engine" = "api/engine.md" }, { "timenet.errors" = "api/errors.md" }, { "timenet.format" = "api/format.md" }, { "timenet.json" = "api/json.md" }, { "timenet.manifest" = "api/manifest.md" }, { "timenet.progress" = "api/progress.md" }, { "timenet.provenance" = "api/provenance.md" }, { "timenet.reader" = "api/reader.md" }, { "timenet.registry" = "api/registry.md" }, { "timenet.storage" = "api/storage.md" }, { "timenet.types" = "api/types.md" }, { "timenet.values_backends" = "api/values_backends.md" }, { "timenet.writer" = "api/writer.md" }, { "timenet_connectors.api" = "api/api.md" }, { "timenet_connectors.bases" = "api/bases.md" }, { "timenet_connectors.discovery" = "api/discovery.md" }, { "timenet_connectors.download" = "api/download.md" }, { "timenet_connectors.sources" = "api/sources.md" }, { "timenet_connectors.splitting" = "api/splitting.md" }, { "timenet_connectors.time_axes" = "api/time_axes.md" }] }, # <<< api-nav <<< ]