From f432e30acfe5509e4ee9769da0a2e9c0b2ecfc2f Mon Sep 17 00:00:00 2001 From: ethernet Date: Sun, 13 Sep 2026 14:46:51 -0400 Subject: [PATCH] test: consolidate root artifact and runtime boundary coverage --- tests/termux_fixtures.py | 65 ++++ tests/test_engines_satisfiable.py | 213 ------------ tests/test_fast_safe_load.py | 43 --- tests/test_fresh_source_install.py | 7 + tests/test_git_safety_boundary.py | 37 ++ tests/test_hermes_bootstrap.py | 126 +++---- tests/test_hermes_logging.py | 275 ++------------- tests/test_hermes_yaml.py | 15 +- tests/test_install_sh_node_deps_workspaces.py | 19 - tests/test_iron_proxy.py | 167 +-------- tests/test_model_tools_async_bridge.py | 138 +------- tests/test_node_resolution.py | 63 ++-- tests/test_old_updater_additional_shims.py | 27 -- tests/test_old_updater_main_shims.py | 68 ---- tests/test_old_updater_shims.py | 63 +++- tests/test_os_marker_gating.py | 101 +++--- tests/test_packaging_build_guard.py | 48 +-- tests/test_packaging_metadata.py | 167 ++------- tests/test_platforms_marker.py | 188 ++-------- tests/test_plugin_historic_compat.py | 14 - tests/test_pm_bionic.py | 200 ++++------- tests/test_project_metadata.py | 128 +------ tests/test_run_tests_parallel.py | 129 ++----- tests/test_stage_apt_repo.py | 329 ++++-------------- tests/test_termux_deb_version.py | 166 +++------ tests/test_termux_python_linkage.py | 33 +- tests/test_termux_retag_wheel.py | 82 +---- tests/test_termux_runtime_libs.py | 85 ++--- tests/test_yaml_indent_consistency_31999.py | 47 +-- 29 files changed, 677 insertions(+), 2366 deletions(-) create mode 100644 tests/termux_fixtures.py delete mode 100644 tests/test_engines_satisfiable.py delete mode 100644 tests/test_fast_safe_load.py create mode 100644 tests/test_git_safety_boundary.py delete mode 100644 tests/test_install_sh_node_deps_workspaces.py diff --git a/tests/termux_fixtures.py b/tests/termux_fixtures.py new file mode 100644 index 0000000000..af851b8f25 --- /dev/null +++ b/tests/termux_fixtures.py @@ -0,0 +1,65 @@ +"""Small, structurally valid Termux archives with independent integrity checks.""" +import base64 +import csv +import hashlib +import io +import tarfile +import zipfile +from pathlib import Path + + +def build_deb(path, control, files=(), compression="gz"): + members = [("debian-binary", b"2.0\n")] + for kind, entries in (("control", {"control": "".join(f"{k}: {v}\n" for k, v in control.items()).encode()}), + ("data", dict(files))): + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode=f"w:{compression}") as archive: + for name, data in entries.items(): + info = name if isinstance(name, tarfile.TarInfo) else tarfile.TarInfo(name) + info.size = len(data) + archive.addfile(info, io.BytesIO(data)) + members.append((f"{kind}.tar.{compression}", buffer.getvalue())) + result = bytearray(b"!\n") + for name, data in members: + result.extend(f"{name:<16}{0:<12}{0:<6}{0:<6}{'100644':<8}{len(data):<10}`\n".encode()) + result.extend(data) + result.extend(b"\n" if len(data) % 2 else b"") + path.write_bytes(result) + + +def record_hash(data): + return "sha256=" + base64.urlsafe_b64encode(hashlib.sha256(data).digest()).rstrip(b"=").decode() + + +def write_wheel(directory, distribution="fakedep", version="1.2.3", platform_tag="linux_aarch64", + *, metadata_version=None, include_so=True): + info = f"{distribution}-{version}.dist-info" + members = { + f"{distribution}/__init__.py": b"", + f"{info}/METADATA": f"Metadata-Version: 2.1\nName: {distribution}\nVersion: {metadata_version or version}\n".encode(), + f"{info}/WHEEL": f"Wheel-Version: 1.0\nRoot-Is-Purelib: false\nTag: py3-none-{platform_tag}\n".encode(), + } + if include_so: + members[f"{distribution}/_native.so"] = b"\x7fELFfake" + buffer = io.StringIO() + csv.writer(buffer, lineterminator="\n").writerows( + [[name, record_hash(data), str(len(data))] for name, data in members.items()] + + [[f"{info}/RECORD", "", ""]] + ) + members[f"{info}/RECORD"] = buffer.getvalue().encode() + path = Path(directory) / f"{distribution}-{version}-py3-none-{platform_tag}.whl" + with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as archive: + for name, data in members.items(): + archive.writestr(name, data) + return path + + +def verify_record(path): + with zipfile.ZipFile(path) as archive: + record, = [name for name in archive.namelist() if name.endswith(".dist-info/RECORD")] + rows = list(csv.reader(io.StringIO(archive.read(record).decode()))) + assert len(rows) == len(archive.namelist()) + assert {row[0] for row in rows} == set(archive.namelist()) + for name, digest, size in rows: + data = archive.read(name) + assert (digest, size) == (("", "") if name == record else (record_hash(data), str(len(data)))) diff --git a/tests/test_engines_satisfiable.py b/tests/test_engines_satisfiable.py deleted file mode 100644 index 72ef27267d..0000000000 --- a/tests/test_engines_satisfiable.py +++ /dev/null @@ -1,213 +0,0 @@ -"""The manifest's ``engines`` must be satisfiable by a toolchain we can actually ship. - -`engine-strict=true` in `.npmrc` makes `engines` a hard gate on every -`npm ci` / `npm install` — the installer's workspace step, `hermes update`'s -dependency refresh, and CI alike. So a floor nobody's toolchain can meet is -not a strict-hygiene win; it is a total install outage. - -That is exactly what happened: `engines.npm` was raised to `>=12.0.0` while -**no Node release bundles npm 12** (Node 26 ships 11.17.0, 24 ships 11.16.0, -22 ships 10.9.8). Every fresh install died at the first `npm ci`, and -`hermes update` left installs in a mixed state. These tests encode the -invariants that would have caught it. - -Deliberately behavioral, not a snapshot: nothing here pins a version we -expect to change. Each test asserts a *relationship* — between the floor we -declare and the toolchain that has to satisfy it. -""" - -from __future__ import annotations - -import json -import re -from pathlib import Path - -import pytest - -REPO_ROOT = Path(__file__).resolve().parents[1] - -# npm releases bundled with a Node major, newest-per-major. Not a catalog -# snapshot: the point is that *some* real, shipping toolchain must clear the -# floor, and these are the ones users actually arrive with. -_STOCK_NPM_BY_NODE_MAJOR = { - 20: "10.8.2", - 22: "10.9.8", - 24: "11.16.0", - 26: "11.17.0", -} - - -def _root_manifest() -> dict: - return json.loads((REPO_ROOT / "package.json").read_text()) - - -def _parse_major_minor_patch(version: str) -> tuple[int, int, int]: - parts = version.split("-", 1)[0].split(".") - nums = [int(p) for p in parts[:3]] - while len(nums) < 3: - nums.append(0) - return nums[0], nums[1], nums[2] - - -def _satisfies_clause(version: str, clause: str) -> bool: - """Evaluate one `>=x.y.z` / `=x.y.z within the same major (x > 0). - return have[0] == want[0] and have >= want - for op in (">=", "<=", "<", ">", "="): - if clause.startswith(op): - bound = clause[len(op) :].strip() - break - else: - op, bound = "=", clause - have = _parse_major_minor_patch(version) - want = _parse_major_minor_patch(bound) - if op == ">=": - return have >= want - if op == "<=": - return have <= want - if op == "<": - return have < want - if op == ">": - return have > want - return have == want - - -def _satisfies_range(version: str, spec: str) -> bool: - """Evaluate the `A || B` / space-joined-AND subset of semver we author.""" - for alternative in spec.split("||"): - clauses = [c for c in alternative.strip().split() if c] - if clauses and all(_satisfies_clause(version, c) for c in clauses): - return True - return False - - -class TestEnginesAreSatisfiable: - def test_npm_floor_is_met_by_a_shipping_node(self): - """Some stock Node must bundle an npm our floor accepts. - - Without this, a fresh install cannot run `npm ci` at all: the - installer provisions a Node from nodejs.org and immediately uses the - npm that came with it. - """ - npm_range = _root_manifest()["engines"]["npm"] - satisfying = { - major: npm - for major, npm in _STOCK_NPM_BY_NODE_MAJOR.items() - if _satisfies_range(npm, npm_range) - } - assert satisfying, ( - f"engines.npm is {npm_range!r}, which no shipping Node bundles " - f"(checked {_STOCK_NPM_BY_NODE_MAJOR}). With engine-strict=true " - "every fresh install fails at the first `npm ci`." - ) - - def test_node_floor_is_met_by_the_managed_runtime(self): - """The Node major the installers provision must clear engines.node.""" - node_range = _root_manifest()["engines"]["node"] - # pm-era install: node is a pm package pinned in pm/lock.json (the - # installers stage it via `pm`, not a NODE_VERSION shell var). - lock = json.loads((REPO_ROOT / "pm" / "lock.json").read_text(encoding="utf-8")) - node_pin = lock["packages"]["node"]["version"] - managed_major = int(node_pin.split(".")[0]) - - # pm fetches the exact pinned version, so compare on the major: the - # pinned node line must clear the floor. A floor in a HIGHER major - # than we provision can never be met. - floor_majors = [ - int(m.group(1)) - for m in re.finditer(r">=\s*v?(\d+)", node_range) - ] - assert floor_majors, f"cannot read a floor out of {node_range!r}" - assert managed_major >= min(floor_majors), ( - f"engines.node is {node_range!r} but pm/lock.json pins Node " - f"{node_pin}. The runtime we ship must satisfy the floor we " - "declare, or the install we just performed cannot install deps." - ) - - def test_managed_node_bundles_an_npm_the_engines_accept(self): - """The Node major install.sh fetches must ship an npm that clears - engines.npm. Node 22 bundles 11.16.0, which is in the excluded - 11.10–11.16 band — fresh Hermes-managed installs then die at - `npm ci` with EBADENGINE (#80769). - """ - npm_range = _root_manifest()["engines"]["npm"] - # pm-era install: node is pinned in pm/lock.json; the npm that - # rides with it is the pm-managed npm (also pinned there). - lock = json.loads((REPO_ROOT / "pm" / "lock.json").read_text(encoding="utf-8")) - managed_major = int(lock["packages"]["node"]["version"].split(".")[0]) - managed_npm = lock["packages"].get("npm", {}).get("version", "") - if managed_npm: - # The pinned npm's own version — clear the floor directly. - assert _satisfies_range(managed_npm, npm_range), ( - f"pm/lock.json pins npm {managed_npm}, but engines.npm is " - f"{npm_range!r}. A fresh Hermes-managed install cannot run npm ci." - ) - else: - stock_npm = _STOCK_NPM_BY_NODE_MAJOR.get(managed_major) - assert stock_npm is not None, ( - f"pm/lock.json pins Node {managed_major} but it is not in the " - f"known stock map {_STOCK_NPM_BY_NODE_MAJOR}" - ) - assert _satisfies_range(stock_npm, npm_range), ( - f"pm/lock.json pins Node {managed_major}.x (stock npm " - f"{stock_npm}), but engines.npm is {npm_range!r}. A fresh " - "Hermes-managed install cannot run npm ci." - ) - - def test_desktop_node_floor_is_not_stricter_than_its_toolchain(self): - """apps/desktop must not demand more Node than its own build tools do. - - Vite is the real constraint (it needs `node:util.styleText`). Raising - the desktop floor beyond it silently force-migrates every user's - toolchain for no dependency reason. - """ - desktop = json.loads((REPO_ROOT / "apps" / "desktop" / "package.json").read_text()) - node_range = desktop["engines"]["node"] - # The tightest floor any dependency actually declares (react-router - # 8.3.0 -> >=22.22.0). If this legitimately rises, the assertion - # documents the reason for the bump rather than blocking it. - assert _satisfies_range("22.22.0", node_range), ( - f"apps/desktop engines.node is {node_range!r}, which rejects Node " - "22.12 — stricter than Vite requires. A desktop floor above the " - "build toolchain's own floor replaces working user toolchains for " - "nothing." - ) - - -class TestExcludedNpmBand: - """npm 11.10–11.16 honor `min-release-age` but ignore `min-release-age-exclude`. - - `.npmrc` sets both, so that band applies the 14-day age gate to packages - we deliberately exempted and installs fail with ETARGET. The floor must - keep excluding them. - """ - - @pytest.mark.parametrize("bad_npm", ["11.10.0", "11.12.1", "11.16.0"]) - def test_band_that_ignores_the_exclude_list_is_rejected(self, bad_npm): - npm_range = _root_manifest()["engines"]["npm"] - assert not _satisfies_range(bad_npm, npm_range), ( - f"engines.npm {npm_range!r} accepts npm {bad_npm}, which supports " - "min-release-age but not min-release-age-exclude — it will fail " - "ETARGET on any freshly published dependency in .npmrc's exclude list." - ) - - @pytest.mark.parametrize("good_npm", ["10.9.8", "11.17.0", "12.0.2"]) - def test_versions_handling_the_exclude_list_are_accepted(self, good_npm): - npm_range = _root_manifest()["engines"]["npm"] - assert _satisfies_range(good_npm, npm_range), ( - f"engines.npm {npm_range!r} rejects npm {good_npm}, which handles " - ".npmrc correctly and should be usable." - ) - - -class TestManifestMirrors: - def test_lockfile_engines_match_the_manifest(self): - """A stale lockfile mirror re-imposes the old floor on `npm ci`.""" - manifest = _root_manifest()["engines"] - lock = json.loads((REPO_ROOT / "package-lock.json").read_text()) - assert lock["packages"][""]["engines"] == manifest diff --git a/tests/test_fast_safe_load.py b/tests/test_fast_safe_load.py deleted file mode 100644 index 3cc9c159d4..0000000000 --- a/tests/test_fast_safe_load.py +++ /dev/null @@ -1,43 +0,0 @@ -"""The fast-load entry point follows the shared YAML policy.""" - -import io - -import pytest - -import hermes_yaml as yaml -from utils import fast_safe_load - - -_DOCS = [ - "", - "a: 1\nb: two\nc: 3.5\n", - "list: [1, 2, 3]\nnested:\n k: v\n flag: true\n empty: null\n", - "name: skill-x\nmetadata:\n hermes:\n tags: [alpha, beta]\n category: devops\n", - "- one\n- two\n- three\n", - "scalar string", - "flags: [on, off, yes, no, y, n]\n", -] - - -def test_equivalent_to_safe_load_for_strings(): - for doc in _DOCS: - assert fast_safe_load(doc) == yaml.safe_load(doc), repr(doc) - - -def test_equivalent_to_safe_load_for_file_objects(): - for doc in _DOCS: - assert fast_safe_load(io.StringIO(doc)) == yaml.safe_load(io.StringIO(doc)), repr(doc) - - -def test_empty_document_returns_none(): - assert fast_safe_load("") is None - - -def test_duplicate_keys_are_rejected_instead_of_silently_overwriting(): - with pytest.raises(yaml.YAMLError): - fast_safe_load("model: first\nmodel: second\n") - - -def test_rejects_arbitrary_python_objects_like_safe_load(): - with pytest.raises(yaml.YAMLError): - fast_safe_load("!!python/object/apply:builtins.str ['must not construct']\n") diff --git a/tests/test_fresh_source_install.py b/tests/test_fresh_source_install.py index 8cd8b3827e..9326a2c202 100644 --- a/tests/test_fresh_source_install.py +++ b/tests/test_fresh_source_install.py @@ -46,6 +46,12 @@ def test_current_installer_publishes_real_dependencies_and_warm_path(tmp_path, s env = {"PATH": os.environ["PATH"], "HOME": str(home), "LANG": "C.UTF-8", "HERMES_HOME": str(home / ".hermes"), "UV_PYTHON_INSTALL_DIR": str(managed), "UV_PYTHON_DOWNLOADS": "never", "UV_CACHE_DIR": str(tmp_path / "cache")} + canary = tmp_path / "ambient-bin" + canary.mkdir() + npm_called = tmp_path / "npm-called" + (canary / "npm").write_text(f'#!/bin/sh\nprintf called > "{npm_called}"\nexit 99\n', encoding="utf-8") + (canary / "npm").chmod(0o755) + env["PATH"] = str(canary) + os.pathsep + env["PATH"] for key in ("SSL_CERT_FILE", "SSL_CERT_DIR", "NIX_SSL_CERT_FILE"): if key in os.environ: env[key] = os.environ[key] @@ -110,6 +116,7 @@ def test_current_installer_publishes_real_dependencies_and_warm_path(tmp_path, s command = ["bash", str(ROOT / "scripts/install.sh"), "--dir", str(install), "--branch", "fixture", "--commit", commit, "--non-interactive", "--json"] result = run(command, expected=1 if fault else 0) + assert not npm_called.exists() if fault: assert not (install / ".hermes-bootstrap-complete").exists() assert not (home / ".local/bin/hermes").exists() diff --git a/tests/test_git_safety_boundary.py b/tests/test_git_safety_boundary.py new file mode 100644 index 0000000000..0d01983f05 --- /dev/null +++ b/tests/test_git_safety_boundary.py @@ -0,0 +1,37 @@ +"""A removed live-checkout guard can damage only disposable repositories here.""" +from pathlib import Path +import shlex +import subprocess + +import pytest + + +@pytest.mark.platforms("posix") +def test_guard_blocks_native_and_shell_git_mutations_without_touching_checkout(tmp_path, monkeypatch): + from tests import conftest + + def git(repo, *args): + result = subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True, check=True) + return result.stdout.strip() + + protected, ordinary = tmp_path / "protected", tmp_path / "ordinary" + for repo in (protected, ordinary): + repo.mkdir() + git(repo, "init") + git(repo, "-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "--allow-empty", "-m", "first") + (repo / "sentinel").write_text("committed", encoding="utf-8") + git(repo, "add", "sentinel") + git(repo, "-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "-m", "second") + monkeypatch.setattr(conftest, "_LIVE_GUARD_PROTECTED_GIT_ROOTS", (protected,)) + head = git(protected, "rev-parse", "HEAD") + (protected / "sentinel").write_bytes(b"uncommitted user data") + for command in (["git", "-C", str(protected), "reset", "--hard", "HEAD~1"], + ["sh", "-c", f"git -C {shlex.quote(str(protected))} checkout -- sentinel"]): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.run(command, check=True) + assert git(protected, "rev-parse", "HEAD") == head + assert (protected / "sentinel").read_bytes() == b"uncommitted user data" + old = git(ordinary, "rev-parse", "HEAD~1") + git(ordinary, "reset", "--hard", old) + assert git(ordinary, "rev-parse", "HEAD") == old + assert not (ordinary / "sentinel").exists() diff --git a/tests/test_hermes_bootstrap.py b/tests/test_hermes_bootstrap.py index d9f2035c7f..031054b4fa 100644 --- a/tests/test_hermes_bootstrap.py +++ b/tests/test_hermes_bootstrap.py @@ -187,86 +187,54 @@ class TestStdioReconfigureErrorHandling: -class TestEntryPointsImportBootstrap: - """Every Hermes entry point must import hermes_bootstrap as its - first non-docstring import. We check this by scanning source files - rather than invoking the entry points (which would require a full - agent context).""" +@pytest.mark.parametrize("path", [ + "hermes_cli/main.py", "run_agent.py", "acp_adapter/entry.py", + "gateway/run.py", "batch_runner.py", "cli.py", +]) +def test_entrypoint_executes_bootstrap_before_application_imports(tmp_path, path): + import subprocess + from pathlib import Path - # Entry points that invoke Hermes as a process. Each one must - # import hermes_bootstrap before doing any file I/O or stdout writes. - ENTRY_POINTS = [ - "hermes_cli/main.py", # hermes CLI (console_script) - "run_agent.py", # hermes-agent (console_script) - "acp_adapter/entry.py", # hermes-acp (console_script) - "gateway/run.py", # gateway - "batch_runner.py", # batch mode - "cli.py", # legacy direct-launch CLI - ] - - @pytest.mark.parametrize("path", ENTRY_POINTS) - def test_entry_point_imports_bootstrap(self, path): - """The file must contain 'import hermes_bootstrap' and that - line must appear before the first 'import' of anything else. - - We're lenient about the docstring (can be arbitrarily long) and - about comment lines — just need to verify the first import - statement is the bootstrap. - - Also lenient about a try/except wrapper around the import: entry - points may guard the import against ``ModuleNotFoundError`` so a - half-finished ``hermes update`` (git-reset landed new code but - ``uv pip install -e .`` didn't finish re-registering - ``hermes_bootstrap`` as a top-level module) leaves hermes - recoverable instead of crashing on every invocation. When the - first top-level node is such a guarded-import block, we peek - inside it to verify bootstrap is the imported module. - """ - # Resolve relative to the hermes-agent repo root. Tests live - # at tests/test_hermes_bootstrap.py, so go up one dir. - import pathlib - here = pathlib.Path(__file__).resolve() - repo_root = here.parent.parent # tests/ -> repo root - full_path = repo_root / path - assert full_path.exists(), f"entry point missing: {full_path}" - - source = full_path.read_text(encoding="utf-8") - - # Find the first non-comment, non-blank line that starts with - # 'import ' or 'from ', or a Try block whose body is the import. - import ast - tree = ast.parse(source) - - first_import_node = None - for node in ast.iter_child_nodes(tree): - if isinstance(node, (ast.Import, ast.ImportFrom)): - first_import_node = node - break - # Accept a guarded-import Try block where the body is a lone - # Import node — this is the recovery-friendly form that lets - # hermes start even when hermes_bootstrap hasn't been - # re-registered in the venv yet. - if isinstance(node, ast.Try) and len(node.body) == 1 and isinstance( - node.body[0], (ast.Import, ast.ImportFrom) - ): - first_import_node = node.body[0] - break - - assert first_import_node is not None, ( - f"{path}: no top-level imports found at all" - ) - - if isinstance(first_import_node, ast.Import): - first_import_name = first_import_node.names[0].name - else: # ImportFrom - first_import_name = first_import_node.module or "" - - assert first_import_name == "hermes_bootstrap", ( - f"{path}: first top-level import is {first_import_name!r}, " - f"but it must be 'hermes_bootstrap' so UTF-8 stdio is " - f"configured before anything else initializes. Move the " - f"'import hermes_bootstrap' line to be the first import." - ) + root = Path(__file__).resolve().parents[1] + entry = tmp_path / "startup.py" + entry.write_bytes((root / path).read_bytes()) + # Stop at the first application import, after executing the REAL bootstrap. + # pm repair is the supported stdlib-only startup, so no update/service runs. + program = r""" +import builtins, os, runpy, sys +root, entry = sys.argv[1:] +sys.path.insert(0, root) +sys.argv = [entry, 'pm', 'repair'] +real_import = builtins.__import__ +class Boundary(BaseException): pass +seen = [] +def guarded(name, globals=None, locals=None, fromlist=(), level=0): + if globals and globals.get('__file__') == entry: + if name == '__future__': + return real_import(name, globals, locals, fromlist, level) + if not seen: + assert name == 'hermes_bootstrap', name + module = real_import(name, globals, locals, fromlist, level) + assert module._pm_repair is True + assert module._bootstrap_applied is (sys.platform == 'win32') + seen.append(name) + return module + raise Boundary() + return real_import(name, globals, locals, fromlist, level) +builtins.__import__ = guarded +try: + runpy.run_path(entry, run_name='__main__') +except Boundary: + assert seen == ['hermes_bootstrap'] + print('bootstrap-before-app') +else: + raise AssertionError('entrypoint never reached the application import boundary') +""" + result = subprocess.run([sys.executable, "-I", "-S", "-c", program, str(root), str(entry)], + cwd=tmp_path, env={**os.environ, "HERMES_HOME": str(tmp_path / "home")}, + capture_output=True, text=True, timeout=30) + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "bootstrap-before-app" class TestHardenImportPath: diff --git a/tests/test_hermes_logging.py b/tests/test_hermes_logging.py index a49215ed4f..a79559f453 100644 --- a/tests/test_hermes_logging.py +++ b/tests/test_hermes_logging.py @@ -66,57 +66,37 @@ def hermes_home(tmp_path, monkeypatch): return home +@pytest.mark.parametrize("mode,component", [("cli", None), ("gateway", "gateway.log"), ("gui", "gui.log")]) +@pytest.mark.parametrize("configured,explicit,minimum", [(None, None, logging.INFO), ("DEBUG", "WARNING", logging.WARNING), ("DEBUG", None, logging.DEBUG)]) +def test_repeated_setup_routes_records_once(hermes_home, mode, component, configured, explicit, minimum): + if configured: + (hermes_home / "config.yaml").write_text(f"logging:\n level: {configured}\n", encoding="utf-8") + for _ in range(2): + assert hermes_logging.setup_logging(hermes_home=hermes_home, mode=mode, log_level=explicit) == hermes_home / "logs" + hermes_logging.set_session_context("routing-session") + sources = ["tools.terminal_tool", "agent.context_compressor", "gateway.run", + "plugins.platforms.telegram.adapter", "hermes_cli.web_server", "tui_gateway.ws"] + for index, source in enumerate(sources): + for level in (logging.DEBUG, logging.INFO, logging.WARNING): + logging.getLogger(source).log(level, "routing-witness-%s-%s", index, level) + hermes_logging.flush_log_queue() + outputs = {path.name: path.read_text(encoding="utf-8-sig") for path in (hermes_home / "logs").glob("*.log")} + assert set(outputs) == {"agent.log", "errors.log"} | ({component} if component else set()) + for filename, content in outputs.items(): + for index, source in enumerate(sources): + for level in (logging.DEBUG, logging.INFO, logging.WARNING): + accepted = { + "agent.log": level >= minimum, + "errors.log": level >= logging.WARNING, + "gateway.log": index in (2, 3) and level >= max(logging.INFO, minimum), + "gui.log": index in (4, 5) and level >= max(logging.INFO, minimum), + }[filename] + witness = f"routing-witness-{index}-{level}" + assert content.count(witness) == int(accepted), (mode, filename, witness, content) + assert "[routing-session]" in content + + class TestSetupLogging: - """setup_logging() creates agent.log + errors.log with RotatingFileHandler.""" - - def test_creates_log_directory(self, hermes_home): - log_dir = hermes_logging.setup_logging(hermes_home=hermes_home) - assert log_dir == hermes_home / "logs" - assert log_dir.is_dir() - - def test_creates_agent_log_handler(self, hermes_home): - hermes_logging.setup_logging(hermes_home=hermes_home) - root = logging.getLogger() - - agent_handlers = [ - h for h in hermes_logging._queued_file_handlers - if isinstance(h, RotatingFileHandler) - and "agent.log" in getattr(h, "baseFilename", "") - ] - assert len(agent_handlers) == 1 - assert agent_handlers[0].level == logging.INFO - - - def test_idempotent_no_duplicate_handlers(self, hermes_home): - hermes_logging.setup_logging(hermes_home=hermes_home) - hermes_logging.setup_logging(hermes_home=hermes_home) # second call — should be no-op - - root = logging.getLogger() - agent_handlers = [ - h for h in hermes_logging._queued_file_handlers - if isinstance(h, RotatingFileHandler) - and "agent.log" in getattr(h, "baseFilename", "") - ] - assert len(agent_handlers) == 1 - - - - - - def test_writes_to_agent_log(self, hermes_home): - hermes_logging.setup_logging(hermes_home=hermes_home) - - test_logger = logging.getLogger("test_hermes_logging.write_test") - test_logger.info("test message for agent.log") - - # Flush handlers - hermes_logging.flush_log_queue() - - agent_log = hermes_home / "logs" / "agent.log" - assert agent_log.exists() - content = agent_log.read_text(encoding="utf-8-sig") - assert "test message for agent.log" in content - def test_profile_routing_follows_context_home(self, hermes_home, tmp_path): """Desktop multiplex cron records are written to their owning profile.""" from hermes_constants import reset_hermes_home_override, set_hermes_home_override @@ -145,162 +125,6 @@ class TestSetupLogging: - def test_explicit_params_override_config(self, hermes_home): - """Explicit function params take precedence over config.yaml.""" - import hermes_yaml as yaml - config = {"logging": {"level": "DEBUG"}} - (hermes_home / "config.yaml").write_text(yaml.safe_dump(config), encoding="utf-8") - - hermes_logging.setup_logging(hermes_home=hermes_home, log_level="WARNING") - - root = logging.getLogger() - agent_handlers = [ - h for h in hermes_logging._queued_file_handlers - if isinstance(h, RotatingFileHandler) - and "agent.log" in getattr(h, "baseFilename", "") - ] - assert agent_handlers[0].level == logging.WARNING - - - -class TestGatewayMode: - """setup_logging(mode='gateway') creates a filtered gateway.log.""" - - def test_gateway_log_created(self, hermes_home): - hermes_logging.setup_logging(hermes_home=hermes_home, mode="gateway") - root = logging.getLogger() - - gw_handlers = [ - h for h in hermes_logging._queued_file_handlers - if isinstance(h, RotatingFileHandler) - and "gateway.log" in getattr(h, "baseFilename", "") - ] - assert len(gw_handlers) == 1 - - def test_gateway_log_not_created_in_cli_mode(self, hermes_home): - hermes_logging.setup_logging(hermes_home=hermes_home, mode="cli") - root = logging.getLogger() - - gw_handlers = [ - h for h in hermes_logging._queued_file_handlers - if isinstance(h, RotatingFileHandler) - and "gateway.log" in getattr(h, "baseFilename", "") - ] - assert len(gw_handlers) == 0 - - - - def test_gateway_log_receives_gateway_records(self, hermes_home): - """gateway.log captures records from gateway.* loggers.""" - hermes_logging.setup_logging(hermes_home=hermes_home, mode="gateway") - - gw_logger = logging.getLogger("plugins.platforms.telegram.adapter") - gw_logger.info("telegram connected") - - hermes_logging.flush_log_queue() - - gw_log = hermes_home / "logs" / "gateway.log" - assert gw_log.exists() - assert "telegram connected" in gw_log.read_text(encoding="utf-8-sig") - - def test_gateway_log_rejects_non_gateway_records(self, hermes_home): - """gateway.log does NOT capture records from tools.*, agent.*, etc.""" - hermes_logging.setup_logging(hermes_home=hermes_home, mode="gateway") - - tool_logger = logging.getLogger("tools.terminal_tool") - tool_logger.info("running command") - - agent_logger = logging.getLogger("agent.context_compressor") - agent_logger.info("compressing context") - - hermes_logging.flush_log_queue() - - gw_log = hermes_home / "logs" / "gateway.log" - if gw_log.exists(): - content = gw_log.read_text(encoding="utf-8-sig") - assert "running command" not in content - assert "compressing context" not in content - - - -class TestGuiMode: - """setup_logging(mode='gui') creates a filtered gui.log.""" - - def test_gui_log_created(self, hermes_home): - hermes_logging.setup_logging(hermes_home=hermes_home, mode="gui") - root = logging.getLogger() - - gui_handlers = [ - h for h in hermes_logging._queued_file_handlers - if isinstance(h, RotatingFileHandler) - and "gui.log" in getattr(h, "baseFilename", "") - ] - assert len(gui_handlers) == 1 - - - def test_gui_log_receives_only_gui_components(self, hermes_home): - hermes_logging.setup_logging(hermes_home=hermes_home, mode="gui") - - logging.getLogger("hermes_cli.web_server").info("dashboard online") - logging.getLogger("tui_gateway.ws").info("ws connected") - logging.getLogger("gateway.run").info("gateway event") - - hermes_logging.flush_log_queue() - - gui_log = hermes_home / "logs" / "gui.log" - assert gui_log.exists() - content = gui_log.read_text(encoding="utf-8-sig") - assert "dashboard online" in content - assert "ws connected" in content - assert "gateway event" not in content - - -class TestSessionContext: - """set_session_context / clear_session_context + _SessionFilter.""" - - def test_session_tag_in_log_output(self, hermes_home): - """When session context is set, log lines include [session_id].""" - hermes_logging.setup_logging(hermes_home=hermes_home) - hermes_logging.set_session_context("abc123") - - test_logger = logging.getLogger("test.session_tag") - test_logger.info("tagged message") - - hermes_logging.flush_log_queue() - - agent_log = hermes_home / "logs" / "agent.log" - content = agent_log.read_text(encoding="utf-8-sig") - assert "[abc123]" in content - assert "tagged message" in content - - - - - - - -class TestComponentFilter: - """Unit tests for _ComponentFilter.""" - - def test_passes_matching_prefix(self): - f = hermes_logging._ComponentFilter(("gateway",)) - record = logging.LogRecord( - "gateway.run", logging.INFO, "", 0, "msg", (), None - ) - assert f.filter(record) is True - - - def test_blocks_non_matching(self): - f = hermes_logging._ComponentFilter(("gateway",)) - record = logging.LogRecord( - "tools.terminal_tool", logging.INFO, "", 0, "msg", (), None - ) - assert f.filter(record) is False - - - - - class TestSetupVerboseLogging: """setup_verbose_logging() adds a DEBUG-level console handler.""" @@ -745,41 +569,10 @@ class TestSafeStderr: assert isinstance(result, io.TextIOWrapper) assert result.encoding == "utf-8" assert result.errors == "replace" - - def test_handler_emits_unicode_without_crash(self, tmp_path): - """StreamHandler with _safe_stderr can emit Unicode messages.""" - import io - - # Create a stderr-like stream with ASCII encoding - class AsciiStream: - encoding = "ascii" - buffer = io.BytesIO() - - def write(self, s): - self.buffer.write(s.encode("ascii", errors="replace")) - - def flush(self): - pass - - # Without the fix, this would crash on cp949/ASCII stderr. - # With the wrapper, the em-dash is replaced with '?' - handler = logging.StreamHandler( - io.TextIOWrapper( - io.BytesIO(), - encoding="utf-8", - errors="replace", - ) - ) - handler.setFormatter(logging.Formatter("%(message)s")) - logger = logging.getLogger("_test_unicode") - logger.addHandler(handler) - logger.setLevel(logging.DEBUG) - try: - # Em-dash U+2014 — the exact character from the bug report - logger.info("Session hygiene: 400 messages — auto-compressing") - finally: - logger.removeHandler(handler) - + handler = logging.StreamHandler(result) + handler.handle(logging.LogRecord("unicode", logging.INFO, "", 0, "Session — 日本", (), None)) + handler.flush() + assert fake.buffer.getvalue() == "Session — 日本\n".encode("utf-8") class TestAsyncQueueLogging: """File logging runs through a QueueListener so emits never block on the diff --git a/tests/test_hermes_yaml.py b/tests/test_hermes_yaml.py index f4c727afc6..d2882d2861 100644 --- a/tests/test_hermes_yaml.py +++ b/tests/test_hermes_yaml.py @@ -6,19 +6,22 @@ from concurrent.futures import ThreadPoolExecutor import pytest import hermes_yaml as yaml +from utils import fast_safe_load -def test_safe_load_accepts_existing_config_boolean_spellings(): +@pytest.mark.parametrize("load", [yaml.safe_load, fast_safe_load]) +def test_safe_load_accepts_existing_config_boolean_spellings(load): document = "flags: [on, off, yes, no, true, false]\nquoted: ['off', 'yes']\n" expected = {"flags": [True, False, True, False, True, False], "quoted": ["off", "yes"]} for stream in (document, document.encode(), io.StringIO(document), io.BytesIO(document.encode())): - assert yaml.safe_load(stream) == expected - assert yaml.safe_load("") is None + assert load(stream) == expected + assert load("") is None -def test_safe_load_rejects_python_object_construction(): +@pytest.mark.parametrize("load", [yaml.safe_load, fast_safe_load]) +def test_safe_load_rejects_python_object_construction(load): with pytest.raises(yaml.YAMLError): - yaml.safe_load("!!python/object/apply:builtins.str ['must not construct']") + load("!!python/object/apply:builtins.str ['must not construct']") def test_safe_dump_preserves_data_and_readable_block_layout(): @@ -62,7 +65,7 @@ def test_roundtrip_preserves_comments_quotes_and_scalar_types(): def test_native_yaml11_scalars_and_duplicate_key_policy(): - for load in (yaml.safe_load, yaml.roundtrip_yaml().load): + for load in (yaml.safe_load, fast_safe_load, yaml.roundtrip_yaml().load): assert load("[y, n, Y, N, 'y', 'n']") == [True, False, True, False, "y", "n"] with pytest.raises(yaml.YAMLError): load("model: first\nmodel: second\n") diff --git a/tests/test_install_sh_node_deps_workspaces.py b/tests/test_install_sh_node_deps_workspaces.py deleted file mode 100644 index 23a1f0c3e4..0000000000 --- a/tests/test_install_sh_node_deps_workspaces.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Bootstrap does not ask npm to resolve an unrelated desktop workspace.""" -import os -from pathlib import Path -import subprocess - -ROOT = Path(__file__).resolve().parents[1] - - -def test_node_stage_does_not_invoke_ambient_npm(tmp_path): - sentinel = tmp_path / "npm-called" - script = f"""source "{(ROOT / 'scripts/install.sh').as_posix()}" --manifest -npm() {{ touch "{sentinel.as_posix()}"; return 99; }} -INSTALL_DIR="{tmp_path.as_posix()}" -stage_node_deps -""" - env = dict(os.environ, HOME=tmp_path.as_posix(), HERMES_HOME=tmp_path.as_posix()) - result = subprocess.run(["bash", "-c", script], env=env, capture_output=True, text=True, timeout=30) - assert result.returncode == 0, result.stderr - assert not sentinel.exists() diff --git a/tests/test_iron_proxy.py b/tests/test_iron_proxy.py index eb6ce58f4f..70146e9f02 100644 --- a/tests/test_iron_proxy.py +++ b/tests/test_iron_proxy.py @@ -59,7 +59,7 @@ def test_management_token_path_is_single_authority(hermes_home): assert not (hermes_home / "proxy").exists() token = ip.ensure_management_token() - assert token + assert token.startswith("hermes-mgmt-") p = ip._management_token_path() assert p.is_file() assert p.read_text(encoding="utf-8-sig").strip() == token @@ -71,15 +71,9 @@ def test_management_token_path_is_single_authority(hermes_home): assert ip._management_token_path().read_text(encoding="utf-8-sig").strip() == rotated - - - - # Unknown providers (no entry in _BEARER_PROVIDERS) are skipped, not warned. - - # --------------------------------------------------------------------------- # Config / mapping serialization # --------------------------------------------------------------------------- @@ -93,8 +87,6 @@ def _sample_mapping(env_name: str = "OPENROUTER_API_KEY") -> ip.TokenMapping: ) - - def test_build_proxy_config_custom_allowed_hosts(tmp_path): m = _sample_mapping("OPENAI_API_KEY") cfg = ip.build_proxy_config( @@ -114,27 +106,11 @@ def test_build_proxy_config_custom_allowed_hosts(tmp_path): # --------------------------------------------------------------------------- - - - - - - # --------------------------------------------------------------------------- # Bind policy (regression: must not bind 0.0.0.0) # --------------------------------------------------------------------------- - - - - - - - - - - # --------------------------------------------------------------------------- # audit_log file pre-creation (parameter still accepted; v0.39 doesn't # wire it into the binary config but ensure_audit_log() still creates @@ -163,61 +139,27 @@ def test_audit_log_kwarg_does_not_inject_audit_path_v039(tmp_path): ) - - - - - - def test_load_mappings_handles_corrupt_json(hermes_home): state = ip._proxy_state_dir() (state / "mappings.json").write_text("{not json", encoding="utf-8") assert ip.load_mappings() == [] - - # --------------------------------------------------------------------------- # Token-preservation on re-setup (regression: clobbered live sandboxes) # --------------------------------------------------------------------------- - - - - - - # --------------------------------------------------------------------------- # Uncovered provider detection (regression: signature-auth providers bypass) # --------------------------------------------------------------------------- - - - - - - - - # --------------------------------------------------------------------------- # Binary discovery + lazy install # --------------------------------------------------------------------------- - - - - - - - - - - - - # ── GPG release-signature verification (maxpetrusenko P1) ──────────────────── def test_verify_checksums_signature_skips_without_gpg(hermes_home, monkeypatch, tmp_path): @@ -228,37 +170,11 @@ def test_verify_checksums_signature_skips_without_gpg(hermes_home, monkeypatch, assert ip._verify_checksums_signature(tmp_path, cks) is False - - - - - - - - - - # --------------------------------------------------------------------------- # Subprocess lifecycle # --------------------------------------------------------------------------- - - - - - - - - - - - - - - - - def test_start_proxy_idempotent_when_already_running(hermes_home, monkeypatch): state = ip._proxy_state_dir() pid_file = state / "iron-proxy.pid" @@ -281,25 +197,11 @@ def test_start_proxy_idempotent_when_already_running(hermes_home, monkeypatch): # --------------------------------------------------------------------------- - - - - - - - - - - # --------------------------------------------------------------------------- # Platform asset name resolution # --------------------------------------------------------------------------- - - - - # --------------------------------------------------------------------------- # Subprocess env minimization (regression: host secrets leaked to proxy) # --------------------------------------------------------------------------- @@ -325,10 +227,6 @@ def test_subprocess_env_strips_unrelated_secrets(hermes_home, monkeypatch): assert env.get("OPENROUTER_API_KEY") == "sk-or-real" - - - - # --------------------------------------------------------------------------- # CA generation TOCTOU (regression: 0o600 only set AFTER copy) # --------------------------------------------------------------------------- @@ -401,60 +299,36 @@ def test_proxy_state_dir_is_0o700(hermes_home): assert mode == 0o700 - - # --------------------------------------------------------------------------- # Mappings clobber refused when corrupt (regression: silent 403s) # --------------------------------------------------------------------------- - - # --------------------------------------------------------------------------- # CA missing → enforce_on_docker semantics (regression: silent fail-open) # --------------------------------------------------------------------------- - - # --------------------------------------------------------------------------- # Docker env collision detection (regression: docker_env silently bypassed proxy) # --------------------------------------------------------------------------- - - # --------------------------------------------------------------------------- # v3 round: bridge-IP parser hardening (P1 #1) # --------------------------------------------------------------------------- - - - - - - # --------------------------------------------------------------------------- # v3: default deny-list adjacency (P2 IPv4-mapped-v6 + CGNAT) # --------------------------------------------------------------------------- - - # --------------------------------------------------------------------------- # Header-auth providers (x-api-key family) — match_headers + aliases # --------------------------------------------------------------------------- - - - - - - - - def test_mappings_roundtrip_preserves_headers_and_aliases(hermes_home): m = ip.TokenMapping( proxy_token=ip.mint_proxy_token("gemini"), @@ -469,30 +343,15 @@ def test_mappings_roundtrip_preserves_headers_and_aliases(hermes_home): assert loaded[0].alias_env_names == ("GOOGLE_API_KEY",) - - - - - - # --------------------------------------------------------------------------- # Management API (hot reload) # --------------------------------------------------------------------------- - - @pytest.mark.platforms("linux") -def test_ensure_management_token_persists_and_is_stable(hermes_home): - t1 = ip.ensure_management_token() - t2 = ip.ensure_management_token() - assert t1 == t2 - assert t1.startswith("hermes-mgmt-") - p = ip._proxy_state_dir() / "management.token" - assert p.exists() - assert (p.stat().st_mode & 0o777) == 0o600 - - +def test_management_token_is_private(hermes_home): + ip.ensure_management_token() + assert (ip._management_token_path().stat().st_mode & 0o777) == 0o600 def test_reload_proxy_refuses_when_not_running(hermes_home, monkeypatch): @@ -501,8 +360,6 @@ def test_reload_proxy_refuses_when_not_running(hermes_home, monkeypatch): ip.reload_proxy() - - def test_reload_proxy_posts_bearer_to_management_endpoint(hermes_home, monkeypatch): monkeypatch.setattr(ip, "_read_pid", lambda: 4242) monkeypatch.setattr(ip, "_pid_alive", lambda pid: True) @@ -582,17 +439,11 @@ def test_start_proxy_injects_management_key_env(hermes_home, monkeypatch): # --------------------------------------------------------------------------- - - - - # --------------------------------------------------------------------------- # v3: stop_proxy SIGKILL suppression on pid recycle (P3 #5 coverage gap) # --------------------------------------------------------------------------- - - # --------------------------------------------------------------------------- # v3: _reset_for_tests actually clears module state (P3 #1) # --------------------------------------------------------------------------- @@ -614,8 +465,6 @@ def test_reset_for_tests_clears_version_cache_and_nonce(): # --------------------------------------------------------------------------- - - # --------------------------------------------------------------------------- # v3: NODE_OPTIONS append-merge in docker env (arshkumarsingh #1) # --------------------------------------------------------------------------- @@ -665,8 +514,6 @@ def test_docker_egress_node_options_uses_sentinel(hermes_home, monkeypatch): # --------------------------------------------------------------------------- - - # --------------------------------------------------------------------------- # v3: persisted nonce roundtrip (stephenschoettler #3 cross-CLI defense) # --------------------------------------------------------------------------- @@ -683,18 +530,12 @@ def test_persisted_nonce_roundtrip(hermes_home, monkeypatch): assert ip._read_persisted_nonce() == "test-nonce-abc123" - - # --------------------------------------------------------------------------- # v4 round (GodsBoy follow-up): bind-host-aware liveness probes + # allow_env_fallback on the partial-secret path # --------------------------------------------------------------------------- - - - - def test_get_status_probes_configured_bind_host(hermes_home, monkeypatch): """get_status must probe the configured bind host (e.g. the docker bridge IP), not loopback unconditionally.""" diff --git a/tests/test_model_tools_async_bridge.py b/tests/test_model_tools_async_bridge.py index e2e070712d..e1d2226828 100644 --- a/tests/test_model_tools_async_bridge.py +++ b/tests/test_model_tools_async_bridge.py @@ -45,31 +45,6 @@ async def _create_and_return_transport(): # --------------------------------------------------------------------------- class TestRunAsyncLoopLifecycle: - """Verify _run_async() keeps the event loop alive after returning.""" - - def test_loop_not_closed_after_run_async(self): - """The loop used by _run_async must still be open after the call.""" - from model_tools import _run_async - - loop = _run_async(_get_current_loop()) - - assert not loop.is_closed(), ( - "_run_async() closed the event loop — cached async clients will " - "crash with 'Event loop is closed' on GC (issue #2104)" - ) - - def test_same_loop_reused_across_calls(self): - """Consecutive _run_async calls should reuse the same loop.""" - from model_tools import _run_async - - loop1 = _run_async(_get_current_loop()) - loop2 = _run_async(_get_current_loop()) - - assert loop1 is loop2, ( - "_run_async() created a new loop on the second call — cached " - "async clients from the first call would be orphaned" - ) - def test_cached_transport_survives_between_calls(self): """A transport/future created in call 1 must be valid in call 2.""" from model_tools import _run_async @@ -84,99 +59,25 @@ class TestRunAsyncLoopLifecycle: assert not loop.is_closed(), "Loop closed before second call" -class TestRunAsyncWorkerThread: - """Verify worker threads get persistent per-thread loops (delegate_task fix).""" +def test_concurrent_workers_reuse_distinct_loops(): + from concurrent.futures import ThreadPoolExecutor + from model_tools import _run_async - def test_worker_thread_loop_not_closed(self): - """A worker thread's loop must stay open after _run_async returns, - so cached httpx/AsyncOpenAI clients don't crash on GC.""" - from concurrent.futures import ThreadPoolExecutor - from model_tools import _run_async + main = _run_async(_get_current_loop()) + barrier = threading.Barrier(3, timeout=10) - def _run_on_worker(): - loop = _run_async(_get_current_loop()) - still_open = not loop.is_closed() - return loop, still_open + def worker(): + loop, future = _run_async(_create_and_return_transport()) + barrier.wait() + assert _run_async(_get_current_loop()) is loop + assert not loop.is_closed() and future.result() == "ok" + return loop, threading.get_ident() - with ThreadPoolExecutor(max_workers=1) as pool: - loop, still_open = pool.submit(_run_on_worker).result() - - assert still_open, ( - "Worker thread's event loop was closed after _run_async — " - "cached async clients will crash with 'Event loop is closed'" - ) - - def test_worker_thread_reuses_loop_across_calls(self): - """Multiple _run_async calls on the same worker thread should - reuse the same persistent loop (not create-and-destroy each time).""" - from concurrent.futures import ThreadPoolExecutor - from model_tools import _run_async - - def _run_twice_on_worker(): - loop1 = _run_async(_get_current_loop()) - loop2 = _run_async(_get_current_loop()) - return loop1, loop2 - - with ThreadPoolExecutor(max_workers=1) as pool: - loop1, loop2 = pool.submit(_run_twice_on_worker).result() - - assert loop1 is loop2, ( - "Worker thread created different loops for consecutive calls — " - "cached clients from the first call would be orphaned" - ) - assert not loop1.is_closed() - - def test_parallel_workers_get_separate_loops(self): - """Different worker threads must get their own loops to avoid - contention (the original reason for the worker-thread branch).""" - from concurrent.futures import ThreadPoolExecutor, as_completed - from model_tools import _run_async - - barrier = threading.Barrier(3, timeout=5) - - def _get_loop_id(): - # Use a barrier to force all 3 threads to be alive simultaneously, - # ensuring the ThreadPoolExecutor actually uses 3 distinct threads. - loop = _run_async(_get_current_loop()) - barrier.wait() - return id(loop), not loop.is_closed(), threading.current_thread().ident - - with ThreadPoolExecutor(max_workers=3) as pool: - futures = [pool.submit(_get_loop_id) for _ in range(3)] - results = [f.result() for f in as_completed(futures)] - - loop_ids = {r[0] for r in results} - thread_ids = {r[2] for r in results} - all_open = all(r[1] for r in results) - - assert all_open, "At least one worker thread's loop was closed" - # The barrier guarantees 3 distinct threads were used - assert len(thread_ids) == 3, f"Expected 3 threads, got {len(thread_ids)}" - # Each thread should have its own loop - assert len(loop_ids) == 3, ( - f"Expected 3 distinct loops for 3 parallel workers, " - f"got {len(loop_ids)} — workers may be contending on a shared loop" - ) - - def test_worker_loop_separate_from_main_loop(self): - """Worker thread loops must be different from the main thread's - persistent loop to avoid cross-thread contention.""" - from concurrent.futures import ThreadPoolExecutor - from model_tools import _run_async, _get_tool_loop - - main_loop = _get_tool_loop() - - def _get_worker_loop_id(): - loop = _run_async(_get_current_loop()) - return id(loop) - - with ThreadPoolExecutor(max_workers=1) as pool: - worker_loop_id = pool.submit(_get_worker_loop_id).result() - - assert worker_loop_id != id(main_loop), ( - "Worker thread used the main thread's loop — this would cause " - "cross-thread contention on the event loop" - ) + with ThreadPoolExecutor(max_workers=3) as pool: + futures = [pool.submit(worker) for _ in range(3)] + results = [future.result(timeout=15) for future in futures] + assert len({thread for _, thread in results}) == 3 + assert len({main, *(loop for loop, _ in results)}) == 4 class TestRunAsyncWithRunningLoop: @@ -191,10 +92,9 @@ class TestRunAsyncWithRunningLoop: async def _simple(): return 42 - result = await asyncio.get_event_loop().run_in_executor( - None, _run_async, _simple() - ) - assert result == 42 + assert _run_async(_simple()) == 42 + loop = _run_async(_get_current_loop()) + assert loop is not asyncio.get_running_loop() @pytest.mark.asyncio async def test_timeout_uses_nonblocking_executor_shutdown(self, monkeypatch): diff --git a/tests/test_node_resolution.py b/tests/test_node_resolution.py index 9864f9aa71..fdecb4437b 100644 --- a/tests/test_node_resolution.py +++ b/tests/test_node_resolution.py @@ -17,6 +17,27 @@ from pm.registry import get_package from pm.store import current_target, tree_digest +def _register_installed_tool(name, executable, companions=()): + executable = Path(executable) + version = subprocess.run([str(executable), "--version"], capture_output=True, text=True, + check=True, timeout=10).stdout.strip().removeprefix("v") + package, target, store = get_package(name), current_target(), paths.store_root() + entry = store / package.store_entry(version, target) + binary = package.binary(entry, target) + assert binary is not None + binary.parent.mkdir(parents=True) + binary.symlink_to(executable) + for companion in companions: + binary.with_name(companion.name).symlink_to(companion) + digest = hashlib.sha256(executable.read_bytes()).hexdigest() + lock = Lockfile(paths.lockfile_path()) + lock.set_pin(name, version, {target: {"url": executable.as_uri(), "sha256": digest}}) + lock.save() + Facts(paths.facts_path()).record(name, version, entry.name, package.env(entry, target), store, + target=target, artifacts=[digest], digest=tree_digest(entry)) + return binary + + @pytest.fixture def node_store(tmp_path, monkeypatch): node = shutil.which("node") @@ -32,24 +53,7 @@ def node_store(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_DISABLE_LAZY_INSTALLS", "1") lock_path = tmp_path / "lock.json" monkeypatch.setattr(paths, "lockfile_path", lambda: lock_path) - lock = Lockfile(lock_path) - target = current_target() - package = get_package("node") - version = subprocess.run( - [node, "--version"], capture_output=True, text=True, check=True, timeout=10, - ).stdout.strip().removeprefix("v") - entry = store / package.store_entry(version, target) - binary = package.binary(entry, target) - assert binary is not None - binary.parent.mkdir(parents=True) - binary.symlink_to(node) - digest = hashlib.sha256(Path(node).read_bytes()).hexdigest() - lock.set_pin("node", version, {target: {"url": Path(node).as_uri(), "sha256": digest}}) - lock.save() - Facts(paths.facts_path()).record( - "node", version, entry.name, package.env(entry, target), store, - target=target, artifacts=[digest], digest=tree_digest(entry), - ) + binary = _register_installed_tool("node", node) return home, binary, node @@ -133,27 +137,8 @@ def test_npm_and_npx_use_the_paired_pm_entry(node_store, monkeypatch): npx = Path(external).with_name("npx") if not npm.is_file() or not npx.is_file(): pytest.skip("requires already-installed npm and npx") - version = subprocess.run( - [str(npm), "--version"], capture_output=True, text=True, check=True, timeout=10, - ).stdout.strip() - package = get_package("npm") - target = current_target() - store = paths.store_root() - entry = store / package.store_entry(version, target) - binary = package.binary(entry, target) - assert binary is not None - binary.parent.mkdir(parents=True) - binary.symlink_to(npm) + binary = _register_installed_tool("npm", npm, [npx]) companion = binary.with_name("npx") - companion.symlink_to(npx) - lock = Lockfile(paths.lockfile_path()) - digest = hashlib.sha256(npm.read_bytes()).hexdigest() - lock.set_pin("npm", version, {target: {"url": npm.as_uri(), "sha256": digest}}) - lock.save() - Facts(paths.facts_path()).record( - "npm", version, entry.name, package.env(entry, target), store, - target=target, artifacts=[digest], digest=tree_digest(entry), - ) legacy = home / "node" / "bin" / "npm" legacy.parent.mkdir(parents=True) legacy.symlink_to(npm) @@ -172,7 +157,7 @@ def test_npm_and_npx_use_the_paired_pm_entry(node_store, monkeypatch): [resolved, "--version"], env=environment, capture_output=True, text=True, check=True, timeout=10, ) - assert result.stdout.strip() == version + assert result.stdout.strip() == Lockfile(paths.lockfile_path()).version("npm") assert paths.facts_path().read_bytes() == before companion.unlink() assert hermes_constants.find_node_executable("npx") is None diff --git a/tests/test_old_updater_additional_shims.py b/tests/test_old_updater_additional_shims.py index f7be4abe65..eede7d25df 100644 --- a/tests/test_old_updater_additional_shims.py +++ b/tests/test_old_updater_additional_shims.py @@ -53,22 +53,6 @@ def test_retired_constants_reload_handoffs_old_gateway_recovery(fresh_child, mon assert vars(hermes_constants) == before -@pytest.mark.parametrize("kwargs", [{}, {"timeout": 120, "capture_output": False}]) -def test_retired_pip_install_handoffs_before_reporting_success(kwargs, fresh_child): - from hermes_cli.tools_config import _pip_install - - with fresh_child.exits(): - result = _pip_install(["--quiet", "honcho-ai"], **kwargs) - pytest.fail(f"retired installer returned a result: {result}") - - -def test_retired_root_handoffs_before_inventing_portable_git_path(fresh_child): - from hermes_cli.update_cmd import get_default_hermes_root - - with fresh_child.exits(): - get_default_hermes_root() / "git" / "mingw64" / "libexec" / "git-core" / "git.exe" - - @pytest.mark.parametrize("prompt", [True, False]) def test_retired_ensure_reports_unavailable_without_installing(prompt, no_external_work): from tools.lazy_deps import ensure @@ -94,17 +78,6 @@ def test_live_dingtalk_dependencies_use_pm_not_retired_installer(monkeypatch): assert requested == ["dingtalk"] -@pytest.mark.parametrize("specs", [[], ["honcho-ai"]]) -def test_retired_install_specs_handoffs_before_reporting_success(specs, fresh_child): - from tools.lazy_deps import install_specs - - before = list(specs) - with fresh_child.exits(): - result = install_specs(specs, timeout=120) - pytest.fail(f"retired installer returned a result: {result}") - assert specs == before - - @pytest.mark.parametrize("handled", [False, True], ids=["unacknowledged", "child-completed"]) def test_historical_payload_survives_bridge_and_cleanup_requires_ack(handled, fresh_child, monkeypatch): from hermes_cli import update_receipt diff --git a/tests/test_old_updater_main_shims.py b/tests/test_old_updater_main_shims.py index 28fb7ddfb5..48428083f9 100644 --- a/tests/test_old_updater_main_shims.py +++ b/tests/test_old_updater_main_shims.py @@ -1,12 +1,10 @@ """Historical main imports must not restart pre-PM updater work after a swap.""" -from copy import deepcopy from pathlib import Path import pytest from tests.compat.old_updater_support import ( - fresh_child as fresh_child, no_external_work as no_external_work, ) @@ -61,69 +59,3 @@ def test_historical_marker_cleanup_preserves_path_and_is_idempotent(historical_m assert historical_main._clear_update_incomplete_marker() is None assert not marker.exists() assert historical_main._clear_update_incomplete_marker() is None - - -@pytest.mark.parametrize("cached", [False, True], ids=["cold-lookup", "cached-export"]) -@pytest.mark.parametrize( - "name,args,kwargs", - [ - ("_capture_active_lazy_features", (), {}), - ("_refresh_active_lazy_features", (), {}), - ("_refresh_active_lazy_features", (["browser"],), {}), - ("_refresh_active_lazy_features", (["uv", "pip"],), - {"env": {"VIRTUAL_ENV": "venv"}, "features": ["browser"]}), - ("_refresh_active_memory_provider_dependencies", (), {}), - ("_npm_lockfile_changed", (Path("checkout"),), {}), - ("_write_update_incomplete_marker", (), {}), - ("_reload_updated_runtime_modules", (), {}), - ], -) -def test_historical_main_lazy_hooks_handoff(name, args, kwargs, cached, historical_main, fresh_child, monkeypatch): - main = historical_main - before = deepcopy((args, kwargs)) - # Exercise PEP 562 even if an earlier test cached this export. Register the - # temporary slot with monkeypatch so it also restores an absent attribute. - monkeypatch.setitem(main.__dict__, name, None) - monkeypatch.delitem(main.__dict__, name) - if cached: - getattr(main, name) - with fresh_child.exits(): - getattr(main, name)(*args, **kwargs) - assert (args, kwargs) == before - - -@pytest.mark.parametrize( - "name,args,kwargs", - [ - ("_desktop_stamp_path", (), {}), - ("_expected_windows_pe_machines", (), {}), - ("_hermes_exe_shims", (Path("venv"),), {}), - ("_insert_python_pin", (["uv", "pip", "install", "-e", "."],), {}), - ("_interpreter_scripts_dir", (), {}), - ("_load_installable_optional_extras", (), {"group": "termux-all"}), - ("_parse_pe_machine", (Path("Hermes.exe"),), {}), - ("_quarantine_running_hermes_exe", (Path("venv"),), {"max_attempts": 1, "failed_out": []}), - ("_repair_broken_lazy_refresh_imports", (["uv", "pip"], ["certifi"]), {"env": {"VIRTUAL_ENV": "venv"}}), - ("_run_install_with_heartbeat", (["uv", "pip", "install", "-e", "."],), - {"env": {"VIRTUAL_ENV": "venv"}, "heartbeat_interval_seconds": 1}), - ("_run_package_only_install", (["uv", "pip", "install", "-e", "."],), {"env": {"VIRTUAL_ENV": "venv"}}), - ("_run_quarantined_install", (["uv", "pip", "install", "-e", "."],), - {"env": {"VIRTUAL_ENV": "venv"}, "scripts_dir": Path("venv"), "strict_quarantine": True}), - ("_run_quarantined_install", (["uv", "pip", "install", "-e", "."],), {}), - ("_run_with_idle_timeout", (["uv", "pip", "install", "-e", "."], Path("venv")), - {"env": {"VIRTUAL_ENV": "venv"}, "idle_timeout_seconds": 1, "indent": ""}), - ("_self", (), {}), - ("_verify_console_scripts_installed", (["uv", "pip"],), {"env": {"VIRTUAL_ENV": "venv"}}), - ("_verify_core_dependencies_installed", (["uv", "pip"],), {"env": {"VIRTUAL_ENV": "venv"}, "group": "all"}), - ("_web_ui_build_needed", (Path("web"),), {}), - ("_windows_native_machine", (), {}), - ("_windows_shim_in_process_chain", (), {}), - ], -) -def test_historical_main_entrypoints_handoff_without_install_or_success_fallback( - name, args, kwargs, historical_main, fresh_child, -): - before = deepcopy((args, kwargs)) - with fresh_child.exits(): - getattr(historical_main, name)(*args, **kwargs) - assert (args, kwargs) == before diff --git a/tests/test_old_updater_shims.py b/tests/test_old_updater_shims.py index eb3b1fa625..32f44bf239 100644 --- a/tests/test_old_updater_shims.py +++ b/tests/test_old_updater_shims.py @@ -17,8 +17,9 @@ from tests.compat.old_updater_support import ( @pytest.mark.parametrize( - "module,name,args,kwargs", + "module,name,args,kwargs,cached", [ + *((f"hermes_cli.{module}", name, args, kwargs, None) for module, name, args, kwargs in [ ("managed_uv", "ensure_uv", (), {}), ("managed_uv", "ensure_uv", (), {"repair_observer": lambda result: pytest.fail("repair observer ran")}), ("managed_uv", "update_managed_uv", (), {}), @@ -53,18 +54,70 @@ from tests.compat.old_updater_support import ( ("update_cmd", "_write_lazy_refresh_incomplete_marker", (), {}), ("update_cmd", "_reload_updated_runtime_modules", (), {}), ("update_cmd_maint", "_reload_updated_runtime_modules", (), {}), + ]), + *(("hermes_cli.main", name, args, kwargs, None) for name, args, kwargs in [ + ("_desktop_stamp_path", (), {}), + ("_expected_windows_pe_machines", (), {}), + ("_hermes_exe_shims", (Path("venv"),), {}), + ("_insert_python_pin", (["uv", "pip", "install", "-e", "."],), {}), + ("_interpreter_scripts_dir", (), {}), + ("_load_installable_optional_extras", (), {"group": "termux-all"}), + ("_parse_pe_machine", (Path("Hermes.exe"),), {}), + ("_quarantine_running_hermes_exe", (Path("venv"),), {"max_attempts": 1, "failed_out": []}), + ("_repair_broken_lazy_refresh_imports", (["uv", "pip"], ["certifi"]), {"env": {"VIRTUAL_ENV": "venv"}}), + ("_run_install_with_heartbeat", (["uv", "pip", "install", "-e", "."],), + {"env": {"VIRTUAL_ENV": "venv"}, "heartbeat_interval_seconds": 1}), + ("_run_package_only_install", (["uv", "pip", "install", "-e", "."],), {"env": {"VIRTUAL_ENV": "venv"}}), + ("_run_quarantined_install", (["uv", "pip", "install", "-e", "."],), + {"env": {"VIRTUAL_ENV": "venv"}, "scripts_dir": Path("venv"), "strict_quarantine": True}), + ("_run_quarantined_install", (["uv", "pip", "install", "-e", "."],), {}), + ("_run_with_idle_timeout", (["uv", "pip", "install", "-e", "."], Path("venv")), + {"env": {"VIRTUAL_ENV": "venv"}, "idle_timeout_seconds": 1, "indent": ""}), + ("_self", (), {}), + ("_verify_console_scripts_installed", (["uv", "pip"],), {"env": {"VIRTUAL_ENV": "venv"}}), + ("_verify_core_dependencies_installed", (["uv", "pip"],), {"env": {"VIRTUAL_ENV": "venv"}, "group": "all"}), + ("_web_ui_build_needed", (Path("web"),), {}), + ("_windows_native_machine", (), {}), + ("_windows_shim_in_process_chain", (), {}), + ]), + *(("hermes_cli.main", name, args, kwargs, cached) for name, args, kwargs in [ + ("_capture_active_lazy_features", (), {}), + ("_refresh_active_lazy_features", (), {}), + ("_refresh_active_lazy_features", (["browser"],), {}), + ("_refresh_active_lazy_features", (["uv", "pip"],), + {"env": {"VIRTUAL_ENV": "venv"}, "features": ["browser"]}), + ("_refresh_active_memory_provider_dependencies", (), {}), + ("_npm_lockfile_changed", (Path("checkout"),), {}), + ("_write_update_incomplete_marker", (), {}), + ("_reload_updated_runtime_modules", (), {}), + ] for cached in (False, True)), + ("hermes_cli.main_web_build", "_run_with_idle_timeout", (["npm", "ci"], Path("web")), {}, None), + ("hermes_cli.main_web_build", "_run_npm_install_deterministic", ("npm", Path("web")), {}, None), + ("hermes_cli.main_web_build", "_nixos_build_env", (), {}, None), + ("hermes_cli.main", "_reexec_dependency_sync_off_windows_shim", (), {}, None), + ("hermes_cli.update_cmd", "get_default_hermes_root", (), {}, None), + ("hermes_cli.tools_config", "_pip_install", (["--quiet", "honcho-ai"],), {}, None), + ("hermes_cli.tools_config", "_pip_install", (["--quiet", "honcho-ai"],), {"timeout": 120, "capture_output": False}, None), + ("tools.lazy_deps", "install_specs", ([],), {"timeout": 120}, None), + ("tools.lazy_deps", "install_specs", (["honcho-ai"],), {"timeout": 120}, None), ], ) -def test_retired_dependency_entrypoints_handoff_without_fallback(module, name, args, kwargs, fresh_child): +def test_retired_dependency_entrypoints_handoff_without_fallback(module, name, args, kwargs, cached, fresh_child, monkeypatch): # Some boundaries (notably psutil_android) hand off during import itself. # Resolve ordinary modules before the guard: their CLI startup is not a shim. - if module != "psutil_android": - importlib.import_module(f"hermes_cli.{module}") + if module != "hermes_cli.psutil_android": + resolved = importlib.import_module(module) + if cached is not None: + # Reset lazy exports even when earlier rows warmed the facade. + monkeypatch.setitem(resolved.__dict__, name, None) + monkeypatch.delitem(resolved.__dict__, name) + if cached: + getattr(resolved, name) # Exceptions have identity equality; preserve the caller's instance too. memo = {id(arg): arg for arg in args if isinstance(arg, BaseException)} before = deepcopy((args, kwargs), memo) with fresh_child.exits(): - getattr(importlib.import_module(f"hermes_cli.{module}"), name)(*args, **kwargs) + getattr(importlib.import_module(module), name)(*args, **kwargs) assert (args, kwargs) == before diff --git a/tests/test_os_marker_gating.py b/tests/test_os_marker_gating.py index 8e4dad72af..e87fa5d28a 100644 --- a/tests/test_os_marker_gating.py +++ b/tests/test_os_marker_gating.py @@ -1,63 +1,46 @@ -"""The collection guard against a test carrying two platforms() markers. - -A module-level gate stacked on a per-test gate ran on no host at all while -both the full-suite and marked lanes reported green — the silent coverage -loss the guard exists for. tests/conftest.py fails collection instead; this -pins that behaviour so the guard can't be dropped silently. -""" - -from __future__ import annotations +"""Run the real collection hook; skip-all and unregistered guards must fail.""" +import os +from pathlib import Path +import subprocess +import sys import pytest -from tests.conftest import _reject_contradictory_platform_marks - -class _FakeItem: - """Stands in for a collected item: the guard reads only these two.""" - - def __init__(self, nodeid: str, *marks) -> None: - self.nodeid = nodeid - self._marks = list(marks) - - def iter_markers(self, name=None): - if name is None: - return iter(self._marks) - return iter(m for m in self._marks if m.name == name) - - -def test_single_platforms_marker_is_accepted(): - items = [ - _FakeItem("t.py::test_linux", pytest.mark.platforms("linux")), - _FakeItem("t.py::test_win", pytest.mark.platforms("windows", arch="arm64")), - _FakeItem("t.py::test_not", pytest.mark.platforms("not macos")), - ] - _reject_contradictory_platform_marks(items) # must not raise - - -def test_unmarked_and_non_platform_markers_are_accepted(): - _reject_contradictory_platform_marks( - [ - _FakeItem("t.py::test_plain"), - _FakeItem("t.py::test_slow", pytest.mark.slow), - ] - ) - - -def test_two_platforms_markers_fail_collection(): - items = [ - _FakeItem("t.py::test_ok", pytest.mark.platforms("linux")), - _FakeItem( - "t.py::test_bad", - pytest.mark.platforms("linux"), - pytest.mark.platforms("windows"), - ), - ] - with pytest.raises(pytest.UsageError) as excinfo: - _reject_contradictory_platform_marks(items) - - message = str(excinfo.value) - assert "t.py::test_bad" in message - assert "at most one platforms()" in message - # The passing item must not be named — the error is a list of offenders. - assert "t.py::test_ok" not in message +@pytest.mark.parametrize("invalid,message", [("", ""), ("stacked", "at most one platforms()"), + ("keyword", "unexpected keyword")]) +def test_native_collection_witnesses(tmp_path, invalid, message): + root = Path(__file__).resolve().parents[1] + host = {"linux": "linux", "darwin": "macos", "win32": "windows"}[sys.platform] + (tmp_path / "conftest.py").write_text( + f"import sys; sys.path.insert(0, {str(root)!r})\n" + "from tests.conftest import pytest_configure, pytest_collection_modifyitems\n", encoding="utf-8") + suite = "import pytest\nfrom pathlib import Path\n" + for name, marker in [ + ("plain", ""), ("any", "@pytest.mark.platforms('any')"), + ("native", f"@pytest.mark.platforms({host!r})"), + ("foreign", f"@pytest.mark.platforms('not {host}')"), + ("arch", "@pytest.mark.platforms('any', arch='nonexistent-architecture')"), + ]: + suite += f"{marker}\ndef test_{name}():\n Path({name!r}).touch()\n" + (tmp_path / "test_valid.py").write_text(suite, encoding="utf-8") + if invalid: + marker = "@pytest.mark.platforms('any')" if invalid == "stacked" else "@pytest.mark.platforms('any', bogus=True)" + module_mark = "pytestmark = pytest.mark.platforms('any')\n" if invalid == "stacked" else "" + (tmp_path / "test_bad.py").write_text( + f"import pytest\n{module_mark}{marker}\ndef test_bad():\n raise AssertionError('must reject collection')\n", + encoding="utf-8") + result = subprocess.run([sys.executable, "-m", "pytest", "-q", "-o", "addopts=", str(tmp_path)], + cwd=tmp_path, env={**os.environ, "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1"}, + capture_output=True, text=True, timeout=30) + output = result.stdout + result.stderr + witnesses = {p.name for p in tmp_path.iterdir() if p.name in {"plain", "any", "native", "foreign", "arch"}} + if invalid: + assert result.returncode == 4, output + assert message in output and "test_bad.py::test_bad" in output + assert "test_valid.py::" not in output + assert not witnesses + else: + assert result.returncode == 0, output + assert "3 passed, 2 skipped" in output + assert witnesses == {"plain", "any", "native"} diff --git a/tests/test_packaging_build_guard.py b/tests/test_packaging_build_guard.py index c4bc3a80bf..9601942d2d 100644 --- a/tests/test_packaging_build_guard.py +++ b/tests/test_packaging_build_guard.py @@ -84,6 +84,19 @@ def test_artifact_build_allows_explicit_nix_package_build_marker(kind, artifact_ if kind == "wheel": with zipfile.ZipFile(artifacts[0]) as wheel: shipped = set(wheel.namelist()) + assert {"pm/__init__.py", "pm/lock.json", "pm/artifact-mirror.json"} <= shipped + installed = tmp_path / "installed" + with zipfile.ZipFile(artifacts[0]) as wheel: + wheel.extractall(installed) + check = subprocess.run( + [sys.executable, "-I", "-S", "-c", + "import sys; sys.path.insert(0, sys.argv[1]); " + "from pm.artifact_mirror import mirror_url; print(mirror_url('0' * 64))", str(installed)], + cwd=tmp_path, text=True, capture_output=True, timeout=30, + ) + assert check.returncode == 0, check.stderr + from pm.artifact_mirror import mirror_url + assert check.stdout.strip() == mirror_url("0" * 64) else: with tarfile.open(artifacts[0]) as sdist: shipped = { @@ -94,38 +107,3 @@ def test_artifact_build_allows_explicit_nix_package_build_marker(kind, artifact_ missing = sorted(expected - shipped) assert not missing, f"{kind} omits bundled plugin manifests: {missing}" - - -def test_wheel_ships_pm_package_and_lock_json(tmp_path): - """The pm/ package manager must survive a sealed wheel build. - - pm is a flat package listed in [tool.setuptools.packages.find] include, - and pm/lock.json is its runtime pin table (uv/python/tool versions + - sha256s) declared via [tool.setuptools.package-data]. If either drops - out of the wheel, installed Hermes has no package manager at all -- - exercise the real PEP 517 build path rather than reading TOML source. - """ - result = _build_artifact("wheel", tmp_path, nix_build=True) - - assert result.returncode == 0, result.stderr - artifacts = list(tmp_path.glob("hermes_agent-*.whl")) - assert artifacts - - with zipfile.ZipFile(artifacts[0]) as wheel: - shipped = set(wheel.namelist()) - - missing = sorted({"pm/__init__.py", "pm/lock.json", "pm/artifact-mirror.json"} - shipped) - assert not missing, f"wheel omits pm package files: {missing}" - - installed = tmp_path / "installed" - with zipfile.ZipFile(artifacts[0]) as wheel: - wheel.extractall(installed) - check = subprocess.run( - [sys.executable, "-I", "-S", "-c", - "import sys; sys.path.insert(0, sys.argv[1]); " - "from pm.artifact_mirror import mirror_url; print(mirror_url('0' * 64))", str(installed)], - cwd=tmp_path, text=True, capture_output=True, timeout=30, - ) - assert check.returncode == 0, check.stderr - from pm.artifact_mirror import mirror_url - assert check.stdout.strip() == mirror_url("0" * 64) diff --git a/tests/test_packaging_metadata.py b/tests/test_packaging_metadata.py index d603f5b693..a59b8578b9 100644 --- a/tests/test_packaging_metadata.py +++ b/tests/test_packaging_metadata.py @@ -1,148 +1,37 @@ -import ast -import re +"""Independent core/optional dependency and reviewed CVE policies.""" import tomllib from pathlib import Path -import pytest - +from packaging.requirements import Requirement +from packaging.version import Version REPO_ROOT = Path(__file__).resolve().parents[1] -def _distribution_name(requirement: str) -> str: - """Extract the PEP 508 distribution name from a requirement string. - - Robust to markers (``; python_version < '3.12'``), direct references - (``name @ https://...``), extras (``name[extra]``) and every version - operator (``==``, ``>=``, ``<=``, ``~=``, ``!=``, ``<``, ``>``), so a - future dep declared with any valid specifier shape doesn't silently - mis-parse here. - """ - spec = requirement.split(";", 1)[0] # drop environment markers - spec = spec.split("@", 1)[0] # drop direct-reference URLs - spec = spec.split("[", 1)[0] # drop extras - spec = re.split(r"[=<>!~]", spec, maxsplit=1)[0] # drop any version operator - return spec.strip().lower() +def test_core_and_optional_speech_dependencies(): + project = tomllib.loads((REPO_ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"] + core = {Requirement(dep).name for dep in project["dependencies"]} + assert "packaging" in core # Runtime code imports it directly, not transitively. + assert "faster-whisper" not in core + assert "faster-whisper" in { + Requirement(dep).name for dep in project["optional-dependencies"]["stt-whisper"] + } -def test_packaging_declared_as_core_dependency(): - """Regression for #40503. - - ``packaging`` is imported directly on three production paths - (plugins/memory/hindsight/__init__.py, pm/extras.py, - hermes_cli/main.py) yet was undeclared, so it only reached users - transitively. The slim Docker image shipped without it, silently - disabling Hindsight append-mode and version-constraint checks. It must - be a declared core dependency so PM includes it in dependency generations. - """ - data = tomllib.loads((REPO_ROOT / "pyproject.toml").read_text(encoding="utf-8")) - core = data["project"]["dependencies"] - names = {_distribution_name(dep) for dep in core} - assert "packaging" in names, ( - "packaging is imported on production paths (hindsight version compare, " - "version constraints, requirement parsing) and must be a " - "declared core dependency, not a transitive — see #40503" - ) - - -def test_faster_whisper_is_not_a_base_dependency(): - data = tomllib.loads((REPO_ROOT / "pyproject.toml").read_text(encoding="utf-8")) - deps = data["project"]["dependencies"] - - assert not any(dep.startswith("faster-whisper") for dep in deps) - - stt_extra = data["project"]["optional-dependencies"]["stt-whisper"] - assert any(dep.startswith("faster-whisper") for dep in stt_extra) - - -# Minimum non-vulnerable Starlette: CVE-2026-48710 ("BadHost") was fixed in -# 1.0.1. Anything below that lets a malformed Host header desync -# ``request.url.path`` from the dispatched ASGI path, bypassing path-based -# authz in middleware/endpoints that gate on ``request.url``. Starlette is a -# transitive dep (fastapi in [web]; sse-starlette/mcp in [mcp]/[computer-use]/ -# [dev]) so we pin it directly in every extra that exposes a server surface and -# enforce the floor in both pyproject and the committed lockfile. -_STARLETTE_CVE_FLOOR = (1, 0, 1) -_UPDATE_DOWNGRADE_GUARD_FLOORS = { - # `hermes update` reinstalls exact pins from pyproject/uv.lock. These - # reviewed CVE pins must not slide back to stale versions that downgrade - # already-patched user environments. - "cryptography": (50, 0, 0), - "starlette": (1, 3, 1), - "python-multipart": (0, 0, 32), -} - - -def _version_tuple(spec: str) -> tuple[int, ...]: - # "1.0.1" -> (1, 0, 1); tolerant of pre/post suffixes by truncating. - head = spec.split("+", 1)[0] - parts = [] - for chunk in head.split("."): - digits = "".join(ch for ch in chunk if ch.isdigit()) - if not digits: - break - parts.append(int(digits)) - return tuple(parts) - - -def test_starlette_pinned_above_cve_2026_48710_floor_in_pyproject(): - """Every extra that declares Starlette must pin a patched (>=1.0.1) version. - - Regression guard for #35067 / CVE-2026-48710. A future edit that drops the - pin (re-exposing the unbounded transitive ``starlette>=0.27`` from mcp / - ``>=0.40.0`` from fastapi) or pins a pre-1.0.1 version fails here instead of - shipping a Host-header auth-bypass to dashboard / MCP-HTTP users. - """ - data = tomllib.loads((REPO_ROOT / "pyproject.toml").read_text(encoding="utf-8")) - extras = data["project"]["optional-dependencies"] - - found = {} - for extra, specs in extras.items(): - for spec in specs: - name = spec.split("==", 1)[0].split(">", 1)[0].split("<", 1)[0].split("[", 1)[0].strip() - if name.lower() == "starlette": - assert "==" in spec, f"[{extra}] must exact-pin starlette, got {spec!r}" - ver = spec.split("==", 1)[1].split(";", 1)[0].strip() - found[extra] = ver - - # The four server-surface extras must each carry the direct pin. - for extra in ("web", "mcp", "computer-use", "dev"): - assert extra in found, ( - f"[{extra}] no longer pins starlette directly — CVE-2026-48710 " - f"regression risk (mcp/fastapi pull it transitively with no upper bound)" - ) - - for extra, ver in found.items(): - assert _version_tuple(ver) >= _STARLETTE_CVE_FLOOR, ( - f"[{extra}] pins starlette=={ver}, below the CVE-2026-48710 fix " - f"floor {'.'.join(map(str, _STARLETTE_CVE_FLOOR))}" - ) - - -def test_locked_starlette_is_not_vulnerable_to_cve_2026_48710(): - """The committed uv.lock must resolve starlette to a patched version. - - pyproject pins protect the declared extras, but the lockfile is what - hash-verified installs (``uv sync --locked``) actually pull. Assert the - resolved version is >= the CVE-2026-48710 fix floor so a stale-lock - regression can't ship a vulnerable Starlette to users. - """ - lock = (REPO_ROOT / "uv.lock").read_text(encoding="utf-8") - versions = [] - in_starlette = False - for line in lock.splitlines(): - if line.startswith("[[package]]"): - in_starlette = False - elif line.strip() == 'name = "starlette"': - in_starlette = True - elif in_starlette and line.startswith("version = "): - versions.append(line.split("=", 1)[1].strip().strip('"')) - in_starlette = False - - assert versions, "starlette not found in uv.lock" - for ver in versions: - assert _version_tuple(ver) >= _STARLETTE_CVE_FLOOR, ( - f"uv.lock resolves starlette=={ver}, below the CVE-2026-48710 fix " - f"floor {'.'.join(map(str, _STARLETTE_CVE_FLOOR))} — regenerate the " - f"lockfile after bumping the pin" - ) +def test_starlette_server_pins_and_lock_exclude_cve_2026_48710(): + # BadHost's reviewed fixed boundary is independent of today's exact pin. + floor = Version("1.0.1") + metadata = tomllib.loads((REPO_ROOT / "pyproject.toml").read_text(encoding="utf-8")) + lock = tomllib.loads((REPO_ROOT / "uv.lock").read_text(encoding="utf-8")) + found = set() + for extra, specs in metadata["project"]["optional-dependencies"].items(): + for requirement in map(Requirement, specs): + if requirement.name != "starlette": + continue + pins = list(requirement.specifier) + assert len(pins) == 1 and pins[0].operator == "==", (extra, requirement) + assert Version(pins[0].version) >= floor, (extra, requirement) + found.add(extra) + assert {"web", "mcp", "computer-use", "dev"} <= found + versions = [Version(row["version"]) for row in lock["package"] if row["name"] == "starlette"] + assert versions and all(version >= floor for version in versions) diff --git a/tests/test_platforms_marker.py b/tests/test_platforms_marker.py index 6d7af9e408..6df5309527 100644 --- a/tests/test_platforms_marker.py +++ b/tests/test_platforms_marker.py @@ -1,163 +1,43 @@ -"""The composable ``platforms`` marker: spec evaluation + collection gating. - -Behavior-contract tests for the gate introduced alongside the fixed -platforms("linux")/platforms("macos")/platforms("windows") trio: any-of semantics, negation, -POSIX grouping, arch filters, and the hard errors on unknown specs and -stray keyword arguments. -""" - -from __future__ import annotations +"""Native platform policy table; collection wiring lives in test_os_marker_gating.""" +import sys import pytest - -from tests.conftest import _host_matches_platforms +from tests.conftest import _host_matches_platforms, _platform_machine -class TestSpecEvaluation: - """Pure evaluation: takes the host as data, no host faking.""" - - # (specs, host_platform, expected_ok) - CASES = [ - (("linux",), "linux", True), - (("linux",), "win32", False), - (("macos",), "darwin", True), - (("windows",), "win32", True), - (("windows",), "linux", False), - (("posix",), "linux", True), - (("posix",), "darwin", True), - (("posix",), "win32", False), - (("not macos",), "linux", True), - (("not macos",), "darwin", False), - (("not windows",), "win32", False), - (("not windows",), "linux", True), - (("linux", "win32host-mismatch"), "win32", False), # unknown spec never matches - (("any",), "linux", True), - ((), "linux", True), # no specs = documentation form, matches all - ] - - @pytest.mark.parametrize(("specs", "host", "expected"), CASES) - def test_spec_matrix(self, specs, host, expected, monkeypatch): - monkeypatch.setattr("tests.conftest.sys.platform", host) - ok, _reason = _host_matches_platforms(specs) - assert ok is expected - - def test_unknown_spec_is_reported_not_matched(self, monkeypatch): - monkeypatch.setattr("tests.conftest.sys.platform", "linux") - ok, reason = _host_matches_platforms(("amiga",)) - assert ok is False +@pytest.mark.parametrize("specs,hosts", [ + (("linux",), {"linux"}), (("macos",), {"darwin"}), + (("windows",), {"win32"}), (("WINDOWS",), {"win32"}), + (("posix",), {"linux", "darwin"}), (("not macos",), {"linux", "win32"}), + (("not windows",), {"linux", "darwin"}), (("linux", "macos"), {"linux", "darwin"}), + (("any",), {"linux", "darwin", "win32"}), ((), {"linux", "darwin", "win32"}), + (("linux", "amiga"), {"linux"}), (("linux", "win32host-mismatch"), {"linux"}), + (("amiga",), set()), (("not amiga",), set()), +]) +def test_native_spec_table(specs, hosts): + ok, reason = _host_matches_platforms(specs) + assert ok is (sys.platform in hosts), reason + if specs in (("amiga",), ("not amiga",)): assert "unknown spec" in reason - def test_negation_of_unknown_spec_is_rejected(self, monkeypatch): - monkeypatch.setattr("tests.conftest.sys.platform", "linux") - ok, reason = _host_matches_platforms(("not amiga",)) - assert ok is False - assert "unknown spec" in reason - def test_case_insensitive_specs(self, monkeypatch): - monkeypatch.setattr("tests.conftest.sys.platform", "win32") - ok, _ = _host_matches_platforms(("WINDOWS",)) - assert ok is True +@pytest.mark.parametrize("negate", [False, True]) +def test_native_arch_filter(negate): + machine = _platform_machine() + for arch, matches in ((machine, True), ("nonexistent-architecture", False)): + ok, reason = _host_matches_platforms(("any",), arch=arch, arch_negate=negate) + assert ok is (matches != negate) + if not ok: + assert machine in reason + if machine == "arm64": + assert _host_matches_platforms(("any",), arch="aarch64", arch_negate=negate)[0] is not negate -class TestArchFilter: - @pytest.mark.parametrize( - ("arch", "machine", "negate", "expected"), - [ - ("arm64", "arm64", False, True), - ("arm64", "x86_64", False, False), - ("aarch64", "arm64", False, True), # alias - ("arm64", "arm64", True, False), - ("arm64", "x86_64", True, True), - ], - ) - def test_arch_matrix(self, arch, machine, negate, expected, monkeypatch): - monkeypatch.setattr("tests.conftest.sys.platform", "win32") - monkeypatch.setattr("tests.conftest._platform_machine", lambda: machine) - ok, reason = _host_matches_platforms(("windows",), arch=arch, arch_negate=negate) - assert ok is expected, reason - - def test_arch_reason_names_the_machine(self, monkeypatch): - monkeypatch.setattr("tests.conftest.sys.platform", "win32") - monkeypatch.setattr("tests.conftest._platform_machine", lambda: "x86_64") - ok, reason = _host_matches_platforms(("windows",), arch="arm64") - assert ok is False - assert "x86_64" in reason - - -class TestAnyOfSemantics: - def test_multiple_specs_are_any_of(self, monkeypatch): - monkeypatch.setattr("tests.conftest.sys.platform", "darwin") - ok, _ = _host_matches_platforms(("linux", "macos")) - assert ok is True - - def test_first_matching_spec_wins_over_later_unknown(self, monkeypatch): - # any-of: a matching spec satisfies the gate even if a later spec - # is garbage — unknown specs only matter when nothing matched. - monkeypatch.setattr("tests.conftest.sys.platform", "linux") - ok, _ = _host_matches_platforms(("linux", "amiga")) - assert ok is True - - -class TestCollectionGating: - """The marker must actually skip/gate collected items on this host.""" - - @pytest.mark.platforms("not " + __import__("sys").platform.split("_")[0]) - def test_never_runs_on_this_host_shape(self): - # The spec is built to exclude whatever this host is (linux → "not - # linux", win32 → "not windows"); if it RUNS the gate is broken. - raise AssertionError("platforms() gate failed to skip this host") - - @pytest.mark.platforms("any") - def test_any_spec_runs_everywhere(self): - assert True - - @pytest.mark.skipif( - __import__("sys").platform == "win32", - reason="linux-host assertion; inverted on the linux lane below", - ) - @pytest.mark.platforms("linux") - def test_runs_on_linux(self): - assert True - - -class TestHardErrors: - def test_stray_kwarg_is_a_usage_error(self): - # The gate raises UsageError (surfaced by pytest as a collection - # error) for keyword arguments it does not understand — evaluated - # directly because the raise happens inside the project conftest's - # collection hook. - import pytest as _pytest - - from tests.conftest import _platforms_gate_reason - - class _Item: - nodeid = "tests/x.py::test_x" - - @staticmethod - def iter_markers(name): - yield _pytest.mark.platforms("linux", bogus=True).mark - - with _pytest.raises(_pytest.UsageError, match="unexpected keyword"): - _platforms_gate_reason(_Item) - - -class TestMachineAliases: - """_platform_machine normalizes the raw platform.machine() spellings.""" - - @pytest.mark.parametrize( - ("raw", "normalized"), - [ - ("AMD64", "x86_64"), - ("x86", "x86_64"), - ("aarch64", "arm64"), - ("arm64", "arm64"), - ("x86_64", "x86_64"), - ], - ) - def test_alias_matrix(self, raw, normalized, monkeypatch): - import platform as _platform - - monkeypatch.setattr(_platform, "machine", lambda: raw) - from tests.conftest import _platform_machine - - assert _platform_machine() == normalized +@pytest.mark.parametrize("raw,normalized", [ + ("AMD64", "x86_64"), ("x86", "x86_64"), ("aarch64", "arm64"), + ("arm64", "arm64"), ("x86_64", "x86_64"), +]) +def test_machine_alias_normalization(raw, normalized, monkeypatch): + # Exercise normalization data, never alter sys.platform or interpreter OS behavior. + monkeypatch.setattr("platform.machine", lambda: raw) + assert _platform_machine() == normalized diff --git a/tests/test_plugin_historic_compat.py b/tests/test_plugin_historic_compat.py index f698133490..97a20d02b5 100644 --- a/tests/test_plugin_historic_compat.py +++ b/tests/test_plugin_historic_compat.py @@ -33,7 +33,6 @@ def _plugin(home, name, *, dependencies=True): def test_candidate_member_dirs_preserves_proposed_home_order_and_extras(isolated_home, monkeypatch, active): from hermes_cli import plugins_admission - assert callable(getattr(plugins_admission, "candidate_member_dirs", None)) home = isolated_home profile = home / "profiles" / "coder" profile.mkdir(parents=True) @@ -63,16 +62,6 @@ def test_candidate_member_dirs_preserves_proposed_home_order_and_extras(isolated assert {p: p.read_bytes() for p in home.rglob("*") if p.is_file()} == before -def test_candidate_member_dirs_historical_defaults_do_not_replace_active_selection(isolated_home): - from hermes_cli import plugins_admission - - assert callable(getattr(plugins_admission, "candidate_member_dirs", None)) - old = _plugin(isolated_home, "old") - _plugin(isolated_home, "new") - (isolated_home / "config.yaml").write_text("plugins:\n enabled: [old]\n", encoding="utf-8") - assert plugins_admission.candidate_member_dirs(["new"]) == [old] - - @pytest.fixture def publication(isolated_home, tmp_path): from hermes_cli.runtime_paths import install_state_dir @@ -112,7 +101,6 @@ sys.path.insert(0, sys.argv[1]) for module in ('pm', 'hermes_cli.config', 'hermes_cli.plugins_cmd'): sys.modules[module] = None from hermes_cli import plugins_transaction -assert callable(getattr(plugins_transaction, 'recover_plugin_publication', None)) row = json.loads(sys.stdin.read()) plugins_transaction.recover_plugin_publication( project=Path(sys.argv[2]), row=row, journal=Path(sys.argv[3]), @@ -149,7 +137,6 @@ def test_old_publication_rolls_back_a_first_install(publication): from hermes_cli import plugins_transaction import shutil - assert callable(getattr(plugins_transaction, "recover_plugin_publication", None)) project, row, journal, _ = publication shutil.rmtree(row["backup"]) row.update(target_existed=False, metadata_before=None) @@ -163,7 +150,6 @@ def test_old_publication_rolls_back_a_first_install(publication): def test_old_publication_refuses_unsafe_or_changed_state_without_writes(publication, tmp_path, invalid): from hermes_cli import plugins_transaction - assert callable(getattr(plugins_transaction, "recover_plugin_publication", None)) project, row, journal, _ = publication if invalid == "edited-metadata": Path(row["metadata"]).write_bytes(b"independent user edit") diff --git a/tests/test_pm_bionic.py b/tests/test_pm_bionic.py index 2e90ee4459..318ccbeeac 100644 --- a/tests/test_pm_bionic.py +++ b/tests/test_pm_bionic.py @@ -9,9 +9,6 @@ this host's installed facts. from __future__ import annotations import json -import tarfile -import io -import zipfile from pathlib import Path import pytest @@ -19,15 +16,6 @@ import pytest REPO_ROOT = Path(__file__).resolve().parents[1] -def _pm(): - import sys - - sys.path.insert(0, str(REPO_ROOT)) - import pm - - return pm - - @pytest.fixture(scope="module") def lock(): return json.loads((REPO_ROOT / "pm" / "lock.json").read_text(encoding="utf-8")) @@ -91,130 +79,70 @@ def test_uv_bionic_row_matches_supplier(lock): """The uv bionic row is an explicit pin of the termux-main pool .deb; the row and Uv.fetch_url(bionic arm) must agree.""" _assert_pinned_bionic_row(lock, "uv", r"/u/uv/uv_(?P[0-9.]+)_aarch64\.deb$") -def _build_fake_deb(path: Path, control: dict[str, str], files: dict[str, bytes]) -> None: - def ar_member(name: str, data: bytes) -> bytes: - hdr = ( - name.ljust(16).encode() - + b"0".ljust(12) - + b"0".ljust(6) - + b"0".ljust(6) - + b"100644".ljust(8) - + str(len(data)).encode().ljust(10) - + b"`\n" - ) - pad = b"\n" if len(data) % 2 else b"" - return hdr + data + pad - - ctrl_buf = io.BytesIO() - with tarfile.open(fileobj=ctrl_buf, mode="w:gz") as tf: - body = "".join(f"{k}: {v}\n" for k, v in control.items()).encode() - info = tarfile.TarInfo("control") - info.size = len(body) - tf.addfile(info, io.BytesIO(body)) - data_buf = io.BytesIO() - with tarfile.open(fileobj=data_buf, mode="w:gz") as tf: - for name, content in files.items(): - info = tarfile.TarInfo(name) - info.size = len(content) - tf.addfile(info, io.BytesIO(content)) - path.write_bytes( - b"!\n" - + ar_member("debian-binary", b"2.0\n") - + ar_member("control.tar.gz", ctrl_buf.getvalue()) - + ar_member("data.tar.gz", data_buf.getvalue()) - ) - - -def test_debpackage_unpack_hardened(tmp_path: Path): - """DebPackage.unpack extracts data members and refuses traversal.""" - from pm.package import DebPackage - - class _P(DebPackage): - name = "test-deb" - - deb = tmp_path / "test.deb" - _build_fake_deb( - deb, - {"Package": "test-deb", "Version": "1.0"}, - {"data/data/com.termux/files/usr/bin/tool": b"\x7fELF"}, - ) - staged = tmp_path / "staged" - staged.mkdir() - _P().unpack(deb, staged, "linux-arm64-bionic") - assert (staged / "data/data/com.termux/files/usr/bin/tool").read_bytes() == b"\x7fELF" - - # traversal member must be refused - evil = tmp_path / "evil.deb" - _build_fake_deb( - evil, {"Package": "evil", "Version": "1.0"}, {"../escape": b"x"} - ) - with pytest.raises(Exception): - _P().unpack(evil, tmp_path / "staged2", "linux-arm64-bionic") - - -@pytest.mark.parametrize("name", ["python", "uv", "node"]) -def test_bionic_verify_is_file_evidence(tmp_path: Path, monkeypatch, name): - """bionic verify never executes the staged binary; presence is the - contract (the digest already proved the bytes).""" - from pm.registry import get_package - - def refuse_exec(*args, **kwargs): - pytest.fail(f"bionic verification attempted execution: {args}") - - monkeypatch.setattr("pm.packages.subprocess.run", refuse_exec) - package = get_package(name) - bin_rel = Path(package.prefix_rel) / package.main_rel("linux-arm64-bionic") - entry = tmp_path / "entry" - (entry / bin_rel).parent.mkdir(parents=True) - (entry / bin_rel).write_bytes(b"bionic-elf-bytes") - assert package.verify(entry, "linux-arm64-bionic") == "" - empty = tmp_path / "empty" - empty.mkdir() - assert "missing" in package.verify(empty, "linux-arm64-bionic") - - -def test_bionic_binary_and_env_contract(tmp_path: Path): - """Bionic binaries retain their staged paths, but only on_path packages - expose them in the environment; internal uv stays private to PM.""" - from pm.registry import get_package - - for name in ("uv", "python", "node"): - pkg = get_package(name) - entry = tmp_path / name - main = entry / pkg.prefix_rel / pkg.main_rel("linux-arm64-bionic") - main.parent.mkdir(parents=True) - main.write_bytes(b"bionic-elf") - - binary = pkg.binary(entry, "linux-arm64-bionic") - assert binary == main, f"{name}.binary() on bionic: {binary}" - - env = pkg.env(entry, "linux-arm64-bionic") - expected_path = [str(main.parent)] if pkg.on_path else None - assert env.get("PATH") == expected_path, ( - f"{name}.env() on bionic does not follow its on_path declaration" - ) - if pkg.internal: - assert "PATH" not in env, f"internal {name} must not leak into public PATH" - - -def test_stage_only_does_not_record_host_facts(tmp_path, monkeypatch): - """stage_only publishes the entry but must not touch this machine's - installed facts -- the fact slot belongs to the HOST target.""" - pm = _pm() +@pytest.mark.parametrize("name,main,on_path", [ + ("python", None, True), ("uv", "bin/uv", False), ("node", "bin/node", True), +]) +def test_registered_bionic_stage_preserves_host_facts(tmp_path, monkeypatch, lock, name, main, on_path): + import hashlib + from pm import paths from pm.ensure import stage_only - from pm.lock import Facts - from pm.paths import facts_path + from pm.lock import Lockfile + from pm.package import InstallError + from pm.registry import get_package + from pm.store import Store + from tests.termux_fixtures import build_deb - def snapshot() -> dict: - path = facts_path() - if not path.is_file(): - return {} - return Facts(path)._packages + target = "linux-arm64-bionic" + monkeypatch.setenv("HERMES_RUNTIME_DIR", str(tmp_path / "runtime")) + monkeypatch.setattr(paths, "lockfile_path", lambda: tmp_path / "lock.json") + facts = paths.facts_path() + facts.parent.mkdir(parents=True, exist_ok=True) + facts.write_bytes(b'{"sentinel": "host state must not change"}') + before = facts.read_bytes() + if main is None: + # Archive filename is the independent supplier authority, not main_rel(). + version = lock["packages"]["python"]["artifacts"][target]["url"].rsplit("/", 1)[1].split("_")[1] + main = "bin/python" + ".".join(version.split(".")[:2]) + relative = "data/data/com.termux/files/usr/" + main + package = get_package(name) + store = Store(paths.store_root()) - before = snapshot() - entry = stage_only("termux-docker", "linux-arm64-bionic") - after = snapshot() - assert before == after - # termux-docker is a pin_only package: stage_only returns the would-be - # entry path (store root + entry name) without staging bytes. - assert "termux-docker" in str(entry) + def archive(files): + deb = tmp_path / "fixture.deb" + build_deb(deb, {"Package": name, "Version": "1.0"}, files) + digest = hashlib.sha256(deb.read_bytes()).hexdigest() + lock = Lockfile(paths.lockfile_path()) + lock.set_pin(name, "1.0", {target: {"url": "https://example.test/fixture.deb", "sha256": digest}}) + lock.save() + cached = store.entry(f"fetch-{digest}") + cached.mkdir(parents=True) + (cached / "fixture.deb").write_bytes(deb.read_bytes()) + + def no_exec(*args, **kwargs): + pytest.fail(f"cross-target staging executed foreign bytes: {args}") + + monkeypatch.setattr("pm.packages.subprocess.run", no_exec) + archive({relative: b"bionic-payload"}) + entry = stage_only(name, target) + assert (entry / relative).read_bytes() == b"bionic-payload" + assert package.binary(entry, target) == entry / relative + assert package.env(entry, target).get("PATH") == ([str((entry / relative).parent)] if on_path else None) + assert stage_only(name, target) == entry + archive({"unrelated": b"not the main executable"}) + with pytest.raises(InstallError, match="missing"): + stage_only(name, target) + assert (entry / relative).read_bytes() == b"bionic-payload" + assert facts.read_bytes() == before + + +def test_deb_rejects_traversal_before_touching_outside(tmp_path): + from pm.package import DebPackage, InstallError + from tests.termux_fixtures import build_deb + + sentinel = tmp_path / "escape" + sentinel.write_bytes(b"owned outside extraction") + deb = tmp_path / "evil.deb" + build_deb(deb, {"Package": "evil"}, {"../escape": b"overwrite"}) + with pytest.raises(InstallError, match="unsafe|escape|traversal"): + DebPackage().unpack(deb, tmp_path / "staged", "linux-arm64-bionic") + assert sentinel.read_bytes() == b"owned outside extraction" diff --git a/tests/test_project_metadata.py b/tests/test_project_metadata.py index 32387727fe..b800206f56 100644 --- a/tests/test_project_metadata.py +++ b/tests/test_project_metadata.py @@ -67,119 +67,25 @@ def test_direct_overrides_preserve_the_declared_exact_version(): assert versions and all(version in requirement.specifier for version in versions) -def test_matrix_extra_not_in_all(): - """The [matrix] extra pulls `mautrix[encryption]` -> `python-olm`, - which has Linux-only wheels and no native build path on Windows or - modern macOS (archived libolm, C++ errors with Clang 21+). +def test_opt_in_extras_stay_out_of_default_recursive_selection(): + from packaging.requirements import Requirement + from pm.extras import ANCHORS - With matrix in [all], `uv sync --locked` on Windows tried to build - python-olm from sdist and failed on `make`. As of 2026-05-12 the - [matrix] extra is excluded from [all] entirely and installs on first - use (pm.ensure_import("matrix")), where the user is expected to have - a toolchain. - """ - optional_dependencies = _load_optional_dependencies() - - assert "matrix" in optional_dependencies, "[matrix] extra must still exist for `uv sync --extra matrix`" - # Must NOT appear in [all] in any form — neither unconditional nor - # platform-gated. Lazy-install handles it. - matrix_in_all = [ - dep for dep in optional_dependencies["all"] - if "matrix" in dep - ] - assert not matrix_in_all, ( - "matrix must not appear in [all] — it installs on first use via " - f"pm.ensure_import('matrix'). Found: {matrix_in_all}" - ) - - -def test_lazy_installable_extras_excluded_from_all(): - """Policy (2026-05-12): opt-in backends stay out of [all]. - - On-demand install exists so one quarantined PyPI release - (e.g. mistralai 2.4.6) can't break every fresh install. Putting a - backend in [all] defeats that — fresh installs eager-install it and - inherit whatever's broken upstream. Opt-in backends are extras that - install at first use via pm.ensure_import(extra). - """ - optional_dependencies = _load_optional_dependencies() - - # The on-demand backends as of 2026-05-12. Deliberately a literal - # list so the test stays a contract — adding a new opt-in backend - # means updating this list AND verifying [all] doesn't contain it. - lazy_covered_extras = { - "anthropic", "bedrock", - "exa", "firecrawl", "parallel-web", - "fal", - "edge-tts", "tts-premium", - "voice", # faster-whisper / sounddevice / numpy (composes stt-whisper + audio-io) - "stt-whisper", - "modal", "daytona", "vercel", - "messaging", "slack", "matrix", "dingtalk", "feishu", - "telegram", "discord", - "wake", "wake-openwakeword", "wake-sherpa", "wake-porcupine", - "google-chat", - "honcho", "hindsight", - "supermemory", "mem0", - "mistral", # mistralai — Voxtral STT/TTS, lazy-installed (stt.mistral / tts.mistral) - } - all_extra_specs = optional_dependencies["all"] - for extra in lazy_covered_extras: - offending = [ - spec for spec in all_extra_specs - if f"hermes-agent[{extra}]" in spec - ] - assert not offending, ( - f"[{extra}] is in [all] but also in LAZY_DEPS. " - f"Remove it from [all] in pyproject.toml — it lazy-installs " - f"at first use. Found in [all]: {offending}" - ) - - -def _exact_pins(specs): - pins = {} - for spec in specs: - requirement = spec.split(";", 1)[0].strip() - if "==" not in requirement: + optional = _load_optional_dependencies() + # Deliberate eager surfaces: core Google integration, ACP launcher, + # dashboard, transcript reader, and the no-op Pillow compatibility alias. + eager = {"google", "acp", "web", "youtube", "vision"} + selected, pending = set(), ["all"] + while pending: + extra = pending.pop() + if extra in selected: continue - package, version = requirement.split("==", 1) - package = package.split("[", 1)[0].lower().replace("_", "-") - pins[package] = version - return pins - - - - -def test_extras_pin_each_package_at_one_version(): - """One package, one version, across every extra. - - tools/lazy_deps.py is gone — pyproject.toml is the single authority for - optional-dependency pins (pm syncs the venv from uv.lock, which resolves - from here). The drift class that killed us before (#31817: two documents - pinning the same package differently, update ping-ponging the version) - is now only possible BETWEEN extras — so pin consistency across extras - is the whole remaining contract. - """ - optional_dependencies = _load_optional_dependencies() - - pins: dict[str, dict[str, set[str]]] = {} - for extra, specs in optional_dependencies.items(): - for package, version in _exact_pins(specs).items(): - pins.setdefault(package, {}).setdefault(version, set()).add(extra) - - drift = { - package: {v: sorted(extras) for v, extras in versions.items()} - for package, versions in pins.items() - if len(versions) > 1 - } - assert not drift, ( - "extras pin the same package at different versions — uv sync would " - f"resolve whichever wins and silently downgrade the other: {drift}" - ) - - - - + selected.add(extra) + for requirement in map(Requirement, optional[extra]): + if requirement.name == "hermes-agent": + pending.extend(requirement.extras) + assert set(ANCHORS) <= optional.keys() + assert not (selected & (set(ANCHORS) - eager)) def test_dingtalk_extra_includes_qrcode_for_qr_auth(): diff --git a/tests/test_run_tests_parallel.py b/tests/test_run_tests_parallel.py index 004a2e90ce..d16c999e15 100644 --- a/tests/test_run_tests_parallel.py +++ b/tests/test_run_tests_parallel.py @@ -275,46 +275,6 @@ def _run_runner(probe_dir: Path, *extra: str) -> subprocess.CompletedProcess: -def test_bare_value_flag_keeps_its_value(tmp_path: Path) -> None: - """``-k test_alpha`` reaches pytest as a selector, not as a path. - - The value token (``test_alpha``) must NOT be swallowed by the runner's - positional-path discovery — if it were, discovery would look for a path - named ``test_alpha``, find nothing, and the run would degrade. We assert - the run succeeds AND only one of the two tests was selected (proving the - ``-k`` filter actually applied inside pytest). - """ - probe_dir = _make_probe_dir(tmp_path) - proc = _run_runner(probe_dir, "-k", "test_alpha") - assert proc.returncode == 0, proc.stdout - # Exactly one test selected: the per-file summary shows "1✓" (1 passed). - # test_beta is deselected by the -k filter. - assert "1✓" in proc.stdout or "1 passed" in proc.stdout, proc.stdout - assert "2✓" not in proc.stdout, ( - f"both tests ran — -k filter did not apply:\n{proc.stdout}" - ) - - - - -def test_positional_path_not_treated_as_flag(tmp_path: Path) -> None: - """A positional path arg still overrides discovery (not routed to pytest).""" - probe_dir = _make_probe_dir(tmp_path) - repo_root = _probe_root(tmp_path) - runner = repo_root / "scripts" / "run_tests_parallel.py" - # Pass the probe dir positionally (no --paths), plus a bare -q. - proc = subprocess.run( - [sys.executable, str(runner), str(probe_dir), "-j", "1", - "--file-timeout", "30", "-q"], - cwd=probe_dir, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - encoding="utf-8", errors="replace", timeout=60, - ) - assert proc.returncode == 0, proc.stdout - # Discovery found the probe file (2 tests), proving the positional path - # was consumed as a root, not forwarded to pytest as a bad flag. - assert "test_flagprobe.py" in proc.stdout, proc.stdout - - def test_file_retry_self_heals_and_prints_both_attempts(tmp_path: Path) -> None: """A pass-on-retry is green, loud, and retains the failing traceback.""" repo_root = _probe_root(tmp_path) @@ -385,65 +345,36 @@ def test_zero_collected_across_run_fails_and_says_so(tmp_path: Path) -> None: -def test_node_id_selector_runs_the_named_test(tmp_path: Path) -> None: - """``file.py::test_alpha`` runs that test instead of discovering nothing.""" - probe_dir = _make_probe_dir(tmp_path) - target = probe_dir / "test_flagprobe.py" - repo_root = _probe_root(tmp_path) - proc = subprocess.run( - [sys.executable, str(repo_root / "scripts" / "run_tests_parallel.py"), - f"{target}::test_alpha", "-j", "1", "--file-timeout", "30"], - cwd=probe_dir, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, timeout=60, - ) - assert proc.returncode == 0, proc.stdout - assert "No test files to run" not in proc.stdout - assert "node id" in proc.stdout # explains the translation - # Ran exactly the one selected test, not both in the file. - assert "1 tests passed" in proc.stdout - - -def test_explicit_k_wins_over_node_id_inference(tmp_path: Path) -> None: - """A caller's own ``-k`` is not overridden by the node-id translation.""" - probe_dir = _make_probe_dir(tmp_path) - target = probe_dir / "test_flagprobe.py" - repo_root = _probe_root(tmp_path) - proc = subprocess.run( - [sys.executable, str(repo_root / "scripts" / "run_tests_parallel.py"), - f"{target}::test_alpha", "-k", "test_beta", - "-j", "1", "--file-timeout", "30"], - cwd=probe_dir, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, timeout=60, - ) - # -k test_beta wins: one test ran, and it wasn't filtered to nothing. - assert proc.returncode == 0, proc.stdout - assert "1 tests passed" in proc.stdout - - -def test_multiple_absolute_paths_split_on_pathsep(tmp_path: Path) -> None: - """``--paths`` accepts ``os.pathsep``-joined absolute paths. - - On Windows the absolute paths contain drive-letter colons, so a naive - ``split(":")`` shreds them into phantom roots and only one (or neither) - of the two probe dirs would be discovered. - """ - dir_a = _make_probe_dir(tmp_path) - dir_b = tmp_path / "probe_b" - dir_b.mkdir() - (dir_b / "test_flagprobe_b.py").write_text( - "def test_gamma():\n assert True\n" - ) - repo_root = _probe_root(tmp_path) - runner = repo_root / "scripts" / "run_tests_parallel.py" - proc = subprocess.run( - [sys.executable, str(runner), - "--paths", os.pathsep.join([str(dir_a), str(dir_b)]), - "-j", "1", "--file-timeout", "30", "-q"], - cwd=tmp_path, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - encoding="utf-8", errors="replace", timeout=60, - ) - assert proc.returncode == 0, proc.stdout - assert "Discovered 2 test files" in proc.stdout, proc.stdout +@pytest.mark.parametrize("form,expected", [ + ("positional", ["alpha", "beta"]), ("bare-k", ["alpha"]), + ("node-id", ["alpha"]), ("explicit-k", ["beta"]), + ("pathsep", ["alpha", "beta", "gamma"]), +]) +def test_runner_selection_records_actual_test_identity(tmp_path, form, expected): + probe = tmp_path / "probe" + other = tmp_path / "other" + probe.mkdir() + other.mkdir() + receipt = tmp_path / "witnesses" + receipt.mkdir() + for directory, filename, names in ((probe, "test_flags.py", ["alpha", "beta"]), + (other, "test_other.py", ["gamma"])): + (directory / filename).write_text("from pathlib import Path\n" + "".join( + f"def test_{name}():\n Path({str(receipt / name)!r}).touch()\n" for name in names + ), encoding="utf-8") + target = str(probe / "test_flags.py") + arguments = { + "positional": [str(probe), "-q"], + "bare-k": ["--paths", str(probe), "-k", "test_alpha"], + "node-id": [target + "::test_alpha"], + "explicit-k": [target + "::test_alpha", "-k", "test_beta"], + "pathsep": ["--paths", os.pathsep.join([str(probe), str(other)])], + }[form] + runner = _probe_root(tmp_path) / "scripts/run_tests_parallel.py" + result = subprocess.run([sys.executable, str(runner), *arguments, "-j", "1", "--file-timeout", "30"], + cwd=tmp_path, capture_output=True, text=True, encoding="utf-8", timeout=60) + assert result.returncode == 0, result.stdout + result.stderr + assert sorted(path.name for path in receipt.iterdir()) == expected @pytest.mark.platforms("windows") diff --git a/tests/test_stage_apt_repo.py b/tests/test_stage_apt_repo.py index d8686006c6..2f2fd5c1ee 100644 --- a/tests/test_stage_apt_repo.py +++ b/tests/test_stage_apt_repo.py @@ -6,16 +6,15 @@ output (no secret logging). """ import gzip -import io import os import shutil import subprocess import sys -import tarfile import tempfile from pathlib import Path import pytest +from tests.termux_fixtures import build_deb REPO_ROOT = Path(__file__).resolve().parents[1] SCRIPTS = REPO_ROOT / "scripts" / "termux" @@ -27,34 +26,9 @@ GPG_PRESENT = shutil.which("gpg") is not None def make_deb(path: Path, package: str, version: str, arch: str = "aarch64", compression: str = "gz") -> None: - """Build a minimal .deb (ar archive with control.tar.gz) using stdlib only.""" - control = ( - f"Package: {package}\n" - f"Version: {version}\n" - f"Architecture: {arch}\n" - f"Maintainer: Test \n" - f"Description: test package {package}\n" - ) - buf = io.BytesIO() - mode = f"w:{compression}" - member = f"control.tar.{compression}" if compression != "tar" else "control.tar" - with tarfile.open(fileobj=buf, mode=mode) as tf: - data = control.encode("utf-8") - ti = tarfile.TarInfo("control") - ti.size = len(data) - tf.addfile(ti, io.BytesIO(data)) - - ar = io.BytesIO() - ar.write(b"!\n") - payload = buf.getvalue() - header = "{:<16}{:<12}{:<6}{:<6}{:<8}{:<10}".format( - member, "0", "0", "0", "100644", str(len(payload)) - ).encode() + b"`\n" - ar.write(header) - ar.write(payload) - if len(payload) % 2: - ar.write(b"\n") - path.write_bytes(ar.getvalue()) + build_deb(path, {"Package": package, "Version": version, "Architecture": arch, + "Maintainer": "Test ", "Description": f"test package {package}"}, + compression=compression) @pytest.fixture @@ -63,37 +37,6 @@ def no_gpg(monkeypatch): monkeypatch.setattr(stage_apt_repo.shutil, "which", lambda _: None) -@pytest.fixture -def fake_gpg(monkeypatch, tmp_path): - """Make the script believe gpg is present, but stub out signing.""" - monkeypatch.setattr(stage_apt_repo.shutil, "which", lambda _: "C:/fake/gpg.exe") - monkeypatch.setattr(stage_apt_repo, "sign", lambda *a, **k: None) - key = tmp_path / "signing.asc" - key.write_text("stub-key\n") - return key - - -def test_stages_xz_control_deb(tmp_path): - """dpkg >= 1.21 emits xz/zst control members; our build uses -Zxz so the - stager must read xz controls (gz is covered by every other test).""" - pool = tmp_path / "pool" - pool.mkdir() - make_deb(pool / "hermes-agent_1.0-1_aarch64.deb", "hermes-agent", "1.0-1", compression="xz") - out = tmp_path / "out" - out.mkdir() - rc = stage_apt_repo.stage(pool, out, "hermes-canary", None) - assert rc == 3 # unsigned (no gpg key file) but staged - - -def test_control_field_extraction(tmp_path): - deb = tmp_path / "pkg_a.deb" - make_deb(deb, "hermes-agent", "1.2.3-1") - fields = stage_apt_repo.deb_control_fields(deb) - assert fields["Package"] == "hermes-agent" - assert fields["Version"] == "1.2.3-1" - assert fields["Architecture"] == "aarch64" - - def test_canary_versions_below_stable(): versions = ["1.2.3-1", "1.2.3~canary.20260831120000-1", "1.2.4~canary.1-1", "1.2.4-1"] ordered = sorted(versions, key=stage_apt_repo.deb_version_key) @@ -105,97 +48,51 @@ def test_canary_versions_below_stable(): ] -def test_dists_layout_and_pool_copy(tmp_path, fake_gpg): - pool = tmp_path / "pool-in" - pool.mkdir() - make_deb(pool / "hermes-agent_1.2.3-1_aarch64.deb", "hermes-agent", "1.2.3-1") - out = tmp_path / "repo" - r = stage_apt_repo.main( - [ - "--pool", str(pool), "--out", str(out), "--suite", "hermes-stable", - "--gpg-key-file", str(fake_gpg), - ] - ) - assert r == 0 - - dists = out / "dists" / "hermes-stable" / "main" / "binary-aarch64" - assert (dists / "Packages").exists() - assert (dists / "Packages.gz").exists() - assert (out / "dists" / "hermes-stable" / "Release").exists() - - deb_out = out / "pool" / "h" / "hermes-agent_1.2.3-1_aarch64.deb" - assert deb_out.exists() - - text = (dists / "Packages").read_text(encoding="utf-8") - assert "Package: hermes-agent" in text - assert "Version: 1.2.3-1" in text - assert "Filename: pool/h/hermes-agent_1.2.3-1_aarch64.deb" in text - assert "SHA256: " in text - - gz_text = gzip.decompress((dists / "Packages.gz").read_bytes()).decode() - assert gz_text == text - - release = (out / "dists" / "hermes-stable" / "Release").read_text() - assert "Suite: hermes-stable" in release - assert "SHA256:" in release - assert "SHA512:" in release - # apt contract (learned from a real device rejecting our first repo): - # Date is mandatory, and the checksum sections must live in the SAME - # deb822 stanza as the header fields -- a blank line ends the record, - # after which apt "provides only weak security information" and - # disables the repository. - assert "Date: " in release - assert "\n\n" not in release, "blank line splits the Release stanza" - head, _, checksums_block = release.partition("SHA256:\n") - assert "Date: " in head, "Date must precede the checksum sections" - - -def test_immutability_refusal(tmp_path, fake_gpg, capsys): - pool = tmp_path / "pool-in" - pool.mkdir() - make_deb(pool / "hermes-agent_1.2.3-1_aarch64.deb", "hermes-agent", "1.2.3-1") - out = tmp_path / "repo" - assert stage_apt_repo.main( - [ - "--pool", str(pool), "--out", str(out), "--suite", "hermes-stable", - "--gpg-key-file", str(fake_gpg), - ] - ) == 0 - with pytest.raises(SystemExit) as ei: - stage_apt_repo.main( - [ - "--pool", str(pool), "--out", str(out), "--suite", "hermes-stable", - "--gpg-key-file", str(fake_gpg), - ] - ) - assert ei.value.code == 2 - assert "already published" in capsys.readouterr().err - - -def test_by_hash_indexes_match_release_and_survive_later_publication(tmp_path): +def test_unsigned_multiversion_publication_and_immutable_indexes(tmp_path, capsys): import hashlib pool = tmp_path / "pool" pool.mkdir() - package = pool / "hermes-agent.deb" - make_deb(package, "hermes-agent", "1.0-1") + versions = ["1.2.3~canary.20260901000000-1", "1.2.3-1"] + for filename, version, compression in zip(("a.deb", "b.deb"), versions, ("gz", "xz")): + make_deb(pool / filename, "hermes-agent", version, compression=compression) out = tmp_path / "repo" - assert stage_apt_repo.stage(pool, out, "hermes-canary", None) == 3 + args = ["--pool", str(pool), "--out", str(out), "--suite", "hermes-canary"] + assert stage_apt_repo.main(args) == 3 binary = out / "dists/hermes-canary/main/binary-aarch64" - original = {} + text = (binary / "Packages").read_text(encoding="utf-8") + records = [dict(line.split(": ", 1) for line in stanza.splitlines()) for stanza in text.strip().split("\n\n")] + assert [row["Version"] for row in records] == versions + for row in records: + assert row["Package"] == "hermes-agent" and row["Architecture"] == "aarch64" + copied = out / row["Filename"] + original = pool / copied.name + assert copied.read_bytes() == original.read_bytes() + assert row["SHA256"] == hashlib.sha256(copied.read_bytes()).hexdigest() + assert row["Size"] == str(copied.stat().st_size) + assert gzip.decompress((binary / "Packages.gz").read_bytes()).decode() == text + release = (out / "dists/hermes-canary/Release").read_text(encoding="utf-8") + assert "Suite: hermes-canary\n" in release and "Acquire-By-Hash: yes\n" in release + assert "\n\n" not in release + assert "Date: " in release.partition("SHA256:\n")[0] + immutable = {} for name in ("Packages", "Packages.gz"): data = (binary / name).read_bytes() for algorithm in ("SHA256", "SHA512"): digest = hashlib.new(algorithm.lower(), data).hexdigest() - immutable = binary / "by-hash" / algorithm / digest - assert immutable.read_bytes() == data - original[immutable] = data - release = (out / "dists/hermes-canary/Release").read_text() - assert "Acquire-By-Hash: yes\n" in release - make_deb(package, "hermes-agent", "1.1-1") - assert stage_apt_repo.stage(pool, out, "hermes-canary", None) == 3 - for path, data in original.items(): - assert path.read_bytes() == data + assert [digest, str(len(data)), f"main/binary-aarch64/{name}"] in [line.split() for line in release.splitlines()] + path = binary / "by-hash" / algorithm / digest + assert path.read_bytes() == data + immutable[path] = data + assert stage_apt_repo.existing_published(out, "hermes-canary") == {("hermes-agent", v) for v in versions} + with pytest.raises(SystemExit) as stopped: + stage_apt_repo.main(args) + assert stopped.value.code == 2 and "already published" in capsys.readouterr().err + for old in pool.iterdir(): + old.unlink() + make_deb(pool / "c.deb", "hermes-agent", "1.2.4-1") + assert stage_apt_repo.main(args) == 3 + assert all(path.read_bytes() == data for path, data in immutable.items()) def test_unsigned_release_exit_3_without_gpg(tmp_path, no_gpg): @@ -212,78 +109,6 @@ def test_unsigned_release_exit_3_without_gpg(tmp_path, no_gpg): assert not (out / "dists" / "hermes-canary" / "Release.gpg").exists() -def test_signing_invoked_when_gpg_and_key_present(tmp_path, monkeypatch): - """No real gpg: assert sign() is called with the right dists dir/key file.""" - calls = [] - - def fake_sign(dists, release_path, gpg_key_file): - calls.append((str(dists), str(release_path), str(gpg_key_file))) - (dists / "InRelease").write_text("stub", encoding="utf-8") - (dists / "Release.gpg").write_text("stub", encoding="utf-8") - - monkeypatch.setattr(stage_apt_repo.shutil, "which", lambda _: "C:/fake/gpg.exe") - monkeypatch.setattr(stage_apt_repo, "sign", fake_sign) - - pool = tmp_path / "pool-in" - pool.mkdir() - make_deb(pool / "hermes-agent_1.2.3-1_aarch64.deb", "hermes-agent", "1.2.3-1") - out = tmp_path / "repo" - keyfile = tmp_path / "signing.asc" - keyfile.write_text("-----BEGIN PGP PRIVATE KEY BLOCK-----\n") - code = stage_apt_repo.main( - [ - "--pool", str(pool), "--out", str(out), "--suite", "hermes-stable", - "--gpg-key-file", str(keyfile), - ] - ) - assert code == 0 - assert len(calls) == 1 - dists, release_path, kf = calls[0] - assert dists == str(out / "dists" / "hermes-stable") - assert release_path == str(out / "dists" / "hermes-stable" / "Release") - assert kf == str(keyfile) - assert (out / "dists" / "hermes-stable" / "InRelease").exists() - - -# --------------------------------------------------------------------------- -# deb822 record separation (multiversion Packages correctness) -# --------------------------------------------------------------------------- - -def _stanza_count(packages_text: str) -> int: - return len([s for s in packages_text.split("\n\n") if s.strip()]) - - -def test_multiversion_packages_records_are_blank_line_separated(tmp_path): - """Multiple versions of one package must be separate deb822 records: - apt splits records on blank lines, so a missing blank line merges two - versions into one garbled stanza and drops the later one.""" - pool = tmp_path / "pool-in" - pool.mkdir() - make_deb(pool / "a.deb", "hermes-agent", "1.2.3~canary.20260901000000-1") - make_deb(pool / "b.deb", "hermes-agent", "1.2.3-1") - out = tmp_path / "repo" - assert stage_apt_repo.main( - ["--pool", str(pool), "--out", str(out), "--suite", "hermes-canary"] - ) == 3 # staged unsigned - - text = (out / "dists" / "hermes-canary" / "main" / "binary-aarch64" / "Packages").read_text() - assert _stanza_count(text) == 2 - assert "Version: 1.2.3~canary.20260901000000-1\n" in text - assert "Version: 1.2.3-1\n" in text - # each stanza carries its own checksum - assert text.count("SHA256: ") == 2 - # the repo's own published-set parser agrees (it feeds immutability) - published = stage_apt_repo.existing_published(out, "hermes-canary") - assert published == { - ("hermes-agent", "1.2.3~canary.20260901000000-1"), - ("hermes-agent", "1.2.3-1"), - } - - -# --------------------------------------------------------------------------- -# Real-GPG behavioral tests (throwaway key in a temp GNUPGHOME) -# --------------------------------------------------------------------------- - def _generate_test_key(home: Path, passphrase: str = "") -> str: """Generate a throwaway ed25519 signing key inside `home` and return its fingerprint. Uses the production _gpg_run wrapper.""" @@ -349,18 +174,19 @@ def tracked_gpg_argv(monkeypatch): @pytest.mark.skipif(not GPG_PRESENT, reason="gpg binary not available") -def test_real_gpg_signs_and_published_public_key_verifies(tmp_path, monkeypatch, tracked_gpg_argv, short_home): +@pytest.mark.parametrize("secret_pass", ["", "correct-horse-battery-staple"]) +def test_real_gpg_signs_and_published_public_key_verifies(tmp_path, monkeypatch, tracked_gpg_argv, short_home, secret_pass): """Full behavior: a staged repo signs in an isolated temp GNUPGHOME, and InRelease + detached Release.gpg verify as GOOD signatures using ONLY the published key.asc (independent gpgv keyring).""" - monkeypatch.delenv("TERMUX_APT_GPG_PASSPHRASE", raising=False) + monkeypatch.setenv("TERMUX_APT_GPG_PASSPHRASE", secret_pass) kh = short_home() - fpr = _generate_test_key(kh) + fpr = _generate_test_key(kh, passphrase=secret_pass) stage_apt_repo._gpg_run( - kh, ["--quick-add-key", fpr, "ed25519", "sign", "never"], passphrase="", + kh, ["--quick-add-key", fpr, "ed25519", "sign", "never"], passphrase=secret_pass, ) keyfile = tmp_path / "signing.asc" - keyfile.write_bytes(_export_secret_key(kh, fpr)) + keyfile.write_bytes(_export_secret_key(kh, fpr, passphrase=secret_pass)) pool = tmp_path / "pool-in" pool.mkdir() @@ -379,6 +205,8 @@ def test_real_gpg_signs_and_published_public_key_verifies(tmp_path, monkeypatch, # the system temp dir, never the user's default keyring. temp_root = stage_apt_repo._gpg_homedir_arg(Path(tempfile.gettempdir())) for argv in tracked_gpg_argv: + if secret_pass: + assert secret_pass not in " ".join(argv) assert "--homedir" in argv, f"gpg called without --homedir: {argv}" homedir = argv[argv.index("--homedir") + 1] assert homedir.startswith(temp_root), homedir @@ -408,38 +236,6 @@ def test_real_gpg_signs_and_published_public_key_verifies(tmp_path, monkeypatch, assert not native.exists() -@pytest.mark.skipif(not GPG_PRESENT, reason="gpg binary not available") -def test_real_gpg_passphrase_reaches_gpg_via_stdin_never_argv(tmp_path, monkeypatch, tracked_gpg_argv, short_home): - """A passphrase-protected signing key works (env var -> stdin fd), and - the passphrase never appears in any spawned argv.""" - secret_pass = "correct-horse-battery-staple" - monkeypatch.setenv("TERMUX_APT_GPG_PASSPHRASE", secret_pass) - kh = short_home() - fpr = _generate_test_key(kh, passphrase=secret_pass) - keyfile = tmp_path / "signing.asc" - keyfile.write_bytes(_export_secret_key(kh, fpr, passphrase=secret_pass)) - - pool = tmp_path / "pool-in" - pool.mkdir() - make_deb(pool / "h.deb", "hermes-agent", "1.2.3-1") - out = tmp_path / "repo" - assert stage_apt_repo.main( - ["--pool", str(pool), "--out", str(out), - "--suite", "hermes-canary", "--gpg-key-file", str(keyfile)] - ) == 0 - - for argv in tracked_gpg_argv: - assert secret_pass not in " ".join(argv), "passphrase leaked into argv" - - dists = out / "dists" / "hermes-canary" - vr = short_home(prefix="apt-test-verify-") - stage_apt_repo._gpg_run(vr, ["--import"], stdin=(out / "key.asc").read_bytes()) - r = _independent_gpgv_verify(vr, dists / "InRelease") - assert r.returncode == 0, r.stderr.decode() - r = _independent_gpgv_verify(vr, dists / "Release.gpg", dists / "Release") - assert r.returncode == 0, r.stderr.decode() - - @pytest.mark.skipif(not GPG_PRESENT, reason="gpg binary not available") def test_real_gpg_tampered_metadata_fails_closed(tmp_path, monkeypatch, short_home): """Fail-closed contract: verification of the signed artifacts is done @@ -458,27 +254,24 @@ def test_real_gpg_tampered_metadata_fails_closed(tmp_path, monkeypatch, short_ho pool.mkdir() make_deb(pool / "h.deb", "hermes-agent", "1.2.3-1") out = tmp_path / "repo" - assert stage_apt_repo.main( - ["--pool", str(pool), "--out", str(out), - "--suite", "hermes-stable", "--gpg-key-file", str(keyfile)] - ) == 0 + real_gpg = stage_apt_repo._gpg_run - dists = out / "dists" / "hermes-stable" - # untouched artifacts verify with the exact signing fingerprint - stage_apt_repo._verify_signature(kh, fpr, dists / "InRelease", None) - stage_apt_repo._verify_signature(kh, fpr, dists / "Release.gpg", dists / "Release") + def tamper_after_sign(home, args, **kwargs): + result = real_gpg(home, args, **kwargs) + if "--detach-sign" in args: + release = Path(args[-1]) + release.write_bytes(release.read_bytes() + b"Architectures: amd64\n") + return result - # tamper with the signed Release -> detached sig no longer validates - # (gpg exits non-zero during re-verification -> fail closed) - release_path = dists / "Release" - release_path.write_text(release_path.read_text() + "Architectures: amd64\n") + monkeypatch.setattr(stage_apt_repo, "_gpg_run", tamper_after_sign) with pytest.raises(stage_apt_repo.StageError): - stage_apt_repo._verify_signature(kh, fpr, dists / "Release.gpg", release_path) - - # and gpgv agrees independently + stage_apt_repo.stage(pool, out, "hermes-stable", keyfile) + assert not (out / "key.asc").exists(), "verification must precede public-key publication" + dists = out / "dists/hermes-stable" vr = short_home(prefix="apt-test-verify-") - stage_apt_repo._gpg_run(vr, ["--import"], stdin=(out / "key.asc").read_bytes()) - assert _independent_gpgv_verify(vr, dists / "Release.gpg", release_path).returncode != 0 + public = real_gpg(kh, ["--armor", "--export", fpr]).stdout + real_gpg(vr, ["--import"], stdin=public) + assert _independent_gpgv_verify(vr, dists / "Release.gpg", dists / "Release").returncode != 0 @pytest.mark.skipif(not GPG_PRESENT, reason="gpg binary not available") diff --git a/tests/test_termux_deb_version.py b/tests/test_termux_deb_version.py index 1d8e15cb3e..e550b50882 100644 --- a/tests/test_termux_deb_version.py +++ b/tests/test_termux_deb_version.py @@ -1,138 +1,56 @@ -"""Unit tests for scripts/termux/deb_version.py (Task 4 of the termux-deb plan).""" - +"""Literal tag boundaries and real Debian ordering (not a second parser).""" +import shutil import subprocess import sys from pathlib import Path import pytest +from scripts.termux.deb_version import channel_for_tag, deb_version_for_tag -HERE = Path(__file__).resolve().parent -SCRIPT = HERE.parent / "scripts" / "termux" / "deb_version.py" - -from scripts.termux.deb_version import channel_for_tag, deb_version_for_tag # noqa: E402 +SCRIPT = Path(__file__).resolve().parents[1] / "scripts/termux/deb_version.py" -def test_canary_tag_shape_matches_canonical(): - """Invariant: the deb versioner accepts EXACTLY the canary tags the - canonical release tooling mints. The canonical shape lives in - hermes_cli/update_channel.py:_CANARY_TAG_RE (8-or-14-digit, 20-prefixed - timestamps); scripts/releases/r2.py:channel_for_tag parses the same shape. - A tag this module accepts but the release flow would never mint (or vice - versa) is version-drift between the .deb channel and the feed channel. - """ - from hermes_cli.update_channel import _CANARY_TAG_RE as _NIGHTLY_TAG_RE - from scripts.termux import deb_version as dv - - samples = [ - "v0.20.6-canary.20260831120000", # canonical canary (14-digit) - "v0.20.6-canary.20260831", # canonical canary (8-digit) - "v1.2.3", # stable - ] - for tag in samples: - assert dv._TAG_RE.match(tag), f"deb versioner rejects canonical tag {tag}" - - never_minted = [ - "v1.2.3-canary.202608311", # 9 digits -- canonical rejects - "v1.2.3-canary.12345678", # non-20 prefix -- canonical rejects - "v1.2.3-canary.202608311200001", # 15 digits -- canonical rejects - ] - for tag in never_minted: - assert not _NIGHTLY_TAG_RE.match(tag), f"sample is actually canonical: {tag}" - assert not dv._TAG_RE.match(tag), f"deb versioner accepts never-minted tag {tag}" +@pytest.mark.parametrize("tag,version,channel", [ + ("v1.2.3", "1.2.3-1", "stable"), + ("v26.8.31", "26.8.31-1", "stable"), + ("v126.8.31", "126.8.31-1", "stable"), + ("v1.234.567", "1.234.567-1", "stable"), + ("v0.20.6-canary.20260831", "0.20.6~canary.20260831-1", "canary"), + ("v0.20.6-canary.20260831120000", "0.20.6~canary.20260831120000-1", "canary"), +]) +def test_tag_mapping(tag, version, channel): + assert deb_version_for_tag(tag) == version + assert channel_for_tag(tag) == channel -def test_stable_tag_maps_to_revision_1(): - assert deb_version_for_tag("v1.2.3") == "1.2.3-1" +@pytest.mark.parametrize("tag", [ + "", "1.2.3", "v1.2", "v1.2.3.4", "v1.2.3-", "v1.2.3-canary", "v1.2.3-canary.abc", + "v1.2.3-beta.1", "v1.2.x", "v-1.2.3", "v1234.1.2", "v99999.0.0", + "v1.2.3-canary.202608311", "v1.2.3-canary.12345678", "v1.2.3-canary.202608311200001", +]) +def test_malformed_tags_rejected_by_both_mappings(tag): + for mapping in (deb_version_for_tag, channel_for_tag): + with pytest.raises(ValueError): + mapping(tag) -def _dpkg_key(v: str) -> str: - # Approximate dpkg ordering for these versions: '~' sorts before everything - # (even the empty string / '-'), so map it low. - return v.replace("~", "\x00") +@pytest.mark.parametrize("args,status,output", [ + (["v9.8.7"], 0, "9.8.7-1"), + (["--channel", "v9.8.7"], 0, "stable"), + (["--channel", "v9.8.7-canary.20260831120000"], 0, "canary"), + (["v1.2"], 1, ""), (["--channel", "v1.2"], 1, ""), +]) +def test_cli_dispatch(args, status, output): + result = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True) + assert result.returncode == status, result.stderr + assert result.stdout.strip() == output + if status: + assert result.stderr -def test_stable_tag_multi_digit(): - assert deb_version_for_tag("v26.8.31") == "26.8.31-1" - - -def test_major_can_be_three_digits(): - assert deb_version_for_tag("v126.8.31") == "126.8.31-1" - - -def test_canary_tag_ranks_below_stable(): - got = deb_version_for_tag("v1.2.3-canary.20260831120000") - assert got == "1.2.3~canary.20260831120000-1" - assert _dpkg_key(got) < _dpkg_key(deb_version_for_tag("v1.2.3")) # dpkg ordering - - -def test_canary_canary_ranking_among_nightlies(): - earlier = deb_version_for_tag("v1.2.3-canary.20260831000000") - later = deb_version_for_tag("v1.2.3-canary.20260831235959") - assert _dpkg_key(earlier) < _dpkg_key(later) < _dpkg_key(deb_version_for_tag("v1.2.3")) - - -@pytest.mark.parametrize( - "bad", - [ - "", - "1.2.3", # missing v prefix - "v1.2", # not three components - "v1.2.3.4", # four components - "v1.2.3-", # empty suffix - "v1.2.3-canary", # canary without timestamp - "v1.2.3-canary.abc", # non-numeric timestamp - "v1.2.3-beta.1", # unknown suffix channel - "v1.2.x", - "v-1.2.3", - ], -) -def test_malformed_tags_raise(bad): - with pytest.raises(ValueError): - deb_version_for_tag(bad) - - -@pytest.mark.parametrize("bad", ["v1234.1.2", "v99999.0.0"]) -def test_major_above_three_digits_rejected(bad): - with pytest.raises(ValueError): - deb_version_for_tag(bad) - - -def test_minor_patch_can_be_three_digits(): - # Cap applies to major only; minor/patch may be wide. - assert deb_version_for_tag("v1.234.567") == "1.234.567-1" - - -def test_cli_invocation(capsys): - r = subprocess.run( - [sys.executable, str(SCRIPT), "v9.8.7"], capture_output=True, text=True - ) - assert r.returncode == 0, r.stderr - assert r.stdout.strip() == "9.8.7-1" - - -def test_channel_matches_canary_shape(): - """--channel derives from the SAME _TAG_RE as the deb version: any tag - that yields a '~canary' version is canary, everything else stable.""" - assert channel_for_tag("v1.2.3") == "stable" - assert channel_for_tag("v26.8.31") == "stable" - assert channel_for_tag("v0.20.6-canary.20260831120000") == "canary" - assert channel_for_tag("v0.20.6-canary.20260831") == "canary" - - -def test_channel_agrees_with_deb_version(): - for tag in ("v1.2.3", "v126.8.31", "v1.2.3-canary.20260831120000"): - assert ("~canary" in deb_version_for_tag(tag)) == (channel_for_tag(tag) == "canary") - - -def test_channel_malformed_tag_raises(): - with pytest.raises(ValueError): - channel_for_tag("v1.2") - - -def test_channel_cli_invocation(): - for tag, expected in [("v9.8.7", "stable"), ("v9.8.7-canary.20260831120000", "canary")]: - r = subprocess.run( - [sys.executable, str(SCRIPT), "--channel", tag], capture_output=True, text=True - ) - assert r.returncode == 0, r.stderr - assert r.stdout.strip() == expected +@pytest.mark.skipif(shutil.which("dpkg") is None, reason="requires native dpkg") +def test_dpkg_orders_canary_and_numeric_versions(): + tags = ["v1.2.3-canary.20260831000000", "v1.2.3-canary.20260831235959", "v1.2.3", "v1.2.10"] + versions = list(map(deb_version_for_tag, tags)) + for earlier, later in zip(versions, versions[1:]): + subprocess.run(["dpkg", "--compare-versions", earlier, "lt", later], check=True) diff --git a/tests/test_termux_python_linkage.py b/tests/test_termux_python_linkage.py index 37c19d27a0..d15070fea6 100644 --- a/tests/test_termux_python_linkage.py +++ b/tests/test_termux_python_linkage.py @@ -18,7 +18,7 @@ def test_python_symbols_gain_an_explicit_library_dependency(tmp_path): library = Path(sysconfig.get_config_var("LIBDIR")) / sysconfig.get_config_var("LDLIBRARY") extension = tmp_path / Path(_cffi_backend.__file__).name - shutil.copy2(_cffi_backend.__file__, extension) + shutil.copyfile(_cffi_backend.__file__, extension) # Writable scratch even from a read-only Nix store. original = subprocess.check_output(["patchelf", "--print-needed", str(extension)], text=True).splitlines() for name in original: if name.startswith("libpython"): @@ -35,33 +35,24 @@ def test_python_symbols_gain_an_explicit_library_dependency(tmp_path): def test_wheel_rewrite_regenerates_record_for_changed_member(tmp_path): - import base64 - import csv - import hashlib - import io import zipfile + from scripts.termux import retag_wheel + from tests.termux_fixtures import write_wheel, verify_record - wheel = tmp_path / "sample-1.0-cp311-cp311-linux_aarch64.whl" - record = "sample-1.0.dist-info/RECORD" - with zipfile.ZipFile(wheel, "w") as archive: - archive.writestr("sample/_native.so", b"unrepaired native bytes") - archive.writestr("sample/__init__.py", b"") - archive.writestr("sample-1.0.dist-info/WHEEL", "Wheel-Version: 1.0\nTag: cp311-cp311-linux_aarch64\n") - archive.writestr(record, "") + wheel = write_wheel(tmp_path) def repair(path, library): assert library == tmp_path / "libpython.so" + assert path.read_bytes() == b"\x7fELFfake" path.write_bytes(b"repaired native bytes") return True python_linkage.repair_wheel(wheel, tmp_path / "libpython.so", repair=repair) + verify_record(wheel) # Retagging must not hide a stale repair RECORD. with zipfile.ZipFile(wheel) as archive: - assert archive.read("sample/_native.so") == b"repaired native bytes" - rows = {r[0]: r[1:] for r in csv.reader(io.StringIO(archive.read(record).decode()))} - for name in archive.namelist(): - if name == record: - assert rows[name] == ["", ""] - continue - data = archive.read(name) - digest = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).rstrip(b"=").decode() - assert rows[name] == ["sha256=" + digest, str(len(data))] + assert archive.read("fakedep/_native.so") == b"repaired native bytes" + retagged = retag_wheel.retag_wheel(str(wheel), "android_24_arm64_v8a") + verify_record(retagged) + with zipfile.ZipFile(retagged) as archive: + assert archive.read("fakedep/_native.so") == b"repaired native bytes" + assert b"Tag: py3-none-android_24_arm64_v8a\n" in archive.read("fakedep-1.2.3.dist-info/WHEEL") diff --git a/tests/test_termux_retag_wheel.py b/tests/test_termux_retag_wheel.py index 466dac4a0b..cf4cee048d 100644 --- a/tests/test_termux_retag_wheel.py +++ b/tests/test_termux_retag_wheel.py @@ -10,15 +10,12 @@ Run: scripts/run_tests.sh tests/test_termux_retag_wheel.py from __future__ import annotations -import base64 -import csv -import hashlib -import io import sys import zipfile from pathlib import Path import pytest +from tests.termux_fixtures import write_wheel, verify_record SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts" / "termux" sys.path.insert(0, str(SCRIPTS_DIR)) @@ -28,60 +25,9 @@ import retag_wheel # noqa: E402 ANDROID_TAG = "android_24_arm64_v8a" -def _record_hash(data: bytes) -> str: - digest = hashlib.sha256(data).digest() - return "sha256=" + base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") - - -def _write_wheel( - path: Path, - distribution: str, - version: str, - platform_tag: str, - *, - metadata_version: str | None = None, - include_so: bool = True, -) -> None: - """Build a tiny fake wheel with the same shape the builder produces.""" - dist_info = f"{distribution}-{version}.dist-info" - members: list[tuple[str, bytes]] = [] - if include_so: - # A fake native extension -- retagging a pure wheel onto a platform - # tag would be a lie, so fixtures default to carrying one. - members.append((f"{distribution}/_native.cpython-314-aarch64-linux-gnu.so", b"\x7fELFfake")) - members.append((f"{distribution}/__init__.py", b"")) - metadata_version = metadata_version or version - members.append((f"{dist_info}/METADATA", f"Metadata-Version: 2.1\nName: {distribution}\nVersion: {metadata_version}\n".encode())) - members.append((f"{dist_info}/WHEEL", f"Wheel-Version: 1.0\nRoot-Is-Purelib: false\nTag: py3-none-{platform_tag}\nGenerator: fixture\n".encode())) - - # RECORD with real hashes so the retagger's ZIP-integrity/consistency - # checks exercise the true path. - rows: list[list[str]] = [] - for name, data in members: - rows.append([name, _record_hash(data), str(len(data))]) - rows.append([f"{dist_info}/RECORD", "", ""]) - buf = io.StringIO() - csv.writer(buf, lineterminator="\n").writerows(rows) - members.append((f"{dist_info}/RECORD", buf.getvalue().encode())) - - filename = f"{distribution}-{version}-py3-none-{platform_tag}.whl" - with zipfile.ZipFile(path / filename, "w", zipfile.ZIP_DEFLATED) as zf: - for name, data in members: - zf.writestr(name, data) - - -def _read_member(zf: zipfile.ZipFile, name: str) -> bytes: - return zf.read(name) - - -def _record_rows(zf: zipfile.ZipFile, dist_info: str) -> dict[str, tuple[str, str]]: - text = zf.read(f"{dist_info}/RECORD").decode("utf-8") - return {row[0]: (row[1], row[2]) for row in csv.reader(io.StringIO(text)) if row} - - @pytest.fixture def wheel(tmp_path: Path) -> Path: - _write_wheel(tmp_path, "fakedep", "1.2.3", "linux_aarch64") + write_wheel(tmp_path, "fakedep", "1.2.3", "linux_aarch64") return tmp_path / "fakedep-1.2.3-py3-none-linux_aarch64.whl" @@ -89,6 +35,7 @@ def test_filename_and_wheel_tags_rewritten_consistently(wheel: Path) -> None: new_path = Path(retag_wheel.retag_wheel(str(wheel), ANDROID_TAG)) assert new_path.name == f"fakedep-1.2.3-py3-none-{ANDROID_TAG}.whl" + verify_record(new_path) assert not wheel.exists(), "the original wheel must be replaced, not left beside the new one" with zipfile.ZipFile(new_path) as zf: @@ -97,27 +44,8 @@ def test_filename_and_wheel_tags_rewritten_consistently(wheel: Path) -> None: assert tag_lines == [f"Tag: py3-none-{ANDROID_TAG}"], wheel_txt -def test_record_rows_valid_after_retag(wheel: Path) -> None: - new_path = Path(retag_wheel.retag_wheel(str(wheel), ANDROID_TAG)) - - with zipfile.ZipFile(new_path) as zf: - rows = _record_rows(zf, "fakedep-1.2.3.dist-info") - for member in zf.namelist(): - if member.endswith("/"): - continue - if member == "fakedep-1.2.3.dist-info/RECORD": - # RECORD's own row is digest-less by spec; checked separately - continue - data = _read_member(zf, member) - digest, size = rows[member] - assert digest == _record_hash(data), f"RECORD hash stale for {member}" - assert size == str(len(data)), f"RECORD size stale for {member}" - record_row = rows["fakedep-1.2.3.dist-info/RECORD"] - assert record_row == ("", ""), "RECORD's own row must be digest-less" - - def test_native_extension_presence_required(tmp_path: Path) -> None: - _write_wheel(tmp_path, "puredist", "0.1.0", "linux_aarch64", include_so=False) + write_wheel(tmp_path, "puredist", "0.1.0", "linux_aarch64", include_so=False) pure = tmp_path / "puredist-0.1.0-py3-none-linux_aarch64.whl" with pytest.raises(retag_wheel.RetagError, match="native"): retag_wheel.retag_wheel(str(pure), ANDROID_TAG) @@ -126,7 +54,7 @@ def test_native_extension_presence_required(tmp_path: Path) -> None: def test_refuses_version_mismatch_between_filename_and_metadata(tmp_path: Path) -> None: # METADATA says 9.9.9 while the filename says 1.2.3 -- a lie the # retagger must refuse rather than launder. - _write_wheel(tmp_path, "fakedep", "1.2.3", "linux_aarch64", metadata_version="9.9.9") + write_wheel(tmp_path, "fakedep", "1.2.3", "linux_aarch64", metadata_version="9.9.9") lying = tmp_path / "fakedep-1.2.3-py3-none-linux_aarch64.whl" with pytest.raises(retag_wheel.RetagError): retag_wheel.retag_wheel(str(lying), ANDROID_TAG) diff --git a/tests/test_termux_runtime_libs.py b/tests/test_termux_runtime_libs.py index 98f8cfd5ac..bb3e969fa3 100644 --- a/tests/test_termux_runtime_libs.py +++ b/tests/test_termux_runtime_libs.py @@ -7,14 +7,13 @@ then unpacked through pm's DebPackage — the same production path. from __future__ import annotations import hashlib -import io import json -import tarfile import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import pytest +from tests.termux_fixtures import build_deb import sys @@ -29,53 +28,9 @@ PREFIX = srl.PREFIX_REL # ---------------------------------------------------------------- fixtures -def _ar_header(name: str, size: int) -> bytes: - hdr = name.ljust(16).encode() - hdr += b"0".ljust(12) # mtime - hdr += b"0".ljust(6) # uid - hdr += b"0".ljust(6) # gid - hdr += b"644".ljust(8) # mode - hdr += str(size).ljust(10).encode() - hdr += b"`\n" - assert len(hdr) == 60 - return hdr - - def _build_deb(path: Path, lib_name: str, content: bytes) -> None: - """A minimal but real .deb: ar{debian-binary, control.tar.gz, data.tar} - with data.tar carrying /lib/.""" - data = io.BytesIO() - with tarfile.open(fileobj=data, mode="w") as tf: - lib_dir = f"{PREFIX}/lib/" - ti = tarfile.TarInfo(lib_dir) - ti.type = tarfile.DIRTYPE - ti.mode = 0o755 - tf.addfile(ti) - payload = f"FAKE-ELF {lib_name}\n".encode() + content - ti = tarfile.TarInfo(lib_dir + lib_name) - ti.size = len(payload) - ti.mode = 0o755 - tf.addfile(ti, io.BytesIO(payload)) - - control = io.BytesIO() - with tarfile.open(fileobj=control, mode="w") as tf: - stanza = f"Package: pkg-{lib_name}\nVersion: 1.0\n".encode() - ti = tarfile.TarInfo("control") - ti.size = len(stanza) - tf.addfile(ti, io.BytesIO(stanza)) - - members = [ - (b"debian-binary/", b"2.0\n"), - (b"control.tar.gz/", control.getvalue()), - (b"data.tar/", data.getvalue()), - ] - out = b"!\n" - for name, body in members: - out += _ar_header(name.decode().rstrip("/"), len(body)) - out += body - if len(body) % 2: - out += b"\n" - path.write_bytes(out) + build_deb(path, {"Package": f"pkg-{lib_name}", "Version": "1.0"}, + {f"{PREFIX}/lib/{lib_name}": f"FAKE-ELF {lib_name}\n".encode() + content}) class _Server: @@ -83,9 +38,16 @@ class _Server: def __init__(self, root: Path) -> None: self.root = root + self.requests = [] + self.available = True + owner = self class H(BaseHTTPRequestHandler): def do_GET(self): + owner.requests.append(self.path) + if not owner.available: + self.send_error(503) + return f = self.server.root / self.path.lstrip("/") # type: ignore[attr-defined] if not f.is_file(): self.send_error(404) @@ -113,6 +75,7 @@ class _Server: def stop(self): self.httpd.shutdown() self.httpd.server_close() + self.thread.join(timeout=5) @pytest.fixture() @@ -158,30 +121,23 @@ def test_stage_cache_correctness(tmp_path, lib_source, corruption): assert {p.name for p in out.glob("*.so*")} == names manifest = json.loads(out.parent.joinpath("manifest.json").read_text()) - # True cache hit: source deleted, no downloads possible. - server.stop() + # Same URL/table throughout: corruption must invalidate output evidence, + # not accidentally trigger the independent table-identity check. + server.available = False import shutil shutil.rmtree(tmp_path / "payload" / ".work") + requests = list(server.requests) assert srl.stage(tmp_path / "payload", table) == out - assert {p.name for p in out.glob("*.so*")} == names - - # Corrupt the cache and restore the source so a rebuild is possible. + assert server.requests == requests + server.available = True if corruption == "missing": (out / "liba.so").unlink() elif corruption == "extra": (out / "libjunk.so").write_bytes(b"bogus") else: (out / "libb.so").write_bytes(b"corrupted bytes") - - src = tmp_path / "debs" - server2 = _Server(src) - for name in table: - table[name]["url"] = f"{server2.url}/{name}.deb" - try: - result = srl.stage(tmp_path / "payload", table) - assert result == out - finally: - server2.stop() + assert srl.stage(tmp_path / "payload", table) == out + assert len(server.requests) > len(requests) assert {p.name for p in out.glob("*.so*")} == names for name in names: assert hashlib.sha256((out / name).read_bytes()).hexdigest() == \ @@ -201,8 +157,9 @@ def test_collision_identical_ok_conflicting_raises(tmp_path, lib_source): (tmp_path / "debs" / "libc.deb").read_bytes()).hexdigest() # Force a miss: current manifest no longer validates for libc's bytes. - with pytest.raises(Exception): + with pytest.raises(srl.StageError, match="soname collision.*libb.so"): srl.stage(tmp_path / "payload", table) + assert not (out.parent / "manifest.json").exists() def test_rebuild_removes_superseded_license_files(tmp_path, lib_source): diff --git a/tests/test_yaml_indent_consistency_31999.py b/tests/test_yaml_indent_consistency_31999.py index 13f4adada6..705e52bb25 100644 --- a/tests/test_yaml_indent_consistency_31999.py +++ b/tests/test_yaml_indent_consistency_31999.py @@ -1,52 +1,26 @@ """All YAML write paths use indented block sequences (#31999).""" -import io - import pytest import hermes_yaml as yaml from utils import atomic_roundtrip_yaml_update, atomic_yaml_write -def test_safe_dump_produces_indented_lists(): - data = {"custom_providers": [{"name": "NVIDIA", "base_url": "https://api.nvidia.com"}]} - out = yaml.safe_dump(data) - assert "\n - " in out - assert yaml.safe_load(out) == data - - -def test_safe_and_roundtrip_writers_use_the_same_layout(): - data = {"items": [{"key": "value1"}, {"key": "value2"}]} - stream = io.StringIO() - yaml.roundtrip_yaml().dump(data, stream) - assert yaml.safe_dump(data, sort_keys=False) == stream.getvalue() - - def test_atomic_write_then_key_update_keeps_layout_and_values(tmp_path): - data = {"custom_providers": [{"name": "Test", "base_url": "https://example.com"}]} + data = {"custom_providers": [{"name": "Tëst 🦀", "base_url": "https://example.com"}]} path = tmp_path / "config.yaml" atomic_yaml_write(path, data) initial = path.read_text(encoding="utf-8") atomic_roundtrip_yaml_update(path, "approvals.mode", "off") content = path.read_text(encoding="utf-8") + assert "Tëst 🦀" in content + assert not list(tmp_path.glob(".config_*.tmp")) + assert yaml.roundtrip_yaml().load(content) == {**data, "approvals": {"mode": "off"}} assert content.startswith(initial) assert "\n - " in content assert yaml.safe_load(content) == {**data, "approvals": {"mode": "off"}} -def test_atomic_yaml_write_preserves_unicode(tmp_path): - path = tmp_path / "config.yaml" - atomic_yaml_write(path, {"name": "Tëst Näme 🦀"}) - assert "Tëst Näme 🦀" in path.read_text(encoding="utf-8") - - -def test_atomic_yaml_write_is_atomic(tmp_path): - path = tmp_path / "config.yaml" - atomic_yaml_write(path, {"key": "value"}) - assert yaml.safe_load(path.read_text(encoding="utf-8")) == {"key": "value"} - assert not list(tmp_path.glob(".config_*.tmp")) - - def test_failed_atomic_yaml_write_keeps_original(tmp_path): path = tmp_path / "config.yaml" original = "# keep original\nkey: value\n" @@ -55,16 +29,3 @@ def test_failed_atomic_yaml_write_keeps_original(tmp_path): atomic_yaml_write(path, {"object": object()}) assert path.read_text(encoding="utf-8") == original assert not list(tmp_path.glob(".config_*.tmp")) - - -def test_atomic_yaml_write_loads_in_roundtrip_editor(tmp_path): - data = { - "custom_providers": [ - {"name": "Provider A", "base_url": "https://a.example.com"}, - {"name": "Provider B", "base_url": "https://b.example.com"}, - ], - "fallback_providers": ["backup1", "backup2"], - } - path = tmp_path / "config.yaml" - atomic_yaml_write(path, data) - assert yaml.roundtrip_yaml().load(path.read_text(encoding="utf-8")) == data