#!/usr/bin/env python3 """Isolierte Vertragstests. Kein HOME-Override, GPU, Netzwerk oder echter systemd-Aufruf.""" import os import json from pathlib import Path import shlex import subprocess import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] RUNNER = ROOT / "libexec/hermes-stack-llama-run" PROFILE = ROOT / "libexec/hermes-stack-llama-profile" RESTART = ROOT / "libexec/hermes-stack-llama-restart" OPTIONS = """--model --hf-repo --mmproj --alias --host --port --ctx-size --parallel --jinja --reasoning --reasoning-budget --reasoning-preserve --no-reasoning-preserve --flash-attn --threads --threads-batch --n-gpu-layers --fit --fit-target --cache-type-k --cache-type-v --sleep-idle-seconds --image-min-tokens --cors-origins --no-ui --rope-scaling --rope-scale --yarn-orig-ctx --device --no-op-offload --no-kv-offload""" class LlamaTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="hermes-test-", dir=ROOT.parent) self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.config = self.root / "config/hermes-stack-maintenance" self.config.mkdir(parents=True) self.bin = self.root / "bin" self.bin.mkdir() self.record = self.root / "calls" self.env = dict(os.environ, XDG_CONFIG_HOME=str(self.config.parent), XDG_STATE_HOME=str(self.root / "state"), XDG_RUNTIME_DIR=str(self.root), PATH=str(self.bin) + ":" + os.environ["PATH"], TEST_RECORD=str(self.record)) self.server = self.bin / "llama-server" self.write_server() self.write_config() self.program("systemctl", 'printf "%s\\n" "$*" >>"$TEST_RECORD"\nexit 0\n') self.program("notify-test", "exit 0\n") self.program("sleep", "exit 0\n") def program(self, name, body): path = self.bin / name path.write_text("#!/usr/bin/env bash\nset -eu\n" + body) path.chmod(0o755) return path def write_server(self, options=OPTIONS): self.program("llama-server", 'if [[ "${1:-}" == --help ]]; then\n' f' printf "%s\\n" {shlex.quote(options)}\n exit 0\nfi\n' 'printf "%s\\n" "$@" >"$TEST_RECORD"\n') def write_config(self, extra="", profile="baseline", model="", hf="test/model:Q4_K_M"): content = (f'LLAMA_SERVER_BIN={shlex.quote(str(self.server))}\n' f'LLAMA_MODEL={shlex.quote(model)}\nLLAMA_HF_REPO={shlex.quote(hf)}\n' f'LLAMA_PROFILE={shlex.quote(profile)}\n' 'LLAMA_ALIAS=qwen38-27b-local\nLLAMA_ARGS=()\n') (self.config / "llama.env").write_text(content + extra) (self.config / "config.env").write_text( "LLAMA_SERVICE=llama-qwen.service\nGATEWAY_SERVICE=hermes-gateway.service\n" "LLAMA_URL=http://127.0.0.1:8080\n" f'NOTIFIER={shlex.quote(str(self.bin / "notify-test"))}\n') def run_tool(self, script, *args, success=True): result = subprocess.run(["bash", str(script), *args], env=self.env, text=True, capture_output=True, timeout=8) if success: self.assertEqual(result.returncode, 0, result.stdout + result.stderr) else: self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) return result def test_profile_values(self): for name, ctx, margin in [("desktop-64k", "65536", "4096"), ("baseline", "65536", "1536"), ("vram-64k", "65536", "1280"), ("balanced", "49152", "1536"), ("vram-plus", "49152", "1280"), ("compact", "32768", "1536")]: with self.subTest(name=name): result = self.run_tool(RUNNER, "--profile", name, "--print") args = shlex.split(result.stdout) self.assertEqual(args[args.index("--ctx-size") + 1], ctx) self.assertEqual(args[args.index("--fit-target") + 1], margin) self.assertEqual(args.count("--ctx-size"), 1) self.assertIn("qwen38-27b-local", args) def test_cpu_compression_template_uses_separate_config(self): model = self.root / "compression model.gguf" model.touch() template = (ROOT / "config/compression.env.example").read_text() template = template.replace('LLAMA_SERVER_BIN=""', f'LLAMA_SERVER_BIN={shlex.quote(str(self.server))}') template = template.replace( 'LLAMA_MODEL="$HOME/.local/share/hermes-stack/models/qwen3-1.7b-compression/Qwen3-1.7B-Q4_K_M.gguf"', f'LLAMA_MODEL={shlex.quote(str(model))}') config = self.config / "compression.env" config.write_text(template) self.env["HERMES_LLAMA_CONFIG"] = str(config) self.run_tool(RUNNER) args = self.record.read_text().splitlines() self.assertEqual(args[args.index("--port") + 1], "8081") self.assertEqual(args[args.index("--device") + 1], "none") self.assertIn("--no-op-offload", args) self.assertIn("--no-kv-offload", args) self.assertIn(str(model), args) self.assertNotIn("qwen38-27b-local", args) self.assertNotIn("--fit-target", args) def test_check_does_not_run_inference(self): self.run_tool(RUNNER, "--check") self.assertFalse(self.record.exists()) def test_unsupported_option_rejected(self): self.write_server(OPTIONS.replace("--fit-target", "")) result = self.run_tool(RUNNER, "--check", success=False) self.assertIn("--fit-target", result.stderr) self.assertFalse(self.record.exists()) def test_execution_uses_array(self): self.run_tool(RUNNER) args = self.record.read_text().splitlines() self.assertIn("--hf-repo", args) self.assertIn("test/model:Q4_K_M", args) def test_local_legacy_config_and_path_spaces(self): model = self.root / "a model.gguf" model.touch() self.write_config(profile="", model=str(model), hf="", extra="LLAMA_ARGS=(--ctx-size 8192 --parallel 1)\n") self.run_tool(RUNNER) args = self.record.read_text().splitlines() self.assertIn(str(model), args) self.assertNotIn("--fit", args) def test_model_source_conflict(self): self.write_config(model="/some/model.gguf") self.run_tool(RUNNER, "--check", success=False) def test_missing_model(self): self.write_config(model="/does/not/exist.gguf", hf="") self.run_tool(RUNNER, "--check", success=False) def test_missing_model_source(self): self.write_config(hf="") self.run_tool(RUNNER, "--check", success=False) def test_unknown_profile(self): self.run_tool(RUNNER, "--profile", "../unknown", "--check", success=False) def test_profile_rejects_legacy_args(self): self.write_config(extra="LLAMA_ARGS=(--ctx-size 64)\n") self.run_tool(RUNNER, "--check", success=False) def test_profile_selection_preserves_config(self): before = (self.config / "llama.env").read_bytes() self.run_tool(PROFILE, "balanced") self.assertEqual((self.config / "llama.env").read_bytes(), before) self.assertFalse(self.record.exists()) result = self.run_tool(RUNNER, "--print") self.assertIn("49152", result.stdout) self.run_tool(PROFILE, "baseline") self.assertEqual(len(list(self.config.glob("llama-profile.backup.*"))), 1) self.assertEqual((self.config / "llama-profile.env").stat().st_mode & 0o777, 0o600) def test_profile_does_not_migrate_implicitly(self): self.write_config(profile="") self.run_tool(PROFILE, "balanced", success=False) self.assertFalse((self.config / "llama-profile.env").exists()) def test_restart_requires_confirmation(self): self.run_tool(RESTART, success=False) self.assertFalse(self.record.exists()) def test_restart_after_check(self): self.run_tool(RESTART, "--jobs-finished") self.assertEqual(self.record.read_text(), "--user restart llama-qwen.service\n") def test_failed_check_does_not_restart(self): self.write_server("") self.run_tool(RESTART, "--jobs-finished", success=False) self.assertFalse(self.record.exists()) def test_lock_blocks_restart_and_profile_change(self): import fcntl with (self.root / "hermes-stack-hermes-update.lock").open("w") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) self.run_tool(RESTART, "--jobs-finished", success=False) self.run_tool(PROFILE, "balanced", success=False) self.assertFalse(self.record.exists()) self.assertFalse((self.config / "llama-profile.env").exists()) def test_failed_http_does_not_restart_backend(self): self.program("hermes", "exit 0\n") self.program("curl", "exit 28\n") self.run_tool(ROOT / "libexec/hermes-stack-health", success=False) self.assertNotIn("restart llama", self.record.read_text()) def test_installer_requires_explicit_ref(self): self.run_tool(ROOT / "scripts/install-llama-cpp.sh", success=False) self.run_tool(ROOT / "install.sh", "--install-llama", success=False) def test_installer_help(self): self.run_tool(ROOT / "scripts/install-llama-cpp.sh", "--help") self.run_tool(ROOT / "install.sh", "--help") def prepare_build_mocks(self): self.env["CUDACXX"] = str(self.program("nvcc", "exit 0\n")) self.program("nvidia-smi", "exit 0\n") self.program("ninja", "exit 0\n") self.program("c++", "exit 0\n") self.program("git", 'if [[ "$*" == *rev-parse* ]]; then\n' 'printf "%s\\n" 0123456789012345678901234567890123456789\nfi\n') self.program("cmake", 'if [[ "${1:-}" == --build ]]; then\n' ' [[ "${TEST_BUILD_FAIL:-0}" != 1 ]] || exit 1\n' ' mkdir -p "$2/bin"\n' ' for name in llama-server llama-cli llama-bench; do\n' ' cp "$CUDACXX" "$2/bin/$name"\n' ' done\nfi\n') def test_build_publishes_all_three_tools(self): self.prepare_build_mocks() prefix = self.root / "managed" self.run_tool(ROOT / "scripts/install-llama-cpp.sh", "--ref", "master", "--prefix", str(prefix)) for name in ("llama-server", "llama-cli", "llama-bench"): self.assertTrue((prefix / "current/bin" / name).is_file()) manifest = (prefix / "current/BUILD-INFO.txt").read_text() self.assertIn("0123456789012345678901234567890123456789", manifest) self.assertIn("cuda_arch=86", manifest) def test_build_preserves_existing_selection(self): self.prepare_build_mocks() prefix = self.root / "managed" prefix.mkdir() old = prefix / "old" old.mkdir() (prefix / "current").symlink_to(old) self.run_tool(ROOT / "scripts/install-llama-cpp.sh", "--ref", "master", "--prefix", str(prefix)) self.assertEqual((prefix / "current").resolve(), old) def test_failed_build_does_not_activate(self): self.prepare_build_mocks() self.env["TEST_BUILD_FAIL"] = "1" prefix = self.root / "managed" self.run_tool(ROOT / "scripts/install-llama-cpp.sh", "--ref", "master", "--prefix", str(prefix), success=False) self.assertFalse((prefix / "current").exists()) def signal_fixture(self, target, response): env_file = self.root / "hermes.env" env_file.write_text('SIGNAL_HTTP_URL=http://127.0.0.1:8090\n' 'SIGNAL_ACCOUNT=test-account\n' f'SIGNAL_HOME_CHANNEL={shlex.quote(target)}\n') with (self.config / "config.env").open("a") as config: config.write(f'HERMES_ENV={shlex.quote(str(env_file))}\n' 'SIGNAL_NOTIFY=1\nTELEGRAM_NOTIFY=0\n') self.program("systemd-cat", "exit 0\n") self.program("curl", 'while (($#)); do\n' ' if [[ "$1" == --data-binary ]]; then\n' ' printf "%s" "$2" >"$TEST_RECORD"\n fi\n shift\ndone\n' f'printf "%s\\n" {shlex.quote(response)}\n') def test_signal_direct_payload(self): self.signal_fixture("test-recipient", '{"result":{"timestamp":1}}') message = 'Test "Anführungszeichen"\nzweite Zeile' self.run_tool(ROOT / "libexec/hermes-stack-notify", message) payload = json.loads(self.record.read_text()) self.assertEqual(payload["params"]["recipient"], ["test-recipient"]) self.assertEqual(payload["params"]["message"], message) def test_signal_group_payload(self): self.signal_fixture("group:test-group", '{"result":{"timestamp":1}}') self.run_tool(ROOT / "libexec/hermes-stack-notify", "Test") payload = json.loads(self.record.read_text()) self.assertEqual(payload["params"]["groupId"], "test-group") self.assertNotIn("recipient", payload["params"]) def test_signal_rpc_failure_is_nonfatal(self): self.signal_fixture("test-recipient", '{"error":{"code":-1}}') self.run_tool(ROOT / "libexec/hermes-stack-notify", "Test") if __name__ == "__main__": unittest.main(verbosity=2)