chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,120 @@
|
||||
"""Integration tests for the catalog stack: precedence, policy gating, search."""
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
import pytest
|
||||
|
||||
from specify_cli.bundler import BundlerError
|
||||
from specify_cli.bundler.models.catalog import CatalogSource, InstallPolicy, Scope
|
||||
from specify_cli.bundler.services.catalog_stack import CatalogStack
|
||||
from tests.bundler_helpers import catalog_entry_dict, catalog_payload
|
||||
|
||||
|
||||
def _source(source_id, priority, policy, url="builtin://x"):
|
||||
return CatalogSource(
|
||||
id=source_id, url=url, priority=priority,
|
||||
install_policy=InstallPolicy(policy), scope=Scope.PROJECT,
|
||||
)
|
||||
|
||||
|
||||
def _stack(sources, payloads):
|
||||
def fetcher(src):
|
||||
return payloads[src.id]
|
||||
return CatalogStack(sources, fetcher)
|
||||
|
||||
|
||||
def test_resolve_prefers_highest_precedence_source():
|
||||
sources = [
|
||||
_source("low", 2, "install-allowed"),
|
||||
_source("high", 1, "discovery-only"),
|
||||
]
|
||||
payloads = {
|
||||
"high": catalog_payload({"b": catalog_entry_dict("b", version="9.0.0")}),
|
||||
"low": catalog_payload({"b": catalog_entry_dict("b", version="1.0.0")}),
|
||||
}
|
||||
resolved = _stack(sources, payloads).resolve("b")
|
||||
assert resolved.source.id == "high"
|
||||
assert resolved.entry.version == "9.0.0"
|
||||
assert resolved.install_allowed is False
|
||||
|
||||
|
||||
def test_resolve_unknown_bundle_errors():
|
||||
stack = _stack(
|
||||
[_source("only", 1, "install-allowed")],
|
||||
{"only": catalog_payload({})},
|
||||
)
|
||||
with pytest.raises(BundlerError, match="not found"):
|
||||
stack.resolve("missing")
|
||||
|
||||
|
||||
def test_search_dedupes_by_precedence_and_filters():
|
||||
sources = [_source("a", 1, "install-allowed"), _source("b", 2, "install-allowed")]
|
||||
payloads = {
|
||||
"a": catalog_payload({
|
||||
"alpha": catalog_entry_dict("alpha", role="developer"),
|
||||
}),
|
||||
"b": catalog_payload({
|
||||
"alpha": catalog_entry_dict("alpha", version="0.0.1"),
|
||||
"beta": catalog_entry_dict("beta", role="qa"),
|
||||
}),
|
||||
}
|
||||
stack = _stack(sources, payloads)
|
||||
|
||||
all_results = stack.search()
|
||||
ids = [r.entry.id for r in all_results]
|
||||
assert ids == ["alpha", "beta"]
|
||||
# alpha resolved from the higher-precedence source 'a'.
|
||||
alpha = next(r for r in all_results if r.entry.id == "alpha")
|
||||
assert alpha.source.id == "a"
|
||||
|
||||
qa_only = stack.search("qa")
|
||||
assert [r.entry.id for r in qa_only] == ["beta"]
|
||||
|
||||
|
||||
def test_search_does_not_surface_a_shadowed_lower_precedence_entry():
|
||||
"""Search must resolve each id at its highest-precedence source, then
|
||||
filter — never fall through to a shadowed lower-precedence entry the query
|
||||
happens to match.
|
||||
|
||||
If the query matched only the lower-precedence copy of an id, search used
|
||||
to return that copy, even though `resolve()`/install always use the
|
||||
higher-precedence one. That advertised a bundle (name/version/source) the
|
||||
user could never actually get.
|
||||
"""
|
||||
sources = [_source("high", 1, "install-allowed"), _source("low", 2, "install-allowed")]
|
||||
payloads = {
|
||||
# Highest-precedence entry for 'shared' does NOT match "widget".
|
||||
"high": catalog_payload({
|
||||
"shared": catalog_entry_dict(
|
||||
"shared", name="Alpha Tool", role="developer",
|
||||
description="nothing relevant", version="2.0.0",
|
||||
),
|
||||
}),
|
||||
# Lower-precedence entry for the same id DOES match "widget".
|
||||
"low": catalog_payload({
|
||||
"shared": catalog_entry_dict(
|
||||
"shared", name="Searchable Widget", version="1.0.0",
|
||||
),
|
||||
}),
|
||||
}
|
||||
stack = _stack(sources, payloads)
|
||||
|
||||
# resolve() uses the high-precedence entry.
|
||||
assert stack.resolve("shared").source.id == "high"
|
||||
|
||||
# A query that only the shadowed low-precedence entry matches returns
|
||||
# nothing — search agrees with resolve().
|
||||
assert stack.search("widget") == []
|
||||
|
||||
# And a query the high-precedence entry matches returns it (from 'high').
|
||||
alpha = stack.search("alpha tool")
|
||||
assert [r.entry.id for r in alpha] == ["shared"]
|
||||
assert alpha[0].source.id == "high"
|
||||
|
||||
|
||||
def test_unreachable_source_raises_named_error():
|
||||
def fetcher(src):
|
||||
raise RuntimeError("boom")
|
||||
stack = CatalogStack([_source("bad", 1, "install-allowed")], fetcher)
|
||||
with pytest.raises(BundlerError, match="bad"):
|
||||
stack.resolve("anything")
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Install-time initialization and integration precedence (T049, T050).
|
||||
|
||||
``specify bundle install`` into an uninitialized directory must scaffold a Spec
|
||||
Kit project first (FR-012), choosing the integration by precedence (FR-013):
|
||||
explicit ``--integration`` override → bundle-declared integration → default.
|
||||
The end-to-end test runs fully offline against bundled assets.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from specify_cli import app
|
||||
from specify_cli.bundler.models.manifest import BundleManifest
|
||||
from specify_cli.commands.bundle import _resolve_init_integration
|
||||
from specify_cli.bundler.services.packager import build_bundle
|
||||
from tests.bundler_helpers import valid_manifest_dict
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
def _manifest(**overrides):
|
||||
data = valid_manifest_dict(**overrides)
|
||||
return BundleManifest.from_dict(data)
|
||||
|
||||
|
||||
def test_precedence_override_wins():
|
||||
manifest = _manifest(integration={"id": "claude"})
|
||||
assert _resolve_init_integration("gemini", manifest) == "gemini"
|
||||
|
||||
|
||||
def test_precedence_bundle_declared_when_no_override():
|
||||
manifest = _manifest(integration={"id": "claude"})
|
||||
assert _resolve_init_integration(None, manifest) == "claude"
|
||||
|
||||
|
||||
def test_precedence_default_when_unspecified():
|
||||
manifest = _manifest()
|
||||
assert _resolve_init_integration(None, manifest) == "copilot"
|
||||
assert _resolve_init_integration(None, None) == "copilot"
|
||||
|
||||
|
||||
def _build_mini(tmp_path: Path) -> Path:
|
||||
bundle = tmp_path / "mini"
|
||||
bundle.mkdir()
|
||||
(bundle / "bundle.yml").write_text(
|
||||
yaml.safe_dump(
|
||||
{
|
||||
"schema_version": "1.0",
|
||||
"bundle": {
|
||||
"id": "mini",
|
||||
"name": "Mini",
|
||||
"version": "1.0.0",
|
||||
"role": "developer",
|
||||
"description": "minimal",
|
||||
"author": "tests",
|
||||
"license": "MIT",
|
||||
},
|
||||
"requires": {"speckit_version": ">=0.1.0"},
|
||||
"provides": {"extensions": [{"id": "agent-context", "version": "1.0.0"}]},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(bundle / "README.md").write_text("# Mini\n", encoding="utf-8")
|
||||
return build_bundle(bundle).artifact_path
|
||||
|
||||
|
||||
def test_install_initializes_uninitialized_project(tmp_path: Path):
|
||||
project = tmp_path / "proj"
|
||||
project.mkdir()
|
||||
artifact = _build_mini(tmp_path)
|
||||
|
||||
previous = Path.cwd()
|
||||
os.chdir(project)
|
||||
try:
|
||||
result = runner.invoke(
|
||||
app, ["bundle", "install", str(artifact), "--offline"]
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
finally:
|
||||
os.chdir(previous)
|
||||
|
||||
assert (project / ".specify").is_dir()
|
||||
marker = project / ".specify" / "integration.json"
|
||||
assert marker.exists()
|
||||
data = json.loads(marker.read_text(encoding="utf-8"))
|
||||
assert "copilot" in json.dumps(data)
|
||||
@@ -0,0 +1,287 @@
|
||||
"""Integration tests for the install → record → remove lifecycle (offline, fake installer).
|
||||
|
||||
Uses :class:`FakeInstaller` so no network or real primitive machinery is touched
|
||||
(Constitution Principle II network-mocking, Principle IV offline-first).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from specify_cli.bundler import BundlerError
|
||||
from specify_cli.bundler.models.manifest import BundleManifest
|
||||
from specify_cli.bundler.models.records import load_records
|
||||
from specify_cli.bundler.services.installer import install_bundle, remove_bundle
|
||||
from specify_cli.bundler.services.resolver import resolve_install_plan
|
||||
from tests.bundler_helpers import FakeInstaller, make_project, valid_manifest_dict
|
||||
|
||||
|
||||
def _plan(manifest):
|
||||
return resolve_install_plan(
|
||||
manifest, speckit_version="0.11.2", active_integration="copilot"
|
||||
)
|
||||
|
||||
|
||||
def test_install_records_and_invokes_primitives(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
|
||||
result = install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
|
||||
assert len(result.installed) == 4
|
||||
assert len(installer.install_calls) == 4
|
||||
records = load_records(tmp_path)
|
||||
assert len(records) == 1
|
||||
assert records[0].bundle_id == "demo-bundle"
|
||||
|
||||
|
||||
def test_install_is_idempotent(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
second = install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
|
||||
# Second install adds nothing and does not duplicate the record.
|
||||
assert second.installed == []
|
||||
assert len(second.skipped) == 4
|
||||
assert len(load_records(tmp_path)) == 1
|
||||
|
||||
|
||||
def test_partial_failure_rolls_back_and_records_nothing(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller(fail_on="preset-a")
|
||||
|
||||
with pytest.raises(BundlerError):
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
|
||||
# ext-a was installed first, then rolled back; no record persisted.
|
||||
assert installer.installed == set()
|
||||
assert load_records(tmp_path) == []
|
||||
|
||||
|
||||
def test_remove_is_non_collateral(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
installer = FakeInstaller()
|
||||
|
||||
# Bundle A provides a shared preset; Bundle B also provides it.
|
||||
data_a = valid_manifest_dict()
|
||||
data_a["bundle"]["id"] = "a"
|
||||
data_b = valid_manifest_dict()
|
||||
data_b["bundle"]["id"] = "b"
|
||||
data_b["provides"] = {"presets": [
|
||||
{"id": "preset-a", "version": "2.0.0", "priority": 10, "strategy": "append"}
|
||||
]}
|
||||
|
||||
man_a = BundleManifest.from_dict(data_a)
|
||||
man_b = BundleManifest.from_dict(data_b)
|
||||
install_bundle(tmp_path, _plan(man_a), installer, manifest=man_a)
|
||||
install_bundle(tmp_path, _plan(man_b), installer, manifest=man_b)
|
||||
|
||||
# Removing B must NOT uninstall preset-a (still needed by A).
|
||||
result = remove_bundle(tmp_path, "b", installer)
|
||||
assert ("presets", "preset-a") in {(c.kind, c.id) for c in result.skipped}
|
||||
assert installer.is_installed(tmp_path, man_a.presets[0]) is True
|
||||
|
||||
remaining = {r.bundle_id for r in load_records(tmp_path)}
|
||||
assert remaining == {"a"}
|
||||
|
||||
|
||||
def test_remove_unknown_bundle_errors(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
with pytest.raises(BundlerError, match="not installed"):
|
||||
remove_bundle(tmp_path, "ghost", FakeInstaller())
|
||||
|
||||
|
||||
def test_remove_reports_uninstalled_not_installed(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
|
||||
result = remove_bundle(tmp_path, "demo-bundle", installer)
|
||||
|
||||
# Removal flows populate the dedicated ``uninstalled`` list; ``installed``
|
||||
# stays empty so the result type is never ambiguous for callers.
|
||||
assert result.installed == []
|
||||
assert len(result.uninstalled) == 4
|
||||
assert installer.installed == set()
|
||||
|
||||
|
||||
def test_remove_counts_only_components_actually_removed(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
|
||||
# Simulate one contributed component already gone from disk (e.g. removed
|
||||
# out of band). It must not be reported as uninstalled and remove() must
|
||||
# not be called for it.
|
||||
gone = manifest.components[0]
|
||||
installer.installed.discard((gone.kind, gone.id))
|
||||
|
||||
result = remove_bundle(tmp_path, "demo-bundle", installer)
|
||||
|
||||
assert len(result.uninstalled) == 3
|
||||
assert (gone.kind, gone.id) not in installer.remove_calls
|
||||
assert gone in result.skipped
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
result = install_bundle(
|
||||
tmp_path, _plan(manifest), installer, manifest=manifest, refresh=True
|
||||
)
|
||||
|
||||
# With refresh, already-installed components are re-applied, not skipped.
|
||||
assert result.skipped == []
|
||||
assert len(result.refreshed) == 4
|
||||
assert len(installer.refresh_calls) == 4
|
||||
assert result.changed is True
|
||||
|
||||
|
||||
def test_refresh_falls_back_to_install_without_hook(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
|
||||
class NoRefreshInstaller(FakeInstaller):
|
||||
refresh = None # type: ignore[assignment]
|
||||
|
||||
installer = NoRefreshInstaller()
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
before = len(installer.install_calls)
|
||||
result = install_bundle(
|
||||
tmp_path, _plan(manifest), installer, manifest=manifest, refresh=True
|
||||
)
|
||||
|
||||
# No refresh hook → re-install path keeps components current.
|
||||
assert len(result.refreshed) == 4
|
||||
assert len(installer.install_calls) == before + 4
|
||||
|
||||
|
||||
def test_update_preserves_original_installed_at(tmp_path: Path):
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
|
||||
original = load_records(tmp_path)[0].installed_at
|
||||
|
||||
# A refresh (bundle update) must not rewrite the original install timestamp.
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest, refresh=True)
|
||||
|
||||
assert load_records(tmp_path)[0].installed_at == original
|
||||
|
||||
|
||||
def test_refresh_does_not_touch_independently_installed_component(tmp_path: Path):
|
||||
# bundle update (refresh) must not re-apply a component installed
|
||||
# independently and tracked by no bundle — refreshing it would be a
|
||||
# collateral change to something the bundle does not own (FR-022).
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
installer.installed.add(("extensions", "ext-a"))
|
||||
|
||||
result = install_bundle(
|
||||
tmp_path, _plan(manifest), installer, manifest=manifest, refresh=True
|
||||
)
|
||||
|
||||
# ext-a is skipped (not refreshed) and never attributed to the bundle.
|
||||
assert ("extensions", "ext-a") not in installer.refresh_calls
|
||||
assert ("extensions", "ext-a") in {(c.kind, c.id) for c in result.skipped}
|
||||
assert ("extensions", "ext-a") not in {(c.kind, c.id) for c in result.refreshed}
|
||||
contributed = {
|
||||
(c.kind, c.id) for c in load_records(tmp_path)[0].contributed_components
|
||||
}
|
||||
assert ("extensions", "ext-a") not in contributed
|
||||
|
||||
|
||||
def test_pre_existing_component_is_not_attributed_or_removed(tmp_path: Path):
|
||||
# A component installed independently (before any bundle) must not be
|
||||
# attributed to the bundle, so removing the bundle never uninstalls it
|
||||
# (FR-022, no collateral removal).
|
||||
make_project(tmp_path)
|
||||
manifest = BundleManifest.from_dict(valid_manifest_dict())
|
||||
installer = FakeInstaller()
|
||||
# Pre-install ext-a independently — no bundle record references it yet.
|
||||
installer.installed.add(("extensions", "ext-a"))
|
||||
|
||||
install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
|
||||
|
||||
contributed = {
|
||||
(c.kind, c.id) for c in load_records(tmp_path)[0].contributed_components
|
||||
}
|
||||
assert ("extensions", "ext-a") not in contributed
|
||||
|
||||
remove_bundle(tmp_path, "demo-bundle", installer)
|
||||
assert ("extensions", "ext-a") in installer.installed
|
||||
|
||||
|
||||
def _bundle(manifest_id, ext_ids, *, version="1.0.0"):
|
||||
data = valid_manifest_dict()
|
||||
data["bundle"]["id"] = manifest_id
|
||||
data["bundle"]["version"] = version
|
||||
data["provides"] = {
|
||||
"extensions": [{"id": e, "version": version} for e in ext_ids]
|
||||
}
|
||||
return BundleManifest.from_dict(data)
|
||||
|
||||
|
||||
def test_update_uninstalls_components_dropped_by_new_version(tmp_path: Path):
|
||||
"""`bundle update` must uninstall components the new version no longer
|
||||
ships, instead of orphaning them (installed on disk, tracked by nothing)."""
|
||||
make_project(tmp_path)
|
||||
installer = FakeInstaller()
|
||||
|
||||
man_v1 = _bundle("demo", ["ext-a", "ext-b"])
|
||||
install_bundle(tmp_path, _plan(man_v1), installer, manifest=man_v1)
|
||||
assert ("extensions", "ext-b") in installer.installed
|
||||
|
||||
man_v2 = _bundle("demo", ["ext-a"], version="2.0.0")
|
||||
result = install_bundle(
|
||||
tmp_path, _plan(man_v2), installer, manifest=man_v2, refresh=True
|
||||
)
|
||||
|
||||
# ext-b was dropped by v2 -> uninstalled and reported.
|
||||
assert ("extensions", "ext-b") in installer.remove_calls
|
||||
assert ("extensions", "ext-b") in {(c.kind, c.id) for c in result.uninstalled}
|
||||
assert ("extensions", "ext-b") not in installer.installed
|
||||
assert ("extensions", "ext-a") in installer.installed
|
||||
|
||||
# The saved record lists only ext-a.
|
||||
rec = next(r for r in load_records(tmp_path) if r.bundle_id == "demo")
|
||||
keys = {(c.kind, c.id) for c in rec.contributed_components}
|
||||
assert ("extensions", "ext-a") in keys
|
||||
assert ("extensions", "ext-b") not in keys
|
||||
|
||||
|
||||
def test_update_keeps_component_still_needed_by_sibling_bundle(tmp_path: Path):
|
||||
"""A dropped component still owned by another bundle stays installed."""
|
||||
make_project(tmp_path)
|
||||
installer = FakeInstaller()
|
||||
|
||||
man_sib = _bundle("sibling", ["ext-b"])
|
||||
install_bundle(tmp_path, _plan(man_sib), installer, manifest=man_sib)
|
||||
|
||||
man_v1 = _bundle("demo", ["ext-a", "ext-b"])
|
||||
install_bundle(tmp_path, _plan(man_v1), installer, manifest=man_v1)
|
||||
|
||||
man_v2 = _bundle("demo", ["ext-a"], version="2.0.0")
|
||||
install_bundle(
|
||||
tmp_path, _plan(man_v2), installer, manifest=man_v2, refresh=True
|
||||
)
|
||||
|
||||
# ext-b is still needed by 'sibling' -> not removed, stays installed.
|
||||
assert ("extensions", "ext-b") not in installer.remove_calls
|
||||
assert ("extensions", "ext-b") in installer.installed
|
||||
|
||||
# But demo's record no longer attributes it.
|
||||
rec = next(r for r in load_records(tmp_path) if r.bundle_id == "demo")
|
||||
assert ("extensions", "ext-b") not in {
|
||||
(c.kind, c.id) for c in rec.contributed_components
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
"""Tests for installing a bundle from a local artifact/path (T045).
|
||||
|
||||
The resolution-level tests are pure; the end-to-end test installs the bundled
|
||||
``agent-context`` extension fully offline from a built ``.zip`` artifact,
|
||||
proving the real in-process primitive dispatch (T044) works without a network.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from specify_cli import app
|
||||
from specify_cli.bundler import BundlerError
|
||||
from specify_cli.commands.bundle import _local_manifest_source
|
||||
from tests.bundler_helpers import make_project, valid_manifest_dict, write_manifest
|
||||
|
||||
|
||||
def test_local_source_none_for_non_path():
|
||||
assert _local_manifest_source("some-catalog-bundle-id") is None
|
||||
|
||||
|
||||
def test_local_source_from_directory(tmp_path: Path):
|
||||
write_manifest(tmp_path, valid_manifest_dict())
|
||||
manifest = _local_manifest_source(str(tmp_path))
|
||||
assert manifest is not None
|
||||
assert manifest.bundle.id == "demo-bundle"
|
||||
|
||||
|
||||
def test_local_source_from_bundle_yml(tmp_path: Path):
|
||||
path = write_manifest(tmp_path, valid_manifest_dict())
|
||||
manifest = _local_manifest_source(str(path))
|
||||
assert manifest is not None
|
||||
assert manifest.bundle.id == "demo-bundle"
|
||||
|
||||
|
||||
def test_local_source_from_zip_artifact(tmp_path: Path):
|
||||
bundle_dir = tmp_path / "bundle"
|
||||
bundle_dir.mkdir()
|
||||
write_manifest(bundle_dir, valid_manifest_dict())
|
||||
(bundle_dir / "README.md").write_text("# demo\n", encoding="utf-8")
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(app, ["bundle", "build", "--path", str(bundle_dir)])
|
||||
assert result.exit_code == 0, result.output
|
||||
artifact = next(bundle_dir.glob("*.zip"))
|
||||
|
||||
manifest = _local_manifest_source(str(artifact))
|
||||
assert manifest is not None
|
||||
assert manifest.bundle.id == "demo-bundle"
|
||||
|
||||
|
||||
def test_local_source_rejects_unknown_file(tmp_path: Path):
|
||||
weird = tmp_path / "thing.txt"
|
||||
weird.write_text("nope", encoding="utf-8")
|
||||
with pytest.raises(BundlerError, match="not a recognised bundle source"):
|
||||
_local_manifest_source(str(weird))
|
||||
|
||||
|
||||
def test_install_bundled_extension_from_zip_offline(tmp_path: Path):
|
||||
"""End-to-end: build → install (offline, local .zip) → list → remove."""
|
||||
project = make_project(tmp_path / "proj")
|
||||
|
||||
bundle_dir = tmp_path / "mini"
|
||||
bundle_dir.mkdir()
|
||||
(bundle_dir / "bundle.yml").write_text(
|
||||
yaml.safe_dump(
|
||||
{
|
||||
"schema_version": "1.0",
|
||||
"bundle": {
|
||||
"id": "mini",
|
||||
"name": "Mini",
|
||||
"version": "1.0.0",
|
||||
"role": "developer",
|
||||
"description": "minimal",
|
||||
"author": "tests",
|
||||
"license": "MIT",
|
||||
},
|
||||
"requires": {"speckit_version": ">=0.1.0"},
|
||||
"provides": {
|
||||
"extensions": [{"id": "agent-context", "version": "1.0.0"}]
|
||||
},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(bundle_dir / "README.md").write_text("# Mini\n", encoding="utf-8")
|
||||
|
||||
runner = CliRunner()
|
||||
previous = Path.cwd()
|
||||
os.chdir(project)
|
||||
try:
|
||||
build = runner.invoke(app, ["bundle", "build", "--path", str(bundle_dir)])
|
||||
assert build.exit_code == 0, build.output
|
||||
artifact = next(bundle_dir.glob("*.zip"))
|
||||
|
||||
install = runner.invoke(app, ["bundle", "install", str(artifact), "--offline"])
|
||||
assert install.exit_code == 0, install.output
|
||||
|
||||
from specify_cli.extensions import ExtensionManager
|
||||
|
||||
assert ExtensionManager(project).registry.is_installed("agent-context")
|
||||
|
||||
listing = runner.invoke(app, ["bundle", "list"])
|
||||
assert "mini" in listing.output
|
||||
|
||||
remove = runner.invoke(app, ["bundle", "remove", "mini"])
|
||||
assert remove.exit_code == 0, remove.output
|
||||
assert not ExtensionManager(project).registry.is_installed("agent-context")
|
||||
finally:
|
||||
os.chdir(previous)
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Offline-first tests (Constitution Principle IV).
|
||||
|
||||
Assert that consume/author flows work with no network access: built-in catalogs
|
||||
resolve offline, file:// catalogs resolve offline, and http(s) sources are
|
||||
refused (never silently attempted) when network is disabled.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from specify_cli.bundler import BundlerError
|
||||
from specify_cli.bundler.models.catalog import CatalogSource, InstallPolicy, Scope
|
||||
from specify_cli.bundler.services.adapters import make_catalog_fetcher
|
||||
from specify_cli.bundler.services.catalog_stack import CatalogStack
|
||||
from tests.bundler_helpers import catalog_entry_dict, write_catalog_file
|
||||
|
||||
|
||||
def _src(source_id, url, priority=1, policy="install-allowed"):
|
||||
return CatalogSource(
|
||||
id=source_id, url=url, priority=priority,
|
||||
install_policy=InstallPolicy(policy), scope=Scope.PROJECT,
|
||||
)
|
||||
|
||||
|
||||
def test_builtin_catalog_resolves_offline():
|
||||
fetcher = make_catalog_fetcher(allow_network=False)
|
||||
stack = CatalogStack([_src("default", "builtin://default")], fetcher)
|
||||
# Built-in default ships empty; search works without network and returns [].
|
||||
assert stack.search() == []
|
||||
|
||||
|
||||
def test_file_catalog_resolves_offline(tmp_path: Path):
|
||||
catalog_path = tmp_path / "catalog.json"
|
||||
write_catalog_file(catalog_path, {"demo": catalog_entry_dict("demo")})
|
||||
fetcher = make_catalog_fetcher(allow_network=False)
|
||||
stack = CatalogStack([_src("local", str(catalog_path))], fetcher)
|
||||
resolved = stack.resolve("demo")
|
||||
assert resolved.entry.id == "demo"
|
||||
|
||||
|
||||
def test_http_source_refused_when_offline():
|
||||
fetcher = make_catalog_fetcher(allow_network=False)
|
||||
stack = CatalogStack([_src("remote", "https://example.com/catalog.json")], fetcher)
|
||||
with pytest.raises(BundlerError, match="Network access disabled"):
|
||||
stack.resolve("anything")
|
||||
|
||||
|
||||
def test_missing_file_catalog_errors_offline(tmp_path: Path):
|
||||
fetcher = make_catalog_fetcher(allow_network=False)
|
||||
stack = CatalogStack([_src("local", str(tmp_path / "nope.json"))], fetcher)
|
||||
with pytest.raises(BundlerError):
|
||||
stack.resolve("anything")
|
||||
|
||||
|
||||
def test_file_url_catalog_resolves_offline(tmp_path: Path):
|
||||
catalog_path = tmp_path / "catalog.json"
|
||||
write_catalog_file(catalog_path, {"demo": catalog_entry_dict("demo")})
|
||||
fetcher = make_catalog_fetcher(allow_network=False)
|
||||
stack = CatalogStack([_src("local", catalog_path.as_uri())], fetcher)
|
||||
resolved = stack.resolve("demo")
|
||||
assert resolved.entry.id == "demo"
|
||||
|
||||
|
||||
def test_plain_http_remote_rejected_before_network():
|
||||
# HTTPS is required for non-localhost catalogs; reject http:// up front.
|
||||
fetcher = make_catalog_fetcher(allow_network=True)
|
||||
stack = CatalogStack([_src("remote", "http://example.com/catalog.json")], fetcher)
|
||||
with pytest.raises(BundlerError, match="must use HTTPS"):
|
||||
stack.resolve("anything")
|
||||
|
||||
|
||||
def test_remote_url_without_host_rejected():
|
||||
fetcher = make_catalog_fetcher(allow_network=True)
|
||||
stack = CatalogStack([_src("remote", "https:///catalog.json")], fetcher)
|
||||
with pytest.raises(BundlerError, match="valid URL with a host"):
|
||||
stack.resolve("anything")
|
||||
@@ -0,0 +1,192 @@
|
||||
"""Security tests: path-traversal / symlink confinement (Constitution Principle V).
|
||||
|
||||
These assert the bundler refuses to read or write outside an allowed root, so a
|
||||
malicious manifest or artifact path cannot escape the project/bundle directory.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from specify_cli.bundler import BundlerError
|
||||
from specify_cli.bundler.lib.yamlio import ensure_within, is_safe_relpath
|
||||
|
||||
|
||||
def test_ensure_within_allows_child(tmp_path: Path):
|
||||
root = tmp_path / "bundle"
|
||||
root.mkdir()
|
||||
child = root / "sub" / "file.txt"
|
||||
assert ensure_within(root, child) == child.resolve()
|
||||
|
||||
|
||||
def test_ensure_within_rejects_parent_traversal(tmp_path: Path):
|
||||
root = tmp_path / "bundle"
|
||||
root.mkdir()
|
||||
escape = root / ".." / "secret.txt"
|
||||
with pytest.raises(BundlerError, match="escapes"):
|
||||
ensure_within(root, escape)
|
||||
|
||||
|
||||
def test_ensure_within_rejects_absolute_outside(tmp_path: Path):
|
||||
root = tmp_path / "bundle"
|
||||
root.mkdir()
|
||||
with pytest.raises(BundlerError):
|
||||
ensure_within(root, Path("/etc/passwd"))
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="symlink semantics differ on Windows")
|
||||
def test_ensure_within_rejects_symlink_escape(tmp_path: Path):
|
||||
root = tmp_path / "bundle"
|
||||
root.mkdir()
|
||||
outside = tmp_path / "outside.txt"
|
||||
outside.write_text("secret", encoding="utf-8")
|
||||
link = root / "link.txt"
|
||||
link.symlink_to(outside)
|
||||
with pytest.raises(BundlerError, match="escapes"):
|
||||
ensure_within(root, link)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("rel,safe", [
|
||||
("a/b.txt", True),
|
||||
("./a.txt", True),
|
||||
("../escape", False),
|
||||
("a/../../escape", False),
|
||||
("/abs", False),
|
||||
("C:/abs", False),
|
||||
("C:\\abs", False),
|
||||
("\\\\server\\share", False),
|
||||
("", False),
|
||||
])
|
||||
def test_is_safe_relpath(rel, safe):
|
||||
assert is_safe_relpath(rel) is safe
|
||||
|
||||
|
||||
def test_build_skips_symlinks(tmp_path: Path):
|
||||
"""Packager must not follow symlinks out of the bundle dir."""
|
||||
import yaml
|
||||
|
||||
from specify_cli.bundler.services.packager import build_bundle
|
||||
from tests.bundler_helpers import valid_manifest_dict
|
||||
|
||||
bundle = tmp_path / "bundle"
|
||||
bundle.mkdir()
|
||||
(bundle / "bundle.yml").write_text(
|
||||
yaml.safe_dump(valid_manifest_dict()), encoding="utf-8"
|
||||
)
|
||||
(bundle / "README.md").write_text("# Demo", encoding="utf-8")
|
||||
|
||||
if os.name != "nt":
|
||||
secret = tmp_path / "secret.txt"
|
||||
secret.write_text("top secret", encoding="utf-8")
|
||||
(bundle / "leak.txt").symlink_to(secret)
|
||||
|
||||
result = build_bundle(bundle, output_dir=tmp_path / "out")
|
||||
import zipfile
|
||||
|
||||
with zipfile.ZipFile(result.artifact_path) as archive:
|
||||
names = archive.namelist()
|
||||
assert "leak.txt" not in names
|
||||
assert "bundle.yml" in names
|
||||
|
||||
|
||||
def test_load_records_refuses_symlinked_specify_escape(tmp_path: Path):
|
||||
# Reading bundle-records.json must honour the same confinement as writes:
|
||||
# a symlinked .specify pointing outside project_root is refused.
|
||||
from specify_cli.bundler.models.records import load_records
|
||||
|
||||
project = tmp_path / "proj"
|
||||
project.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
(outside / "bundle-records.json").write_text(
|
||||
'{"schema_version": "1.0", "bundles": []}', encoding="utf-8"
|
||||
)
|
||||
(project / ".specify").symlink_to(outside, target_is_directory=True)
|
||||
|
||||
with pytest.raises(BundlerError, match="escapes the allowed root"):
|
||||
load_records(project)
|
||||
|
||||
|
||||
def test_active_integration_refuses_symlinked_specify_escape(tmp_path: Path):
|
||||
# Reading the integration marker must not follow a .specify symlink that
|
||||
# resolves outside project_root; an escape is treated as "not determinable".
|
||||
from specify_cli.bundler.lib.project import active_integration
|
||||
|
||||
project = tmp_path / "proj"
|
||||
project.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
(outside / "integration.json").write_text(
|
||||
'{"integration": "leaked"}', encoding="utf-8"
|
||||
)
|
||||
(project / ".specify").symlink_to(outside, target_is_directory=True)
|
||||
|
||||
assert active_integration(project) is None
|
||||
|
||||
|
||||
def test_read_catalog_config_refuses_symlinked_specify_escape(tmp_path: Path):
|
||||
from specify_cli.bundler.commands_impl import catalog_config as cc
|
||||
|
||||
project = tmp_path / "proj"
|
||||
project.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
(outside / "bundle-catalogs.yml").write_text(
|
||||
"schema_version: '1.0'\ncatalogs: []\n", encoding="utf-8"
|
||||
)
|
||||
(project / ".specify").symlink_to(outside, target_is_directory=True)
|
||||
|
||||
with pytest.raises(BundlerError, match="escapes the allowed root"):
|
||||
cc._read(project)
|
||||
|
||||
|
||||
def test_load_source_stack_refuses_symlinked_specify_dir(tmp_path: Path):
|
||||
from specify_cli.bundler.models.catalog import load_source_stack
|
||||
|
||||
project = tmp_path / "project"
|
||||
project.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
(outside / "bundle-catalogs.yml").write_text("catalogs: []\n", encoding="utf-8")
|
||||
try:
|
||||
(project / ".specify").symlink_to(outside, target_is_directory=True)
|
||||
except (OSError, NotImplementedError):
|
||||
pytest.skip("symlinks not supported on this platform")
|
||||
with pytest.raises(BundlerError, match="escapes the allowed root"):
|
||||
load_source_stack(project)
|
||||
|
||||
|
||||
def test_find_project_root_ignores_symlinked_specify(tmp_path: Path):
|
||||
from specify_cli.bundler.lib.project import find_project_root
|
||||
|
||||
real = tmp_path / "real-specify"
|
||||
real.mkdir()
|
||||
project = tmp_path / "project"
|
||||
project.mkdir()
|
||||
try:
|
||||
(project / ".specify").symlink_to(real, target_is_directory=True)
|
||||
except (OSError, NotImplementedError):
|
||||
pytest.skip("symlinks not supported on this platform")
|
||||
# A symlinked .specify must not be accepted as a project root.
|
||||
assert find_project_root(project) is None
|
||||
|
||||
|
||||
def test_find_project_root_override_errors_on_symlinked_specify(tmp_path: Path, monkeypatch):
|
||||
"""The SPECIFY_INIT_DIR override path refuses a symlinked .specify too,
|
||||
matching the cwd loop path (regression: the override returned early and
|
||||
skipped the symlink guard)."""
|
||||
from specify_cli.bundler.lib.project import find_project_root
|
||||
|
||||
real = tmp_path / "real-specify"
|
||||
real.mkdir()
|
||||
project = tmp_path / "project"
|
||||
project.mkdir()
|
||||
try:
|
||||
(project / ".specify").symlink_to(real, target_is_directory=True)
|
||||
except (OSError, NotImplementedError):
|
||||
pytest.skip("symlinks not supported on this platform")
|
||||
monkeypatch.setenv("SPECIFY_INIT_DIR", str(project))
|
||||
with pytest.raises(BundlerError, match="symlinked \\.specify"):
|
||||
find_project_root(None)
|
||||
Reference in New Issue
Block a user