mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-10 12:47:04 +00:00
kevinveenbirkenbachandClaude Opus 5.5
4984f506c7
Every front-proxy vhost wrote its body_filter Lua through a separate stack_host_template task in sys-front-inj-all: two remote round trips when unchanged (stat + file) and four to five when changed, 191 s over 54 vhosts per echoserver deploy and 179 s over 48 per galaxyserver deploy. The Lua now joins http.conf and the vhost in the Deploy task's stack_host_contents call, written first as before, rendered through template_shallow with application_id and domain pinned to the vhost. - sys-front-inj-all/tasks/main.yml no longer writes the Lua; the new standalone.yml includes it and keeps the separate write for matomo's re-inject path, which has no vhost Deploy task - stack_host_contents compares bytes instead of a universal-newline text read, so content with a CR stops reporting changed on every run and an existing non-UTF-8 file is replaced instead of crashing the module - a new required diff_paths argument keeps --diff for the Lua only; vhost and http.conf content stay out of the diff A Lua-only change still reports changed and fires reload-openresty once; the served probe and the fallback redeploy keep their old triggers. Check mode is unchanged. Tests: the real stack_host_template action against the fold expression across 99 roles and TLS/Tor/injection scenarios, byte-identical, with trim and newline negative controls; owner, group and mode parity with the copy module; byte compare with CR and non-UTF-8 files; the wiring of both callers. The saving (about 184 s and 174 s per deploy, net) is projected until the next deploys. 🎯 Fix: 92 % 💥 Break: 2 % 🌱 Efficiency: – ⚡ Speed: – 🛡️ Security: ~100 % 🏆 Score: 90 🚦 Gate: ✅ 90/5, efficiency and speed <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
365 lines
12 KiB
Python
365 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
from ansible.module_utils.basic import AnsibleModule
|
|
from ansible.module_utils.testing import patch_module_args
|
|
from ansible.modules import copy as copy_module
|
|
|
|
from utils.cache.files import read_text
|
|
|
|
from . import PROJECT_ROOT
|
|
|
|
if TYPE_CHECKING:
|
|
from types import ModuleType
|
|
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
MODULE_PATH = PROJECT_ROOT / "library" / "stack_host_contents.py"
|
|
ROLES = PROJECT_ROOT / "roles"
|
|
TODAY_TEMPLATES = (
|
|
ROLES / "sys-front-tls" / "templates" / "http.conf.j2",
|
|
ROLES / "sys-front-inj-all" / "templates" / "body_filter.lua.j2",
|
|
*sorted((ROLES / "sys-svc-proxy" / "templates" / "vhost").glob("*.conf.j2")),
|
|
*sorted(ROLES.glob("*/templates/proxy.conf.j2")),
|
|
)
|
|
|
|
|
|
def _load_stack_host_contents() -> ModuleType:
|
|
spec = importlib.util.spec_from_file_location(
|
|
"library_stack_host_contents", str(MODULE_PATH)
|
|
)
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError(f"Could not load: {MODULE_PATH}")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod) # type: ignore[attr-defined]
|
|
return mod
|
|
|
|
|
|
_MOD = _load_stack_host_contents()
|
|
|
|
|
|
class WriteFileTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.path = Path(self._tmp.name) / "redirect.conf"
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _module(self, *, check_mode=False):
|
|
args = {
|
|
"files": [],
|
|
"mode": "0644",
|
|
"diff_paths": [],
|
|
"_ansible_check_mode": check_mode,
|
|
}
|
|
with patch_module_args(args):
|
|
return AnsibleModule(
|
|
argument_spec=_MOD.ARGUMENT_SPEC, supports_check_mode=True
|
|
)
|
|
|
|
def _write(self, content, *, check_mode=False):
|
|
return _MOD.write_file(
|
|
self._module(check_mode=check_mode), str(self.path), content, "0644"
|
|
)
|
|
|
|
def _mode(self):
|
|
return stat.S_IMODE(self.path.stat().st_mode)
|
|
|
|
def _content(self):
|
|
return self.path.read_text(
|
|
encoding="utf-8"
|
|
) # nocheck: cache-read the file under test changes between reads
|
|
|
|
def test_creates_missing_file_with_mode(self):
|
|
self.assertTrue(self._write("server {}\n"))
|
|
self.assertEqual(self._content(), "server {}\n")
|
|
self.assertEqual(self._mode(), 0o644)
|
|
|
|
def test_unchanged_content_reports_no_change(self):
|
|
self._write("server {}\n")
|
|
self.assertFalse(self._write("server {}\n"))
|
|
|
|
def test_replaces_differing_content(self):
|
|
self._write("server {}\n")
|
|
self.assertTrue(self._write("server { listen 80; }\n"))
|
|
self.assertEqual(self._content(), "server { listen 80; }\n")
|
|
|
|
def test_repairs_mode_drift(self):
|
|
self._write("server {}\n")
|
|
self.path.chmod(0o600)
|
|
self.assertTrue(self._write("server {}\n"))
|
|
self.assertEqual(self._mode(), 0o644)
|
|
|
|
def test_check_mode_reports_without_writing(self):
|
|
self.assertTrue(self._write("server {}\n", check_mode=True))
|
|
self.assertFalse(self.path.exists())
|
|
self._write("server {}\n")
|
|
self.assertTrue(self._write("other\n", check_mode=True))
|
|
self.assertEqual(self._content(), "server {}\n")
|
|
|
|
def test_leaves_no_staging_file_behind(self):
|
|
self._write("server {}\n")
|
|
self._write("other\n")
|
|
self.assertEqual([p.name for p in self.path.parent.iterdir()], [self.path.name])
|
|
|
|
def _bytes(self):
|
|
return self.path.read_bytes() # nocheck: cache-read rewritten between reads
|
|
|
|
def test_carriage_returns_are_stable_on_the_second_run(self):
|
|
content = "a\r\nb\rc\n"
|
|
self.assertTrue(self._write(content))
|
|
self.assertEqual(self._bytes(), content.encode("utf-8"))
|
|
self.assertFalse(self._write(content))
|
|
|
|
def test_a_text_read_turns_carriage_returns_into_a_change(self):
|
|
self._write("a\r\nb\n")
|
|
self.assertNotEqual(self._content(), "a\r\nb\n")
|
|
|
|
def test_a_non_utf8_file_is_replaced_without_a_crash(self):
|
|
self.path.write_bytes(b"\xff\xfe\xe4 latin-1\n")
|
|
with self.assertRaises(UnicodeDecodeError):
|
|
self._content()
|
|
self.assertTrue(self._write("server {}\n"))
|
|
self.assertEqual(self._bytes(), b"server {}\n")
|
|
self.assertFalse(self._write("server {}\n"))
|
|
|
|
def test_todays_templates_written_as_text_stay_unchanged(self):
|
|
self.assertGreater(len(TODAY_TEMPLATES), 20)
|
|
for source in TODAY_TEMPLATES:
|
|
content = read_text(str(source))
|
|
with self.subTest(template=str(source.relative_to(PROJECT_ROOT))):
|
|
self.path.write_text(content, encoding="utf-8")
|
|
self.assertEqual(self._content(), content)
|
|
self.assertFalse(self._write(content))
|
|
|
|
|
|
def _run_module(main, args: dict) -> dict:
|
|
output = io.StringIO()
|
|
with (
|
|
patch_module_args(args),
|
|
contextlib.redirect_stdout(output),
|
|
contextlib.suppress(SystemExit),
|
|
):
|
|
main()
|
|
return json.loads(output.getvalue())
|
|
|
|
|
|
def _run_main(args: dict) -> dict:
|
|
return _run_module(_MOD.main, args)
|
|
|
|
|
|
class MainTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self._tmp.name)
|
|
self.lua = self.root / "body.lua"
|
|
self.vhost = self.root / "vhost.conf"
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _run(self, lua: str, vhost: str, *, diff=True, check=False) -> dict:
|
|
return _run_main(
|
|
{
|
|
"files": [
|
|
{"path": str(self.lua), "content": lua},
|
|
{"path": str(self.vhost), "content": vhost},
|
|
],
|
|
"mode": "0644",
|
|
"diff_paths": [str(self.lua)],
|
|
"_ansible_diff": diff,
|
|
"_ansible_check_mode": check,
|
|
}
|
|
)
|
|
|
|
def test_a_lua_only_change_is_changed_and_names_only_the_lua(self):
|
|
self._run("a\n", "v\n")
|
|
result = self._run("b\n", "v\n")
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(result["changed_paths"], [str(self.lua)])
|
|
|
|
def test_an_unchanged_run_reports_ok(self):
|
|
self._run("a\n", "v\n")
|
|
result = self._run("a\n", "v\n")
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(result["changed_paths"], [])
|
|
self.assertEqual(result["diff"], [])
|
|
|
|
def test_diff_mode_reports_only_the_listed_path(self):
|
|
self._run("a\n", "v\n")
|
|
result = self._run("b\n", "secret\n")
|
|
self.assertEqual(
|
|
result["diff"],
|
|
[
|
|
{
|
|
"before_header": str(self.lua),
|
|
"after_header": str(self.lua),
|
|
"before": "a\n",
|
|
"after": "b\n",
|
|
}
|
|
],
|
|
)
|
|
|
|
def test_a_new_file_diffs_from_empty(self):
|
|
result = self._run("a\n", "v\n")
|
|
self.assertEqual(result["diff"][0]["before"], "")
|
|
self.assertEqual(result["diff"][0]["after"], "a\n")
|
|
|
|
def test_without_diff_mode_no_content_leaves_the_module(self):
|
|
self._run("a\n", "v\n", diff=False)
|
|
result = self._run("b\n", "w\n", diff=False)
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(result["diff"], [])
|
|
|
|
def test_check_mode_diffs_without_writing(self):
|
|
self._run("a\n", "v\n")
|
|
result = self._run("b\n", "v\n", check=True)
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(result["diff"][0]["after"], "b\n")
|
|
self.assertEqual(
|
|
self.lua.read_text(
|
|
encoding="utf-8"
|
|
), # nocheck: cache-read the file under test changes between reads
|
|
"a\n",
|
|
)
|
|
|
|
def test_carriage_returns_neither_change_nor_diff_on_the_second_run(self):
|
|
self._run("a\r\n", "v\r\n")
|
|
result = self._run("a\r\n", "v\r\n")
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(result["diff"], [])
|
|
|
|
def test_a_non_utf8_file_diffs_with_replacement_characters(self):
|
|
self.lua.write_bytes(b"old \xff\n")
|
|
self.vhost.write_bytes(b"\xe4\n")
|
|
result = self._run("new\n", "v\n")
|
|
self.assertEqual(result["changed_paths"], [str(self.lua), str(self.vhost)])
|
|
self.assertEqual(result["diff"][0]["before"], "old �\n")
|
|
|
|
def test_the_diff_paths_argument_is_required(self):
|
|
result = _run_main(
|
|
{"files": [{"path": str(self.lua), "content": "a\n"}], "mode": "0644"}
|
|
)
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("diff_paths", result["msg"])
|
|
|
|
|
|
class AttributeParityWithTheCopyModuleTests(unittest.TestCase):
|
|
"""The template action ends in the copy module; the fold ends in write_file."""
|
|
|
|
CONTENT = 'ngx.arg[1] = whole or ""\n'
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self._tmp.name)
|
|
self.group = self._assignable_secondary_group()
|
|
|
|
def _assignable_secondary_group(self) -> int:
|
|
probe = self.root / "probe"
|
|
probe.touch()
|
|
for gid in os.getgroups():
|
|
if gid == os.getegid():
|
|
continue
|
|
try:
|
|
os.chown(probe, -1, gid)
|
|
except OSError:
|
|
continue
|
|
return gid
|
|
return os.getegid()
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _twin_dirs(self, name: str, *, setgid: bool) -> tuple[Path, Path]:
|
|
dirs = (self.root / f"{name}-copy", self.root / f"{name}-fold")
|
|
for directory in dirs:
|
|
directory.mkdir()
|
|
if setgid:
|
|
os.chown(directory, -1, self.group)
|
|
directory.chmod(0o2775)
|
|
return dirs
|
|
|
|
def _copy(self, dest: Path) -> None:
|
|
staged = self.root / f"staged-{dest.parent.name}"
|
|
staged.write_text(self.CONTENT, encoding="utf-8")
|
|
result = _run_module(
|
|
copy_module.main,
|
|
{
|
|
"src": str(staged),
|
|
"dest": str(dest),
|
|
"mode": "0644",
|
|
"_original_basename": "body_filter.lua.j2",
|
|
},
|
|
)
|
|
self.assertNotIn("failed", result, result)
|
|
|
|
def _fold(self, dest: Path, mode: str = "0644") -> None:
|
|
with patch_module_args({"files": [], "mode": mode, "diff_paths": []}):
|
|
module = AnsibleModule(
|
|
argument_spec=_MOD.ARGUMENT_SPEC, supports_check_mode=True
|
|
)
|
|
_MOD.write_file(module, str(dest), self.CONTENT, mode)
|
|
|
|
@staticmethod
|
|
def _attributes(path: Path) -> tuple:
|
|
info = path.stat()
|
|
return (
|
|
info.st_uid,
|
|
info.st_gid,
|
|
stat.S_IMODE(info.st_mode),
|
|
path.read_bytes(), # nocheck: cache-read the file under test changes between reads
|
|
)
|
|
|
|
def _assert_parity(self, name: str, *, setgid=False, existing=None) -> None:
|
|
copy_dir, fold_dir = self._twin_dirs(name, setgid=setgid)
|
|
for directory in (copy_dir, fold_dir):
|
|
if existing is not None:
|
|
content, mode, group = existing
|
|
target = directory / "body.lua"
|
|
target.write_text(content, encoding="utf-8")
|
|
os.chown(target, -1, group)
|
|
target.chmod(mode)
|
|
self._copy(copy_dir / "body.lua")
|
|
self._fold(fold_dir / "body.lua")
|
|
self.assertEqual(
|
|
self._attributes(fold_dir / "body.lua"),
|
|
self._attributes(copy_dir / "body.lua"),
|
|
)
|
|
|
|
def test_a_new_file(self):
|
|
self._assert_parity("new")
|
|
|
|
def test_a_new_file_in_a_setgid_directory(self):
|
|
self._assert_parity("setgid", setgid=True)
|
|
|
|
def test_a_changed_file_keeps_its_group_and_gets_the_mode(self):
|
|
self._assert_parity("changed", existing=("old\n", 0o600, self.group))
|
|
|
|
def test_an_unchanged_file_only_gets_the_mode(self):
|
|
self._assert_parity("unchanged", existing=(self.CONTENT, 0o600, self.group))
|
|
|
|
def test_a_mode_difference_is_detected(self):
|
|
copy_dir, fold_dir = self._twin_dirs("control", setgid=False)
|
|
self._copy(copy_dir / "body.lua")
|
|
self._fold(fold_dir / "body.lua", mode="0600")
|
|
self.assertNotEqual(
|
|
self._attributes(fold_dir / "body.lua"),
|
|
self._attributes(copy_dir / "body.lua"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|