From 5320cb102f6eda4fa98d40eddd55ef3e6446077e Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 14:06:39 +0100 Subject: [PATCH 1/4] feat(coordinator)!: plan and run reads through the list of sources A read now has its own coordinator, in timenet.coordinator.read. plan_read() returns the plan of a read without running it. read_release() runs the plan and opens the release with its payload sources. The client uses read_release() only for a remote registry so far. A read opens a release where it is. The first source that provides the release gives a ReadStep, and the read copies nothing. Each payload source resolves on its own through the same list, at its exact version. So a release and its payload sources can come from different registries. The pins of the release must match each payload source. The read and the build share one planner. A read is a fourth outcome of the walk of the build planner, next to no step, a copy and a build. The walk takes needed_in_local_registry: a build needs a complete copy in the store, and a read needs only a source that serves the release. If only InstalledConnectors() provides a release, the walk plans its build into the store, with the whole list as the sources of its parents. If a build also needs a release that the read found in a registry, the release goes into the store, and the read opens it there. A read that needs a build without a store raises TimeFValidationError. Before any step runs, the plan checks the pins of each payload source. Then it checks the size of each file of each release that it reads from a local registry. A copy that a build needs can still repair a damaged copy that the read found first. A damaged copy in the store also fails a read that no registry serves. So the read never falls through to a later source and never builds over the damage. The plan and the builds that it plans share one set of answers and manifests. read_release() opens each release with the manifest that the plan got. So a remote read fetches each manifest once. The walk takes the answer for the requested release, so a read of the latest version does not ask again for the exact version. BaseRegistry gets open_release(), which opens one exact release without its payload sources. It takes the manifest if the caller has it, and it may use that manifest. A local registry checks the size of each file. A remote registry uses a complete cached copy, else the remote files. open_cached_or_remote_release() holds that one-release open. Step is now ReadStep | CopyStep | BuildStep, and run_plan() runs nothing for a read step. The planner shares Lookups, plan_steps() and describe_searched_sources() with the read coordinator. BaseRegistry.open_version() goes away, with its implementations and open_cached_or_remote(). open_release() and download_release() handle one release without its payload sources, and copy_release() copies a release with them. A read of a release with its payload sources goes through the read coordinator. A read can need one release twice. For example, a child reads P1 and P2, and the card of P1 declares P2. The release then has one step, and its build must match the pins of each reader. If two readers pin one file of the release as different files, no build can match both, and the plan raises TimeFFormatError. A cycle through the requested release names that release first. The plan builds a release with the InstalledBuilder that its lookup read, so it reads the card of the connector once. The error for a missing payload source names the sources in the same way as the error of the plan, with the version that the installed connector builds. BREAKING CHANGE: BaseRegistry has a new abstract method, open_release(). A registry of another package must implement it. The abstract LazyReadRegistry._open_remote_version() takes an optional manifest. BaseRegistry.open_version() is gone from every registry. To read a release with its payload sources from one registry, use TimeNet(sources=[registry]).open(). Part of TNET-157. --- .../chengsenwang/tsqa/tests/test_connector.py | 4 +- .../tsaqa_benchmark/tests/test_connector.py | 4 +- .../tests/test_build_plan.py | 8 +- packages/timenet/src/timenet/client.py | 3 +- .../timenet/src/timenet/coordinator/build.py | 7 +- .../timenet/src/timenet/coordinator/plan.py | 432 +++++++++---- .../timenet/src/timenet/coordinator/read.py | 120 ++++ packages/timenet/src/timenet/registry/base.py | 31 +- .../src/timenet/registry/dependencies.py | 46 +- .../timenet/src/timenet/registry/lazy_read.py | 35 +- .../timenet/src/timenet/registry/local.py | 35 +- .../src/timenet/registry/remote/registry.py | 11 +- packages/timenet/src/timenet/registry/s3.py | 17 +- packages/timenet/tests/_demo_child.py | 32 + .../tests/{registry => }/_fake_registry.py | 0 packages/timenet/tests/conftest.py | 6 + .../timenet/tests/coordinator/test_build.py | 8 +- .../timenet/tests/coordinator/test_read.py | 578 ++++++++++++++++++ .../tests/reader/test_shared_storage.py | 4 +- packages/timenet/tests/registry/conftest.py | 10 +- .../tests/registry/test_conformance.py | 34 +- .../timenet/tests/registry/test_registry.py | 37 +- .../timenet/tests/registry/test_remote.py | 4 +- packages/timenet/tests/registry/test_s3.py | 10 +- .../timenet/tests/registry/test_version.py | 11 +- packages/timenet/tests/test_client.py | 25 +- packages/timenet/tests/test_engine.py | 7 +- 27 files changed, 1201 insertions(+), 318 deletions(-) create mode 100644 packages/timenet/src/timenet/coordinator/read.py create mode 100644 packages/timenet/tests/_demo_child.py rename packages/timenet/tests/{registry => }/_fake_registry.py (100%) create mode 100644 packages/timenet/tests/coordinator/test_read.py diff --git a/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py b/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py index 53e01c97e..2ccb73c89 100644 --- a/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py +++ b/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py @@ -8,7 +8,7 @@ from timenet.dataset import TimeFDataset from timenet.errors import TimeFValidationError from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry +from timenet.registry import LocalRegistry, open_local_version from timenet.types import AnswerTask from timenet.types.splits import Split from timenet.types.text import read_value @@ -188,7 +188,7 @@ def test_splits_survive_storage_round_trip(tmp_path): original = TSQAConnector().convert(_refs(tmp_path, _balanced_rows())) registry = LocalRegistry(tmp_path / "registry") registry.put_dataset(original) - with TimeFReader(registry.open_version("chengsenwang/tsqa", "1.2.0")) as reader: + with TimeFReader(open_local_version(registry.root / "chengsenwang/tsqa/1.2.0")) as reader: restored = reader.read() assert {read_value(task.prompt): task.split for task in restored.tasks} == { read_value(task.prompt): task.split for task in original.tasks diff --git a/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py b/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py index 93f4c09ba..f97e78213 100644 --- a/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py +++ b/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py @@ -13,7 +13,7 @@ from timenet.errors import TimeFFormatError, TimeFValidationError from timenet.json import JsonValue from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry +from timenet.registry import LocalRegistry, open_local_version from timenet.types import AnswerTask, ClassificationTask, Split from timenet_connectors.bases.huggingface import HubFileRef from timenet_connectors.datasets.tsaqa.tsaqa_benchmark import connector @@ -127,7 +127,7 @@ def test_storage_round_trip_keeps_the_whole_conversion(tmp_path): registry = LocalRegistry(tmp_path / "registry") registry.put_dataset(original) published = _published() - with TimeFReader(registry.open_version("tsaqa/tsaqa-benchmark", "1.0.0")) as reader: + with TimeFReader(open_local_version(registry.root / "tsaqa/tsaqa-benchmark/1.0.0")) as reader: restored = reader.read() assert {a.key: a.value for a in restored.annotations if a.key not in {"task", "question_type"}} == { "source_repository": "TSAQA/TSAQA-Benchmark", diff --git a/packages/timenet-connectors/tests/test_build_plan.py b/packages/timenet-connectors/tests/test_build_plan.py index e61241255..531d73db3 100644 --- a/packages/timenet-connectors/tests/test_build_plan.py +++ b/packages/timenet-connectors/tests/test_build_plan.py @@ -20,7 +20,7 @@ TimeNetMissingParentError, ) from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry, WritableRegistry +from timenet.registry import LocalRegistry, WritableRegistry, open_local_version from timenet.registry._paths import release_dir from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset @@ -176,7 +176,7 @@ def _child_of(registry: LocalRegistry, reference: DatasetRef, parent: DatasetRef update={"dataset_id": reference.dataset_id, "parents": {ParentAlias("parent"): parent}} ) child = TimeFDataset(metadata=metadata) - with TimeFReader(registry.open_version(parent.dataset_id, str(parent.version))) as reader: + with TimeFReader(open_local_version(release_dir(registry.root, parent))) as reader: child.import_from(reader, records=[record_id]) own = Signal.from_values( np.arange(16, dtype=np.float32), spec=_SPEC, name="own", time_axis=RegularAxis.from_rate_hz(16) @@ -435,8 +435,8 @@ def get_manifest(self, dataset_id, version=None): def open_file(self, dataset_id, version, relpath): return self._local.open_file(dataset_id, version, relpath) - def open_version(self, dataset_id, version=None): - return self._local.open_version(dataset_id, version) + def open_release(self, reference, **options): + return self._local.open_release(reference, **options) def put_dataset(self, dataset, **options): return self._local.put_dataset(dataset, **options) diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index cf2dca4d2..da7ece6d4 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -17,6 +17,7 @@ from timenet.builders import installed_builder from timenet.config import settings from timenet.coordinator.build import plan_release_build, run_plan +from timenet.coordinator.read import read_release from timenet.dataset import TimeFDataset from timenet.errors import ( TimeFValidationError, @@ -256,7 +257,7 @@ def open( "fetch_from copies releases into a local registry; a remote client reads from its own registry" ) self._reject_unhosted_access(dataset_id, version) - return TimeFReader(registry.open_version(dataset_id, version)) + return TimeFReader(read_release(dataset_id, version, sources=[registry], store=None)) source = open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from try: return TimeFReader(_prepare(registry, dataset_id, version, auto_build=auto_build, fetch_from=source)) diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py index 9aa22fdd5..139c9d1fc 100644 --- a/packages/timenet/src/timenet/coordinator/build.py +++ b/packages/timenet/src/timenet/coordinator/build.py @@ -4,7 +4,7 @@ from typing import assert_never from timenet.builders import BuildOptions, declared -from timenet.coordinator.plan import BuildStep, CopyStep, Plan, plan_steps +from timenet.coordinator.plan import BuildStep, CopyStep, Plan, ReadStep, plan_steps from timenet.manifest import PayloadSource from timenet.progress import StepFinished, StepStarted from timenet.registry import copy_release @@ -60,6 +60,8 @@ def plan_release_build( # ruff: ignore[too-many-arguments] def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = None) -> None: """Run a plan's copy and build steps in order. + A read step runs nothing, because its release stays where it is. + Args: plan: The steps to run in dependency order. store: The local registry to copy and build into. @@ -73,6 +75,9 @@ def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = options = options or BuildOptions() for step in plan.steps: match step: + case ReadStep(): + # A read opens the release where it is, so it has nothing to run. + pass case CopyStep(): copy_release( step.release, diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index d86ac72f2..e9f0d7bf7 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -1,7 +1,8 @@ """Plan release copies and builds with dependencies first, before any connector runs.""" -from collections.abc import Mapping, Sequence +from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field, replace +from functools import cached_property from graphlib import CycleError, TopologicalSorter from typing import ClassVar, Literal, TypeAlias @@ -22,7 +23,9 @@ validate_manifest_reference, ) from timenet.registry.local import LocalRegistry +from timenet.registry.version import DatasetVersion from timenet.sources import ( + Found, FoundInConnector, FoundInRegistry, InstalledConnectors, @@ -37,6 +40,19 @@ from timenet.types.version import Version +@dataclass(frozen=True) +class ReadStep: + """A planned read of one release in the registry that holds it.""" + + release: DatasetRef + registry: BaseRegistry + """The registry to read from.""" + depends_on: tuple[DatasetRef, ...] = () + """The releases with steps that must finish before this step runs.""" + action: ClassVar[Literal["read"]] = "read" + """The kind of step, for display.""" + + @dataclass(frozen=True) class CopyStep: """A planned copy of one release from a registry into the local registry.""" @@ -72,13 +88,13 @@ class BuildStep: """The kind of step, for display.""" -Step: TypeAlias = CopyStep | BuildStep -"""A planned copy or build of one release.""" +Step: TypeAlias = ReadStep | CopyStep | BuildStep +"""A planned read, copy, or build of one release.""" @dataclass(frozen=True) class Plan: - """The copy and build steps for a dataset, in execution order.""" + """The steps of a build or read in execution order.""" steps: tuple[Step, ...] @@ -94,70 +110,226 @@ class Plan: @dataclass(frozen=True) class _Release: - """A required release with its role, planned step and dependencies.""" + """A required release with its role, its need, its planned step and its dependencies.""" role: Role """The release's role as a dataset or a source of payload files.""" + needed_in_local_registry: bool + """True if the plan needs a complete copy in the output, False if a source that serves the release is enough.""" step: Step | None - """The copy or build step before dependencies are set, or ``None`` for an existing release.""" + """The step before dependencies are set, or ``None`` for a release that the output holds.""" dependencies: tuple[DatasetRef, ...] """The parent or payload source releases that must be ready first.""" +@dataclass +class Lookups: + """A cache of source answers and release manifests for one plan. + + The cache stays valid only while no step runs, so use one instance for one plan only. + """ + + answers: dict[tuple[int, DatasetRef], ReleaseLookup] = field(default_factory=dict) + """Source answers keyed by source identity and release. + + Keep each source alive while its answers are cached. + """ + manifests: dict[tuple[int, DatasetRef], Manifest | None] = field(default_factory=dict) + """Cached manifests keyed by registry identity and release, with ``None`` for missing releases.""" + + def resolve(self, sources: Sequence[ReleaseSource], dataset_id: str, version: str | None) -> ReleaseLookup: + """Resolve a release from ordered sources and cache answers for exact versions. + + Args: + sources: The sources to search, in order. + dataset_id: The dataset ID. + version: An exact version, ``"latest"``, or ``None``, as in :func:`~timenet.sources.resolve_release`. + + Returns: + The selected ``Found`` answer, otherwise the first ``NotServed`` answer or ``NotFound()``. + """ + return resolve_release(sources, dataset_id, version, ask=self._ask) + + def ask(self, source: ReleaseSource, reference: DatasetRef) -> ReleaseLookup: + """Return a cached answer for an exact release, or ask the source and cache it. + + Args: + source: The source to query. + reference: The exact release. + + Returns: + The source's answer for the release. + """ + key = (id(source), reference) + if key not in self.answers: + self.answers[key] = source.resolve(reference.dataset_id, reference.version) + return self.answers[key] + + def manifest(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: + """Return a cached release manifest, or read and cache it. + + A manifest in a cached source answer takes precedence over a registry read. + + Returns: + The manifest, or ``None`` if the release is absent from the registry. + """ + key = (id(registry), reference) + if key not in self.manifests: + answer = self.answers.get(key) + if isinstance(answer, FoundInRegistry | NotServed) and answer.manifest is not None: + validate_manifest_reference(answer.manifest, reference) + self.manifests[key] = answer.manifest + else: + self.manifests[key] = registry.find_manifest(reference) + return self.manifests[key] + + def _ask(self, source: ReleaseSource, dataset_id: str, version: Version | None) -> ReleaseLookup: + """Return the answer of a source for a release, from the cache for an exact version.""" + if version is not None: + return self.ask(source, DatasetRef(dataset_id=dataset_id, version=version)) + answer = source.resolve(dataset_id, None) + if isinstance(answer, Found): + # A later request for the exact version gets the same answer. + self.answers.setdefault((id(source), DatasetRef(dataset_id=dataset_id, version=answer.version)), answer) + return answer + + @dataclass class _Graph: - """A graph of required releases and their copy or build steps.""" + """A graph of required releases and their read, copy or build steps.""" - output: LocalRegistry + output: LocalRegistry | None + """The local registry for copies and builds, or ``None`` for a read that copies and builds nothing.""" sources: Sequence[ReleaseSource] - """Sources for missing releases in preference order, excluding the output.""" + """The sources in preference order. A read reads the output like any other source.""" open_only: bool """True if each release that the walk visits must declare ``access: open``.""" releases: dict[DatasetRef, _Release] = field(default_factory=dict) """The nodes, in the order in which the walk found them.""" pins: dict[DatasetRef, dict[DatasetRef, PayloadSource]] = field(default_factory=dict) """For each payload source: each release that reads its files, with its pins on them.""" - answers: dict[tuple[int, DatasetRef], ReleaseLookup] = field(default_factory=dict) - """Cached source answers by source identity and release.""" - manifests: dict[tuple[int, DatasetRef], Manifest | None] = field(default_factory=dict) - """Cached manifests keyed by registry identity and release, with ``None`` for missing releases.""" + lookups: Lookups = field(default_factory=Lookups) + """The cache of source answers and release manifests.""" damage: dict[DatasetRef, TimeNetFileError | None] = field(default_factory=dict) """The first file error per checked local release, or ``None`` if all file sizes match.""" - def visit(self, reference: DatasetRef, needed_by: DatasetRef, pinned: PayloadSource | None = None) -> None: + @cached_property + def build_sources(self) -> list[ReleaseSource]: + """The sources without the output, which is never a source of its own copies and builds.""" + output = self.output + return [source for source in self.sources if output is None or not _is_same_registry(source, output)] + + def visit( + self, + reference: DatasetRef, + needed_by: DatasetRef | None, + pinned: PayloadSource | None = None, + *, + needed_in_local_registry: bool, + answer: ReleaseLookup | None = None, + ) -> None: """Add a release and its dependencies to the graph. - The graph revisits an existing entry only when its role changes from payload source to dataset. + A node is planned again when a later visit needs more of it: a dataset instead of a payload source, or a + copy in the output instead of a read. + + Args: + reference: The exact release to add. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + pinned: File requirements from ``needed_by``, or ``None`` for a parent dataset. + needed_in_local_registry: True if the plan needs a complete copy in the output. + answer: The answer of the sources for ``reference``, or ``None`` to ask them. + """ + role: Role = "dataset" if pinned is None else "payload-source" + if pinned is not None and needed_by is not None: + self.pins.setdefault(reference, {})[needed_by] = pinned + found = self.releases.get(reference) + if found is not None: + role = "dataset" if "dataset" in {found.role, role} else "payload-source" + needed_in_local_registry = found.needed_in_local_registry or needed_in_local_registry + if (role, needed_in_local_registry) == (found.role, found.needed_in_local_registry): + return + if needed_in_local_registry: + self.local(reference, needed_by, role) + else: + self.read(reference, needed_by, role, answer) + + def read( + self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role, answer: ReleaseLookup | None + ) -> None: + """Plan a release that a source can serve, and plan its payload sources if it is a dataset. + + Args: + reference: The exact release. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + role: Whether the release is a dataset or a payload source. + answer: The answer of the sources for ``reference``, or ``None`` to ask them. + + Raises: + TimeNetFileError: If a source other than a registry provides the release and the output holds a + damaged copy. + TimeFValidationError: If only installed connectors provide the release and the plan has no output. + TimeNetDatasetNotFoundError: If the registry of the answer no longer holds the release. + TimeNetAccessError: If a source lists the release but no source serves it. + TimeNetMissingParentError: If no source provides the payload source. + """ # ruff: ignore[docstring-extraneous-exception] - the file, access and missing errors are raised as values + if answer is None: + answer = self.lookups.resolve(self.sources, reference.dataset_id, str(reference.version)) + if isinstance(answer, FoundInRegistry): + manifest = self.lookups.manifest(answer.registry, reference) + if manifest is None: + raise TimeNetDatasetNotFoundError(f"{answer.registry} no longer holds {reference}") + # The check of the local files waits for the end of the walk, because a later copy can repair them. + sources = list(manifest.payload_sources.values()) if role == "dataset" else [] + step = ReadStep(reference, answer.registry) + self.releases[reference] = _Release(role, False, step, tuple(source.release for source in sources)) + for source in sources: + self.visit(source.release, reference, source, needed_in_local_registry=False) + return + # Nothing but a registry replaces a damaged copy in the output. + damage = self.in_output(reference)[1] + if damage is not None: + raise damage + if isinstance(answer, FoundInConnector): + if self.output is None: + raise TimeFValidationError( + f"{reference} needs a build by {InstalledConnectors()}, but the read has no local registry to " + "build into" + ) + self.local(reference, needed_by, role) + elif isinstance(answer, NotServed): + raise answer.access_error(reference.dataset_id) + else: + raise self.missing(reference, needed_by, role, self.sources) + + def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role) -> None: + """Plan a complete copy of a release in the output, from the output itself, a copy or a build. Args: - reference: The release to add. - needed_by: The release that needs ``reference``. - pinned: File pins from ``needed_by``, or ``None`` for a parent dataset. + reference: The exact release. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + role: Whether the release is a dataset or a payload source. Raises: + TimeFValidationError: If the plan has no output. TimeNetFileError: If the output holds a damaged copy that no other registry provides. TimeNetAccessError: If a source lists the release but no source serves it. TimeNetMissingParentError: If no source provides the release. - """ # ruff: ignore[docstring-extraneous-exception] - the file and access errors are raised as values - role: Role = "dataset" if pinned is None else "payload-source" - if pinned is not None: - self.pins.setdefault(reference, {})[needed_by] = pinned - found = self.releases.get(reference) - # If the walk found a payload source first, and later finds it as a parent, it looks at the - # release again with its payload sources. - if found is not None and (found.role == "dataset" or role == "payload-source"): - return + """ # ruff: ignore[docstring-extraneous-exception] - the file, access and missing errors are raised as values + output = self.output + if output is None: + raise TimeFValidationError(f"{reference} must be in a local registry, but the read has none") held, damage = self.in_output(reference) if held is not None: if self.open_only: require_open_access(held.metadata, held.lineage) sources = list(held.payload_sources.values()) if role == "dataset" else [] - self.releases[reference] = _Release(role, None, tuple(source.release for source in sources)) + self.releases[reference] = _Release(role, True, None, tuple(source.release for source in sources)) for source in sources: - self.visit(source.release, reference, source) + self.visit(source.release, reference, source, needed_in_local_registry=True) return - answer = self.resolve(reference) - if isinstance(answer, FoundInRegistry) and not _is_same_registry(answer.registry, self.output): + answer = self.lookups.resolve(self.build_sources, reference.dataset_id, str(reference.version)) + if isinstance(answer, FoundInRegistry): self.copy(reference, role, answer.registry) elif damage is not None: # A copy from another registry repairs a damaged copy. Nothing else replaces it. @@ -168,9 +340,7 @@ def visit(self, reference: DatasetRef, needed_by: DatasetRef, pinned: PayloadSou raise answer.access_error(reference.dataset_id) else: # No source provides the release, or only the output does. - raise TimeNetMissingParentError( - reference, needed_by, self.describe_searched_sources(reference), payload_source=role == "payload-source" - ) + raise self.missing(reference, needed_by, role, (output, *self.build_sources)) def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> None: """Add a copy step and plan the payload sources that the copy does not bring. @@ -179,15 +349,15 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non lacks gets its own step first. Args: - reference: The release to copy. + reference: The exact release to copy. role: Whether the release is a dataset or a payload source. registry: The registry to copy from. Raises: - TimeNetDatasetNotFoundError: If the registry no longer holds the release. + TimeNetDatasetNotFoundError: If the release is absent from the registry. TimeFFormatError: If a payload source that the copy brings or reuses fails a file pin. """ # ruff: ignore[docstring-extraneous-exception] - raised by find_payload_sources - manifest = self.manifest(registry, reference) + manifest = self.lookups.manifest(registry, reference) if manifest is None: raise TimeNetDatasetNotFoundError(f"{registry} no longer holds {reference}") if self.open_only: @@ -207,9 +377,9 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non step = CopyStep(reference, registry, with_payload_sources=role == "dataset") # The node holds its final dependencies before the walk visits them. So a release of the walk # that depends on it closes a cycle and does not look it up again. - self.releases[reference] = _Release(role, step, tuple(sources.missing)) + self.releases[reference] = _Release(role, True, step, tuple(sources.missing)) for source, pinned in sources.missing.items(): - self.visit(source, reference, pinned) + self.visit(source, reference, pinned, needed_in_local_registry=True) def build(self, reference: DatasetRef, role: Role, builder: InstalledBuilder) -> None: """Add a build step and plan the parents declared in its card. @@ -223,43 +393,31 @@ def build(self, reference: DatasetRef, role: Role, builder: InstalledBuilder) -> if self.open_only: require_open_access(card) parents = tuple(card.parents.values()) - self.releases[reference] = _Release(role, BuildStep(reference, builder), parents) + self.releases[reference] = _Release(role, True, BuildStep(reference, builder), parents) # A build opens each of its parents with the payload sources of that parent. for parent in parents: - self.visit(parent, reference) + self.visit(parent, reference, needed_in_local_registry=True) - def resolve(self, reference: DatasetRef) -> ReleaseLookup: - """Find an exact release across the sources in preference order. + @staticmethod + def missing( + reference: DatasetRef, needed_by: DatasetRef | None, role: Role, searched: Iterable[ReleaseSource] + ) -> TimeNetDatasetNotFoundError: + """Create the error for a release that no source provides. Args: - reference: The exact release to find. - - Returns: - The first ``Found`` answer, otherwise the first ``NotServed`` answer or ``NotFound()``. - """ - return resolve_release(self.sources, reference.dataset_id, str(reference.version), ask=self._ask) - - def ask(self, source: ReleaseSource, reference: DatasetRef) -> ReleaseLookup: - """Get and cache a source's answer for an exact release. - - Args: - source: The source to query. - reference: The exact release to find. + reference: The missing release. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + role: Whether the release is a dataset or a payload source. + searched: The searched sources, in order. Returns: - The cached or newly fetched answer. + A ``TimeNetDatasetNotFoundError`` for the requested release, else a ``TimeNetMissingParentError``. """ - key = (id(source), reference) - if key not in self.answers: - self.answers[key] = source.resolve(reference.dataset_id, reference.version) - return self.answers[key] - - def _ask(self, source: ReleaseSource, dataset_id: str, version: Version | None) -> ReleaseLookup: - """Return the answer of a source for a release, from :meth:`ask` for an exact version.""" - # The plan asks only for exact versions, so it keeps only those answers. - if version is None: - return source.resolve(dataset_id, None) - return self.ask(source, DatasetRef(dataset_id=dataset_id, version=version)) + if needed_by is None: + names = ", ".join(str(source) for source in searched) + return TimeNetDatasetNotFoundError(f"{reference} is in none of the sources: {names}.") + described = describe_searched_sources(searched, reference) + return TimeNetMissingParentError(reference, needed_by, described, payload_source=role == "payload-source") def listed(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: """Get a release's manifest if the registry holds the release, also if it does not serve it. @@ -267,35 +425,22 @@ def listed(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | No Returns: The manifest, or ``None`` if the registry does not hold the release. """ - if isinstance(self.ask(registry, reference), NotFound): + if isinstance(self.lookups.ask(registry, reference), NotFound): return None - return self.manifest(registry, reference) - - def manifest(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: - """Read and cache a release's manifest. - - A manifest in a cached source answer takes precedence over a registry read. - - Returns: - The manifest, or ``None`` if the registry does not hold the release. - """ - key = (id(registry), reference) - if key not in self.manifests: - answer = self.answers.get(key) - if isinstance(answer, FoundInRegistry | NotServed) and answer.manifest is not None: - validate_manifest_reference(answer.manifest, reference) - self.manifests[key] = answer.manifest - else: - self.manifests[key] = registry.find_manifest(reference) - return self.manifests[key] + return self.lookups.manifest(registry, reference) def in_output(self, reference: DatasetRef) -> tuple[Manifest | None, TimeNetFileError | None]: - """Check a release in the output for missing files or incorrect file sizes. + """Find a release in the output and check local files for damage. + + Missing files and incorrect file sizes count as damage in a local registry. Returns: - ``(manifest, None)`` if intact, ``(None, error)`` if damaged, else ``(None, None)``. + ``(manifest, None)`` if found, ``(None, error)`` if damaged, or ``(None, None)`` if absent or if the + plan has no output. """ - manifest = self.manifest(self.output, reference) + if self.output is None: + return None, None + manifest = self.lookups.manifest(self.output, reference) if manifest is None: return manifest, None if reference not in self.damage: @@ -303,47 +448,51 @@ def in_output(self, reference: DatasetRef) -> tuple[Manifest | None, TimeNetFile damage = self.damage[reference] return (manifest, None) if damage is None else (None, damage) - def describe_searched_sources(self, reference: DatasetRef) -> str: - """Describe the output and sources searched for a missing release. - - Connector descriptions include their declared version when it differs from the requested version. - - Args: - reference: The missing release. - - Returns: - The source descriptions joined with ``" or "``. - """ - names = [str(self.output)] - for source in self.sources: - card = find_card(reference.dataset_id) if isinstance(source, InstalledConnectors) else None - if card is not None and card.dataset_version != reference.version: - names.append(f"{source} ({reference.dataset_id} builds {card.dataset_version})") - else: - names.append(str(source)) - return " or ".join(names) - def check(self) -> None: - """Check each pinned release against the pins of every release that reads it. + """Check the pins of each pinned release, then the files of each release that a read opens in place. Raises: - TimeFFormatError: If a copied or held release fails a file pin, or two readers of a build pin one file - differently. - TimeNetDatasetNotFoundError: If a registry no longer holds a copied or held release. - """ # ruff: ignore[docstring-extraneous-exception] - raised by check_pins + TimeFFormatError: If a read, copied or held release fails a file pin, or two readers of a build pin one + file differently. + TimeNetDatasetNotFoundError: If a registry no longer holds a release of the plan. + TimeNetFileError: If a local file of a release that a read opens is missing or has another size. + """ # ruff: ignore[docstring-extraneous-exception] - raised by check_pins and open_release for reference, readers in self.pins.items(): step = self.releases[reference].step if isinstance(step, BuildStep): # One build must match every reader, so the readers must pin each file alike. _check_agreement(reference, readers) continue - # A copy reads the release from its registry. A release without a step is intact in the output. + # A read or a copy uses the release in its registry. A release without a step is intact in the output. registry = self.output if step is None else step.registry - manifest = self.manifest(registry, reference) + manifest = None if registry is None else self.lookups.manifest(registry, reference) if manifest is None: raise TimeNetDatasetNotFoundError(f"{registry} no longer holds {reference}") for reader, pinned in readers.items(): check_pins(reader, pinned, manifest) + for reference, release in self.releases.items(): + step = release.step + if isinstance(step, ReadStep) and isinstance(step.registry, LocalRegistry): + # Opening a local release checks each of its files. + step.registry.open_release(reference, manifest=self.lookups.manifest(step.registry, reference)) + + def open(self, reference: DatasetRef) -> DatasetVersion: + """Open a release of the plan without its payload sources, after the steps of the plan ran. + + Returns: + The release in the registry of its read step, else in the output. + + Raises: + TimeFValidationError: If the release has no read step and the plan has no output. + """ + step = self.releases[reference].step + if isinstance(step, ReadStep): + return step.registry.open_release(reference, manifest=self.lookups.manifest(step.registry, reference)) + if self.output is None: + # Without an output, the planner plans only read steps. So this is a bug of the planner. + raise TimeFValidationError(f"{reference} has no read step and the plan of the read has no output") + # The steps changed the output, so the release opens with a manifest that is read again. + return self.output.open_release(reference) def order(self) -> list[DatasetRef]: """Order the releases so that each release comes after the releases that it depends on. @@ -426,27 +575,26 @@ def plan_steps( # ruff: ignore[too-many-arguments] open_access_only: bool, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> tuple[DatasetRef, list[Step]]: - """Plan a dataset build and its dependencies. + """Plan a dataset build and all required releases. A damaged copy of the requested release in the output gets a forced build. Args: builder: The installed connector that builds the dataset. - output: The local registry that the build writes to. - parents_from: Sources for missing dependencies, in preference order. - force_build: Plan a rebuild even if the requested release exists. - open_access_only: Require ``access: open`` for every needed release. - pinned_by: File pins grouped by dependent release that the requested build must preserve. + output: The registry to write built releases to. + parents_from: The sources to search for missing releases, in order. + force_build: Rebuild the requested release even if the output holds it. + open_access_only: Require open access for every release. + pinned_by: The releases that pin files of the requested release, with their file requirements. Returns: - The requested release and ordered steps, with its own build last if needed. + The requested release and ordered steps, ending with its build if needed. """ - # The output is never a source for itself. - parents_from = [source for source in parents_from if not _is_same_registry(source, output)] card = builder.card if open_access_only: require_open_access(card) root = builder.release + # The graph leaves the output out of the sources of its copies and builds. graph = _Graph(output, parents_from, open_only=open_access_only) if pinned_by: graph.pins[root] = dict(pinned_by) @@ -460,16 +608,14 @@ def plan_steps( # ruff: ignore[too-many-arguments] # closes a cycle, and the walk does not look it up again. if held is None: step = BuildStep(root, builder, force=force_build or damage is not None) - graph.releases[root] = _Release(role="dataset", step=step, dependencies=tuple(card.parents.values())) + graph.releases[root] = _Release("dataset", True, step, tuple(card.parents.values())) for parent in card.parents.values(): - graph.visit(parent, root) + graph.visit(parent, root, needed_in_local_registry=True) else: sources = list(held.payload_sources.values()) - graph.releases[root] = _Release( - role="dataset", step=None, dependencies=tuple(source.release for source in sources) - ) + graph.releases[root] = _Release("dataset", True, None, tuple(source.release for source in sources)) for source in sources: - graph.visit(source.release, root, source) + graph.visit(source.release, root, source, needed_in_local_registry=True) graph.check() return root, graph.create_ordered_steps() @@ -487,3 +633,23 @@ def _is_same_registry(source: ReleaseSource, output: LocalRegistry) -> bool: if source is output: return True return isinstance(source, LocalRegistry) and source.root.resolve() == output.root.resolve() + + +def describe_searched_sources(sources: Iterable[ReleaseSource], reference: DatasetRef) -> str: + """Describe the sources searched for a missing release. + + Args: + sources: The searched sources, in order. + reference: The missing release. + + Returns: + Source names joined with " or ", including connector versions that differ from the requested version. + """ + names = [] + for source in sources: + card = find_card(reference.dataset_id) if isinstance(source, InstalledConnectors) else None + if card is not None and card.dataset_version != reference.version: + names.append(f"{source} ({reference.dataset_id} builds {card.dataset_version})") + else: + names.append(str(source)) + return " or ".join(names) diff --git a/packages/timenet/src/timenet/coordinator/read.py b/packages/timenet/src/timenet/coordinator/read.py new file mode 100644 index 000000000..a1a771529 --- /dev/null +++ b/packages/timenet/src/timenet/coordinator/read.py @@ -0,0 +1,120 @@ +"""Plan and read releases from ordered sources. + +A read opens each release in the registry that provides it, so a release and its payload sources can come from +different registries. A release that only installed connectors provide is built into the local registry first. +""" + +from collections.abc import Sequence +from typing import cast + +from timenet.builders import declared, find_card +from timenet.coordinator.build import run_plan +from timenet.coordinator.plan import Plan, ReadStep, _Graph, describe_searched_sources +from timenet.errors import TimeNetDatasetNotFoundError +from timenet.registry.dependencies import attach_payload_sources +from timenet.registry.local import LocalRegistry +from timenet.registry.version import DatasetVersion +from timenet.sources import InstalledConnectors, NotFound, ReleaseSource +from timenet.types.references import DatasetRef, is_latest + + +def plan_read( + dataset_id: str, version: str | None, *, sources: Sequence[ReleaseSource], store: LocalRegistry | None +) -> Plan: + """Plan and validate the steps needed to read a release. + + A release that a registry provides gets a ``"read"`` step, and a release that only installed connectors + provide gets the steps of its build into ``store``. This checks local files and payload source pins before + any step runs. + + Args: + dataset_id: The dataset ID. + version: An exact version, ``"latest"``, or ``None``, as in :func:`~timenet.sources.resolve_release`. + sources: Registries and ``InstalledConnectors()`` to search, in order. + store: The local registry for copies and builds, or ``None`` to copy and build nothing. + + Returns: + The ordered plan, ending with the requested release's step if it has one. + + Raises: + TimeNetDatasetNotFoundError: If no source provides the release. + TimeNetVersionUnavailableError: If the installed connector builds a different version. + TimeNetAccessError: If a source lists a needed release but no source serves it. + TimeNetMissingParentError: If a required payload source or build parent is unavailable. + TimeNetFileError: If a required local copy has missing files or incorrect file sizes. + TimeFFormatError: If a payload source conflicts with a pin or two pins conflict. + TimeFValidationError: If a build has no local registry to build into, or the releases form a cycle. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the planner and the checks + graph, _ = _walk(dataset_id, version, sources=sources, store=store) + return Plan(tuple(graph.create_ordered_steps())) + + +def read_release( + dataset_id: str, + version: str | None, + *, + sources: Sequence[ReleaseSource], + store: LocalRegistry | None, +) -> DatasetVersion: + """Open a release with its payload sources after completing required copies and builds. + + This can raise any error listed by :func:`plan_read`. + A local registry is required only for copies and builds. + + Args: + dataset_id: The dataset ID. + version: An exact version, ``"latest"``, or ``None``, as in :func:`~timenet.sources.resolve_release`. + sources: Registries and ``InstalledConnectors()`` to search, in order. + store: The local registry for copies and builds, or ``None`` to copy and build nothing. + + Returns: + The release with attached payload sources that match its file pins. + + Raises: + TimeNetDatasetNotFoundError: If no source provides the release. + TimeNetVersionConflictError: If a copied release conflicts with the files in ``store``. + TimeFFormatError: If a build changes files pinned by other releases. + TimeNetBuildError: If a build fails. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the plan, the run and the open + graph, release = _walk(dataset_id, version, sources=sources, store=store) + plan = Plan(tuple(graph.create_ordered_steps())) + if any(not isinstance(step, ReadStep) for step in plan.steps): + # The plan copies and builds only into a local registry. + run_plan(plan, cast("LocalRegistry", store)) + return attach_payload_sources( + graph.open(release), graph.open, searched=lambda: describe_searched_sources(sources, release) + ) + + +def _walk( + dataset_id: str, version: str | None, *, sources: Sequence[ReleaseSource], store: LocalRegistry | None +) -> tuple[_Graph, DatasetRef]: + """Plan a read in one walk of the release graph, and check the plan. + + Returns: + The graph, and the requested release with its exact version. + + Raises: + TimeNetDatasetNotFoundError: If no source provides the release. + """ + graph = _Graph(store, sources, open_only=False) + answer = graph.lookups.resolve(sources, dataset_id, version) + if isinstance(answer, NotFound): + requested = None if is_latest(version) else DatasetRef.parse(f"{dataset_id}@{version}") + if requested is not None: + # Nothing but a registry replaces a damaged copy in the local registry. + damage = graph.in_output(requested)[1] + if damage is not None: + raise damage + connectors = any(isinstance(source, InstalledConnectors) for source in sources) + if connectors and find_card(dataset_id) is not None: + # The installed connector builds another version. + declared(dataset_id, requested.version) + names = ", ".join(str(source) for source in sources) + wanted = dataset_id if requested is None else str(requested) + raise TimeNetDatasetNotFoundError(f"{wanted} is in none of the sources: {names}.") + release = DatasetRef(dataset_id=dataset_id, version=answer.version) + # The walk takes the answer, so that a read without an exact version does not ask again for the exact version. + graph.visit(release, None, needed_in_local_registry=False, answer=answer) + graph.check() + return graph, release diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index eb9791782..2ad9157bc 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -2,7 +2,7 @@ A registry serves compiled TimeF versions. It does not run connector code. Concrete backends must implement the four data-access methods: :meth:`list_datasets`, :meth:`get_manifest`, -:meth:`open_file`, and :meth:`open_version`. :meth:`BaseRegistry.search` is shared. It filters +:meth:`open_file`, and :meth:`open_release`. :meth:`BaseRegistry.search` is shared. It filters the output of :meth:`list_datasets` and uses :meth:`get_manifest` for the type filters. """ @@ -35,7 +35,11 @@ class BaseRegistry(ABC): - """A registry that provides TimeF dataset metadata, manifests, and files.""" + """A searchable source of TimeF datasets, manifests, and files. + + :meth:`open_release` and :meth:`download_release` handle one release without its payload + sources. :func:`~timenet.registry.copy_release` copies a release with its payload sources. + """ def __str__(self) -> str: return type(self).__name__ @@ -109,24 +113,25 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: """ @abstractmethod - def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: - """Open a committed dataset version as a random-access handle. + def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open one exact release without its payload sources. - The returned :class:`~timenet.registry.version.DatasetVersion` object bundles the parsed - manifest with a handle to the files of the version. The handle is rooted in a filesystem. - A reader built from this object does not re-open the registry. It also does not re-read - ``manifest.json``. The filesystem of the handle MUST serve range reads. The reader uses it - to read Parquet footers and value slices out of order. + Opening reads only metadata. The handle has no payload sources, + so a caller can open each of them from another source and attach it with + :func:`~timenet.registry.dependencies.attach_payload_sources`. Args: - dataset_id: The dataset id. - version: The version string, or ``None`` for the latest version. + reference: The exact release. + manifest: The manifest of the release, if the caller already read it from this registry. + The registry may use it instead of reading the manifest again. Returns: - A handle to the manifest and files of the committed version. + A handle to the manifest and files of the release, without its payload sources. Raises: - TimeNetDatasetNotFoundError: If the dataset id or version is unknown. + TimeNetDatasetNotFoundError: If the release is unknown. + TimeFFormatError: If the manifest declares another release. + TimeNetFileError: If a file of a local copy is missing or has another size than recorded. """ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup: diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index 29b533bd4..897d79bbb 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -104,11 +104,11 @@ def open_source(reference: DatasetRef) -> DatasetVersion: raise TimeNetDatasetNotFoundError(f"no manifest at {source_manifest}") return DatasetVersion(read_manifest(source_manifest), version.filesystem, str(source_dir)) - return attach_payload_sources(version, open_source, searched=str(root)) + return attach_payload_sources(version, open_source, searched=lambda: str(root)) def attach_payload_sources( - version: DatasetVersion, open_source: Callable[[DatasetRef], DatasetVersion], *, searched: str + version: DatasetVersion, open_source: Callable[[DatasetRef], DatasetVersion], *, searched: Callable[[], str] ) -> DatasetVersion: """Open a release's payload sources and check each against the files the release pins. @@ -119,7 +119,7 @@ def attach_payload_sources( version: The opened release, without its payload sources. open_source: Open one exact release, without its payload sources. It raises :class:`~timenet.errors.TimeNetDatasetNotFoundError` when the release is absent. - searched: Where ``open_source`` looks, for the error that names a missing source. + searched: Describe where ``open_source`` looks, only for the error that names a missing source. Returns: The release with each payload source attached. @@ -136,7 +136,7 @@ def attach_payload_sources( try: source = open_source(pinned.release) except TimeNetDatasetNotFoundError as error: - raise TimeNetMissingParentError(pinned.release, release, searched, payload_source=True) 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}") @@ -324,39 +324,33 @@ def open_complete(ref: DatasetRef, *, requested: bool) -> DatasetVersion: return DatasetVersion(release, pafs.LocalFileSystem(), str(directory)) version = open_complete(reference, requested=True) - return attach_payload_sources(version, lambda ref: open_complete(ref, requested=False), searched=str(root)) + return attach_payload_sources(version, lambda ref: open_complete(ref, requested=False), searched=lambda: str(root)) -def open_cached_or_remote( - root: Path, - reference: DatasetRef, - open_remote_version: Callable[[DatasetRef], DatasetVersion], - *, - searched: str, +def open_cached_or_remote_release( + root: Path, reference: DatasetRef, open_remote_version: Callable[[DatasetRef], DatasetVersion] ) -> DatasetVersion: - """Open a release and its payload sources from complete cached copies or remotely. + """Open one release without its payload sources, from a complete cached copy or remotely. Args: - root: The cache directory, checked for complete releases before using remote files. + root: The cache directory, checked for a complete copy of the release before using remote files. reference: The exact dataset ID and version to open. open_remote_version: Open one remote release without its payload sources. - searched: The remote registry, for the error that names a missing source. Returns: - A local or remote handle with each payload source attached and checked against its pins. - No catalogs or payload files are opened here. - """ + A local handle over the complete cached copy, else the remote handle. No catalogs or payload + files are opened here. - def open_one(ref: DatasetRef) -> DatasetVersion: - directory = release_dir(root, ref) - cached = _load_complete_local_manifest(directory, ref) - if cached is not None: - return DatasetVersion(cached, pafs.LocalFileSystem(), str(directory)) - return open_remote_version(ref) - - version = open_one(reference) + Raises: + TimeFFormatError: If the manifest declares another release. + """ # ruff: ignore[docstring-extraneous-exception] - raised by validate_manifest_reference + directory = release_dir(root, reference) + cached = _load_complete_local_manifest(directory, reference) + if cached is not None: + return DatasetVersion(cached, pafs.LocalFileSystem(), str(directory)) + version = open_remote_version(reference) validate_manifest_reference(version.manifest, reference) - return attach_payload_sources(version, open_one, searched=f"{searched} or its cache {root}") + return version def _ensure_dataset_version_available_locally( # ruff: ignore[too-many-arguments] diff --git a/packages/timenet/src/timenet/registry/lazy_read.py b/packages/timenet/src/timenet/registry/lazy_read.py index dca91c38b..54c45a7a0 100644 --- a/packages/timenet/src/timenet/registry/lazy_read.py +++ b/packages/timenet/src/timenet/registry/lazy_read.py @@ -9,12 +9,12 @@ from timenet.dataset import TimeFDataset from timenet.errors import TimeFValidationError -from timenet.manifest import PayloadSource +from timenet.manifest import Manifest, PayloadSource from timenet.registry._paths import release_dir -from timenet.registry.dependencies import open_cached_or_remote +from timenet.registry.dependencies import open_cached_or_remote_release from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry -from timenet.types.references import DatasetRef, is_latest +from timenet.types.references import DatasetRef from timenet.writer import TimeFWriter, WriteProgressEvent @@ -81,29 +81,30 @@ def put_dataset( finally: shutil.rmtree(staging, ignore_errors=True) - def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: - """Open a release and its payload sources, then read remote files on demand. - - Opening reads only manifests. Each payload source must be in this registry and match the - files the release pins. + def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open one release without its payload sources, then read remote files on demand. Args: - dataset_id: The dataset ID. - version: The release version, or ``None`` for the latest. + reference: The exact release. + manifest: The manifest of the release, if the caller already read it from this registry. + A complete cached copy uses its own manifest instead. Returns: - A handle that prefers complete cached releases and otherwise uses remote range reads. + A handle over the complete cached copy of the release, if :attr:`local_root` holds one. + Else a handle that reads the remote files with range reads. """ - if is_latest(version): - version = str(self.get_manifest(dataset_id, version).metadata.dataset_version) - return open_cached_or_remote( - self.local_root, DatasetRef.parse(f"{dataset_id}@{version}"), self._open_remote_version, searched=str(self) + return open_cached_or_remote_release( + self.local_root, reference, lambda ref: self._open_remote_version(ref, manifest=manifest) ) @abstractmethod - def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: + def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: """Bind one remote release to a range-capable filesystem, without its payload sources. + Args: + reference: The exact release. + manifest: The manifest of the release, if the caller already read it from this registry. + Returns: - A handle whose payload sources :meth:`open_version` attaches. + A handle without its payload sources. """ diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index e0a077263..5555f1c51 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -13,7 +13,7 @@ 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 check_file_sizes, open_local_release +from timenet.registry.dependencies import check_file_sizes, validate_manifest_reference from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry from timenet.sources import FoundInRegistry, NotFound, ReleaseLookup @@ -268,25 +268,30 @@ def damage(self, release: DatasetRef, manifest: Manifest | None = None) -> TimeN 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. + def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open one committed release as a local handle, without its payload sources. - Reuses the manifest :meth:`get_manifest` already parsed and validated, so a missing version or a - misplaced artifact surfaces there. Roots the handle at the version's directory over a - :class:`pyarrow.fs.LocalFileSystem`, which is zero network and already seekable. Opening reads - only metadata, as :func:`~timenet.registry.open_local_version` does, and finds payload sources - in this registry. + Opening reads only metadata. It checks that each file of the release is present with its + recorded size. Args: - dataset_id: The dataset id. - version: The version string, or ``None`` for the latest. + reference: The exact release. + manifest: The manifest of the release, if the caller already read it from this registry. Returns: - A handle to the committed version's manifest and files. - """ - manifest = self.get_manifest(dataset_id, version) - reference = DatasetRef(dataset_id=dataset_id, version=manifest.metadata.dataset_version) - return open_local_release(release_dir(self._root, reference), manifest, sources_root=self._root) + A handle rooted at the directory of the release over a :class:`pyarrow.fs.LocalFileSystem`. + + Raises: + TimeNetDatasetNotFoundError: If this registry has no committed manifest of the release. + TimeFFormatError: If the manifest declares another release. + TimeNetFileError: If a file of the release is missing or has another size than recorded. + """ # ruff: ignore[docstring-extraneous-exception] - raised by get_manifest and the checks + if manifest is None: + manifest = self.get_manifest(reference.dataset_id, str(reference.version)) + validate_manifest_reference(manifest, reference) + version = DatasetVersion(manifest, pafs.LocalFileSystem(), str(release_dir(self._root, reference))) + check_file_sizes(version, manifest.files.all_files(), owner=str(reference)) + return version def _dataset_dir(self, dataset_id: str) -> Path: """Return the directory holding a dataset's versions. diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index d33cece8f..30b763061 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -118,13 +118,18 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: buffer.seek(0) return buffer - def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: + def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: """Bind one remote manifest to a range-capable filesystem. + Args: + reference: The exact release. + manifest: The manifest of the release, if the caller already fetched it from this registry. + Returns: - A handle without its payload sources, which ``open_version`` attaches. + A handle without its payload sources. """ - manifest = self.get_manifest(reference.dataset_id, str(reference.version)) + if manifest is None: + manifest = self.get_manifest(reference.dataset_id, str(reference.version)) root = f"{reference.dataset_id}/{reference.version}" files = {f"{root}/{part.path}": part.size for part in manifest.files.all_files()} filesystem = RegistryFileSystem(self._base_url, files, token=self._http.token, transport=self._http.transport) diff --git a/packages/timenet/src/timenet/registry/s3.py b/packages/timenet/src/timenet/registry/s3.py index 4ec31d639..326c0e8e8 100644 --- a/packages/timenet/src/timenet/registry/s3.py +++ b/packages/timenet/src/timenet/registry/s3.py @@ -4,9 +4,9 @@ matches the hosted registry's bucket layout. The manifest's relative file paths do not change. Only the key prefix differs from a :class:`~timenet.registry.LocalRegistry`. Credentials, region, and an optional endpoint override come from boto3's default session (``AWS_*`` vars, ``AWS_PROFILE``, -``AWS_ENDPOINT_URL``). Opening a version reads its payload sources' manifests upfront and reads remote -payloads on demand. Complete cached releases are reused. The backend has no catalog, so listing and -search are unsupported. Use it to publish and to fetch by an explicit ``org/name@version``. +``AWS_ENDPOINT_URL``). Opening a version reads remote payloads on demand. Complete cached releases +are reused. The backend has no catalog, so listing and search are unsupported. Use it to publish and +to fetch by an explicit ``org/name@version``. boto3 is optional. Install the ``timenet[s3]`` extra to use this backend. """ @@ -140,13 +140,18 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: raise TimeNetDatasetNotFoundError(f"no object {relpath!r} for {dataset_id!r} version {version!r}") from exc return response["Body"] - def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: + def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: """Bind one remote manifest to Arrow's native S3 filesystem. + Args: + reference: The exact release. + manifest: The manifest of the release, if the caller already fetched it from this registry. + Returns: - A handle without its payload sources, which ``open_version`` attaches. + A handle without its payload sources. """ - manifest = self.get_manifest(reference.dataset_id, str(reference.version)) + if manifest is None: + manifest = self.get_manifest(reference.dataset_id, str(reference.version)) root = f"{self._bucket}/{self._key(reference.dataset_id, str(reference.version))}" return DatasetVersion(manifest, self._s3_filesystem(), root) diff --git a/packages/timenet/tests/_demo_child.py b/packages/timenet/tests/_demo_child.py new file mode 100644 index 000000000..4316831c2 --- /dev/null +++ b/packages/timenet/tests/_demo_child.py @@ -0,0 +1,32 @@ +"""A local registry with the demo release and a child that reads one of its Records.""" + +from pathlib import Path + +from timenet.dataset import TimeFDataset +from timenet.reader import TimeFReader +from timenet.registry import LocalRegistry, open_local_version +from timenet.registry._paths import release_dir +from timenet.testing import make_dataset +from timenet.types.reference_keys import ParentAlias +from timenet.types.references import DatasetRef + + +def _child_of_the_demo(root: Path) -> tuple[LocalRegistry, DatasetRef]: + """Store the demo dataset and ``test/child``, which reads one of its Records, in a local registry. + + Returns: + The registry, and the demo release, which is the payload source of the child. + """ + parent = make_dataset() + registry = LocalRegistry(root) + registry.put_dataset(parent) + reference = DatasetRef(dataset_id=parent.metadata.dataset_id, version=parent.metadata.dataset_version) + child = TimeFDataset( + metadata=parent.metadata.model_copy( + update={"dataset_id": "test/child", "parents": {ParentAlias("parent"): reference}} + ) + ) + with TimeFReader(open_local_version(release_dir(root, reference))) as reader: + child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) + registry.put_dataset(child) + return registry, reference diff --git a/packages/timenet/tests/registry/_fake_registry.py b/packages/timenet/tests/_fake_registry.py similarity index 100% rename from packages/timenet/tests/registry/_fake_registry.py rename to packages/timenet/tests/_fake_registry.py diff --git a/packages/timenet/tests/conftest.py b/packages/timenet/tests/conftest.py index 3e7a2ecf6..cc6be6f89 100644 --- a/packages/timenet/tests/conftest.py +++ b/packages/timenet/tests/conftest.py @@ -5,10 +5,16 @@ """ import os +from pathlib import Path +import sys from hypothesis import HealthCheck, settings +# Make the test helpers of this directory (e.g. ``_demo_child``) importable by name under pytest's +# importlib import mode, which does not add test dirs to ``sys.path``. +sys.path.insert(0, str(Path(__file__).parent)) + settings.register_profile("dev", max_examples=10) settings.register_profile("ci", max_examples=100, print_blob=True) settings.register_profile( diff --git a/packages/timenet/tests/coordinator/test_build.py b/packages/timenet/tests/coordinator/test_build.py index 103af8160..16e22d550 100644 --- a/packages/timenet/tests/coordinator/test_build.py +++ b/packages/timenet/tests/coordinator/test_build.py @@ -10,7 +10,7 @@ from timenet.errors import TimeNetAccessError, TimeNetVersionUnavailableError from timenet.progress import StepFinished, StepStarted from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry +from timenet.registry import LocalRegistry, open_local_version from timenet.registry._paths import release_dir from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset @@ -112,7 +112,7 @@ def _store_child_without_parent(registry): parent = make_dataset() registry.put_dataset(parent) child = TimeFDataset(metadata=_child_card()) - with TimeFReader(registry.open_version(PARENT_REF.dataset_id, str(PARENT_REF.version))) as reader: + with TimeFReader(open_local_version(release_dir(registry.root, PARENT_REF))) as reader: child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) registry.put_dataset(child) shutil.rmtree(registry.root / PARENT_REF.dataset_id) @@ -257,8 +257,8 @@ def test_plan_release_build_with_open_access_only_refuses_a_held_release_with_a_ store.put_dataset(_demo(_card(REF_AB))) dataset = TimeFDataset(metadata=child_card) with ( - TimeFReader(store.open_version(PARENT_REF.dataset_id, str(PARENT_REF.version))) as gated_reader, - TimeFReader(store.open_version(REF_AB.dataset_id, str(REF_AB.version))) as open_reader, + TimeFReader(open_local_version(release_dir(store.root, PARENT_REF))) as gated_reader, + TimeFReader(open_local_version(release_dir(store.root, REF_AB))) as open_reader, ): dataset.import_from(gated_reader, alias=ParentAlias("parent")) dataset.import_from(open_reader, alias=ParentAlias("open"), records=[make_dataset().records[0].id]) diff --git a/packages/timenet/tests/coordinator/test_read.py b/packages/timenet/tests/coordinator/test_read.py new file mode 100644 index 000000000..15324741e --- /dev/null +++ b/packages/timenet/tests/coordinator/test_read.py @@ -0,0 +1,578 @@ +from pathlib import Path +import re +import shutil + +from _demo_child import _child_of_the_demo +from _fake_registry import build_fake, build_service_fake +import httpx +import numpy as np +import pytest + +from timenet import builders +from timenet.coordinator.plan import BuildStep, CopyStep, ReadStep +from timenet.coordinator.read import plan_read, read_release +from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset +from timenet.errors import ( + TimeFFormatError, + TimeFValidationError, + TimeNetAccessError, + TimeNetDatasetNotFoundError, + TimeNetFileError, + TimeNetMissingParentError, + TimeNetVersionUnavailableError, +) +from timenet.reader import TimeFReader +from timenet.registry import LocalRegistry, RemoteRegistry, open_local_version +from timenet.registry._paths import release_dir +from timenet.sources import InstalledConnectors, ReleaseSource +from timenet.testing import install_backends, make_dataset +from timenet.types import Access, TimeSeriesSpec, ureg +from timenet.types.access import AccessRequirement +from timenet.types.reference_keys import ParentAlias, SourceKey +from timenet.types.references import DatasetRef + + +DEMO = DatasetRef.parse("timenet/hello-world@1.0.0") +CHILD = DatasetRef.parse("test/child@1.0.0") +PARENT = DatasetRef.parse("test/parent@1.0.0") +OTHER = DatasetRef.parse("test/other@1.0.0") +_GATED = {"access": Access.CREDENTIALED, "access_url": "https://example.org/dua"} + + +def _build(release: DatasetRef, **fields) -> BuildStep: + """Return the build step of ``release``, with the builder that is installed for it.""" + return BuildStep(release, builders.declared(release.dataset_id), **fields) + + +def _card(reference, **update): + """Return the card of the demo dataset under another release.""" + return make_dataset().metadata.model_copy( + update={"dataset_id": reference.dataset_id, "dataset_version": reference.version, **update} + ) + + +def _demo(card): + """Return the demo dataset under ``card``.""" + dataset = make_dataset() + dataset._metadata = card + return dataset + + +class _Backend: + """A builder backend of another package. + + It builds CHILD as ``_child_of_the_demo()`` does, and the demo dataset under each other card. A + build hands its pins to the writer, which publishes the release only if it reproduces them. + """ + + def __init__(self, cards, calls): + self.cards = {card.dataset_id: card for card in cards} + self.calls = calls + + def dataset_ids(self): + return list(self.cards) + + def card(self, dataset_id): + return self.cards.get(dataset_id) + + def build(self, release, store, *, force, pinned_by, options): + self.calls.append((release, force, dict(pinned_by))) + if release == CHILD: + _child_of_the_demo(store.root) + else: + store.put_dataset(_demo(self.cards[release.dataset_id]), force=force, pinned_by=pinned_by or None) + + +def _install_demo_backend(monkeypatch, *cards) -> list: + """Make one backend the only installed one. It builds the demo release, and a release for each card. + + Returns: + The list that receives ``(release, force, pinned_by)`` for each build. + """ + calls = [] + backend = _Backend((make_dataset().metadata, *cards), calls) + install_backends(monkeypatch, backend) + return calls + + +def _child_reading(store, *cards, child_release=CHILD): + """Store a child, which reads one Record of the demo dataset under each card, without those releases. + + Args: + store: The registry to store the child in. + *cards: The cards of the releases that the child reads. + child_release: The release of the child. + + Returns: + The pins of the child on each release that it reads, in the order of its manifest. + """ + parents = { + ParentAlias(f"parent{index}"): DatasetRef(dataset_id=card.dataset_id, version=card.dataset_version) + for index, card in enumerate(cards) + } + child = TimeFDataset(metadata=_card(child_release, parents=parents)) + for (alias, parent), card in zip(parents.items(), cards, strict=True): + store.put_dataset(_demo(card)) + with TimeFReader(open_local_version(release_dir(store.root, parent))) as reader: + child.import_from(reader, alias=alias, records=[make_dataset().records[0].id]) + store.put_dataset(child) + for parent in parents.values(): + shutil.rmtree(release_dir(store.root, parent)) + manifest = store.get_manifest(child_release.dataset_id) + return {pinned.release: pinned for pinned in manifest.payload_sources.values()} + + +def _gated_remote(tmp_path) -> RemoteRegistry: + """Return a remote registry that lists DEMO as credentialed, so it does not serve its files.""" + origin = LocalRegistry(tmp_path / "origin") + origin.put_dataset(_demo(make_dataset().metadata.model_copy(update=_GATED))) + transport, _ = build_fake(release_dir(origin.root, DEMO)) + return RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache") + + +def _offline(request): + raise httpx.ConnectError("the network is down", request=request) + + +def _count_reads(monkeypatch, registry) -> dict[str, int]: + """Count each manifest that is read from a registry, by the release that it was asked for.""" + reads, get_manifest = {}, registry.get_manifest + + def counted(dataset_id, version=None): + reads[f"{dataset_id}@{version}"] = reads.get(f"{dataset_id}@{version}", 0) + 1 + return get_manifest(dataset_id, version) + + monkeypatch.setattr(registry, "get_manifest", counted) + return reads + + +def test_a_read_reads_in_place_from_the_first_source_that_holds_the_release(tmp_path): + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + plan = plan_read(DEMO.dataset_id, None, sources=[store, origin], store=store) + assert plan.steps == (ReadStep(DEMO, origin),) + version = read_release(DEMO.dataset_id, None, sources=[store, origin], store=store) + assert Path(version.root).is_relative_to(origin.root) + assert not store.root.exists() + + +def test_a_damaged_copy_fails_the_read_before_any_step_runs(tmp_path, monkeypatch): + store, _ = _child_of_the_demo(tmp_path / "store") + # No registry holds DEMO, so the plan builds it before the read opens CHILD. + shutil.rmtree(release_dir(store.root, DEMO)) + # Origin and the connector also provide CHILD, but the read stops at the damaged copy, which comes first. + origin = LocalRegistry(tmp_path / "origin") + shutil.copytree(store.root, origin.root) + (release_dir(store.root, CHILD) / "control.duckdb").write_bytes(b"cut short") + calls = _install_demo_backend(monkeypatch, _card(CHILD, parents={ParentAlias("parent"): DEMO})) + with pytest.raises(TimeNetFileError, match=r"^test/child@1\.0\.0: file 'control\.duckdb' has 9 bytes"): + read_release(CHILD.dataset_id, None, sources=[store, origin, InstalledConnectors()], store=store) + assert calls == [] + + +@pytest.mark.parametrize("answer", ["build", "not-found", "not-served"]) +def test_a_damaged_copy_in_the_store_fails_a_read_that_no_registry_serves(tmp_path, monkeypatch, answer): + # The store is not in the list, so no registry answers for its damaged copy. + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + (release_dir(store.root, DEMO) / "control.duckdb").write_bytes(b"cut short") + _install_demo_backend(monkeypatch) + sources: list[ReleaseSource] + if answer == "build": + sources = [InstalledConnectors()] + elif answer == "not-found": + sources = [LocalRegistry(tmp_path / "empty")] + else: + # The damage error comes before the access error of the registry that lists DEMO. + sources = [_gated_remote(tmp_path)] + with pytest.raises(TimeNetFileError, match=r"^timenet/hello-world@1\.0\.0: file 'control\.duckdb' has 9 bytes"): + plan_read(DEMO.dataset_id, "1.0.0", sources=sources, store=store) + + +def test_a_payload_source_resolves_on_its_own(tmp_path): + store, _ = _child_of_the_demo(tmp_path / "store") + shutil.rmtree(release_dir(store.root, DEMO)) + origin = LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + plan = plan_read(CHILD.dataset_id, None, sources=[store, origin], store=store) + assert plan.steps == (ReadStep(DEMO, origin), ReadStep(CHILD, store, depends_on=(DEMO,))) + version = read_release(CHILD.dataset_id, None, sources=[store, origin], store=store) + assert Path(version.root).is_relative_to(store.root) + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(origin.root) + assert not store.exists(DEMO.dataset_id, str(DEMO.version)) + + +def test_a_payload_source_with_other_files_than_the_pins_fails_the_plan(tmp_path): + store, _ = _child_of_the_demo(tmp_path / "store") + shutil.rmtree(release_dir(store.root, DEMO)) + origin = LocalRegistry(tmp_path / "origin") + # The same release with its values in another format: CHILD pins files that it does not have. + origin.put_dataset(make_dataset(), values_backend="zarr") + with pytest.raises(TimeFFormatError, match=r"^payload source timenet/hello-world@1\.0\.0 differs"): + plan_read(CHILD.dataset_id, None, sources=[store, origin], store=store) + + +def test_a_missing_payload_source_builds_against_the_pins_of_the_release(tmp_path, monkeypatch): + store, _ = _child_of_the_demo(tmp_path / "store") + shutil.rmtree(release_dir(store.root, DEMO)) + pins = store.get_manifest(CHILD.dataset_id).payload_sources[SourceKey(str(DEMO))] + calls = _install_demo_backend(monkeypatch) + sources = [store, InstalledConnectors()] + assert plan_read(CHILD.dataset_id, None, sources=sources, store=store).steps == ( + _build(DEMO, pinned_by={CHILD: pins}), + ReadStep(CHILD, store, depends_on=(DEMO,)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(DEMO, False, {CHILD: pins})] + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +def test_a_read_builds_a_release_that_only_a_connector_provides(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + calls = _install_demo_backend(monkeypatch, _card(CHILD, parents={ParentAlias("parent"): DEMO})) + sources = [store, InstalledConnectors()] + # The build of CHILD reads DEMO, which the store holds, so DEMO gets no step. + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + assert plan.steps == (_build(CHILD),) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(CHILD, False, {})] + assert Path(version.root).is_relative_to(store.root) + # A payload source without a step opens from the store. + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +@pytest.mark.parametrize(("version", "release"), [(None, "x/y"), ("1.0.0", "x/y@1.0.0")]) +def test_a_release_that_no_source_provides_names_every_source(tmp_path, monkeypatch, version, release): + store = LocalRegistry(tmp_path / "store") + _install_demo_backend(monkeypatch) + with pytest.raises(TimeNetDatasetNotFoundError) as caught: + plan_read("x/y", version, sources=[store, InstalledConnectors()], store=store) + assert type(caught.value) is TimeNetDatasetNotFoundError + assert str(caught.value) == f"{release} is in none of the sources: {store}, installed connectors." + + +def test_a_missing_payload_source_names_the_version_that_the_connector_builds(tmp_path, monkeypatch): + origin, _ = _child_of_the_demo(tmp_path / "origin") + shutil.rmtree(release_dir(origin.root, DEMO)) + _install_demo_backend(monkeypatch, _card(DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"))) + with pytest.raises(TimeNetMissingParentError) as caught: + plan_read(CHILD.dataset_id, None, sources=[origin, InstalledConnectors()], store=None) + assert str(caught.value) == ( + f"payload source {DEMO} of {CHILD} is not in {origin} or installed connectors ({DEMO.dataset_id} builds 2.0.0)." + ) + + +def test_pins_that_the_store_copy_of_a_payload_source_breaks_fail_the_read_before_any_step(tmp_path, monkeypatch): + # Origin holds CHILD, which reads OTHER and DEMO. Only the connectors provide them. The store + # holds DEMO with its values in another format, so the plan builds nothing that checks the pins + # of CHILD on DEMO. The plan builds OTHER first. + origin, store = LocalRegistry(tmp_path / "origin"), LocalRegistry(tmp_path / "store") + _child_reading(origin, _card(OTHER), make_dataset().metadata) + store.put_dataset(make_dataset(), values_backend="zarr") + calls = _install_demo_backend(monkeypatch, _card(OTHER)) + with pytest.raises(TimeFFormatError, match=r"^payload source timenet/hello-world@1\.0\.0 differs"): + read_release(CHILD.dataset_id, None, sources=[origin, InstalledConnectors()], store=store) + assert calls == [] + + +def test_a_release_that_a_source_lists_without_serving_raises_the_access_error(tmp_path): + with pytest.raises(TimeNetAccessError) as caught: + read_release(DEMO.dataset_id, None, sources=[_gated_remote(tmp_path)], store=None) + assert caught.value.requirements == (AccessRequirement(DEMO, Access.CREDENTIALED, _GATED["access_url"]),) + + +def test_an_exact_version_that_the_connector_does_not_declare_raises_version_unavailable(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + _install_demo_backend(monkeypatch) + with pytest.raises(TimeNetVersionUnavailableError) as caught: + read_release(DEMO.dataset_id, "2.0.0", sources=[store, InstalledConnectors()], store=store) + assert (caught.value.release, caught.value.declared) == (DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"), DEMO.version) + + +def test_latest_without_the_connectors_in_the_list_reads_the_older_copy_and_builds_nothing(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + # The installed connector builds a newer version, but the list does not hold the connectors. + calls = _install_demo_backend(monkeypatch, _card(DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"))) + assert plan_read(DEMO.dataset_id, "latest", sources=[store], store=store).steps == (ReadStep(DEMO, store),) + version = read_release(DEMO.dataset_id, "latest", sources=[store], store=store) + assert version.manifest.metadata.dataset_version == DEMO.version + assert calls == [] + + +def test_a_read_without_a_version_stops_at_the_first_source_that_serves_it(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + # A later registry cannot be reached, and the installed connector builds a newer version. + down = RemoteRegistry("http://down.local", transport=httpx.MockTransport(_offline), cache_dir=tmp_path / "c") + calls = _install_demo_backend(monkeypatch, _card(DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"))) + + version = read_release(DEMO.dataset_id, None, sources=[store, down, InstalledConnectors()], store=store) + + assert version.manifest.metadata.dataset_version == DEMO.version + assert calls == [] + + +def test_a_read_that_builds_reads_the_card_of_the_connector_once(tmp_path, monkeypatch): + reads = [] + + class _Counted(_Backend): + def card(self, dataset_id): + reads.append(dataset_id) + return super().card(dataset_id) + + install_backends(monkeypatch, _Counted((make_dataset().metadata,), [])) + plan_read(DEMO.dataset_id, None, sources=[InstalledConnectors()], store=LocalRegistry(tmp_path / "store")) + assert reads == [DEMO.dataset_id] + + +def test_a_read_that_finds_every_release_asks_the_connectors_nothing(tmp_path, monkeypatch): + asked = [] + + class _Counted(_Backend): + def dataset_ids(self): + asked.append("dataset_ids") + return super().dataset_ids() + + def card(self, dataset_id): + asked.append(dataset_id) + return super().card(dataset_id) + + install_backends(monkeypatch, _Counted((make_dataset().metadata,), [])) + store, _ = _child_of_the_demo(tmp_path / "store") + read_release(CHILD.dataset_id, None, sources=[store, InstalledConnectors()], store=store) + # Only the error for a missing payload source names the sources, with the versions that the connectors build. + assert asked == [] + + +def test_a_read_that_needs_a_build_without_a_store_raises(monkeypatch): + _install_demo_backend(monkeypatch) + with pytest.raises(TimeFValidationError, match=r"timenet/hello-world@1\.0\.0 needs a build"): + read_release(DEMO.dataset_id, None, sources=[InstalledConnectors()], store=None) + + +@pytest.mark.parametrize("parent", [PARENT, DatasetRef.parse("x/parent@1.0.0")], ids=["build-first", "read-first"]) +def test_a_release_that_a_build_copies_is_read_from_the_store_and_appears_once(tmp_path, monkeypatch, parent): + # CHILD reads the parent and DEMO. Only the connectors build the parent, and its card declares + # DEMO, which origin holds. So the build of the parent copies DEMO into the store, and the read + # takes it there. The read plans the payload sources in the order of their names. + parent_card = _card(parent, parents={ParentAlias("demo"): DEMO}) + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + pins = _child_reading(store, parent_card, make_dataset().metadata) + origin.put_dataset(make_dataset()) + calls = _install_demo_backend(monkeypatch, parent_card) + sources = [store, origin, InstalledConnectors()] + reads = _count_reads(monkeypatch, origin) + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + # The read and the build of the parent share what the plan read. + assert reads == {str(DEMO): 1} + assert plan.steps == ( + CopyStep(DEMO, origin), + _build(parent, depends_on=(DEMO,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=tuple(pins)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(parent, False, {CHILD: pins[parent]})] + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +def test_a_step_that_two_builds_need_appears_once(tmp_path, monkeypatch): + # CHILD reads PARENT and OTHER. Only the connectors build them, and both of their cards declare + # DEMO, which origin holds. So both builds copy DEMO first. + cards = [_card(reference, parents={ParentAlias("demo"): DEMO}) for reference in (PARENT, OTHER)] + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + pins = _child_reading(store, *cards) + origin.put_dataset(make_dataset()) + _install_demo_backend(monkeypatch, *cards) + + plan = plan_read(CHILD.dataset_id, None, sources=[store, origin, InstalledConnectors()], store=store) + + assert [(step.release, step.action) for step in plan.steps] == [ + (DEMO, "copy"), + *((reference, "build") for reference in pins), + (CHILD, "read"), + ] + + +@pytest.mark.parametrize( + "parent", [PARENT, DatasetRef.parse("a/parent@1.0.0")], ids=["pinned-build-first", "unpinned-build-first"] +) +def test_two_builds_of_one_release_merge_into_one_build_with_the_pins_of_each(tmp_path, monkeypatch, parent): + # CHILD reads the parent and OTHER, and the card of the parent declares OTHER. So the build of + # the parent builds OTHER without pins, and the read builds OTHER with the pins of CHILD. The read + # plans the payload sources in the order of their names. + cards = [_card(parent, parents={ParentAlias("other"): OTHER}), _card(OTHER)] + store = LocalRegistry(tmp_path / "store") + pins = _child_reading(store, *cards) + calls = _install_demo_backend(monkeypatch, *cards) + sources = [store, InstalledConnectors()] + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + # The one build of OTHER comes before the build of the parent, which needs it. + assert plan.steps == ( + _build(OTHER, pinned_by={CHILD: pins[OTHER]}), + _build(parent, depends_on=(OTHER,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=tuple(pins)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(OTHER, False, {CHILD: pins[OTHER]}), (parent, False, {CHILD: pins[parent]})] + assert Path(version.payload_sources[SourceKey(str(OTHER))].root).is_relative_to(store.root) + + +def test_two_readers_that_pin_one_file_differently_fail_the_read_before_any_step(tmp_path, monkeypatch): + # CHILD reads PARENT and OTHER. The card of PARENT declares X, which the store holds, and which + # reads OTHER too. So the read builds OTHER with the pins of CHILD, and the build of PARENT builds + # OTHER with the pins of X. X pins one file of OTHER with another checksum, so no build matches both. + x = DatasetRef.parse("test/x@1.0.0") + cards = [_card(PARENT, parents={ParentAlias("x"): x}), _card(OTHER)] + store = LocalRegistry(tmp_path / "store") + _child_reading(store, cards[1], child_release=x) + _child_reading(store, *cards) + manifest = store.get_manifest(x.dataset_id) + ((key, pinned),) = manifest.payload_sources.items() + changed = pinned.files[0].model_copy(update={"checksum": "sha256:" + "0" * 64}) + pinned = pinned.model_copy(update={"files": (changed, *pinned.files[1:])}) + (release_dir(store.root, x) / "manifest.json").write_text( + manifest.model_copy(update={"payload_sources": {key: pinned}}).to_json() + ) + # A new registry, so that no manifest that it read before stays. + store = LocalRegistry(store.root) + calls = _install_demo_backend(monkeypatch, *cards) + message = f"{CHILD} and {x} pin {changed.path!r} of {OTHER} as different files" + with pytest.raises(TimeFFormatError, match=f"^{re.escape(message)}"): + read_release(CHILD.dataset_id, None, sources=[store, InstalledConnectors()], store=store) + assert calls == [] + + +def test_a_remote_read_fetches_each_manifest_once(tmp_path, monkeypatch): + origin, _ = _child_of_the_demo(tmp_path / "origin") + transport, _ = build_service_fake(origin.root) + remote = RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache") + reads = _count_reads(monkeypatch, remote) + + version = read_release(CHILD.dataset_id, None, sources=[remote], store=None) + + assert version.manifest == origin.get_manifest(CHILD.dataset_id) + assert version.payload_sources[SourceKey(str(DEMO))].manifest == origin.get_manifest(DEMO.dataset_id) + # The registry answers "latest" with the manifest, and the plan and the open use that manifest. + assert reads == {f"{CHILD.dataset_id}@None": 1, str(DEMO): 1} + + +def test_latest_reads_the_manifest_of_the_version_that_the_answer_found(tmp_path, monkeypatch): + origin = LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + manifest = origin.get_manifest(DEMO.dataset_id, str(DEMO.version)) + reads = _count_reads(monkeypatch, origin) + # A local registry answers "latest" without its manifest, so the plan reads it at the exact version. + version = read_release(DEMO.dataset_id, None, sources=[origin], store=None) + assert version.manifest == manifest + assert reads == {str(DEMO): 1} + + +@pytest.mark.parametrize("parent", [PARENT, DatasetRef.parse("x/parent@1.0.0")], ids=["build-first", "read-first"]) +def test_a_copy_that_a_build_needs_repairs_a_damaged_copy_that_the_read_found_first(tmp_path, monkeypatch, parent): + # CHILD reads the parent and DEMO. The store comes first in the list and holds DEMO damaged, and origin + # holds DEMO. Only the connectors build the parent, and its card declares DEMO. So the build copies DEMO + # from origin over the damaged copy, and the read takes it there. + parent_card = _card(parent, parents={ParentAlias("demo"): DEMO}) + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + pins = _child_reading(store, parent_card, make_dataset().metadata) + for registry in (store, origin): + registry.put_dataset(make_dataset()) + (release_dir(store.root, DEMO) / "control.duckdb").write_bytes(b"cut short") + _install_demo_backend(monkeypatch, parent_card) + sources = [store, origin, InstalledConnectors()] + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + assert plan.steps == ( + CopyStep(DEMO, origin), + _build(parent, depends_on=(DEMO,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=tuple(pins)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert store.damage(DEMO) is None + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +_SPEC = TimeSeriesSpec(spec_type="sine", name="Sine", unit_value=ureg.dimensionless) + + +def _store_reading(registry: LocalRegistry, release: DatasetRef, source: DatasetRef) -> None: + """Store ``release``, which keeps one Record of ``source`` and adds one Record of its own values.""" + dataset = TimeFDataset(metadata=_card(release, parents={ParentAlias("source"): source})) + with TimeFReader(open_local_version(release_dir(registry.root, source))) as reader: + dataset.import_from(reader, alias=ParentAlias("source"), records=[make_dataset().records[0].id]) + own = Signal.from_values( + np.arange(16, dtype=np.float32), spec=_SPEC, name="own", time_axis=RegularAxis.from_rate_hz(16) + ) + dataset.add_record(record=Record(sources=(Source(name="own", signals=(own,)),))) + registry.put_dataset(dataset) + + +def test_a_payload_source_that_a_build_needs_as_a_parent_comes_with_its_own_payload_sources(tmp_path, monkeypatch): + # CHILD reads the own Record of SOURCE, which reads DEMO, and a Record of the parent. Origin holds SOURCE + # and DEMO. Only the connectors build the parent, and its card declares SOURCE. The read finds SOURCE in + # origin first. The build then needs SOURCE in the store, so the copy takes DEMO along. + source, parent = DatasetRef.parse("test/source@1.0.0"), DatasetRef.parse("x/parent@1.0.0") + parent_card = _card(parent, parents={ParentAlias("source"): source}) + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + store.put_dataset(make_dataset()) + _store_reading(store, source, DEMO) + store.put_dataset(_demo(parent_card)) + child = TimeFDataset(metadata=_card(CHILD, parents={ParentAlias("source"): source, ParentAlias("parent"): parent})) + for alias, reference, record in (("source", source, 2), ("parent", parent, make_dataset().records[0].id)): + with TimeFReader(open_local_version(release_dir(store.root, reference))) as reader: + child.import_from(reader, alias=ParentAlias(alias), records=[record]) + store.put_dataset(child) + for reference in (DEMO, source, parent): + shutil.copytree(release_dir(store.root, reference), release_dir(origin.root, reference)) + shutil.rmtree(release_dir(store.root, reference)) + shutil.rmtree(release_dir(origin.root, parent)) + pins = {pinned.release: pinned for pinned in store.get_manifest(CHILD.dataset_id).payload_sources.values()} + assert list(pins) == [source, parent] + _install_demo_backend(monkeypatch, parent_card) + sources = [store, origin, InstalledConnectors()] + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + assert plan.steps == ( + CopyStep(source, origin), + _build(parent, depends_on=(source,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=(source, parent)), + ) + read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert store.exists(DEMO.dataset_id, str(DEMO.version)) + + +def test_a_read_of_the_latest_version_asks_each_source_once(tmp_path, monkeypatch): + # The first registry holds DEMO 1.0.0, and the second holds DEMO 2.0.0. + newer = DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0") + registries = [] + for name, dataset in (("first", make_dataset()), ("second", _demo(_card(newer)))): + LocalRegistry(tmp_path / name).put_dataset(dataset) + transport, _ = build_service_fake(tmp_path / name) + registries.append(RemoteRegistry(f"http://{name}.local", transport=transport, cache_dir=tmp_path / f"{name}-c")) + reads = _count_reads(monkeypatch, registries[0]) + + version = read_release(DEMO.dataset_id, "latest", sources=registries, store=None) + + assert version.manifest.metadata.dataset_version == newer.version + # The plan takes the answer of the sources for the latest version, and asks for 2.0.0 nowhere. + assert reads == {f"{DEMO.dataset_id}@None": 1} + + +def test_a_cycle_through_the_requested_release_starts_at_it(tmp_path, monkeypatch): + # CHILD reads PARENT, which only the connectors build, and the card of PARENT declares CHILD. + parent_card = _card(PARENT, parents={ParentAlias("child"): CHILD}) + store = LocalRegistry(tmp_path / "store") + _child_reading(store, parent_card) + _install_demo_backend(monkeypatch, parent_card) + + message = f"dataset parent cycle: {CHILD} -> {PARENT} -> {CHILD}" + with pytest.raises(TimeFValidationError, match=f"^{re.escape(message)}$"): + plan_read(CHILD.dataset_id, None, sources=[store, InstalledConnectors()], store=store) diff --git a/packages/timenet/tests/reader/test_shared_storage.py b/packages/timenet/tests/reader/test_shared_storage.py index 55730194a..0d9a7b49b 100644 --- a/packages/timenet/tests/reader/test_shared_storage.py +++ b/packages/timenet/tests/reader/test_shared_storage.py @@ -394,8 +394,6 @@ def test_opening_a_child_detects_a_forced_replace_of_its_source(tmp_path): message = rf"payload source test/p@1\.0\.0 differs from the release test/c@1\.0\.0 was written against: test/c@1\.0\.0 pins '{first}' as \d+ bytes" with pytest.raises(TimeFFormatError, match=message): open_local_version(child_dir) - with pytest.raises(TimeFFormatError, match=message): - LocalRegistry(tmp_path).open_version("test/c", "1.0.0") def test_storing_a_child_detects_a_forced_replace_of_its_source(tmp_path): @@ -420,7 +418,7 @@ def test_a_child_is_refused_by_a_registry_that_holds_another_release_of_its_sour ours, foreign = LocalRegistry(tmp_path / "ours"), LocalRegistry(tmp_path / "foreign") ours.put_dataset(_release("test/p", [_record_in_memory("p", "Ours. " * 300, [0, 1_000])])) foreign.put_dataset(_release("test/p", [_record_in_memory("p", "Foreign. " * 300, [0, 3_000])])) - with TimeFReader(ours.open_version("test/p", "1.0.0")) as reader: + with TimeFReader(open_local_version(ours.root / "test/p/1.0.0")) as reader: loaded = reader.read() with pytest.raises( TimeFFormatError, diff --git a/packages/timenet/tests/registry/conftest.py b/packages/timenet/tests/registry/conftest.py index 4202abcb6..2b736d6c5 100644 --- a/packages/timenet/tests/registry/conftest.py +++ b/packages/timenet/tests/registry/conftest.py @@ -1,16 +1,8 @@ from pathlib import Path -import sys import pytest -from timenet.dataset import Record, Source - - -# Make sibling test helpers (e.g. ``_fake_registry``) importable by name under -# pytest's importlib import mode, which does not add test dirs to ``sys.path``. -sys.path.insert(0, str(Path(__file__).parent)) - -from timenet.dataset import Signal, TimeFDataset +from timenet.dataset import Record, Signal, Source, TimeFDataset from timenet.dataset.axis import RegularAxis from timenet.testing import make_dataset, sine_loader from timenet.types import ( diff --git a/packages/timenet/tests/registry/test_conformance.py b/packages/timenet/tests/registry/test_conformance.py index e22356e5d..fea9190e2 100644 --- a/packages/timenet/tests/registry/test_conformance.py +++ b/packages/timenet/tests/registry/test_conformance.py @@ -9,6 +9,7 @@ from typing import NamedTuple import uuid +from _demo_child import _child_of_the_demo from _fake_registry import build_service_fake import pytest @@ -16,6 +17,7 @@ from timenet.dataset.axis import RegularAxis from timenet.errors import TimeNetDatasetNotFoundError, TimeNetFileError from timenet.registry import LocalRegistry, RemoteRegistry, S3Registry, WritableRegistry +from timenet.registry._paths import release_dir from timenet.sources import FoundInRegistry, NotFound from timenet.testing import make_dataset, sine_loader from timenet.types import ( @@ -184,13 +186,35 @@ def test_open_file_rejects_traversal(backend): backend.registry.open_file("timenet/hello-world", "1.0.0", "../../escape") -# ---- open_version ----------------------------------------------------------------------------- +# ---- open_release ----------------------------------------------------------------------------- -def test_open_version_manifest_matches(backend): - version = backend.registry.open_version("timenet/hello-world") # no pin -> latest - assert version.manifest.metadata.dataset_id == "timenet/hello-world" - assert str(version.manifest.metadata.dataset_version) == "1.1.0" +def test_open_release_opens_one_release_without_its_payload_sources(backend, tmp_path): + source, _ = _child_of_the_demo(tmp_path / "source") + child = DatasetRef.parse("test/child@1.0.0") + manifest = source.get_manifest(child.dataset_id) + backend.registry.put_release(release_dir(source.root, child)) + version = backend.registry.open_release(child) + # The child reads files of the demo release, but the handle opens none of them. + assert manifest.payload_sources + assert version.payload_sources == {} + assert version.manifest == backend.registry.get_manifest(child.dataset_id, str(child.version)) + + +def test_open_release_uses_the_manifest_that_the_caller_passes(backend, tmp_path, monkeypatch): + source, _ = _child_of_the_demo(tmp_path / "source") + child = DatasetRef.parse("test/child@1.0.0") + manifest = source.get_manifest(child.dataset_id) + backend.registry.put_release(release_dir(source.root, child)) + + def no_read(*args, **kwargs): + raise AssertionError("the registry reads the manifest that the caller passed") + + monkeypatch.setattr(backend.registry, "get_manifest", no_read) + # No cache holds a copy of the child, so S3 and the remote registry open their remote files. + version = backend.registry.open_release(child, manifest=manifest) + assert version.manifest == manifest + assert version.payload_sources == {} # ---- exists ----------------------------------------------------------------------------------- diff --git a/packages/timenet/tests/registry/test_registry.py b/packages/timenet/tests/registry/test_registry.py index 01e1438f4..7459a30ba 100644 --- a/packages/timenet/tests/registry/test_registry.py +++ b/packages/timenet/tests/registry/test_registry.py @@ -4,13 +4,12 @@ from pydantic import ValidationError import pytest -from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError, TimeNetRegistryError +from timenet.errors import TimeFFormatError, TimeNetFileError, TimeNetRegistryError from timenet.manifest import Manifest from timenet.reader import TimeFReader from timenet.registry import ( TIMENET_REGISTRY_URL, BaseRegistry, - DatasetVersion, LocalRegistry, RemoteRegistry, S3Registry, @@ -182,36 +181,10 @@ def test_latest_version_ignores_staging_dirs(registry_root): assert manifest.metadata.dataset_version.major == 2 -# ---- open_version ----------------------------------------------------------------------------- - - -def test_open_version_returns_handle(registry_root): - version = LocalRegistry(registry_root).open_version("demo/ecg") - assert isinstance(version, DatasetVersion) - assert version.manifest.metadata.dataset_id == "demo/ecg" - assert version.root == str(registry_root / "demo/ecg" / "2.0.0") - - -def test_open_version_resolves_latest_to_a_concrete_directory(registry_root): - # ``None`` means latest, but the handle roots at that resolved version, never the string "latest" - version = LocalRegistry(registry_root).open_version("demo/ecg", version=None) - assert version.root.endswith("/2.0.0") - - -def test_open_version_filesystem_reads_a_file(registry_root): - version = LocalRegistry(registry_root).open_version("demo/ecg") - with version.filesystem.open_input_file(version.resolve_file_path("manifest.json")) as handle: - assert b"demo/ecg" in handle.read() - - -def test_open_version_unknown_raises(registry_root): - with pytest.raises(TimeNetDatasetNotFoundError): - LocalRegistry(registry_root).open_version("does/not-exist") - - -def test_open_version_unknown_version_raises(registry_root): - with pytest.raises(TimeNetDatasetNotFoundError): - LocalRegistry(registry_root).open_version("demo/ecg", version="9.9.9") +def test_open_release_checks_the_size_of_each_file(registry_root): + (registry_root / "demo/ecg/2.0.0/control.duckdb").write_bytes(b"cut short") + with pytest.raises(TimeNetFileError, match=r"^demo/ecg@2\.0\.0: file 'control\.duckdb' has 9 bytes"): + LocalRegistry(registry_root).open_release(DatasetRef.parse("demo/ecg@2.0.0")) # ---- search ----------------------------------------------------------------------------------- diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py index c00426a6a..7d8d6ef7c 100644 --- a/packages/timenet/tests/registry/test_remote.py +++ b/packages/timenet/tests/registry/test_remote.py @@ -401,7 +401,7 @@ def test_put_dataset_refuses_a_payload_source_the_service_does_not_hold_before_p local.put_dataset(make_dataset()) transport, state = build_publish_fake(token="tok_rw") registry = RemoteRegistry("http://api.local", token="tok_rw", transport=transport, cache_dir=tmp_path / "cache") - with TimeFReader(local.open_version(make_dataset().metadata.dataset_id)) as reader: + with TimeFReader(open_local_version(local.root / "timenet/hello-world/1.0.0")) as reader: parent = reader.read() child = TimeFDataset.from_parts( metadata=parent.metadata.model_copy(update={"dataset_id": "test/child"}), @@ -508,7 +508,7 @@ def test_child_streams_physical_sources_and_downloads_for_offline_reads(tmp_path parent._import_state.parent_cards[ParentAlias("build-only")] = ParentDatasetCard(metadata=build_only) local.put_dataset(parent, values_backend=backend) parent_id = parent.metadata.dataset_id - with TimeFReader(local.open_version(parent_id, "1.0.0")) as reader: + with TimeFReader(open_local_version(local.root / parent_id / "1.0.0")) as reader: reference = DatasetRef(dataset_id=parent_id, version=parent.metadata.dataset_version) child = TimeFDataset( metadata=parent.metadata.model_copy( diff --git a/packages/timenet/tests/registry/test_s3.py b/packages/timenet/tests/registry/test_s3.py index 62d5f95f5..96f2f30a2 100644 --- a/packages/timenet/tests/registry/test_s3.py +++ b/packages/timenet/tests/registry/test_s3.py @@ -15,7 +15,7 @@ TimeNetRegistryError, ) from timenet.reader import TimeFReader # ruff: ignore[module-import-not-at-top-of-file] -from timenet.registry import LocalRegistry, S3Registry # ruff: ignore[module-import-not-at-top-of-file] +from timenet.registry import LocalRegistry, S3Registry, open_local_version # ruff: ignore[module-import-not-at-top-of-file] from timenet.testing import assert_datasets_equal, make_dataset # ruff: ignore[module-import-not-at-top-of-file] from timenet.types import DatasetSchema, Version # ruff: ignore[module-import-not-at-top-of-file] from timenet.types.references import DatasetRef # ruff: ignore[module-import-not-at-top-of-file] @@ -59,7 +59,9 @@ def test_put_dataset_and_remote_load_round_trip(s3_root, tmp_path, backend, entr assert_datasets_equal(make_dataset(), reader.read()) else: assert_datasets_equal(make_dataset(), client.load(make_dataset().metadata.dataset_id)) - assert isinstance(registry.open_version(make_dataset().metadata.dataset_id).filesystem, pafs.S3FileSystem) + assert isinstance( + registry.open_release(DatasetRef.parse("timenet/hello-world@1.0.0")).filesystem, pafs.S3FileSystem + ) assert not (tmp_path / "cache" / make_dataset().metadata.dataset_id).exists() @@ -68,7 +70,7 @@ def test_put_dataset_checks_payload_sources_against_the_bucket(s3_root, tmp_path local.put_dataset(make_dataset()) # moto keeps one bucket state per process, so this test uses a prefix no other test writes. registry = S3Registry("s3://tn-test/payload-sources", cache_dir=tmp_path / "cache") - with TimeFReader(local.open_version(make_dataset().metadata.dataset_id)) as reader: + with TimeFReader(open_local_version(local.root / "timenet/hello-world/1.0.0")) as reader: parent = reader.read() child = TimeFDataset.from_parts( metadata=parent.metadata.model_copy(update={"dataset_id": "test/child"}), @@ -104,7 +106,7 @@ def test_download_then_load_serves_from_cache(s3_root, tmp_path): client = TimeNet(registry=registry, storage_path=tmp_path / "cache") path = client.download(dataset_id) # boto3 GetObject per file, into the cache assert (path / "manifest.json").exists() - handle = registry.open_version(dataset_id) + handle = registry.open_release(DatasetRef.parse(f"{dataset_id}@1.0.0")) assert isinstance(handle.filesystem, pafs.LocalFileSystem) # cache-first: no S3 reads assert_datasets_equal(make_dataset(), client.load(dataset_id)) diff --git a/packages/timenet/tests/registry/test_version.py b/packages/timenet/tests/registry/test_version.py index d8e235e24..d0e5c3b4b 100644 --- a/packages/timenet/tests/registry/test_version.py +++ b/packages/timenet/tests/registry/test_version.py @@ -10,7 +10,7 @@ from timenet.format.checksums import file_checksum from timenet.manifest import PayloadSource from timenet.registry import LocalRegistry, open_local_version -from timenet.registry.dependencies import open_cached_or_remote, prepare_local_copy +from timenet.registry.dependencies import prepare_local_copy from timenet.registry.shared import complete_copy from timenet.types.reference_keys import SourceKey from timenet.types.references import DatasetRef @@ -153,14 +153,6 @@ def test_preparation_needs_only_the_direct_payload_sources(registry_root, tmp_pa with pytest.raises(TimeNetDatasetNotFoundError, match=rf"payload source {parent_ref} of {reference} is not in"): prepare_local_copy(cache, reference) - def missing_remote(ref): - raise TimeNetDatasetNotFoundError(f"no remote release {ref}") - - with pytest.raises( - TimeNetDatasetNotFoundError, match=rf"payload source {parent_ref} .* is not in remote or its cache" - ): - open_cached_or_remote(cache, reference, missing_remote, searched="remote") - prepare_local_copy(cache, reference, fetch_from=LocalRegistry(registry_root)) cached_parent = cache / "timenet/hello-world/1.0.0" assert (cached_parent / "control.duckdb").is_file() @@ -175,7 +167,6 @@ def missing_remote(ref): prepared = prepare_local_copy(cache, reference) assert set(prepared.payload_sources) == {SourceKey(str(parent_ref))} assert prepared.resolve_payload_source(SourceKey(str(parent_ref))).payload_sources == {} - assert open_cached_or_remote(cache, reference, missing_remote, searched="remote").payload_sources @pytest.mark.parametrize("suffix", ["different/ecg/2.0.0", "demo/ecg/3.0.0", "demo/ecg/latest"]) diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index d75e84864..5c94a5523 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -1,7 +1,7 @@ -from pathlib import Path import shutil from types import SimpleNamespace +from _demo_child import _child_of_the_demo import httpx from pydantic import ValidationError import pytest @@ -35,27 +35,6 @@ from timenet.writer import TimeFWriter -def _child_of_the_demo(root: Path) -> tuple[LocalRegistry, DatasetRef]: - """Store the demo dataset and ``test/child``, which reads one of its Records, in a local registry. - - Returns: - The registry, and the demo release, which is the payload source of the child. - """ - parent = make_dataset() - registry = LocalRegistry(root) - registry.put_dataset(parent) - reference = DatasetRef(dataset_id=parent.metadata.dataset_id, version=parent.metadata.dataset_version) - child = TimeFDataset( - metadata=parent.metadata.model_copy( - update={"dataset_id": "test/child", "parents": {ParentAlias("parent"): reference}} - ) - ) - with TimeNet(registry).open(parent.metadata.dataset_id) as reader: - child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) - registry.put_dataset(child) - return registry, reference - - class _Builder: """A builder backend whose cards declare ``VERSION``. @@ -441,7 +420,7 @@ def counting_close(reader): def test_load_reads_in_place_without_downloading(registry_root, tmp_path): - # load() reads through the registry's open_version handle, not download-then-read, so it must not + # load() reads the release where it is, not download-then-read, so it must not # write a copy into local storage the way download() does. storage = tmp_path / "store" client = TimeNet(registry_root, storage_path=storage) diff --git a/packages/timenet/tests/test_engine.py b/packages/timenet/tests/test_engine.py index 0dd8fdee5..431b73b86 100644 --- a/packages/timenet/tests/test_engine.py +++ b/packages/timenet/tests/test_engine.py @@ -11,7 +11,8 @@ from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError from timenet.manifest import Manifest from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry +from timenet.registry import LocalRegistry, open_local_version +from timenet.registry._paths import release_dir from timenet.testing import make_dataset from timenet.types import Access, Version from timenet.types.access import effective_access @@ -396,8 +397,8 @@ def test_a_build_context_made_directly_closes_its_parent_readers(tmp_path, monke for dataset in (make_dataset(), second): registry.put_dataset(dataset) readers = { - ParentAlias("parent"): TimeFReader(registry.open_version(_PARENT.dataset_id, str(_PARENT.version))), - ParentAlias("second"): TimeFReader(registry.open_version("test/second", "1.0.0")), + ParentAlias("parent"): TimeFReader(open_local_version(release_dir(registry.root, _PARENT))), + ParentAlias("second"): TimeFReader(open_local_version(registry.root / "test/second/1.0.0")), } closed = [] close = TimeFReader.close From 72a9ea6981c03042ec7b981cf1aaf7e23ee405ee Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 19:03:41 +0100 Subject: [PATCH 2/4] docs: shorten the docstrings and docs of the read coordinator Say only what a caller of the read plan and of open_release() needs to know. Part of TNET-157. --- packages/timenet/src/timenet/registry/base.py | 3 +-- packages/timenet/src/timenet/registry/dependencies.py | 5 ++--- packages/timenet/src/timenet/registry/lazy_read.py | 6 ++---- 3 files changed, 5 insertions(+), 9 deletions(-) diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index 2ad9157bc..c7777fa91 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -122,8 +122,7 @@ def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = Non Args: reference: The exact release. - manifest: The manifest of the release, if the caller already read it from this registry. - The registry may use it instead of reading the manifest again. + manifest: The manifest of the release from this registry, if you already read it. Returns: A handle to the manifest and files of the release, without its payload sources. diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index 897d79bbb..c9b5afe39 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -333,13 +333,12 @@ def open_cached_or_remote_release( """Open one release without its payload sources, from a complete cached copy or remotely. Args: - root: The cache directory, checked for a complete copy of the release before using remote files. + root: The cache directory to check for a complete copy first. reference: The exact dataset ID and version to open. open_remote_version: Open one remote release without its payload sources. Returns: - A local handle over the complete cached copy, else the remote handle. No catalogs or payload - files are opened here. + A local handle over the complete cached copy, else the remote handle. Raises: TimeFFormatError: If the manifest declares another release. diff --git a/packages/timenet/src/timenet/registry/lazy_read.py b/packages/timenet/src/timenet/registry/lazy_read.py index 54c45a7a0..dd0a217b7 100644 --- a/packages/timenet/src/timenet/registry/lazy_read.py +++ b/packages/timenet/src/timenet/registry/lazy_read.py @@ -86,12 +86,10 @@ def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = Non Args: reference: The exact release. - manifest: The manifest of the release, if the caller already read it from this registry. - A complete cached copy uses its own manifest instead. + manifest: The manifest of the release from this registry, if you already read it. Returns: - A handle over the complete cached copy of the release, if :attr:`local_root` holds one. - Else a handle that reads the remote files with range reads. + A handle over a complete cached copy in :attr:`local_root`, else over remote files. """ return open_cached_or_remote_release( self.local_root, reference, lambda ref: self._open_remote_version(ref, manifest=manifest) From 01652d292ca55a337ef5cde272b68619f1aef2e8 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 00:22:03 +0100 Subject: [PATCH 3/4] docs: shorten the docstrings of the read coordinator Shorter docstrings for the code that this change adds or changes. Part of TNET-157. --- .../timenet/src/timenet/coordinator/build.py | 12 ++++++------ packages/timenet/src/timenet/registry/base.py | 17 ++++++++--------- .../src/timenet/registry/dependencies.py | 10 ++++++---- .../timenet/src/timenet/registry/lazy_read.py | 16 ++++++++-------- packages/timenet/src/timenet/registry/local.py | 17 ++++++++--------- .../src/timenet/registry/remote/registry.py | 10 +++++----- packages/timenet/src/timenet/registry/s3.py | 10 +++++----- 7 files changed, 46 insertions(+), 46 deletions(-) diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py index 139c9d1fc..b62364dbc 100644 --- a/packages/timenet/src/timenet/coordinator/build.py +++ b/packages/timenet/src/timenet/coordinator/build.py @@ -58,18 +58,18 @@ def plan_release_build( # ruff: ignore[too-many-arguments] def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = None) -> None: - """Run a plan's copy and build steps in order. + """Run the copy and build steps of a plan in order. - A read step runs nothing, because its release stays where it is. + Skip read steps. Args: - plan: The steps to run in dependency order. + plan: The steps to run. store: The local registry to copy and build into. - options: Build settings, including the progress callback for copies and builds. + options: Build settings, including the progress callback. Raises: - TimeNetVersionConflictError: If ``store`` holds different files for a copied release. - TimeFFormatError: If a build fails to reproduce pinned files. + TimeNetVersionConflictError: If a copied release conflicts with the files in ``store``. + TimeFFormatError: If a build changes files pinned by other releases. TimeNetBuildError: If a build fails. """ # ruff: ignore[docstring-extraneous-exception] - raised by the copy and the backends options = options or BuildOptions() diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index c7777fa91..aa4c0364f 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -114,23 +114,22 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: @abstractmethod def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: - """Open one exact release without its payload sources. + """Open an exact release without attaching its payload sources. - Opening reads only metadata. The handle has no payload sources, - so a caller can open each of them from another source and attach it with + Opening reads metadata only. Attach payload sources with :func:`~timenet.registry.dependencies.attach_payload_sources`. Args: - reference: The exact release. - manifest: The manifest of the release from this registry, if you already read it. + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle to the manifest and files of the release, without its payload sources. + A handle to the release's manifest and files. Raises: - TimeNetDatasetNotFoundError: If the release is unknown. - TimeFFormatError: If the manifest declares another release. - TimeNetFileError: If a file of a local copy is missing or has another size than recorded. + TimeNetDatasetNotFoundError: If the registry lacks the release. + TimeFFormatError: If the manifest declares a different release. + TimeNetFileError: If a local file is missing or has an incorrect size. """ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup: diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index c9b5afe39..280f0c275 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -330,18 +330,20 @@ def open_complete(ref: DatasetRef, *, requested: bool) -> DatasetVersion: def open_cached_or_remote_release( root: Path, reference: DatasetRef, open_remote_version: Callable[[DatasetRef], DatasetVersion] ) -> DatasetVersion: - """Open one release without its payload sources, from a complete cached copy or remotely. + """Open a complete cached release or fall back to remote storage. + + Payload sources remain unattached. Args: - root: The cache directory to check for a complete copy first. + root: The cache directory to check before opening remote files. reference: The exact dataset ID and version to open. open_remote_version: Open one remote release without its payload sources. Returns: - A local handle over the complete cached copy, else the remote handle. + A handle to the complete cached copy, or a remote handle if the cache is incomplete. Raises: - TimeFFormatError: If the manifest declares another release. + TimeFFormatError: If the manifest declares a different release. """ # ruff: ignore[docstring-extraneous-exception] - raised by validate_manifest_reference directory = release_dir(root, reference) cached = _load_complete_local_manifest(directory, reference) diff --git a/packages/timenet/src/timenet/registry/lazy_read.py b/packages/timenet/src/timenet/registry/lazy_read.py index dd0a217b7..9368a596e 100644 --- a/packages/timenet/src/timenet/registry/lazy_read.py +++ b/packages/timenet/src/timenet/registry/lazy_read.py @@ -82,14 +82,14 @@ def put_dataset( shutil.rmtree(staging, ignore_errors=True) def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: - """Open one release without its payload sources, then read remote files on demand. + """Open a cached or remote release without attaching its payload sources. Args: - reference: The exact release. - manifest: The manifest of the release from this registry, if you already read it. + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle over a complete cached copy in :attr:`local_root`, else over remote files. + A handle to a complete copy in :attr:`local_root`, or to remote files read on demand. """ return open_cached_or_remote_release( self.local_root, reference, lambda ref: self._open_remote_version(ref, manifest=manifest) @@ -97,12 +97,12 @@ def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = Non @abstractmethod def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: - """Bind one remote release to a range-capable filesystem, without its payload sources. + """Open a remote release on a filesystem that serves range reads. Args: - reference: The exact release. - manifest: The manifest of the release, if the caller already read it from this registry. + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle without its payload sources. + A handle to the remote release without attached payload sources. """ diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 5555f1c51..ab4e8fb43 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -269,22 +269,21 @@ def damage(self, release: DatasetRef, manifest: Manifest | None = None) -> TimeN return None def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: - """Open one committed release as a local handle, without its payload sources. + """Open a local release without attaching its payload sources. - Opening reads only metadata. It checks that each file of the release is present with its - recorded size. + This checks that every file exists with its recorded size. Args: - reference: The exact release. - manifest: The manifest of the release, if the caller already read it from this registry. + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle rooted at the directory of the release over a :class:`pyarrow.fs.LocalFileSystem`. + A handle to the release's directory on the local filesystem. Raises: - TimeNetDatasetNotFoundError: If this registry has no committed manifest of the release. - TimeFFormatError: If the manifest declares another release. - TimeNetFileError: If a file of the release is missing or has another size than recorded. + TimeNetDatasetNotFoundError: If the registry lacks a committed manifest for the release. + TimeFFormatError: If the manifest declares a different release. + TimeNetFileError: If a file is missing or has an incorrect size. """ # ruff: ignore[docstring-extraneous-exception] - raised by get_manifest and the checks if manifest is None: manifest = self.get_manifest(reference.dataset_id, str(reference.version)) diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index 30b763061..ad093591c 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -32,7 +32,7 @@ class RemoteRegistry(LazyReadRegistry): - """Read and publish datasets through the hosted TimeNet HTTP service.""" + """A dataset registry backed by the hosted TimeNet HTTP service.""" def __init__( self, @@ -119,14 +119,14 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: return buffer def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: - """Bind one remote manifest to a range-capable filesystem. + """Open a remote release with on-demand file reads. Args: - reference: The exact release. - manifest: The manifest of the release, if the caller already fetched it from this registry. + reference: The dataset ID and exact version to open. + manifest: A manifest already fetched from this registry, or ``None`` to fetch it. Returns: - A handle without its payload sources. + A handle to the remote release without attached payload sources. """ if manifest is None: manifest = self.get_manifest(reference.dataset_id, str(reference.version)) diff --git a/packages/timenet/src/timenet/registry/s3.py b/packages/timenet/src/timenet/registry/s3.py index 326c0e8e8..4bdf97d5f 100644 --- a/packages/timenet/src/timenet/registry/s3.py +++ b/packages/timenet/src/timenet/registry/s3.py @@ -38,7 +38,7 @@ class S3Registry(LazyReadRegistry): - """Serves and publishes datasets under an ``s3://bucket/prefix`` root.""" + """A registry that reads and writes datasets under an ``s3://bucket/prefix`` root.""" _staging_prefix = "timenet-s3-" @@ -141,14 +141,14 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: return response["Body"] def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: - """Bind one remote manifest to Arrow's native S3 filesystem. + """Open a release through Arrow's S3 filesystem. Args: - reference: The exact release. - manifest: The manifest of the release, if the caller already fetched it from this registry. + reference: The dataset ID and exact version to open. + manifest: A manifest already fetched from this registry, or ``None`` to fetch it. Returns: - A handle without its payload sources. + A handle to the S3 release without attached payload sources. """ if manifest is None: manifest = self.get_manifest(reference.dataset_id, str(reference.version)) From a2a0fd14712dfd811ed0fa05404153dda838f69f Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 10:24:50 +0100 Subject: [PATCH 4/4] test: trim the tests of the read coordinator Drop tests that cannot fail or repeat other tests, and sharpen the rest. Part of TNET-157. --- packages/timenet/tests/registry/test_conformance.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/timenet/tests/registry/test_conformance.py b/packages/timenet/tests/registry/test_conformance.py index fea9190e2..a02a657a3 100644 --- a/packages/timenet/tests/registry/test_conformance.py +++ b/packages/timenet/tests/registry/test_conformance.py @@ -214,7 +214,6 @@ def no_read(*args, **kwargs): # No cache holds a copy of the child, so S3 and the remote registry open their remote files. version = backend.registry.open_release(child, manifest=manifest) assert version.manifest == manifest - assert version.payload_sources == {} # ---- exists -----------------------------------------------------------------------------------