test: consolidate build contracts around executed artifacts
This commit is contained in:
@@ -1,114 +1,27 @@
|
||||
import { spawnSync } from 'node:child_process'
|
||||
import path from 'node:path'
|
||||
import fs from 'node:fs'
|
||||
import os from 'node:os'
|
||||
import path from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { expect, test, vi } from 'vitest'
|
||||
import { generateIcons } from '../scripts/generate-icons.mjs'
|
||||
|
||||
test('icon builds delegate environment preparation to PM using the prepared Python', () => {
|
||||
const run = vi.fn(() => ({ status: 0 }))
|
||||
const root = path.resolve('icon-build-fixture')
|
||||
const env = { PATH: 'tools', HERMES_PYTHON: 'prepared-python', VIRTUAL_ENV: 'runtime-venv', PYTHONPATH: 'payload-libraries', PYTHONHOME: 'payload-python' }
|
||||
expect(generateIcons(['--check'], { root, run, env })).toBe(0)
|
||||
expect(run).toHaveBeenCalledExactlyOnceWith(env.HERMES_PYTHON, [
|
||||
path.join(root, 'scripts', 'build', 'icon_environment.py'), '--source', root, '--out', root, '--check'
|
||||
], { cwd: root, stdio: 'inherit', windowsHide: true, env: { PATH: 'tools', HERMES_PYTHON: env.HERMES_PYTHON, VIRTUAL_ENV: 'runtime-venv' } })
|
||||
expect(env.PYTHONPATH).toBe('payload-libraries')
|
||||
})
|
||||
|
||||
test('icon preparation passes explicit source and independent output roots', () => {
|
||||
const run = vi.fn(() => ({ status: 0 }))
|
||||
const source = path.resolve('source with spaces')
|
||||
const out = path.resolve('icon outputs')
|
||||
expect(generateIcons(['--source', source, '--out', out], { run, env: {} })).toBe(0)
|
||||
const [python, args, options] = run.mock.calls[0]
|
||||
expect(python).toBe('python')
|
||||
expect(args.slice(-4)).toEqual(['--source', source, '--out', out])
|
||||
expect(options.cwd).toBe(source)
|
||||
})
|
||||
|
||||
test('on-demand icon preparation preserves admission intent at the PM boundary', () => {
|
||||
const run = vi.fn(() => ({ status: 0 }))
|
||||
expect(generateIcons(['--on-demand'], { run, env: {} })).toBe(0)
|
||||
expect(run.mock.calls[0][1]).toContain('--on-demand')
|
||||
})
|
||||
|
||||
test('the PM driver owns a temporary group-only environment and preserves generator argv and status', () => {
|
||||
const result = spawnSync(process.env.HERMES_PYTHON || 'python', ['-c', `
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
from scripts.build import icon_environment
|
||||
|
||||
with TemporaryDirectory() as directory:
|
||||
source = Path(directory) / 'source with spaces'
|
||||
source.mkdir()
|
||||
argv = ['--source', str(source), '--out', str(Path(directory) / 'icon outputs'), '--check']
|
||||
outputs = []
|
||||
explicit = True
|
||||
def build(**options):
|
||||
output = options.pop('out')
|
||||
assert output.parent.is_dir() and not output.exists()
|
||||
assert output.name == 'venv'
|
||||
assert options == dict(source=source, groups=['icon-build'], only_groups=True,
|
||||
explicit=explicit, cache=source / '.cache/icon-build')
|
||||
outputs.append(output)
|
||||
return Path(sys.executable)
|
||||
def generate(command, **options):
|
||||
assert outputs[0].parent.is_dir()
|
||||
assert command == [sys.executable, '-I', str(Path(icon_environment.__file__).resolve().parents[1] / 'generate_icons.py'), *argv]
|
||||
assert options == dict(cwd=source)
|
||||
return subprocess.CompletedProcess(command, 7)
|
||||
with patch.object(icon_environment.pm, 'build_environment', side_effect=build) as prepare, patch.object(icon_environment.subprocess, 'run', side_effect=generate) as run:
|
||||
assert icon_environment.main(argv) == 7
|
||||
prepare.assert_called_once()
|
||||
run.assert_called_once()
|
||||
explicit = False
|
||||
outputs.clear()
|
||||
assert icon_environment.main([*argv, '--on-demand']) == 7
|
||||
assert not outputs[0].parent.exists()
|
||||
`], { cwd: fileURLToPath(new URL('..', import.meta.url)), encoding: 'utf8' })
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.status, result.stderr).toBe(0)
|
||||
})
|
||||
|
||||
test.each([
|
||||
{ HERMES_PAYLOAD_TAG: 'v1.2.3', HERMES_BUILD_COMMIT: '' },
|
||||
{ HERMES_PAYLOAD_TAG: 'v1.2.3-canary.20260911010203', HERMES_BUILD_COMMIT: '' },
|
||||
{ HERMES_PAYLOAD_TAG: '', HERMES_BUILD_COMMIT: 'abcdef0'.padEnd(40, '1') }
|
||||
])('build identity reaches the child unchanged with separate source/output roots: %j', (identity) => {
|
||||
const root = fs.mkdtempSync(path.join(os.tmpdir(), 'hermes-icon-env-'))
|
||||
test('the real wrapper builds and checks native icon artifacts in an independent output', () => {
|
||||
const out = fs.mkdtempSync(path.join(os.tmpdir(), 'hermes-icons-'))
|
||||
const env = { ...process.env, HERMES_HOME: path.join(out, 'home'),
|
||||
HERMES_RUNTIME_DIR: path.join(out, 'tools'), HERMES_PAYLOAD_TAG: 'v1.2.3', HERMES_BUILD_COMMIT: '',
|
||||
PYTHONPATH: path.join(out, 'foreign-site'), PYTHONHOME: path.join(out, 'foreign-python') }
|
||||
try {
|
||||
const source = path.join(root, 'separate source')
|
||||
const out = path.join(root, 'generated output')
|
||||
const driver = path.join(root, 'scripts', 'build')
|
||||
fs.mkdirSync(driver, { recursive: true })
|
||||
fs.mkdirSync(source)
|
||||
// Observe the real subprocess environment at the PM-driver boundary.
|
||||
fs.writeFileSync(path.join(driver, 'icon_environment.py'), `
|
||||
import json, os, pathlib, sys
|
||||
out = pathlib.Path(sys.argv[sys.argv.index('--out') + 1])
|
||||
out.mkdir()
|
||||
(out / 'child.json').write_text(json.dumps({
|
||||
'identity': {key: os.environ.get(key) for key in ['HERMES_PAYLOAD_TAG', 'HERMES_BUILD_COMMIT']},
|
||||
'source': sys.argv[sys.argv.index('--source') + 1],
|
||||
'cwd': os.getcwd(),
|
||||
}))
|
||||
`)
|
||||
expect(generateIcons(['--source', source, '--out', out], {
|
||||
root, env: { ...process.env, ...identity }
|
||||
})).toBe(0)
|
||||
expect(JSON.parse(fs.readFileSync(path.join(out, 'child.json'), 'utf8'))).toEqual({
|
||||
identity, source, cwd: source
|
||||
})
|
||||
expect(fs.readdirSync(source)).toEqual([])
|
||||
const args = ['--source', fileURLToPath(new URL('..', import.meta.url)), '--out', out]
|
||||
expect(generateIcons(args, { env })).toBe(0)
|
||||
const png = fs.readFileSync(path.join(out, 'apps/desktop/assets/icon.png'))
|
||||
expect(png.subarray(0, 8)).toEqual(Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]))
|
||||
expect([png.readUInt32BE(16), png.readUInt32BE(20)]).toEqual([1024, 1024])
|
||||
expect(generateIcons([...args, '--check'], { env })).toBe(0)
|
||||
expect(env.PYTHONPATH).toBe(path.join(out, 'foreign-site'))
|
||||
} finally {
|
||||
fs.rmSync(root, { recursive: true, force: true })
|
||||
fs.rmSync(out, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
}, 180_000)
|
||||
|
||||
test('failed icon processes cannot report a successful build', () => {
|
||||
expect(generateIcons([], { run: () => ({ status: 7 }), env: {} })).toBe(7)
|
||||
|
||||
@@ -142,13 +142,17 @@ def test_fixed_root_keeps_privilege_shim_and_resolves_venv_command_symlink(tmp_p
|
||||
out, data = inputs_fixture(tmp_path)
|
||||
# Docker source already occupies /opt/hermes; its bin/hermes belongs to s6.
|
||||
shutil.copytree(data["code"], out, dirs_exist_ok=True)
|
||||
data.update(placement="fixed", repo=".", code=str(out), project=str(out / "pyproject.toml"), bin_dir="libexec")
|
||||
data.update(placement="fixed", repo=".", code=str(out), project=str(out / "pyproject.toml"), bin_dir="libexec", python=sys.executable)
|
||||
bindir = out / "bin"
|
||||
bindir.mkdir()
|
||||
(bindir / "probe").write_text("privilege shim", encoding="utf-8")
|
||||
stale = out / "hermes_cli/web_dist/stale"
|
||||
stale.parent.mkdir(parents=True)
|
||||
stale.write_bytes(b"old surface")
|
||||
result = build_cli(data, out, tmp_path)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert (bindir / "probe").read_text() == "privilege shim"
|
||||
assert not stale.exists()
|
||||
link = out / "venv/bin/probe"
|
||||
link.parent.mkdir()
|
||||
link.symlink_to("../../libexec/probe")
|
||||
@@ -185,7 +189,7 @@ def test_source_metadata_keeps_declared_requirements_extras_and_entrypoints(tmp_
|
||||
|
||||
|
||||
@pytest.mark.platforms("posix")
|
||||
@pytest.mark.parametrize("fault", ["missing-pm", "missing-resource", "missing-tui", "unknown-field", "bad-repo", "bad-target"])
|
||||
@pytest.mark.parametrize("fault", ["missing-pm", "missing-resource", "missing-tui", "missing-web", "unknown-field", "bad-repo", "bad-target"])
|
||||
def test_failed_inputs_cannot_leave_a_completion_claim(tmp_path, fault):
|
||||
out, data = inputs_fixture(tmp_path)
|
||||
(out / "manifest.json").write_text('{"runtime": {"commands": {}}}', encoding="utf-8")
|
||||
@@ -195,6 +199,8 @@ def test_failed_inputs_cannot_leave_a_completion_claim(tmp_path, fault):
|
||||
data["resources"]["skills"] = str(tmp_path / "missing skills")
|
||||
elif fault == "missing-tui":
|
||||
(Path(data["frontends"]["tui"]) / "dist/entry.js").unlink()
|
||||
elif fault == "missing-web":
|
||||
(Path(data["frontends"]["web"]) / "index.html").unlink()
|
||||
elif fault == "unknown-field":
|
||||
data["typo"] = True
|
||||
elif fault == "bad-repo":
|
||||
@@ -216,23 +222,14 @@ def test_incremental_copy_drops_removed_source_without_deleting_provider_files(t
|
||||
sentinel = out / "provider-file"
|
||||
sentinel.write_text("retain", encoding="utf-8")
|
||||
(source / "obsolete.py").unlink()
|
||||
(out / "app/hermes_cli/web_dist/stale").write_text("old", encoding="utf-8")
|
||||
assert build_cli(data, out, tmp_path).returncode == 0
|
||||
assert not (out / "app/obsolete.py").exists()
|
||||
assert not (out / "app/hermes_cli/web_dist/stale").exists()
|
||||
assert (out / "app/hermes_cli/tui_dist/entry.js").read_bytes() == (Path(data["frontends"]["tui"]) / "dist/entry.js").read_bytes()
|
||||
assert sentinel.read_text() == "retain"
|
||||
|
||||
|
||||
@pytest.mark.platforms("posix")
|
||||
def test_fixed_launcher_accepts_explicit_external_python_without_path_guessing(tmp_path):
|
||||
out, data = inputs_fixture(tmp_path)
|
||||
data.update(placement="fixed", python=sys.executable)
|
||||
result = build_cli(data, out, tmp_path)
|
||||
assert result.returncode == 0, result.stderr
|
||||
run = subprocess.run([str(out / "bin/probe")], cwd=tmp_path,
|
||||
env={"PATH": os.environ["PATH"], "HOME": str(tmp_path / "home")},
|
||||
capture_output=True, text=True)
|
||||
assert run.returncode == 7, run.stderr
|
||||
|
||||
|
||||
@pytest.mark.platforms("posix")
|
||||
def test_payload_smoke_uses_relocated_manifest_commands(tmp_path):
|
||||
out, data = inputs_fixture(tmp_path)
|
||||
|
||||
@@ -55,26 +55,3 @@ def test_canary_publication_verifies_native_identity_and_uploads_bundle_before_p
|
||||
assert not any(method == 'PUT' and path.endswith('canary.appinstaller')
|
||||
for method, path, _ in r2_server.requests)
|
||||
assert r2_server.store[pointer] == previous
|
||||
|
||||
|
||||
@pytest.mark.parametrize('channel', ['stable', 'canary', 'light/stable', 'light/canary'])
|
||||
def test_descriptor_cli_preserves_package_facts_and_subscription_uri(tmp_path, channel):
|
||||
identity, publisher = 'Product&<"test">', 'CN=Publisher & "Team"'
|
||||
self_uri = f'https://releases.example/releases/win32/{channel}/update.appinstaller?a=1&b=2'
|
||||
out = tmp_path / 'descriptor.appinstaller'
|
||||
for version in ['1.2.3.0', '1.2.4.31']:
|
||||
artifact_uri = f'https://releases.example/releases/tag/v{version}/app.msixbundle?a=1&b=2'
|
||||
subprocess.run([
|
||||
sys.executable, '-m', 'scripts.bundles.release_artifacts', 'appinstaller',
|
||||
'--root', str(tmp_path), '--out', str(out), '--identity', identity,
|
||||
'--publisher', publisher, '--version', version,
|
||||
'--self-uri', self_uri, '--artifact-uri', artifact_uri,
|
||||
], cwd=Path(__file__).resolve().parents[2], check=True)
|
||||
descriptor = ET.parse(out).getroot()
|
||||
assert descriptor.tag == '{http://schemas.microsoft.com/appx/appinstaller/2017/2}AppInstaller'
|
||||
assert descriptor.attrib == {'Uri': self_uri, 'Version': version}
|
||||
assert descriptor.find('{*}MainBundle').attrib == {
|
||||
'Name': identity, 'Publisher': publisher, 'Version': version, 'Uri': artifact_uri,
|
||||
}
|
||||
assert descriptor.find('{*}UpdateSettings/{*}OnLaunch').attrib == {'HoursBetweenUpdateChecks': '12'}
|
||||
assert descriptor.find('{*}MainPackage') is None
|
||||
|
||||
@@ -23,10 +23,13 @@ def test_bundle_stages_git_tree_and_runs_native_children_before_manifest(tmp_pat
|
||||
import inspect
|
||||
|
||||
from hermes_cli.runtime_paths import site_packages
|
||||
from pm.lock import Facts, Lockfile
|
||||
from pm.store import Store
|
||||
from tests.pm._fixtures import _wheel
|
||||
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path / "home")
|
||||
output = tmp_path / "payload"
|
||||
target_python = output / "staged-python" / ("python.exe" if os.name == "nt" else "bin/python")
|
||||
target_python = output / "tools/cached-python" / ("python.exe" if os.name == "nt" else "bin/python3")
|
||||
target_python.parent.mkdir(parents=True)
|
||||
# PM seals a payload-owned base interpreter, not an external venv launcher.
|
||||
# The POSIX host supplies its stdlib; Windows needs it beside the executable.
|
||||
@@ -36,16 +39,48 @@ def test_bundle_stages_git_tree_and_runs_native_children_before_manifest(tmp_pat
|
||||
shutil.copy2(Path(getattr(sys, "_base_executable")).resolve(), target_python)
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
pm_project = Path(__file__).resolve().parents[2] / "pm"
|
||||
(repo / "pm").mkdir()
|
||||
for name in ("pyproject.toml", "uv.lock", "lock.json"):
|
||||
shutil.copy2(pm_project / name, repo / "pm" / name)
|
||||
(repo / "pyproject.toml").write_text('[project]\nname="fixture"\nversion="1.0.0"\nrequires-python=">=3.11"\n[project.scripts]\nprobe="entry:main"\n[project.optional-dependencies]\npayloadtest=[]\n[tool.uv]\npackage=false\n', encoding="utf-8")
|
||||
source = Path(__file__).resolve().parents[2]
|
||||
shutil.copytree(source / "pm", repo / "pm", ignore=shutil.ignore_patterns("__pycache__"))
|
||||
(repo / "hermes_cli").mkdir()
|
||||
for name in ("__init__.py", "runtime_paths.py", "runtime_state.py"):
|
||||
shutil.copy2(source / "hermes_cli" / name, repo / "hermes_cli" / name)
|
||||
shutil.copy2(source / "hermes_constants.py", repo / "hermes_constants.py")
|
||||
wheels = repo / "wheels"
|
||||
wheels.mkdir()
|
||||
witness = tmp_path / "inventory-python.json"
|
||||
wheel = _wheel(wheels, "bundle_probe")
|
||||
import zipfile
|
||||
with zipfile.ZipFile(wheel, "a") as archive:
|
||||
archive.writestr("bundle_probe/witness/__init__.py", "import json,pathlib,sys\n"
|
||||
f"pathlib.Path({str(witness)!r}).write_text(json.dumps(sys.executable))\n")
|
||||
archive.writestr("bundle_probe/witness/present.py", "")
|
||||
(repo / "pyproject.toml").write_text(
|
||||
'[project]\nname="fixture"\nversion="1.0.0"\nrequires-python=">=3.11"\n'
|
||||
'[project.scripts]\nprobe="entry:main"\n[project.optional-dependencies]\npayloadtest=["bundle-probe==1.0"]\n'
|
||||
'[tool.uv]\npackage=false\nno-index=true\nfind-links=["wheels"]\n', encoding="utf-8")
|
||||
selected = {"agent-browser", "chromium", "uv", "python"}
|
||||
stale = {"chromium-headless-shell", "retired-tool"}
|
||||
user_store = tmp_path / "user-tools"
|
||||
for store in (output / "tools", user_store):
|
||||
store.mkdir(parents=True, exist_ok=True)
|
||||
facts = Facts(store / "facts.json")
|
||||
for name in selected | stale:
|
||||
entry = store / f"cached-{name}"
|
||||
entry.mkdir(exist_ok=True)
|
||||
(entry / "payload").write_text(name, encoding="utf-8")
|
||||
facts.record(name, f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", entry.name, {}, store)
|
||||
(store / "orphaned-version").mkdir()
|
||||
(store / ".partials").mkdir()
|
||||
user_before = {path.relative_to(user_store): path.read_bytes() if path.is_file() else None for path in user_store.rglob("*")}
|
||||
selection = Lockfile(tmp_path / "selection.json")
|
||||
for name in ("agent-browser", "uv"):
|
||||
selection.set_pin(name, "fixture", {})
|
||||
selection.save()
|
||||
from scripts.build.inputs import RESOURCE_ENV
|
||||
for name in RESOURCE_ENV:
|
||||
(repo / name).mkdir()
|
||||
(repo / name / "asset").write_text("required", encoding="utf-8")
|
||||
(repo / "entry.py").write_text("def main(): return 0\n", encoding="utf-8")
|
||||
(repo / "entry.py").write_text("import bundle_probe\ndef main(): print(bundle_probe.__version__); return 7\n", encoding="utf-8")
|
||||
uv = shutil.which("uv")
|
||||
assert uv, "native bundle test requires uv"
|
||||
env = {**os.environ, "UV_OFFLINE": "1", "UV_PYTHON_DOWNLOADS": "never", "UV_CACHE_DIR": str(tmp_path / "cache")}
|
||||
@@ -54,21 +89,18 @@ def test_bundle_stages_git_tree_and_runs_native_children_before_manifest(tmp_pat
|
||||
subprocess.run(["git", "add", "."], cwd=repo, check=True)
|
||||
subprocess.run(["git", "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "-m", "fixture"], cwd=repo, check=True, capture_output=True)
|
||||
monkeypatch.setattr("pm.paths.repo_root", lambda: repo)
|
||||
monkeypatch.setattr(native, "_bundle_package_names", lambda: [])
|
||||
monkeypatch.setattr(native, "_install_names", lambda names: 0)
|
||||
monkeypatch.setattr(native, "_store", lambda: SimpleNamespace(root=output / "tools", entry=lambda _: target_python.parent))
|
||||
monkeypatch.setattr(native, "_facts", lambda: SimpleNamespace(get=lambda _: {"entry": "python", "version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"}, entries_in_use=lambda: []))
|
||||
monkeypatch.setattr(native, "get_package", lambda _: SimpleNamespace(binary=lambda *args: target_python))
|
||||
(repo / "untracked").write_text("must not ship", encoding="utf-8")
|
||||
monkeypatch.setattr(native, "_lockfile", lambda: selection)
|
||||
monkeypatch.setattr(native, "_install_names", lambda names: 0) # prepared artifacts, real facts/prune
|
||||
monkeypatch.setattr(native, "_store", lambda: Store(output / "tools"))
|
||||
monkeypatch.setattr(native, "_facts", lambda: Facts(output / "tools/facts.json"))
|
||||
monkeypatch.setattr("pm.client.is_runtime", lambda: True)
|
||||
monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable)))
|
||||
monkeypatch.setattr(native, "_arch_guard", lambda store: [])
|
||||
monkeypatch.setattr("scripts.bundles.payload.relativize_links", lambda root: 0)
|
||||
monkeypatch.setattr("pm.extras.ANCHORS", {"payloadtest": "bundle_probe.present"})
|
||||
monkeypatch.setattr("pm.extras.ANCHORS", {"payloadtest": "bundle_probe.witness.present"})
|
||||
import pm
|
||||
real_build = pm.build_environment
|
||||
install_timeout = inspect.signature(real_build).parameters["timeout"].default
|
||||
calls = []
|
||||
witness = tmp_path / "inventory-python.json"
|
||||
fail_inventory = False
|
||||
|
||||
def build(**kwargs):
|
||||
@@ -84,39 +116,57 @@ def test_bundle_stages_git_tree_and_runs_native_children_before_manifest(tmp_pat
|
||||
site = site_packages(output / "venv")
|
||||
if fail_inventory:
|
||||
shutil.rmtree(site)
|
||||
else:
|
||||
package = site / "bundle_probe"
|
||||
package.mkdir()
|
||||
(package / "__init__.py").write_text(
|
||||
"import json, pathlib, sys\n"
|
||||
f"pathlib.Path({str(witness)!r}).write_text(json.dumps(sys.executable), encoding='utf-8')\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(package / "present.py").write_text("", encoding="utf-8")
|
||||
return result
|
||||
|
||||
monkeypatch.setattr(pm, "build_environment", build)
|
||||
monkeypatch.setenv("HERMES_RUNTIME_DIR", str(tmp_path / "original"))
|
||||
monkeypatch.setenv("HERMES_RUNTIME_DIR", str(user_store))
|
||||
monkeypatch.setenv("UV_CACHE_DIR", str(tmp_path / "cache"))
|
||||
assert native._stage_native(SimpleNamespace(out=str(output), ref="HEAD")) == 0
|
||||
assert calls[0]["all_extras"] is True
|
||||
assert calls[0]["cache"] == tmp_path / "cache"
|
||||
assert (output / "hermes-agent/pyproject.toml").is_file()
|
||||
assert not (output / "hermes-agent/untracked").exists()
|
||||
assert not (output / "hermes-agent/.git").exists()
|
||||
facts = Facts(output / "tools/facts.json")
|
||||
for name in stale:
|
||||
assert facts.get(name) is None
|
||||
assert not (output / "tools" / f"cached-{name}").exists()
|
||||
for name in selected:
|
||||
assert facts.get(name)["entry"] == f"cached-{name}"
|
||||
assert (output / "tools" / f"cached-{name}" / "payload").read_text(encoding="utf-8-sig") == name
|
||||
assert not (output / "tools/orphaned-version").exists()
|
||||
assert (output / "tools/.partials").is_dir()
|
||||
assert user_before == {path.relative_to(user_store): path.read_bytes() if path.is_file() else None for path in user_store.rglob("*")}
|
||||
|
||||
manifest = json.loads((output / "manifest.json").read_text())
|
||||
assert manifest["repo"] == "hermes-agent"
|
||||
command = "bin/probe.exe" if os.name == "nt" else "bin/probe"
|
||||
assert manifest["runtime"]["commands"] == {"probe": command}
|
||||
feature_file = output / "enabled-features.json"
|
||||
assert json.loads(feature_file.read_text(encoding="utf-8"))["extras"] == ["payloadtest"]
|
||||
assert Path(json.loads(witness.read_text(encoding="utf-8"))) == target_python
|
||||
assert os.environ["HERMES_RUNTIME_DIR"] == str(tmp_path / "original")
|
||||
assert os.environ["HERMES_RUNTIME_DIR"] == str(user_store)
|
||||
|
||||
moved = tmp_path / "installed elsewhere"
|
||||
output.rename(moved)
|
||||
try:
|
||||
run = subprocess.run([str(moved / command)], cwd=tmp_path, capture_output=True, text=True, timeout=30)
|
||||
assert run.returncode == 7, run.stderr
|
||||
assert run.stdout.strip() == "1.0"
|
||||
from pm import runtime as runtime_api, paths
|
||||
with monkeypatch.context() as patch:
|
||||
patch.setattr(paths, "repo_root", lambda: moved / "hermes-agent")
|
||||
run = subprocess.run(runtime_api.runtime_command(moved / "hermes-agent/pm/launch.py", ["status"]),
|
||||
cwd=tmp_path, env=runtime_api.runtime_environment(), capture_output=True, text=True, timeout=30)
|
||||
assert run.returncode == 0, run.stderr
|
||||
assert "no pm sync receipt" in run.stdout
|
||||
finally:
|
||||
moved.rename(output)
|
||||
assert json.loads(feature_file.read_text(encoding="utf-8"))["extras"] == ["payloadtest"]
|
||||
assert Path(json.loads(witness.read_text(encoding="utf-8-sig"))) == target_python
|
||||
before = feature_file.read_bytes()
|
||||
fail_inventory = True
|
||||
assert native._stage_native(SimpleNamespace(out=str(output), ref="HEAD")) == 1
|
||||
assert not (output / "manifest.json").exists()
|
||||
assert feature_file.read_bytes() == before
|
||||
assert os.environ["HERMES_RUNTIME_DIR"] == str(tmp_path / "original")
|
||||
assert os.environ["HERMES_RUNTIME_DIR"] == str(user_store)
|
||||
|
||||
from pm.package import InstallError
|
||||
def fail_build(**kwargs):
|
||||
@@ -124,7 +174,7 @@ def test_bundle_stages_git_tree_and_runs_native_children_before_manifest(tmp_pat
|
||||
monkeypatch.setattr(pm, "build_environment", fail_build)
|
||||
assert native._stage_native(SimpleNamespace(out=str(output), ref="HEAD")) == 1
|
||||
assert not (output / "manifest.json").exists()
|
||||
assert os.environ["HERMES_RUNTIME_DIR"] == str(tmp_path / "original")
|
||||
assert os.environ["HERMES_RUNTIME_DIR"] == str(user_store)
|
||||
|
||||
import pytest
|
||||
(repo / "pm/lock.json").write_text("{}", encoding="utf-8")
|
||||
@@ -259,65 +309,6 @@ def test_staged_cache_rebuilds_venv_offline_without_build_sources_or_zips(tmp_pa
|
||||
assert probe.stdout.splitlines() == ["installed from cached wheel", "1.0"]
|
||||
|
||||
|
||||
def test_native_dispatch_reuses_pm_cache_offline(tmp_path, monkeypatch):
|
||||
import pm
|
||||
from pm.packages import uv_cache_dir
|
||||
from tests.pm._fixtures import _wheel
|
||||
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path / "home")
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "setup-pm"))
|
||||
monkeypatch.setenv("HERMES_RUNTIME_DIR", str(tmp_path / "setup-pm/tools"))
|
||||
monkeypatch.delenv("UV_CACHE_DIR", raising=False)
|
||||
monkeypatch.setattr("scripts.build.windows_deps.prepare_windows_environment", lambda **kwargs: dict(kwargs["env"]))
|
||||
uv = shutil.which("uv")
|
||||
assert uv, "native bundle test requires uv"
|
||||
monkeypatch.setattr("pm.client.is_runtime", lambda: True)
|
||||
monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable)))
|
||||
cache = uv_cache_dir()
|
||||
wheels = tmp_path / "wheels"
|
||||
wheels.mkdir()
|
||||
_wheel(wheels, "cache_probe", "1.0")
|
||||
wheel, = wheels.glob("*.whl")
|
||||
server = ThreadingHTTPServer(("127.0.0.1", 0), partial(SimpleHTTPRequestHandler, directory=str(wheels)))
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
requirement = f"cache-probe @ http://127.0.0.1:{server.server_port}/{wheel.name}"
|
||||
try:
|
||||
pm.build_requirements_environment(
|
||||
[requirement], out=tmp_path / "first", explicit=True,
|
||||
)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join(timeout=5)
|
||||
shutil.rmtree(wheels)
|
||||
pm.prune_cache(cache)
|
||||
original = dict(os.environ)
|
||||
run = subprocess.run
|
||||
homes = []
|
||||
|
||||
def child(command, *, cwd, env):
|
||||
assert Path(env["UV_CACHE_DIR"]) == cache
|
||||
homes.append(Path(env["HOME"]))
|
||||
# The server and source wheel are gone. Only the cache restored for
|
||||
# PM can satisfy this install inside the payload's isolated HOME.
|
||||
return run([sys.executable, "-c",
|
||||
"import os, subprocess, sys; from pathlib import Path; import pm, pm._uv; "
|
||||
"pm.client.is_runtime = lambda: True; "
|
||||
f"pm._uv._toolchain = lambda **kw: (Path({uv!r}), Path(sys.executable)); "
|
||||
"python = pm.build_requirements_environment([sys.argv[1]], "
|
||||
"out=Path(os.environ['HOME'])/'venv', "
|
||||
"cache=Path(os.environ['UV_CACHE_DIR']), offline=True, explicit=True); "
|
||||
"subprocess.run([str(python), '-I', '-c', 'import cache_probe'], check=True)",
|
||||
requirement], cwd=cwd, env=env, check=True)
|
||||
|
||||
monkeypatch.setattr(native.subprocess, "run", child)
|
||||
for name in ("first-payload", "second-payload"):
|
||||
assert native.stage_native(SimpleNamespace(out=tmp_path / name, ref="HEAD")) == 0
|
||||
assert all(not home.exists() for home in homes)
|
||||
assert dict(os.environ) == original
|
||||
|
||||
|
||||
def test_native_dispatch_isolates_process_state_on_real_child_failure(tmp_path, monkeypatch):
|
||||
# Compiler provisioning has its own native test; this probe must stop
|
||||
# at the invalid revision without installing tools on a developer host.
|
||||
@@ -334,64 +325,53 @@ def test_native_dispatch_isolates_process_state_on_real_child_failure(tmp_path,
|
||||
assert not list(out.glob(".build-*"))
|
||||
|
||||
|
||||
def test_native_dispatch_keeps_cache_across_failed_children(tmp_path, monkeypatch):
|
||||
@pytest.mark.parametrize("cache_source,explicit_compilers,status", [
|
||||
("explicit", True, 17), ("ambient", False, 0), ("default", False, 17),
|
||||
])
|
||||
def test_native_dispatch_child_environment(tmp_path, monkeypatch, cache_source, explicit_compilers, status):
|
||||
from pm.packages import uv_cache_dir
|
||||
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path / "host")
|
||||
monkeypatch.setattr("scripts.build.windows_deps.prepare_windows_environment", lambda **kwargs: dict(kwargs["env"]))
|
||||
out = tmp_path / "payload"
|
||||
cache = tmp_path / "persistent-cache"
|
||||
monkeypatch.setenv("UV_CACHE_DIR", str(tmp_path / "ambient-cache"))
|
||||
original = dict(os.environ)
|
||||
run = subprocess.run
|
||||
attempts = []
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "user"))
|
||||
monkeypatch.delenv("UV_CACHE_DIR", raising=False)
|
||||
ambient = tmp_path / "ambient"
|
||||
if cache_source != "default":
|
||||
monkeypatch.setenv("UV_CACHE_DIR", str(ambient))
|
||||
cache = {"explicit": tmp_path / "explicit", "ambient": ambient, "default": uv_cache_dir()}[cache_source]
|
||||
compilers = {}
|
||||
for key, name in (("CARGO_HOME", ".cargo"), ("RUSTUP_HOME", ".rustup")):
|
||||
home = tmp_path / ("custom" if explicit_compilers else "host") / name
|
||||
home.mkdir(parents=True)
|
||||
(home / "fixture-state").write_text(key, encoding="utf-8")
|
||||
compilers[key] = str(home)
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
if explicit_compilers:
|
||||
monkeypatch.setenv(key, str(home))
|
||||
before, run, homes = dict(os.environ), subprocess.run, []
|
||||
observed = tmp_path / "child.json"
|
||||
|
||||
def child(command, *, cwd, env):
|
||||
assert command[command.index("-m") + 1] == "scripts.bundles.native"
|
||||
assert Path(env["UV_CACHE_DIR"]) == cache
|
||||
attempts.append(Path(env["HOME"]))
|
||||
# Run a real child using the actual dispatch environment. Its cache
|
||||
# write must survive the failing process and temporary-HOME cleanup.
|
||||
return run([sys.executable, "-c",
|
||||
"import os,sys; from pathlib import Path; "
|
||||
"p=Path(os.environ['UV_CACHE_DIR']); p.mkdir(exist_ok=True); "
|
||||
"f=p/'reused'; f.write_text(f.read_text()+'x' if f.exists() else 'x'); sys.exit(17)"],
|
||||
cwd=cwd, env=env)
|
||||
"import os,json,sys; from pathlib import Path; "
|
||||
"Path(sys.argv[1]).write_text(json.dumps(dict(os.environ))); "
|
||||
"assert all((Path(os.environ[k])/'fixture-state').read_text() == k for k in ('CARGO_HOME','RUSTUP_HOME')); "
|
||||
"p=Path(os.environ['UV_CACHE_DIR']); p.mkdir(parents=True,exist_ok=True); "
|
||||
"f=p/'reused'; f.write_text(f.read_text()+'x' if f.exists() else 'x'); sys.exit(int(sys.argv[2]))",
|
||||
str(observed), str(status)], cwd=cwd, env=env)
|
||||
|
||||
monkeypatch.setattr(native.subprocess, "run", child)
|
||||
out = tmp_path / "payload"
|
||||
for _ in range(2):
|
||||
assert native.stage_native(SimpleNamespace(out=out, ref="HEAD", cache=cache)) == 17
|
||||
assert (cache / "reused").read_text() == "xx"
|
||||
assert all(not home.exists() for home in attempts)
|
||||
assert not (tmp_path / "ambient-cache").exists()
|
||||
assert dict(os.environ) == original
|
||||
|
||||
|
||||
@pytest.mark.parametrize("explicit", [False, True])
|
||||
def test_native_dispatch_preserves_compiler_homes_inside_isolated_home(tmp_path, monkeypatch, explicit):
|
||||
host_home = tmp_path / "host"
|
||||
monkeypatch.setattr(Path, "home", lambda: host_home)
|
||||
monkeypatch.setattr("scripts.build.windows_deps.prepare_windows_environment", lambda **kwargs: dict(kwargs["env"]))
|
||||
homes = {}
|
||||
for key, name in (("CARGO_HOME", ".cargo"), ("RUSTUP_HOME", ".rustup")):
|
||||
directory = (tmp_path / "custom" if explicit else host_home) / name
|
||||
directory.mkdir(parents=True)
|
||||
(directory / "fixture-state").write_text(key, encoding="utf-8")
|
||||
homes[key] = str(directory)
|
||||
if explicit:
|
||||
monkeypatch.setenv(key, str(directory))
|
||||
else:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
before = dict(os.environ)
|
||||
run = subprocess.run
|
||||
|
||||
def child(command, *, cwd, env):
|
||||
assert env["HOME"] != str(host_home)
|
||||
assert env["USERPROFILE"] == env["HOME"]
|
||||
assert {key: env.get(key) for key in homes} == homes
|
||||
return run([sys.executable, "-c",
|
||||
"import os; from pathlib import Path; "
|
||||
"assert all((Path(os.environ[k]) / 'fixture-state').read_text() == k "
|
||||
"for k in ('CARGO_HOME', 'RUSTUP_HOME'))"],
|
||||
cwd=cwd, env=env, check=True)
|
||||
|
||||
monkeypatch.setattr(native.subprocess, "run", child)
|
||||
assert native.stage_native(SimpleNamespace(out=tmp_path / "out", ref="HEAD")) == 0
|
||||
assert native.stage_native(SimpleNamespace(out=out, ref="HEAD", cache=cache if cache_source == "explicit" else None)) == status
|
||||
env = json.loads(observed.read_text(encoding="utf-8-sig"))
|
||||
homes.append(Path(env["HOME"]))
|
||||
assert env["HOME"] == env["USERPROFILE"] != str(tmp_path / "host")
|
||||
assert {key: env[key] for key in compilers} == compilers
|
||||
assert Path(env["UV_CACHE_DIR"]) == cache
|
||||
assert Path(env["HERMES_RUNTIME_DIR"]) == out / "tools"
|
||||
assert Path(env["HERMES_HOME"]) == homes[-1] / ".hermes"
|
||||
assert (cache / "reused").read_text(encoding="utf-8-sig") == "xx"
|
||||
assert all(not home.exists() for home in homes)
|
||||
assert dict(os.environ) == before
|
||||
|
||||
@@ -16,44 +16,11 @@ from scripts.build.launchers import posix_launcher
|
||||
from scripts.bundles.desktop import release_version
|
||||
|
||||
|
||||
def test_snapshot_preserves_declared_entries_in_both_layouts(tmp_path):
|
||||
source = tmp_path / "source"
|
||||
source.mkdir()
|
||||
subprocess.run(["git", "init", str(source)], check=True, capture_output=True)
|
||||
project = '[project]\nname="fixture"\nversion="1.2.3"\n[project.scripts]\ncustom="entry:run"\n'
|
||||
(source / "pyproject.toml").write_text(project, encoding="utf-8")
|
||||
subprocess.run(["git", "add", "."], cwd=source, check=True)
|
||||
subprocess.run(["git", "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "-m", "fixture"], cwd=source, check=True, capture_output=True)
|
||||
(source / "untracked").write_text("must not ship", encoding="utf-8")
|
||||
for repo_name, target in [("app", "linux-arm64-bionic"), ("hermes-agent", "win32-arm64")]:
|
||||
root = tmp_path / repo_name
|
||||
root.mkdir()
|
||||
snapshot(source, "HEAD", root / repo_name)
|
||||
assert project_entries(root / repo_name / "pyproject.toml") == {"custom": "entry:run"}
|
||||
assert not (root / repo_name / "untracked").exists()
|
||||
assert not (root / repo_name / ".git").exists()
|
||||
assert release_version(source, "v1.2.3") == "1.2.3"
|
||||
def test_release_version_must_match_project(tmp_path):
|
||||
(tmp_path / "pyproject.toml").write_text('[project]\nversion="1.2.3"\n', encoding="utf-8")
|
||||
assert release_version(tmp_path, "v1.2.3") == "1.2.3"
|
||||
with pytest.raises(ValueError):
|
||||
release_version(source, "v1.2.4")
|
||||
|
||||
|
||||
def test_surfaces_require_complete_outputs_and_replace_stale_files(tmp_path):
|
||||
source, repo = tmp_path / "build", tmp_path / "payload"
|
||||
tui = source / "tui/dist"
|
||||
web = source / "hermes_cli/web_dist"
|
||||
tui.mkdir(parents=True)
|
||||
web.mkdir(parents=True)
|
||||
(tui / "entry.js").write_text("built tui", encoding="utf-8")
|
||||
(tui.parent / "package.json").write_text('{"type":"module"}', encoding="utf-8")
|
||||
(web / "index.html").write_text("built web", encoding="utf-8")
|
||||
plant_surfaces(repo, {"tui": tui.parent, "web": web})
|
||||
(repo / "hermes_cli/web_dist/stale").write_text("old", encoding="utf-8")
|
||||
plant_surfaces(repo, {"tui": tui.parent, "web": web})
|
||||
assert not (repo / "hermes_cli/web_dist/stale").exists()
|
||||
assert (repo / "hermes_cli/tui_dist/entry.js").read_text(encoding="utf-8-sig") == "built tui"
|
||||
(web / "index.html").unlink()
|
||||
with pytest.raises(FileNotFoundError):
|
||||
plant_surfaces(repo, {"tui": tui.parent, "web": web})
|
||||
release_version(tmp_path, "v1.2.4")
|
||||
|
||||
|
||||
def test_wrapper_rejects_unresolved_template_fields(tmp_path, monkeypatch):
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
"""Reusing a bundle cache must not ship packages removed from its selection."""
|
||||
from pm.lock import Facts, Lockfile
|
||||
from scripts.bundles import native
|
||||
|
||||
|
||||
def test_cached_bundle_prunes_unselected_facts_and_entries(tmp_path, monkeypatch):
|
||||
output = tmp_path / "payload"
|
||||
staged_store = output / "tools"
|
||||
user_store = tmp_path / "user-tools"
|
||||
selected = {"agent-browser", "chromium", "uv", "python"}
|
||||
stale = {"chromium-headless-shell", "retired-tool"}
|
||||
for root in (staged_store, user_store):
|
||||
root.mkdir(parents=True)
|
||||
facts = Facts(root / "facts.json")
|
||||
for name in selected | stale:
|
||||
entry = f"cached-{name}"
|
||||
(root / entry).mkdir()
|
||||
(root / entry / "payload").write_text(name, encoding="utf-8")
|
||||
facts.record(name, "fixture", entry, {}, root)
|
||||
(root / "orphaned-version").mkdir()
|
||||
(root / ".partials").mkdir()
|
||||
user_facts_before = (user_store / "facts.json").read_bytes()
|
||||
user_entries_before = {p.name for p in user_store.iterdir()}
|
||||
|
||||
# Chromium is a dependency, not a selected root; Python is supplied by
|
||||
# bundle selection and uv must survive despite being internal.
|
||||
lock = Lockfile(tmp_path / "lock.json")
|
||||
for name in ("agent-browser", "uv"):
|
||||
lock.set_pin(name, "fixture", {})
|
||||
lock.save()
|
||||
monkeypatch.setattr(native, "_lockfile", lambda: lock)
|
||||
monkeypatch.setenv("HERMES_RUNTIME_DIR", str(user_store))
|
||||
|
||||
native.prune_staged_store(staged_store, native._bundle_package_names())
|
||||
|
||||
remaining = Facts(staged_store / "facts.json")
|
||||
for name in stale:
|
||||
assert remaining.get(name) is None
|
||||
assert not (staged_store / f"cached-{name}").exists()
|
||||
for name in selected:
|
||||
fact = remaining.get(name)
|
||||
assert fact is not None
|
||||
assert fact["entry"] == f"cached-{name}"
|
||||
assert (staged_store / f"cached-{name}" / "payload").read_text(encoding="utf-8") == name
|
||||
assert not (staged_store / "orphaned-version").exists()
|
||||
assert (staged_store / ".partials").is_dir()
|
||||
assert native._store().root == user_store
|
||||
assert (user_store / "facts.json").read_bytes() == user_facts_before
|
||||
assert {p.name for p in user_store.iterdir()} == user_entries_before
|
||||
@@ -27,14 +27,9 @@ _WRAPPER = _REPO / "scripts" / "build" / "launcher_wrapper.py"
|
||||
def _load(env=None):
|
||||
"""Import the wrapper with its placeholders substituted, like the build
|
||||
script does, and return its module namespace."""
|
||||
text = _WRAPPER.read_text(encoding="utf-8")
|
||||
for placeholder, value in {
|
||||
"__HERMES_ENTRY_MODULE__": "stubmod.entry",
|
||||
"__HERMES_ENTRY_FUNC__": "main",
|
||||
"__HERMES_REPO_REL__": "../repo",
|
||||
"__HERMES_SITE_REL__": "../venv/Lib/site-packages",
|
||||
}.items():
|
||||
text = text.replace(placeholder, value)
|
||||
from scripts.build.launchers import render_wrapper
|
||||
|
||||
text = render_wrapper("stubmod.entry:main", "../repo", "../venv/Lib/site-packages")
|
||||
namespace: dict = {"__name__": "launcher_wrapper_under_test"}
|
||||
exec(compile(text, str(_WRAPPER), "exec"), namespace) # noqa: S102 - test fixture
|
||||
return namespace
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
"""Exercise the frontend COPY closure before an image/network build.
|
||||
|
||||
PM environment construction is the only substituted boundary. The copied icon
|
||||
provider must load, ask for its real dependency group, and invoke the copied
|
||||
generator. Image-level runtime/permission coverage stays in tests/docker.
|
||||
"""
|
||||
from pathlib import Path
|
||||
import runpy
|
||||
import shlex
|
||||
import shutil
|
||||
from types import SimpleNamespace
|
||||
from types import ModuleType
|
||||
|
||||
from pathspec import PathSpec
|
||||
|
||||
import pm
|
||||
|
||||
|
||||
def test_frontend_copy_closure_reaches_the_icon_provider(tmp_path, monkeypatch):
|
||||
repo = Path(__file__).resolve().parents[2]
|
||||
# Apply the frontend's local COPY declarations, not a duplicated filename
|
||||
# allowlist. Runtime-base supplies PM itself and is separately image-tested.
|
||||
frontend = False
|
||||
ignored = PathSpec.from_lines("gitignore", (repo / ".dockerignore").read_text().splitlines())
|
||||
|
||||
def excluded(directory, names):
|
||||
return [name for name in names if ignored.match_file(
|
||||
(Path(directory) / name).relative_to(repo).as_posix() +
|
||||
("/" if (Path(directory) / name).is_dir() else ""))]
|
||||
for line in (repo / "Dockerfile").read_text().splitlines():
|
||||
if not line.startswith(("FROM ", "COPY ")):
|
||||
continue
|
||||
words = shlex.split(line.rstrip("\\"), comments=True)
|
||||
if not words:
|
||||
continue
|
||||
if words[0] == "FROM":
|
||||
frontend = words[-1] == "frontend_build"
|
||||
if not frontend or words[0] != "COPY" or any(word.startswith("--") for word in words[1:]):
|
||||
continue
|
||||
sources, destination = words[1:-1], tmp_path / words[-1]
|
||||
for pattern in sources:
|
||||
for source in repo.glob(pattern):
|
||||
target = destination / source.name if words[-1].endswith("/") else destination
|
||||
if source.is_dir():
|
||||
shutil.copytree(source, destination, dirs_exist_ok=True,
|
||||
ignore=excluded)
|
||||
else:
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(source, target)
|
||||
acquired = []
|
||||
|
||||
def environment(**kwargs):
|
||||
acquired.append(kwargs)
|
||||
return Path("/prepared/icon-python")
|
||||
|
||||
launched = []
|
||||
monkeypatch.setattr(pm, "build_environment", environment)
|
||||
provider = runpy.run_path(str(tmp_path / "scripts/build/icon_environment.py"))
|
||||
|
||||
def run(argv, *, cwd):
|
||||
assert Path(argv[2]).is_file(), "generator must also be in the stage"
|
||||
assert cwd == tmp_path
|
||||
launched.append(argv)
|
||||
return SimpleNamespace(returncode=0)
|
||||
|
||||
monkeypatch.setattr(provider["subprocess"], "run", run)
|
||||
assert provider["main"](["--source", str(tmp_path), "--out", str(tmp_path / "icons")]) == 0
|
||||
assert acquired[0]["source"] == tmp_path
|
||||
assert acquired[0]["groups"] == ["icon-build"]
|
||||
assert acquired[0]["only_groups"] is True
|
||||
assert launched[0][:2] == ["/prepared/icon-python", "-I"]
|
||||
|
||||
# Exercise the generator's own source loader. Rasterization is not needed
|
||||
# to prove that every composed SVG's artwork made it into the build context.
|
||||
monkeypatch.setitem(__import__("sys").modules, "resvg_py", ModuleType("resvg_py"))
|
||||
generator = runpy.run_path(str(tmp_path / "scripts/generate_icons.py"))
|
||||
monkeypatch.setitem(generator["IconArt"].__init__.__globals__, "girl_bbox", lambda *args: (0, 0, 512, 512))
|
||||
art = generator["IconArt"](tmp_path)
|
||||
assert art.master_mac and art.master_mac_dark
|
||||
@@ -1,281 +1,66 @@
|
||||
"""Tests for the encoding-direction footgun rules in
|
||||
``scripts/check-windows-footguns.py``.
|
||||
|
||||
Two rules enforce the repo encoding policy (reads ``utf-8-sig``, writes
|
||||
``utf-8``):
|
||||
|
||||
- ``read with encoding='utf-8' (BOM-intolerant — use 'utf-8-sig')`` —
|
||||
Windows tooling (PowerShell Set-Content/Out-File, some editors)
|
||||
BOM-prefixes files it touches; a plain-utf-8 read hands ``'\\ufeff{...'``
|
||||
to ``json.load``, which fails with "Expecting value". Live case: PR #3 —
|
||||
a BOM'd ``install-stamp.json`` made ``read_build_info`` demote the tree
|
||||
to ``unknown``.
|
||||
- ``write with encoding='utf-8-sig' (emits a BOM)`` — the inverse: writing
|
||||
with ``utf-8-sig`` emits the exact bytes the read rule exists to
|
||||
tolerate.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
"""Encoding-direction diagnostics through the real file scanner, not a cloned loop."""
|
||||
import importlib.util
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
LINTER_PATH = REPO_ROOT / "scripts" / "check-windows-footguns.py"
|
||||
|
||||
READ_RULE = "read with encoding='utf-8' (BOM-intolerant — use 'utf-8-sig')"
|
||||
WRITE_RULE = "write with encoding='utf-8-sig' (emits a BOM)"
|
||||
|
||||
|
||||
def _load_linter_module():
|
||||
"""Import the linter script as a module (it's not a package).
|
||||
|
||||
Register the module in sys.modules BEFORE exec_module so that
|
||||
``@dataclass`` can resolve ``cls.__module__`` (CPython 3.11+).
|
||||
"""
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"check_windows_footguns", LINTER_PATH
|
||||
)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
sys.modules["check_windows_footguns"] = mod
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
READ = "read with encoding='utf-8' (BOM-intolerant — use 'utf-8-sig')"
|
||||
WRITE = "write with encoding='utf-8-sig' (emits a BOM)"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def linter():
|
||||
return _load_linter_module()
|
||||
path = Path(__file__).resolve().parents[2] / "scripts/check-windows-footguns.py"
|
||||
spec = importlib.util.spec_from_file_location("encoding_footguns", path)
|
||||
assert spec is not None and spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[spec.name] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _find_footgun(linter, name: str):
|
||||
for fg in linter.FOOTGUNS:
|
||||
if fg.name == name:
|
||||
return fg
|
||||
pytest.fail(f"Footgun rule '{name}' not found in FOOTGUNS")
|
||||
|
||||
|
||||
def _scan_line(linter, line: str, footgun_name: str) -> bool:
|
||||
"""Return True if the given line triggers the named footgun rule.
|
||||
|
||||
Replicates the relevant checks from scan_file(): suppression marker,
|
||||
guard hints, comment stripping, then pattern + post_filter — so the
|
||||
test exercises the real detection path.
|
||||
"""
|
||||
fg = _find_footgun(linter, footgun_name)
|
||||
if linter.SUPPRESS_MARKER.search(line):
|
||||
return False
|
||||
if any(hint in line for hint in linter.GUARD_HINTS):
|
||||
return False
|
||||
code = linter._strip_code(line)
|
||||
if not code.strip():
|
||||
return False
|
||||
match = fg.pattern.search(code)
|
||||
if not match:
|
||||
return False
|
||||
if fg.post_filter is not None:
|
||||
try:
|
||||
if not fg.post_filter(match, line):
|
||||
return False
|
||||
except (IndexError, AttributeError):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# READ rule — plain utf-8 on read-shaped lines SHOULD be flagged
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestReadRuleDetection:
|
||||
def test_flags_read_text_plain_utf8(self, linter):
|
||||
line = ' data = path.read_text(encoding="utf-8")'
|
||||
assert _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_flags_open_read_mode_plain_utf8(self, linter):
|
||||
line = " with open(path, 'r', encoding='utf-8') as f:"
|
||||
assert _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_flags_open_omitted_mode_plain_utf8(self, linter):
|
||||
# open() defaults to mode 'r' — still a read.
|
||||
line = " with open(path, encoding='utf-8') as f:"
|
||||
assert _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_flags_fdopen_read_mode_plain_utf8(self, linter):
|
||||
line = " f = os.fdopen(fd, 'r', encoding='utf-8')"
|
||||
assert _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_flags_underscore_spelling(self, linter):
|
||||
line = ' data = path.read_text(encoding="utf_8")'
|
||||
assert _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_flags_mode_keyword_read(self, linter):
|
||||
line = " with open(path, mode='r', encoding='utf-8') as f:"
|
||||
assert _scan_line(linter, line, READ_RULE)
|
||||
|
||||
|
||||
class TestReadRuleNegatives:
|
||||
def test_does_not_flag_utf8_sig_read(self, linter):
|
||||
# The policy-compliant form. The pattern must stop at the closing
|
||||
# quote of 'utf-8' and never match 'utf-8-sig'.
|
||||
line = ' data = path.read_text(encoding="utf-8-sig")'
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_write_text_plain_utf8(self, linter):
|
||||
# Writes with plain utf-8 are the CORRECT form.
|
||||
line = ' path.write_text(data, encoding="utf-8")'
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_open_write_mode_plain_utf8(self, linter):
|
||||
line = " with open(path, 'w', encoding='utf-8') as f:"
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_append_mode(self, linter):
|
||||
line = " with open(path, 'a', encoding='utf-8') as f:"
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_plus_mode(self, linter):
|
||||
# 'r+' can write — treat as write-capable, not a read.
|
||||
line = " with open(path, 'r+', encoding='utf-8') as f:"
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_subprocess_encoding(self, linter):
|
||||
# Not a file read; the rule stays quiet on unclassifiable lines.
|
||||
line = " subprocess.run(cmd, text=True, encoding='utf-8')"
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_suppressed_line(self, linter):
|
||||
line = (
|
||||
" data = path.read_text(encoding='utf-8')"
|
||||
" # windows-footgun: ok — file is repo-owned, never BOM'd"
|
||||
)
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_comment_only_line(self, linter):
|
||||
line = " # use path.read_text(encoding='utf-8') here"
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WRITE rule — utf-8-sig on write-shaped lines SHOULD be flagged
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestWriteRuleDetection:
|
||||
def test_flags_write_text_utf8_sig(self, linter):
|
||||
line = ' path.write_text(data, encoding="utf-8-sig")'
|
||||
assert _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_flags_open_write_mode_utf8_sig(self, linter):
|
||||
line = " with open(path, 'w', encoding='utf-8-sig') as f:"
|
||||
assert _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_flags_append_mode_utf8_sig(self, linter):
|
||||
line = " with open(path, 'a', encoding='utf-8-sig') as f:"
|
||||
assert _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_flags_fdopen_write_mode_utf8_sig(self, linter):
|
||||
line = " f = os.fdopen(fd, 'w', encoding='utf-8-sig')"
|
||||
assert _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_flags_path_open_append_utf8_sig(self, linter):
|
||||
# Path.open puts the mode in the FIRST argument — the original
|
||||
# classifier only understood the builtin second-arg shape, so the
|
||||
# live sweep rewrote 18 append/write sites to utf-8-sig (BOM per
|
||||
# append in jsonl logs; caught by test_lifecycle_ledger).
|
||||
line = ' with path.open("a", encoding="utf-8-sig") as fh:'
|
||||
assert _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_flags_path_open_plus_mode_utf8_sig(self, linter):
|
||||
# r+/a+ can write at position 0 — write-shaped.
|
||||
line = ' lock = path.open("a+", encoding="utf-8-sig")'
|
||||
assert _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_flags_underscore_spelling(self, linter):
|
||||
line = ' path.write_text(data, encoding="utf_8_sig")'
|
||||
assert _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
|
||||
class TestWriteRuleNegatives:
|
||||
def test_does_not_flag_utf8_sig_read(self, linter):
|
||||
# Reads with utf-8-sig are the CORRECT form.
|
||||
line = ' data = path.read_text(encoding="utf-8-sig")'
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_does_not_flag_open_read_mode_utf8_sig(self, linter):
|
||||
line = " with open(path, 'r', encoding='utf-8-sig') as f:"
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_does_not_flag_open_omitted_mode_utf8_sig(self, linter):
|
||||
# Omitted mode defaults to 'r' — a read.
|
||||
line = " with open(path, encoding='utf-8-sig') as f:"
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_does_not_flag_plain_utf8_write(self, linter):
|
||||
line = ' path.write_text(data, encoding="utf-8")'
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_does_not_flag_unclassifiable_line(self, linter):
|
||||
# No file-opening call — conservative default is no flag.
|
||||
line = " codec = 'utf-8-sig'"
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_does_not_flag_nested_call_first_arg(self, linter):
|
||||
# Regression: the first live sweep false-positived this line from
|
||||
# _startup_fast.py — the old mode regex swallowed the nested
|
||||
# ``join(`` paren and captured "install-stamp.json" as the mode
|
||||
# (its 'a' made it look write-shaped). Omitted mode = read.
|
||||
line = (
|
||||
' with open(os.path.join(root, "install-stamp.json"), '
|
||||
'encoding="utf-8-sig") as handle:'
|
||||
)
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_does_not_flag_path_open_read_mode_utf8_sig(self, linter):
|
||||
line = ' with path.open("r", encoding="utf-8-sig") as fh:'
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
def test_read_rule_does_not_rewrite_path_open_append(self, linter):
|
||||
# The inverse guard on the READ rule: an append-mode Path.open with
|
||||
# plain utf-8 is CORRECT and must never be flagged for conversion.
|
||||
line = ' with path.open("a", encoding="utf-8") as fh:'
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_read_rule_skips_method_open_without_mode(self, linter):
|
||||
# `.open(` with no recognizable mode is unclassified — the sweep
|
||||
# must never rewrite on a guess about the receiver.
|
||||
line = ' with something.open(encoding="utf-8") as fh:'
|
||||
assert not _scan_line(linter, line, READ_RULE)
|
||||
|
||||
def test_does_not_flag_suppressed_line(self, linter):
|
||||
line = (
|
||||
" path.write_text(data, encoding='utf-8-sig')"
|
||||
" # windows-footgun: ok — downstream consumer requires a BOM"
|
||||
)
|
||||
assert not _scan_line(linter, line, WRITE_RULE)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Suppression marker cannot be a bare comment mention (the real invariant —
|
||||
# a comment that merely NAMES the marker text elsewhere must not exempt).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestShapeHelpers:
|
||||
def test_read_shaped_read_text(self, linter):
|
||||
assert linter._is_read_shaped('x = p.read_text(encoding="utf-8")')
|
||||
|
||||
def test_read_shaped_rejects_write_text(self, linter):
|
||||
assert not linter._is_read_shaped('p.write_text(d, encoding="utf-8")')
|
||||
|
||||
def test_write_shaped_x_mode(self, linter):
|
||||
assert linter._is_write_shaped("open(p, 'x', encoding='utf-8-sig')")
|
||||
|
||||
def test_write_shaped_rejects_no_call(self, linter):
|
||||
assert not linter._is_write_shaped("enc = 'utf-8-sig'")
|
||||
|
||||
def test_read_shaped_rejects_no_call(self, linter):
|
||||
# No open/fdopen/read_text on the line at all.
|
||||
assert not linter._is_read_shaped("enc = 'utf-8'")
|
||||
@pytest.mark.parametrize("source,expected", [
|
||||
('data = path.read_text(encoding="utf-8")', [(1, READ)]),
|
||||
("with open(path, 'r', encoding='utf-8') as f:", [(1, READ)]),
|
||||
("with open(path, encoding='utf-8') as f:", [(1, READ)]),
|
||||
("f = os.fdopen(fd, 'r', encoding='utf-8')", [(1, READ)]),
|
||||
('data = path.read_text(encoding="utf_8")', [(1, READ)]),
|
||||
("with open(path, mode='r', encoding='utf-8') as f:", [(1, READ)]),
|
||||
('path.read_text(encoding="utf-8-sig")', []),
|
||||
('path.write_text(data, encoding="utf-8")', []),
|
||||
("open(path, 'w', encoding='utf-8')", []),
|
||||
("open(path, 'a', encoding='utf-8')", []),
|
||||
("open(path, 'r+', encoding='utf-8')", []),
|
||||
("subprocess.run(cmd, text=True, encoding='utf-8')", []),
|
||||
("path.read_text(encoding='utf-8') # windows-footgun: ok — owned file", []),
|
||||
("# use path.read_text(encoding='utf-8') here", []),
|
||||
('path.write_text(data, encoding="utf-8-sig")', [(1, WRITE)]),
|
||||
("open(path, 'w', encoding='utf-8-sig')", [(1, WRITE)]),
|
||||
("open(path, 'a', encoding='utf-8-sig')", [(1, WRITE)]),
|
||||
("os.fdopen(fd, 'w', encoding='utf-8-sig')", [(1, WRITE)]),
|
||||
('path.open("a", encoding="utf-8-sig")', [(1, WRITE)]),
|
||||
('path.open("a+", encoding="utf-8-sig")', [(1, WRITE)]),
|
||||
('path.open("r+", encoding="utf-8-sig")', [(1, WRITE)]),
|
||||
('path.write_text(data, encoding="utf_8_sig")', [(1, WRITE)]),
|
||||
("open(path, 'r', encoding='utf-8-sig')", []),
|
||||
("open(path, encoding='utf-8-sig')", []),
|
||||
("codec = 'utf-8-sig'", []),
|
||||
('open(os.path.join(root, "install-stamp.json"), encoding="utf-8-sig")', []),
|
||||
('path.open("r", encoding="utf-8-sig")', []),
|
||||
('path.open("a", encoding="utf-8")', []),
|
||||
('something.open(encoding="utf-8")', []),
|
||||
("path.write_text(data, encoding='utf-8-sig') # windows-footgun: ok — BOM required", []),
|
||||
("open(p, 'x', encoding='utf-8-sig')", [(1, WRITE)]),
|
||||
("enc = 'utf-8'", []),
|
||||
("path.read_text(encoding='utf-8') # merely mentions windows-footgun", [(1, READ)]),
|
||||
("path.read_text(encoding='utf-8') if hasattr(path, 'read_text') else ''", [(1, READ)]),
|
||||
("path.read_text(encoding='utf-8') if sys.platform != 'win32' else ''", []),
|
||||
('"""Example:\npath.read_text(encoding="utf-8")\n"""\npath.read_text(encoding="utf-8")', [(4, READ)]),
|
||||
])
|
||||
def test_encoding_diagnostics(linter, tmp_path, source, expected):
|
||||
path = tmp_path / "input.py"
|
||||
path.write_text(source + "\n", encoding="utf-8")
|
||||
rules = [rule for rule in linter.FOOTGUNS if rule.name in {READ, WRITE}]
|
||||
assert {rule.name for rule in rules} == {READ, WRITE}
|
||||
assert [(line, rule.name) for line, _, rule in linter.scan_file(path, rules)] == expected
|
||||
|
||||
@@ -3,6 +3,10 @@ import colorsys
|
||||
import io
|
||||
import itertools
|
||||
import os
|
||||
import json
|
||||
import shutil
|
||||
import tomllib
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
import struct
|
||||
import subprocess
|
||||
@@ -18,30 +22,61 @@ ROOT = Path(__file__).resolve().parents[2]
|
||||
@pytest.fixture(scope="module")
|
||||
def generate(tmp_path_factory):
|
||||
root = tmp_path_factory.mktemp("icon-flavors")
|
||||
source = root / "source with spaces"
|
||||
source.mkdir()
|
||||
foreign = root / "foreign-site"
|
||||
foreign.mkdir()
|
||||
(foreign / "sitecustomize.py").write_text("raise SystemExit('foreign interpreter path leaked')\n", encoding="utf-8")
|
||||
shutil.copytree(ROOT / "assets", source / "assets")
|
||||
# A real app-only wheel makes --only-group load-bearing: startup dies if
|
||||
# the driver accidentally includes application dependencies in the renderer.
|
||||
from tests.pm._fixtures import _wheel
|
||||
wheel = _wheel(source, "application_only")
|
||||
with zipfile.ZipFile(wheel, "a") as archive:
|
||||
archive.writestr("application_only.pth", "import sys; sys.exit('application dependency leaked into icon renderer')\n")
|
||||
group = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8-sig"))["dependency-groups"]["icon-build"]
|
||||
(source / "pyproject.toml").write_text(
|
||||
'[project]\nname="icon-fixture"\nversion="1"\nrequires-python=">=3.11"\n'
|
||||
'dependencies=["application-only==1.0"]\n[dependency-groups]\nicon-build=' + json.dumps(group) + '\n'
|
||||
'[tool.uv]\npackage=false\n[tool.uv.sources]\napplication-only={path=' + json.dumps(wheel.as_posix()) + '}\n', encoding="utf-8")
|
||||
uv, node = shutil.which("uv"), shutil.which("node")
|
||||
assert uv and node, "icon acceptance requires prepared uv and Node"
|
||||
subprocess.run([uv, "lock", "--python", sys.executable], cwd=source, check=True, capture_output=True, timeout=60)
|
||||
outputs = {}
|
||||
sequence = itertools.count()
|
||||
|
||||
def build(tag="", commit="", *, rejected=False):
|
||||
key = (tag, commit)
|
||||
def build(tag="", commit="", *, rejected=False, on_demand=False):
|
||||
key = (tag, commit, on_demand)
|
||||
if key not in outputs:
|
||||
out = root / str(next(sequence))
|
||||
env = {**os.environ, "HERMES_HOME": str(root / "home"),
|
||||
"HERMES_RUNTIME_DIR": str(root / "tools"),
|
||||
"HERMES_PAYLOAD_TAG": tag, "HERMES_BUILD_COMMIT": commit}
|
||||
command = [sys.executable, str(ROOT / "scripts/build/icon_environment.py"),
|
||||
"--source", str(ROOT), "--out", str(out)]
|
||||
"HERMES_PAYLOAD_TAG": tag, "HERMES_BUILD_COMMIT": commit,
|
||||
"HERMES_PYTHON": sys.executable, "PYTHONPATH": str(root / "foreign-site"),
|
||||
"PYTHONHOME": str(root / "foreign-python"), "HERMES_DISABLE_LAZY_INSTALLS": "1"}
|
||||
command = [node, str(ROOT / "scripts/generate-icons.mjs"),
|
||||
"--source", str(source), "--out", str(out), *(["--on-demand"] if on_demand else [])]
|
||||
result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=180)
|
||||
if rejected:
|
||||
assert result.returncode != 0, "invalid build identity generated icons"
|
||||
assert not out.exists()
|
||||
if on_demand:
|
||||
assert "disabled" in (result.stdout + result.stderr).lower()
|
||||
return
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
if not outputs:
|
||||
checked = subprocess.run([*command, "--check"], env=env, capture_output=True, text=True, timeout=180)
|
||||
assert checked.returncode == 0, checked.stdout + checked.stderr
|
||||
outputs[key] = out
|
||||
return outputs[key]
|
||||
|
||||
return build
|
||||
|
||||
|
||||
def test_on_demand_build_obeys_disabled_lazy_install_admission(generate):
|
||||
generate(on_demand=True, rejected=True)
|
||||
|
||||
|
||||
def frames(path):
|
||||
"""Read every native frame, including ICO entries Pillow's n_frames misses."""
|
||||
data = path.read_bytes()
|
||||
|
||||
@@ -111,49 +111,3 @@ def test_pm_builder_ignores_ambient_uv_configuration(tmp_path, monkeypatch, pois
|
||||
monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable)))
|
||||
executable = stage_manager_runtime(python=Path(sys.executable), destination=tmp_path / "runtime")
|
||||
assert executable.is_file()
|
||||
|
||||
|
||||
def test_native_stage_builds_pm_before_application_environment(tmp_path, monkeypatch):
|
||||
from scripts.bundles import native, payload
|
||||
from types import SimpleNamespace
|
||||
|
||||
class StopAfterPM(Exception):
|
||||
pass
|
||||
|
||||
class Facts:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def retain(self, names):
|
||||
pass
|
||||
|
||||
def entries_in_use(self):
|
||||
return set()
|
||||
|
||||
def get(self, name):
|
||||
return {"entry": "python"}
|
||||
|
||||
calls = []
|
||||
lock = tmp_path / "hermes-agent/pm/lock.json"
|
||||
lock.parent.mkdir(parents=True)
|
||||
shutil.copy2(Path(__file__).resolve().parents[2] / "pm/lock.json", lock)
|
||||
monkeypatch.setattr(payload, "snapshot", lambda *args: None)
|
||||
monkeypatch.setattr(native, "_bundle_package_names", lambda: [])
|
||||
monkeypatch.setattr(native, "_install_names", lambda names: 0)
|
||||
monkeypatch.setattr(native, "Facts", Facts)
|
||||
monkeypatch.setattr(native, "_facts", Facts)
|
||||
monkeypatch.setattr(native, "_store", lambda: SimpleNamespace(entry=lambda name: tmp_path / "tools" / name))
|
||||
monkeypatch.setattr(native, "get_package", lambda name: SimpleNamespace(binary=lambda path, target: path / "python"))
|
||||
|
||||
cache_dir = tmp_path / "cache"
|
||||
monkeypatch.setenv("UV_CACHE_DIR", str(cache_dir))
|
||||
|
||||
def staged(root, python, repo, *, cache):
|
||||
assert cache == cache_dir
|
||||
calls.append((root, python, repo))
|
||||
raise StopAfterPM
|
||||
|
||||
monkeypatch.setattr(native, "stage_pm_runtime", staged)
|
||||
with pytest.raises(StopAfterPM):
|
||||
native._stage_native(SimpleNamespace(out=str(tmp_path), ref="HEAD"))
|
||||
assert calls == [(tmp_path, tmp_path / "tools/python/python", tmp_path / "hermes-agent")]
|
||||
|
||||
@@ -12,10 +12,11 @@ import pytest
|
||||
from pm.lock import Lockfile
|
||||
from pm.paths import lockfile_path
|
||||
from pm.store import current_target
|
||||
from tests.pm._fixtures import build_worker, client, isolated_python # noqa: F401
|
||||
|
||||
|
||||
@pytest.mark.parametrize("extras", [[], ["dev"]], ids=["runtime", "tests"])
|
||||
def test_development_setup_keeps_test_groups_out_of_the_runtime(tmp_path, monkeypatch, extras):
|
||||
def test_development_setup_keeps_test_groups_out_of_the_runtime(tmp_path, monkeypatch, extras, build_worker):
|
||||
from types import SimpleNamespace
|
||||
import shutil
|
||||
|
||||
@@ -36,17 +37,10 @@ def test_development_setup_keeps_test_groups_out_of_the_runtime(tmp_path, monkey
|
||||
'[tool.uv]\npackage=false\nno-index=true\n'
|
||||
f'find-links=[{json.dumps(wheels.as_posix())}]\n', encoding="utf-8",
|
||||
)
|
||||
uv = shutil.which("uv")
|
||||
assert uv
|
||||
monkeypatch.setattr("pm.client.is_runtime", lambda: True)
|
||||
monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable)))
|
||||
lock_project(core, python=Path(sys.executable), offline=True, explicit=True)
|
||||
home = tmp_path / "ci-home"
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setattr("pm.paths.repo_root", lambda: core)
|
||||
# This test exercises the worker-side CI environment split with offline uv.
|
||||
# Dispatch into that worker is covered by test_runtime_entrypoints.
|
||||
monkeypatch.setattr("pm.runtime.is_runtime", lambda: True)
|
||||
files = {name: tmp_path / name for name in ("GITHUB_ENV", "GITHUB_OUTPUT", "GITHUB_PATH")}
|
||||
for name, file in files.items():
|
||||
monkeypatch.setenv(name, str(file))
|
||||
|
||||
@@ -113,91 +113,15 @@ def test_observer_preserves_no_desktop_and_refuses_incomplete_app(tmp_path):
|
||||
|
||||
|
||||
@pytest.mark.platforms("posix")
|
||||
def test_observer_checks_real_compiler_receipts_without_repairing(tmp_path):
|
||||
verify = runpy.run_path(str(ASSETS / "source_driver.py"))["verify_products"]
|
||||
path = os.pathsep.join(p for p in os.get_exec_path() if ".hermes" not in Path(p).parts)
|
||||
node = shutil.which("node", path=path)
|
||||
assert node, "source-driver tests require the prepared Node tool"
|
||||
root = tmp_path / "source"
|
||||
build = root / "scripts/build"
|
||||
build.mkdir(parents=True)
|
||||
repo = ASSETS.parents[2]
|
||||
for name in ("freshness.mjs", "frontend-common.mjs"):
|
||||
shutil.copy2(repo / "scripts/build" / name, build / name)
|
||||
outputs = {"tui": root / "ui-tui/dist", "web": root / "hermes_cli/web_dist",
|
||||
"desktop": root / "apps/desktop/release/linux-unpacked/resources/app.asar.unpacked/dist"}
|
||||
for out in outputs.values():
|
||||
out.mkdir(parents=True)
|
||||
(out / "index.html").write_text("fixture renderer", encoding="utf-8")
|
||||
(root / "apps/desktop/release/linux-unpacked/hermes").write_text("fixture executable", encoding="utf-8")
|
||||
record = '''import { buildInputs, recordProduct } from './scripts/build/freshness.mjs';
|
||||
const source = process.cwd();
|
||||
for (const [product, out] of Object.entries(JSON.parse(process.argv[1]))) {
|
||||
recordProduct({ source, product, out, inputs: buildInputs(source, product) });
|
||||
}
|
||||
'''
|
||||
subprocess.run([node, "--input-type=module", "-e", record, json.dumps({k: str(v) for k, v in outputs.items()})],
|
||||
cwd=root, check=True, timeout=30)
|
||||
def snapshot():
|
||||
return {str(p.relative_to(root)): (p.read_bytes(), p.stat().st_mtime_ns)
|
||||
for p in root.rglob("*") if p.is_file()}
|
||||
before = snapshot()
|
||||
verify(root, "present", Path(node))
|
||||
assert snapshot() == before
|
||||
# The successful receipts cannot bless damaged or missing outputs.
|
||||
damaged = outputs["web"] / "index.html"
|
||||
damaged.write_text("damaged renderer", encoding="utf-8")
|
||||
before = snapshot()
|
||||
with pytest.raises(RuntimeError, match="web output"):
|
||||
verify(root, "present", Path(node))
|
||||
assert snapshot() == before
|
||||
damaged.unlink()
|
||||
with pytest.raises(RuntimeError, match="web output"):
|
||||
verify(root, "present", Path(node))
|
||||
assert not damaged.exists()
|
||||
|
||||
|
||||
@pytest.mark.platforms("posix")
|
||||
def test_pm_probe_rejects_foreign_launcher_before_any_bootstrap(tmp_path, monkeypatch):
|
||||
from hermes_cli import _launchers
|
||||
|
||||
root = tmp_path / "installed source"
|
||||
foreign = tmp_path / "other source"
|
||||
repo = ASSETS.parents[2]
|
||||
# Real published-launcher protocol, with only the stdlib pre-boot closure.
|
||||
for tree in (root, foreign):
|
||||
for name in ("hermes_constants.py", "hermes_cli/__init__.py", "hermes_cli/_launchers.py",
|
||||
"hermes_cli/runtime_paths.py"):
|
||||
dest = tree / name
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(repo / name, dest)
|
||||
(tree / "pm").mkdir()
|
||||
(tree / "pm/lock.json").write_text("{}", encoding="utf-8")
|
||||
(tree / "hermes_bootstrap.py").write_text("raise RuntimeError('bootstrap must not run')", encoding="utf-8")
|
||||
home = tmp_path / "home"
|
||||
store = home / "tools"
|
||||
store.mkdir(parents=True)
|
||||
interpreter = Path(sys._base_executable).resolve()
|
||||
(store / "facts.json").write_text(json.dumps({"packages": {"python": {
|
||||
"entry": str(interpreter.parents[1])}}}), encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setenv("HERMES_RUNTIME_DIR", str(store))
|
||||
bin_dir = root / ".hermes/bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
launcher = _launchers.mint_launcher("hermes", foreign, bin_dir, interpreter, None)
|
||||
assert launcher is not None
|
||||
result = subprocess.run([sys.executable, "-B", str(ASSETS / "source_driver.py"),
|
||||
"--root", str(root), "--launcher", str(launcher), "--desktop", "absent"],
|
||||
cwd=tmp_path, env=dict(os.environ, HOME=str(tmp_path)),
|
||||
capture_output=True, text=True, timeout=30)
|
||||
assert result.returncode != 0
|
||||
assert "belongs to another installation" in result.stderr
|
||||
assert "bootstrap must not run" not in result.stderr
|
||||
assert not (root / "venv").exists()
|
||||
|
||||
|
||||
@pytest.mark.platforms("posix")
|
||||
def test_pm_observer_accepts_ready_fixture_and_leaves_failed_fixture_untouched(tmp_path):
|
||||
@pytest.mark.parametrize("fault,error", [
|
||||
("uv-lock", "dependency generation is not current"),
|
||||
("no-desktop", "dependency generation is not current"),
|
||||
("foreign-launcher", "belongs to another installation"),
|
||||
("missing-launcher", ""),
|
||||
("incomplete", "incomplete"),
|
||||
("web-changed", "web output"), ("web-missing", "web output"),
|
||||
])
|
||||
def test_pm_observer_accepts_ready_fixture_and_leaves_failed_fixture_untouched(tmp_path, fault, error):
|
||||
# This exercises the complete observer process, not a whole install. The
|
||||
# fixture borrows prepared test dependencies and records real PM input
|
||||
# stamps / compiler receipts. It never resolves or acquires a package.
|
||||
@@ -250,14 +174,18 @@ ensure_install_launchers(root, root / '.hermes/bin')
|
||||
record = '''import { mkdirSync, writeFileSync } from 'node:fs';
|
||||
import { buildInputs, recordProduct } from './scripts/build/freshness.mjs';
|
||||
const source = process.cwd();
|
||||
for (const [product, out] of [['tui', 'ui-tui/dist'], ['web', 'hermes_cli/web_dist']]) {
|
||||
for (const [product, out] of [['tui', 'ui-tui/dist'], ['web', 'hermes_cli/web_dist'], ['desktop', 'apps/desktop/release/linux-unpacked/resources/app.asar.unpacked/dist']]) {
|
||||
mkdirSync(out, { recursive: true }); writeFileSync(out + '/index.html', 'fixture product');
|
||||
recordProduct({source, product, out, inputs: buildInputs(source, product)});
|
||||
}
|
||||
'''
|
||||
subprocess.run([node, "--input-type=module", "-e", record], cwd=root, env=env, check=True, timeout=30)
|
||||
(root / "apps/desktop/release/linux-unpacked/hermes").write_bytes(b"fixture executable")
|
||||
command = [sys.executable, "-B", str(ASSETS / "source_driver.py"), "--root", str(root),
|
||||
"--launcher", str(root / ".hermes/bin/hermes"), "--desktop", "absent"]
|
||||
"--launcher", str(root / ".hermes/bin/hermes"), "--desktop", "present"]
|
||||
if fault == "no-desktop":
|
||||
shutil.rmtree(root / "apps/desktop")
|
||||
command[-1] = "absent"
|
||||
# The passive query must not write even import caches: -I ignores the
|
||||
# environment's bytecode switch, so the published query enforces it.
|
||||
def snapshot():
|
||||
@@ -267,9 +195,27 @@ for (const [product, out] of [['tui', 'ui-tui/dist'], ['web', 'hermes_cli/web_di
|
||||
result = subprocess.run(command, env=env, cwd=tmp_path, capture_output=True, text=True, timeout=60)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
assert snapshot() == before
|
||||
(root / "uv.lock").write_text("changed graph without preparation", encoding="utf-8")
|
||||
if fault in {"uv-lock", "no-desktop"}:
|
||||
(root / "uv.lock").write_text("changed graph without preparation", encoding="utf-8")
|
||||
elif fault == "foreign-launcher":
|
||||
foreign = tmp_path / "foreign"
|
||||
shutil.copytree(root, foreign)
|
||||
from hermes_cli._launchers import mint_launcher
|
||||
launcher = mint_launcher("hermes", foreign, root / ".hermes/bin", Path(sys.executable), None)
|
||||
assert launcher is not None
|
||||
elif fault == "missing-launcher":
|
||||
(root / ".hermes/bin/hermes").unlink()
|
||||
elif fault == "incomplete":
|
||||
(root / ".update-incomplete").write_text("incomplete", encoding="utf-8")
|
||||
else:
|
||||
damaged = root / "hermes_cli/web_dist/index.html"
|
||||
if fault == "web-missing":
|
||||
damaged.unlink()
|
||||
else:
|
||||
damaged.write_bytes(b"damaged")
|
||||
before = snapshot()
|
||||
result = subprocess.run(command, env=env, cwd=tmp_path, capture_output=True, text=True, timeout=60)
|
||||
assert result.returncode != 0
|
||||
assert "dependency generation is not current" in result.stderr
|
||||
assert error in result.stderr
|
||||
assert "bootstrap must not run" not in result.stderr
|
||||
assert snapshot() == before
|
||||
@@ -51,11 +51,6 @@ def _git(repo: Path, *args: str) -> str:
|
||||
return out.stdout.strip()
|
||||
|
||||
_SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "verify-bootstrap-version-stamp.py"
|
||||
_spec = importlib.util.spec_from_file_location("verify_bootstrap_version_stamp", _SCRIPT)
|
||||
if _spec is None or _spec.loader is None:
|
||||
raise ImportError("Failed to load verify-bootstrap-version-stamp.py")
|
||||
_mod = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(_mod)
|
||||
|
||||
|
||||
def _install_repo(tmp_path: Path) -> Path:
|
||||
@@ -72,73 +67,34 @@ def _install_repo(tmp_path: Path) -> Path:
|
||||
return repo
|
||||
|
||||
|
||||
def _stamp(repo: Path, **overrides: object) -> Path:
|
||||
stamp = {
|
||||
"schemaVersion": 1,
|
||||
"pinnedCommit": _git(repo, "rev-parse", "HEAD"),
|
||||
"pinnedBranch": "main",
|
||||
"completedAt": "2026-08-30T12:00:00.000Z",
|
||||
}
|
||||
stamp.update(overrides)
|
||||
@pytest.mark.parametrize("changes,expect,error", [
|
||||
({}, [], None),
|
||||
({"pinnedCommit": "a" * 40}, [], "installed HEAD"),
|
||||
({}, ["--expect-commit", "b" * 40], "expected"),
|
||||
({}, ["--expect-branch", "release"], "pinnedBranch"),
|
||||
({"schemaVersion": 2}, [], "schemaVersion"),
|
||||
({"pinnedCommit": "deadbeef"}, [], "40-char"),
|
||||
({"completedAt": "not-a-time"}, [], "ISO-8601"),
|
||||
({"completedAt": "2026-08-30T12:00:00+01:00"}, [], "not UTC"),
|
||||
({"missing": "stamp"}, [], "cannot read stamp"),
|
||||
({"missing": "version"}, [], "no __version__"),
|
||||
])
|
||||
def test_verifier_cli(tmp_path, changes, expect, error):
|
||||
repo = _install_repo(tmp_path)
|
||||
stamp = {"schemaVersion": 1, "pinnedCommit": _git(repo, "rev-parse", "HEAD"),
|
||||
"pinnedBranch": "main", "completedAt": "2026-08-30T12:00:00.000Z", **changes}
|
||||
path = repo / ".hermes-bootstrap-complete"
|
||||
path.write_text(json.dumps(stamp, indent=2) + "\n", encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def test_honest_stamp_verifies(tmp_path: Path):
|
||||
repo = _install_repo(tmp_path)
|
||||
errors = _mod.verify_stamp(_stamp(repo), repo, None, None)
|
||||
assert errors == []
|
||||
|
||||
|
||||
def test_pinned_commit_must_match_installed_head(tmp_path: Path):
|
||||
repo = _install_repo(tmp_path)
|
||||
liar = "a" * 40
|
||||
errors = _mod.verify_stamp(_stamp(repo, pinnedCommit=liar), repo, None, None)
|
||||
assert any("pinnedCommit" in e and "HEAD" in e for e in errors), errors
|
||||
|
||||
|
||||
def test_expect_commit_and_branch_are_enforced(tmp_path: Path):
|
||||
repo = _install_repo(tmp_path)
|
||||
head = _git(repo, "rev-parse", "HEAD")
|
||||
# Matching expectations pass.
|
||||
assert _mod.verify_stamp(_stamp(repo), repo, head, "main") == []
|
||||
# A stale expectation fails.
|
||||
errors = _mod.verify_stamp(_stamp(repo), repo, "b" * 40, "release")
|
||||
assert len(errors) == 2, errors
|
||||
|
||||
|
||||
def test_bad_shape_and_timestamps_are_refused(tmp_path: Path):
|
||||
repo = _install_repo(tmp_path)
|
||||
errors = _mod.verify_stamp(
|
||||
_stamp(
|
||||
repo,
|
||||
schemaVersion=2,
|
||||
pinnedCommit="deadbeef",
|
||||
completedAt="not-a-time",
|
||||
),
|
||||
repo,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
joined = "\n".join(errors)
|
||||
assert "schemaVersion" in joined
|
||||
assert "40-char" in joined
|
||||
assert "ISO-8601" in joined
|
||||
|
||||
|
||||
def test_missing_stamp_or_version_is_refused(tmp_path: Path):
|
||||
repo = _install_repo(tmp_path)
|
||||
errors = _mod.verify_stamp(repo / "nope.json", repo, None, None)
|
||||
assert errors and "cannot read stamp" in errors[0]
|
||||
|
||||
empty = tmp_path / "empty"
|
||||
empty.mkdir()
|
||||
stamp_path = _stamp(repo) # valid stamp...
|
||||
(repo / "hermes_cli" / "__init__.py").unlink()
|
||||
_git(repo, "add", "-A")
|
||||
_git(repo, "commit", "-m", "drop version")
|
||||
# The stamp's commit no longer matches HEAD AND no version ships.
|
||||
errors = _mod.verify_stamp(stamp_path, repo, None, None)
|
||||
joined = "\n".join(errors)
|
||||
assert "no __version__" in joined
|
||||
if changes.get("missing") != "stamp":
|
||||
path.write_text(json.dumps(stamp), encoding="utf-8")
|
||||
if changes.get("missing") == "version":
|
||||
(repo / "hermes_cli/__init__.py").unlink()
|
||||
if not error:
|
||||
expect = ["--expect-commit", stamp["pinnedCommit"], "--expect-branch", "main"]
|
||||
env = dict(os.environ)
|
||||
if os.name == "nt":
|
||||
env.setdefault("SystemRoot", r"C:\Windows")
|
||||
env.setdefault("ComSpec", r"C:\Windows\system32\cmd.exe")
|
||||
result = subprocess.run([sys.executable, str(_SCRIPT), "--repo", str(repo), "--stamp", str(path), *expect],
|
||||
env=env, capture_output=True, text=True, timeout=30)
|
||||
assert result.returncode == (1 if error else 0), result.stderr
|
||||
assert error in result.stderr if error else "stamp verified" in result.stdout
|
||||
|
||||
@@ -60,6 +60,11 @@ def home(tmp_path):
|
||||
directory-only, so the scanner cannot recurse into a dependency graph and
|
||||
the test needs no network/Torch."""
|
||||
h = tmp_path / "hermes-home"
|
||||
if os.name != "nt":
|
||||
vpp.seed_fixtures(h, tmp_path / "external-mnemosyne-runtime")
|
||||
(h / "profiles/e2e-preserve").rename(h / "profiles/work")
|
||||
return h
|
||||
# Retained native NTFS fixture until the shared-seed successor runs on Windows.
|
||||
# active-home plugin: directory wrapper with marker + payload
|
||||
plugin = h / "plugins" / "mnemosyne-wrapper"
|
||||
plugin.mkdir(parents=True)
|
||||
@@ -79,162 +84,65 @@ def home(tmp_path):
|
||||
return h
|
||||
|
||||
|
||||
def _snapshot(home, out):
|
||||
snap = vpp.snapshot_home(str(home))
|
||||
with open(out, "w", encoding="utf-8") as fh:
|
||||
json.dump(snap, fh)
|
||||
return snap
|
||||
|
||||
|
||||
def _verify(home, snap):
|
||||
return vpp.verify_home(str(home), snap)
|
||||
|
||||
|
||||
def test_snapshot_records_all_trees(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
keys = set(snap["entries"])
|
||||
assert "plugins/mnemosyne-wrapper/mnemosyne-wrapper.json" in keys
|
||||
assert "plugins/mnemosyne-wrapper/plugin.py" in keys
|
||||
assert "profiles/work/plugins/second-plugin/data.bin" in keys
|
||||
assert snap["roots"] == [
|
||||
"<home>/plugins",
|
||||
"<home>/profiles/work/plugins",
|
||||
]
|
||||
|
||||
|
||||
def test_snapshot_captures_bytes_and_symlinks(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
e = snap["entries"]
|
||||
assert e["plugins/mnemosyne-wrapper/plugin.py"]["sha256"] != ""
|
||||
assert e["plugins/mnemosyne-wrapper/plugin.py"]["size"] == 16
|
||||
link = e["plugins/mnemosyne-wrapper/runtime"]
|
||||
assert link["kind"] == "symlink"
|
||||
assert link["target_resolves"] is True
|
||||
assert link["target_kind"] == "dir"
|
||||
tree = link["target_tree"]
|
||||
assert tree["sidecar-witness.txt"]["sha256"] != ""
|
||||
assert tree["engine.bin"]["kind"] == "file"
|
||||
|
||||
|
||||
def test_untouched_home_verifies_clean(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is True
|
||||
assert report["counts"]["deleted"] == 0
|
||||
assert report["counts"]["modified"] == 0
|
||||
|
||||
|
||||
def test_catches_plugin_file_deletion(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
(home / "plugins" / "mnemosyne-wrapper" / "plugin.py").unlink()
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is False
|
||||
assert "plugins/mnemosyne-wrapper/plugin.py" in report["deleted"]
|
||||
|
||||
|
||||
def test_catches_whole_plugin_root_deletion(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
shutil.rmtree(home / "plugins")
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is False
|
||||
assert report["counts"]["deleted"] > 0
|
||||
|
||||
|
||||
def test_catches_wrapper_marker_modification(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
(home / "plugins" / "mnemosyne-wrapper" / "mnemosyne-wrapper.json").write_text(
|
||||
'{"wrapper": false}\n', encoding="utf-8"
|
||||
)
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is False
|
||||
assert "plugins/mnemosyne-wrapper/mnemosyne-wrapper.json" in report["modified"]
|
||||
|
||||
|
||||
def test_catches_symlink_target_repoint(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
link = home / "plugins" / "mnemosyne-wrapper" / "runtime"
|
||||
other = tmp_path / "other-runtime"
|
||||
other.mkdir()
|
||||
(other / "sidecar-witness.txt").write_text("different\n", encoding="utf-8")
|
||||
_remove_link(link)
|
||||
_make_link(other, link)
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is False
|
||||
assert "plugins/mnemosyne-wrapper/runtime" in report["modified"]
|
||||
|
||||
|
||||
def test_catches_external_witness_modification(home, tmp_path):
|
||||
# The externally-owned sidecar witness lives OUTSIDE the home; an upgrade
|
||||
# that tramples it must still be caught through the symlink fingerprint.
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
witness = tmp_path / "external-mnemosyne-runtime" / "sidecar-witness.txt"
|
||||
witness.write_text("external-witness-TAMPERED\n", encoding="utf-8")
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is False
|
||||
# Depending on the platform's walk, the tamper surfaces either as the
|
||||
# link entry (target fingerprint) or as the linked file itself.
|
||||
assert any(
|
||||
"runtime" in key for key in list(report["modified"]) + report["deleted"]
|
||||
)
|
||||
|
||||
|
||||
def test_catches_external_witness_deletion(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
(tmp_path / "external-mnemosyne-runtime" / "engine.bin").unlink()
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is False
|
||||
|
||||
|
||||
def test_catches_profile_plugin_deletion(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
(home / "profiles" / "work" / "plugins" / "second-plugin" / "data.bin").unlink()
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is False
|
||||
assert "profiles/work/plugins/second-plugin/data.bin" in report["deleted"]
|
||||
|
||||
|
||||
def test_added_entries_do_not_fail(home, tmp_path):
|
||||
# An upgrade may ADD files (new bundled plugin, caches); only taking
|
||||
# away or changing existing entries is a violation.
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
newp = home / "plugins" / "fresh-from-upgrade"
|
||||
newp.mkdir()
|
||||
(newp / "b.txt").write_text("new\n", encoding="utf-8")
|
||||
report = _verify(home, snap)
|
||||
assert report["ok"] is True
|
||||
assert "plugins/fresh-from-upgrade/b.txt" in report["added"]
|
||||
|
||||
|
||||
def test_verifier_is_read_only_against_home(home, tmp_path):
|
||||
snap = _snapshot(home, tmp_path / "snap.json")
|
||||
before = sorted(
|
||||
(p, p.stat().st_size if p.is_file() else "dir")
|
||||
for p in home.rglob("*")
|
||||
)
|
||||
_verify(home, snap)
|
||||
_verify(home, snap)
|
||||
after = sorted(
|
||||
(p, p.stat().st_size if p.is_file() else "dir")
|
||||
for p in home.rglob("*")
|
||||
)
|
||||
assert before == after
|
||||
|
||||
|
||||
def test_catches_empty_dir_deletion(home, tmp_path):
|
||||
# A plugin directory emptied (or an empty dir removed) must be caught:
|
||||
# directories themselves are recorded, not skipped.
|
||||
empty = home / "plugins" / "wrapper-b" / "empty-cache"
|
||||
@pytest.mark.parametrize("relative,action,category", [
|
||||
("plugins/mnemosyne-wrapper/plugin.py", "delete", "deleted"),
|
||||
("plugins", "tree", "deleted"),
|
||||
("plugins/mnemosyne-wrapper/mnemosyne-wrapper.json", "change", "modified"),
|
||||
("plugins/mnemosyne-wrapper/runtime", "repoint", "modified"),
|
||||
("plugins/mnemosyne-wrapper/runtime/sidecar-witness.txt", "change", "modified"),
|
||||
("plugins/mnemosyne-wrapper/runtime/engine.bin", "delete", "modified"),
|
||||
("profiles/work/plugins/second-plugin/data.bin", "delete", "deleted"),
|
||||
("plugins/wrapper-b/empty-cache", "tree", "deleted"),
|
||||
("plugins/fresh-from-upgrade/b.txt", "add", "added"),
|
||||
])
|
||||
def test_preservation_cli_fault_matrix(home, tmp_path, relative, action, category):
|
||||
empty = home / "plugins/wrapper-b/empty-cache"
|
||||
empty.mkdir(parents=True)
|
||||
snap2 = _snapshot(home, tmp_path / "snap2.json")
|
||||
assert snap2["entries"]["plugins/wrapper-b/empty-cache"] == {"kind": "dir"}
|
||||
empty.rmdir()
|
||||
(home / "plugins" / "wrapper-b").rmdir()
|
||||
report = _verify(home, snap2)
|
||||
assert report["ok"] is False
|
||||
assert "plugins/wrapper-b/empty-cache" in report["deleted"]
|
||||
snapshot = tmp_path / "snap.json"
|
||||
report = tmp_path / "report.json"
|
||||
command = [sys.executable, VERIFIER, "verify", "--home", str(home),
|
||||
"--snapshot", str(snapshot), "--report", str(report)]
|
||||
def fingerprint():
|
||||
return {str(p): (p.read_bytes() if p.is_file() else None, p.lstat().st_mtime_ns,
|
||||
os.readlink(p) if p.is_symlink() or p.is_junction() else None)
|
||||
for root in (home, tmp_path / "external-mnemosyne-runtime") for p in root.rglob("*")}
|
||||
before = fingerprint()
|
||||
result = subprocess.run([sys.executable, VERIFIER, "snapshot", "--home", str(home), "--out", str(snapshot)],
|
||||
capture_output=True, text=True, timeout=30)
|
||||
assert result.returncode == 0, result.stderr
|
||||
entries = json.loads(snapshot.read_text(encoding="utf-8-sig"))["entries"]
|
||||
assert {"plugins/mnemosyne-wrapper/plugin.py", "profiles/work/plugins/second-plugin/data.bin",
|
||||
"plugins/mnemosyne-wrapper/mnemosyne-wrapper.json", "plugins/wrapper-b/empty-cache"} <= entries.keys()
|
||||
assert entries["plugins/wrapper-b/empty-cache"] == {"kind": "dir"}
|
||||
link = entries["plugins/mnemosyne-wrapper/runtime"]
|
||||
assert link["kind"] == "symlink" and link["target_resolves"] and link["target_kind"] == "dir"
|
||||
assert {"engine.bin", "sidecar-witness.txt"} <= link["target_tree"].keys()
|
||||
assert subprocess.run(command, capture_output=True, text=True, timeout=30).returncode == 0
|
||||
assert fingerprint() == before
|
||||
target = home / relative
|
||||
if action == "repoint":
|
||||
other = tmp_path / "other-runtime"
|
||||
other.mkdir()
|
||||
_remove_link(target)
|
||||
_make_link(other, target)
|
||||
elif action == "tree":
|
||||
shutil.rmtree(target)
|
||||
elif action == "delete":
|
||||
target.unlink()
|
||||
else:
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_bytes(b"tampered or added")
|
||||
before = fingerprint()
|
||||
result = subprocess.run(command, capture_output=True, text=True, timeout=30)
|
||||
assert result.returncode == (0 if action == "add" else 1), result.stderr
|
||||
data = json.loads(report.read_text(encoding="utf-8-sig"))
|
||||
affected = "plugins/mnemosyne-wrapper/runtime" if "/runtime/" in relative else relative
|
||||
assert affected in data[category]
|
||||
assert data["ok"] == (action == "add")
|
||||
assert fingerprint() == before
|
||||
|
||||
|
||||
def test_empty_snapshot_is_inconclusive(home, tmp_path):
|
||||
def test_empty_snapshot_is_inconclusive(tmp_path):
|
||||
# Zero recorded entries cannot prove anything: the CLI refuses.
|
||||
empty_home = tmp_path / "bare-home"
|
||||
empty_home.mkdir()
|
||||
@@ -268,7 +176,7 @@ def test_unreadable_path_is_hard_error(home, tmp_path):
|
||||
os.chmod(secret, 0o000)
|
||||
try:
|
||||
with pytest.raises((OSError, vpp.ScanError)):
|
||||
_snapshot(home, tmp_path / "snap2.json")
|
||||
vpp.snapshot_home(str(home))
|
||||
finally:
|
||||
os.chmod(secret, 0o755)
|
||||
|
||||
@@ -282,28 +190,6 @@ def test_missing_home_fails_snapshot(tmp_path):
|
||||
assert proc.returncode == 2
|
||||
|
||||
|
||||
def test_cli_roundtrip_end_to_end(home, tmp_path):
|
||||
"""The exact command shape the E2E drivers use."""
|
||||
snap_file = tmp_path / "snap.json"
|
||||
r1 = subprocess.run(
|
||||
[sys.executable, VERIFIER, "snapshot", "--home", str(home), "--out", str(snap_file)],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
assert r1.returncode == 0, r1.stderr
|
||||
r2 = subprocess.run(
|
||||
[sys.executable, VERIFIER, "verify", "--home", str(home), "--snapshot", str(snap_file)],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
assert r2.returncode == 0, r2.stderr
|
||||
(home / "plugins" / "mnemosyne-wrapper" / "plugin.py").unlink()
|
||||
r3 = subprocess.run(
|
||||
[sys.executable, VERIFIER, "verify", "--home", str(home), "--snapshot", str(snap_file)],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
assert r3.returncode == 1
|
||||
assert "PLUGIN PRESERVATION FAILED" in r3.stderr
|
||||
|
||||
|
||||
def test_release_fixture_seed_is_shared_and_never_repairs_damage(tmp_path):
|
||||
home, external = tmp_path / "home", tmp_path / "external"
|
||||
args = [sys.executable, VERIFIER, "seed", "--home", str(home), "--external", str(external)]
|
||||
|
||||
@@ -1,113 +1,57 @@
|
||||
"""The emitted stamp must preserve build identity and reach the runtime reader."""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from scripts.write_install_stamp import build_stamp
|
||||
|
||||
|
||||
def test_build_stamp_keeps_provenance_separate_from_distribution():
|
||||
stamp = build_stamp(commit="a" * 40, source="ci", distribution="docker", update_mechanism="external")
|
||||
|
||||
assert stamp["source"] == "ci"
|
||||
assert stamp["distribution"] == "docker"
|
||||
|
||||
|
||||
def test_default_build_is_a_bootstrap_artifact_regardless_of_tag(monkeypatch):
|
||||
monkeypatch.delenv("HERMES_DESKTOP_VARIANT", raising=False)
|
||||
monkeypatch.setenv("HERMES_PAYLOAD_TAG", "v9.9.9")
|
||||
|
||||
stamp = build_stamp(commit="a" * 40, update_mechanism="self")
|
||||
|
||||
assert stamp["payload"] == "bootstrap"
|
||||
assert stamp["tag"] is None
|
||||
|
||||
|
||||
def test_explicit_bootstrap_variant_matches_the_default(monkeypatch):
|
||||
monkeypatch.setenv("HERMES_DESKTOP_VARIANT", "bootstrap")
|
||||
monkeypatch.setenv("HERMES_PAYLOAD_TAG", "v9.9.9")
|
||||
|
||||
stamp = build_stamp(commit="a" * 40, update_mechanism="self")
|
||||
|
||||
assert stamp["payload"] == "bootstrap"
|
||||
assert stamp["tag"] is None
|
||||
|
||||
|
||||
def test_bundled_variant_records_payload_and_tag(monkeypatch):
|
||||
monkeypatch.setenv("HERMES_DESKTOP_VARIANT", "bundled")
|
||||
monkeypatch.setenv("HERMES_PAYLOAD_TAG", "v0.18.0")
|
||||
|
||||
stamp = build_stamp(commit="b" * 40, update_mechanism="self")
|
||||
|
||||
assert stamp["payload"] == "bundled"
|
||||
assert stamp["tag"] == "v0.18.0"
|
||||
|
||||
|
||||
def test_light_variant_records_payload_and_tag(monkeypatch):
|
||||
monkeypatch.setenv("HERMES_DESKTOP_VARIANT", "light")
|
||||
monkeypatch.setenv("HERMES_PAYLOAD_TAG", "v0.18.0")
|
||||
|
||||
stamp = build_stamp(commit="b" * 40, update_mechanism="self")
|
||||
|
||||
assert stamp["payload"] == "light"
|
||||
assert stamp["tag"] == "v0.18.0"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("variant", ["bundled", "light"])
|
||||
def test_self_updating_variants_without_tag_stop_the_build(monkeypatch, variant):
|
||||
monkeypatch.setenv("HERMES_DESKTOP_VARIANT", variant)
|
||||
monkeypatch.delenv("HERMES_PAYLOAD_TAG", raising=False)
|
||||
|
||||
with pytest.raises(SystemExit, match="HERMES_PAYLOAD_TAG"):
|
||||
build_stamp(commit="b" * 40, update_mechanism="self")
|
||||
|
||||
|
||||
def test_unknown_variant_stops_the_build(monkeypatch):
|
||||
monkeypatch.setenv("HERMES_DESKTOP_VARIANT", "chonky")
|
||||
monkeypatch.setenv("HERMES_PAYLOAD_TAG", "v0.18.0")
|
||||
|
||||
with pytest.raises(SystemExit, match="unknown HERMES_DESKTOP_VARIANT"):
|
||||
build_stamp(commit="b" * 40, update_mechanism="self")
|
||||
|
||||
|
||||
def test_desktop_app_is_a_valid_distribution():
|
||||
stamp = build_stamp(commit="c" * 40, source="ci", distribution="desktop-app", update_mechanism="electron-updater")
|
||||
|
||||
assert stamp["distribution"] == "desktop-app"
|
||||
|
||||
|
||||
def test_distribution_defaults_to_null():
|
||||
stamp = build_stamp(commit="c" * 40, update_mechanism="self")
|
||||
|
||||
assert stamp["distribution"] is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mechanism", ["electron-updater", "app-installer", "external"])
|
||||
def test_cli_stamp_is_accepted_by_runtime_readers(tmp_path, monkeypatch, mechanism):
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
out = tmp_path / "stamp.json"
|
||||
repo_root = Path(__file__).resolve().parents[2]
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
str(repo_root / "scripts" / "write_install_stamp.py"),
|
||||
"--output", str(out),
|
||||
"--commit", "d" * 40,
|
||||
"--distribution", "desktop-app",
|
||||
"--update-mechanism", mechanism,
|
||||
],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("variant,distribution,mechanism,payload,tag", [
|
||||
("", None, "self", "bootstrap", None),
|
||||
("bootstrap", "docker", "external", "bootstrap", None),
|
||||
("bundled", "desktop-app", "electron-updater", "bundled", "v0.18.0"),
|
||||
("bundled", "desktop-app", "app-installer", "bundled", "v0.18.0"),
|
||||
("bundled", "desktop-app", "external", "bundled", "v0.18.0"),
|
||||
("light", "desktop-app", "electron-updater", "light", "v0.18.0"),
|
||||
])
|
||||
def test_cli_stamp_roundtrip(tmp_path, monkeypatch, variant, distribution, mechanism, payload, tag):
|
||||
out = tmp_path / "install-stamp.json"
|
||||
script = Path(__file__).resolve().parents[2] / "scripts/write_install_stamp.py"
|
||||
env = {**os.environ, "HERMES_DESKTOP_VARIANT": variant, "HERMES_PAYLOAD_TAG": "v0.18.0", "HERMES_BUILD_COMMIT": ""}
|
||||
args = [sys.executable, str(script), "--output", str(out), "--commit", "d" * 40,
|
||||
"--source", "ci", "--update-mechanism", mechanism]
|
||||
if distribution:
|
||||
args += ["--distribution", distribution]
|
||||
result = subprocess.run(args, env=env, capture_output=True, text=True, timeout=30)
|
||||
assert result.returncode == 0, result.stderr
|
||||
data = json.loads(out.read_text(encoding="utf-8-sig"))
|
||||
assert data["distribution"] == "desktop-app"
|
||||
assert data["updateMechanism"] == mechanism
|
||||
assert {key: data[key] for key in ("source", "distribution", "updateMechanism", "payload", "tag", "commit")} == {
|
||||
"source": "ci", "distribution": distribution, "updateMechanism": mechanism,
|
||||
"payload": payload, "tag": tag, "commit": "d" * 40}
|
||||
from hermes_cli.version_info import _stamp_version_info
|
||||
from hermes_cli.venv_sync import _is_sealed
|
||||
|
||||
out.rename(tmp_path / "install-stamp.json")
|
||||
monkeypatch.setenv("HERMES_INSTALL_ROOT", str(tmp_path))
|
||||
assert _stamp_version_info() is not None
|
||||
assert _is_sealed(tmp_path)
|
||||
if payload == "light":
|
||||
with pytest.raises(RuntimeError, match="light"):
|
||||
_stamp_version_info()
|
||||
else:
|
||||
info = _stamp_version_info()
|
||||
assert info is not None and info.commit == "d" * 40
|
||||
assert _is_sealed(tmp_path)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("variant,tag,error", [
|
||||
("bundled", "", "HERMES_PAYLOAD_TAG"), ("light", "", "HERMES_PAYLOAD_TAG"),
|
||||
("chonky", "v0.18.0", "unknown HERMES_DESKTOP_VARIANT"),
|
||||
])
|
||||
def test_invalid_variant_cannot_emit_stamp(tmp_path, variant, tag, error):
|
||||
out = tmp_path / "install-stamp.json"
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(Path(__file__).resolve().parents[2] / "scripts/write_install_stamp.py"),
|
||||
"--output", str(out), "--commit", "d" * 40, "--update-mechanism", "self"],
|
||||
env={**os.environ, "HERMES_DESKTOP_VARIANT": variant, "HERMES_PAYLOAD_TAG": tag, "HERMES_BUILD_COMMIT": ""},
|
||||
capture_output=True, text=True, timeout=30)
|
||||
assert result.returncode != 0 and error in result.stderr
|
||||
assert not out.exists()
|
||||
|
||||
Reference in New Issue
Block a user