vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-02 20:29:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 17

Preserve auto-update consent during manual version changes

Browse source 
An update, rollback or reinstall must not silently enable unattended updates. Share the existing-install opt-in prompt and retain affirmative CLI and interactive choices without changing helper refresh or fresh-install defaults.

Exercise all five existing main flows and add bounded configuration/timer intent observations to the signed published lifecycle rehearsal, including changed and unavailable negative controls.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot] 2026-10-01 11:50:33 +01:00
parent cafe38823f
commit eb7e70135d
7 changed files with 403 additions and 112 deletions
Download patch file Download diff file Expand all files Collapse all files

24
docs/release-control/v6/internal/subsystems/deployment-installability.md
View file

@ -2570,6 +2570,30 @@ artifact-selection behaviour.
## Current State
### Existing-install auto-update consent (1 October 2026)
A manual update, version-pinned rollback or reinstall is not consent to turn
on unattended updates. The root server installer's three existing-install
prompt paths share `offer_existing_auto_updates`: missing timers and explicitly
disabled settings default to **No**, including Enter and non-TTY reads. Only
`y`/`yes` (case-insensitive), or the existing explicit enable option, opts in.
Explicit CLI choices are not prompted again. Existing enabled or disabled timer
assets still refresh without changing their enablement; fresh installs remain
opt-in. Readiness, signature validation and persistent-data backup are unchanged.
`auto_update_intent_test.go` executes the real main flows for update, rollback,
same-version reinstall and both menu actions, covering absent/disabled/enabled
timers, Enter, EOF, invalid/no/affirmative input and explicit CLI choices.
`root_install_sh_test.go` binds those paths to the shared choice, and
`build_release_assets_test.go` binds the signed published lifecycle rehearsal to
its new intent check. The rehearsal now retains only the boolean choice and
timer enablement/activity before and after upgrade and rollback, fails changed
or unavailable observations, and does not pass a disable flag to hide the bug.
Its Python tests execute the observer/comparator against intact, changed and
unreadable fixtures. These source proofs are not native installed acceptance;
the published containing installer and both actual lifecycle phases still need
their terminal observations.
### Credential-safe Proxmox bootstrap (1 October 2026)
Current PVE auto-registration metadata accepts the credential-free setup artifact: all command aliases use a private-file handoff and `downloadURL` equals the tokenless script URL. This replaces earlier requirements to embed setup tokens in commands/URLs. The older coherent server artifact is accepted read-only during upgrades, never executed. Host, type, canonical filename/URL, masked hint and live expiry remain required. Root-installer JSON parsing and registration pass secrets through descriptor/stdin input rather than Python/curl argv, and the setup response is no longer persisted as a plaintext /tmp diagnostic. No install source, API scope, trust exception, release selector or success condition is widened.

148
install.sh
View file

