diff --git a/packages/timenet-connectors/src/timenet_connectors/api.py b/packages/timenet-connectors/src/timenet_connectors/api.py index bb2458733..4674cdeda 100644 --- a/packages/timenet-connectors/src/timenet_connectors/api.py +++ b/packages/timenet-connectors/src/timenet_connectors/api.py @@ -24,31 +24,31 @@ def build( # ruff: ignore[too-many-arguments] fetch_from: "str | BaseRegistry | None" = None, build_parents: bool = False, ) -> Path: - # ruff: ignore[docstring-extraneous-exception] (raised by build_release, not directly here) + # ruff: ignore[docstring-extraneous-exception] (raised by the plan and its run, not directly here) """Build a dataset release in a local registry. - The build reuses an existing release unless you set ``force``. The connector runs in an environment - with its required packages, or in this interpreter if ``TIMENET_ISOLATION=off``. + The build reuses an existing release unless ``force`` is set. The connector runs in an environment + from its ``requirements.txt``, or in this interpreter if ``TIMENET_ISOLATION=off``. Args: - dataset_id: The connector's dataset ID (``org/name``). - version: The version to build, or ``None`` for the connector's declared version. - out: The output directory, or ``None`` to use the default local registry. - force: Rebuild the release even if it exists in the registry. - keep_cache: Keep downloaded source files after a successful build. - fetch_from: The registry, or its URI, to copy missing parent releases from. - build_parents: Build missing parent releases from their installed connectors. + dataset_id: The ``org/name`` ID declared in the connector's card. + version: The required card version, or ``None`` to accept any version. + out: The local registry directory, or ``None`` to use the default local registry. + force: Rebuild an existing release. + keep_cache: Keep the raw download cache after a successful build. + fetch_from: The registry or URI to copy missing parents from before trying builds. + build_parents: Build missing parents with installed connectors. Returns: - The release's version directory. + The version directory of the release. Raises: TimeNetDatasetNotFoundError: If no installed connector builds ``dataset_id``. - TimeNetVersionUnavailableError: If ``version`` differs from the connector's declared version. - TimeNetMissingParentError: If a required parent release is still missing. + TimeNetVersionUnavailableError: If the card's version differs from ``version``. + TimeNetMissingParentError: If a required release cannot be copied or built. """ from timenet.builders import BuildOptions, declared # ruff: ignore[import-outside-top-level] - from timenet.coordinator.build import build_release # ruff: ignore[import-outside-top-level] + from timenet.coordinator.build import plan_release_build, run_plan # ruff: ignore[import-outside-top-level] from timenet.errors import TimeNetMissingParentError # ruff: ignore[import-outside-top-level] from timenet.registry import ( # ruff: ignore[import-outside-top-level] LocalRegistry, @@ -56,24 +56,50 @@ def build( # ruff: ignore[too-many-arguments] open_registry, ) from timenet.registry._paths import release_dir # ruff: ignore[import-outside-top-level, import-private-name] + from timenet.sources import InstalledConnectors # ruff: ignore[import-outside-top-level] release = declared(dataset_id).release root = Path(out).expanduser() if out is not None else default_registry_path() + source = open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from + store = LocalRegistry(root) try: - build_release( + plan = plan_release_build( f"{dataset_id}@{version}" if version else dataset_id, - store=LocalRegistry(root), - fetch_from=open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from, - build_parents=build_parents, + store=store, + # TODO(adrian): Remove temporary glue which maps fetch_from and build_parents onto a + # list of sources until the producer API replaces this function. + parents_from=[ + *([source] if source is not None else []), + *([InstalledConnectors()] if build_parents else []), + ], force=force, - options=BuildOptions(keep_cache=keep_cache), ) + run_plan(plan, store, options=BuildOptions(keep_cache=keep_cache)) except TimeNetMissingParentError as error: - error.hint = "Copy it with fetch_from= or build it with build_parents=True." + error.hint = _missing_hint(fetch_from=fetch_from, build_parents=build_parents) raise return release_dir(root, release) +def _missing_hint(*, fetch_from: "str | BaseRegistry | None", build_parents: bool) -> str | None: + """Suggest options to obtain a missing release. + + Args: + fetch_from: The selected registry or URI, or ``None``. + build_parents: Whether parent builds are enabled. + + Returns: + A sentence naming unused options, or ``None`` if both are enabled. + """ + if fetch_from is None and not build_parents: + return "Copy it with fetch_from= or build it with build_parents=True." + if fetch_from is None: + return "Copy it with fetch_from=." + if not build_parents: + return "Build it with build_parents=True." + return None + + def load( dataset_id: str, version: str | None = None, diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py index 7b43b2d7f..a49ac0511 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py @@ -15,7 +15,7 @@ from timenet.cli.runner import run_cli from timenet.cli.ui import console, progress_renderer from timenet.config import settings -from timenet.coordinator.build import build_release +from timenet.coordinator.build import plan_release_build, run_plan from timenet.errors import ( TimeNetAccessError, TimeNetFileError, @@ -33,6 +33,7 @@ open_writable_registry, ) from timenet.registry._paths import release_dir # ruff: ignore[import-private-name] +from timenet.sources import InstalledConnectors from timenet.types.references import DatasetRef, is_latest, split_ref from timenet_connectors.builder.backend import ConnectorBuilder from timenet_connectors.discovery import known_ids @@ -160,7 +161,7 @@ def build( ), ) except TimeNetMissingParentError as error: - error.hint = "Fetch it with --fetch-from or build it with --build-parents." + error.hint = _missing_hint(fetch_from=fetch_from, build_parents=build_parents) raise except TimeNetFileError as error: error.hint = ( @@ -175,6 +176,27 @@ def build( typer.echo(result) +def _missing_hint(*, fetch_from: str | None, build_parents: bool) -> str | None: + """Suggest flags to obtain a missing release. + + Args: + fetch_from: The value of ``--fetch-from``, or ``None``. + build_parents: Whether ``--build-parents`` is set. + + Returns: + A sentence naming unused flags, or ``None`` if both are set. + """ + ways = [] + if fetch_from is None: + ways.append("fetch it with --fetch-from ") + if not build_parents: + ways.append("build it with --build-parents") + if not ways: + return None + hint = " or ".join(ways) + return f"{hint[0].upper()}{hint[1:]}." + + def _build_to( # ruff: ignore[too-many-arguments] dataset_id: str, target: WritableRegistry | Path, @@ -192,12 +214,12 @@ def _build_to( # ruff: ignore[too-many-arguments] dataset_id: The dataset ID. target: The local directory or remote registry to write to. fetch_from: The registry to copy missing parents from. ``None`` uses a remote output, if any. - build_parents: Build missing parents with their installed connectors. - force: Rebuild the release even if the local registry holds it. + build_parents: Build missing parents with installed connectors. + force: Rebuild an existing local release. options: Settings for each build. Returns: - The version directory for a local output, or the version string for a remote output. + The local version directory or the remote version string. Raises: BadParameter: If ``force`` is set and the remote output already holds the version. @@ -221,15 +243,19 @@ def _build_to( # ruff: ignore[too-many-arguments] fetch_from = fetch_from if fetch_from is not None else remote # A remote registry holds only open releases. So for a remote output, the build refuses another # release before any download, as a build into that registry would. - plan = build_release( + plan = plan_release_build( dataset_id, store=store, - fetch_from=fetch_from, - build_parents=build_parents, + # TODO(adrian): Remove temporary glue which maps --fetch-from and --build-parents onto a + # list of sources until the flags of the defaults replace them. + parents_from=[ + *([fetch_from] if fetch_from is not None else []), + *([InstalledConnectors()] if build_parents else []), + ], force=force, open_access_only=remote is not None, - options=options, ) + run_plan(plan, store, options=options) if remote is not None: copy_release(release, store, remote, progress=options.progress) return str(release.version) diff --git a/packages/timenet-connectors/tests/test_build_plan.py b/packages/timenet-connectors/tests/test_build_plan.py index 82b3b35ee..e61241255 100644 --- a/packages/timenet-connectors/tests/test_build_plan.py +++ b/packages/timenet-connectors/tests/test_build_plan.py @@ -8,22 +8,21 @@ from timenet import builders from timenet.connectors import BaseConnector -from timenet.coordinator.build import build_release -from timenet.coordinator.plan import Step, plan_parents +from timenet.coordinator.build import plan_release_build, run_plan +from timenet.coordinator.plan import BuildStep, CopyStep, Step from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset from timenet.engine import open_parents, run_pipeline from timenet.errors import ( TimeFFormatError, TimeFValidationError, TimeNetAccessError, - TimeNetDatasetNotFoundError, TimeNetFileError, - TimeNetInvalidCardError, TimeNetMissingParentError, ) from timenet.reader import TimeFReader from timenet.registry import LocalRegistry, WritableRegistry from timenet.registry._paths import release_dir +from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset from timenet.types import Access, TimeSeriesSpec, ureg from timenet.types.access import AccessRequirement @@ -44,6 +43,17 @@ runner = CliRunner() +def _parent_steps(dataset_id: str, store: LocalRegistry, **options) -> list[Step]: + """Plan the build of ``dataset_id`` and return each step but its own build.""" + steps = plan_release_build(dataset_id, store=store, **options).steps + return [step for step in steps if not (step.action == "build" and step.release.dataset_id == dataset_id)] + + +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) + + class _CardBackend: """A backend of another package, not ConnectorBuilder: it serves cards by dataset ID, and the plan never builds through it.""" @@ -98,20 +108,34 @@ def _install_cards(monkeypatch, **parents: dict[str, DatasetRef]) -> dict: @pytest.mark.parametrize( ("grand_in", "expected"), [ - ("output", [Step(PARENT, "build")]), - ("origin", [Step(GRAND, "copy"), Step(PARENT, "build", depends_on=(GRAND,))]), - ("nowhere", [Step(GRAND, "build"), Step(PARENT, "build", depends_on=(GRAND,))]), + ("output", lambda origin: [_build(PARENT)]), + ( + "origin", + lambda origin: [ + CopyStep(GRAND, origin), + _build(PARENT, depends_on=(GRAND,)), + ], + ), + ( + "nowhere", + lambda origin: [ + _build(GRAND), + _build(PARENT, depends_on=(GRAND,)), + ], + ), ], ) -def test_plan_uses_fetches_or_builds_each_parent_once_before_its_children(tmp_path, monkeypatch, grand_in, expected): +def test_plan_copies_or_builds_each_parent_once_before_its_children(tmp_path, monkeypatch, grand_in, expected): _install_cards(monkeypatch, child={"parent": PARENT, "grand": GRAND}, parent={"grand": GRAND}) registries = {name: LocalRegistry(tmp_path / name) for name in ("output", "origin")} if grand_in in registries: registries[grand_in].put_dataset(make_dataset()) - steps = plan_parents(CHILD.dataset_id, registries["output"], fetch_from=registries["origin"], build_parents=True) + steps = _parent_steps( + CHILD.dataset_id, registries["output"], parents_from=[registries["origin"], InstalledConnectors()] + ) - assert steps == expected + assert steps == expected(registries["origin"]) @pytest.mark.parametrize( @@ -123,20 +147,14 @@ def test_plan_uses_fetches_or_builds_each_parent_once_before_its_children(tmp_pa TimeNetMissingParentError, r"^parent release test/parent@1\.0\.0 of test/child@1\.0\.0 is not in local registry", ), - ( - {"child": {"parent": DatasetRef.parse("test/parent@2.0.0")}, "parent": {}}, - {"build_parents": True}, - TimeFValidationError, - "not that exact version", - ), ( {"child": {"parent": PARENT}, "parent": {"child": CHILD}}, - {"build_parents": True}, + {"parents_from": [InstalledConnectors()]}, TimeFValidationError, "cycle: test/child@1.0.0 -> test/parent@1.0.0 -> test/child@1.0.0", ), ], - ids=["missing", "other-version", "cycle"], + ids=["missing", "cycle"], ) # ruff: ignore[too-many-arguments, too-many-positional-arguments] def test_plan_rejects_unobtainable_parents_before_anything_runs(tmp_path, monkeypatch, cards, options, error, match): @@ -144,7 +162,7 @@ def test_plan_rejects_unobtainable_parents_before_anything_runs(tmp_path, monkey output = LocalRegistry(tmp_path / "output") with pytest.raises(error, match=match): - plan_parents(CHILD.dataset_id, output, **options) + _parent_steps(CHILD.dataset_id, output, **options) _SPEC = TimeSeriesSpec(spec_type="sine", name="Sine", unit_value=ureg.dimensionless) @@ -191,21 +209,15 @@ def test_a_declared_parent_needs_only_its_direct_payload_sources(tmp_path, monke release_dir(origin.root, SOURCE).parent.mkdir(parents=True) shutil.move(release_dir(output.root, SOURCE), release_dir(origin.root, SOURCE)) - steps = plan_parents(CHILD.dataset_id, output, fetch_from=origin) - - # DECLARED opens with SOURCE, its direct payload source, which a copy of DECLARED brings. So the - # plan does not name GRAND, SOURCE's own source. - assert steps == ([] if where == "output" else [Step(DECLARED, "copy")]) - if where == "source-in-origin": - # The output holds DECLARED, so the run copies SOURCE with its own payload sources. That - # includes GRAND, which no registry holds, so the copy fails before it copies anything. - with pytest.raises( - TimeNetMissingParentError, match=r"^payload source timenet/hello-world@1\.0\.0 of test/source" - ): - build_release(CHILD.dataset_id, store=output, fetch_from=origin) - assert child_builds == [] - return - build_release(CHILD.dataset_id, store=output, fetch_from=origin) + # DECLARED opens with SOURCE, its direct payload source. So the plan needs neither GRAND, which + # SOURCE reads and no registry holds, nor a step for SOURCE where the output holds it. + steps = { + "output": [], + "origin": [CopyStep(DECLARED, origin)], + "source-in-origin": [CopyStep(SOURCE, origin, with_payload_sources=False)], + } + assert _parent_steps(CHILD.dataset_id, output, parents_from=[origin]) == steps[where] + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[origin]), output) for reference in (DECLARED, SOURCE): assert output.exists(reference.dataset_id, str(reference.version)) assert not output.exists(GRAND.dataset_id, str(GRAND.version)) @@ -221,9 +233,9 @@ def test_a_payload_source_that_is_later_a_declared_parent_brings_its_own_payload _declared_without_its_sources_source(output) pins = output.get_manifest(SOURCE.dataset_id).payload_sources[SourceKey(str(GRAND))] - steps = plan_parents(CHILD.dataset_id, output, build_parents=True) + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) - assert steps == [Step(GRAND, "build", pinned_by={SOURCE: pins})] + assert steps == [_build(GRAND, pinned_by={SOURCE: pins})] def test_the_plan_reads_each_card_from_the_backend_that_claims_the_dataset(tmp_path, monkeypatch): @@ -233,26 +245,19 @@ def test_the_plan_reads_each_card_from_the_backend_that_claims_the_dataset(tmp_p # which only the second backend knows. install_backends(monkeypatch, ConnectorBuilder(), _CardBackend(cards)) - steps = plan_parents(CHILD.dataset_id, LocalRegistry(tmp_path), build_parents=True) + steps = _parent_steps(CHILD.dataset_id, LocalRegistry(tmp_path), parents_from=[InstalledConnectors()]) - assert steps == [Step(GRAND, "build"), Step(PARENT, "build", depends_on=(GRAND,))] + assert steps == [ + _build(GRAND), + _build(PARENT, depends_on=(GRAND,)), + ] cards[PARENT.dataset_id] = cards[PARENT.dataset_id].model_copy(update={"dataset_version": "2.0.0"}) - with pytest.raises(TimeFValidationError, match=r"test/parent@1\.0\.0 is missing .* builds version 2\.0\.0"): - plan_parents(CHILD.dataset_id, LocalRegistry(tmp_path), build_parents=True) - - -def test_a_parent_whose_card_cannot_be_read_fails_the_plan_with_the_card_error(tmp_path, monkeypatch): - (tmp_path / "dataset.yaml").write_text("dataset_id: [unclosed\n") - # The in-repo backend claims PARENT, whose card cannot be read. The second backend serves CHILD. - monkeypatch.setattr(backend_module, "known_ids", lambda: [PARENT.dataset_id]) - monkeypatch.setattr(backend_module, "connector_dir", lambda dataset_id: tmp_path) - install_backends(monkeypatch, ConnectorBuilder(), _CardBackend(_cards(child={"parent": PARENT}))) - - with pytest.raises(TimeNetInvalidCardError, match="could not read dataset card") as caught: - plan_parents(CHILD.dataset_id, LocalRegistry(tmp_path / "output"), build_parents=True) - - # The plan does not report a card that cannot be read as a missing connector. - assert not isinstance(caught.value, (LookupError, TimeNetDatasetNotFoundError)) + with pytest.raises( + TimeNetMissingParentError, + match=r"^parent release test/parent@1\.0\.0 of test/child@1\.0\.0 is not in local registry .* or " + r"installed connectors \(test/parent builds 2\.0\.0\)\.$", + ): + _parent_steps(CHILD.dataset_id, LocalRegistry(tmp_path), parents_from=[InstalledConnectors()]) class _Grand(BaseConnector[str]): @@ -275,18 +280,6 @@ def _parent_without_its_source(output: LocalRegistry) -> None: shutil.rmtree(release_dir(output.root, GRAND)) -def test_a_missing_payload_source_names_the_release_that_reads_it(tmp_path, monkeypatch): - _install_cards(monkeypatch, child={"parent": PARENT}, parent={"parent": GRAND}) - output = LocalRegistry(tmp_path / "output") - _parent_without_its_source(output) - - with pytest.raises(TimeNetMissingParentError) as caught: - plan_parents(CHILD.dataset_id, output) - - assert (caught.value.missing, caught.value.needed_by, caught.value.payload_source) == (GRAND, PARENT, True) - assert str(caught.value).startswith("payload source timenet/hello-world@1.0.0 of test/parent@1.0.0 is not in") - - @pytest.mark.parametrize("backend", [ValuesBackend.PARQUET, ValuesBackend.ZARR]) def test_a_pinned_source_is_rebuilt_in_staging_and_published_only_if_it_matches(tmp_path, monkeypatch, backend): _install_cards(monkeypatch, child={"parent": PARENT}, parent={"parent": GRAND}) @@ -295,9 +288,9 @@ def test_a_pinned_source_is_rebuilt_in_staging_and_published_only_if_it_matches( _parent_without_its_source(output) pins = output.get_manifest(PARENT.dataset_id).payload_sources[SourceKey(str(GRAND))] - steps = plan_parents(CHILD.dataset_id, output, build_parents=True) + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) - assert steps == [Step(GRAND, "build", pinned_by={PARENT: pins})] + assert steps == [_build(GRAND, pinned_by={PARENT: pins})] built = [] def child(release, store_root, *, pinned_by, **_): @@ -311,13 +304,13 @@ def child(release, store_root, *, pinned_by, **_): monkeypatch.setattr(backend_module, "run_isolated_build", child) if backend == ValuesBackend.PARQUET: - build_release(CHILD.dataset_id, store=output, build_parents=True) + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[InstalledConnectors()]), output) assert output.exists(GRAND.dataset_id, str(GRAND.version)) assert built == [GRAND, CHILD] return # A Zarr rebuild writes other files than the Parquet files PARENT reads, so nothing is published. with pytest.raises(TimeFFormatError, match=r"Nothing was published\. Fetch the exact release .* or rebuild"): - build_release(CHILD.dataset_id, store=output, build_parents=True) + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[InstalledConnectors()]), output) assert built == [GRAND] assert not output.exists(GRAND.dataset_id, str(GRAND.version)) assert not list(release_dir(output.root, GRAND).parent.glob("*")) @@ -337,9 +330,46 @@ def test_a_payload_source_that_two_releases_read_is_built_against_the_pins_of_bo for reader in (PARENT, sibling) } - steps = plan_parents(CHILD.dataset_id, output, build_parents=True) + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) + + assert steps == [_build(GRAND, pinned_by=pins)] + + +def _repin(registry: LocalRegistry, reader: DatasetRef) -> None: + """Give the first file that ``reader`` pins of GRAND another checksum, as if another build of GRAND wrote it.""" + path = release_dir(registry.root, reader) / "manifest.json" + manifest = registry.get_manifest(reader.dataset_id) + pinned = manifest.payload_sources[SourceKey(str(GRAND))] + first, *rest = pinned.files + files = (first.model_copy(update={"checksum": "sha256:" + "0" * 64}), *rest) + sources = {**manifest.payload_sources, SourceKey(str(GRAND)): pinned.model_copy(update={"files": files})} + path.write_text(manifest.model_copy(update={"payload_sources": sources}).to_json()) + + +def test_two_readers_that_pin_one_file_differently_fail_the_plan_of_its_build(tmp_path, monkeypatch): + sibling = DatasetRef.parse("test/sibling@1.0.0") + _install_cards(monkeypatch, child={"parent": PARENT, "sibling": sibling}) + output = LocalRegistry(tmp_path / "output") + output.put_dataset(make_dataset()) + for reader in (PARENT, sibling): + _child_of(output, reader, GRAND, make_dataset().records[0].id) + _repin(output, sibling) + shutil.rmtree(release_dir(output.root, GRAND)) - assert steps == [Step(GRAND, "build", pinned_by=pins)] + # No build of GRAND can match both readers, so the plan fails before any step runs. + with pytest.raises(TimeFFormatError, match=f"pin .* of {GRAND} as different files, so no build"): + plan_release_build(CHILD.dataset_id, store=output, parents_from=[InstalledConnectors()]) + + +def test_a_held_payload_source_that_breaks_a_pin_fails_the_plan(tmp_path, monkeypatch): + _install_cards(monkeypatch, child={"parent": PARENT}) + output = LocalRegistry(tmp_path / "output") + output.put_dataset(make_dataset()) + _child_of(output, PARENT, GRAND, make_dataset().records[0].id) + _repin(output, PARENT) + + with pytest.raises(TimeFFormatError, match=f"payload source {GRAND} differs from the release {PARENT}"): + plan_release_build(CHILD.dataset_id, store=output) def test_a_built_release_needs_only_its_payload_sources(tmp_path, monkeypatch): @@ -349,20 +379,23 @@ def test_a_built_release_needs_only_its_payload_sources(tmp_path, monkeypatch): pins = output.get_manifest(PARENT.dataset_id).payload_sources[SourceKey(str(GRAND))] with pytest.raises(TimeNetMissingParentError, match=r"^payload source timenet/hello-world@1\.0\.0 of test/parent"): - plan_parents(PARENT.dataset_id, output) + _parent_steps(PARENT.dataset_id, output) + connectors = [InstalledConnectors()] # The source that PARENT reads builds against its pins. - assert plan_parents(PARENT.dataset_id, output, build_parents=True) == [ - Step(GRAND, "build", pinned_by={PARENT: pins}) + assert _parent_steps(PARENT.dataset_id, output, parents_from=connectors) == [ + _build(GRAND, pinned_by={PARENT: pins}) ] # A rebuild of PARENT needs its declared parents instead, and pins the new files. - assert plan_parents(PARENT.dataset_id, output, build_parents=True, force_build=True) == [Step(GRAND, "build")] + assert _parent_steps(PARENT.dataset_id, output, parents_from=connectors, force=True) == [_build(GRAND)] # A registry that holds the source fills the gap: the local PARENT is reused and GRAND is copied. origin = LocalRegistry(tmp_path / "origin") origin.put_dataset(make_dataset()) - assert plan_parents(PARENT.dataset_id, output, fetch_from=origin) == [Step(PARENT, "copy")] + assert _parent_steps(PARENT.dataset_id, output, parents_from=[origin]) == [ + CopyStep(GRAND, origin, with_payload_sources=False) + ] -def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkeypatch, child_builds): +def test_a_damaged_parent_is_copied_again_but_never_built_over(tmp_path, monkeypatch, child_builds): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) output, origin = LocalRegistry(tmp_path / "output"), LocalRegistry(tmp_path / "origin") parent = make_dataset() @@ -372,12 +405,14 @@ def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkey (release_dir(output.root, PARENT) / "control.duckdb").write_bytes(b"cut short") with pytest.raises(TimeNetFileError, match=r"'control\.duckdb' has 9 bytes, expected [\d,]+ bytes\.$") as caught: - plan_parents(CHILD.dataset_id, output, build_parents=True) + _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) assert caught.value.release == PARENT - assert plan_parents(CHILD.dataset_id, output, fetch_from=origin) == [Step(PARENT, "copy")] - build_release(CHILD.dataset_id, store=output, fetch_from=origin) - assert plan_parents(CHILD.dataset_id, output) == [] + assert _parent_steps(CHILD.dataset_id, output, parents_from=[origin, InstalledConnectors()]) == [ + CopyStep(PARENT, origin) + ] + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[origin]), output) + assert _parent_steps(CHILD.dataset_id, output) == [] assert child_builds == [(CHILD.dataset_id, False)] @@ -413,21 +448,17 @@ def put_release(self, from_directory): _GATED = {"access": Access.CREDENTIALED, "access_url": "https://example.org/dua"} -@pytest.mark.parametrize("gated", ["child-card", "held-parent", "parent-card"]) -def test_open_access_only_refuses_a_plan_that_visits_a_release_that_is_not_open(tmp_path, monkeypatch, gated): +def test_open_access_only_refuses_a_parent_to_build_whose_card_is_not_open(tmp_path, monkeypatch): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) - gated_ref = CHILD if gated == "child-card" else PARENT - cards[gated_ref.dataset_id] = cards[gated_ref.dataset_id].model_copy(update=_GATED) + cards[PARENT.dataset_id] = cards[PARENT.dataset_id].model_copy(update=_GATED) store = LocalRegistry(tmp_path / "store") - if gated == "held-parent": - store.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id])) with pytest.raises(TimeNetAccessError) as caught: - plan_parents(CHILD.dataset_id, store, build_parents=True, open_access_only=True) + _parent_steps(CHILD.dataset_id, store, parents_from=[InstalledConnectors()], open_access_only=True) error = caught.value - requirement = AccessRequirement(gated_ref, Access.CREDENTIALED, _GATED["access_url"]) - assert (error.dataset, error.access, error.requirements) == (gated_ref, Access.CREDENTIALED, (requirement,)) + requirement = AccessRequirement(PARENT, Access.CREDENTIALED, _GATED["access_url"]) + assert (error.dataset, error.access, error.requirements) == (PARENT, Access.CREDENTIALED, (requirement,)) def test_open_access_only_names_the_access_page_of_a_held_parent_from_its_lineage(tmp_path, monkeypatch): @@ -440,7 +471,7 @@ def test_open_access_only_names_the_access_page_of_a_held_parent_from_its_lineag store.put_dataset(held) with pytest.raises(TimeNetAccessError) as caught: - plan_parents(CHILD.dataset_id, store, build_parents=True, open_access_only=True) + _parent_steps(CHILD.dataset_id, store, parents_from=[InstalledConnectors()], open_access_only=True) assert (caught.value.dataset, caught.value.access) == (PARENT, Access.CREDENTIALED) assert caught.value.requirements == (AccessRequirement(GRAND, Access.CREDENTIALED, _GATED["access_url"]),) @@ -554,39 +585,6 @@ def test_load_copies_a_missing_payload_source_through_fetch_from(tmp_path, monke assert len(dataset.records) == 2 -class _NoDownloads(_Hosted): - """A registry that is not local and fails the test if a file is read from it.""" - - def open_file(self, dataset_id, version, relpath): - pytest.fail(f"downloaded {dataset_id}@{version}/{relpath}") - - -@pytest.mark.parametrize("page", ["own", "ancestor"]) -def test_a_fetch_of_a_release_that_is_not_open_fails_before_any_download(tmp_path, monkeypatch, page): - cards = _install_cards(monkeypatch, child={"grand": GRAND, "parent": PARENT}, parent={}) - origin = _NoDownloads(tmp_path / "origin") - # GRAND is open and would be fetched first. - origin.put_dataset(make_dataset()) - ancestor = DatasetRef.parse("test/ancestor@1.0.0") - if page == "own": - parent, requirement = TimeFDataset(metadata=cards[PARENT.dataset_id].model_copy(update=_GATED)), PARENT - else: - # The card of PARENT stays open. Its lineage holds an ancestor, which is credentialed. - parent = TimeFDataset(metadata=cards[PARENT.dataset_id]) - card = make_dataset().metadata.model_copy(update={"dataset_id": ancestor.dataset_id, **_GATED}) - parent._import_state.lineage[str(ancestor)] = ParentDatasetCard(metadata=card) - requirement = ancestor - origin.put_dataset(parent) - output = LocalRegistry(tmp_path / "out") - - with pytest.raises(TimeNetAccessError) as caught: - build_release(CHILD.dataset_id, store=output, fetch_from=origin) - - assert caught.value.dataset == PARENT - assert caught.value.requirements == (AccessRequirement(requirement, Access.CREDENTIALED, _GATED["access_url"]),) - assert not output.exists(GRAND.dataset_id, str(GRAND.version)) - - def test_a_pinned_parent_build_hands_its_pins_to_the_isolated_child(tmp_path, monkeypatch): cards = _install_cards(monkeypatch, parent={"parent": GRAND}) output = LocalRegistry(tmp_path / "output") @@ -762,3 +760,129 @@ def test_force_is_a_usage_error_if_the_remote_out_holds_the_version(tmp_path, mo assert "a published version never changes" in " ".join(output.replace("│", " ").split()) assert calls == [(CHILD.dataset_id, False)] assert _files(local.root) == held + + +def test_the_plan_drops_the_store_from_parents_from(tmp_path, monkeypatch): + _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) + store = LocalRegistry(tmp_path / "store") + + # The list holds the store itself and another registry over its directory. The plan drops both, + # so the error names the store once. + with pytest.raises(TimeNetMissingParentError) as missing: + _parent_steps(CHILD.dataset_id, store, parents_from=[store, LocalRegistry(store.root)]) + + assert missing.value.searched == str(store) + + +def test_a_parent_that_a_registry_lists_without_serving_is_built_by_the_connectors(tmp_path, monkeypatch): + cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) + hosted = _Hosted(tmp_path / "hosted") + hosted.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id].model_copy(update=_GATED))) + output = LocalRegistry(tmp_path / "out") + + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[hosted, InstalledConnectors()]) + + assert steps == [_build(PARENT)] + # Without the connectors, the plan names the access page of the release. + with pytest.raises(TimeNetAccessError) as caught: + _parent_steps(CHILD.dataset_id, output, parents_from=[hosted]) + assert caught.value.requirements == (AccessRequirement(PARENT, Access.CREDENTIALED, _GATED["access_url"]),) + + +@pytest.mark.parametrize("answer", ["not-found", "not-served"]) +def test_a_damaged_copy_in_the_output_stops_the_plan_if_no_registry_serves_the_release(tmp_path, monkeypatch, answer): + cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) + output, hosted = LocalRegistry(tmp_path / "output"), _Hosted(tmp_path / "hosted") + parent = make_dataset() + parent._metadata = cards[PARENT.dataset_id] + output.put_dataset(parent) + (release_dir(output.root, PARENT) / "control.duckdb").write_bytes(b"cut short") + if answer == "not-served": + hosted.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id].model_copy(update=_GATED))) + + # The damage, not the missing release or its access page, is what the caller must fix. + with pytest.raises(TimeNetFileError) as caught: + _parent_steps(CHILD.dataset_id, output, parents_from=[hosted]) + + assert caught.value.release == PARENT + + +def test_the_requested_build_waits_for_a_payload_source_of_a_held_parent(tmp_path, monkeypatch): + _install_cards(monkeypatch, child={"declared": DECLARED}) + output, origin = LocalRegistry(tmp_path / "output"), LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + _child_of(origin, SOURCE, GRAND, make_dataset().records[0].id) + _child_of(origin, DECLARED, SOURCE, 2) + # The output holds DECLARED, but not SOURCE, whose files DECLARED reads. + shutil.copytree(release_dir(origin.root, DECLARED), release_dir(output.root, DECLARED)) + + plan = plan_release_build(CHILD.dataset_id, store=output, parents_from=[origin]) + + # DECLARED has no step, so the build of CHILD waits for the step of SOURCE behind it. + assert plan.steps == ( + CopyStep(SOURCE, origin, with_payload_sources=False), + _build(CHILD, depends_on=(SOURCE,)), + ) + + +@pytest.mark.parametrize( + ("entry", "hint"), [("cli", "Build it with --build-parents."), ("api", "Copy it with fetch_from=.")] +) +def test_the_hint_names_only_the_option_that_the_build_did_not_use(tmp_path, monkeypatch, entry, hint): + # No registry holds the parent, and no installed connector builds it. The CLI build uses + # --fetch-from, and the API build uses build_parents=True. + cards = _install_cards(monkeypatch, child={"parent": PARENT}) + monkeypatch.setattr(cli_module, "known_ids", lambda: list(cards)) + monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) + out = tmp_path / "out" + + if entry == "cli": + options = ["--fetch-from", str(tmp_path / "origin")] + error = runner.invoke(app, ["build", CHILD.dataset_id, "--out", str(out), *options]).exception + else: + with pytest.raises(TimeNetMissingParentError) as caught: + timenet_connectors.build(CHILD.dataset_id, out=out, build_parents=True) + error = caught.value + + assert isinstance(error, TimeNetMissingParentError) + assert error.hint == hint + + +def _count_reads(monkeypatch, registry) -> dict[str, int]: + """Count each manifest that is read from a registry, by release.""" + 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_plan_reads_each_manifest_once_from_each_registry(tmp_path, monkeypatch, child_builds): + # SOURCE is a declared parent and a payload source of DECLARED, so the walk meets it twice. The + # registry answers for SOURCE first as the payload source. + parents = {"declared": DECLARED, "source": SOURCE} + _install_cards(monkeypatch, child=parents) + layout = LocalRegistry(tmp_path / "hosted") + layout.put_dataset(make_dataset()) + _child_of(layout, SOURCE, GRAND, make_dataset().records[0].id) + _child_of(layout, DECLARED, SOURCE, 2) + hosted, output = _Hosted(layout.root), LocalRegistry(tmp_path / "output") + hosted_reads, output_reads = _count_reads(monkeypatch, hosted), _count_reads(monkeypatch, output) + + plan = plan_release_build(CHILD.dataset_id, store=output, parents_from=[hosted]) + + assert plan.steps[:-1] == tuple(CopyStep(reference, hosted) for reference in parents.values()) + # The registry that is not local answers through its manifest, and the plan reads that manifest + # only once. + assert hosted_reads == {str(reference): 1 for reference in (DECLARED, SOURCE, GRAND)} + # The output is read for the requested release too, which it does not hold. + assert output_reads == {str(reference): 1 for reference in (CHILD, DECLARED, SOURCE)} + hosted_reads.clear() + run_plan(plan, output) + # Each copy reads the manifests of its release and of the payload sources that it brings once + # more, from the registry as it is when the copy runs. SOURCE is in both copies. + assert hosted_reads == {str(DECLARED): 1, str(SOURCE): 2, str(GRAND): 1} + assert child_builds == [(CHILD.dataset_id, False)] diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index 6f4234227..cf2dca4d2 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -16,7 +16,7 @@ from timenet.builders import installed_builder from timenet.config import settings -from timenet.coordinator.build import build_release +from timenet.coordinator.build import plan_release_build, run_plan from timenet.dataset import TimeFDataset from timenet.errors import ( TimeFValidationError, @@ -32,6 +32,7 @@ from timenet.registry.base import ProgressCallback from timenet.registry.dependencies import prepare_local_copy from timenet.registry.version import DatasetVersion +from timenet.sources import InstalledConnectors from timenet.types import DatasetMetadata, Domain, License, Task from timenet.types.access import require_open_access from timenet.types.references import DatasetRef, is_latest, split_ref @@ -393,19 +394,18 @@ def _missing_hint(*, auto_build: bool, fetch_from: BaseRegistry | None) -> str | def _prepare( registry: LocalRegistry, dataset_id: str, version: str | None, *, auto_build: bool, fetch_from: BaseRegistry | None ) -> DatasetVersion: - """Prepare a local dataset version and its payload sources. + """Prepare a local dataset version and the releases that supply its payload files. - This function reuses local releases, tries ``fetch_from``, then builds missing releases if - ``auto_build`` is true. Each rebuilt payload source must match the version and file pins the - dataset requires. + The function copies missing releases from ``fetch_from`` or builds them when ``auto_build`` is set. + A payload source build must reproduce the exact version and pinned files. Returns: The release with each payload source attached. Raises: - TimeNetMissingParentError: If copying and building cannot provide a missing payload source. - TimeNetInvalidCardError: If reading a connector's card fails. - """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain, claim and build_release + TimeNetMissingParentError: If a missing payload source cannot be copied or built. + TimeNetInvalidCardError: If a connector's card cannot be read. + """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain, claim and the build try: manifest = registry.get_manifest(dataset_id, version) except TimeNetDatasetNotFoundError as miss: @@ -425,11 +425,15 @@ def _prepare( if found is None or found.release != source: raise pins = next(pinned for pinned in manifest.payload_sources.values() if pinned.release == source) - # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto the - # coordinator until the client takes a list of sources. - build_release( - str(source), store=registry, fetch_from=fetch_from, build_parents=True, pinned_by={reference: pins} + # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto a list + # of sources until the client takes the list itself. + plan = plan_release_build( + str(source), + store=registry, + parents_from=[*([fetch_from] if fetch_from is not None else []), InstalledConnectors()], + pinned_by={reference: pins}, ) + run_plan(plan, registry) built.add(source) @@ -442,18 +446,18 @@ def _obtain( # ruff: ignore[too-many-arguments] auto_build: bool, fetch_from: BaseRegistry | None, ) -> Manifest: - """Get a missing version's manifest from another registry or a local build. + """Get a missing version's manifest from ``fetch_from``, or build the version locally. - Try ``fetch_from`` first, then build if ``auto_build`` is true. + A local build needs ``auto_build`` and an installed connector for the dataset. Returns: - The manifest from ``fetch_from`` without copying its files, or the local manifest after a build. + The manifest from ``fetch_from`` without its files, or the local manifest after a build. Raises: TimeNetDatasetNotFoundError: If neither ``fetch_from`` nor a connector provides the version. TimeNetVersionUnavailableError: If the connector declares a different version. - TimeNetInvalidCardError: If reading the connector's card fails. - """ # ruff: ignore[docstring-extraneous-exception] - raised by build_release + TimeNetInvalidCardError: If the connector's card cannot be read. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the build if fetch_from is not None: try: return fetch_from.get_manifest(dataset_id, version) @@ -464,14 +468,14 @@ def _obtain( # ruff: ignore[too-many-arguments] if hint is None: raise miss raise TimeNetDatasetNotFoundError(f"{miss}. {hint}") from miss - # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto the coordinator - # until the client takes a list of sources. The build prepares the declared parents of the - # version in the same order: reuse, copy, then build. Sentinels such as "latest" are not pins, - # so a fresh build satisfies them. - build_release( + # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto a list of + # sources until the client takes the list itself. The build prepares the declared parents of + # the version in the same order: reuse, copy, then build. Sentinels such as "latest" are not + # pins, so a fresh build satisfies them. + plan = plan_release_build( dataset_id if is_latest(version) else f"{dataset_id}@{version}", store=registry, - fetch_from=fetch_from, - build_parents=True, + parents_from=[*([fetch_from] if fetch_from is not None else []), InstalledConnectors()], ) + run_plan(plan, registry) return registry.get_manifest(dataset_id, version) diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py index 1fab16f92..9aa22fdd5 100644 --- a/packages/timenet/src/timenet/coordinator/build.py +++ b/packages/timenet/src/timenet/coordinator/build.py @@ -1,146 +1,106 @@ -"""Prepare parent releases and build a dataset into a local registry.""" +"""Plan and run release builds with dependencies first.""" -from collections.abc import Mapping +from collections.abc import Mapping, Sequence +from typing import assert_never from timenet.builders import BuildOptions, declared -from timenet.coordinator.plan import Plan, Step, plan_parents -from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError -from timenet.manifest import Manifest, PayloadSource -from timenet.progress import ProgressCallback, StepFinished, StepStarted +from timenet.coordinator.plan import BuildStep, CopyStep, Plan, plan_steps +from timenet.manifest import PayloadSource +from timenet.progress import StepFinished, StepStarted from timenet.registry import copy_release -from timenet.registry._paths import release_dir -from timenet.registry.base import BaseRegistry from timenet.registry.dependencies import prepare_local_copy from timenet.registry.local import LocalRegistry -from timenet.registry.shared import complete_copy +from timenet.sources import ReleaseSource from timenet.types.references import DatasetRef, is_latest, split_ref -def build_release( # ruff: ignore[too-many-arguments] - dataset: str, +def plan_release_build( # ruff: ignore[too-many-arguments] + dataset_id: str, *, store: LocalRegistry, - fetch_from: BaseRegistry | None = None, - build_parents: bool = False, + parents_from: Sequence[ReleaseSource] = (), force: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, open_access_only: bool = False, - options: BuildOptions | None = None, ) -> Plan: - """Prepare a release's dependencies and build it into a local registry. + """Plan a release build in a local registry. - Copy missing dependencies from ``fetch_from``, or build them if ``build_parents`` is set. - Reuse an existing release unless ``force`` is set. + The plan reuses an intact release unless ``force`` is true. + Planning leaves the registry unchanged. Args: - dataset: A dataset ID such as ``org/name``, or an exact release such as ``org/name@1.2.0``. - store: The local registry to build into. - fetch_from: The registry to copy missing parents from, or ``None`` to copy nothing. - build_parents: Build missing parents with installed connectors. - force: Rebuild the requested release and prepare its parents even if the release exists. - pinned_by: Releases that already read the requested release's files, with the pins the new build must match. - open_access_only: Require ``access: open`` for all releases and their lineage before running the plan. - options: Build settings, or ``None`` for defaults. + dataset_id: A dataset ID such as ``org/name``, or an exact release such as ``org/name@1.2.0``. + store: The local registry to write to. + parents_from: Sources to copy or build missing dependencies from, in preference order. + force: Plan a forced rebuild and its parents even if the requested release already exists. + pinned_by: File pins grouped by dependent release that the build must preserve. + open_access_only: Require ``access: open`` for every needed release. Returns: - The completed plan, with steps in execution order. + A plan with steps in execution order. Raises: - TimeNetDatasetNotFoundError: If no installed backend builds the dataset or a missing parent. - TimeNetVersionUnavailableError: If the connector declares a different version. - TimeNetVersionConflictError: If ``store`` holds a different copy of a release being copied. - TimeFValidationError: If the dataset reference, backend, or dependency plan is invalid. - TimeNetMissingParentError: If a parent is missing and ``build_parents`` is not set. - TimeNetAccessError: If a release or its lineage lacks required open access. - TimeNetInvalidCardError: If an installed connector's card cannot be read. - TimeNetFileError: If ``store`` holds a damaged parent and no replacement is available. - TimeFFormatError: If a build omits or changes a pinned file. - TimeNetBuildError: If a build fails. - """ # ruff: ignore[docstring-extraneous-exception] - raised by the plan, the copy and the backends - options = options or BuildOptions() - dataset_id, version = split_ref(dataset) - # A connector builds only the version of its card, so "latest" asks for that version. - found = declared(dataset_id, None if is_latest(version) else version) - card, requested = found.card, found.release - steps = plan_parents( - dataset_id, + TimeNetDatasetNotFoundError: If no installed connector builds the dataset. + TimeNetVersionUnavailableError: If the requested version differs from the connector's card. + TimeNetMissingParentError: If no source provides a parent or a payload source. + """ # ruff: ignore[docstring-extraneous-exception] - raised by declared and the plan + dataset_id, version = split_ref(dataset_id) + _, steps = plan_steps( + # A connector builds only the version of its card, so "latest" asks for that version. + declared(dataset_id, None if is_latest(version) else version), store, - fetch_from=fetch_from, - build_parents=build_parents, + parents_from=parents_from, force_build=force, open_access_only=open_access_only, + pinned_by=pinned_by, ) - if force or not store.exists(requested.dataset_id, str(requested.version)): - planned = {step.release for step in steps} - depends_on = tuple(parent for parent in card.parents.values() if parent in planned) - steps.append(Step(requested, "build", depends_on=depends_on, pinned_by=pinned_by or {})) - if fetch_from is None and any(step.action == "copy" for step in steps): - raise TimeFValidationError(f"the plan of {requested} copies a release, but has no registry to copy from") - for step in steps: - if step.action == "build": - _build(step, store, force=force and step.release == requested, options=options) - elif fetch_from is not None: - _copy(step.release, fetch_from, store, progress=options.progress) return Plan(tuple(steps)) -def _copy( - release: DatasetRef, fetch_from: BaseRegistry, store: LocalRegistry, *, progress: ProgressCallback | None -) -> None: - """Copy a release and its required payload sources into the local registry. - - If the release is intact, copy only its missing or damaged payload sources. - - Args: - release: The release to copy. - fetch_from: The registry to copy from. - store: The destination registry. - progress: The callback for copy progress events, or ``None``. - """ - held = _intact_manifest(store, release) - if held is None: - copy_release(release, fetch_from, store, progress=progress) - return - for pinned in held.payload_sources.values(): - if _intact_manifest(store, pinned.release) is None: - copy_release(pinned.release, fetch_from, store, progress=progress) - # The copied payload sources must hold the files that the release pins. - prepare_local_copy(store.root, release) - - -def _intact_manifest(store: LocalRegistry, reference: DatasetRef) -> Manifest | None: - """Read a release's manifest and check its file sizes. +def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = None) -> None: + """Run a plan's copy and build steps in order. Args: - store: The local registry. - reference: The release to check. + plan: The steps to run in dependency order. + store: The local registry to copy and build into. + options: Build settings, including the progress callback for copies and builds. - Returns: - The manifest, or ``None`` if the release or a file is missing or a file has the wrong size. - """ - try: - manifest = store.get_manifest(reference.dataset_id, str(reference.version)) - except TimeNetDatasetNotFoundError: - return None - return manifest if complete_copy(release_dir(store.root, reference), manifest) else None - - -def _build(step: Step, store: LocalRegistry, *, force: bool, options: BuildOptions) -> None: - """Build a planned release and check its required files. + Raises: + TimeNetVersionConflictError: If ``store`` holds different files for a copied release. + TimeFFormatError: If a build fails to reproduce pinned files. + TimeNetBuildError: If a build fails. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the copy and the backends + options = options or BuildOptions() + for step in plan.steps: + match step: + case CopyStep(): + copy_release( + step.release, + step.registry, + store, + with_payload_sources=step.with_payload_sources, + progress=options.progress, + ) + case BuildStep(): + _build(step, store, options=options) + case _: + assert_never(step) + + +def _build(step: BuildStep, store: LocalRegistry, *, options: BuildOptions) -> None: + """Build an exact release with the builder of its step and check its required files. Args: - step: The build step. + step: The build step, with ``force`` set to rebuild an existing release. store: The local registry to build into. - force: Rebuild even if ``store`` already holds the release. options: The build settings. Raises: - TimeNetDatasetNotFoundError: If no backend builds this version or the result lacks a release or payload source. - """ # ruff: ignore[docstring-extraneous-exception] - raised by declared and prepare_local_copy - builder = declared(step.release.dataset_id, step.release.version).backend + TimeNetDatasetNotFoundError: If the result lacks the release or a payload source. + """ # ruff: ignore[docstring-extraneous-exception] - raised by prepare_local_copy if options.progress is not None: options.progress(StepStarted(step.release, "build")) - builder.build(step.release, store, force=force, pinned_by=step.pinned_by, options=options) + step.builder.backend.build(step.release, store, force=step.force, pinned_by=step.pinned_by, options=options) # After a successful build, the local registry must hold the exact release with its payload sources. prepare_local_copy(store.root, step.release) if options.progress is not None: diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index cf4d0fac3..d86ac72f2 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -1,16 +1,11 @@ -"""Plan how to obtain the parent releases that a build needs, before any connector runs. +"""Plan release copies and builds with dependencies first, before any connector runs.""" -Each :class:`Step` copies or builds one release after its dependencies are ready. -""" - -from collections.abc import Mapping -from dataclasses import dataclass, field +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field, replace from graphlib import CycleError, TopologicalSorter -from typing import Literal +from typing import ClassVar, Literal, TypeAlias -import pyarrow.fs as pafs - -from timenet.builders import find_card +from timenet.builders import InstalledBuilder, find_card from timenet.errors import ( TimeFFormatError, TimeFValidationError, @@ -18,25 +13,52 @@ TimeNetFileError, TimeNetMissingParentError, ) -from timenet.manifest import Manifest, PayloadSource +from timenet.manifest import FilePart, Manifest, PayloadSource from timenet.registry.base import BaseRegistry -from timenet.registry.dependencies import check_file_sizes +from timenet.registry.dependencies import ( + CopySources, + check_pins, + find_payload_sources, + validate_manifest_reference, +) from timenet.registry.local import LocalRegistry -from timenet.registry.version import DatasetVersion -from timenet.types import DatasetMetadata +from timenet.sources import ( + FoundInConnector, + FoundInRegistry, + InstalledConnectors, + NotFound, + NotServed, + ReleaseLookup, + ReleaseSource, + resolve_release, +) from timenet.types.access import require_open_access from timenet.types.references import DatasetRef +from timenet.types.version import Version @dataclass(frozen=True) -class Step: - """A plan step that copies or builds one release. +class CopyStep: + """A planned copy of one release from a registry into the local registry.""" - A copy includes the release and its missing payload sources. - """ + release: DatasetRef + registry: BaseRegistry + """The registry to copy from.""" + depends_on: tuple[DatasetRef, ...] = () + """The releases with steps that must finish before this step runs.""" + with_payload_sources: bool = True + """Also copy the payload sources that the manifest of the release lists.""" + action: ClassVar[Literal["copy"]] = "copy" + """The kind of step, for display.""" + + +@dataclass(frozen=True) +class BuildStep: + """A planned build of one release by an installed connector into the local registry.""" release: DatasetRef - action: Literal["copy", "build"] + builder: InstalledBuilder = field(hash=False) + """The installed connector that builds the release.""" depends_on: tuple[DatasetRef, ...] = () """The releases with steps that must finish before this step runs.""" pinned_by: Mapping[DatasetRef, PayloadSource] = field(default_factory=dict, hash=False) @@ -44,6 +66,14 @@ class Step: A build must reproduce every recorded file before it publishes the release. """ + force: bool = False + """Build the release again, also if the local registry holds it.""" + action: ClassVar[Literal["build"]] = "build" + """The kind of step, for display.""" + + +Step: TypeAlias = CopyStep | BuildStep +"""A planned copy or build of one release.""" @dataclass(frozen=True) @@ -53,83 +83,267 @@ class Plan: steps: tuple[Step, ...] +Role = Literal["dataset", "payload-source"] +"""The release's role as a dataset or a source of payload files. + +A release needs exactly the releases that its manifest lists as payload sources, never the payload sources of a +payload source. So a ``"dataset"`` needs its payload sources to open, and a ``"payload-source"`` supplies only its +own files. +""" + + @dataclass(frozen=True) class _Release: - """A required release, its dependencies, and its planned action.""" + """A required release with its role, planned step and dependencies.""" - opened: bool - """True if the build opens the release with its payload sources. False if the release is only a - payload source of another release, and the build reads only its pinned files.""" - action: Literal["copy", "build"] | None - """The planned action, or ``None`` if the output already holds the release.""" + role: Role + """The release's role as a dataset or a source of payload files.""" + step: Step | None + """The copy or build step before dependencies are set, or ``None`` for an existing release.""" dependencies: tuple[DatasetRef, ...] - """The releases that must be in place before this release. A release to build depends on the - parents on its card. An opened release that the plan does not build depends on its payload - sources. Other releases depend on nothing.""" + """The parent or payload source releases that must be ready first.""" @dataclass class _Graph: - """The releases a build needs and the dependencies between them.""" + """A graph of required releases and their copy or build steps.""" output: LocalRegistry - fetch_from: BaseRegistry | None - build_parents: bool + sources: Sequence[ReleaseSource] + """Sources for missing releases in preference order, excluding the output.""" open_only: bool - """Whether each visited release and its lineage must declare ``access: open``.""" - searched: str - """The registries that the plan looks in, for the message of a missing release.""" + """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.""" - fetched: set[DatasetRef] = field(default_factory=set) - """The payload sources that the plan copies as part of another release.""" 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.""" + 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: - """Add a release and its required dependencies to the graph. + """Add a release and its dependencies to the graph. - Use :meth:`order` to check the completed graph for cycles. + The graph revisits an existing entry only when its role changes from payload source to dataset. Args: reference: The release to add. needed_by: The release that needs ``reference``. - pinned: File pins from ``needed_by``, or ``None`` to open ``reference`` as a parent. - """ - opened = pinned is None + pinned: File pins from ``needed_by``, or ``None`` for a parent dataset. + + Raises: + 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.opened or not opened): + if found is not None and (found.role == "dataset" or role == "payload-source"): return - manifest, action = _locate(self.output, self.fetch_from, reference, open_only=self.open_only) - if manifest is None: - card = _card_to_build( - reference, - needed_by, - build_parents=self.build_parents, - open_only=self.open_only, - pinned=not opened, - searched=self.searched, - ) - self.releases[reference] = _Release(opened, "build", tuple(card.parents.values())) - # A build opens each of its parents with the payload sources of that parent. - for parent in card.parents.values(): - self.visit(parent, reference) - elif opened: - # To open a release, the build needs its direct payload sources. These are the releases - # whose files the release reads. The build does not need their own payload sources. - sources = list(manifest.payload_sources.values()) - self.releases[reference] = _Release(opened, action, tuple(source.release for source in sources)) + 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)) for source in sources: self.visit(source.release, reference, source) + return + answer = self.resolve(reference) + if isinstance(answer, FoundInRegistry) and not _is_same_registry(answer.registry, self.output): + self.copy(reference, role, answer.registry) + elif damage is not None: + # A copy from another registry repairs a damaged copy. Nothing else replaces it. + raise damage + elif isinstance(answer, FoundInConnector): + self.build(reference, role, answer.builder) + elif isinstance(answer, NotServed): + raise answer.access_error(reference.dataset_id) else: - self.releases[reference] = _Release(opened, action, ()) - # A copy of the release that reads this payload source also copies the payload source. - if action == "copy": - self.fetched.add(reference) + # 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" + ) + + 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. + + A dataset copy brings each payload source that ``registry`` serves. Each other payload source that the output + lacks gets its own step first. + + Args: + reference: The 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. + 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) + if manifest is None: + raise TimeNetDatasetNotFoundError(f"{registry} no longer holds {reference}") + if self.open_only: + require_open_access(manifest.metadata, manifest.lineage) + sources = CopySources() + if role == "dataset": + sources = find_payload_sources( + reference, + manifest, + registry, + find=lambda source: self.listed(registry, source), + held=lambda source: self.in_output(source)[0], + ) + if self.open_only: + for found in (*sources.fetched.values(), *sources.held.values()): + require_open_access(found.metadata, found.lineage) + 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)) + for source, pinned in sources.missing.items(): + self.visit(source, reference, pinned) + + def build(self, reference: DatasetRef, role: Role, builder: InstalledBuilder) -> None: + """Add a build step and plan the parents declared in its card. + + Args: + reference: The release to build. + role: Whether the release is a dataset or a payload source. + builder: The installed connector that builds the release. + """ + card = builder.card + if self.open_only: + require_open_access(card) + parents = tuple(card.parents.values()) + self.releases[reference] = _Release(role, 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) + + def resolve(self, reference: DatasetRef) -> ReleaseLookup: + """Find an exact release across the sources in preference order. + + 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. + + Returns: + The cached or newly fetched answer. + """ + 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)) + + 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. + + Returns: + The manifest, or ``None`` if the registry does not hold the release. + """ + if isinstance(self.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] + + def in_output(self, reference: DatasetRef) -> tuple[Manifest | None, TimeNetFileError | None]: + """Check a release in the output for missing files or incorrect file sizes. + + Returns: + ``(manifest, None)`` if intact, ``(None, error)`` if damaged, else ``(None, None)``. + """ + manifest = self.manifest(self.output, reference) + if manifest is None: + return manifest, None + if reference not in self.damage: + self.damage[reference] = self.output.damage(reference, manifest) + 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. + + 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 + 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. + registry = self.output if step is None else step.registry + manifest = self.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) def order(self) -> list[DatasetRef]: """Order the releases so that each release comes after the releases that it depends on. @@ -156,219 +370,120 @@ def order(self) -> list[DatasetRef]: path = " -> ".join(str(reference) for reference in (*cycle, cycle[0])) raise TimeFValidationError(f"dataset parent cycle: {path}") from error - def steps(self, requested: DatasetRef) -> list[Step]: - """Create copy and build steps with dependencies first. - - Each step names only dependencies that also have steps. - - Args: - requested: The requested release, which gets a copy step if needed but no build step. + def create_ordered_steps(self) -> list[Step]: + """Create copy and build steps in dependency order. Returns: - The steps in dependency order. + The ordered steps, including the requested build if needed. Raises: - TimeFValidationError: If the releases form a dependency cycle. + TimeFValidationError: If the dependencies form a cycle. """ # ruff: ignore[docstring-extraneous-exception] - order() raises the cycle error - chosen: dict[DatasetRef, Literal["copy", "build"]] = {} + nearest: dict[DatasetRef, tuple[DatasetRef, ...]] = {} + steps: list[Step] = [] for reference in self.order(): release = self.releases[reference] - if reference == requested and release.action == "build": - continue - if release.action == "build": - chosen[reference] = "build" - # A copy of an opened release also copies each payload source that the output does not - # hold. So if the plan copies a payload source only with the release that reads it, the - # plan copies that release again, even if the output holds it. - elif release.opened and ( - release.action == "copy" or any(source in self.fetched for source in release.dependencies) - ): - chosen[reference] = "copy" - return [ - Step( - reference, - action, - depends_on=tuple(need for need in self.releases[reference].dependencies if need in chosen), - pinned_by=self.pins.get(reference, {}) if action == "build" else {}, - ) - for reference, action in chosen.items() - ] + # The order puts each dependency first, so its nearest releases with a step are known. + needs: dict[DatasetRef, None] = {} + for need in release.dependencies: + needs.update(dict.fromkeys((need,) if self.releases[need].step is not None else nearest[need])) + nearest[reference] = tuple(needs) + step = release.step + if isinstance(step, BuildStep): + step = replace(step, pinned_by=self.pins.get(reference, {})) + if step is not None: + steps.append(replace(step, depends_on=nearest[reference])) + return steps + + +def _check_agreement(release: DatasetRef, readers: Mapping[DatasetRef, PayloadSource]) -> None: + """Check that the readers of a release pin each of its files alike. + Args: + release: The release whose files are pinned. + readers: The releases that pin files of ``release``, with their file requirements. -def plan_parents( # ruff: ignore[too-many-arguments] - dataset_id: str, + Raises: + TimeFFormatError: If two readers pin one file as different files. + """ + pinned: dict[str, tuple[DatasetRef, FilePart]] = {} + for reader, source in readers.items(): + for part in source.files: + first_reader, first_part = pinned.setdefault(part.path, (reader, part)) + if first_part != part: + raise TimeFFormatError( + f"{first_reader} and {reader} pin {part.path!r} of {release} as different files, " + f"so no build of {release} can match both" + ) + + +def plan_steps( # ruff: ignore[too-many-arguments] + builder: InstalledBuilder, output: LocalRegistry, *, - fetch_from: BaseRegistry | None = None, - build_parents: bool = False, - force_build: bool = False, - open_access_only: bool = False, -) -> list[Step]: - """Plan the copy and build steps for a dataset's parents. + parents_from: Sequence[ReleaseSource], + force_build: bool, + open_access_only: bool, + pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, +) -> tuple[DatasetRef, list[Step]]: + """Plan a dataset build and its dependencies. - Copy missing parents and their payload sources from ``fetch_from``, or build them if enabled. + A damaged copy of the requested release in the output gets a forced build. Args: - dataset_id: The dataset to build. + builder: The installed connector that builds the dataset. output: The local registry that the build writes to. - fetch_from: The registry to copy missing parents from, or ``None``. - build_parents: Build missing parents with their installed connectors. - force_build: Plan parents for a rebuild even if the requested release already exists. - open_access_only: Require ``access: open`` for each release, including the requested release. + 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. Returns: - The copy and build steps, with dependencies first. - - Raises: - LookupError: If no installed connector declares a card for ``dataset_id``. - TimeNetInvalidCardError: If an installed connector's card cannot be read. - TimeFValidationError: If parents form a cycle or versions conflict. - TimeNetMissingParentError: If a parent or payload source is missing and ``build_parents`` is false. - TimeNetDatasetNotFoundError: If a parent is missing and no installed connector builds it. - TimeNetFileError: If a local parent is damaged and no replacement is available. - TimeNetAccessError: If a release or its lineage lacks required open access. - """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers of the walk - card = find_card(dataset_id) - if card is None: - raise LookupError(f"no installed connector declares a card for {dataset_id!r}") + The requested release and ordered steps, with its own build last 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) - # For a dataset without parents, the plan does not read the output, so the build makes no extra request. - if not card.parents: - return [] - root = DatasetRef(dataset_id=card.dataset_id, version=card.dataset_version) + root = builder.release + graph = _Graph(output, parents_from, open_only=open_access_only) + if pinned_by: + graph.pins[root] = dict(pinned_by) # If the output already holds the requested release, a reader needs only its payload sources to # open it. Its declared parents matter only to a rebuild. - held = None if force_build else _manifest(output, root) + held, damage = (None, None) if force_build else graph.in_output(root) if held is not None and open_access_only: # The installed card has no lineage, so the check above does not see an ancestor that is not open. require_open_access(held.metadata, held.lineage) - searched = str(output) if fetch_from is None else f"{output} or {fetch_from}" - graph = _Graph(output, fetch_from, build_parents=build_parents, open_only=open_access_only, searched=searched) # The graph holds the requested release before the walk starts. So a release that depends on it # closes a cycle, and the walk does not look it up again. if held is None: - graph.releases[root] = _Release(opened=True, action="build", dependencies=tuple(card.parents.values())) + 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())) for parent in card.parents.values(): graph.visit(parent, root) else: sources = list(held.payload_sources.values()) graph.releases[root] = _Release( - opened=True, action=None, dependencies=tuple(source.release for source in sources) + role="dataset", step=None, dependencies=tuple(source.release for source in sources) ) for source in sources: graph.visit(source.release, root, source) - return graph.steps(root) - - -def _manifest(registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: - """Read the manifest of one release. - - This reads no other file and resolves no payload source. - - Returns: - The manifest, or ``None`` if the registry does not hold the release. - - Raises: - TimeFFormatError: If the manifest declares another dataset ID or version than ``reference``. - """ - try: - manifest = registry.get_manifest(reference.dataset_id, str(reference.version)) - except TimeNetDatasetNotFoundError: - return None - if manifest.dataset_id != reference.dataset_id or manifest.metadata.dataset_version != reference.version: - raise TimeFFormatError(f"manifest for {reference} declares a different dataset ID or version") - return manifest + graph.check() + return root, graph.create_ordered_steps() -def _locate( - output: LocalRegistry, fetch_from: BaseRegistry | None, reference: DatasetRef, *, open_only: bool -) -> tuple[Manifest | None, Literal["copy"] | None]: - # The damage is raised from a variable, the access error by require_open_access. - # ruff: ignore[docstring-extraneous-exception] - """Find a release in the output or the source registry. +def _is_same_registry(source: ReleaseSource, output: LocalRegistry) -> bool: + """Check whether a source is the output registry. - A local copy is intact when all files exist at their recorded sizes. - Enforce open access for the release and its lineage when ``open_only`` or a copy requires it. + Args: + source: The source to compare. + output: The destination registry. Returns: - The manifest and action (``None`` to reuse, ``"copy"`` to fetch), or ``(None, None)`` if absent. - - Raises: - TimeNetFileError: If the local copy is damaged and the source registry lacks the release. - TimeNetAccessError: If the release or its lineage lacks required open access. + ``True`` for the same object or local registries with the same resolved directory. """ - manifest = _manifest(output, reference) - damage = None - if manifest is not None: - directory = output.root / reference.dataset_id / str(reference.version) - try: - check_file_sizes( - DatasetVersion(manifest, pafs.LocalFileSystem(), str(directory)), - manifest.files.all_files(), - owner=f"committed release {reference} in {output}", - ) - except TimeNetFileError as error: - manifest, damage = None, error - if manifest is not None: - if open_only: - require_open_access(manifest.metadata, manifest.lineage) - return manifest, None - fetched = None if fetch_from is None else _manifest(fetch_from, reference) - if fetched is not None: - # A registry that requires open access serves only releases whose cards declare access: open. - # So the plan refuses another release before any download. - if open_only or (fetch_from is not None and fetch_from.requires_open_access): - require_open_access(fetched.metadata, fetched.lineage) - return fetched, "copy" - if damage is not None: - raise damage - return None, None - - -def _card_to_build( # ruff: ignore[too-many-arguments] - reference: DatasetRef, - needed_by: DatasetRef, - *, - build_parents: bool, - open_only: bool, - pinned: bool, - searched: str, -) -> DatasetMetadata: - """Find a connector card for building a missing release. - - Returns: - The connector card for the exact dataset ID and version in ``reference``. - - Raises: - TimeNetMissingParentError: If ``build_parents`` is false. - TimeNetAccessError: If ``open_only`` is set and the card does not declare ``access: open``. - """ # ruff: ignore[docstring-extraneous-exception] - the access error is raised by require_open_access - if not build_parents: - raise TimeNetMissingParentError(reference, needed_by, searched, payload_source=pinned) - card = _exact_card(reference) - if open_only: - require_open_access(card) - return card - - -def _exact_card(reference: DatasetRef) -> DatasetMetadata: - """Return the card of the installed connector that builds exactly ``reference``. - - Returns: - The card of the connector. - - Raises: - TimeNetDatasetNotFoundError: If no installed connector builds the dataset. - TimeFValidationError: If the installed connector declares another version. - TimeNetInvalidCardError: If the card of the installed connector cannot be read. - """ # ruff: ignore[docstring-extraneous-exception] - find_card raises the card error - card = find_card(reference.dataset_id) - if card is None: - raise TimeNetDatasetNotFoundError(f"parent release {reference} is missing and no installed connector builds it") - if card.dataset_version != reference.version: - raise TimeFValidationError( - f"parent release {reference} is missing and the installed connector builds " - f"version {card.dataset_version}, not that exact version" - ) - return card + if source is output: + return True + return isinstance(source, LocalRegistry) and source.root.resolve() == output.root.resolve() diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py index b9ec678e9..db7ad0b6f 100644 --- a/packages/timenet/src/timenet/registry/_copy.py +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -27,24 +27,24 @@ def copy_release( with_payload_sources: bool = True, progress: ProgressCallback | None = None, ) -> None: - """Copy a release and its payload sources to another registry. + """Copy a release and its required payload sources to another registry. - Copy payload sources first, reuse complete releases in ``dest``, and fetch damaged local copies again. - All checks run before the first copy, so a failed check leaves ``dest`` unchanged. + The copy reuses complete releases in ``dest`` and fetches damaged local copies again. + It checks access, file records and pins before copying payload sources and then the requested release. Args: reference: The exact release to copy. source: The registry to copy from. dest: The registry to copy to. with_payload_sources: Also copy the payload sources that the manifest of the release lists. - progress: Callback for ``StepStarted`` and ``StepFinished`` events for each copied or repaired release. + progress: Callback for ``StepStarted`` and ``StepFinished`` events for each copied release. Raises: - TimeNetVersionConflictError: If ``dest`` holds the release or a payload source with other files or another card. - TimeNetAccessError: If a release lacks open access where the source or destination requires it. + TimeNetVersionConflictError: If ``dest`` holds other files or another card for a release being copied. + TimeNetAccessError: If a release fails an open access requirement of ``dest`` or ``source``. TimeNetDatasetNotFoundError: If ``source`` does not hold the release. - TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` lacks a complete copy. - TimeFFormatError: If a manifest names the wrong release or pinned files differ. + TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` holds no complete copy of it. + TimeFFormatError: If a manifest names a different release or a pin fails. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers manifests = _manifests(reference, source, dest, with_payload_sources=with_payload_sources) # Check every release before anything is copied, so that a refusal leaves dest unchanged. A payload @@ -146,7 +146,7 @@ def _manifests( if listed is not None: # The source lists the payload source, but does not serve its files. require_open_access(listed.metadata, listed.lineage) - raise TimeNetMissingParentError(missing, reference, str(source), payload_source=True) + raise TimeNetMissingParentError(missing, reference, f"{source} or {dest}", payload_source=True) return {**sources.fetched, **sources.held, reference: manifest} diff --git a/packages/timenet/src/timenet/sources.py b/packages/timenet/src/timenet/sources.py index a72aaaea0..3689f8baa 100644 --- a/packages/timenet/src/timenet/sources.py +++ b/packages/timenet/src/timenet/sources.py @@ -1,6 +1,6 @@ """Find dataset releases across registries and installed connectors.""" -from collections.abc import Sequence +from collections.abc import Callable, Sequence from dataclasses import dataclass, field from typing import TYPE_CHECKING, TypeAlias, Union @@ -128,7 +128,13 @@ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup: # return FoundInConnector(builder) -def resolve_release(sources: Sequence[ReleaseSource], dataset_id: str, version: str | None) -> ReleaseLookup: +Ask: TypeAlias = Callable[[ReleaseSource, str, Version | None], ReleaseLookup] +"""A function that asks one source for a release: ``ask(source, dataset_id, version)``.""" + + +def resolve_release( + sources: Sequence[ReleaseSource], dataset_id: str, version: str | None, *, ask: Ask | None = None +) -> ReleaseLookup: """Find a source that provides the requested dataset release. An exact version comes from the first source that serves it, and no version means the newest version of the first @@ -139,6 +145,7 @@ def resolve_release(sources: Sequence[ReleaseSource], dataset_id: str, version: sources: The sources, in order. dataset_id: The dataset ID. version: An exact version, ``"latest"``, or ``None`` or ``""`` for no version. + ask: The function that asks one source, or ``None`` to call the source's ``resolve()``. Returns: A ``Found`` result if available, otherwise the first ``NotServed`` result or ``NotFound``. @@ -146,16 +153,24 @@ def resolve_release(sources: Sequence[ReleaseSource], dataset_id: str, version: Raises: TimeFValidationError: If ``version`` is neither ``"latest"`` nor a ``major.minor.patch`` version. """ # ruff: ignore[docstring-extraneous-exception] - raised by Version.parse + ask = ask or _ask if version == LATEST_VERSION: - return _resolve_latest(sources, dataset_id) - return _resolve_first(sources, dataset_id, Version.parse(version) if version else None) + return _resolve_latest(sources, dataset_id, ask) + return _resolve_first(sources, dataset_id, Version.parse(version) if version else None, ask) + + +def _ask(source: ReleaseSource, dataset_id: str, version: Version | None) -> ReleaseLookup: + """Return the answer of a source for a release.""" + return source.resolve(dataset_id, version) -def _resolve_first(sources: Sequence[ReleaseSource], dataset_id: str, version: Version | None) -> ReleaseLookup: +def _resolve_first( + sources: Sequence[ReleaseSource], dataset_id: str, version: Version | None, ask: Ask +) -> ReleaseLookup: """Return the answer of the first source that serves ``version``, or its newest for ``None``, else the first listing.""" listed: NotServed | None = None for source in sources: - answer = source.resolve(dataset_id, version) + answer = ask(source, dataset_id, version) if isinstance(answer, Found): return answer if isinstance(answer, NotServed) and listed is None: @@ -163,23 +178,25 @@ def _resolve_first(sources: Sequence[ReleaseSource], dataset_id: str, version: V return NotFound() if listed is None else listed -def _resolve_latest(sources: Sequence[ReleaseSource], dataset_id: str) -> ReleaseLookup: +def _resolve_latest(sources: Sequence[ReleaseSource], dataset_id: str, ask: Ask) -> ReleaseLookup: """Return the highest version that any source provides, from the first source that provides it.""" best: Found | None = None listed: list[tuple[ReleaseSource, NotServed]] = [] before_best = 0 for source in sources: - answer = source.resolve(dataset_id, None) + answer = ask(source, dataset_id, None) if isinstance(answer, NotServed): listed.append((source, answer)) elif isinstance(answer, Found) and (best is None or answer.version > best.version): best, before_best = answer, len(listed) if best is None: return listed[0][1] if listed else NotFound() - return _ask_again(listed[:before_best], dataset_id, best.version) or best + return _ask_again(listed[:before_best], dataset_id, best.version, ask) or best -def _ask_again(listed: Sequence[tuple[ReleaseSource, NotServed]], dataset_id: str, version: Version) -> Found | None: +def _ask_again( + listed: Sequence[tuple[ReleaseSource, NotServed]], dataset_id: str, version: Version, ask: Ask +) -> Found | None: """Ask each source that lists a higher version without serving it whether it serves ``version``. Returns: @@ -187,7 +204,7 @@ def _ask_again(listed: Sequence[tuple[ReleaseSource, NotServed]], dataset_id: st """ for source, answer in listed: if answer.version > version: - again = source.resolve(dataset_id, version) + again = ask(source, dataset_id, version) if isinstance(again, Found): return again return None diff --git a/packages/timenet/tests/coordinator/test_build.py b/packages/timenet/tests/coordinator/test_build.py index be5ca53cc..103af8160 100644 --- a/packages/timenet/tests/coordinator/test_build.py +++ b/packages/timenet/tests/coordinator/test_build.py @@ -2,16 +2,17 @@ import pytest +from timenet import builders from timenet.builders import BuildOptions -from timenet.coordinator.build import build_release -from timenet.coordinator.plan import Step +from timenet.coordinator.build import plan_release_build, run_plan +from timenet.coordinator.plan import BuildStep, CopyStep from timenet.dataset import TimeFDataset -from timenet.errors import TimeNetAccessError, TimeNetDatasetNotFoundError, TimeNetVersionUnavailableError +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._paths import release_dir -from timenet.registry.shared import complete_copy +from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset from timenet.types import Access from timenet.types.reference_keys import ParentAlias, SourceKey @@ -25,6 +26,11 @@ _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( @@ -119,56 +125,32 @@ def pins(store): return _store_child_without_parent(store) -def test_build_release_builds_into_the_store_and_returns_the_plan(store, fake): - plan = build_release("a/b", store=store) - assert [(s.action, s.release) for s in plan.steps] == [("build", REF_AB)] - assert store.exists("a/b", "1.0.0") - assert fake.calls == [(REF_AB, False, {})] - - -def test_build_release_calls_the_claiming_backend_with_the_pins(store, fake_parent, fake_child, pins): - plan = build_release(CHILD_REF.dataset_id, store=store, build_parents=True) - # (release, force, pinned_by): the backend of another package gets the pins of the child. - assert fake_parent.calls == [(PARENT_REF, False, {CHILD_REF: pins})] - # The store holds the child, so the plan builds only its missing payload source. - assert fake_child.calls == [] - assert plan.steps == (Step(PARENT_REF, "build", pinned_by={CHILD_REF: pins}),) - assert store.exists(PARENT_REF.dataset_id, str(PARENT_REF.version)) - - -def test_build_release_passes_the_pins_of_the_caller_to_the_backend(store, fake_parent, pins): - plan = build_release(PARENT_REF.dataset_id, store=store, pinned_by={CHILD_REF: pins}) +def test_a_build_passes_the_pins_of_the_caller_to_the_backend(store, fake_parent, pins): + plan = plan_release_build(PARENT_REF.dataset_id, store=store, pinned_by={CHILD_REF: pins}) + run_plan(plan, store) assert fake_parent.calls == [(PARENT_REF, False, {CHILD_REF: pins})] - assert plan.steps == (Step(PARENT_REF, "build", pinned_by={CHILD_REF: pins}),) + assert plan.steps == (_build(PARENT_REF, pinned_by={CHILD_REF: pins}),) assert store.exists(PARENT_REF.dataset_id, str(PARENT_REF.version)) -def test_build_release_rejects_a_version_the_connector_does_not_declare(store, fake): +def test_plan_release_build_rejects_a_version_the_connector_does_not_declare(store, fake): with pytest.raises(TimeNetVersionUnavailableError) as caught: - build_release("a/b@2.0.0", store=store) + plan_release_build("a/b@2.0.0", store=store) assert (caught.value.release, caught.value.declared) == (DatasetRef.parse("a/b@2.0.0"), REF_AB.version) - assert str(caught.value) == ( - "connector for 'a/b' builds version 1.0.0, not the requested 2.0.0. " - "Ask for version 1.0.0, or leave out the version." - ) - assert fake.calls == [] - - -def test_build_release_accepts_the_version_that_the_connector_declares(store, fake): - build_release("a/b@1.0.0", store=store) - assert fake.calls == [(REF_AB, False, {})] -def test_build_release_copies_a_missing_parent_and_reports_each_step_once(tmp_path, store, fake_parent, fake_child): +def test_a_build_copies_a_missing_parent_and_reports_each_step_once(tmp_path, store, fake_parent, fake_child): origin = LocalRegistry(tmp_path / "origin") origin.put_dataset(make_dataset()) events = [] - plan = build_release( - CHILD_REF.dataset_id, store=store, fetch_from=origin, options=BuildOptions(progress=events.append) - ) + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[origin]) + run_plan(plan, store, options=BuildOptions(progress=events.append)) - assert plan.steps == (Step(PARENT_REF, "copy"), Step(CHILD_REF, "build", depends_on=(PARENT_REF,))) + assert plan.steps == ( + CopyStep(PARENT_REF, origin), + _build(CHILD_REF, depends_on=(PARENT_REF,)), + ) assert events == [ StepStarted(PARENT_REF, "copy"), StepFinished(PARENT_REF, "copy"), @@ -179,31 +161,7 @@ def test_build_release_copies_a_missing_parent_and_reports_each_step_once(tmp_pa assert store.exists(PARENT_REF.dataset_id, str(PARENT_REF.version)) -@pytest.mark.usefixtures("fake_child") -@pytest.mark.parametrize("payload_source", ["missing", "damaged"]) -def test_build_release_reuses_a_held_release_and_copies_only_its_payload_source( - tmp_path, store, fake_parent, pins, payload_source -): - # The store holds the child, but not an intact copy of its payload source. The registry to copy - # from holds only the payload source. - origin = LocalRegistry(tmp_path / "origin") - origin.put_dataset(make_dataset()) - if payload_source == "damaged": - store.put_dataset(make_dataset()) - (release_dir(store.root, PARENT_REF) / "control.duckdb").write_bytes(b"cut short") - events = [] - - plan = build_release( - CHILD_REF.dataset_id, store=store, fetch_from=origin, options=BuildOptions(progress=events.append) - ) - - assert plan.steps == (Step(CHILD_REF, "copy"),) - assert events == [StepStarted(PARENT_REF, "copy"), StepFinished(PARENT_REF, "copy")] - assert fake_parent.calls == [] - assert complete_copy(release_dir(store.root, PARENT_REF), origin.get_manifest(PARENT_REF.dataset_id)) - - -def test_build_release_copies_a_release_whose_payload_source_only_the_store_holds( +def test_a_build_copies_a_release_whose_payload_source_only_the_store_holds( tmp_path, store, fake_parent, fake_child, backends ): # The registry to copy from holds the child, but not the parent whose files the child reads. @@ -213,15 +171,19 @@ def test_build_release_copies_a_release_whose_payload_source_only_the_store_hold fake_grandchild = _Backend(_grandchild_card()) backends.append(fake_grandchild) - plan = build_release(GRANDCHILD_REF.dataset_id, store=store, fetch_from=origin) + plan = plan_release_build(GRANDCHILD_REF.dataset_id, store=store, parents_from=[origin]) + run_plan(plan, store) - assert plan.steps == (Step(CHILD_REF, "copy"), Step(GRANDCHILD_REF, "build", depends_on=(CHILD_REF,))) + assert plan.steps == ( + CopyStep(CHILD_REF, origin), + _build(GRANDCHILD_REF, depends_on=(CHILD_REF,)), + ) assert fake_parent.calls == fake_child.calls == [] assert fake_grandchild.calls == [(GRANDCHILD_REF, False, {})] assert store.get_manifest(CHILD_REF.dataset_id) == origin.get_manifest(CHILD_REF.dataset_id) -def test_build_release_builds_a_payload_source_of_a_release_that_it_copies( +def test_a_build_builds_a_payload_source_of_a_release_that_it_copies( tmp_path, store, fake_parent, fake_child, backends ): # The registry to copy from holds the child, but not the parent whose files the child reads. @@ -230,12 +192,13 @@ def test_build_release_builds_a_payload_source_of_a_release_that_it_copies( fake_grandchild = _Backend(_grandchild_card()) backends.append(fake_grandchild) - plan = build_release(GRANDCHILD_REF.dataset_id, store=store, fetch_from=origin, build_parents=True) + plan = plan_release_build(GRANDCHILD_REF.dataset_id, store=store, parents_from=[origin, InstalledConnectors()]) + run_plan(plan, store) assert plan.steps == ( - Step(PARENT_REF, "build", pinned_by={CHILD_REF: pins}), - Step(CHILD_REF, "copy", depends_on=(PARENT_REF,)), - Step(GRANDCHILD_REF, "build", depends_on=(CHILD_REF,)), + _build(PARENT_REF, pinned_by={CHILD_REF: pins}), + CopyStep(CHILD_REF, origin, depends_on=(PARENT_REF,)), + _build(GRANDCHILD_REF, depends_on=(CHILD_REF,)), ) assert fake_parent.calls == [(PARENT_REF, False, {CHILD_REF: pins})] assert fake_child.calls == [] @@ -243,53 +206,48 @@ def test_build_release_builds_a_payload_source_of_a_release_that_it_copies( assert store.get_manifest(CHILD_REF.dataset_id) == origin.get_manifest(CHILD_REF.dataset_id) -def test_build_release_forces_only_the_requested_release(store, fake_parent, fake_child): +def test_a_forced_build_forces_only_the_requested_release(store, fake_parent, fake_child): store.put_dataset(_demo(_child_card())) - plan = build_release(CHILD_REF.dataset_id, store=store, build_parents=True, force=True) + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()], force=True) + run_plan(plan, store) - assert [(s.action, s.release) for s in plan.steps] == [("build", PARENT_REF), ("build", CHILD_REF)] + assert plan.steps == (_build(PARENT_REF), _build(CHILD_REF, depends_on=(PARENT_REF,), force=True)) assert fake_parent.calls == [(PARENT_REF, False, {})] assert fake_child.calls == [(CHILD_REF, True, {})] -def test_build_release_skips_a_release_the_store_holds(store, fake): +def test_a_build_skips_a_release_the_store_holds(store, fake): + store.put_dataset(_demo(_card(REF_AB))) + assert plan_release_build("a/b", store=store).steps == () + + +def test_a_damaged_copy_of_the_requested_release_gets_a_forced_build(store, fake): store.put_dataset(_demo(_card(REF_AB))) - plan = build_release("a/b", store=store) - assert plan.steps == () - assert fake.calls == [] + (release_dir(store.root, REF_AB) / "control.duckdb").write_bytes(b"cut short") + + plan = plan_release_build("a/b", store=store) + + assert plan.steps == (_build(REF_AB, force=True),) + run_plan(plan, store) + assert fake.calls == [(REF_AB, True, {})] + assert store.damage(REF_AB) is None -def test_build_release_with_open_access_only_refuses_a_credentialed_card(store, backends): +def test_plan_release_build_with_open_access_only_refuses_a_credentialed_card(store, backends): fake_credentialed = _Backend(_card(REF_AB, **_GATED)) backends.append(fake_credentialed) with pytest.raises(TimeNetAccessError) as caught: - build_release("a/b", store=store, open_access_only=True) + plan_release_build("a/b", store=store, open_access_only=True) assert caught.value.dataset == REF_AB - assert fake_credentialed.calls == [] # Without the flag, the local store takes a release that is not open. - build_release("a/b", store=store) + run_plan(plan_release_build("a/b", store=store), store) assert fake_credentialed.calls == [(REF_AB, False, {})] -@pytest.mark.parametrize("held", [False, True]) -def test_build_release_with_open_access_only_refuses_a_credentialed_parent(store, backends, held): - gated = _Backend(make_dataset().metadata.model_copy(update=_GATED)) - child = _Backend(_child_card()) - backends.extend([gated, child]) - if held: - store.put_dataset(_demo(gated.cards[PARENT_REF.dataset_id])) - - with pytest.raises(TimeNetAccessError) as caught: - build_release(CHILD_REF.dataset_id, store=store, build_parents=True, open_access_only=True) - - assert caught.value.dataset == PARENT_REF - assert gated.calls == child.calls == [] - - -def test_build_release_with_open_access_only_refuses_a_held_release_with_a_credentialed_ancestor(store, backends): +def test_plan_release_build_with_open_access_only_refuses_a_held_release_with_a_credentialed_ancestor(store, backends): # The card of the child is open. The child keeps a Record of the open release a/b, which the store # lacks. It reads nothing from the credentialed parent, which is only in its lineage. child_card = _card(CHILD_REF, parents={ParentAlias("parent"): PARENT_REF, ParentAlias("open"): REF_AB}) @@ -312,12 +270,37 @@ def test_build_release_with_open_access_only_refuses_a_held_release_with_a_crede assert list(held.payload_sources) == [SourceKey(str(REF_AB))] with pytest.raises(TimeNetAccessError) as caught: - build_release(CHILD_REF.dataset_id, store=store, build_parents=True, open_access_only=True) + plan_release_build( + CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()], open_access_only=True + ) assert caught.value.dataset == CHILD_REF - assert other.calls == child.calls == [] -def test_build_release_without_a_backend_raises_not_found(store, backends): - with pytest.raises(TimeNetDatasetNotFoundError, match="no installed connector builds 'a/b'"): - build_release("a/b", store=store) +def test_a_plan_runs_only_through_run_plan(store, fake_parent, fake_child): + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()]) + + assert plan.steps == ( + _build(PARENT_REF), + _build(CHILD_REF, depends_on=(PARENT_REF,)), + ) + assert not store.root.exists() + assert fake_parent.calls == fake_child.calls == [] + run_plan(plan, store) + assert fake_parent.calls == [(PARENT_REF, False, {})] + assert fake_child.calls == [(CHILD_REF, False, {})] + + +def test_a_plan_reads_the_card_of_each_connector_once(store, backends): + reads = [] + + class _Counted(_Backend): + def card(self, dataset_id): + reads.append(dataset_id) + return super().card(dataset_id) + + backends.extend([_Counted(make_dataset().metadata), _Counted(_child_card())]) + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()]) + run_plan(plan, store) + assert reads == [CHILD_REF.dataset_id, PARENT_REF.dataset_id] + assert store.exists(CHILD_REF.dataset_id, str(CHILD_REF.version)) diff --git a/packages/timenet/tests/registry/test_copy.py b/packages/timenet/tests/registry/test_copy.py index a82858fe6..43514741f 100644 --- a/packages/timenet/tests/registry/test_copy.py +++ b/packages/timenet/tests/registry/test_copy.py @@ -290,9 +290,13 @@ def test_copy_release_checks_the_access_of_a_payload_source_that_only_dest_holds def test_copy_release_refuses_a_payload_source_that_no_registry_holds(source, dest): + origin = _without_parent(source) with pytest.raises(TimeNetMissingParentError) as raised: - copy_release(child_ref, _without_parent(source), dest) + copy_release(child_ref, origin, dest) assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) + # The error names both registries that the copy looked in. + assert raised.value.searched == f"{origin} or {dest}" + assert str(raised.value) == f"payload source {parent_ref} of {child_ref} is not in {origin} or {dest}." assert not dest.exists(*child) diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index 2307858d7..d75e84864 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -25,6 +25,7 @@ from timenet.manifest.files import FilePart from timenet.reader import Loading, LoadPlan, TimeFReader from timenet.registry import TIMENET_REGISTRY_URL, LocalRegistry, RemoteRegistry, open_local_version +from timenet.sources import InstalledConnectors from timenet.testing import assert_datasets_equal, install_backends, make_dataset from timenet.types import Access, DatasetMetadata, Domain, License, Version from timenet.types.access import AccessRequirement @@ -83,19 +84,19 @@ def build(self, release, store, *, force, pinned_by, options): def _record_build_requests(monkeypatch): - """Record the dataset, ``fetch_from`` and ``build_parents`` of each build that the client asks for. + """Record the dataset and the ``parents_from`` list of each build that the client asks for. Returns: - The list that receives one ``(dataset, fetch_from, build_parents)`` entry for each request. + The list that receives one ``(dataset, parents_from)`` entry for each request. """ requests = [] - build_release = client_module.build_release + plan_release_build = client_module.plan_release_build - def record(dataset, **options): - requests.append((dataset, options.get("fetch_from"), options.get("build_parents"))) - return build_release(dataset, **options) + def record(dataset_id, **options): + requests.append((dataset_id, options.get("parents_from"))) + return plan_release_build(dataset_id, **options) - monkeypatch.setattr(client_module, "build_release", record) + monkeypatch.setattr(client_module, "plan_release_build", record) return requests @@ -136,7 +137,7 @@ def test_a_missing_version_is_copied_from_fetch_from_before_it_is_built(tmp_path assert (tmp_path / "local/test/child/1.0.0/manifest.json").is_file() return # The build gets the same registry to copy its own missing parents from. - assert requests == [("test/child", origin, True)] + assert requests == [("test/child", [origin, InstalledConnectors()])] assert calls == [(DatasetRef.parse("test/child@1.0.0"), False, {})] @@ -229,7 +230,9 @@ def test_a_missing_payload_source_builds_with_its_own_connector_against_the_pins assert signal.to_arrow().equals(make_dataset().records[0].signals[0].to_arrow()) # The source builds with its own connector, which prepares its parents in the same order, against # the pins of the child that reads it. - assert requests == [(str(reference), fetch_from, True)] + assert requests == [ + (str(reference), [InstalledConnectors()] if fetch_from is None else [fetch_from, InstalledConnectors()]) + ] assert calls == [(reference, False, {DatasetRef.parse("test/child@1.0.0"): pins})]