From 4be71dab3415d5fcb1312cb3556db9fe77b01550 Mon Sep 17 00:00:00 2001 From: muhammadumer-waheed Date: Tue, 6 Oct 2026 13:43:00 +0500 Subject: [PATCH] fix(bundles): install pinned catalog releases via exact-release selection Bundle manifests pin component versions, but extension and preset installs compared the pin against the version the catalog currently advertises and refused the install when they differed, even when the catalog still listed the pinned release (#4712). The bundler now selects the pinned release from the winning catalog entry with the exact-release support added in #4726 (extensions) and #4823 (presets), downloads that record with its own URL and SHA-256 via download_extension_info / download_pack_info, and passes the pin as expected_id / expected_version so the archive must declare the pinned component. A pin the winning entry does not list fails with an error naming the pinned and advertised versions, without substituting the advertised release or falling through to another catalog. Unpinned components, entries advertising no version, workflows and bundled assets keep their existing behavior. Fixes #4712 --- docs/reference/bundles.md | 2 + src/specify_cli/bundles/primitives.py | 53 ++++- .../bundles/test_command_install.py | 7 +- tests/specify_cli/bundles/test_primitives.py | 191 +++++++++++++++++- 4 files changed, 242 insertions(+), 11 deletions(-) diff --git a/docs/reference/bundles.md b/docs/reference/bundles.md index cd0b23472a..d03826eef5 100644 --- a/docs/reference/bundles.md +++ b/docs/reference/bundles.md @@ -97,6 +97,8 @@ specify bundle update [] Re-resolves a bundle and **refreshes** its components through each primitive's update path, bringing already-installed components up to the bundle's newly pinned versions while preserving primitive-level overrides (such as preset priority). Provide a bundle id, or use `--all` to update everything installed. +**Pinned catalog releases.** An extension or preset pinned to a version other than the one its catalog currently advertises installs that exact release when the winning catalog entry lists it under `releases`, using that release's own download URL and SHA-256 digest. The downloaded archive must declare the pinned ID and version. If the winning catalog entry has no release for the pinned version, install stops with an error rather than substituting the advertised release or falling through to a lower-priority catalog. Workflows and components bundled with Spec Kit still require the pin to match the version they resolve to. + > **Pin enforcement is install-time only.** Idempotency checks are id-based, not version-aware: a component owned by a bundle that is already present is skipped during `install` without comparing its on-disk version to the manifest pin. Version pins are therefore guaranteed to be applied only when the bundler actually installs a component for the first time or refreshes it. Run `specify bundle update ` for catalog bundles or `specify bundle install --refresh` for local sources to re-apply owned components at their pinned versions. ## Remove a Bundle diff --git a/src/specify_cli/bundles/primitives.py b/src/specify_cli/bundles/primitives.py index c885d62443..26a0518d44 100644 --- a/src/specify_cli/bundles/primitives.py +++ b/src/specify_cli/bundles/primitives.py @@ -56,6 +56,33 @@ def _assert_pinned_version( ) +def _select_pinned_release( + kind: str, component: ComponentRef, info: dict, select_release +) -> tuple[dict, str | None]: + """Select the catalog release a bundle pin names. + + Returns the selected release record and the version the archive must + declare (``None`` when the pin cannot be enforced). Selection stays within + the winning catalog entry, so a pinned release missing from it is an error + rather than a silent fall-through to the advertised release or to a + lower-priority catalog. An entry advertising no version cannot enforce the + pin, so it is installed as resolved (mirrors ``_assert_pinned_version``). + """ + pinned = component.version + advertised = info.get("version") + if not pinned or advertised is None or not str(advertised).strip(): + return info, None + selected = select_release(info, pinned) + if selected is None: + raise BundlerError( + f"{kind} '{component.id}' is pinned to version {pinned} in the bundle " + f"manifest, but its catalog has no release for that version (it " + f"advertises {str(advertised).strip()}). Update the bundle's pinned " + "version or the catalog before installing." + ) + return selected, selected["version"] + + def _bundled_manifest_version(manifest_path: Path, root_key: str) -> str | None: """Best-effort read of a bundled asset's declared version from its manifest. @@ -219,10 +246,12 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: f"Preset '{component.id}' is from a discovery-only catalog; " "installation is not allowed." ) - _assert_pinned_version( - "Preset", component.id, component.version, info.get("version") + from ..presets._catalog_versions import select_release + + info, expected_version = _select_pinned_release( + "Preset", component, info, select_release ) - zip_path = catalog.download_pack(component.id) + zip_path = catalog.download_pack_info(info) try: self._manager.install_from_zip( zip_path, @@ -230,6 +259,11 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: priority, catalog_name=info.get("_catalog_name"), **({"force": True} if force else {}), + **( + {"expected_id": component.id, "expected_version": expected_version} + if expected_version is not None + else {} + ), ) finally: with contextlib.suppress(Exception): @@ -312,10 +346,12 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: f"Extension '{component.id}' is from a discovery-only catalog; " "installation is not allowed." ) - _assert_pinned_version( - "Extension", component.id, component.version, info.get("version") + from ..extensions._catalog_versions import select_release + + info, expected_version = _select_pinned_release( + "Extension", component, info, select_release ) - zip_path = catalog.download_extension(component.id) + zip_path = catalog.download_extension_info(info) try: manifest = self._manager.install_from_zip( zip_path, @@ -323,6 +359,11 @@ def _do_install(self, component: ComponentRef, *, force: bool) -> None: priority=priority, force=force, catalog_name=info.get("_catalog_name"), + **( + {"expected_id": component.id, "expected_version": expected_version} + if expected_version is not None + else {} + ), ) self._manager.scaffold_config(manifest.id) finally: diff --git a/tests/specify_cli/bundles/test_command_install.py b/tests/specify_cli/bundles/test_command_install.py index fb6c81e380..4f1cc538e1 100644 --- a/tests/specify_cli/bundles/test_command_install.py +++ b/tests/specify_cli/bundles/test_command_install.py @@ -409,7 +409,8 @@ def test_local_refresh_catalog_extension_requires_network( version = "1.0.0" downloads = [] - def download_extension(self, extension_id): + def download_extension_info(self, info): + extension_id = info["id"] downloads.append((extension_id, version)) artifact = tmp_path / "extension.zip" extension = { @@ -439,7 +440,9 @@ def download_extension(self, extension_id): "get_extension_info", lambda self, cid: {"id": cid, "version": version, "_install_allowed": True}, ) - monkeypatch.setattr(ExtensionCatalog, "download_extension", download_extension) + monkeypatch.setattr( + ExtensionCatalog, "download_extension_info", download_extension_info + ) data = valid_manifest_dict( provides={"extensions": [{"id": "catalog-ext", "version": version}]} ) diff --git a/tests/specify_cli/bundles/test_primitives.py b/tests/specify_cli/bundles/test_primitives.py index 31c729ab65..9cf6d3adb4 100644 --- a/tests/specify_cli/bundles/test_primitives.py +++ b/tests/specify_cli/bundles/test_primitives.py @@ -220,7 +220,9 @@ def install_from_zip(self, *args, **kwargs): "_catalog_name": "bundle-preset-catalog", }, ) - monkeypatch.setattr(PresetCatalog, "download_pack", lambda _self, _id: archive) + monkeypatch.setattr( + PresetCatalog, "download_pack_info", lambda _self, _info: archive + ) manager = primitive_manager("presets", tmp_path, allow_network=True) manager._manager = _FakeManager() @@ -266,7 +268,7 @@ def scaffold_config(self, extension_id): }, ) monkeypatch.setattr( - ExtensionCatalog, "download_extension", lambda _self, _id: archive + ExtensionCatalog, "download_extension_info", lambda _self, _info: archive ) manager = primitive_manager("extensions", tmp_path, allow_network=True) @@ -411,7 +413,7 @@ def test_catalog_extension_install_scaffolds_config(tmp_path: Path, monkeypatch) lambda self, eid: {"id": eid, "_install_allowed": True}, ) monkeypatch.setattr( - ExtensionCatalog, "download_extension", lambda self, eid: zip_path + ExtensionCatalog, "download_extension_info", lambda self, info: zip_path ) manager = primitive_manager("extensions", project, allow_network=True) @@ -652,3 +654,186 @@ def _boom(step_id, *args, **kwargs): # A rollback must be a rollback: the entry comes back byte-for-byte, not # re-registered with fresh ``installed_at`` / ``updated_at`` stamps. assert restored.get("my-step") == seeded + + +_HISTORICAL_SHA = "a" * 64 + + +def _versioned_entry(cid: str) -> dict: + """A catalog entry advertising 0.5.1 that keeps 0.4.12 under ``releases``.""" + return { + "id": cid, + "name": cid, + "version": "0.5.1", + "download_url": f"https://example.com/{cid}/v0.5.1/{cid}.zip", + "sha256": "b" * 64, + "releases": { + "0.4.12": { + "download_url": f"https://example.com/{cid}/v0.4.12/{cid}.zip", + "sha256": _HISTORICAL_SHA, + } + }, + "_install_allowed": True, + "_catalog_name": "bundle-catalog", + } + + +def _patch_extension_catalog(monkeypatch, entry: dict, archive: Path, downloads: list): + import specify_cli._assets as assets + from specify_cli.extensions import ExtensionCatalog + + monkeypatch.setattr(assets, "_locate_bundled_extension", lambda _id: None) + monkeypatch.setattr( + ExtensionCatalog, "get_extension_info", lambda _self, _id: entry + ) + + def _download(_self, info): + downloads.append(info) + return archive + + monkeypatch.setattr(ExtensionCatalog, "download_extension_info", _download) + + +def _patch_preset_catalog(monkeypatch, entry: dict, archive: Path, downloads: list): + import specify_cli._assets as assets + from specify_cli.presets import PresetCatalog + + monkeypatch.setattr(assets, "_locate_bundled_preset", lambda _id: None) + monkeypatch.setattr(PresetCatalog, "get_pack_info", lambda _self, _id: entry) + + def _download(_self, info): + downloads.append(info) + return archive + + monkeypatch.setattr(PresetCatalog, "download_pack_info", _download) + + +def test_extension_pin_selects_historical_catalog_release(tmp_path: Path, monkeypatch): + """A pin older than the advertised release installs that exact release + (its own URL and digest) instead of refusing the install (#4712).""" + archive = tmp_path / "ext.zip" + archive.write_bytes(b"placeholder") + downloads: list = [] + installs: list = [] + _patch_extension_catalog( + monkeypatch, _versioned_entry("pinned-ext"), archive, downloads + ) + + class _FakeManager: + def install_from_zip(self, *args, **kwargs): + installs.append(kwargs) + return SimpleNamespace(id="pinned-ext") + + def scaffold_config(self, _extension_id): + pass + + manager = primitive_manager("extensions", tmp_path, allow_network=True) + manager._manager = _FakeManager() + manager.install(ComponentRef(kind="extensions", id="pinned-ext", version="0.4.12")) + + assert [d["version"] for d in downloads] == ["0.4.12"] + assert downloads[0]["download_url"].endswith("/v0.4.12/pinned-ext.zip") + assert downloads[0]["sha256"] == _HISTORICAL_SHA + assert installs[0]["expected_id"] == "pinned-ext" + assert installs[0]["expected_version"] == "0.4.12" + assert installs[0]["catalog_name"] == "bundle-catalog" + + +def test_preset_pin_selects_historical_catalog_release(tmp_path: Path, monkeypatch): + archive = tmp_path / "preset.zip" + archive.write_bytes(b"placeholder") + downloads: list = [] + installs: list = [] + _patch_preset_catalog( + monkeypatch, _versioned_entry("pinned-preset"), archive, downloads + ) + + class _FakeManager: + def install_from_zip(self, *args, **kwargs): + installs.append(kwargs) + + manager = primitive_manager("presets", tmp_path, allow_network=True) + manager._manager = _FakeManager() + manager.install(ComponentRef(kind="presets", id="pinned-preset", version="0.4.12")) + + assert [d["version"] for d in downloads] == ["0.4.12"] + assert downloads[0]["download_url"].endswith("/v0.4.12/pinned-preset.zip") + assert downloads[0]["sha256"] == _HISTORICAL_SHA + assert installs[0]["expected_id"] == "pinned-preset" + assert installs[0]["expected_version"] == "0.4.12" + + +@pytest.mark.parametrize("kind", ["extensions", "presets"]) +def test_pin_missing_from_catalog_releases_refuses_before_download( + tmp_path: Path, monkeypatch, kind: str +): + """A pin the winning catalog entry does not carry fails clearly, without + substituting the advertised release.""" + archive = tmp_path / "a.zip" + downloads: list = [] + patch = _patch_extension_catalog if kind == "extensions" else _patch_preset_catalog + patch(monkeypatch, _versioned_entry("c"), archive, downloads) + + manager = primitive_manager(kind, tmp_path, allow_network=True) + with pytest.raises( + BundlerError, + match=r"pinned to version 0\.3\.0 .* no release for that version " + r"\(it advertises 0\.5\.1\)", + ): + manager.install(ComponentRef(kind=kind, id="c", version="0.3.0")) + assert downloads == [] + + +@pytest.mark.parametrize("kind", ["extensions", "presets"]) +def test_unpinned_component_installs_advertised_release_unverified( + tmp_path: Path, monkeypatch, kind: str +): + archive = tmp_path / "a.zip" + archive.write_bytes(b"placeholder") + downloads: list = [] + installs: list = [] + patch = _patch_extension_catalog if kind == "extensions" else _patch_preset_catalog + patch(monkeypatch, _versioned_entry("c"), archive, downloads) + + class _FakeManager: + def install_from_zip(self, *args, **kwargs): + installs.append(kwargs) + return SimpleNamespace(id="c") + + def scaffold_config(self, _extension_id): + pass + + manager = primitive_manager(kind, tmp_path, allow_network=True) + manager._manager = _FakeManager() + manager.install(ComponentRef(kind=kind, id="c")) + + assert [d["version"] for d in downloads] == ["0.5.1"] + assert "expected_version" not in installs[0] + + +def test_extension_pin_refuses_archive_declaring_another_version( + tmp_path: Path, monkeypatch +): + """The selected release's archive must declare the pinned version: a + mislabeled historical asset is refused and nothing is installed.""" + import zipfile + + from specify_cli.extensions import ExtensionError + + project = tmp_path / "project" + ext_source = tmp_path / "ext-source" + _write_extension_with_config(ext_source) # declares my-ext 1.0.0 + zip_path = tmp_path / "my-ext.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + for f in ext_source.rglob("*"): + if f.is_file(): + zf.write(f, f.relative_to(ext_source)) + + entry = _versioned_entry("my-ext") + _patch_extension_catalog(monkeypatch, entry, zip_path, []) + + manager = primitive_manager("extensions", project, allow_network=True) + with pytest.raises(ExtensionError, match="0.4.12"): + manager.install(ComponentRef(kind="extensions", id="my-ext", version="0.4.12")) + assert not manager.is_installed(ComponentRef(kind="extensions", id="my-ext")) + assert not zip_path.exists()