@ -699,6 +699,38 @@ safe_read_with_default() {
return 0
}
# A version change is not consent to unattended updates. Keep the existing
# choice (including a disabled timer) and require an affirmative answer when
# offering updates on an installation without an enabled configuration.
offer_existing_auto_updates() {
if [[ "$AUTO_UPDATE_CHOICE_EXPLICIT" == "true" ]] || [[ "$ENABLE_AUTO_UPDATES" == "true" ]] || [[ "$IN_DOCKER" == "true" ]]; then
return 0
fi
local prompt_reason=""
if ! update_timer_exists; then
prompt_reason="missing"
elif [[ -f "$CONFIG_DIR/system.json" ]] && grep -Eq '"autoUpdateEnabled"[[:space:]]*:[[:space:]]*false' "$CONFIG_DIR/system.json"; then
prompt_reason="disabled"
fi
[[ -n "$prompt_reason" ]] || return 0
echo
if [[ "$prompt_reason" == "disabled" ]]; then
echo -e "${YELLOW}Auto-updates are currently disabled.${NC}"
else
echo "Automatic updates are not configured."
fi
echo "Pulse can automatically install stable updates daily (between 2-6 AM)"
echo "Leave this disabled to keep control of version changes and rollbacks."
local enable_updates=""
safe_read_with_default "Enable auto-updates? [y/N]: " enable_updates "n"
if [[ "$enable_updates" =~ ^([Yy]|[Yy][Ee][Ss])$ ]]; then
ENABLE_AUTO_UPDATES=true
fi
return 0
}
wait_for_pulse_ready() {
local pulse_url="$1"
local retries="${2:-60}"
@ -4870,42 +4902,8 @@ main() {
print_info "${action_word} version ${FORCE_VERSION}..."
LATEST_RELEASE="${FORCE_VERSION}"
# Check if auto-updates should be offered when using --version
# Same logic as update/reinstall paths
if [[ "$AUTO_UPDATE_CHOICE_EXPLICIT" != "true" ]] && [[ "$ENABLE_AUTO_UPDATES" != "true" ]] && [[ "$IN_DOCKER" != "true" ]]; then
local should_ask_about_updates=false
local prompt_reason=""
if ! update_timer_exists; then
# Timer doesn't exist - new feature
should_ask_about_updates=true
prompt_reason="new"
elif [[ -f "$CONFIG_DIR/system.json" ]]; then
# Timer exists, check if it's properly configured
if grep -q '"autoUpdateEnabled":\s*false' "$CONFIG_DIR/system.json" 2>/dev/null; then
should_ask_about_updates=true
prompt_reason="disabled"
fi
fi
if [[ "$should_ask_about_updates" == "true" ]]; then
echo
if [[ "$prompt_reason" == "disabled" ]]; then
echo -e "${YELLOW}Auto-updates are currently disabled.${NC}"
echo "Would you like to enable automatic updates?"
else
echo -e "${YELLOW}New feature: Automatic updates!${NC}"
fi
echo "Pulse can automatically install stable updates daily (between 2-6 AM)"
echo "This keeps your installation secure and up-to-date."
safe_read_with_default "Enable auto-updates? [Y/n]: " enable_updates "y"
# Default to yes for this prompt since they're already updating
if [[ ! "$enable_updates" =~ ^[Nn]$ ]]; then
ENABLE_AUTO_UPDATES=true
fi
fi
fi
offer_existing_auto_updates
# Detect the actual service name before trying to stop it
SERVICE_NAME=$(detect_service_name)
@ -5079,43 +5077,7 @@ main() {
print_info "${action_word} $target_version..."
LATEST_RELEASE="$target_version"
# Check if auto-updates should be offered to the user
# Offer if: not already forced by flag, not in Docker, and either:
# 1. Timer doesn't exist (new feature), OR
# 2. Timer exists but autoUpdateEnabled is false (misconfigured)
if [[ "$AUTO_UPDATE_CHOICE_EXPLICIT" != "true" ]] && [[ "$ENABLE_AUTO_UPDATES" != "true" ]] && [[ "$IN_DOCKER" != "true" ]]; then
local should_ask_about_updates=false
local prompt_reason=""
if ! update_timer_exists; then
# Timer doesn't exist - new feature
should_ask_about_updates=true
prompt_reason="new"
elif [[ -f "$CONFIG_DIR/system.json" ]]; then
# Timer exists, check if it's properly configured
if grep -q '"autoUpdateEnabled":\s*false' "$CONFIG_DIR/system.json" 2>/dev/null; then
should_ask_about_updates=true
prompt_reason="disabled"
fi
fi
if [[ "$should_ask_about_updates" == "true" ]]; then
echo
if [[ "$prompt_reason" == "disabled" ]]; then
echo -e "${YELLOW}Auto-updates are currently disabled.${NC}"
echo "Would you like to enable automatic updates?"
else
echo -e "${YELLOW}New feature: Automatic updates!${NC}"
fi
echo "Pulse can automatically install stable updates daily (between 2-6 AM)"
echo "This keeps your installation secure and up-to-date."
safe_read_with_default "Enable auto-updates? [Y/n]: " enable_updates "y"
# Default to yes for this prompt since they're already updating
if [[ ! "$enable_updates" =~ ^[Nn]$ ]]; then
ENABLE_AUTO_UPDATES=true
fi
fi
fi
offer_existing_auto_updates
if ! run_upgrade_readiness_preflight "$CURRENT_VERSION" "$LATEST_RELEASE"; then
exit 1
@ -5147,44 +5109,8 @@ main() {
exit 0
;;
reinstall)
# Check if auto-updates should be offered to the user
# Offer if: not already forced by flag, not in Docker, and either:
# 1. Timer doesn't exist (new feature), OR
# 2. Timer exists but autoUpdateEnabled is false (misconfigured)
if [[ "$AUTO_UPDATE_CHOICE_EXPLICIT" != "true" ]] && [[ "$ENABLE_AUTO_UPDATES" != "true" ]] && [[ "$IN_DOCKER" != "true" ]]; then
local should_ask_about_updates=false
local prompt_reason=""
if ! update_timer_exists; then
# Timer doesn't exist - new feature
should_ask_about_updates=true
prompt_reason="new"
elif [[ -f "$CONFIG_DIR/system.json" ]]; then
# Timer exists, check if it's properly configured
if grep -q '"autoUpdateEnabled":\s*false' "$CONFIG_DIR/system.json" 2>/dev/null; then
should_ask_about_updates=true
prompt_reason="disabled"
fi
fi
if [[ "$should_ask_about_updates" == "true" ]]; then
echo
if [[ "$prompt_reason" == "disabled" ]]; then
echo -e "${YELLOW}Auto-updates are currently disabled.${NC}"
echo "Would you like to enable automatic updates?"
else
echo -e "${YELLOW}New feature: Automatic updates!${NC}"
fi
echo "Pulse can automatically install stable updates daily (between 2-6 AM)"
echo "This keeps your installation secure and up-to-date."
safe_read_with_default "Enable auto-updates? [Y/n]: " enable_updates "y"
# Default to yes for this prompt
if [[ ! "$enable_updates" =~ ^[Nn]$ ]]; then
ENABLE_AUTO_UPDATES=true
fi
fi
fi
offer_existing_auto_updates
backup_existing
stop_pulse_for_update
create_user

151
scripts/installtests/auto_update_intent_test.go Normal file
View file

@ -0,0 +1,151 @@
package installtests
import (
"fmt"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
)
// Exercise the real main flow and read-default wrapper, not a model of the
// prompt. External install mutations are stubs confined to each fixture.
func TestRootInstallExistingAutoUpdatesRequireAffirmativeChoice(t *testing.T) {
type scenario struct {
name, config, answer string
timer, enabled, eof, explicit, optIn, wantSetup bool
}
cases := []scenario{
{name: "missing_timer_enter"},
{name: "missing_timer_non_tty", eof: true},
{name: "disabled_enter", timer: true, config: `{"autoUpdateEnabled":false}`},
{name: "disabled_non_tty", timer: true, config: `{"autoUpdateEnabled":false}`, eof: true},
{name: "disabled_no", timer: true, config: `{"autoUpdateEnabled":false}`, answer: "n"},
{name: "invalid_answer", answer: "maybe"},
{name: "enabled_preserved", timer: true, enabled: true, config: `{"autoUpdateEnabled":true}`},
{name: "disabled_timer_preserved", timer: true, config: `{"autoUpdateEnabled":true}`},
{name: "explicit_disable", explicit: true, answer: "y"},
{name: "explicit_enable", explicit: true, optIn: true, wantSetup: true},
{name: "affirmative_y", answer: "y", wantSetup: true},
{name: "affirmative_yes", answer: "Yes", timer: true, config: `{"autoUpdateEnabled":false}`, wantSetup: true},
}
for _, flow := range []string{"version_upgrade", "version_rollback", "version_reinstall", "menu_update", "menu_reinstall"} {
for _, tc := range cases {
t.Run(flow+"/"+tc.name, func(t *testing.T) {
dir := t.TempDir()
config := filepath.Join(dir, "system.json")
if tc.config != "" {
if err := os.WriteFile(config, []byte(tc.config), 0600); err != nil {
t.Fatal(err)
}
}
installer, err := filepath.Abs(filepath.Join("..", "..", "install.sh"))
if err != nil {
t.Fatal(err)
}
script := `
source "$INSTALLER_UNDER_TEST"
CONFIG_DIR="$FIXTURE_DIR"
INSTALL_DIR="$FIXTURE_DIR/install"
CURRENT_VERSION=v6.4.5
LATEST_RELEASE=v6.4.6
FORCE_VERSION=""
case "$FLOW" in
version_upgrade) FORCE_VERSION=v6.4.6 ;;
version_rollback) FORCE_VERSION=v6.4.1 ;;
version_reinstall) FORCE_VERSION=v6.4.5 ;;
esac
AUTO_UPDATE_CHOICE_EXPLICIT="$EXPLICIT"
ENABLE_AUTO_UPDATES="$OPT_IN"
UPDATE_CHANNEL=stable
for fn in print_header check_root detect_os check_docker_environment backup_existing stop_pulse_for_update create_user download_pulse setup_directories setup_update_command ensure_systemd_service_installed install_systemd_service start_pulse; do
eval "$fn() { :; }"
done
check_proxmox_host() { return 1; }
check_existing_installation() { return 0; }
detect_service_name() { echo pulse; }
resolve_latest_release_tag_for_channel() { echo v6.4.6; }
read_configured_update_channel() { echo stable; }
curl() { :; }
timeout() { shift; "$@"; }
run_upgrade_readiness_preflight() { return 0; }
update_timer_exists() { [[ "$TIMER" == true ]]; }
refresh_auto_updates() { echo REFRESH; }
setup_auto_updates() {
echo SETUP
printf '{"autoUpdateEnabled":true}' > "$CONFIG_DIR/system.json"
TIMER=true
TIMER_ENABLED=true
}
create_marker_file() { echo INSTALL_FINISHED; }
print_completion() { printf 'TIMER_ENABLED=%s\n' "$TIMER_ENABLED"; }
safe_read() {
if [[ "$1" == "Select option "* ]]; then
local selection=1
[[ "$FLOW" == menu_reinstall ]] && selection=2
printf -v "$2" '%s' "$selection"
else
echo AUTO_UPDATE_PROMPT
[[ "$READ_EOF" == true ]] && return 1
printf -v "$2" '%s' "$ANSWER"
fi
}
main
`
cmd := exec.Command("bash", "-c", script)
cmd.Env = append(os.Environ(), "INSTALLER_UNDER_TEST="+installer, "FIXTURE_DIR="+dir,
"FLOW="+flow, "TIMER="+fmt.Sprint(tc.timer), "TIMER_ENABLED="+fmt.Sprint(tc.enabled),
"EXPLICIT="+fmt.Sprint(tc.explicit), "OPT_IN="+fmt.Sprint(tc.optIn),
"READ_EOF="+fmt.Sprint(tc.eof), "ANSWER="+tc.answer)
out, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("main: %v\n%s", err, out)
}
if !strings.Contains(string(out), "INSTALL_FINISHED") {
t.Fatalf("install did not complete:\n%s", out)
}
if got := strings.Contains(string(out), "\nSETUP\n"); got != tc.wantSetup {
t.Fatalf("auto-update setup=%v, want %v:\n%s", got, tc.wantSetup, out)
}
if !tc.wantSetup {
got, readErr := os.ReadFile(config)
if tc.config == "" {
if !os.IsNotExist(readErr) {
t.Fatalf("absent config was changed: %s, %v", got, readErr)
}
} else if readErr != nil || string(got) != tc.config {
t.Fatalf("existing choice was changed: %s, %v", got, readErr)
}
if !strings.Contains(string(out), "TIMER_ENABLED="+fmt.Sprint(tc.enabled)) {
t.Fatalf("timer intent changed:\n%s", out)
}
if tc.timer && !strings.Contains(string(out), "REFRESH") {
t.Fatalf("existing helper was not refreshed:\n%s", out)
}
}
if (tc.explicit || tc.enabled) && strings.Contains(string(out), "AUTO_UPDATE_PROMPT") {
t.Fatalf("existing/explicit choice was re-prompted:\n%s", out)
}
})
}
}
}
func TestRootInstallSafeReadNonTTYDefaultsToNoAutoUpdates(t *testing.T) {
setsid, err := exec.LookPath("setsid")
if err != nil {
t.Skip("setsid unavailable for a controlling-terminal-free read")
}
script := `set -euo pipefail
print_info() { :; }
` + extractRootInstallShellFunction(t, "safe_read") + "\n" + extractRootInstallShellFunction(t, "safe_read_with_default") + `
answer=""
safe_read_with_default "Enable auto-updates? [y/N]: " answer "n"
[[ "$answer" == n ]]
`
out, err := exec.Command(setsid, "bash", "-c", script).CombinedOutput()
if err != nil {
t.Fatalf("non-TTY fallback: %v\n%s", err, out)
}
}

24
scripts/installtests/build_release_assets_test.go
View file

@ -144,6 +144,30 @@ func TestBuildReleaseUsesV6InstallScripts(t *testing.T) {
}
}
// A no-flag, signed published-installer rehearsal must catch re-enabling after
// a version change. The install smoke's explicit disable flag alone cannot.
func TestPublishedLifecycleRehearsalPreservesAutoUpdateChoice(t *testing.T) {
content, err := os.ReadFile(repoFile("scripts", "release_lifecycle_rehearsal.sh"))
if err != nil {
t.Fatal(err)
}
for _, required := range []string{
`auto_update_snapshot > "${WORK_DIR}/state/auto-updates.baseline.tsv"`,
`check_auto_update_intent upgrade || true`,
`check_auto_update_intent rollback || true`,
`cexec '/bin/update --version "$TARGET"'`,
} {
if !strings.Contains(string(content), required) {
t.Fatalf("published installer lifecycle proof missing %s", required)
}
}
for _, line := range strings.Split(string(content), "\n") {
if strings.Contains(line, "/bin/update --version") && strings.Contains(line, "--disable-auto-updates") {
t.Fatal("lifecycle proof must observe installer consent, not bypass it")
}
}
}
func TestSecurityScanRevalidatesLatestStableDelivery(t *testing.T) {
content, err := os.ReadFile(repoFile(".github", "workflows", "security-scan.yml"))
if err != nil {

15
scripts/installtests/root_install_sh_test.go
View file

@ -1480,6 +1480,21 @@ func TestRootInstallScriptUpdateFlowsRefreshExistingAutoUpdateAssets(t *testing.
}
}
// The tested --version, menu-update and menu-reinstall paths must all use the
// same affirmative-only prompt; a new inline default-yes branch is unsafe.
func TestRootInstallExistingFlowsShareExplicitAutoUpdateChoice(t *testing.T) {
main := extractRootInstallShellFunction(t, "main")
if got := strings.Count(main, "offer_existing_auto_updates"); got != 3 {
t.Fatalf("expected three existing-install consent calls, found %d", got)
}
if strings.Contains(main, `Enable auto-updates? [Y/n]`) || strings.Contains(main, `Re-enable auto-updates? [Y/n]`) {
t.Fatal("existing install must not default to enabling auto-updates")
}
if !strings.Contains(main, `safe_read_with_default "Enable auto-updates? [y/N]: " enable_updates "n"`) {
t.Fatal("fresh-install consent must remain opt-in")
}
}
// Regression test for #1526 (and the earlier #1396): when the installer is piped
// to bash (curl ... | bash) there is no source file, so BASH_SOURCE is unset.
// The "am I being sourced?" guard must default the lookup or `set -u` aborts the

58
scripts/release_lifecycle_rehearsal.sh
View file

@ -391,6 +391,59 @@ assert_runtime() {
fi
}
# Only bounded, non-secret intent fields leave the disposable install. A failed
# read is not an absent timer or a disabled setting.
auto_update_snapshot() {
cexec 'set -euo pipefail
config=absent
if [[ -e "$DATA_DIR/system.json" ]]; then
config=$(jq -er '\''if type != "object" then error("invalid system settings")
elif has("autoUpdateEnabled") then
if .autoUpdateEnabled == true then "enabled"
elif .autoUpdateEnabled == false then "disabled"
else error("invalid auto-update choice") end
else "absent" end'\'' "$DATA_DIR/system.json")
fi
load=$(systemctl show -p LoadState --value pulse-update.timer)
case "$load" in
not-found) enabled=absent; active=absent ;;
loaded)
enabled=$(systemctl is-enabled pulse-update.timer || true)
active=$(systemctl is-active pulse-update.timer || true)
case "$enabled" in
enabled|enabled-runtime|disabled|masked|masked-runtime|static|indirect|linked|linked-runtime|generated|transient) ;;
*) exit 1 ;;
esac
case "$active" in
active|inactive|failed|activating|deactivating|reloading) ;;
*) exit 1 ;;
esac ;;
*) exit 1 ;;
esac
printf "%s\t%s\t%s\n" "$config" "$enabled" "$active"' "DATA_DIR=${DATA_DIR}"
}
check_auto_update_intent() {
local label="$1" current before_config before_enabled before_active config enabled active
if [[ ! -s "${WORK_DIR}/state/auto-updates.baseline.tsv" ]] \
|| ! current=$(auto_update_snapshot); then
note_failure "${label}: auto-update intent could not be read or has no baseline"
return 1
fi
IFS=$'\t' read -r before_config before_enabled before_active < "${WORK_DIR}/state/auto-updates.baseline.tsv"
IFS=$'\t' read -r config enabled active <<< "$current"
# A newly persisted false is equivalent to an absent opt-in, not permission
# to enable the timer. Preserve enabled and masked timer choices exactly.
if [[ "$config" != "$before_config" \
&& ! ( "$config" != enabled && "$before_config" != enabled ) ]] \
|| [[ "$enabled" != "$before_enabled" || "$active" != "$before_active" ]]; then
note_failure "${label}: auto-update choice or timer enablement/activity changed"
return 1
fi
printf '%s\n' "$current" > "${WORK_DIR}/state/auto-updates.${label}.tsv"
PHASE_UNITS="${PHASE_UNITS}; auto-updates ${config}/${enabled}/${active} preserved"
}
api_status() {
# api_status METHOD PATH [auth: none|token|basic] [body]
cexec 'args=(-sS -o /dev/null -w "%{http_code}" -X "$METHOD")
@ -626,6 +679,9 @@ phase_seed() {
note_failure "could not record ${DATA_DIR} (or it has no .encryption.key)"
PHASE_DATADIR="not recorded"
fi
if ! auto_update_snapshot > "${WORK_DIR}/state/auto-updates.baseline.tsv"; then
note_failure "could not record the installed auto-update choice and timer state"
fi
record_phase "2. seed" "$FROM_TAG" "-" "-" "$PHASE_SETTINGS" "$PHASE_DATADIR" "-"
}
@ -650,6 +706,7 @@ phase_upgrade() {
log "Phase 3: upgrade with /bin/update --version ${TO_TAG}"
run_updater "$TO_TAG" upgrade || true
assert_runtime "$TO_TAG"
check_auto_update_intent upgrade || true
check_settings upgrade || true
check_datadir upgrade || true
record_phase "3. upgrade (/bin/update)" "$TO_TAG" "$PHASE_VERSION" "$PHASE_HEALTH" "$PHASE_SETTINGS" "$PHASE_DATADIR" "$PHASE_UNITS"
@ -660,6 +717,7 @@ phase_rollback() {
log "Phase 4: roll back with /bin/update --version ${FROM_TAG}"
run_updater "$FROM_TAG" rollback || true
assert_runtime "$FROM_TAG"
check_auto_update_intent rollback || true
check_settings rollback || true
check_datadir rollback || true
record_phase "4. rollback (/bin/update)" "$FROM_TAG" "$PHASE_VERSION" "$PHASE_HEALTH" "$PHASE_SETTINGS" "$PHASE_DATADIR" "$PHASE_UNITS"

95
scripts/tests/test_release_lifecycle_rehearsal.py
View file

@ -5,6 +5,7 @@ from __future__ import annotations
import importlib.util
import json
import os
import re
import subprocess
import sys
@ -150,7 +151,7 @@ class HarnessContractTest(unittest.TestCase):
def test_every_post_change_phase_checks_identity_health_settings_and_data(self) -> None:
for phase in ("phase_upgrade", "phase_rollback"):
body = self.harness.split(f"{phase}() {{", 1)[1].split("\n}\n", 1)[0]
for check in ("assert_runtime", "check_settings", "check_datadir", "record_phase"):
for check in ("assert_runtime", "check_auto_update_intent", "check_settings", "check_datadir", "record_phase"):
with self.subTest(phase=phase, check=check):
self.assertIn(check, body)
@ -176,6 +177,98 @@ class HarnessContractTest(unittest.TestCase):
self.assertEqual(result.returncode, 2, result.stdout + result.stderr)
class AutoUpdateIntentTest(unittest.TestCase):
"""Execute the real observer/comparator, with guest systemctl observations."""
def check(self, baseline: str | None, config: object = None, *,
load: str = "not-found", enabled: str = "disabled",
active: str = "inactive") -> subprocess.CompletedProcess:
harness = HARNESS_PATH.read_text(encoding="utf-8")
functions = "\n".join(
name + "() {" + harness.split(name + "() {", 1)[1].split("\n}\n", 1)[0] + "\n}"
for name in ("auto_update_snapshot", "check_auto_update_intent")
)
script = r'''set -euo pipefail
PHASE_UNITS="pulse active/enabled"
note_failure() { echo "::error::$*"; }
systemctl() {
case "$*" in
"show -p LoadState --value pulse-update.timer") echo "$LOAD" ;;
"is-enabled pulse-update.timer") echo "$ENABLED"; [[ "$ENABLED" == enabled ]] ;;
"is-active pulse-update.timer") echo "$ACTIVE"; [[ "$ACTIVE" == active ]] ;;
*) return 1 ;;
esac
}
export -f systemctl
cexec() { bash -c "$1"; }
''' + functions + '\ncheck_auto_update_intent rollback\nprintf "%s\\n" "$PHASE_UNITS"\n'
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)
(work / "state").mkdir()
data = work / "data"
data.mkdir()
if baseline is not None:
(work / "state" / "auto-updates.baseline.tsv").write_text(baseline + "\n", encoding="utf-8")
if config is not None:
(data / "system.json").write_text(
config if isinstance(config, str) else json.dumps(config), encoding="utf-8")
return subprocess.run(
["bash", "-c", script], capture_output=True, text=True, check=False,
env={**os.environ, "WORK_DIR": str(work), "DATA_DIR": str(data),
"LOAD": load, "ENABLED": enabled, "ACTIVE": active},
)
def test_absent_timer_and_unset_choice_stay_absent(self) -> None:
result = self.check("absent\tabsent\tabsent")
self.assertEqual(0, result.returncode, result.stdout + result.stderr)
self.assertIn("auto-updates absent/absent/absent preserved", result.stdout)
def test_disabled_enabled_and_masked_choices_are_preserved(self) -> None:
for config, enabled, active in ((False, "disabled", "inactive"),
(True, "enabled", "active"),
(False, "masked", "inactive")):
with self.subTest(config=config, timer=enabled):
result = self.check(f"{'enabled' if config else 'disabled'}\t{enabled}\t{active}",
{"autoUpdateEnabled": config}, load="loaded", enabled=enabled, active=active)
self.assertEqual(0, result.returncode, result.stdout + result.stderr)
def test_persisted_false_is_not_a_new_opt_in(self) -> None:
result = self.check("absent\tabsent\tabsent", {"autoUpdateEnabled": False})
self.assertEqual(0, result.returncode, result.stdout + result.stderr)
def test_config_timer_or_activity_changes_fail(self) -> None:
for config, enabled, active in ((True, "disabled", "inactive"),
(False, "enabled", "inactive"),
(False, "disabled", "active")):
with self.subTest(config=config, timer=enabled, active=active):
result = self.check("disabled\tdisabled\tinactive", {"autoUpdateEnabled": config},
load="loaded", enabled=enabled, active=active)
self.assertEqual(1, result.returncode, result.stdout + result.stderr)
self.assertIn("auto-update choice or timer enablement/activity changed", result.stdout)
def test_enabled_choice_cannot_be_silently_disabled(self) -> None:
result = self.check("enabled\tenabled\tactive", {"autoUpdateEnabled": False},
load="loaded", enabled="enabled", active="active")
self.assertEqual(1, result.returncode, result.stdout + result.stderr)
def test_unreadable_choice_or_timer_is_not_assumed_disabled(self) -> None:
for config, load, enabled, active in (("not json", "not-found", "disabled", "inactive"),
({"autoUpdateEnabled": "true"}, "not-found", "disabled", "inactive"),
({"autoUpdateEnabled": None}, "not-found", "disabled", "inactive"),
(None, "", "disabled", "inactive"),
(None, "loaded", "", "inactive"),
(None, "loaded", "disabled", "unknown")):
with self.subTest(config=config, load=load, enabled=enabled, active=active):
result = self.check("absent\tabsent\tabsent", config, load=load, enabled=enabled, active=active)
self.assertEqual(1, result.returncode, result.stdout + result.stderr)
self.assertIn("could not be read", result.stdout)
def test_missing_baseline_cannot_pass(self) -> None:
result = self.check(None)
self.assertEqual(1, result.returncode, result.stdout + result.stderr)
self.assertIn("has no baseline", result.stdout)
GOOD_SNAPSHOT = {
"auth": {"api_token": "200", "password": "200", "requiresAuth": True, "unauthenticated": "401"},
"node": {