mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 13:10:03 +00:00
rcourtman
d8806e3c20
The 2026-10-10 server service discovery change lists the two historical unit names with one pattern read and refuses to continue when that read exits non-zero. systemctl list-unit-files exits 1, printing nothing on stdout or stderr, when none of the named unit files exist. That is every fresh host, so install.sh refused with "Cannot identify the Pulse systemd service: unit inventory failed" before it did any work. The first beta's install.sh smoke test caught it on a clean Debian 12 container; no release carries it, and v6.5.0 listed the whole inventory and grepped it. The tests could not see it. Their systemctl double reports "no units" as exit 0, and test_partial_failed_inventory_is_not_absence asserted that an empty inventory with exit 1 must be refused, which is the defect itself. Add one inventory read shared by the installer and the unattended updater, and use it for the auto-update mask check, which had the same hazard. It accepts exactly the silent exit 1 as an empty inventory, and only after a second read confirms there is still no output and no message. A timeout, a signal, any other status, or an exit 1 that carries output or a message stays a failed inventory, so a real failure still cannot pass as absence. Make the doubles model systemd: a failure now prints its message, silent exit 1 is absence, and a new test runs real systemctl --root over an empty unit root for both scripts. Run against the unfixed scripts, the four new fresh-host tests fail; the rest of the suite is unchanged. The registered runtime proofs for install.sh and the unattended updater, which drive this suite, now require the fresh-host controls to have run and passed, so they cannot be skipped or deleted unnoticed. Refs: #2785
405 lines
20 KiB
Python
405 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Actual installer timer discovery and refresh, confined to temporary files.
|
|
|
|
The server/install producers and systemd manager operations are doubles. The
|
|
optional real systemctl inventory uses --root, never the host's unit manager.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
INSTALLER = Path(os.environ.get("PULSE_INSTALLER_UNDER_TEST", ROOT / "install.sh"))
|
|
REAL_SYSTEMCTL = shutil.which("systemctl")
|
|
|
|
SYSTEMCTL = r'''#!/usr/bin/env bash
|
|
set -eu
|
|
printf '%s\n' "$*" >> "$FIXTURE/calls"
|
|
case "$1" in
|
|
list-unit-files)
|
|
# The one-unit read discovers the timer; the two-unit read admits
|
|
# replacement only after checking the effective masks of both units.
|
|
if [[ "$#" == 7 ]]; then
|
|
if [[ "$GUARD_INVENTORY" == real ]]; then
|
|
exec "$REAL_SYSTEMCTL" --root="$FIXTURE/root" "$@"
|
|
fi
|
|
if [[ "$GUARD_INVENTORY" == hang ]]; then exec sleep 30; fi
|
|
printf '%s' "$GUARD_ROWS"
|
|
printf '%s' "${GUARD_ERR:-}" >&2
|
|
exit "$GUARD_EXIT"
|
|
fi
|
|
if [[ "$INVENTORY" == real ]]; then
|
|
exec "$REAL_SYSTEMCTL" --root="$FIXTURE/root" "$@"
|
|
fi
|
|
if [[ "$INVENTORY" == hang ]]; then exec sleep 30; fi
|
|
printf '%s' "$ROWS"
|
|
exit "$QUERY_EXIT"
|
|
;;
|
|
daemon-reload) exit 0 ;;
|
|
*) echo "Unexpected manager operation: $*" >&2; exit 97 ;;
|
|
esac
|
|
'''
|
|
|
|
HARNESS = r'''
|
|
source "$INSTALLER"
|
|
IN_CONTAINER=true
|
|
IN_DOCKER=false
|
|
CURRENT_VERSION=v6.5.0
|
|
print_header() { :; }
|
|
print_info() { echo "$*"; }
|
|
print_warn() { echo "$*"; }
|
|
print_success() { echo "$*"; }
|
|
print_error() { echo "$*" >&2; }
|
|
check_root() { :; }
|
|
detect_os() { :; }
|
|
check_docker_environment() { :; }
|
|
check_existing_installation() { [[ "$FLOW" != fresh ]]; }
|
|
detect_service_name() { echo "$SERVICE_NAME"; }
|
|
run_upgrade_readiness_preflight() { :; }
|
|
backup_existing() { :; }
|
|
create_user() { :; }
|
|
install_dependencies() { :; }
|
|
setup_directories() { :; }
|
|
download_pulse() { :; }
|
|
build_from_source() { :; }
|
|
setup_update_command() { :; }
|
|
ensure_systemd_service_installed() { :; }
|
|
install_systemd_service() { :; }
|
|
start_pulse() { :; }
|
|
create_marker_file() { :; }
|
|
print_completion() { :; }
|
|
resolve_latest_release_tag_for_channel() {
|
|
if [[ "$1" == stable ]]; then echo v6.6.0; else echo v6.7.0-rc.1; fi
|
|
}
|
|
safe_read() {
|
|
printf '%s\n' "$1" >> "$FIXTURE/prompts"
|
|
if [[ "$2" == enable_updates ]]; then
|
|
printf -v "$2" '%s' n
|
|
elif [[ "$FLOW" == reinstall ]]; then
|
|
printf -v "$2" '%s' 3
|
|
else
|
|
printf -v "$2" '%s' 1
|
|
fi
|
|
}
|
|
case "$FLOW" in
|
|
query)
|
|
if update_timer_exists; then echo PRESENT; else echo ABSENT; fi
|
|
exit 0 ;;
|
|
unavailable)
|
|
command() {
|
|
[[ "$*" != '-v systemctl' ]] || return 1
|
|
builtin command "$@"
|
|
}
|
|
if update_timer_exists; then echo PRESENT; else echo ABSENT; fi
|
|
exit 0 ;;
|
|
refresh-unavailable)
|
|
command() {
|
|
[[ "$*" != '-v systemctl' ]] || return 1
|
|
builtin command "$@"
|
|
}
|
|
refresh_auto_updates
|
|
exit 0 ;;
|
|
refresh) refresh_auto_updates; exit 0 ;;
|
|
setup) ENABLE_AUTO_UPDATES=true; setup_auto_updates; exit 0 ;;
|
|
repair-entry) exec bash "$INSTALLER" --repair-auto-update-units ;;
|
|
assets)
|
|
result=0
|
|
install_auto_update_assets || result=$?
|
|
exit "$result" ;;
|
|
version) FORCE_VERSION=v6.6.0 ;;
|
|
source) BUILD_FROM_SOURCE=true ;;
|
|
update|reinstall|fresh) : ;;
|
|
*) exit 98 ;;
|
|
esac
|
|
main
|
|
'''
|
|
|
|
|
|
class ServerUpdateTimerTest(unittest.TestCase):
|
|
def exercise(self, *, flow="query", state="enabled", service="pulse",
|
|
rows=None, query_exit=0, inventory="double", mask="",
|
|
config_enabled=True, mask_location="configured", guard_rows=None,
|
|
guard_exit=0, guard_inventory=None, guard_err=""):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
fixture = Path(temporary)
|
|
config = fixture / "config"
|
|
install = fixture / "install"
|
|
units = fixture / "root/etc/systemd/system"
|
|
tools = fixture / "tools"
|
|
for directory in (config, install / "scripts", units, tools):
|
|
directory.mkdir(parents=True)
|
|
helper = tools / "auto-update"
|
|
helper.write_text("#!/bin/bash\necho stale-helper\n")
|
|
(install / "scripts/pulse-auto-update.sh").write_text(
|
|
"#!/bin/bash\necho current-helper\n")
|
|
settings = json.dumps({"autoUpdateEnabled": config_enabled, "updateChannel": "stable"})
|
|
(config / "system.json").write_text(settings)
|
|
timer_name = f"{service}-update.timer"
|
|
service_name = f"{service}-update.service"
|
|
timer = units / timer_name
|
|
updater = units / service_name
|
|
installed_units = units
|
|
if mask and mask_location != "configured":
|
|
installed_units = fixture / "root/usr/lib/systemd/system"
|
|
installed_units.mkdir(parents=True)
|
|
if mask_location == "persistent-other":
|
|
# Caller-selected destinations are not the directories in
|
|
# systemd's effective search path. Neither is the mask.
|
|
destination = fixture / "root/opt/selected-units"
|
|
destination.mkdir(parents=True)
|
|
timer = destination / timer_name
|
|
updater = destination / service_name
|
|
(installed_units / timer_name).write_text(
|
|
"[Timer]\nOnCalendar=daily\n[Install]\nWantedBy=timers.target\n")
|
|
(installed_units / service_name).write_text("[Service]\nExecStart=/bin/true\n")
|
|
if state == "enabled":
|
|
wants = units / "timers.target.wants"
|
|
wants.mkdir()
|
|
(wants / timer_name).symlink_to(
|
|
"/" + str((installed_units / timer_name).relative_to(fixture / "root")))
|
|
masked = None
|
|
if mask:
|
|
mask_name = timer_name if mask == "timer" else service_name
|
|
if mask_location == "configured":
|
|
masked = timer if mask == "timer" else updater
|
|
else:
|
|
mask_dir = (fixture / "root/run/systemd/system"
|
|
if mask_location == "runtime" else units)
|
|
mask_dir.mkdir(parents=True, exist_ok=True)
|
|
masked = mask_dir / mask_name
|
|
if masked:
|
|
if masked.exists() or masked.is_symlink():
|
|
masked.unlink()
|
|
masked.symlink_to("/dev/null")
|
|
systemctl = tools / "systemctl"
|
|
systemctl.write_text(SYSTEMCTL)
|
|
systemctl.chmod(0o755)
|
|
if rows is None:
|
|
rows = f"{timer_name} {state} enabled\n"
|
|
if guard_rows is None:
|
|
timer_state = "masked" if mask == "timer" else state
|
|
service_state = "masked" if mask == "service" else "disabled"
|
|
guard_rows = f"{timer_name} {timer_state} enabled\n{service_name} {service_state} enabled\n"
|
|
env = {**os.environ, "INSTALLER": str(INSTALLER), "FIXTURE": str(fixture),
|
|
"FLOW": flow, "ROWS": rows, "QUERY_EXIT": str(query_exit),
|
|
"GUARD_ROWS": guard_rows, "GUARD_EXIT": str(guard_exit), "GUARD_ERR": guard_err,
|
|
"GUARD_INVENTORY": guard_inventory or inventory,
|
|
"INVENTORY": inventory, "REAL_SYSTEMCTL": REAL_SYSTEMCTL or "",
|
|
"PATH": f"{tools}:{os.environ['PATH']}", "PULSE_SERVICE_NAME": service,
|
|
"PULSE_INSTALL_DIR": str(install), "PULSE_CONFIG_DIR": str(config),
|
|
"PULSE_UPDATE_SERVICE_PATH": str(updater), "PULSE_UPDATE_TIMER_PATH": str(timer),
|
|
"PULSE_AUTO_UPDATE_DEST": str(helper)}
|
|
|
|
def assets():
|
|
result = {}
|
|
for path in sorted(fixture.rglob("*")):
|
|
if path.name in ("calls", "prompts"):
|
|
continue
|
|
mode = path.lstat().st_mode
|
|
if path.is_symlink():
|
|
value = {"mode": mode, "link": os.readlink(path)}
|
|
elif path.is_file():
|
|
value = {"mode": mode, "sha256": hashlib.sha256(path.read_bytes()).hexdigest()}
|
|
else:
|
|
value = {"mode": mode}
|
|
result[str(path.relative_to(fixture))] = value
|
|
return result
|
|
|
|
def effective_states():
|
|
if inventory != "real":
|
|
return None
|
|
observed = subprocess.run(
|
|
[REAL_SYSTEMCTL, f"--root={fixture / 'root'}", "list-unit-files",
|
|
"--no-legend", "--no-pager", "--full", "--", timer_name, service_name],
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(observed.returncode, 0, observed.stdout + observed.stderr)
|
|
return dict(row.split()[:2] for row in observed.stdout.splitlines())
|
|
|
|
before_assets = assets()
|
|
before_states = effective_states()
|
|
result = subprocess.run(["bash", "-c", HARNESS], env=env,
|
|
capture_output=True, text=True, timeout=12)
|
|
def lines(name):
|
|
path = fixture / name
|
|
return path.read_text().splitlines() if path.exists() else []
|
|
evidence = {"result": result, "helper": helper.read_text(),
|
|
"settings": (config / "system.json").read_text(), "original_settings": settings,
|
|
"calls": lines("calls"), "prompts": lines("prompts"),
|
|
"before_assets": before_assets, "after_assets": assets(),
|
|
"before_states": before_states, "after_states": effective_states(),
|
|
"timer_masked": timer.is_symlink() and os.readlink(timer) == "/dev/null",
|
|
"service_masked": updater.is_symlink() and os.readlink(updater) == "/dev/null",
|
|
"timer_enabled": (units / "timers.target.wants" / timer_name).is_symlink()}
|
|
if os.environ.get("PULSE_TIMER_TRACE") == "1":
|
|
def digest(value):
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
|
|
print(json.dumps({"flow": flow, "state": state, "service": service,
|
|
"mask": mask, "mask_location": mask_location, "inventory": inventory,
|
|
"guard_inventory": guard_inventory or inventory, "guard_exit": guard_exit,
|
|
"exit": result.returncode, "calls": evidence["calls"],
|
|
"before_states": before_states, "after_states": evidence["after_states"],
|
|
"before_assets_sha256": digest(before_assets),
|
|
"after_assets_sha256": digest(evidence["after_assets"])}, indent=2))
|
|
return evidence
|
|
|
|
def assert_discovery(self, evidence, present):
|
|
r = evidence["result"]
|
|
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
|
|
self.assertEqual(r.stdout.strip(), "PRESENT" if present else "ABSENT")
|
|
|
|
def assert_refreshed(self, evidence, enabled):
|
|
r = evidence["result"]
|
|
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
|
|
self.assertIn("current-helper", evidence["helper"])
|
|
self.assertNotIn("stale-helper", evidence["helper"])
|
|
self.assertEqual(evidence["settings"], evidence["original_settings"])
|
|
self.assertEqual(evidence["timer_enabled"], enabled)
|
|
mutations = [call for call in evidence["calls"] if not call.startswith("list-unit-files")]
|
|
self.assertEqual(mutations, ["daemon-reload"])
|
|
|
|
def assert_preserved(self, evidence, *, status=0):
|
|
r = evidence["result"]
|
|
self.assertEqual(r.returncode, status, r.stdout + r.stderr)
|
|
self.assertIn("stale-helper", evidence["helper"])
|
|
self.assertEqual(evidence["before_assets"], evidence["after_assets"])
|
|
self.assertEqual(evidence["before_states"], evidence["after_states"])
|
|
self.assertFalse(any(not call.startswith("list-unit-files") for call in evidence["calls"]))
|
|
|
|
def test_discovers_full_systemctl_rows_regardless_of_enablement(self):
|
|
for state in ("enabled", "enabled-runtime", "disabled", "static", "indirect",
|
|
"linked", "linked-runtime", "masked", "masked-runtime"):
|
|
with self.subTest(state=state):
|
|
self.assert_discovery(self.exercise(state=state), True)
|
|
|
|
def test_exact_unit_field_rejects_prefixes_and_regex_impostors(self):
|
|
for rows in ("", "pulse-update.timer-other enabled enabled\n",
|
|
"pulse-updateXtimer enabled enabled\n", "other-update.timer enabled enabled\n"):
|
|
with self.subTest(rows=rows):
|
|
self.assert_discovery(self.exercise(rows=rows), False)
|
|
self.assert_discovery(self.exercise(rows="other.timer enabled enabled\n"
|
|
"\tpulse-update.timer\tdisabled\tenabled\n"), True)
|
|
|
|
def test_query_is_bounded_and_scoped_to_full_exact_instance_name(self):
|
|
service = "pulse-" + "department-" * 12
|
|
e = self.exercise(service=service)
|
|
self.assert_discovery(e, True)
|
|
self.assertEqual(e["calls"], [f"list-unit-files --no-legend --no-pager --full -- {service}-update.timer"])
|
|
|
|
def test_failed_query_never_accepts_its_stdout(self):
|
|
self.assert_discovery(self.exercise(query_exit=1), False)
|
|
|
|
def test_missing_systemctl_never_runs_a_query(self):
|
|
e = self.exercise(flow="unavailable")
|
|
self.assert_discovery(e, False)
|
|
self.assertEqual(e["calls"], [])
|
|
|
|
def test_hung_inventory_times_out_without_manager_mutation(self):
|
|
e = self.exercise(inventory="hang")
|
|
self.assert_discovery(e, False)
|
|
self.assertEqual(len(e["calls"]), 1)
|
|
|
|
def test_all_install_flows_reach_real_refresh_without_changing_consent(self):
|
|
for flow in ("version", "source", "update", "reinstall", "fresh"):
|
|
for state, enabled in (("enabled", True), ("disabled", False)):
|
|
with self.subTest(flow=flow, state=state):
|
|
e = self.exercise(flow=flow, state=state, config_enabled=enabled)
|
|
self.assert_refreshed(e, enabled)
|
|
self.assertNotIn("Automatic updates are not configured.", e["result"].stdout)
|
|
if enabled:
|
|
self.assertFalse(any("Enable auto-updates?" in p for p in e["prompts"]))
|
|
|
|
def test_missing_or_failed_inventory_does_not_refresh_or_enable(self):
|
|
for rows, code in (("", 0), ("pulse-update.timer enabled enabled\n", 1)):
|
|
e = self.exercise(flow="version", rows=rows, query_exit=code)
|
|
self.assertEqual(e["result"].returncode, 0)
|
|
self.assertIn("stale-helper", e["helper"])
|
|
self.assertEqual(e["settings"], e["original_settings"])
|
|
self.assertFalse(any(call == "daemon-reload" for call in e["calls"]))
|
|
|
|
def test_timer_and_updater_masks_are_not_replaced_during_refresh(self):
|
|
for mask in ("timer", "service"):
|
|
with self.subTest(mask=mask):
|
|
e = self.exercise(flow="version", mask=mask)
|
|
self.assert_preserved(e)
|
|
self.assertTrue(e[f"{mask}_masked"])
|
|
self.assertIn("stale-helper", e["helper"])
|
|
self.assertIn("masked", e["result"].stdout)
|
|
self.assertEqual(e["settings"], e["original_settings"])
|
|
self.assertFalse(any(call == "daemon-reload" for call in e["calls"]))
|
|
|
|
@unittest.skipUnless(REAL_SYSTEMCTL, "systemctl filesystem inventory is unavailable")
|
|
def test_all_consumers_preserve_effective_masks_outside_destinations(self):
|
|
for flow in ("version", "source", "update", "reinstall", "fresh"):
|
|
for location, expected in (("runtime", "masked-runtime"), ("persistent-other", "masked")):
|
|
for mask in ("timer", "service"):
|
|
with self.subTest(flow=flow, location=location, mask=mask):
|
|
e = self.exercise(flow=flow, state="disabled", mask=mask,
|
|
mask_location=location, inventory="real", config_enabled=False)
|
|
self.assertEqual(e["before_states"][f"pulse-update.{mask}"], expected)
|
|
self.assert_preserved(e)
|
|
self.assertIn(expected, e["result"].stdout)
|
|
|
|
@unittest.skipUnless(REAL_SYSTEMCTL, "systemctl filesystem inventory is unavailable")
|
|
def test_asset_repair_and_explicit_setup_cannot_replace_effectively_masked_units(self):
|
|
for flow in ("assets", "setup", "repair-entry"):
|
|
for mask in ("timer", "service"):
|
|
with self.subTest(flow=flow, mask=mask):
|
|
e = self.exercise(flow=flow, state="disabled", mask=mask,
|
|
mask_location="runtime", inventory="real", config_enabled=False)
|
|
self.assert_preserved(e, status=1 if flow in ("assets", "repair-entry") else 0)
|
|
self.assertIn("masked-runtime", e["result"].stdout)
|
|
|
|
def test_failed_unknown_and_duplicate_effective_inventory_cannot_admit_writes(self):
|
|
failure = "Failed to get unit file list: No such file or directory\n"
|
|
for rows, code, err in (("pulse-update.timer disabled enabled\n", 1, ""),
|
|
("pulse-update.timer not-understood enabled\n", 0, ""),
|
|
("pulse-update.timer\n", 0, ""),
|
|
("pulse-updateXtimer disabled enabled\n", 0, ""),
|
|
("pulse-update.timer disabled enabled\npulse-update.timer disabled enabled\n", 0, ""),
|
|
("", 1, failure)):
|
|
with self.subTest(rows=rows, code=code, err=err):
|
|
self.assert_preserved(self.exercise(flow="version", guard_rows=rows, guard_exit=code,
|
|
guard_err=err))
|
|
|
|
def test_effective_inventory_is_bounded_and_missing_systemctl_preserves_assets(self):
|
|
self.assert_preserved(self.exercise(flow="version", guard_inventory="hang"))
|
|
e = self.exercise(flow="refresh-unavailable")
|
|
self.assert_preserved(e)
|
|
self.assertEqual(e["calls"], [])
|
|
|
|
def test_successful_absent_or_unmasked_inventory_keeps_asset_repair_working(self):
|
|
# systemd exits 1 and prints nothing when neither unit file exists: absence, not failure.
|
|
for rows, code in (("", 0), ("", 1),
|
|
("pulse-update.timer disabled enabled\npulse-update.service disabled enabled\n", 0)):
|
|
with self.subTest(rows=rows, code=code):
|
|
self.assert_refreshed(self.exercise(flow="assets", state="disabled", guard_rows=rows,
|
|
guard_exit=code), False)
|
|
|
|
def test_effective_inventory_names_are_exact_full_custom_instance_names(self):
|
|
service = "pulse-" + "department-" * 12
|
|
e = self.exercise(flow="version", state="disabled", service=service)
|
|
self.assert_refreshed(e, False)
|
|
self.assertEqual(e["calls"][-2],
|
|
f"list-unit-files --no-legend --no-pager --full -- {service}-update.timer {service}-update.service")
|
|
self.assertTrue(e["calls"][:-2])
|
|
self.assertTrue(all(call == f"list-unit-files --no-legend --no-pager --full -- {service}-update.timer"
|
|
for call in e["calls"][:-2]))
|
|
|
|
@unittest.skipUnless(REAL_SYSTEMCTL, "systemctl filesystem inventory is unavailable")
|
|
def test_real_systemctl_inventory_and_complete_flow_use_only_fixture_root(self):
|
|
for state, enabled in (("enabled", True), ("disabled", False)):
|
|
with self.subTest(state=state):
|
|
self.assert_discovery(self.exercise(inventory="real", state=state), True)
|
|
self.assert_refreshed(self.exercise(flow="version", inventory="real", state=state), enabled)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|