mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-07 19:32:17 +00:00
kevinveenbirkenbachandClaude Opus 5.5
73a16bef09
Every app's compose pass ran three remote tasks before compose up: create the directories, compose ps for running containers, and stat the pending marker (53 apps x ~2.7 s in a production run). compose_instance_prepare does all three in one call from 02_stack_host_prep.yml, sharing the directory creation with directory_batch through utils/directory_batch.py; compose ps runs only in compose mode. up.yml evaluates the registered result, named compose_instance_before_files because it predates 04_files: the running check is skipped whenever the compose, env or override file changed in this run (those changes notify compose-up themselves).
Replay harness old vs new: rendered files and the compose-up, marker and swarm handlers identical in every scenario; the only difference is the skipped running check when only the override changed, where compose-up still runs. About 85 s per full production run.
Merge note: store-i18n's a4c7fda974 hoists the removed marker stat into a block; resolve up.yml with this side.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
270 lines
10 KiB
Python
270 lines
10 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.testing import patch_module_args
|
|
|
|
from . import PROJECT_ROOT
|
|
|
|
if TYPE_CHECKING:
|
|
from types import ModuleType
|
|
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from utils import directory_batch
|
|
|
|
MODULE_PATH = PROJECT_ROOT / "library" / "directory_batch.py"
|
|
|
|
|
|
def _load_directory_batch() -> ModuleType:
|
|
sys.modules.setdefault("ansible.module_utils.directory_batch", directory_batch)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"library_directory_batch", 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_directory_batch()
|
|
|
|
|
|
class DirectoryBatchTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self._tmp.name)
|
|
|
|
def tearDown(self):
|
|
for path in self.root.rglob("*"): # nocheck: project-walk
|
|
if path.is_dir() and not path.is_symlink():
|
|
path.chmod(0o700)
|
|
self._tmp.cleanup()
|
|
|
|
def _run(self, args, *, check_mode=False, diff=False):
|
|
buffer = io.StringIO()
|
|
module_args = {**args, "_ansible_check_mode": check_mode, "_ansible_diff": diff}
|
|
with (
|
|
patch_module_args(module_args),
|
|
contextlib.redirect_stdout(buffer),
|
|
self.assertRaises(SystemExit),
|
|
):
|
|
_MOD.main()
|
|
return json.loads(buffer.getvalue())
|
|
|
|
def _path(self, *parts):
|
|
return str(self.root.joinpath(*parts))
|
|
|
|
def _mode(self, path):
|
|
return stat.S_IMODE(Path(path).stat().st_mode)
|
|
|
|
def _mkdir(self, path, mode):
|
|
Path(path).mkdir()
|
|
Path(path).chmod(mode)
|
|
|
|
def test_creates_missing_directory_with_its_mode(self):
|
|
target = self._path("cache")
|
|
result = self._run({"entries": [{"path": target, "mode": "0700"}]})
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(result["changed_paths"], [target])
|
|
self.assertEqual(self._mode(target), 0o700)
|
|
|
|
def test_missing_parents_receive_the_same_mode(self):
|
|
target = self._path("a", "b", "c")
|
|
self._run({"entries": [{"path": target, "mode": "0750"}]})
|
|
self.assertEqual(self._mode(self._path("a")), 0o750)
|
|
self.assertEqual(self._mode(self._path("a", "b")), 0o750)
|
|
self.assertEqual(self._mode(target), 0o750)
|
|
|
|
def test_existing_parent_keeps_its_mode(self):
|
|
self._mkdir(self._path("a"), 0o701)
|
|
self._run({"entries": [{"path": self._path("a", "b"), "mode": "0755"}]})
|
|
self.assertEqual(self._mode(self._path("a")), 0o701)
|
|
|
|
def test_existing_directory_with_same_mode_is_unchanged(self):
|
|
target = self._path("conf.d")
|
|
self._run({"entries": [{"path": target, "mode": "0755"}]})
|
|
result = self._run({"entries": [{"path": target, "mode": "0755"}]})
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(result["changed_paths"], [])
|
|
|
|
def test_mode_drift_is_repaired_and_reported(self):
|
|
target = self._path("conf.d")
|
|
self._run({"entries": [{"path": target, "mode": "0755"}]})
|
|
Path(target).chmod(0o700)
|
|
result = self._run({"entries": [{"path": target, "mode": "0755"}]})
|
|
self.assertEqual(result["changed_paths"], [target])
|
|
self.assertEqual(self._mode(target), 0o755)
|
|
|
|
def test_entry_without_mode_keeps_the_existing_mode(self):
|
|
target = self._path("kept")
|
|
self._mkdir(target, 0o701)
|
|
result = self._run({"entries": [{"path": target}]})
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(self._mode(target), 0o701)
|
|
|
|
def test_only_the_drifted_entry_is_reported(self):
|
|
same = self._path("same")
|
|
drifted = self._path("drifted")
|
|
entries = [{"path": same, "mode": "0755"}, {"path": drifted, "mode": "0755"}]
|
|
self._run({"entries": entries})
|
|
Path(drifted).chmod(0o700)
|
|
self.assertEqual(self._run({"entries": entries})["changed_paths"], [drifted])
|
|
|
|
def test_paths_share_the_top_level_mode(self):
|
|
first = self._path("instance")
|
|
second = self._path("instance", "volumes")
|
|
result = self._run({"paths": [first, second], "mode": "0755"})
|
|
self.assertEqual(result["changed_paths"], [first, second])
|
|
self.assertEqual(self._mode(first), 0o755)
|
|
self.assertEqual(self._mode(second), 0o755)
|
|
|
|
def test_trailing_slash_path_is_created(self):
|
|
target = self._path("servers") + "/"
|
|
result = self._run({"paths": [target], "mode": "0755"})
|
|
self.assertEqual(result["changed_paths"], [target])
|
|
self.assertTrue(Path(target).is_dir())
|
|
|
|
def test_empty_paths_change_nothing(self):
|
|
result = self._run({"paths": [], "mode": "0755"})
|
|
self.assertFalse(result["changed"])
|
|
|
|
def test_check_mode_reports_missing_path_without_creating_it(self):
|
|
target = self._path("new", "child")
|
|
result = self._run({"paths": [target], "mode": "0755"}, check_mode=True)
|
|
self.assertEqual(result["changed_paths"], [target])
|
|
self.assertFalse(Path(self._path("new")).exists())
|
|
|
|
def test_check_mode_reports_mode_drift_without_chmod(self):
|
|
target = self._path("conf.d")
|
|
self._mkdir(target, 0o700)
|
|
result = self._run({"paths": [target], "mode": "0755"}, check_mode=True)
|
|
self.assertEqual(result["changed_paths"], [target])
|
|
self.assertEqual(self._mode(target), 0o700)
|
|
|
|
def test_check_mode_reports_owner_drift_without_chown(self):
|
|
target = self._path("owned")
|
|
Path(target).mkdir()
|
|
foreign_uid = str(Path(target).stat().st_uid + 1)
|
|
result = self._run({"paths": [target], "owner": foreign_uid}, check_mode=True)
|
|
self.assertEqual(result["changed_paths"], [target])
|
|
self.assertEqual(Path(target).stat().st_uid, os.getuid())
|
|
|
|
def test_current_owner_and_group_are_unchanged(self):
|
|
target = self._path("owned")
|
|
Path(target).mkdir()
|
|
info = Path(target).stat()
|
|
result = self._run(
|
|
{
|
|
"entries": [
|
|
{
|
|
"path": target,
|
|
"owner": str(info.st_uid),
|
|
"group": str(info.st_gid),
|
|
}
|
|
]
|
|
}
|
|
)
|
|
self.assertFalse(result["changed"])
|
|
|
|
def test_unknown_owner_fails_loudly(self):
|
|
target = self._path("owned")
|
|
Path(target).mkdir()
|
|
result = self._run({"paths": [target], "owner": "no-such-user-dirbatch"})
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("no-such-user-dirbatch", result["msg"])
|
|
|
|
def test_symlink_to_directory_is_followed(self):
|
|
real = self._path("real")
|
|
link = self._path("link")
|
|
self._mkdir(real, 0o700)
|
|
Path(link).symlink_to(real)
|
|
result = self._run({"paths": [link], "mode": "0755"})
|
|
self.assertEqual(result["changed_paths"], [link])
|
|
self.assertEqual(self._mode(real), 0o755)
|
|
self.assertTrue(Path(link).is_symlink())
|
|
|
|
def test_existing_file_fails_loudly(self):
|
|
target = self._path("nginx.conf")
|
|
Path(target).write_text("events {}\n", encoding="utf-8")
|
|
result = self._run({"paths": [target], "mode": "0755"})
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("already exists as a file", result["msg"])
|
|
|
|
def test_relative_path_fails_loudly(self):
|
|
result = self._run({"paths": ["relative/dir"], "mode": "0755"})
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("not an absolute path", result["msg"])
|
|
|
|
@unittest.skipIf(os.geteuid() == 0, "root ignores directory permissions")
|
|
def test_unwritable_parent_fails_loudly(self):
|
|
parent = self._path("locked")
|
|
self._mkdir(parent, 0o500)
|
|
result = self._run({"paths": [self._path("locked", "child")], "mode": "0755"})
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("Could not create", result["msg"])
|
|
|
|
def test_failure_reports_the_paths_changed_before_it(self):
|
|
created = self._path("created")
|
|
blocker = self._path("blocker")
|
|
Path(blocker).write_text("", encoding="utf-8")
|
|
result = self._run({"paths": [created, blocker], "mode": "0755"})
|
|
self.assertTrue(result["failed"])
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(result["changed_paths"], [created])
|
|
|
|
def test_entries_and_paths_are_mutually_exclusive(self):
|
|
result = self._run(
|
|
{"entries": [{"path": self._path("a")}], "paths": [self._path("b")]}
|
|
)
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("mutually exclusive", result["msg"])
|
|
|
|
def test_entries_and_top_level_mode_are_mutually_exclusive(self):
|
|
result = self._run({"entries": [{"path": self._path("a")}], "mode": "0755"})
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("mutually exclusive", result["msg"])
|
|
|
|
def test_entries_or_paths_is_required(self):
|
|
result = self._run({"mode": "0755"})
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("one of the following is required", result["msg"])
|
|
|
|
def test_unknown_entry_key_is_rejected(self):
|
|
result = self._run({"entries": [{"path": self._path("a"), "recurse": True}]})
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn("recurse", result["msg"])
|
|
|
|
def test_diff_mode_lists_each_changed_path(self):
|
|
target = self._path("conf.d")
|
|
result = self._run({"paths": [target], "mode": "0755"}, diff=True)
|
|
self.assertEqual(len(result["diff"]), 1)
|
|
self.assertEqual(result["diff"][0]["before"]["state"], "absent")
|
|
self.assertEqual(result["diff"][0]["after"]["state"], "directory")
|
|
|
|
def test_diff_mode_records_the_mode_change(self):
|
|
target = self._path("conf.d")
|
|
self._mkdir(target, 0o700)
|
|
result = self._run({"paths": [target], "mode": "0755"}, diff=True)
|
|
self.assertEqual(result["diff"][0]["before"]["mode"], "0700")
|
|
self.assertEqual(result["diff"][0]["after"]["mode"], "0755")
|
|
|
|
def test_no_diff_outside_diff_mode(self):
|
|
result = self._run({"paths": [self._path("conf.d")], "mode": "0755"})
|
|
self.assertNotIn("diff", result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|