pipu-music/deploy/test_deployment_scripts.py

"""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;; 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)

    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()