Merge pull request #967 from infinito-nexus/fix/workspace/distro-ca-and-minitest

fix(workspace): CA anchors, minitest and mmdc's browser on non-Debian distros
This commit is contained in:
kevinveenbirkenbach authored and GitHub committed 2026-10-15 23:51:10 +02:00
commit fea6f31d9f
13 files changed
+712 -117

No files matched your search

+1 -1
View File
@@ -893,7 +893,7 @@ test-unit-python: install
bash scripts/tests/code/wrapper.sh scripts/tests/code/run.sh
.PHONY: test-unit-ruby
# Run the Ruby unit test suite (minitest, stdlib).
# Run the Ruby unit test suite (minitest).
test-unit-ruby:
@INFINITO_TEST_TYPE="unit-ruby" \
bash scripts/tests/code/wrapper.sh scripts/tests/unit/ruby.sh
+21 -7
View File
@@ -1,15 +1,22 @@
#!/usr/bin/env bash
# Install the frontend CA into the runner's trust store. Idempotent.
# See docs/contributing/environment/cache.md.
# Param: CA_ROOT prefix for every absolute path below; empty in production.
set -eu
CA_SRC="/opt/package-frontend-ca.crt"
CA_ROOT="${CA_ROOT:-}"
CA_SRC="${CA_ROOT}/opt/package-frontend-ca.crt"
CA_NAME="infinito-package-frontend.crt"
if command -v update-ca-certificates >/dev/null 2>&1; then
CA_DST="/usr/local/share/ca-certificates/infinito-package-frontend.crt"
CA_DSTS=("${CA_ROOT}/usr/local/share/ca-certificates/${CA_NAME}")
CA_REBUILD=(update-ca-certificates)
elif command -v update-ca-trust >/dev/null 2>&1; then
CA_DST="/etc/ca-certificates/trust-source/anchors/infinito-package-frontend.crt"
# Exception: Arch and the RedHat family both run update-ca-trust but read different anchor dirs, and with-ca-trust.sh creates both dirs on both families, so the anchor goes into each instead of a probed one.
CA_DSTS=(
"${CA_ROOT}/etc/pki/ca-trust/source/anchors/${CA_NAME}"
"${CA_ROOT}/etc/ca-certificates/trust-source/anchors/${CA_NAME}"
)
CA_REBUILD=(update-ca-trust extract)
else
echo "[package-frontend-ca] no supported CA trust tool" >&2
@@ -21,11 +28,18 @@ if [ ! -s "${CA_SRC}" ]; then
exit 0
fi
if cmp -s "${CA_SRC}" "${CA_DST}" 2>/dev/null; then
changed=0
for dst in "${CA_DSTS[@]}"; do
if ! cmp -s "${CA_SRC}" "${dst}" 2>/dev/null; then
install -d -m 0755 "$(dirname "${dst}")"
install -m 0644 "${CA_SRC}" "${dst}"
changed=1
fi
done
if [ "${changed}" -eq 0 ]; then
exit 0
fi
install -d -m 0755 "$(dirname "${CA_DST}")"
install -m 0644 "${CA_SRC}" "${CA_DST}"
"${CA_REBUILD[@]}" >/dev/null 2>&1
echo "[package-frontend-ca] installed ${CA_DST}" >&2
echo "[package-frontend-ca] installed ${CA_DSTS[*]}" >&2
+22 -16
View File
@@ -19,25 +19,24 @@
# * Debian/Ubuntu: /usr/local/share/ca-certificates (name must end in .crt),
# rebuilt by `update-ca-certificates`.
# * Arch and Fedora/CentOS both rebuild with `update-ca-trust extract` but
# read different anchor dirs -- Arch has no /etc/pki, the RedHat family
# has no /etc/ca-certificates -- so the dir is probed, not assumed.
# read different anchor dirs, and with-ca-trust.sh creates both dirs on
# both families, so the anchor goes into each.
# Param: CA_ROOT prefix for every absolute path below; empty in production.
set -eu
CA_DIR="/opt/registry-cache-ca"
CA_ROOT="${CA_ROOT:-}"
CA_DIR="${CA_ROOT}/opt/registry-cache-ca"
CA_SRC="${CA_DIR}/ca.crt"
CA_NAME="infinito-registry-cache.crt"
if command -v update-ca-certificates >/dev/null 2>&1; then
CA_DST="/usr/local/share/ca-certificates/infinito-registry-cache.crt"
CA_DSTS=("${CA_ROOT}/usr/local/share/ca-certificates/${CA_NAME}")
CA_REBUILD=(update-ca-certificates)
elif command -v update-ca-trust >/dev/null 2>&1; then
if [ -d /etc/pki/ca-trust/source/anchors ]; then
CA_DST="/etc/pki/ca-trust/source/anchors/infinito-registry-cache.crt"
elif [ -d /etc/ca-certificates/trust-source/anchors ]; then
CA_DST="/etc/ca-certificates/trust-source/anchors/infinito-registry-cache.crt"
else
echo "[registry-cache-ca] update-ca-trust present but no known anchor dir (/etc/pki/ca-trust/source/anchors, /etc/ca-certificates/trust-source/anchors)" >&2
exit 1
fi
CA_DSTS=(
"${CA_ROOT}/etc/pki/ca-trust/source/anchors/${CA_NAME}"
"${CA_ROOT}/etc/ca-certificates/trust-source/anchors/${CA_NAME}"
)
CA_REBUILD=(update-ca-trust extract)
else
echo "[registry-cache-ca] no supported CA trust tool (update-ca-certificates or update-ca-trust)" >&2
@@ -56,11 +55,18 @@ if [ ! -s "${CA_SRC}" ]; then
exit 1
fi
if cmp -s "${CA_SRC}" "${CA_DST}" 2>/dev/null; then
changed=0
for dst in "${CA_DSTS[@]}"; do
if ! cmp -s "${CA_SRC}" "${dst}" 2>/dev/null; then
install -d -m 0755 "$(dirname "${dst}")"
install -m 0644 "${CA_SRC}" "${dst}"
changed=1
fi
done
if [ "${changed}" -eq 0 ]; then
exit 0
fi
install -d -m 0755 "$(dirname "${CA_DST}")"
install -m 0644 "${CA_SRC}" "${CA_DST}"
"${CA_REBUILD[@]}" >/dev/null
echo "[registry-cache-ca] installed ${CA_DST}" >&2
echo "[registry-cache-ca] installed ${CA_DSTS[*]}" >&2
+1 -1
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env bash
# shellcheck shell=bash
#
# Runs the Ruby unit suite through minitest, which ships with Ruby itself.
# Runs the Ruby unit suite through minitest, which scripts/install/ruby.sh provisions.
set -euo pipefail
Whitespace-only changes.
Whitespace-only changes.
Whitespace-only changes.
@@ -0,0 +1,115 @@
from __future__ import annotations
import os
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
from utils.cache.files import PROJECT_ROOT
SCRIPT = PROJECT_ROOT / "scripts" / "docker" / "cache" / "package-frontend" / "ca.sh"
NAME = "infinito-package-frontend.crt"
STUB = '#!/usr/bin/env bash\nprintf "%s\\n" "$0 $*" >>"${STUB_LOG}"\n'
PKI = "etc/pki/ca-trust/source/anchors"
ARCH = "etc/ca-certificates/trust-source/anchors"
TOOLS = ("bash", "cmp", "dirname", "install")
class TestPackageFrontendCa(unittest.TestCase):
def fake_root(self, tool: str, anchor_dirs: tuple[str, ...]) -> Path:
"""Build a fake root that offers one trust tool and a frontend CA.
Args:
tool: the trust rebuild command the fake system provides.
anchor_dirs: anchor directories that already exist in the fake root.
Returns:
The fake root.
"""
root = Path(self.enterContext(tempfile.TemporaryDirectory()))
(root / "opt").mkdir()
(root / "opt" / "package-frontend-ca.crt").write_text("CA\n", encoding="utf-8")
for anchor in anchor_dirs:
(root / anchor).mkdir(parents=True)
(root / "bin").mkdir()
for needed in TOOLS:
(root / "bin" / needed).symlink_to(shutil.which(needed))
(root / "bin" / tool).write_text(STUB, encoding="utf-8")
(root / "bin" / tool).chmod(0o755)
return root
def run_script(self, root: Path) -> subprocess.CompletedProcess:
"""Run ca.sh against a fake root.
Args:
root: fake root from ``fake_root``.
Returns:
The finished process.
"""
env = {k: v for k, v in os.environ.items() if k not in ("BASH_ENV", "CA_ROOT")}
env.update(
CA_ROOT=str(root),
PATH=str(root / "bin"),
STUB_LOG=str(root / "stub.log"),
)
return subprocess.run(
[str(root / "bin" / "bash"), str(SCRIPT)],
env=env,
capture_output=True,
text=True,
check=False,
)
def rebuilds(self, root: Path) -> int:
log = root / "stub.log"
if not log.exists():
return 0
text = log.read_text(encoding="utf-8") # nocheck: cache-read per-run tempdir
return len(text.splitlines())
def test_the_redhat_family_gets_the_pki_anchor(self) -> None:
root = self.fake_root("update-ca-trust", (PKI,))
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertTrue((root / PKI / NAME).is_file())
self.assertEqual(self.rebuilds(root), 1)
def test_arch_with_both_anchor_dirs_gets_its_trust_source_anchor(self) -> None:
root = self.fake_root("update-ca-trust", (PKI, ARCH))
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertTrue((root / ARCH / NAME).is_file())
def test_a_fresh_update_ca_trust_root_gets_both_anchors(self) -> None:
root = self.fake_root("update-ca-trust", ())
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertTrue((root / PKI / NAME).is_file())
self.assertTrue((root / ARCH / NAME).is_file())
def test_debian_uses_update_ca_certificates(self) -> None:
root = self.fake_root("update-ca-certificates", ())
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertTrue((root / "usr/local/share/ca-certificates" / NAME).is_file())
self.assertFalse((root / "etc").exists())
def test_a_second_run_does_not_rebuild(self) -> None:
root = self.fake_root("update-ca-trust", (PKI,))
self.run_script(root)
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertEqual(self.rebuilds(root), 1)
def test_no_trust_tool_fails_loudly(self) -> None:
root = self.fake_root("unrelated-tool", ())
done = self.run_script(root)
self.assertNotEqual(done.returncode, 0)
self.assertFalse(any(root.glob(f"**/{NAME}")))
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,119 @@
from __future__ import annotations
import os
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
from utils.cache.files import PROJECT_ROOT
SCRIPT = PROJECT_ROOT / "scripts" / "docker" / "cache" / "registry-ca.sh"
NAME = "infinito-registry-cache.crt"
STUB = '#!/usr/bin/env bash\nprintf "%s\\n" "$0 $*" >>"${STUB_LOG}"\n'
PKI = "etc/pki/ca-trust/source/anchors"
ARCH = "etc/ca-certificates/trust-source/anchors"
TOOLS = ("bash", "cmp", "dirname", "install")
class TestRegistryCacheCa(unittest.TestCase):
def fake_root(
self, tool: str, anchor_dirs: tuple[str, ...], ca: str | None
) -> Path:
"""Build a fake root that offers one trust tool.
Args:
tool: the trust rebuild command the fake system provides.
anchor_dirs: anchor directories that already exist in the fake root.
ca: content of the mounted ``ca.crt``; ``None`` leaves the CA dir unmounted.
Returns:
The fake root.
"""
root = Path(self.enterContext(tempfile.TemporaryDirectory()))
if ca is not None:
(root / "opt" / "registry-cache-ca").mkdir(parents=True)
(root / "opt" / "registry-cache-ca" / "ca.crt").write_text(
ca, encoding="utf-8"
)
for anchor in anchor_dirs:
(root / anchor).mkdir(parents=True)
(root / "bin").mkdir()
for needed in TOOLS:
(root / "bin" / needed).symlink_to(shutil.which(needed))
(root / "bin" / tool).write_text(STUB, encoding="utf-8")
(root / "bin" / tool).chmod(0o755)
return root
def run_script(self, root: Path) -> subprocess.CompletedProcess:
"""Run registry-ca.sh against a fake root.
Args:
root: fake root from ``fake_root``.
Returns:
The finished process.
"""
env = {k: v for k, v in os.environ.items() if k not in ("BASH_ENV", "CA_ROOT")}
env.update(
CA_ROOT=str(root),
PATH=str(root / "bin"),
STUB_LOG=str(root / "stub.log"),
)
return subprocess.run(
[str(root / "bin" / "bash"), str(SCRIPT)],
env=env,
capture_output=True,
text=True,
check=False,
)
def rebuilds(self, root: Path) -> int:
log = root / "stub.log"
if not log.exists():
return 0
text = log.read_text(encoding="utf-8") # nocheck: cache-read per-run tempdir
return len(text.splitlines())
def test_the_redhat_family_gets_the_pki_anchor(self) -> None:
root = self.fake_root("update-ca-trust", (PKI,), "CA\n")
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertTrue((root / PKI / NAME).is_file())
self.assertEqual(self.rebuilds(root), 1)
def test_arch_with_both_anchor_dirs_gets_its_trust_source_anchor(self) -> None:
root = self.fake_root("update-ca-trust", (PKI, ARCH), "CA\n")
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertTrue((root / ARCH / NAME).is_file())
def test_debian_uses_update_ca_certificates(self) -> None:
root = self.fake_root("update-ca-certificates", (), "CA\n")
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertTrue((root / "usr/local/share/ca-certificates" / NAME).is_file())
def test_a_second_run_does_not_rebuild(self) -> None:
root = self.fake_root("update-ca-trust", (ARCH,), "CA\n")
self.run_script(root)
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertEqual(self.rebuilds(root), 1)
def test_an_unmounted_ca_dir_is_skipped(self) -> None:
root = self.fake_root("update-ca-trust", (PKI,), None)
done = self.run_script(root)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertEqual(self.rebuilds(root), 0)
def test_a_mounted_dir_without_ca_is_refused(self) -> None:
root = self.fake_root("update-ca-trust", (PKI,), "")
done = self.run_script(root)
self.assertNotEqual(done.returncode, 0)
self.assertFalse(any(root.glob(f"**/{NAME}")))
if __name__ == "__main__":
unittest.main()
@@ -17,10 +17,17 @@ if TYPE_CHECKING:
from collections.abc import Iterator
REVISION = "155.0.8059.39"
LDD = (
"\tlinux-vdso.so.1 (0x00007f5999895000)\n"
"\tlibnss3.so => not found\n"
"\tlibglib-2.0.so.0 => /usr/lib/libglib-2.0.so.0 (0x00007f598d09d000)\n"
"\tlibgbm.so.1 => not found\n"
"\tlibnss3.so => not found\n"
)
def _done(returncode: int) -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess([], returncode, stdout="out\n", stderr="err\n")
def _done(returncode: int, stdout: str = "out\n") -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess([], returncode, stdout=stdout, stderr="err\n")
def _write_core(node_modules: Path, revision: str) -> None:
@@ -64,10 +71,12 @@ def _fake_mmdc(
return mmdc
def _install_browser(cache: Path, revision: str) -> None:
def _install_browser(cache: Path, revision: str) -> Path:
build = cache / "chrome-headless-shell" / f"linux-{revision}" / "linux64"
build.mkdir(parents=True)
(build / "chrome-headless-shell").write_text("", encoding="utf-8")
build.mkdir(parents=True, exist_ok=True)
executable = build / "chrome-headless-shell"
executable.write_text("", encoding="utf-8")
return executable
@contextlib.contextmanager
@@ -89,45 +98,64 @@ def _environment(
mock.patch.dict(os.environ, {"PUPPETEER_CACHE_DIR": cache}),
mock.patch.object(mermaid_cli, "which", return_value=True),
mock.patch.object(mermaid_cli.shutil, "which", return_value=str(mmdc)),
mock.patch.object(mermaid_cli, "missing_libraries", return_value=[]),
):
yield Path(prefix), Path(cache)
class TestBrowserProvisioning(unittest.TestCase):
def provision(self, *, euid: int, results: list[int]) -> tuple[list, list[str]]:
def provision(
self, *, euid: int, results: list[int], apt: bool = True
) -> list[mock._Call]:
"""Run ``ensure()`` with mmdc present and a stubbed puppeteer.
Args:
euid: effective uid the run pretends to have.
results: exit codes the successive puppeteer calls return.
results: exit codes the successive puppeteer calls return; a 0
lays the pinned browser into the cache.
apt: whether apt-get is on the PATH.
Returns:
The recorded subprocess calls and the warnings emitted.
The recorded subprocess calls.
"""
warnings: list[str] = []
with (
_environment(),
mock.patch.object(mermaid_cli.os, "geteuid", return_value=euid),
mock.patch.object(mermaid_cli, "warn", side_effect=warnings.append),
mock.patch.object(
mermaid_cli.subprocess,
"run",
side_effect=[_done(code) for code in results],
) as run,
):
mermaid_cli.ensure()
return run.call_args_list, warnings
calls: list[mock._Call] = []
codes = iter(results)
with _environment() as (_, cache):
def test_root_installs_the_browser_with_its_libraries(self) -> None:
calls, warnings = self.provision(euid=0, results=[0])
def run(*args, **kwargs):
calls.append(mock.call(*args, **kwargs))
code = next(codes)
if code == 0:
_install_browser(cache, REVISION)
return _done(code)
with (
mock.patch.object(mermaid_cli.os, "geteuid", return_value=euid),
mock.patch.object(
mermaid_cli,
"which",
side_effect=lambda cmd: apt or cmd != "apt-get",
),
mock.patch.object(mermaid_cli.subprocess, "run", side_effect=run),
):
mermaid_cli.ensure()
return calls
def test_root_on_apt_installs_the_browser_with_its_libraries(self) -> None:
calls = self.provision(euid=0, results=[0])
self.assertEqual(len(calls), 1)
self.assertEqual(
calls[0].args[0][-2:], ["chrome-headless-shell", "--install-deps"]
)
self.assertEqual(warnings, [])
def test_root_without_apt_leaves_the_libraries_to_the_soname_install(
self,
) -> None:
calls = self.provision(euid=0, results=[0], apt=False)
self.assertNotIn("--install-deps", calls[0].args[0])
def test_the_browser_comes_from_the_puppeteer_next_to_mmdc(self) -> None:
calls, _ = self.provision(euid=0, results=[0])
calls = self.provision(euid=0, results=[0])
self.assertTrue(
calls[0]
.args[0][0]
@@ -135,12 +163,12 @@ class TestBrowserProvisioning(unittest.TestCase):
)
def test_a_user_installs_the_browser_without_libraries(self) -> None:
calls, _ = self.provision(euid=1000, results=[0])
calls = self.provision(euid=1000, results=[0])
self.assertNotIn("--install-deps", calls[0].args[0])
def test_the_browser_install_downloads_despite_the_mmdc_skip_flag(self) -> None:
with mock.patch.dict(os.environ, {"PUPPETEER_SKIP_DOWNLOAD": "1"}):
calls, _ = self.provision(euid=0, results=[0])
calls = self.provision(euid=0, results=[0])
self.assertNotIn("PUPPETEER_SKIP_DOWNLOAD", calls[0].kwargs["env"])
def test_a_failed_install_clears_its_revision_and_retries_once(self) -> None:
@@ -153,17 +181,18 @@ class TestBrowserProvisioning(unittest.TestCase):
def run(*_args, **_kwargs):
seen.append((partial.exists(), other.exists()))
return _done(1 if len(seen) == 1 else 0)
if len(seen) == 1:
return _done(1)
_install_browser(cache, REVISION)
return _done(0)
with mock.patch.object(mermaid_cli.subprocess, "run", side_effect=run):
mermaid_cli.ensure()
self.assertEqual(seen, [(True, True), (False, True)])
def test_a_second_failure_warns_with_puppeteers_output(self) -> None:
calls, warnings = self.provision(euid=0, results=[1, 1])
self.assertEqual(len(calls), 2)
self.assertEqual(len(warnings), 1)
self.assertIn("out\nerr\n", warnings[0])
def test_a_second_failure_raises_with_puppeteers_output(self) -> None:
with self.assertRaisesRegex(RuntimeError, "out\nerr\n"):
self.provision(euid=0, results=[1, 1])
def test_the_pinned_browser_is_not_provisioned_again(self) -> None:
with _environment() as (_, cache):
@@ -175,23 +204,35 @@ class TestBrowserProvisioning(unittest.TestCase):
def test_a_browser_of_another_revision_is_replaced(self) -> None:
with _environment() as (_, cache):
_install_browser(cache, "155.0.8059.79")
with mock.patch.object(
mermaid_cli.subprocess, "run", return_value=_done(0)
) as run:
mermaid_cli.ensure()
run.assert_called_once()
def test_mmdc_without_its_own_puppeteer_warns(self) -> None:
def run(*_args, **_kwargs):
_install_browser(cache, REVISION)
return _done(0)
with mock.patch.object(
mermaid_cli.subprocess, "run", side_effect=run
) as stub:
mermaid_cli.ensure()
stub.assert_called_once()
def test_a_reported_success_without_the_browser_raises(self) -> None:
with (
_environment(),
mock.patch.object(mermaid_cli.subprocess, "run", return_value=_done(0)),
self.assertRaisesRegex(RuntimeError, "still missing"),
):
mermaid_cli.ensure()
def test_mmdc_without_its_own_puppeteer_raises(self) -> None:
with (
_environment(revision=None),
mock.patch.object(mermaid_cli, "warn") as warn,
mock.patch.object(mermaid_cli.subprocess, "run") as run,
self.assertRaisesRegex(RuntimeError, "no puppeteer next to mmdc"),
):
mermaid_cli.ensure()
run.assert_not_called()
warn.assert_called_once()
def test_a_timeout_warns_instead_of_raising(self) -> None:
def test_a_timeout_propagates(self) -> None:
with (
_environment(),
mock.patch.object(
@@ -199,10 +240,112 @@ class TestBrowserProvisioning(unittest.TestCase):
"run",
side_effect=subprocess.TimeoutExpired("puppeteer", 900),
),
mock.patch.object(mermaid_cli, "warn") as warn,
self.assertRaises(subprocess.TimeoutExpired),
):
mermaid_cli.ensure()
warn.assert_called_once()
class TestMmdcInstall(unittest.TestCase):
def test_a_failed_npm_install_propagates(self) -> None:
with (
mock.patch.object(mermaid_cli, "which", return_value=False),
mock.patch.object(mermaid_cli, "install_command_via_pkg"),
mock.patch.object(
mermaid_cli, "npm_install_global", side_effect=RuntimeError("npm")
),
mock.patch.object(mermaid_cli, "_ensure_browser") as browser,
self.assertRaisesRegex(RuntimeError, "npm"),
):
mermaid_cli.ensure()
browser.assert_not_called()
def test_mmdc_still_missing_after_npm_raises(self) -> None:
with (
mock.patch.object(mermaid_cli, "which", return_value=False),
mock.patch.object(mermaid_cli, "install_command_via_pkg"),
mock.patch.object(mermaid_cli, "npm_install_global"),
mock.patch.object(mermaid_cli, "_ensure_browser") as browser,
self.assertRaisesRegex(RuntimeError, "mmdc"),
):
mermaid_cli.ensure()
browser.assert_not_called()
class TestBrowserLibraries(unittest.TestCase):
def test_ldd_not_found_lines_become_sorted_sonames(self) -> None:
with mock.patch.object(
mermaid_cli.subprocess, "run", return_value=_done(0, LDD)
) as run:
missing = mermaid_cli.missing_libraries(Path("/opt/chs"))
self.assertEqual(missing, ["libgbm.so.1", "libnss3.so"])
self.assertEqual(run.call_args.args[0], ["ldd", "/opt/chs"])
def ensure_libraries(
self, *, manager: str, missing: list[list[str]], files: dict[str, str]
) -> mock.Mock:
"""Run the library step with stubbed ldd, package manager and file DB.
Args:
manager: the package manager the host reports.
missing: successive ``missing_libraries`` results.
files: ``pacman -Fq`` stdout per queried path.
Returns:
The stubbed ``run_privileged``.
"""
def query(argv, **_kwargs):
return _done(0, files.get(argv[-1], ""))
with (
mock.patch.object(mermaid_cli, "missing_libraries", side_effect=missing),
mock.patch.object(
mermaid_cli, "detect_package_manager", return_value=manager
),
mock.patch.object(mermaid_cli.subprocess, "run", side_effect=query),
mock.patch.object(mermaid_cli, "run_privileged") as privileged,
):
mermaid_cli._ensure_browser_libraries(Path("/opt/chs"))
return privileged
def test_resolved_libraries_install_nothing(self) -> None:
privileged = self.ensure_libraries(manager="dnf", missing=[[]], files={})
privileged.assert_not_called()
def test_the_redhat_family_installs_by_soname_provide(self) -> None:
privileged = self.ensure_libraries(
manager="dnf", missing=[["libgbm.so.1", "libnss3.so"], []], files={}
)
privileged.assert_called_once_with(
["dnf", "-y", "install", "libgbm.so.1()(64bit)", "libnss3.so()(64bit)"]
)
def test_arch_takes_the_first_package_owning_each_soname(self) -> None:
privileged = self.ensure_libraries(
manager="pacman",
missing=[["libgbm.so.1", "libnss3.so"], []],
files={
"usr/lib/libgbm.so.1": "extra/mesa\nextra/mesa-amber\n",
"usr/lib/libnss3.so": "core/nss\n",
},
)
self.assertEqual(
privileged.call_args_list,
[
mock.call(["pacman", "-Fy"]),
mock.call(["pacman", "-Syu", "--noconfirm", "--needed", "mesa", "nss"]),
],
)
def test_apt_without_libraries_fails_without_installing(self) -> None:
with self.assertRaisesRegex(RuntimeError, "--install-deps"):
self.ensure_libraries(manager="apt-get", missing=[["libnss3.so"]], files={})
def test_libraries_still_missing_after_the_install_raise(self) -> None:
with self.assertRaisesRegex(RuntimeError, "still lacks libnss3.so"):
self.ensure_libraries(
manager="dnf", missing=[["libnss3.so"], ["libnss3.so"]], files={}
)
class TestBrowserInstalled(unittest.TestCase):
@@ -213,6 +356,14 @@ class TestBrowserInstalled(unittest.TestCase):
_install_browser(cache, REVISION)
self.assertTrue(mermaid_cli.browser_installed())
def test_a_browser_with_missing_libraries_is_not_installed(self) -> None:
with _environment() as (_, cache):
_install_browser(cache, REVISION)
with mock.patch.object(
mermaid_cli, "missing_libraries", return_value=["libnss3.so"]
):
self.assertFalse(mermaid_cli.browser_installed())
def test_the_revision_is_read_from_mmdcs_puppeteer_core(self) -> None:
with _environment():
self.assertEqual(mermaid_cli.browser_revision(), REVISION)
+59 -3
View File
@@ -9,19 +9,75 @@ from utils.install import ruby as ruby_mod
class TestEnsureRubyPresent(unittest.TestCase):
def test_delegates_to_the_shared_provisioner(self) -> None:
with mock.patch.object(ruby_mod, "ensure_command_present") as ensure:
def test_provisions_ruby_then_minitest(self) -> None:
calls = mock.Mock()
with (
mock.patch.object(ruby_mod, "ensure_command_present", calls.ruby),
mock.patch.object(ruby_mod, "ensure_minitest_present", calls.minitest),
):
ruby_mod.ensure_ruby_present()
ensure.assert_called_once_with("ruby")
self.assertEqual(
calls.mock_calls, [mock.call.ruby("ruby"), mock.call.minitest()]
)
def test_propagates_the_hard_failure(self) -> None:
with (
mock.patch.object(
ruby_mod, "ensure_command_present", side_effect=RuntimeError("boom")
),
mock.patch.object(ruby_mod, "ensure_minitest_present") as minitest,
self.assertRaises(RuntimeError),
):
ruby_mod.ensure_ruby_present()
minitest.assert_not_called()
class TestEnsureMinitestPresent(unittest.TestCase):
def run_with(self, *, loadable: list[bool], manager: str) -> mock.Mock:
"""Run ``ensure_minitest_present`` with stubbed probe and installer.
Args:
loadable: successive results of the minitest probe.
manager: package manager the host reports.
Returns:
The stubbed ``install_package_candidates``.
"""
with (
mock.patch.object(ruby_mod, "minitest_loadable", side_effect=loadable),
mock.patch.object(ruby_mod, "detect_package_manager", return_value=manager),
mock.patch.object(ruby_mod, "install_package_candidates") as install,
mock.patch.object(ruby_mod, "log"),
):
ruby_mod.ensure_minitest_present()
return install
def test_a_loadable_minitest_installs_nothing(self) -> None:
install = self.run_with(loadable=[True], manager="pacman")
install.assert_not_called()
def test_arch_installs_ruby_minitest(self) -> None:
install = self.run_with(loadable=[False, True], manager="pacman")
install.assert_called_once_with("pacman", ["ruby-minitest"])
def test_the_redhat_family_installs_rubygem_minitest(self) -> None:
install = self.run_with(loadable=[False, True], manager="dnf")
install.assert_called_once_with("dnf", ["rubygem-minitest"])
def test_still_missing_after_the_install_fails_loudly(self) -> None:
with self.assertRaises(RuntimeError):
self.run_with(loadable=[False, False], manager="pacman")
def test_an_unmapped_manager_fails_without_installing(self) -> None:
with (
mock.patch.object(ruby_mod, "minitest_loadable", return_value=False),
mock.patch.object(ruby_mod, "detect_package_manager", return_value="brew"),
mock.patch.object(ruby_mod, "install_package_candidates") as install,
mock.patch.object(ruby_mod, "log"),
self.assertRaises(RuntimeError),
):
ruby_mod.ensure_minitest_present()
install.assert_not_called()
if __name__ == "__main__":
+125 -42
View File
@@ -2,34 +2,38 @@
mmdc comes from npm with ``PUPPETEER_SKIP_DOWNLOAD`` set; the browser is
provisioned afterwards through the puppeteer that npm installed next to mmdc,
at the revision its puppeteer-core pins, as root with its shared libraries
(``--install-deps``). ``unzip`` is ensured first because puppeteer extracts the
browser with it. A failed install warns instead of aborting ``install-lint``, so
sandboxes without npm or browser access still get the rest of the lint
toolchain.
at the revision its puppeteer-core pins. ``unzip`` is ensured first because
puppeteer extracts the browser with it. Puppeteer installs the browser's
shared libraries only through apt-get (``--install-deps``); on dnf, yum and
pacman the libraries ``ldd`` reports missing are installed by soname. Every
failure aborts ``install-lint``.
"""
from __future__ import annotations
import contextlib
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
from utils.install.npm import npm_install_global
from utils.install.primitives import log, warn, which
from utils.install.system_pkg import install_command_via_pkg
from utils.install.primitives import log, run_privileged, which
from utils.install.system_pkg import detect_package_manager, install_command_via_pkg
BROWSER = "chrome-headless-shell"
PROVISION_TIMEOUT_S = 900
_INSTALL_ARGV = {
"dnf": ["dnf", "-y", "install"],
"yum": ["yum", "-y", "install"],
"pacman": ["pacman", "-Syu", "--noconfirm", "--needed"],
}
def _ensure_unzip() -> None:
if which("unzip"):
return
with contextlib.suppress(Exception):
if not which("unzip"):
install_command_via_pkg("unzip")
@@ -75,15 +79,42 @@ def browser_revision() -> str | None:
return match.group(1) if match else None
def browser_installed() -> bool:
def browser_executable() -> Path | None:
revision = browser_revision()
if revision is None:
return False
return any(
path.is_file() for path in browser_cache().glob(f"*-{revision}/*/{BROWSER}")
return None
return next(
(
path
for path in browser_cache().glob(f"*-{revision}/*/{BROWSER}")
if path.is_file()
),
None,
)
def missing_libraries(executable: Path) -> list[str]:
"""Return the shared libraries the dynamic loader cannot resolve.
Args:
executable: the browser binary.
Returns:
Sorted sonames ``ldd`` reports as ``not found``.
"""
done = subprocess.run(
["ldd", str(executable)], capture_output=True, text=True, check=False
)
return sorted(
{line.split()[0] for line in done.stdout.splitlines() if "=> not found" in line}
)
def browser_installed() -> bool:
executable = browser_executable()
return executable is not None and not missing_libraries(executable)
def _provision_browser(puppeteer: Path) -> subprocess.CompletedProcess[str]:
argv = [
str(puppeteer.parent / ".bin" / "puppeteer"),
@@ -91,7 +122,7 @@ def _provision_browser(puppeteer: Path) -> subprocess.CompletedProcess[str]:
"install",
BROWSER,
]
if os.geteuid() == 0:
if os.geteuid() == 0 and which("apt-get"):
argv.append("--install-deps")
env = {k: v for k, v in os.environ.items() if k != "PUPPETEER_SKIP_DOWNLOAD"}
return subprocess.run(
@@ -104,43 +135,95 @@ def _provision_browser(puppeteer: Path) -> subprocess.CompletedProcess[str]:
)
def _ensure_browser() -> None:
if browser_installed():
return
def _download_browser() -> None:
puppeteer = _mmdc_puppeteer()
if puppeteer is None:
warn(f"{BROWSER} provisioning skipped: no puppeteer next to mmdc")
return
raise RuntimeError(f"Cannot provision {BROWSER}: no puppeteer next to mmdc.")
log(f"Provisioning {BROWSER} for mmdc via its own puppeteer.")
try:
done = _provision_browser(puppeteer)
if done.returncode != 0:
revision = browser_revision()
if revision is not None:
for partial in browser_cache().glob(f"*-{revision}"):
shutil.rmtree(partial, ignore_errors=True)
done = _provision_browser(puppeteer)
except (OSError, subprocess.TimeoutExpired) as exc:
warn(f"{BROWSER} provisioning skipped (optional lint tool): {exc}")
return
done = _provision_browser(puppeteer)
if done.returncode != 0:
warn(
f"{BROWSER} provisioning failed (optional lint tool):\n"
f"{done.stdout}{done.stderr}"
revision = browser_revision()
if revision is not None:
for partial in browser_cache().glob(f"*-{revision}"):
shutil.rmtree(partial, ignore_errors=True)
done = _provision_browser(puppeteer)
if done.returncode != 0:
raise RuntimeError(
f"{BROWSER} provisioning failed:\n{done.stdout}{done.stderr}"
)
def _library_packages(manager: str, sonames: list[str]) -> list[str]:
"""Map sonames to what the package manager installs them by.
Args:
manager: the detected package manager.
sonames: libraries the browser cannot resolve.
Returns:
Install arguments for ``manager``.
Raises:
RuntimeError: the manager cannot install by soname.
"""
if manager in ("dnf", "yum"):
suffix = "()(64bit)" if sys.maxsize > 2**32 else ""
return [f"{soname}{suffix}" for soname in sonames]
if manager == "pacman":
run_privileged(["pacman", "-Fy"])
packages: set[str] = set()
for soname in sonames:
found = subprocess.run(
["pacman", "-Fq", f"usr/lib/{soname}"],
capture_output=True,
text=True,
check=False,
).stdout.split()
if found:
packages.add(found[0].rsplit("/", 1)[-1])
return sorted(packages)
raise RuntimeError(
f"{BROWSER} lacks {', '.join(sonames)} and {manager} cannot install "
f"libraries by soname; on apt-get they come from --install-deps, which "
f"needs root."
)
def _ensure_browser_libraries(executable: Path) -> None:
missing = missing_libraries(executable)
if not missing:
return
manager = detect_package_manager()
packages = _library_packages(manager, missing)
log(f"Installing {BROWSER} libraries via {manager}: {', '.join(packages)}")
if packages:
run_privileged([*_INSTALL_ARGV[manager], *packages])
missing = missing_libraries(executable)
if missing:
raise RuntimeError(
f"{BROWSER} still lacks {', '.join(missing)} after installing "
f"{packages} via {manager}."
)
def _ensure_browser() -> None:
executable = browser_executable()
if executable is None:
_download_browser()
executable = browser_executable()
if executable is None:
raise RuntimeError(f"{BROWSER} is still missing after provisioning.")
_ensure_browser_libraries(executable)
def ensure() -> None:
_ensure_unzip()
if not which("mmdc"):
log("Missing command 'mmdc'. Installing @mermaid-js/mermaid-cli via npm.")
os.environ["PUPPETEER_SKIP_DOWNLOAD"] = "1"
try:
npm_install_global("@mermaid-js/mermaid-cli")
except Exception as exc: # noqa: BLE001 - best-effort optional lint tool
warn(f"mermaid-cli install skipped (optional lint tool): {exc}")
return
npm_install_global("@mermaid-js/mermaid-cli")
if not which("mmdc"):
warn("Command 'mmdc' is still unavailable after installation.")
return
raise RuntimeError(
"Command 'mmdc' is still unavailable after installation."
)
_ensure_browser()
+54 -3
View File
@@ -2,16 +2,67 @@
from __future__ import annotations
from utils.install.system_pkg import ensure_command_present
import subprocess
from utils.install.primitives import log
from utils.install.system_pkg import (
detect_package_manager,
ensure_command_present,
install_package_candidates,
)
_MINITEST_PACKAGES: dict[str, list[str]] = {
"pacman": ["ruby-minitest"],
"dnf": ["rubygem-minitest"],
"yum": ["rubygem-minitest"],
"apt-get": ["ruby-minitest"],
}
def minitest_loadable() -> bool:
"""Return whether the installed ruby can require minitest.
Returns:
``True`` when ``ruby -e 'require "minitest"'`` exits 0.
"""
probe = subprocess.run(
["ruby", "-e", 'require "minitest"'], capture_output=True, check=False
)
return probe.returncode == 0
def ensure_minitest_present() -> None:
"""Make minitest loadable, installing its package if ruby lacks it.
Raises:
RuntimeError: no package mapping for the detected package manager, or
minitest is still not loadable after the install.
"""
if minitest_loadable():
return
manager = detect_package_manager()
packages = _MINITEST_PACKAGES.get(manager)
if packages is None:
raise RuntimeError(f"No installer mapping defined for minitest on '{manager}'.")
log(f"Ruby cannot load minitest. Attempting installation via {manager}.")
install_package_candidates(manager, packages)
if not minitest_loadable():
raise RuntimeError(
f"minitest still not loadable after installing {packages} via {manager}"
)
def ensure_ruby_present() -> None:
"""Install the Ruby interpreter through the system package manager.
"""Install the Ruby interpreter and minitest through the system package manager.
Raises:
RuntimeError: ruby is still absent after the install attempt.
RuntimeError: ruby or minitest is still absent after the install attempt.
"""
ensure_command_present("ruby")
ensure_minitest_present()
if __name__ == "__main__":