fix(release): preserve failed installer retry status
Cover all retry call sites with isolated regressions in focused checks and installer CI. Add EN/DE operating guidance and record the unreleased Xrechnung and installer audit follow-ups without changing immutable release artifacts. Refs #54
This commit is contained in:
@@ -0,0 +1,154 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user