govoplan/tests/test_webui_release_dependency_retries.py

from __future__ import annotations

import os
from pathlib import Path
import subprocess
import tempfile
import textwrap
import unittest


META_ROOT = Path(__file__).resolve().parents[1]
INSTALLER = META_ROOT / "tools" / "release" / "install-webui-release-dependencies.sh"
STAGES = ("base", "clone", "modules")


class WebUIReleaseDependencyRetryTests(unittest.TestCase):
    def _run_installer(
        self, stage: str, statuses: tuple[int, ...]
    ) -> tuple[subprocess.CompletedProcess[str], list[str]]:
        with tempfile.TemporaryDirectory(prefix="govoplan-installer-retry-test-") as directory:
            root = Path(directory)
            stub_bin = root / "bin"
            stub_bin.mkdir()
            core_root = root / "core"
            webui = core_root / "web ui"
            webui.mkdir(parents=True)
            work_root = root / "work"
            work_root.mkdir()
            log = root / "commands.log"
            stub = "#!/usr/bin/env bash\nset -euo pipefail\n" + textwrap.dedent(
                r"""
                case "${0##*/}" in
                  node)
                    # Supply the shell's dependency list without requiring Node.
                    printf '%s\t%s\n' '@govoplan/example-webui' \
                      'git+https://example.invalid/module.git#v1.0.0' > "$GOVOPLAN_DEPS"
                    printf 'node\n' >> "$RETRY_TEST_LOG"
                    exit 0
                    ;;
                  sleep)
                    printf 'sleep %s\n' "$*" >> "$RETRY_TEST_LOG"
                    exit 0
                    ;;
                  npm)
                    case "${1:-}" in
                      cache)
                        printf 'cache\n' >> "$RETRY_TEST_LOG"
                        exit 0
                        ;;
                      install)
                        stage=base
                        for argument in "$@"; do
                          if [[ "$argument" == --no-save ]]; then
                            stage=modules
                          fi
                        done
                        ;;
                      *) exit 98 ;;
                    esac
                    ;;
                  git)
                    [[ "${1:-}" == clone ]] || exit 98
                    stage=clone
                    ;;
                  *) exit 98 ;;
                esac

                status=0
                if [[ "$stage" == "$RETRY_TEST_STAGE" ]]; then
                  attempt=0
                  counter="$RETRY_TEST_ROOT/$stage.count"
                  if [[ -f "$counter" ]]; then
                    read -r attempt < "$counter"
                  fi
                  read -r -a statuses <<< "$RETRY_TEST_STATUSES"
                  status="${statuses[$attempt]:-99}"
                  printf '%s\n' "$((attempt + 1))" > "$counter"
                fi
                printf '%s %s\n' "$stage" "$status" >> "$RETRY_TEST_LOG"
                exit "$status"
                """
            )
            for name in ("node", "npm", "git", "sleep"):
                executable = stub_bin / name
                executable.write_text(stub, encoding="utf-8")
                executable.chmod(0o755)

            env = os.environ.copy()
            env.update(
                {
                    "PATH": f"{stub_bin}:{os.defpath}",
                    "TMPDIR": str(work_root),
                    "GOVOPLAN_CORE_ROOT": str(core_root),
                    "GOVOPLAN_WEBUI_PACKAGE_LOCK": "",
                    "GOVOPLAN_WEBUI_PACKAGE_DIR": "",
                    "RETRY_TEST_ROOT": str(root),
                    "RETRY_TEST_LOG": str(log),
                    "RETRY_TEST_STAGE": stage,
                    "RETRY_TEST_STATUSES": " ".join(map(str, statuses)),
                }
            )
            result = subprocess.run(
                [
                    "bash",
                    "-c",
                    'set -euo pipefail; bash "$1" "$2"; '
                    'printf "caller-continued\\n" >> "$RETRY_TEST_LOG"',
                    "retry-test-caller",
                    str(INSTALLER),
                    str(webui),
                ],
                cwd=root,
                env=env,
                text=True,
                capture_output=True,
                timeout=10,
                check=False,
            )
            self.assertEqual(list(work_root.iterdir()), [], result.stderr)
            return result, log.read_text(encoding="utf-8").splitlines()

    def _assert_attempts(self, statuses: tuple[int, ...]) -> None:
        for retried_stage in STAGES:
            with self.subTest(stage=retried_stage, statuses=statuses):
                result, commands = self._run_installer(retried_stage, statuses)
                expected = ["node", "cache"]
                for stage in STAGES:
                    attempts = statuses if stage == retried_stage else (0,)
                    for index, status in enumerate(attempts):
                        expected.append(f"{stage} {status}")
                        if status and index < 2:
                            expected.append(f"sleep {(index + 1) * 10}")
                    if attempts[-1]:
                        break
                if statuses[-1] == 0:
                    expected.append("caller-continued")
                self.assertEqual(result.returncode, statuses[-1], result.stderr)
                self.assertEqual(commands, expected, result.stderr)

    def test_success_on_first_attempt(self) -> None:
        self._assert_attempts((0,))

    def test_success_on_second_attempt(self) -> None:
        self._assert_attempts((17, 0))

    def test_success_on_third_attempt(self) -> None:
        self._assert_attempts((17, 23, 0))

    def test_exhaustion_preserves_final_status_and_stops_callers(self) -> None:
        self._assert_attempts((17, 23, 47))


if __name__ == "__main__":
    unittest.main()