mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 21:20:00 +00:00
pulse-triage[bot]
b155982e30
Use the reviewed server-only selector in the existing-installation menu and bind automatic selection to available channel rows. A current selected release is a no-op; unavailable metadata or an automatic downgrade stops before mutations. Preview selection retains its channel, numeric beta/RC revisions compare correctly, and deliberate reinstall pins the actual current version. Execute the actual menu through confined offline transport and stopped mutation boundaries. Preserve explicit version, interactive removal, signature and all-unit quiescence contracts; reconcile the inherited comment-only preview assertion with the real selector consumer. Change-source: pulse-maintainer
256 lines
12 KiB
Python
256 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise the actual existing-installation menu without host mutations.
|
|
|
|
Release transport and all mutation boundaries are confined doubles. These are
|
|
ordinary-user selection/consent controls, not a native installation proof.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
INSTALLER = Path(os.environ.get("PULSE_INSTALLER_UNDER_TEST", ROOT / "install.sh"))
|
|
|
|
|
|
def release(tag, *, draft=False, prerelease=False):
|
|
return {"tag_name": tag, "draft": draft, "prerelease": prerelease}
|
|
|
|
|
|
HARNESS = r'''
|
|
source "$INSTALLER"
|
|
FORCE_VERSION="$PIN"
|
|
FORCE_CHANNEL="$CHANNEL"
|
|
CURRENT_VERSION="$INSTALLED"
|
|
IN_CONTAINER=true
|
|
IN_DOCKER="$DOCKER"
|
|
print_header() { :; }
|
|
print_info() { printf '%s\n' "$*"; }
|
|
print_warn() { printf '%s\n' "$*"; }
|
|
print_error() { printf '%s\n' "$*" >&2; }
|
|
check_root() { :; }
|
|
detect_os() { :; }
|
|
check_docker_environment() { :; }
|
|
check_existing_installation() { return 0; }
|
|
check_proxmox_host() { return 1; }
|
|
create_lxc_container() { exit 96; }
|
|
timeout() { shift; "$@"; }
|
|
curl() {
|
|
local url="${@: -1}"
|
|
printf '%s\n' "$url" >> "$FIXTURE/transport"
|
|
case "$url" in
|
|
https://api.github.com/repos/rcourtman/Pulse/releases)
|
|
printf '%s' "$RELEASES_JSON"
|
|
return "$API_EXIT" ;;
|
|
https://github.com/rcourtman/Pulse/releases/latest)
|
|
[[ -n "$REDIRECT_TAG" ]] || return 22
|
|
printf 'https://github.com/rcourtman/Pulse/releases/tag/%s' "$REDIRECT_TAG" ;;
|
|
*) printf 'unexpected transport: %s\n' "$url" >&2; exit 97 ;;
|
|
esac
|
|
}
|
|
if [[ "$NO_JQ" == true ]]; then
|
|
command() {
|
|
[[ "$*" != '-v jq' ]] || return 1
|
|
builtin command "$@"
|
|
}
|
|
fi
|
|
safe_read() {
|
|
[[ "$MENU_CHOICE" != eof ]] || return 1
|
|
printf -v "$2" '%s' "$MENU_CHOICE"
|
|
}
|
|
# Every reachable mutation is stopped before the real implementation, including
|
|
# removal. A wrong automatic choice is evidence, never an actual uninstall.
|
|
offer_existing_auto_updates() { echo CONSENT_PROMPT >> "$FIXTURE/mutations"; }
|
|
detect_service_name() { echo pulse; }
|
|
run_upgrade_readiness_preflight() { echo PREFLIGHT >> "$FIXTURE/mutations"; }
|
|
backup_existing() { echo BACKUP >> "$FIXTURE/mutations"; }
|
|
create_user() { echo USER >> "$FIXTURE/mutations"; }
|
|
download_pulse() {
|
|
printf 'DOWNLOAD=%s CHANNEL=%s\n' "${LATEST_RELEASE:-unpinned}" "${UPDATE_CHANNEL:-}" >> "$FIXTURE/mutations"
|
|
exit 0
|
|
}
|
|
quiesce_pulse_for_removal() { echo REMOVE >> "$FIXTURE/mutations"; exit 77; }
|
|
main
|
|
'''
|
|
|
|
|
|
class ServerUpdateMenuTest(unittest.TestCase):
|
|
def exercise(self, releases, installed="v6.5.0", channel="stable", *,
|
|
choice="eof", docker=False, configured="stable", api_exit=0,
|
|
redirect="", no_jq=False, pinned="", compact=False):
|
|
payload = (releases if isinstance(releases, str) else
|
|
json.dumps(releases, indent=None if compact else 2))
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
fixture = Path(temporary)
|
|
config = fixture / "config"
|
|
config.mkdir()
|
|
(config / "system.json").write_text(json.dumps({"updateChannel": configured}))
|
|
env = {**os.environ, "INSTALLER": str(INSTALLER), "FIXTURE": str(fixture),
|
|
"PULSE_CONFIG_DIR": str(config), "RELEASES_JSON": payload,
|
|
"INSTALLED": installed, "CHANNEL": channel, "MENU_CHOICE": choice,
|
|
"DOCKER": str(docker).lower(), "API_EXIT": str(api_exit),
|
|
"REDIRECT_TAG": redirect, "NO_JQ": str(no_jq).lower(), "PIN": pinned}
|
|
result = subprocess.run(["bash", "-c", HARNESS], env=env,
|
|
capture_output=True, text=True, timeout=15)
|
|
def lines(name):
|
|
return (fixture / name).read_text().splitlines() if (fixture / name).exists() else []
|
|
return result, lines("mutations"), lines("transport")
|
|
|
|
def assert_unchanged(self, result, mutations, *, refused=False):
|
|
self.assertEqual(result.returncode, 1 if refused else 0, result.stdout + result.stderr)
|
|
self.assertEqual(mutations, [], result.stdout + result.stderr)
|
|
|
|
def assert_download(self, result, mutations, version, channel):
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(mutations[-1:], [f"DOWNLOAD={version} CHANNEL={channel}"])
|
|
self.assertNotIn("REMOVE", mutations)
|
|
|
|
def test_current_preview_without_stable_never_autoselects_removal(self):
|
|
data = [release("v6.6.0-rc.1", prerelease=True)]
|
|
for docker in (False, True):
|
|
with self.subTest(docker=docker):
|
|
r, mutations, _ = self.exercise(data, "v6.6.0-rc.1", "rc", docker=docker)
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_preview_only_row_is_selected_by_its_actual_index(self):
|
|
data = [release("v6.6.0-rc.1", prerelease=True), release("v6.5.0")]
|
|
for docker in (False, True):
|
|
r, mutations, _ = self.exercise(data, channel="rc", docker=docker)
|
|
self.assert_download(r, mutations, "v6.6.0-rc.1", "rc")
|
|
|
|
def test_stable_at_current_version_does_not_switch_to_preview_or_reinstall(self):
|
|
data = [release("v6.6.0-rc.1", prerelease=True), release("v6.5.0")]
|
|
for docker in (False, True):
|
|
r, mutations, _ = self.exercise(data, docker=docker)
|
|
self.assert_unchanged(r, mutations)
|
|
r, mutations, _ = self.exercise([release("v6.5.0")])
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_chart_and_draft_metadata_do_not_enter_the_menu(self):
|
|
data = [release("helm-chart-6.7.0-rc.1", prerelease=True),
|
|
release("v6.7.0-rc.1", draft=True, prerelease=True),
|
|
release("helm-chart-6.6.0"), release("v6.6.0-rc.1", prerelease=True),
|
|
release("v6.5.0")]
|
|
for compact in (False, True):
|
|
r, mutations, _ = self.exercise(data, channel="rc", compact=compact)
|
|
self.assert_download(r, mutations, "v6.6.0-rc.1", "rc")
|
|
self.assertNotIn("Update to helm-chart", r.stdout)
|
|
self.assertNotIn("Update to v6.7.0", r.stdout)
|
|
|
|
def test_new_stable_and_preview_keep_distinct_targets(self):
|
|
data = [release("v6.7.0-rc.1", prerelease=True), release("v6.6.0")]
|
|
for channel, target in (("stable", "v6.6.0"), ("rc", "v6.7.0-rc.1")):
|
|
r, mutations, _ = self.exercise(data, channel=channel)
|
|
self.assert_download(r, mutations, target, channel)
|
|
|
|
def test_preview_channel_can_follow_stable_without_losing_channel_intent(self):
|
|
data = [release("v6.6.0"), release("v6.6.0-rc.1", prerelease=True)]
|
|
r, mutations, _ = self.exercise(data, channel="rc")
|
|
self.assert_download(r, mutations, "v6.6.0", "rc")
|
|
|
|
def test_configured_channel_and_explicit_override_are_respected(self):
|
|
data = [release("v6.6.0-rc.1", prerelease=True), release("v6.5.0")]
|
|
r, mutations, _ = self.exercise(data, channel="", configured="rc")
|
|
self.assert_download(r, mutations, "v6.6.0-rc.1", "rc")
|
|
r, mutations, _ = self.exercise(data, channel="stable", configured="rc")
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_unavailable_or_malformed_metadata_never_falls_into_another_action(self):
|
|
for payload in ([], "", "null", '[{"tag_name":"v6.6.0"',
|
|
[release("helm-chart-6.6.0")], [release("v6.6.0", draft=True)]):
|
|
for channel in ("stable", "rc"):
|
|
with self.subTest(payload=payload, channel=channel):
|
|
r, mutations, _ = self.exercise(payload, channel=channel)
|
|
self.assert_unchanged(r, mutations, refused=True)
|
|
self.assertIn("published Pulse server release", r.stderr)
|
|
|
|
def test_failed_api_payload_and_missing_parser_do_not_admit_preview(self):
|
|
data = [release("v6.6.0-rc.1", prerelease=True), release("v6.5.0")]
|
|
for options in ({"api_exit": 22}, {"no_jq": True}):
|
|
r, mutations, _ = self.exercise(data, channel="rc", **options)
|
|
self.assert_unchanged(r, mutations, refused=True)
|
|
r, mutations, _ = self.exercise(data, channel="rc", redirect="v6.5.0", **options)
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_automatic_downgrade_is_refused_but_deliberate_menu_choice_is_retained(self):
|
|
data = [release("v6.5.0")]
|
|
r, mutations, _ = self.exercise(data, "v6.6.0-rc.1")
|
|
self.assert_unchanged(r, mutations, refused=True)
|
|
self.assertIn("refusing to downgrade automatically", r.stderr)
|
|
r, mutations, _ = self.exercise(data, "v6.6.0-rc.1", choice="1")
|
|
self.assert_download(r, mutations, "v6.5.0", "stable")
|
|
|
|
def test_equivalent_version_spelling_is_an_automatic_noop(self):
|
|
r, mutations, _ = self.exercise([release("6.5.0")])
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_preview_revision_upgrade_uses_numeric_not_lexical_order(self):
|
|
for stage in ("beta", "rc"):
|
|
with self.subTest(stage=stage):
|
|
target = f"v6.6.0-{stage}.10"
|
|
data = [release(target, prerelease=True), release("v6.5.0")]
|
|
r, mutations, _ = self.exercise(data, f"v6.6.0-{stage}.9", "rc")
|
|
self.assert_download(r, mutations, target, "rc")
|
|
r, mutations, _ = self.exercise(data, target, "rc")
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_preview_stage_and_reverse_revision_are_not_downgrades_in_disguise(self):
|
|
data = [release("v6.6.0-rc.1", prerelease=True), release("v6.5.0")]
|
|
r, mutations, _ = self.exercise(data, "v6.6.0-beta.99", "rc")
|
|
self.assert_download(r, mutations, "v6.6.0-rc.1", "rc")
|
|
for target, installed in (("v6.6.0-rc.9", "v6.6.0-rc.10"),
|
|
("v6.6.0-beta.99", "v6.6.0-rc.1")):
|
|
data = [release(target, prerelease=True), release("v6.5.0")]
|
|
r, mutations, _ = self.exercise(data, installed, "rc")
|
|
self.assert_unchanged(r, mutations, refused=True)
|
|
|
|
def test_explicit_reinstall_pins_current_version_not_latest(self):
|
|
data = [release("v6.7.0-rc.1", prerelease=True), release("v6.6.0")]
|
|
r, mutations, _ = self.exercise(data, choice="3")
|
|
self.assert_download(r, mutations, "v6.5.0", "stable")
|
|
r, mutations, _ = self.exercise(data, installed="unknown", choice="3")
|
|
self.assert_unchanged(r, mutations, refused=True)
|
|
|
|
def test_deliberate_update_rows_remove_and_cancel_have_exact_meaning(self):
|
|
data = [release("v6.7.0-rc.1", prerelease=True), release("v6.6.0")]
|
|
for choice, target, channel in (("1", "v6.6.0", "stable"),
|
|
("2", "v6.7.0-rc.1", "rc")):
|
|
r, mutations, _ = self.exercise(data, choice=choice)
|
|
self.assert_download(r, mutations, target, channel)
|
|
r, mutations, _ = self.exercise(data, choice="4")
|
|
self.assertEqual(r.returncode, 77)
|
|
self.assertEqual(mutations, ["REMOVE"])
|
|
r, mutations, _ = self.exercise(data, choice="5")
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_manual_rows_stay_correct_when_stable_update_is_absent(self):
|
|
data = [release("v6.6.0-rc.1", prerelease=True), release("v6.5.0")]
|
|
r, mutations, _ = self.exercise(data, choice="1")
|
|
self.assert_download(r, mutations, "v6.6.0-rc.1", "rc")
|
|
r, mutations, _ = self.exercise(data, choice="2")
|
|
self.assert_download(r, mutations, "v6.5.0", "stable")
|
|
r, mutations, _ = self.exercise(data, choice="3")
|
|
self.assertEqual(r.returncode, 77)
|
|
self.assertEqual(mutations, ["REMOVE"])
|
|
r, mutations, _ = self.exercise(data, choice="4")
|
|
self.assert_unchanged(r, mutations)
|
|
|
|
def test_invalid_or_empty_manual_choice_never_mutates(self):
|
|
for choice in ("", "0", "99", "not-a-number"):
|
|
r, mutations, _ = self.exercise([release("v6.6.0")], choice=choice)
|
|
self.assert_unchanged(r, mutations, refused=True)
|
|
|
|
def test_explicit_version_bypasses_menu_and_discovery(self):
|
|
r, mutations, transport = self.exercise([], pinned="v6.6.0-rc.1")
|
|
self.assert_download(r, mutations, "v6.6.0-rc.1", "")
|
|
self.assertEqual(transport, [])
|
|
self.assertNotIn("What would you like to do?", r.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|