mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
fix(post-install): preserve Bashrc on reruns and write failures
This commit is contained in:
martino
@@ -0,0 +1,349 @@
|
||||
"""Run only the Bashrc functions, with all root files in a private fixture."""
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shlex
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
FLOWS = ("auto_post_install.sh", "customizable_post_install.sh")
|
||||
BEGIN = b"# BEGIN PMX_CORE_BASHRC\n"
|
||||
END = b"# END PMX_CORE_BASHRC\n"
|
||||
# The unmarked block shipped in v1.1.3 (fc974995), without its leading blank line.
|
||||
LEGACY = rb'''# ProxMenux customizations
|
||||
export HISTTIMEFORMAT="%d/%m/%y %T "
|
||||
export PS1="\[\e[31m\][\[\e[m\]\[\e[38;5;172m\]\u\[\e[m\]@\[\e[38;5;153m\]\h\[\e[m\] \[\e[38;5;214m\]\W\[\e[m\]\[\e[31m\]]\[\e[m\]\\$ "
|
||||
alias l='ls -CF'
|
||||
alias la='ls -A'
|
||||
alias ll='ls -alF'
|
||||
alias ls='ls --color=auto'
|
||||
alias grep='grep --color=auto'
|
||||
alias fgrep='fgrep --color=auto'
|
||||
alias egrep='egrep --color=auto'
|
||||
source /etc/profile.d/bash_completion.sh
|
||||
'''
|
||||
|
||||
|
||||
def extract(source, name):
|
||||
match = re.search(rf"(?ms)^{name}\(\) \{{\n.*?^\}}$", source)
|
||||
if match is None:
|
||||
raise AssertionError(f"missing function: {name}")
|
||||
return match.group()
|
||||
|
||||
|
||||
class BashrcMigrationTests(unittest.TestCase):
|
||||
def run_flow(self, flow, original, times=1, injection="", style=""):
|
||||
with tempfile.TemporaryDirectory(prefix="bashrc-migration-") as temporary:
|
||||
work = Path(temporary)
|
||||
bashrc = work / ".bashrc"
|
||||
bashrc.write_bytes(original)
|
||||
bashrc.chmod(0o640)
|
||||
before = bashrc.stat()
|
||||
(work / "bin").mkdir()
|
||||
# The embedded Python must use the interpreter running this test.
|
||||
(work / "bin/python3").symlink_to(sys.executable)
|
||||
source = (ROOT / "scripts/post_install" / flow).read_text()
|
||||
functions = "\n".join(extract(source, name) for name in
|
||||
("_migrate_proxmenux_bashrc", "customize_bashrc"))
|
||||
# Relocate only the fixture copy; never source the administrative script.
|
||||
functions = functions.replace("/root/", str(work) + "/")
|
||||
prelude = f'source {shlex.quote(str(ROOT / "scripts/global/pmx_journal.sh"))}\n'
|
||||
prelude += r'''
|
||||
translate() { printf '%s' "$1"; }
|
||||
msg_info() { :; }
|
||||
msg_info2() { :; }
|
||||
msg_warn() { :; }
|
||||
msg_error() { printf '%s\n' "$*" >&2; }
|
||||
msg_ok() { printf 'success\n' >> "$FIXTURE_ROOT/messages"; }
|
||||
register_tool() { printf '%s|%s|%s\n' "$@" >> "$FIXTURE_ROOT/registry"; }
|
||||
'''
|
||||
calls = "\n".join(
|
||||
f'customize_bashrc; rc=$?; printf "%s\\n" "$rc" >> "$FIXTURE_ROOT/status"; '
|
||||
f'cp "$FIXTURE_ROOT/.bashrc" "$FIXTURE_ROOT/pass-{i}"'
|
||||
for i in range(times))
|
||||
env = os.environ.copy()
|
||||
env.update(HOME=str(work), FIXTURE_ROOT=str(work), TMPDIR=str(work),
|
||||
PMX_JOURNAL_ROOT=str(work / "journal"), PMX_BASHRC_PATH_STYLE=style,
|
||||
PATH=str(work / "bin") + os.pathsep + env.get("PATH", ""))
|
||||
result = subprocess.run(
|
||||
["bash", "--noprofile", "--norc", "-c",
|
||||
prelude + functions + "\n" + injection + "\n" + calls],
|
||||
cwd=work, env=env, stdin=subprocess.DEVNULL, capture_output=True,
|
||||
text=True, timeout=15)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
after = bashrc.stat()
|
||||
def read(name):
|
||||
path = work / name
|
||||
return path.read_bytes() if path.exists() else b""
|
||||
return {
|
||||
"passes": [read(f"pass-{i}") for i in range(times)],
|
||||
"status": read("status").splitlines(), "registry": read("registry"),
|
||||
"messages": read("messages"), "backup": read(".bashrc.bak"),
|
||||
"profile": read(".bash_profile"), "stderr": result.stderr,
|
||||
"detected_style": read("style"),
|
||||
"before": (stat.S_IMODE(before.st_mode), before.st_uid, before.st_gid),
|
||||
"after": (stat.S_IMODE(after.st_mode), after.st_uid, after.st_gid),
|
||||
"inode_before": before.st_ino, "inode_after": after.st_ino,
|
||||
"leftovers": [p.name for p in work.iterdir() if "proxmenux" in p.name],
|
||||
"journal": [json.loads(p.read_text()) for p in
|
||||
sorted((work / "journal/spool").glob("*.json"))],
|
||||
}
|
||||
|
||||
def test_inspect_detects_style_without_rewriting_the_original(self):
|
||||
for flow in FLOWS:
|
||||
for block in (LEGACY, LEGACY.replace(rb"\W", rb"\w")):
|
||||
with self.subTest(flow=flow, full=rb"\w" in block):
|
||||
injection = r'''customize_bashrc() {
|
||||
_migrate_proxmenux_bashrc "$FIXTURE_ROOT/.bashrc" inspect > "$FIXTURE_ROOT/style"
|
||||
}'''
|
||||
run = self.run_flow(flow, block, injection=injection)
|
||||
self.assertEqual(run["status"], [b"0"], run["stderr"])
|
||||
self.assertEqual(run["passes"], [block])
|
||||
self.assertEqual(run["detected_style"], b"full\n" if rb"\w" in block else b"short\n")
|
||||
self.assertEqual(run["inode_before"], run["inode_after"])
|
||||
self.assertEqual(run["backup"], b"")
|
||||
self.assertEqual(run["registry"], b"")
|
||||
self.assertEqual(run["journal"], [])
|
||||
|
||||
def test_historical_blocks_migrate_once_with_prompt_backup_and_journal(self):
|
||||
for flow in FLOWS:
|
||||
for full in (False, True):
|
||||
legacy = LEGACY.replace(rb"\W", rb"\w") if full else LEGACY
|
||||
blocks = (legacy, b"${marker_begin}\n" + legacy + b"${marker_end}\n",
|
||||
BEGIN + legacy + END, legacy + BEGIN + legacy + END)
|
||||
for block in blocks:
|
||||
with self.subTest(flow=flow, full=full, block=block[:40]):
|
||||
original = b"# user prefix\n" + block + b"# user suffix\n"
|
||||
run = self.run_flow(flow, original, times=2)
|
||||
self.assertEqual(run["status"], [b"0", b"0"], run["stderr"])
|
||||
first, second = run["passes"]
|
||||
self.assertEqual(first, second)
|
||||
self.assertTrue(first.startswith(b"# user prefix\n# user suffix\n"))
|
||||
self.assertEqual(first.count(BEGIN), 1)
|
||||
self.assertEqual(first.count(END), 1)
|
||||
self.assertEqual(first.count(b"alias ll="), 1)
|
||||
self.assertNotIn(b"${marker_", first)
|
||||
self.assertIn(rb"\w" if full else rb"\W", first)
|
||||
self.assertEqual(run["backup"], original)
|
||||
self.assertEqual(len(run["profile"].splitlines()), 1)
|
||||
self.assertEqual(run["registry"], b"bashrc_custom|true|1.3\n" * 2)
|
||||
self.assertEqual(run["before"], run["after"])
|
||||
self.assertEqual(run["leftovers"], [])
|
||||
entries = [e for e in run["journal"] if e["target"].endswith("/.bashrc")]
|
||||
self.assertEqual(len(entries), 1)
|
||||
entry = entries[0]
|
||||
self.assertEqual(entry["operation"], "write_file")
|
||||
self.assertEqual(entry["exactness"], "exact")
|
||||
self.assertEqual(entry["capture"], "present")
|
||||
self.assertNotEqual(entry["before"], entry["after"])
|
||||
self.assertTrue(all("proxmenux." not in e["target"] for e in run["journal"]))
|
||||
syntax = subprocess.run(["bash", "-n"], input=first, capture_output=True)
|
||||
self.assertEqual(syntax.returncode, 0, syntax.stderr)
|
||||
|
||||
def test_incomplete_and_oversized_ranges_are_preserved_on_rerun(self):
|
||||
fragments = (LEGACY.rsplit(b"source", 1)[0], END,
|
||||
b"${marker_begin}\n# user\n", b"${marker_end}\n",
|
||||
BEGIN + b"# user\n" * 33 + END,
|
||||
b"${marker_begin}\n" + b"# user\n" * 33 + b"${marker_end}\n",
|
||||
BEGIN + b"# user\n${marker_begin}\n# inner\n${marker_end}\n# suffix\n")
|
||||
for flow in FLOWS:
|
||||
for fragment in fragments:
|
||||
with self.subTest(flow=flow, fragment=fragment[:40]):
|
||||
run = self.run_flow(flow, fragment, times=2)
|
||||
self.assertEqual(run["status"], [b"0", b"0"], run["stderr"])
|
||||
# Complete inner ranges can migrate, but their outer prefix cannot.
|
||||
preserved = BEGIN + b"# user\n# suffix\n" if b"# inner" in fragment else fragment
|
||||
self.assertTrue(run["passes"][0].startswith(preserved))
|
||||
self.assertEqual(run["passes"][0], run["passes"][1])
|
||||
|
||||
def test_nested_ranges_with_outer_end_are_preserved_on_repeated_runs(self):
|
||||
markers = ((BEGIN, END), (b"${marker_begin}\n", b"${marker_end}\n"))
|
||||
for flow in FLOWS:
|
||||
for outer_begin, outer_end in markers:
|
||||
for inner_begin, inner_end in markers:
|
||||
with self.subTest(flow=flow, outer=outer_begin, inner=inner_begin):
|
||||
original = (outer_begin + b"# OUTER_USER_PREFIX\n"
|
||||
b"user_function() { :; }\n" + inner_begin
|
||||
+ b"# old inner block\n" + inner_end
|
||||
+ b"# OUTER_USER_SUFFIX\n" + outer_end)
|
||||
run = self.run_flow(flow, original, times=3)
|
||||
self.assertEqual(run["status"], [b"0"] * 3, run["stderr"])
|
||||
for content in run["passes"]:
|
||||
self.assertTrue(content.startswith(original), content)
|
||||
self.assertEqual(content.count(BEGIN), original.count(BEGIN) + 1)
|
||||
self.assertEqual(content.count(END), original.count(END) + 1)
|
||||
self.assertEqual(content.count(b"# ProxMenux core customizations\n"), 1)
|
||||
self.assertEqual(run["passes"], [run["passes"][0]] * 3)
|
||||
self.assertEqual(run["backup"], original)
|
||||
|
||||
def test_crossed_marker_families_preserve_every_byte_on_repeated_runs(self):
|
||||
markers = ((BEGIN, END), (b"${marker_begin}\n", b"${marker_end}\n"))
|
||||
for flow in FLOWS:
|
||||
canonical = self.run_flow(flow, b"")["passes"][0]
|
||||
for outer, inner in (markers, markers[::-1]):
|
||||
for newline in (b"\n", b"\r\n"):
|
||||
with self.subTest(flow=flow, outer=outer[0], newline=newline):
|
||||
original = (outer[0] + b"user_prefix() { :; } # \xff\n"
|
||||
+ inner[0] + outer[1]
|
||||
+ b"user_suffix() { :; }\n" + inner[1])
|
||||
original = original.replace(b"\n", newline)
|
||||
run = self.run_flow(flow, original, times=3)
|
||||
self.assertEqual(run["status"], [b"0"] * 3, run["stderr"])
|
||||
self.assertEqual(run["passes"], [original + canonical] * 3)
|
||||
self.assertEqual(run["backup"], original)
|
||||
self.assertEqual(run["registry"], b"bashrc_custom|true|1.3\n" * 3)
|
||||
|
||||
def test_marker_ambiguity_matrix_preserves_spans_without_capturing_separate_blocks(self):
|
||||
markers = ((BEGIN, END), (b"${marker_begin}\n", b"${marker_end}\n"))
|
||||
prefix = b"user_prefix() { :; }\n"
|
||||
suffix = b"user_suffix() { :; }\n"
|
||||
for flow in FLOWS:
|
||||
canonical = self.run_flow(flow, b"")["passes"][0]
|
||||
for (begin, end), (foreign_begin, foreign_end) in (markers, markers[::-1]):
|
||||
# Each malformed complete span is ambiguous, even if the foreign
|
||||
# token has no partner. Never expose its user bytes on a rerun.
|
||||
ambiguous = {
|
||||
"unmatched-inner-begin": begin + prefix + foreign_begin + suffix + end,
|
||||
"unmatched-foreign-end": begin + prefix + foreign_end + suffix + end,
|
||||
"inner-begin-extra-end": begin + prefix + foreign_begin + end + suffix + end,
|
||||
"nested-with-stray-end": (begin + prefix + foreign_begin + end
|
||||
+ foreign_end + suffix + end),
|
||||
"overlap-chain": (begin + prefix + foreign_begin + end + begin
|
||||
+ foreign_end + suffix + end),
|
||||
"ambiguous-legacy-signature": (begin + prefix + foreign_end
|
||||
+ LEGACY + suffix + end),
|
||||
"oversized-ambiguous-span": (begin + prefix + foreign_end
|
||||
+ b"# user\n" * 33 + suffix + end),
|
||||
}
|
||||
cases = [(name, fragment, fragment) for name, fragment in ambiguous.items()]
|
||||
# An unclosed old BEGIN is not an enclosing range. Its prefix
|
||||
# must survive, but independent complete blocks must still migrate.
|
||||
cases.extend([
|
||||
("orphan-before-same-family", begin + prefix + begin + b"old\n" + end + suffix,
|
||||
begin + prefix + suffix),
|
||||
("orphan-before-other-family", begin + prefix + foreign_begin + b"old\n"
|
||||
+ foreign_end + suffix, begin + prefix + suffix),
|
||||
("missing-outer-end", begin + prefix + foreign_begin + b"old\n"
|
||||
+ foreign_end + suffix + foreign_end, begin + prefix + suffix + foreign_end),
|
||||
("separated-valid-blocks", foreign_end + prefix + begin + b"old\n" + end
|
||||
+ suffix + foreign_begin + b"old\n" + foreign_end + foreign_begin,
|
||||
foreign_end + prefix + suffix + foreign_begin),
|
||||
("limit-32", begin + b"# old\n" * 31 + end + suffix, suffix),
|
||||
("over-limit-33", begin + b"# user\n" * 32 + end + suffix,
|
||||
begin + b"# user\n" * 32 + end + suffix),
|
||||
])
|
||||
for name, original, preserved in cases:
|
||||
with self.subTest(flow=flow, family=begin, case=name):
|
||||
run = self.run_flow(flow, original, times=3)
|
||||
self.assertEqual(run["status"], [b"0"] * 3, run["stderr"])
|
||||
self.assertEqual(run["passes"], [preserved + canonical] * 3)
|
||||
self.assertEqual(run["backup"], original)
|
||||
self.assertEqual(run["before"], run["after"])
|
||||
self.assertEqual(run["leftovers"], [])
|
||||
|
||||
def test_path_override_and_empty_install_remain_idempotent(self):
|
||||
for flow in FLOWS:
|
||||
for style in ("short", "full"):
|
||||
with self.subTest(flow=flow, style=style):
|
||||
run = self.run_flow(flow, b"", times=2, style=style)
|
||||
self.assertEqual(run["status"], [b"0", b"0"], run["stderr"])
|
||||
self.assertEqual(run["passes"][0], run["passes"][1])
|
||||
self.assertIn(rb"\w" if style == "full" else rb"\W", run["passes"][0])
|
||||
|
||||
def test_preparation_and_commit_failures_leave_bashrc_untouched(self):
|
||||
failures = {
|
||||
"inspect": 'python3() { [[ "$3" == inspect ]] && return 71; command python3 "$@"; }',
|
||||
"migrate": 'python3() { [[ "$3" == migrate ]] && return 71; command python3 "$@"; }',
|
||||
"replace": '''python3() {
|
||||
command python3 -c 'import os, sys
|
||||
sys.argv = sys.argv[1:]
|
||||
def fail(*args):
|
||||
raise OSError("injected replace failure")
|
||||
os.replace = fail
|
||||
exec(sys.stdin.read())' "$@"
|
||||
}''',
|
||||
"backup": 'cp() { [[ "${@: -1}" == *.bak ]] && return 71; command cp "$@"; }',
|
||||
"mktemp": 'mktemp() { return 71; }',
|
||||
"copy": 'cp() { if [[ "$1" == -p ]]; then printf partial > "$3"; return 71; fi; command cp "$@"; }',
|
||||
"profile": 'pmx_append_file() { return 71; }',
|
||||
"commit": 'mv() { [[ "${@: -1}" == "$FIXTURE_ROOT/.bashrc" ]] && return 71; command mv "$@"; }',
|
||||
}
|
||||
original = b"# user\n" + LEGACY
|
||||
for flow in FLOWS:
|
||||
for failure, injection in failures.items():
|
||||
with self.subTest(flow=flow, failure=failure):
|
||||
run = self.run_flow(flow, original, injection=injection)
|
||||
self.assertNotEqual(run["status"], [b"0"], run["stderr"])
|
||||
self.assertEqual(run["passes"], [original])
|
||||
self.assertEqual(run["before"], run["after"])
|
||||
self.assertEqual(run["inode_before"], run["inode_after"])
|
||||
self.assertEqual(run["registry"], b"")
|
||||
self.assertEqual(run["messages"], b"")
|
||||
self.assertEqual(run["leftovers"], [])
|
||||
self.assertFalse(any(e["target"].endswith("/.bashrc") for e in run["journal"]))
|
||||
with self.subTest(flow=flow, failure="invalid-style"):
|
||||
run = self.run_flow(flow, original, style="invalid")
|
||||
self.assertNotEqual(run["status"], [b"0"])
|
||||
self.assertEqual(run["passes"], [original])
|
||||
self.assertEqual(run["registry"], b"")
|
||||
|
||||
def test_migration_preserves_unrelated_non_utf8_bytes_and_line_endings(self):
|
||||
prefix = b"# user bytes: \xff\r\n"
|
||||
suffix = b"user_function() { :; }\r\n"
|
||||
original = prefix + BEGIN + b"old block\n" + END + suffix
|
||||
for flow in FLOWS:
|
||||
with self.subTest(flow=flow):
|
||||
run = self.run_flow(flow, original, times=2)
|
||||
self.assertEqual(run["status"], [b"0", b"0"], run["stderr"])
|
||||
self.assertTrue(run["passes"][0].startswith(prefix + suffix))
|
||||
self.assertEqual(run["passes"][0], run["passes"][1])
|
||||
self.assertEqual(run["before"], run["after"])
|
||||
|
||||
def test_failed_block_write_preserves_original_without_success(self):
|
||||
original = b"# user content\n" + BEGIN + b"old managed block\n" + END
|
||||
# Fail the real stdin writer after a partial canonical-block write.
|
||||
# The override is confined to this shell fixture, not production.
|
||||
injection = r'''
|
||||
cat() {
|
||||
if [[ $# == 0 ]]; then
|
||||
local first
|
||||
IFS= read -r first
|
||||
if [[ "$first" == '# BEGIN PMX_CORE_BASHRC' ]]; then
|
||||
printf '# partial block'
|
||||
return 71
|
||||
fi
|
||||
printf '%s\n' "$first"
|
||||
fi
|
||||
command cat "$@"
|
||||
}
|
||||
'''
|
||||
for flow in FLOWS:
|
||||
with self.subTest(flow=flow):
|
||||
run = self.run_flow(flow, original, injection=injection)
|
||||
self.assertNotEqual(run["status"], [b"0"], run["stderr"])
|
||||
self.assertEqual(run["passes"], [original])
|
||||
self.assertEqual(run["before"], run["after"])
|
||||
self.assertEqual(run["inode_before"], run["inode_after"])
|
||||
self.assertEqual(run["registry"], b"")
|
||||
self.assertEqual(run["messages"], b"")
|
||||
self.assertEqual(run["leftovers"], [])
|
||||
|
||||
def test_orphan_begin_cannot_consume_user_content_on_rerun(self):
|
||||
original = BEGIN + b"user_function() { :; }\n# user suffix\n"
|
||||
for flow in FLOWS:
|
||||
with self.subTest(flow=flow):
|
||||
run = self.run_flow(flow, original, times=2)
|
||||
self.assertEqual(run["status"], [b"0", b"0"], run["stderr"])
|
||||
for content in run["passes"]:
|
||||
self.assertTrue(content.startswith(original), content)
|
||||
self.assertEqual(run["passes"][0], run["passes"][1])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user