"""Exercise deployment shell scripts with isolated host paths and command stubs.""" import os import shutil import subprocess import tempfile import unittest from pathlib import Path BASH = ("C:/Program Files/Git/bin/bash.exe" if os.name == "nt" else shutil.which("bash")) SCRIPTS = Path(__file__).resolve().parent @unittest.skipUnless(BASH and Path(BASH).exists(), "Bash unavailable") class DeploymentTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.deploy = self.root / "deploy" self.bin = self.root / "bin" self.systemd = self.root / "systemd" for directory in (self.deploy, self.bin, self.systemd): directory.mkdir() self.trace = self.root / "trace" self.active = self.root / "active" # Git Bash on Windows prepends its binaries when importing Windows PATH. # Source our isolated PATH after shell initialization, including re-exec. shell_bin = self.bin.as_posix() if os.name == "nt": shell_bin = "/" + shell_bin[0].lower() + shell_bin[2:] shell_env = self.root / "bash-env" shell_env.write_text('export PATH="' + shell_bin + ':$PATH"\n', encoding="utf-8") self.env = {**os.environ, "PATH": str(self.bin) + os.pathsep + os.environ["PATH"], "BASH_ENV": shell_env.as_posix(), "TRACE": self.trace.as_posix(), "ACTIVE": self.active.as_posix()} for key in ("PIPU_UPDATE_REEXEC", "PIPU_UPDATE_SOURCE", "PIPU_UPDATE_COMMIT", "PIPU_UPDATE_REF", "PIPU_DATA", "SUDO_USER"): self.env.pop(key, None) (self.deploy / ".env").write_text("PIPU_DATA=" + (self.root / "data").as_posix() + "\nPIPU_VERSION=1.9.0\nJWT_PRIVATE_KEY=fixture\nSTREAM_SIGNING_KEY=fixture\nGITEA_URL=https://git.example.org\nGITEA_REPO=pipu/music\n", encoding="utf-8") self.stub("systemctl", 'echo "systemctl $*" >> "$TRACE"\ncase "$1" in is-active) test -f "$ACTIVE";; restart) touch "$ACTIVE";; esac\n') self.stub("id", 'echo "${TEST_UID:-0}"\n') self.stub("chown", 'exit 0\n') self.stub("python3", 'exit 0\n') self.stub("flock", 'exit 0\n') def stub(self, name, body): path = self.bin / name path.write_text("#!/usr/bin/env bash\n" + body, encoding="utf-8", newline="\n") path.chmod(0o755) def run_script(self, name, *args, extra=None): return subprocess.run([BASH, (self.deploy / name).as_posix(), *args], cwd=self.deploy, env={**self.env, **(extra or {})}, text=True, encoding="utf-8", errors="replace", capture_output=True, timeout=20) def install_fixture(self): source = (SCRIPTS / "install-management.sh").read_text(encoding="utf-8") # Relocate host constants only; run the real installer logic unchanged. for original, replacement in (("/etc/systemd/system", self.systemd), ("/var/lib/pipu-management", self.root / "state"), ("/usr/local/lib/pipu", self.root / "installed")): source = source.replace(original, replacement.as_posix()) (self.deploy / "install-management.sh").write_text(source, encoding="utf-8", newline="\n") (self.deploy / "server-agent.py").write_text("# worker fixture\n", encoding="utf-8") def test_ensure_installs_starts_and_preserves_unchanged_running_worker(self): self.install_fixture() first = self.run_script("install-management.sh", "--ensure") self.assertEqual(0, first.returncode, first.stderr) self.assertTrue((self.root / "installed/server-agent.py").is_file()) self.assertTrue((self.root / "data/management").is_dir()) unit = (self.systemd / "pipu-management.service").read_text(encoding="utf-8") canonical = subprocess.run([BASH, "-c", 'cd "$1" && pwd -P', "test", self.deploy.as_posix()], capture_output=True, text=True, check=True).stdout.strip() self.assertIn(canonical, unit) self.assertTrue(self.active.exists()) self.trace.write_text("", encoding="utf-8") again = self.run_script("install-management.sh", "--ensure") self.assertEqual(0, again.returncode, again.stderr) self.assertNotIn("restart", self.trace.read_text()) self.assertNotIn("daemon-reload", self.trace.read_text()) (self.deploy / "server-agent.py").write_text("# upgraded worker\n", encoding="utf-8") updated = self.run_script("install-management.sh", "--ensure") self.assertEqual(0, updated.returncode, updated.stderr) self.assertIn("restart", self.trace.read_text()) def test_ensure_restarts_stopped_worker(self): self.install_fixture() self.assertEqual(0, self.run_script("install-management.sh", "--ensure").returncode) self.active.unlink() result = self.run_script("install-management.sh", "--ensure") self.assertEqual(0, result.returncode, result.stderr) self.assertTrue(self.active.exists()) def test_agent_owned_update_never_replaces_or_restarts_itself(self): self.install_fixture() self.active.touch() result = self.run_script("install-management.sh", "--ensure", extra={"PIPU_UPDATE_SOURCE": "https://git.example.org/pipu/music"}) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual("systemctl is-active --quiet pipu-management.service\n", self.trace.read_text()) self.assertFalse((self.root / "installed").exists()) def test_nonroot_setup_delegates_privileged_step_to_sudo(self): self.install_fixture() self.stub("sudo", 'printf "%s\\n" "$@" > "$TRACE"\n') result = self.run_script("install-management.sh", "--ensure", extra={"TEST_UID": "1000"}) self.assertEqual(0, result.returncode, result.stderr) arguments = self.trace.read_text().splitlines() self.assertEqual(["--", "env"], arguments[:2]) self.assertEqual("PIPU_DATA=" + (self.root / "data").as_posix(), arguments[2]) self.assertEqual("--ensure", arguments[-1]) def update_fixture(self, install_exit=0): shutil.copyfile(SCRIPTS / "update.sh", self.deploy / "update.sh") target = self.root / "target-installer" target.write_text('echo "setup $*" >> "$TRACE"\nexit ' + str(install_exit) + '\n', encoding="utf-8", newline="\n") self.env["TARGET_INSTALLER"] = target.as_posix() (self.deploy / "install-management.sh").write_text("echo 'old setup must not run' >&2; exit 99\n", encoding="utf-8") self.stub("git", 'echo "git $*" >> "$TRACE"\ncase "$1" in checkout) cp "$TARGET_INSTALLER" install-management.sh;; describe) echo v1.9.1;; rev-parse) echo aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa;; esac\n') self.stub("docker", 'echo "docker $*" >> "$TRACE"\n') def test_update_installs_target_revision_worker_before_container_operations(self): self.update_fixture() result = self.run_script("update.sh", "main") self.assertEqual(0, result.returncode, result.stderr) trace = self.trace.read_text() self.assertLess(trace.index("git checkout"), trace.index("setup --ensure")) self.assertLess(trace.index("setup --ensure"), trace.index("docker compose")) self.assertIn("up -d --remove-orphans", trace) self.assertIn("PIPU_COMMIT=" + "a" * 40, (self.deploy / ".env").read_text()) def test_failed_host_setup_stops_update_before_container_operations(self): self.update_fixture(install_exit=1) result = self.run_script("update.sh", "main") self.assertNotEqual(0, result.returncode) self.assertNotIn("docker ", self.trace.read_text()) self.assertIn("PIPU_VERSION=1.9.0", (self.deploy / ".env").read_text()) if __name__ == "__main__": unittest.main()