102 lines
4.4 KiB
Python
102 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the actual maintenance script with isolated state and fake external tools."""
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
RUNNER = ROOT / "libexec/hermes-stack-hermes-update"
|
|
|
|
|
|
class HermesUpdateTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="hermes-update-", dir=ROOT.parent)
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.bin = self.root / "bin"
|
|
self.bin.mkdir()
|
|
config = self.root / "config/hermes-stack-maintenance"
|
|
config.mkdir(parents=True)
|
|
(self.root / "hermes-agent/.git").mkdir(parents=True)
|
|
self.env = dict(os.environ, XDG_CONFIG_HOME=str(config.parent),
|
|
XDG_STATE_HOME=str(self.root / "state"),
|
|
XDG_RUNTIME_DIR=str(self.root), TEST_ROOT=str(self.root),
|
|
PATH=str(self.bin) + ":" + os.environ["PATH"],
|
|
TEST_UPDATE_EXIT="0", TEST_CHANGED="0", TEST_HEALTHY="1")
|
|
(config / "config.env").write_text(
|
|
f'HERMES_REPO={shlex.quote(str(self.root / "hermes-agent"))}\n'
|
|
f'NOTIFIER={shlex.quote(str(self.bin / "notify-test"))}\n'
|
|
'HERMES_REMOTE=origin\nHERMES_BRANCH=main\n'
|
|
'GATEWAY_SERVICE=test-gateway.service\n'
|
|
'GATEWAY_HEALTH_ATTEMPTS=1\nGATEWAY_HEALTH_DELAY=0\n'
|
|
'LLAMA_URL=http://unused.invalid\n')
|
|
self.program("git", '''case "$*" in
|
|
*"status --porcelain"*|*"fetch --no-tags"*) exit 0 ;;
|
|
*"rev-parse HEAD"*)
|
|
if [[ -f "$TEST_ROOT/changed" ]]; then printf 'c%.0s' {1..40};
|
|
else printf 'a%.0s' {1..40}; fi ;;
|
|
*"rev-parse refs/remotes/"*) printf 'b%.0s' {1..40} ;;
|
|
*) exit 99 ;;
|
|
esac
|
|
''')
|
|
self.program("hermes", '''case "$*" in
|
|
update)
|
|
[[ "$TEST_CHANGED" == 0 ]] || touch "$TEST_ROOT/changed"
|
|
echo "Simulated updater completion"
|
|
exit "$TEST_UPDATE_EXIT" ;;
|
|
"gateway status") [[ "$TEST_HEALTHY" == 1 ]] ;;
|
|
"config check"|"config migrate") exit 0 ;;
|
|
*) exit 99 ;;
|
|
esac
|
|
''')
|
|
self.program("systemctl", 'printf "%s\\n" "$*" >>"$TEST_ROOT/systemctl-calls"\n')
|
|
self.program("tar", 'printf "%s\\n" "$*" >>"$TEST_ROOT/tar-calls"\n')
|
|
self.program("curl", "echo '{\"status\":\"ok\"}'\n")
|
|
self.program("notify-test", 'printf "%s\\n" "$*" >>"$TEST_ROOT/notices"\n')
|
|
self.program("systemd-cat", "cat >/dev/null\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)
|
|
|
|
def run_update(self, **settings):
|
|
return subprocess.run(["bash", str(RUNNER)], env=dict(self.env, **settings),
|
|
capture_output=True, text=True, timeout=15)
|
|
|
|
def test_stable_unchanged_with_newer_main_succeeds(self):
|
|
result = self.run_update()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("bereits aktuell", result.stdout)
|
|
self.assertIn("restart test-gateway.service", (self.root / "systemctl-calls").read_text())
|
|
self.assertNotIn("zurückgerollt", (self.root / "notices").read_text())
|
|
|
|
def test_unchanged_failed_updater_still_fails(self):
|
|
result = self.run_update(TEST_UPDATE_EXIT="7")
|
|
self.assertEqual(result.returncode, 1, result.stderr)
|
|
self.assertIn("Updater-Exitcode 7", result.stdout)
|
|
|
|
def test_changed_commit_succeeds(self):
|
|
result = self.run_update(TEST_CHANGED="1")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("erfolgreich aktualisiert", result.stdout)
|
|
|
|
def test_changed_commit_with_restart_warning(self):
|
|
result = self.run_update(TEST_CHANGED="1", TEST_UPDATE_EXIT="7")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("Updater-Restartwarnung", result.stdout)
|
|
|
|
def test_unhealthy_gateway_still_rolls_back(self):
|
|
result = self.run_update(TEST_HEALTHY="0")
|
|
self.assertEqual(result.returncode, 1, result.stderr)
|
|
self.assertIn("zurückgerollt", result.stdout)
|
|
self.assertIn("-xf", (self.root / "tar-calls").read_text())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|