import hashlib import importlib.util import json import os from pathlib import Path import shutil import subprocess import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location("prepare", ROOT / "scripts/prepare-workstation.py") prepare = importlib.util.module_from_spec(spec) spec.loader.exec_module(prepare) class PreparationTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.data = b"test model weights" self.item = {"name": "weights.gguf", "sha256": hashlib.sha256(self.data).hexdigest(), "size": len(self.data)} self.group = {"repo": "owner/model", "revision": "a" * 40, "directory": "model/revision", "files": [self.item]} def test_download_uses_fixed_revision_and_publishes_only_verified_file(self): def download(args, **kwargs): self.assertIn("/resolve/" + "a" * 40 + "/weights.gguf", args[-3]) Path(args[-1]).write_bytes(self.data) with patch.object(prepare.subprocess, "run", side_effect=download): paths = prepare.ensure_model(self.group, self.root, True) self.assertEqual(paths[0].read_bytes(), self.data) self.assertFalse(paths[0].with_name("weights.gguf.part").exists()) def test_corrupt_download_cannot_become_live_model(self): with patch.object(prepare.subprocess, "run", side_effect=lambda args, **_: Path(args[-1]).write_bytes(b"broken")): with self.assertRaisesRegex(ValueError, "nicht aktiviert"): prepare.ensure_model(self.group, self.root, True) self.assertFalse((self.root / "model/revision/weights.gguf").exists()) def test_existing_wrong_file_is_preserved_without_download(self): path = self.root / "model/revision/weights.gguf" path.parent.mkdir(parents=True) path.write_bytes(b"user file") with patch.object(prepare.subprocess, "run") as download, self.assertRaises(ValueError): prepare.ensure_model(self.group, self.root, True) download.assert_not_called() self.assertEqual(path.read_bytes(), b"user file") def test_config_merge_preserves_unrelated_settings_and_external_provider(self): config = {"terminal": {"backend": "docker"}, "plugins": {"enabled": ["private"]}, "custom_providers": [{"name": "Other", "base_url": "https://other.invalid/v1", "model": "other"}]} updated = prepare.configure_hermes(config, True) self.assertEqual(updated["terminal"]["backend"], "docker") self.assertEqual(updated["plugins"]["enabled"], ["private"]) self.assertEqual(updated["custom_providers"][0]["model"], "other") self.assertEqual(updated["custom_providers"][1]["extra_body"]["max_tokens"], 2048) self.assertEqual(updated["auxiliary"]["compression"]["base_url"], "http://127.0.0.1:8081/v1") def test_existing_profiles_preserved_and_explicit_selection_backed_up(self): path = self.root / "llama.env" path.write_text("custom tuning\n") prepare.write_config(path, "generated\n") self.assertEqual(path.read_text(), "custom tuning\n") prepare.write_config(path, "generated\n", replace=True) prepare.write_config(path, "generated\n", replace=True) self.assertEqual(len(list(self.root.glob("llama.env.before-setup-*"))), 1) self.assertEqual(path.stat().st_mode & 0o777, 0o600) def test_profiles_quote_other_user_paths_and_do_not_use_remote_hf_start(self): text = prepare.profile_config("routine", [self.root / "user space/model.gguf", self.root / "projector.gguf"]) result = subprocess.run(["bash", "-c", text + '\nprintf "%s|%s|%s" "$LLAMA_MODEL" "$LLAMA_HF_REPO" "${LLAMA_ARGS[*]}"'], text=True, capture_output=True, check=True) self.assertTrue(result.stdout.startswith(str(self.root / "user space/model.gguf") + "||")) self.assertIn("--n-gpu-layers 99 --fit off", result.stdout) def test_prepare_whole_new_workstation_offline_without_home_override(self): repository = self.root / "repository" (repository / "config").mkdir(parents=True) shutil.copy2(ROOT / "config/config.env.example", repository / "config") (repository / "config/models.lock.json").write_text(json.dumps({"models": {"routine": self.group}})) path = self.root / "models/model/revision/weights.gguf" path.parent.mkdir(parents=True) path.write_bytes(self.data) config = self.root / "config/hermes-stack-maintenance" hermes = self.root / "hermes/config.yaml" args = ["prepare", "--models", "routine", "--model-dir", str(self.root / "models"), "--config-dir", str(config), "--hermes-config", str(hermes), "--configure-hermes", "--select", "routine"] with patch.object(prepare, "ROOT", repository), patch("sys.argv", args), \ patch.dict(os.environ, {"XDG_RUNTIME_DIR": str(self.root)}), \ patch.object(prepare.subprocess, "check_output", return_value="inactive\n"): prepare.main() self.assertEqual((config / "llama.env").read_text(), (config / "llama-routine.env").read_text()) self.assertIn("LLAMA_MAX_REQUEST_SECONDS=600", (config / "config.env").read_text()) self.assertIn("run_budget_seconds: 1800", hermes.read_text()) def test_local_cron_unpin_does_not_touch_other_provider_jobs(self): path = self.root / "jobs.json" path.write_text(json.dumps({"jobs": [ {"id": "local", "model": "qwen38-27b-local", "provider": "custom"}, {"id": "cloud", "model": "qwen38-27b-local", "provider": "other"}, {"id": "remote", "model": "qwen38-27b-local", "provider": "custom", "base_url": "https://other.invalid/v1"}]})) with patch.object(prepare.subprocess, "run") as edit: prepare.unpin_local_jobs(path) edit.assert_called_once_with(["hermes", "cron", "edit", "local", "--unpin"], check=True) class ModelSwitchTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.bin = self.root / "bin" self.bin.mkdir() self.config = self.root / "config/hermes-stack-maintenance" self.config.mkdir(parents=True) (self.config / "llama.env").write_text("LLAMA_MODEL=old.gguf\n") (self.config / "llama-routine.env").write_text("LLAMA_MODEL=new.gguf\n") shutil.copy2(ROOT / "libexec/hermes-stack-model", self.bin) for name, body in {"hermes-stack-llama-run": "exit 0", "systemctl": 'echo "$*" >>"$TEST_CALLS"; if [[ "$*" == *"restart llama-qwen"* && "${TEST_FAIL:-0}" == 1 ]]; then exit 1; fi', "curl": "echo '{\"status\":\"ok\"}'"}.items(): p = self.bin / name p.write_text("#!/usr/bin/env bash\n" + body + "\n") p.chmod(0o755) self.env = dict(os.environ, PATH=str(self.bin) + ":" + os.environ["PATH"], XDG_CONFIG_HOME=str(self.config.parent), XDG_RUNTIME_DIR=str(self.root), TEST_CALLS=str(self.root / "calls")) def test_successful_switch_through_path_entrypoint(self): (self.root / "entrypoint").symlink_to(self.bin / "hermes-stack-model") result = subprocess.run(["bash", str(self.root / "entrypoint"), "routine"], env=self.env, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.config / "llama.env").read_text(), "LLAMA_MODEL=new.gguf\n") self.assertIn("start hermes-gateway.service", (self.root / "calls").read_text()) def test_failed_switch_restores_configuration_and_still_attempts_gateway_start(self): result = subprocess.run(["bash", str(self.bin / "hermes-stack-model"), "routine"], env=dict(self.env, TEST_FAIL="1"), capture_output=True, text=True) self.assertNotEqual(result.returncode, 0) self.assertEqual((self.config / "llama.env").read_text(), "LLAMA_MODEL=old.gguf\n") self.assertIn("start hermes-gateway.service", (self.root / "calls").read_text()) if __name__ == "__main__": unittest.main()