mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 13:10:03 +00:00
pulse-triage[bot]
ac4c4cebe4
Compose upstream bed5052eb5 without replacing reviewed history. Keep the accepted one complete stdout/stderr observation and blank-line refusal instead of upstream second-sample replacement. Retain upstream real-systemctl fresh-host assertions and its platform-availability decorator alongside every reviewed failure-retention control. No release, installed recovery or LXC stop-cause claim.
Change-source: pulse-maintainer
443 lines
22 KiB
Python
443 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute server service discovery and its callers with confined manager doubles.
|
|
|
|
Archive/source fixtures use fake versioned executables, not trusted releases or
|
|
real builds. Only the unit-writer fixture relocates its fixed /etc destination.
|
|
The optional real inventory uses systemctl --root, never the host unit manager.
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
INSTALLER = Path(os.environ.get("PULSE_INSTALLER_UNDER_TEST", ROOT / "install.sh")).resolve()
|
|
UPDATER = Path(os.environ.get("PULSE_AUTO_UPDATER_UNDER_TEST", ROOT / "scripts/pulse-auto-update.sh")).resolve()
|
|
REAL_SYSTEMCTL = shutil.which("systemctl")
|
|
QUERY = ["list-unit-files", "--no-legend", "--no-pager", "--full", "--",
|
|
"pulse-backend.service", "pulse.service"]
|
|
# What systemctl prints when the listing itself fails, as opposed to matching nothing.
|
|
FAILED_LISTING = "Failed to get unit file list: No such file or directory\n"
|
|
|
|
SYSTEMCTL = r'''#!/usr/bin/env python3
|
|
import json, os, sys, time
|
|
from pathlib import Path
|
|
f = Path(os.environ["FIXTURE"])
|
|
args = sys.argv[1:]
|
|
with (f / "calls").open("a") as stream:
|
|
stream.write(json.dumps(args) + "\n")
|
|
if args[0] == "list-unit-files":
|
|
mode = os.environ["MODE"]
|
|
if mode == "real":
|
|
os.execv(os.environ["REAL_SYSTEMCTL"],
|
|
[os.environ["REAL_SYSTEMCTL"], "--root=" + str(f / "root")] + args)
|
|
if mode == "first-error-then-absent":
|
|
if len((f / "calls").read_text().splitlines()) == 1:
|
|
os.write(2, b"Failed to get unit file list: Input/output error\n")
|
|
sys.exit(1)
|
|
if mode == "hang":
|
|
time.sleep(30)
|
|
if mode == "large":
|
|
# An early grep match followed by more than a pipe buffer makes the
|
|
# parent's short read fail deterministically, not just emit a warning.
|
|
os.write(1, b"pulse-backend.service enabled enabled\n")
|
|
if "--" not in args:
|
|
try:
|
|
for i in range(512):
|
|
os.write(1, b"unrelated.service disabled enabled\n" * 128)
|
|
except BrokenPipeError:
|
|
(f / "broken-pipe").write_text("producer failed after early close\n")
|
|
os.write(2, b"Failed to print table: Broken pipe\n")
|
|
os._exit(1)
|
|
sys.exit(0)
|
|
os.write(1, os.environ["ROWS"].encode())
|
|
if os.environ.get("ERR"):
|
|
os.write(2, os.environ["ERR"].encode())
|
|
sys.exit(int(os.environ["QUERY_EXIT"]))
|
|
if args[0] == "show":
|
|
state = "inactive" if (f / "stopped").exists() else "active"
|
|
if "--value" in args:
|
|
print(state)
|
|
else:
|
|
print("LoadState=loaded\nActiveState=" + state)
|
|
sys.exit(0)
|
|
if args[0] == "is-active":
|
|
sys.exit(1 if (f / "stopped").exists() else 0)
|
|
if args[0] == "stop":
|
|
(f / "stopped").write_text(args[1])
|
|
sys.exit(0)
|
|
if args[0] == "daemon-reload":
|
|
sys.exit(0)
|
|
sys.exit(97)
|
|
'''
|
|
|
|
HARNESS = r'''
|
|
source "$INSTALLER"
|
|
if [[ "$PRODUCER" == updater ]]; then
|
|
source "$UPDATER"
|
|
# Discovery/handoff uses a manager double and fake script executables,
|
|
# not a managed process. Check the callback's selected service here;
|
|
# real process identity is mandatory in the separate updater controls.
|
|
verify_update_running_executable() {
|
|
printf '%s\n' "$1" > "$FIXTURE/process-readback-service"
|
|
[[ "$1" == pulse-backend ]]
|
|
}
|
|
fi
|
|
print_header() { :; }
|
|
print_info() { echo "$*" >&2; }
|
|
print_warn() { echo "$*" >&2; }
|
|
print_error() { echo "$*" >&2; }
|
|
print_success() { echo "$*" >&2; }
|
|
log() { echo "$*" >&2; }
|
|
check_root() { :; }
|
|
check_proxmox_host() { return 1; }
|
|
detect_os() { :; }
|
|
check_docker_environment() { :; }
|
|
IN_CONTAINER=true
|
|
IN_DOCKER=false
|
|
mark_mutation() { echo "$*" >> "$FIXTURE/mutations"; return 97; }
|
|
case "$FLOW" in
|
|
unavailable)
|
|
command() {
|
|
[[ "$*" != '-v systemctl' ]] || return 1
|
|
builtin command "$@"
|
|
}
|
|
;;
|
|
main|late-main)
|
|
FORCE_VERSION=v6.6.0
|
|
offer_existing_auto_updates() { :; }
|
|
run_upgrade_readiness_preflight() { :; }
|
|
backup_existing() { mark_mutation backup; }
|
|
create_user() { mark_mutation user; }
|
|
install_dependencies() { mark_mutation packages; }
|
|
setup_directories() { mark_mutation directories; }
|
|
download_pulse() { mark_mutation download; }
|
|
install_systemd_service() { mark_mutation unit; }
|
|
ensure_systemd_service_installed() { mark_mutation unit; }
|
|
setup_update_command() { mark_mutation helper; }
|
|
setup_auto_updates() { mark_mutation timer; }
|
|
refresh_auto_updates() { mark_mutation refresh; }
|
|
start_pulse() { mark_mutation start; }
|
|
create_marker_file() { mark_mutation marker; }
|
|
print_completion() { :; }
|
|
if [[ "$FLOW" == late-main ]]; then
|
|
check_existing_installation() { CURRENT_VERSION=v6.5.0; return 0; }
|
|
fi
|
|
;;
|
|
archive|build|automatic)
|
|
verify_release_signature() { :; }
|
|
validate_pulse_binary_architecture() { :; }
|
|
detect_pulse_architecture() { echo amd64; }
|
|
chown() { :; }
|
|
install_additional_agent_binaries() { :; }
|
|
deploy_agent_scripts() { :; }
|
|
if [[ "$FLOW" == automatic ]]; then
|
|
curl() {
|
|
# Fixed fixture response, never a provider request.
|
|
local target="${@: -1}"
|
|
if [[ "$*" == *.sshsig* ]]; then
|
|
printf 'fixture signature\n' > "$target"
|
|
else
|
|
cp "$FIXTURE/served-installer.sh" "$target"
|
|
fi
|
|
}
|
|
fi
|
|
if [[ "$FLOW" == build ]]; then
|
|
apt-get() { :; }
|
|
go() { echo 'go version go1.26.9 linux/amd64'; }
|
|
npm() { :; }
|
|
make() { cp "$FIXTURE/new-pulse" pulse; }
|
|
git() {
|
|
if [[ "$1" == clone ]]; then
|
|
mkdir -p "${@: -1}/frontend-modern"
|
|
else
|
|
echo abc1234
|
|
fi
|
|
}
|
|
fi
|
|
;;
|
|
uninstall)
|
|
quiesce_pulse_for_removal() { mark_mutation "remove $*"; }
|
|
;;
|
|
reset)
|
|
read_pulse_reset_unit_state() { mark_mutation "reset $*"; }
|
|
;;
|
|
unit|ensure-unit)
|
|
# Only fixed path relocation; the function's discovery, branch and
|
|
# redirection remain production code. Never write the host's /etc.
|
|
source "$FIXTURE/unit-writer.sh"
|
|
;;
|
|
esac
|
|
status=0
|
|
# Deliberately invoke inside an OR-list: callers must handle discovery errors
|
|
# explicitly, even when Bash disables errexit throughout the function.
|
|
case "$FLOW" in
|
|
query|unavailable) selected=$(detect_service_name) || status=$?
|
|
[[ "$status" -ne 0 ]] || printf 'SERVICE=%s\n' "$selected" ;;
|
|
existing) check_existing_installation || status=$?
|
|
[[ "$status" -ne 0 ]] || printf 'SERVICE=%s\n' "$SERVICE_NAME" ;;
|
|
main|late-main) main || status=$? ;;
|
|
archive) install_pulse_archive "$FIXTURE/pulse-v6.6.0-linux-amd64.tar.gz" v6.6.0 || status=$? ;;
|
|
build) build_from_source main || status=$? ;;
|
|
automatic) perform_update v6.6.0 || status=$? ;;
|
|
uninstall) uninstall_pulse || status=$? ;;
|
|
reset) reset_pulse || status=$? ;;
|
|
unit) install_systemd_service || status=$? ;;
|
|
ensure-unit) ensure_systemd_service_installed || status=$? ;;
|
|
*) exit 98 ;;
|
|
esac
|
|
exit "$status"
|
|
'''
|
|
|
|
|
|
class ServiceDiscovery(unittest.TestCase):
|
|
def run_fixture(self, *, rows="", mode="rows", query_exit=0, flow="query", explicit=None,
|
|
producer="installer", err="", units_present=True):
|
|
with tempfile.TemporaryDirectory(prefix="pulse-service-discovery-") as directory:
|
|
f = Path(directory)
|
|
for name in ["bin", "install/bin", "config", "units", "root/etc/systemd/system"]:
|
|
(f / name).mkdir(parents=True)
|
|
(f / "bin/systemctl").write_text(SYSTEMCTL)
|
|
(f / "bin/systemctl").chmod(0o755)
|
|
old = b"#!/bin/sh\necho 'Pulse v6.5.0'\n"
|
|
new = b"#!/bin/sh\necho 'Pulse v6.6.0'\n"
|
|
(f / "install/bin/pulse").write_bytes(old)
|
|
(f / "install/bin/pulse").chmod(0o755)
|
|
(f / "new-pulse").write_bytes(new)
|
|
(f / "new-pulse").chmod(0o755)
|
|
(f / "served-installer.sh").write_text(
|
|
'#!/bin/bash\nset -eu\n'
|
|
'printf "%s\\n" "$PULSE_SERVICE_NAME" > "$FIXTURE/installer-service"\n'
|
|
'cp "$FIXTURE/new-pulse" "$PULSE_INSTALL_DIR/bin/pulse"\n')
|
|
archive = f / "pulse-v6.6.0-linux-amd64.tar.gz"
|
|
with tarfile.open(archive, "w:gz") as stream:
|
|
entry = tarfile.TarInfo("pulse")
|
|
entry.size, entry.mode = len(new), 0o755
|
|
stream.addfile(entry, io.BytesIO(new))
|
|
Path(str(archive) + ".sshsig").write_text("signature verification is a double\n")
|
|
source = INSTALLER.read_text()
|
|
start = source.index("install_systemd_service() {")
|
|
end = source.index("\n}\n", start) + 3
|
|
body = source[start:end].replace("/etc/systemd/system/", str(f / "units") + "/")
|
|
(f / "unit-writer.sh").write_text(body)
|
|
unit = "[Unit]\nDescription=Fixture only\n[Service]\nExecStart=/bin/true\n"
|
|
if units_present:
|
|
(f / "root/etc/systemd/system/pulse-backend.service").write_text(unit)
|
|
(f / "root/etc/systemd/system/pulse.service").write_text(unit)
|
|
env = {k: v for k, v in os.environ.items() if not k.startswith("PULSE_")}
|
|
env.update(FIXTURE=str(f), INSTALLER=str(INSTALLER), UPDATER=str(UPDATER),
|
|
PRODUCER=producer, FLOW=flow, MODE=mode,
|
|
ROWS=rows, QUERY_EXIT=str(query_exit), ERR=err,
|
|
REAL_SYSTEMCTL=REAL_SYSTEMCTL or "",
|
|
PATH=str(f / "bin") + os.pathsep + os.environ["PATH"],
|
|
PULSE_INSTALL_DIR=str(f / "install"), PULSE_CONFIG_DIR=str(f / "config"),
|
|
PULSE_BINARY_LINK_PATH=str(f / "pulse-link"))
|
|
if explicit is not None:
|
|
env["PULSE_SERVICE_NAME"] = explicit
|
|
start_time = time.monotonic()
|
|
result = subprocess.run(["bash", "-c", HARNESS], env=env, cwd=f,
|
|
text=True, capture_output=True, timeout=15)
|
|
return dict(result=result, seconds=time.monotonic() - start_time,
|
|
calls=[json.loads(line) for line in (f / "calls").read_text().splitlines()]
|
|
if (f / "calls").exists() else [],
|
|
mutations=(f / "mutations").read_text() if (f / "mutations").exists() else "",
|
|
binary=(f / "install/bin/pulse").read_bytes(), old=old, new=new,
|
|
units=[p.name for p in (f / "units").iterdir()],
|
|
installer_service=(f / "installer-service").read_text()
|
|
if (f / "installer-service").exists() else None,
|
|
process_readback_service=(f / "process-readback-service").read_text()
|
|
if (f / "process-readback-service").exists() else None,
|
|
broken_pipe=(f / "broken-pipe").exists())
|
|
|
|
def assert_refused(self, e):
|
|
self.assertNotEqual(e["result"].returncode, 0, e["result"])
|
|
self.assertIn("Cannot identify the Pulse systemd service", e["result"].stderr)
|
|
self.assertNotIn("SERVICE=", e["result"].stdout)
|
|
self.assertEqual(e["binary"], e["old"])
|
|
self.assertEqual(e["mutations"], "")
|
|
self.assertEqual(e["units"], [])
|
|
self.assertTrue(all(call == QUERY for call in e["calls"]), e["calls"])
|
|
|
|
def test_large_inventory_cannot_short_read_and_hide_legacy_service(self):
|
|
e = self.run_fixture(mode="large")
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=pulse-backend\n")
|
|
self.assertEqual(e["calls"], [QUERY])
|
|
self.assertFalse(e["broken_pipe"])
|
|
|
|
def test_exact_names_and_legacy_precedence_with_state_columns(self):
|
|
for rows, expected in [("", "pulse"), ("\n", "pulse"),
|
|
("pulse.service disabled enabled\n", "pulse"),
|
|
("pulse-backend.service disabled\n", "pulse-backend"),
|
|
("pulse.service static -\npulse-backend.service masked-runtime enabled\n", "pulse-backend"),
|
|
("pulse-backend.service enabled enabled\npulse.service alias -\n", "pulse-backend")]:
|
|
with self.subTest(rows=rows):
|
|
e = self.run_fixture(rows=rows)
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=" + expected + "\n")
|
|
self.assertEqual(e["calls"], [QUERY])
|
|
|
|
def test_explicit_instances_skip_default_inventory(self):
|
|
for service in ["pulse", "pulse-backend", "pulse.department" + "x" * 140]:
|
|
with self.subTest(service=service):
|
|
e = self.run_fixture(explicit=service, query_exit=1)
|
|
self.assertEqual(e["result"].returncode, 0)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=" + service + "\n")
|
|
self.assertEqual(e["calls"], [])
|
|
|
|
def test_non_systemd_install_keeps_default(self):
|
|
e = self.run_fixture(flow="unavailable")
|
|
self.assertEqual(e["result"].returncode, 0)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=pulse\n")
|
|
self.assertEqual(e["calls"], [])
|
|
|
|
def test_partial_failed_inventory_is_not_absence(self):
|
|
failure = "Failed to get unit file list: No such file or directory\n"
|
|
for rows, err in [("", failure), ("pulse-backend.service enabled enabled\n", ""),
|
|
("pulse.service disabled enabled\n", ""),
|
|
("pulse-backend.service enabled enabled\n", failure)]:
|
|
with self.subTest(rows=rows, err=err):
|
|
self.assert_refused(self.run_fixture(rows=rows, err=err, query_exit=1))
|
|
|
|
def test_first_inventory_failure_is_not_replaced_by_silent_absence(self):
|
|
for producer in ("installer", "updater"):
|
|
with self.subTest(producer=producer):
|
|
e = self.run_fixture(producer=producer, mode="first-error-then-absent")
|
|
self.assert_refused(e)
|
|
self.assertEqual(e["calls"], [QUERY],
|
|
"A failed read must not be replaced by a second sample")
|
|
|
|
def test_only_silent_exit_one_is_admitted_as_absence(self):
|
|
for producer in ("installer", "updater"):
|
|
for code in (2, 124, 137):
|
|
with self.subTest(producer=producer, code=code):
|
|
e = self.run_fixture(producer=producer, query_exit=code)
|
|
self.assert_refused(e)
|
|
self.assertEqual(e["calls"], [QUERY])
|
|
|
|
def test_blank_line_output_is_not_a_silent_failed_inventory(self):
|
|
for producer in ("installer", "updater"):
|
|
for rows, err in (("\n", ""), ("", "\n"), ("\n\n", "\n")):
|
|
with self.subTest(producer=producer, rows=rows, err=err):
|
|
e = self.run_fixture(producer=producer, rows=rows, err=err, query_exit=1)
|
|
self.assert_refused(e)
|
|
self.assertEqual(e["calls"], [QUERY])
|
|
|
|
def test_fresh_host_no_matching_unit_files_is_an_empty_inventory(self):
|
|
# systemd exits 1 and prints nothing when none of the named unit files exist, which is
|
|
# every fresh host. The installer and the update helper must both select the default.
|
|
for producer in ("installer", "updater"):
|
|
with self.subTest(producer=producer):
|
|
e = self.run_fixture(producer=producer, query_exit=1)
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=pulse\n")
|
|
# Absence is decided from one complete stdout/stderr observation.
|
|
self.assertEqual(e["calls"], [QUERY])
|
|
|
|
def test_malformed_duplicate_impostor_and_unknown_inventory_stop(self):
|
|
for rows in ["pulse-backendXservice enabled\n", "pulse-backend.service.extra enabled\n",
|
|
"pulseXservice enabled\n", "unrelated.service enabled\n",
|
|
"pulse-backend.service\n", "pulse.service bad enabled\n",
|
|
"pulse.service future-state enabled\n", "pulse.service enabled enabled extra\n",
|
|
"pulse.service enabled\npulse.service disabled\n",
|
|
"pulse-backend.service enabled\npulse-backend.service enabled\n"]:
|
|
with self.subTest(rows=rows):
|
|
self.assert_refused(self.run_fixture(rows=rows))
|
|
|
|
def test_inventory_timeout_is_bounded_and_refuses(self):
|
|
e = self.run_fixture(mode="hang")
|
|
self.assert_refused(e)
|
|
self.assertLess(e["seconds"], 7)
|
|
|
|
def test_existing_install_retains_legacy_service(self):
|
|
e = self.run_fixture(flow="existing", mode="large")
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=pulse-backend\n")
|
|
|
|
def test_failed_discovery_is_distinct_from_no_installation(self):
|
|
e = self.run_fixture(flow="existing", query_exit=1,
|
|
err="Failed to get unit file list: No such file or directory\n")
|
|
self.assert_refused(e)
|
|
self.assertEqual(e["result"].returncode, 2)
|
|
|
|
def test_all_mutating_callers_refuse_when_errexit_is_disabled(self):
|
|
for flow in ["main", "late-main", "archive", "build", "uninstall", "reset", "unit", "ensure-unit"]:
|
|
with self.subTest(flow=flow):
|
|
self.assert_refused(self.run_fixture(flow=flow, query_exit=1, err=FAILED_LISTING))
|
|
|
|
def test_archive_and_source_paths_stop_only_the_selected_legacy_unit(self):
|
|
for flow in ["archive", "build"]:
|
|
with self.subTest(flow=flow):
|
|
e = self.run_fixture(flow=flow, mode="large")
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["binary"], e["new"])
|
|
self.assertIn(["stop", "pulse-backend"], e["calls"])
|
|
self.assertNotIn(["stop", "pulse"], e["calls"])
|
|
self.assertFalse(e["broken_pipe"])
|
|
|
|
@unittest.skipUnless(REAL_SYSTEMCTL, "systemctl unavailable")
|
|
def test_real_systemctl_on_a_host_without_pulse_units_selects_the_default(self):
|
|
# Real systemctl, empty unit root: the genuine exit status for "nothing matched".
|
|
for producer in ("installer", "updater"):
|
|
with self.subTest(producer=producer):
|
|
e = self.run_fixture(mode="real", producer=producer, units_present=False)
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=pulse\n")
|
|
|
|
@unittest.skipUnless(REAL_SYSTEMCTL, "systemctl unavailable")
|
|
def test_real_systemctl_fixture_inventory_has_compatible_rows(self):
|
|
e = self.run_fixture(mode="real")
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=pulse-backend\n")
|
|
self.assertEqual(e["calls"], [QUERY])
|
|
|
|
def test_automatic_helper_preserves_exact_names_and_explicit_instances(self):
|
|
for rows, explicit, expected in [("", None, "pulse"),
|
|
("pulse.service disabled enabled\n", None, "pulse"),
|
|
("pulse-backend.service masked-runtime enabled\n", None, "pulse-backend"),
|
|
("", "pulse.department" + "x" * 140, "pulse.department" + "x" * 140)]:
|
|
with self.subTest(rows=rows, explicit=explicit):
|
|
e = self.run_fixture(producer="updater", rows=rows, explicit=explicit)
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["result"].stdout, "SERVICE=" + expected + "\n")
|
|
self.assertEqual(e["calls"], [] if explicit else [QUERY])
|
|
|
|
def test_automatic_helper_cannot_short_read_the_installer_handoff(self):
|
|
e = self.run_fixture(producer="updater", mode="large", flow="automatic")
|
|
self.assertEqual(e["result"].returncode, 0, e["result"].stderr)
|
|
self.assertEqual(e["installer_service"], "pulse-backend\n")
|
|
self.assertEqual(e["process_readback_service"], "pulse-backend\n")
|
|
self.assertEqual(e["binary"], e["new"])
|
|
self.assertEqual(e["calls"][0], QUERY)
|
|
self.assertFalse(e["broken_pipe"])
|
|
|
|
def test_automatic_helper_refuses_before_backup_or_installer_even_in_or_list(self):
|
|
for rows, status in [("", 1), ("pulse-backend.service enabled enabled\n", 1),
|
|
("pulse-backend.service future-state\n", 0),
|
|
("pulse-backend.service enabled\npulse-backend.service enabled\n", 0),
|
|
("pulse-backendXservice enabled\n", 0),
|
|
("pulse.service enabled enabled extra\n", 0)]:
|
|
with self.subTest(rows=rows, status=status):
|
|
e = self.run_fixture(producer="updater", rows=rows, query_exit=status, flow="automatic",
|
|
err=FAILED_LISTING if (rows, status) == ("", 1) else "")
|
|
self.assert_refused(e)
|
|
self.assertIsNone(e["installer_service"])
|
|
|
|
def test_automatic_helper_timeout_stops_before_service_observation(self):
|
|
e = self.run_fixture(producer="updater", mode="hang", flow="automatic")
|
|
self.assert_refused(e)
|
|
self.assertLess(e["seconds"], 7)
|
|
self.assertIsNone(e["installer_service"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|