feat: preflight as a release asset - download, verify, unpack, then run the tree preflight (#151, C)
Files changed: - .gitea/workflows/release.yml - CHANGES.md - INSTALL.md - README.md - VERSION - instructions/dev/testing-conventions.md - instructions/preflight.md - tools/CONTRACT.md - tools/README.md - tools/chemenu/tests/test_preflight.py - tools/chemenu/tests/test_preflight_pwsh.py - tools/preflight.ps1 - tools/preflight.sh
This commit is contained in:
torben committed
1 parent
210e0c8286
commit
c33e8cdfb1
13 files changed
+1020
-88
No files matched your search
@@ -11,12 +11,14 @@ preflight for real.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -28,7 +30,7 @@ TOOLS = config._PACKAGE_ROOT / "tools"
|
||||
# What preflight.sh, the launcher and trace-hook call besides shell built-ins
|
||||
# and the tools under test. `iconv` is optional in the script itself.
|
||||
UTILITIES = ("dirname", "uname", "sed", "tr", "wc", "iconv", "tail", "cat", "mv", "head",
|
||||
"cut", "grep", "mkdir", "cp", "rm")
|
||||
"cut", "grep", "mkdir", "cp", "rm", "tar", "gzip", "sha256sum", "mktemp")
|
||||
|
||||
|
||||
def _shells() -> list[str]:
|
||||
@@ -93,6 +95,7 @@ class Machine:
|
||||
self.base = base
|
||||
self.root = base / root_name
|
||||
self.tools = self.root / "tools"
|
||||
self.script = self.tools / "preflight.sh"
|
||||
self.tools.mkdir(parents=True)
|
||||
for name in ("preflight.sh", "preflight.ps1", "prerequisites.txt", "wikitool", "wikitool.ps1",
|
||||
"run_wikitool.py", "trace-hook"):
|
||||
@@ -108,6 +111,7 @@ class Machine:
|
||||
self.pip_log = base / "pip.log"
|
||||
self.extra_env: dict[str, str] = {}
|
||||
self.path_dirs: list[Path] = [self.stubs, self.sysbin]
|
||||
self.cwd: Path | None = None
|
||||
|
||||
def stub(self, name: str, text: str, where: Path | None = None) -> Path:
|
||||
return _executable((where or self.stubs) / name, text)
|
||||
@@ -132,8 +136,8 @@ class Machine:
|
||||
**self.extra_env,
|
||||
}
|
||||
return subprocess.run(
|
||||
[shell, str(self.tools / "preflight.sh"), *args],
|
||||
capture_output=True, text=True, env=env, timeout=60,
|
||||
[shell, str(self.script), *args],
|
||||
capture_output=True, text=True, env=env, timeout=60, cwd=self.cwd,
|
||||
)
|
||||
|
||||
@property
|
||||
@@ -354,6 +358,294 @@ def test_manifest_names_the_tools_the_stack_starts():
|
||||
assert [t.name for t in manifest.tools_for("windows")] == ["python", "git", "rg", "pwsh"]
|
||||
|
||||
|
||||
# --- asset mode (D33, D38-D41) ------------------------------------------------------
|
||||
|
||||
RELEASE_VERSION = "9.9.9"
|
||||
DOWNLOAD_BASE = f"https://example.test/torben/chemenu/releases/download/v{RELEASE_VERSION}"
|
||||
STACK_FILES = ("preflight.sh", "preflight.ps1", "prerequisites.txt", "wikitool", "wikitool.ps1",
|
||||
"run_wikitool.py", "trace-hook")
|
||||
|
||||
# The two lines each script keeps for the release build to fill: (empty, filled).
|
||||
PLACEHOLDERS = {
|
||||
"preflight.sh": (
|
||||
("RELEASE_ARCHIVE_URL=''", f"RELEASE_ARCHIVE_URL='{DOWNLOAD_BASE}/chemenu-stack-{RELEASE_VERSION}.tar.gz'"),
|
||||
("RELEASE_CHECKSUM_URL=''", f"RELEASE_CHECKSUM_URL='{DOWNLOAD_BASE}/chemenu-stack-{RELEASE_VERSION}.tar.gz.sha256'"),
|
||||
),
|
||||
"preflight.ps1": (
|
||||
("$ReleaseArchiveUrl = ''", f"$ReleaseArchiveUrl = '{DOWNLOAD_BASE}/chemenu-stack-{RELEASE_VERSION}.tar.gz'"),
|
||||
("$ReleaseChecksumUrl = ''", f"$ReleaseChecksumUrl = '{DOWNLOAD_BASE}/chemenu-stack-{RELEASE_VERSION}.tar.gz.sha256'"),
|
||||
),
|
||||
}
|
||||
|
||||
CURL_STUB = """#!/bin/sh
|
||||
# Serves the files of $RELEASE_DIR by the last part of the URL, and logs every request.
|
||||
while [ $# -gt 0 ]; do
|
||||
case "$1" in
|
||||
-o) dest=$2; shift ;;
|
||||
-*) ;;
|
||||
*) url=$1 ;;
|
||||
esac
|
||||
shift
|
||||
done
|
||||
echo "$url" >> "$CURL_LOG"
|
||||
src="$RELEASE_DIR/${url##*/}"
|
||||
[ -f "$src" ] || exit 22
|
||||
cp "$src" "$dest"
|
||||
"""
|
||||
|
||||
|
||||
def build_release(base: Path, *, limit: int | None = None, tops: tuple[str, ...] | None = None) -> Path:
|
||||
"""A release tarball and its .sha256 as CI builds them; returns the tarball."""
|
||||
name = f"chemenu-stack-{RELEASE_VERSION}"
|
||||
stage = base / "stage"
|
||||
for top in tops or (name,):
|
||||
(stage / top / "tools").mkdir(parents=True)
|
||||
for file in STACK_FILES:
|
||||
shutil.copy2(TOOLS / file, stage / top / "tools" / file)
|
||||
(stage / top / "tools" / "requirements.txt").write_text("PyYAML\n", encoding="utf-8")
|
||||
if limit is not None:
|
||||
manifest = stage / top / "tools" / "prerequisites.txt"
|
||||
manifest.write_text(
|
||||
manifest.read_text(encoding="utf-8").replace(
|
||||
"limit|install_dir_max|95", f"limit|install_dir_max|{limit}"),
|
||||
encoding="utf-8")
|
||||
release = base / "release"
|
||||
release.mkdir()
|
||||
tarball = release / f"{name}.tar.gz"
|
||||
with tarfile.open(tarball, "w:gz") as archive:
|
||||
for top in tops or (name,):
|
||||
archive.add(stage / top, arcname=top)
|
||||
digest = hashlib.sha256(tarball.read_bytes()).hexdigest()
|
||||
(release / f"{tarball.name}.sha256").write_text(f"{digest} {tarball.name}\n", encoding="utf-8")
|
||||
return tarball
|
||||
|
||||
|
||||
class AssetMachine(Machine):
|
||||
"""The release asset on a machine with nothing unpacked yet: just the script, alone in a folder."""
|
||||
|
||||
def __init__(self, base: Path, *, filled: bool = False, **release):
|
||||
super().__init__(base)
|
||||
self.tarball = build_release(base, **release)
|
||||
self.download = base / "download"
|
||||
self.download.mkdir()
|
||||
for name, pairs in PLACEHOLDERS.items():
|
||||
text = (TOOLS / name).read_text(encoding="utf-8")
|
||||
for empty, full in pairs:
|
||||
assert empty in text, f"{name} lost its placeholder {empty}"
|
||||
if filled:
|
||||
text = text.replace(empty, full)
|
||||
(self.download / name).write_text(text, encoding="utf-8")
|
||||
self.script = self.download / "preflight.sh"
|
||||
self.root = self.download / "chemenu"
|
||||
self.tools = self.root / "tools"
|
||||
self.cwd = self.download
|
||||
self.extra_env.update({"RELEASE_DIR": str(self.tarball.parent), "CURL_LOG": str(base / "curl.log")})
|
||||
self.standard()
|
||||
|
||||
def unpacked(self, root: Path | None = None) -> bool:
|
||||
return (root or self.root).exists()
|
||||
|
||||
def leftovers(self) -> list[str]:
|
||||
found = [str(path) for path in self.base.rglob(".chemenu-unpack.*")]
|
||||
found += [str(path) for path in Path(os.environ.get("TMPDIR", "/tmp")).glob("chemenu-preflight.*")]
|
||||
return found
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def asset(tmp_path: Path) -> AssetMachine:
|
||||
return AssetMachine(tmp_path.resolve())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_asset_unpacks_next_to_itself_and_runs_the_tree_copy(asset, shell):
|
||||
result = asset.run("--archive", str(asset.tarball), shell=shell)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
assert "sha256 OK" in result.stdout and "Preflight passed" in result.stdout
|
||||
assert (asset.tools / "preflight.sh").is_file() and (asset.tools / "prerequisites.txt").is_file()
|
||||
assert asset.recorded()["complete"] is True
|
||||
assert (asset.tools / ".venv").is_dir()
|
||||
assert not asset.leftovers()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_into_chooses_the_target_and_creates_missing_parents(asset, shell):
|
||||
target = asset.base / "deep" / "er" / "wiki"
|
||||
result = asset.run("--archive", str(asset.tarball), "--into", str(target), shell=shell)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
assert (target / "tools" / "preflight.sh").is_file()
|
||||
assert (target / ".wikitool-tools.json").is_file()
|
||||
assert not asset.unpacked()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_a_relative_into_resolves_against_the_working_directory(asset, shell):
|
||||
asset.cwd = asset.base
|
||||
result = asset.run("--archive", str(asset.tarball), "--into", "here", shell=shell)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
assert (asset.base / "here" / "tools" / "preflight.sh").is_file()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_an_existing_target_is_refused_and_left_alone(asset, shell):
|
||||
asset.root.mkdir()
|
||||
(asset.root / "mine.txt").write_text("keep", encoding="utf-8")
|
||||
result = asset.run("--archive", str(asset.tarball), shell=shell)
|
||||
assert result.returncode == 1
|
||||
assert "already exists" in result.stderr and "--into" in result.stderr
|
||||
assert [path.name for path in asset.root.iterdir()] == ["mine.txt"]
|
||||
assert not asset.leftovers()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_a_wrong_sha256_exits_1_and_unpacks_nothing(asset, shell):
|
||||
checksum = asset.tarball.parent / f"{asset.tarball.name}.sha256"
|
||||
checksum.write_text("0" * 64 + f" {asset.tarball.name}\n", encoding="utf-8")
|
||||
result = asset.run("--archive", str(asset.tarball), shell=shell)
|
||||
assert result.returncode == 1
|
||||
assert "does not match" in result.stderr and "nothing was unpacked" in result.stderr
|
||||
assert not asset.unpacked() and not asset.leftovers()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_archive_without_a_checksum_file_exits_1(asset, shell):
|
||||
(asset.tarball.parent / f"{asset.tarball.name}.sha256").unlink()
|
||||
result = asset.run("--archive", str(asset.tarball), shell=shell)
|
||||
assert result.returncode == 1
|
||||
assert ".sha256 is missing" in result.stderr
|
||||
assert not asset.unpacked()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_an_archive_with_two_top_level_folders_exits_1(tmp_path, shell):
|
||||
asset = AssetMachine(tmp_path.resolve(), tops=(f"chemenu-stack-{RELEASE_VERSION}", "stray"))
|
||||
result = asset.run("--archive", str(asset.tarball), shell=shell)
|
||||
assert result.returncode == 1
|
||||
assert "exactly one top-level folder" in result.stderr
|
||||
assert not asset.unpacked() and not asset.leftovers()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_a_copy_without_download_addresses_says_it_is_no_release_asset(asset, shell):
|
||||
result = asset.run(shell=shell)
|
||||
assert result.returncode == 1
|
||||
assert "does not come from a release" in result.stderr
|
||||
assert not asset.unpacked()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_download_fetches_both_files_of_the_same_release(tmp_path, shell):
|
||||
asset = AssetMachine(tmp_path.resolve(), filled=True)
|
||||
asset.stub("curl", CURL_STUB)
|
||||
result = asset.run(shell=shell)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
requested = (asset.base / "curl.log").read_text(encoding="utf-8").split()
|
||||
assert requested == [f"{DOWNLOAD_BASE}/{asset.tarball.name}", f"{DOWNLOAD_BASE}/{asset.tarball.name}.sha256"]
|
||||
assert (asset.tools / "preflight.sh").is_file()
|
||||
assert not asset.leftovers()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_a_failed_download_exits_1_and_unpacks_nothing(tmp_path, shell):
|
||||
asset = AssetMachine(tmp_path.resolve(), filled=True)
|
||||
asset.stub("curl", CURL_STUB)
|
||||
asset.extra_env["RELEASE_DIR"] = str(asset.base / "nothing-here")
|
||||
result = asset.run(shell=shell)
|
||||
assert result.returncode == 1
|
||||
assert "the download failed" in result.stderr
|
||||
assert not asset.unpacked() and not asset.leftovers()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
@pytest.mark.parametrize("missing", ["curl", "tar", "sha256sum"])
|
||||
def test_a_missing_download_tool_is_a_stop_with_guidance(tmp_path, shell, missing):
|
||||
asset = AssetMachine(tmp_path.resolve(), filled=True)
|
||||
asset.stub("curl", CURL_STUB)
|
||||
(asset.stubs / "curl").unlink() if missing == "curl" else (asset.sysbin / missing).unlink()
|
||||
result = asset.run(shell=shell)
|
||||
assert result.returncode == 42
|
||||
assert_guidance(result.stdout, f"could not be found: {missing}")
|
||||
assert not asset.unpacked() and not (asset.base / "curl.log").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_with_an_archive_no_download_tool_is_needed(asset, shell):
|
||||
assert not (asset.stubs / "curl").exists() and not (asset.sysbin / "curl").exists()
|
||||
assert asset.run("--archive", str(asset.tarball), shell=shell).returncode == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
def test_set_and_the_exit_code_pass_through_to_the_tree_copy(asset, shell):
|
||||
refused = asset.run("--archive", str(asset.tarball), "--set", f"rg={asset.base / 'nowhere' / 'rg'}", shell=shell)
|
||||
assert refused.returncode == 42
|
||||
assert_guidance(refused.stdout, "The path given for ripgrep (rg) does not work")
|
||||
assert (asset.tools / "preflight.sh").is_file() and not asset.tools_file.exists()
|
||||
|
||||
elsewhere = asset.stub("rg", "#!/bin/sh\necho 'ripgrep 14.1.1'\n", asset.base / "opt")
|
||||
second = AssetMachine(asset.base / "second")
|
||||
second.stub("rg", "#!/bin/sh\necho 'ripgrep 14.1.1'\n", asset.base / "opt")
|
||||
ok = second.run("--archive", str(second.tarball), "--set", f"rg={elsewhere}", shell=shell)
|
||||
assert ok.returncode == 0, ok.stdout
|
||||
assert second.recorded()["tools"]["rg"] == str(elsewhere)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
@pytest.mark.parametrize("flag", ["--into", "--archive"])
|
||||
def test_asset_options_are_a_usage_error_inside_a_tree(machine, shell, flag):
|
||||
machine.standard()
|
||||
result = machine.run(flag, str(machine.base), shell=shell)
|
||||
assert result.returncode == 1
|
||||
assert "belong to the release asset" in result.stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shell", SHELLS)
|
||||
@pytest.mark.parametrize("length, longpaths, limit, expected", [
|
||||
(95, "0", None, 0),
|
||||
(96, "0", None, 42),
|
||||
(96, "1", None, 0),
|
||||
(110, "0", 120, 0),
|
||||
(121, "0", 120, 42),
|
||||
])
|
||||
def test_the_folder_limit_is_judged_at_the_final_target_from_the_archive(
|
||||
tmp_path, shell, length, longpaths, limit, expected):
|
||||
base = tmp_path.resolve()
|
||||
name_length = length - len(str(base / "download")) - 1
|
||||
assert name_length > 0, "tmp_path is too long for this test"
|
||||
asset = AssetMachine(base, limit=limit)
|
||||
asset.standard(pwsh=True)
|
||||
asset.extra_env.update({"CHEMENU_PREFLIGHT_PLATFORM": "windows", "CHEMENU_PREFLIGHT_LONGPATHS": longpaths})
|
||||
target = asset.download / ("t" * name_length)
|
||||
assert len(str(target)) == length
|
||||
result = asset.run("--archive", str(asset.tarball), "--into", str(target), shell=shell)
|
||||
assert result.returncode == expected, result.stdout + result.stderr
|
||||
if expected == 42:
|
||||
assert_guidance(result.stdout, f"too long ({length} characters, at most {limit or 95})")
|
||||
assert not asset.unpacked(target) and not asset.leftovers()
|
||||
else:
|
||||
assert (target / "tools" / "preflight.sh").is_file()
|
||||
|
||||
|
||||
def test_the_release_workflow_fills_exactly_the_placeholders_the_scripts_keep():
|
||||
workflow = Path(__file__).resolve().parents[3] / ".gitea" / "workflows" / "release.yml"
|
||||
if not workflow.is_file():
|
||||
pytest.skip("a distributed instance carries no release workflow")
|
||||
text = workflow.read_text(encoding="utf-8")
|
||||
|
||||
def as_written_in_the_workflow(line: str) -> str:
|
||||
return (line.replace("$Release", "\\$Release")
|
||||
.replace(DOWNLOAD_BASE, "${download}")
|
||||
.replace(f"chemenu-stack-{RELEASE_VERSION}", "${name}"))
|
||||
|
||||
for script, pairs in PLACEHOLDERS.items():
|
||||
source = (TOOLS / script).read_text(encoding="utf-8")
|
||||
assert f'tools/{script} > "${{BUILD_DIR}}/{script}"' in text
|
||||
for empty, filled in pairs:
|
||||
assert source.count(empty) == 1, f"{script} must hold {empty} exactly once"
|
||||
expression = f"s|^{as_written_in_the_workflow(empty)}|{as_written_in_the_workflow(filled)}|"
|
||||
assert expression in text, f"release.yml does not run {expression}"
|
||||
assert 'download="${PUBLIC_BASE_URL}/${GITHUB_REPOSITORY}/releases/download/${TAG}"' in text
|
||||
assert 'for asset in "${NAME}.tar.gz" "${NAME}.tar.gz.sha256" preflight.sh preflight.ps1; do' in text
|
||||
|
||||
|
||||
# --- the launcher -----------------------------------------------------------------
|
||||
|
||||
|
||||
|
||||
@@ -13,17 +13,21 @@ compares them byte for byte.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import http.server
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from chemenu import prerequisites
|
||||
from chemenu.tests.test_preflight import (
|
||||
ECHO_PYTHON, SHELLS, TOOLS, Machine, _machine_with_root_of, assert_guidance,
|
||||
DOWNLOAD_BASE, ECHO_PYTHON, SHELLS, TOOLS, AssetMachine, Machine, _machine_with_root_of,
|
||||
assert_guidance,
|
||||
)
|
||||
|
||||
PWSH = shutil.which("pwsh")
|
||||
@@ -195,6 +199,172 @@ def test_install_folder_limit_at_the_boundary(tmp_path, length, longpaths, expec
|
||||
assert not (machine.tools / ".venv").exists()
|
||||
|
||||
|
||||
# --- asset mode (D33, D38-D41) ------------------------------------------------------
|
||||
#
|
||||
# The sh tests in test_preflight.py cover the same cases against preflight.sh; what is
|
||||
# here is the twin's own wiring - Get-FileHash, tar.exe, Invoke-WebRequest - and the
|
||||
# hand-over to the tree copy in a child pwsh.
|
||||
|
||||
|
||||
def _asset_run(asset: AssetMachine, *args: str) -> subprocess.CompletedProcess:
|
||||
env = {
|
||||
"PATH": os.pathsep.join(str(d) for d in asset.path_dirs),
|
||||
"HOME": str(asset.base),
|
||||
"PIP_LOG": str(asset.pip_log),
|
||||
"DOTNET_CLI_TELEMETRY_OPTOUT": "1",
|
||||
"POWERSHELL_TELEMETRY_OPTOUT": "1",
|
||||
**asset.extra_env,
|
||||
}
|
||||
return subprocess.run(
|
||||
[PWSH, "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", str(asset.download / "preflight.ps1"), *args],
|
||||
capture_output=True, text=True, env=env, timeout=120, cwd=asset.cwd,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def asset(tmp_path: Path) -> AssetMachine:
|
||||
return AssetMachine(tmp_path.resolve())
|
||||
|
||||
|
||||
def test_asset_unpacks_next_to_itself_and_runs_the_tree_copy(asset):
|
||||
result = _asset_run(asset, "--archive", str(asset.tarball))
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
assert "sha256 OK" in result.stdout and "Preflight passed" in result.stdout
|
||||
assert (asset.tools / "preflight.ps1").is_file()
|
||||
assert asset.recorded()["complete"] is True
|
||||
assert not asset.leftovers()
|
||||
|
||||
|
||||
def test_into_chooses_the_target_and_a_relative_one_follows_the_working_directory(asset):
|
||||
far = asset.base / "deep" / "er" / "wiki"
|
||||
assert _asset_run(asset, "--archive", str(asset.tarball), "--into", str(far)).returncode == 0
|
||||
assert (far / "tools" / "preflight.ps1").is_file() and not asset.unpacked()
|
||||
asset.cwd = asset.base
|
||||
assert _asset_run(asset, "--archive", str(asset.tarball), "--into", "here").returncode == 0
|
||||
assert (asset.base / "here" / "tools" / "preflight.ps1").is_file()
|
||||
|
||||
|
||||
def test_an_existing_target_is_refused_and_left_alone(asset):
|
||||
asset.root.mkdir()
|
||||
(asset.root / "mine.txt").write_text("keep", encoding="utf-8")
|
||||
result = _asset_run(asset, "--archive", str(asset.tarball))
|
||||
assert result.returncode == 1
|
||||
assert "already exists" in result.stderr and "--into" in result.stderr
|
||||
assert [path.name for path in asset.root.iterdir()] == ["mine.txt"]
|
||||
|
||||
|
||||
def test_a_wrong_sha256_exits_1_and_unpacks_nothing(asset):
|
||||
checksum = asset.tarball.parent / f"{asset.tarball.name}.sha256"
|
||||
checksum.write_text("0" * 64 + f" {asset.tarball.name}\n", encoding="utf-8")
|
||||
result = _asset_run(asset, "--archive", str(asset.tarball))
|
||||
assert result.returncode == 1
|
||||
assert "does not match" in result.stderr and "nothing was unpacked" in result.stderr
|
||||
assert not asset.unpacked() and not asset.leftovers()
|
||||
|
||||
|
||||
def test_archive_without_a_checksum_file_exits_1(asset):
|
||||
(asset.tarball.parent / f"{asset.tarball.name}.sha256").unlink()
|
||||
result = _asset_run(asset, "--archive", str(asset.tarball))
|
||||
assert result.returncode == 1 and ".sha256 is missing" in result.stderr
|
||||
assert not asset.unpacked()
|
||||
|
||||
|
||||
def test_an_archive_with_two_top_level_folders_exits_1(tmp_path):
|
||||
asset = AssetMachine(tmp_path.resolve(), tops=("chemenu-stack-9.9.9", "stray"))
|
||||
result = _asset_run(asset, "--archive", str(asset.tarball))
|
||||
assert result.returncode == 1 and "exactly one top-level folder" in result.stderr
|
||||
assert not asset.unpacked() and not asset.leftovers()
|
||||
|
||||
|
||||
def test_a_copy_without_download_addresses_says_it_is_no_release_asset(asset):
|
||||
result = _asset_run(asset)
|
||||
assert result.returncode == 1 and "does not come from a release" in result.stderr
|
||||
assert not asset.unpacked()
|
||||
|
||||
|
||||
def test_a_missing_tar_is_a_stop_with_guidance(asset):
|
||||
(asset.sysbin / "tar").unlink()
|
||||
result = _asset_run(asset, "--archive", str(asset.tarball))
|
||||
assert result.returncode == 42
|
||||
assert_guidance(result.stdout, "(tar) could not be found")
|
||||
assert not asset.unpacked()
|
||||
|
||||
|
||||
def test_set_and_the_exit_code_pass_through_to_the_tree_copy(asset):
|
||||
refused = _asset_run(asset, "--archive", str(asset.tarball), "--set", f"rg={asset.base / 'nowhere' / 'rg'}")
|
||||
assert refused.returncode == 42
|
||||
assert_guidance(refused.stdout, "The path given for ripgrep (rg) does not work")
|
||||
assert (asset.tools / "preflight.ps1").is_file() and not asset.tools_file.exists()
|
||||
|
||||
second = AssetMachine(asset.base / "second")
|
||||
elsewhere = second.stub("rg", "#!/bin/sh\necho 'ripgrep 14.1.1'\n", asset.base / "opt")
|
||||
ok = _asset_run(second, "--archive", str(second.tarball), "--set", f"rg={elsewhere}")
|
||||
assert ok.returncode == 0, ok.stdout
|
||||
assert second.recorded()["tools"]["rg"] == str(elsewhere)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("flag", ["--into", "--archive"])
|
||||
def test_asset_options_are_a_usage_error_inside_a_tree(machine, flag):
|
||||
result = _preflight(machine, flag, str(machine.base))
|
||||
assert result.returncode == 1 and "belong to the release asset" in result.stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize("length, longpaths, limit, expected", [
|
||||
(95, "0", None, 0),
|
||||
(96, "0", None, 42),
|
||||
(96, "1", None, 0),
|
||||
(110, "0", 120, 0),
|
||||
(121, "0", 120, 42),
|
||||
])
|
||||
def test_the_folder_limit_is_judged_at_the_final_target_from_the_archive(
|
||||
tmp_path, length, longpaths, limit, expected):
|
||||
base = tmp_path.resolve()
|
||||
name_length = length - len(str(base / "download")) - 1
|
||||
assert name_length > 0, "tmp_path is too long for this test"
|
||||
asset = AssetMachine(base, limit=limit)
|
||||
asset.standard(pwsh=True)
|
||||
asset.extra_env.update({"CHEMENU_PREFLIGHT_PLATFORM": "windows", "CHEMENU_PREFLIGHT_LONGPATHS": longpaths})
|
||||
target = asset.download / ("t" * name_length)
|
||||
result = _asset_run(asset, "--archive", str(asset.tarball), "--into", str(target))
|
||||
assert result.returncode == expected, result.stdout + result.stderr
|
||||
if expected == 42:
|
||||
assert_guidance(result.stdout, f"too long ({length} characters, at most {limit or 95})")
|
||||
assert not asset.unpacked(target) and not asset.leftovers()
|
||||
else:
|
||||
assert (target / "tools" / "preflight.ps1").is_file()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def release_server(asset):
|
||||
handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory=str(asset.tarball.parent))
|
||||
handler.log_message = lambda *args: None
|
||||
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
||||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||||
script = asset.download / "preflight.ps1"
|
||||
script.write_text(
|
||||
(TOOLS / "preflight.ps1").read_text(encoding="utf-8").replace(
|
||||
"$ReleaseArchiveUrl = \'\'", f"$ReleaseArchiveUrl = \'http://127.0.0.1:{server.server_port}/{asset.tarball.name}\'").replace(
|
||||
"$ReleaseChecksumUrl = \'\'", f"$ReleaseChecksumUrl = \'http://127.0.0.1:{server.server_port}/{asset.tarball.name}.sha256\'"),
|
||||
encoding="utf-8")
|
||||
yield asset
|
||||
server.shutdown()
|
||||
|
||||
|
||||
def test_download_fetches_both_files_and_unpacks(release_server):
|
||||
result = _asset_run(release_server)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
assert "Downloading http://127.0.0.1" in result.stdout and "sha256 OK" in result.stdout
|
||||
assert (release_server.tools / "preflight.ps1").is_file()
|
||||
assert not release_server.leftovers()
|
||||
|
||||
|
||||
def test_a_failed_download_exits_1_and_unpacks_nothing(release_server):
|
||||
(release_server.tarball.parent / f"{release_server.tarball.name}.sha256").unlink()
|
||||
result = _asset_run(release_server)
|
||||
assert result.returncode == 1 and "the download failed" in result.stderr
|
||||
assert not release_server.unpacked() and not release_server.leftovers()
|
||||
|
||||
|
||||
# --- execution policy and Mark of the Web (D16, T6) --------------------------------
|
||||
|
||||
OPEN = "MachinePolicy=Undefined,UserPolicy=Undefined,Process=Undefined,CurrentUser=Undefined,LocalMachine=RemoteSigned"
|
||||
|
||||
Reference in new issue
Block a user