mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-10 20:56:39 +00:00
kevinveenbirkenbachandClaude Opus 5.5
a643871997
When the tree check finds differing entries, pack exactly the differing files into one tar archive in source walk order, transfer it once into the remote tmpdir and let the new copy_batch module write each file through atomic_move with the copy module's directory, mode and unsafe_writes semantics. Mode-only drift and leaf directories settle in the same call; unchanged files are not sent. Every shape the regular copy treats differently makes the module refuse before it writes: a symlink, directory or other non-regular entry at a destination, a missing path below a non-directory or a symlink loop, and two source paths reaching one remote entry. The regular copy then runs. Check mode, diff mode, unprivileged become, a dest holding $ and an archive above 16 MiB keep the regular copy. A write failure fails the task and names the file. Executed with ansible-playbook -c local, pipelining on and off: the trees this action leaves equal the builtin copy's, including the symlink fallback; a 6-file changed tree took 2 module calls and 1 transfer instead of 14 and 4. Unit: a flat 15-file tree with every file changed drops from 31 module calls and 15 transfers to 2 and 1. Two dialectic rounds with three independent skeptics; every hole they found is fixed and tested. A third independent review compared end state, changed, check and diff mode against the builtin copy in a real playbook run and found them equal; one deliberate difference remains: a write failing partway reports changed=true with the files already written, where the builtin copy reports changed=false. | 🎯 Fix | 💥 Break | 🌱 Efficiency | ⚡ Speed | 🛡️ Security | 🏆 Score | Gate | | --- | --- | --- | --- | --- | --- | --- | | 90 % | 2 % | 100 % | 15 % (estimated from round trips) | ~100 % | 100 | ⏳ score passes, gate.py not run | Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
947 lines
33 KiB
Python
947 lines
33 KiB
Python
import hashlib
|
|
import os
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import ClassVar
|
|
from unittest.mock import patch
|
|
|
|
from ansible.plugins.action import copy as copy_action
|
|
|
|
from plugins.action.stack_host_copy import (
|
|
ActionModule,
|
|
_walk_dirs_without_bytecode_caches,
|
|
)
|
|
|
|
|
|
class _FakeTask:
|
|
def __init__(self, *, args=None):
|
|
self.args = {} if args is None else dict(args)
|
|
self.async_val = 0
|
|
self.check_mode = False
|
|
self.diff = False
|
|
self.no_log = False
|
|
self.action = "stack_host_copy"
|
|
|
|
def get_search_path(self):
|
|
return []
|
|
|
|
|
|
class _FakeTemplar:
|
|
def __init__(self, *, is_stack_host=True):
|
|
self._is_stack_host = is_stack_host
|
|
self.available_variables: dict[str, object] = {}
|
|
|
|
def template(self, raw):
|
|
return self._is_stack_host
|
|
|
|
|
|
def _make_action(task, templar):
|
|
action = object.__new__(ActionModule)
|
|
action._task = task
|
|
action._templar = templar
|
|
return action
|
|
|
|
|
|
class TestStackHostCopy(unittest.TestCase):
|
|
def test_skips_silently_on_non_stack_host(self):
|
|
task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})
|
|
action = _make_action(task, _FakeTemplar(is_stack_host=False))
|
|
|
|
result = action.run(task_vars={})
|
|
|
|
self.assertTrue(result["skipped"])
|
|
self.assertFalse(result["changed"])
|
|
self.assertIn("IS_STACK_HOST", result["skip_reason"])
|
|
|
|
def test_skips_when_templar_returns_string_false(self):
|
|
task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})
|
|
action = _make_action(task, _FakeTemplar(is_stack_host="False"))
|
|
|
|
result = action.run(task_vars={})
|
|
|
|
self.assertTrue(result["skipped"])
|
|
|
|
def test_delegates_to_copy_on_stack_host(self):
|
|
task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})
|
|
action = _make_action(task, _FakeTemplar(is_stack_host=True))
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": True},
|
|
) as super_run:
|
|
action.run(task_vars={})
|
|
|
|
super_run.assert_called_once()
|
|
|
|
def test_seeds_templar_with_task_vars_before_rendering(self):
|
|
task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})
|
|
templar = _FakeTemplar(is_stack_host=True)
|
|
action = _make_action(task, templar)
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": True},
|
|
):
|
|
action.run(task_vars={"IS_STACK_HOST": True, "ANSIBLE_VERSION": "9.0"})
|
|
|
|
self.assertEqual(templar.available_variables.get("IS_STACK_HOST"), True)
|
|
self.assertEqual(templar.available_variables.get("ANSIBLE_VERSION"), "9.0")
|
|
|
|
def test_defaults_mode_to_0644_when_unset(self):
|
|
task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})
|
|
action = _make_action(task, _FakeTemplar(is_stack_host=True))
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": True},
|
|
):
|
|
action.run(task_vars={})
|
|
|
|
self.assertEqual(task.args["mode"], "0644")
|
|
|
|
def test_keeps_caller_supplied_mode(self):
|
|
task = _FakeTask(
|
|
args={"src": "x", "dest": "/opt/compose/app/x", "mode": "0600"}
|
|
)
|
|
action = _make_action(task, _FakeTemplar(is_stack_host=True))
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": True},
|
|
):
|
|
action.run(task_vars={})
|
|
|
|
self.assertEqual(task.args["mode"], "0600")
|
|
|
|
def test_file_source_never_takes_the_tree_check(self):
|
|
task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})
|
|
action = _make_action(task, _FakeTemplar(is_stack_host=True))
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": True},
|
|
) as super_run:
|
|
result = action.run(task_vars={})
|
|
|
|
super_run.assert_called_once()
|
|
self.assertTrue(result["changed"])
|
|
|
|
def test_is_stack_host_expr_carries_trust_tag(self):
|
|
from ansible.template import is_trusted_as_template
|
|
|
|
from plugins.action.stack_host_copy import _IS_STACK_HOST_EXPR
|
|
|
|
self.assertTrue(
|
|
is_trusted_as_template(_IS_STACK_HOST_EXPR),
|
|
"The IS_STACK_HOST template expression must carry the TrustedAsTemplate "
|
|
"tag; otherwise Ansible 2.19+ returns the literal string unchanged.",
|
|
)
|
|
|
|
|
|
class _FakeShell:
|
|
def __init__(self):
|
|
self.tmpdir = None
|
|
|
|
def join_path(self, *parts):
|
|
return os.path.join(*parts) # noqa: PTH118 - mirrors the sh shell plugin, whose join_path keeps a trailing slash
|
|
|
|
def path_has_trailing_slash(self, path):
|
|
return path.endswith("/")
|
|
|
|
|
|
class _FakeConnection:
|
|
def __init__(self):
|
|
self._shell = _FakeShell()
|
|
self.become = None
|
|
|
|
|
|
class _FakeLoader:
|
|
def get_real_file(self, path, decrypt=True):
|
|
return path
|
|
|
|
def cleanup_tmp_file(self, path):
|
|
return None
|
|
|
|
def path_dwim_relative_stack(self, paths, dirname, needle):
|
|
return needle
|
|
|
|
|
|
class _FakePlayContext:
|
|
no_log = False
|
|
|
|
|
|
def _sha1(data):
|
|
return hashlib.sha1(data).hexdigest() # noqa: S324 - the checksum Ansible copy compares is SHA-1
|
|
|
|
|
|
class _Dir:
|
|
def __init__(self, mode):
|
|
self.mode = mode
|
|
|
|
|
|
class _Link:
|
|
def __init__(self, data):
|
|
self.data = data
|
|
|
|
|
|
def _mode_bits(mode):
|
|
return mode if isinstance(mode, int) else int(mode, 8)
|
|
|
|
|
|
class _FakeRemote:
|
|
def __init__(self, entries):
|
|
self.entries = dict(entries)
|
|
self.calls = []
|
|
self.args = []
|
|
self.transfers = []
|
|
self.fixups = 0
|
|
|
|
def transfer(self, local, remote_path):
|
|
if not remote_path.endswith(".tar"):
|
|
self.transfers.append(Path(local).read_bytes()) # nocheck: cache-read
|
|
return remote_path
|
|
with tarfile.open(local) as bundle:
|
|
self.transfers.append(
|
|
{
|
|
member.name: bundle.extractfile(member).read()
|
|
for member in bundle.getmembers()
|
|
}
|
|
)
|
|
return remote_path
|
|
|
|
def fixup(self, *args, **kwargs):
|
|
self.fixups += 1
|
|
|
|
def __call__(self, module_name=None, module_args=None, task_vars=None, **kwargs):
|
|
self.calls.append(module_name)
|
|
self.args.append(module_args)
|
|
if module_name == "ansible.legacy.file_checksums":
|
|
return {
|
|
"changed": False,
|
|
"states": {p: self._state(p) for p in module_args["paths"]},
|
|
}
|
|
if module_name == "ansible.legacy.copy_batch":
|
|
return self._copy_batch(module_args)
|
|
if module_name == "ansible.legacy.stat":
|
|
entry = self.entries.get(module_args["path"])
|
|
if entry is None:
|
|
return {"stat": {"exists": False}}
|
|
if isinstance(entry, _Dir):
|
|
return {"stat": {"exists": True, "isdir": True}}
|
|
if isinstance(entry, _Link):
|
|
digest = _sha1(entry.data) if module_args["follow"] else ""
|
|
return {
|
|
"stat": {
|
|
"exists": True,
|
|
"isdir": False,
|
|
"islnk": True,
|
|
"checksum": digest,
|
|
}
|
|
}
|
|
return {
|
|
"stat": {
|
|
"exists": True,
|
|
"isdir": False,
|
|
"islnk": False,
|
|
"checksum": _sha1(entry[0]),
|
|
}
|
|
}
|
|
if module_name == "ansible.legacy.file" and module_args["state"] == "directory":
|
|
return self._ensure_directory(module_args["path"], module_args["mode"])
|
|
if module_name == "ansible.legacy.file":
|
|
entry = self.entries[
|
|
module_args["dest"] + module_args["_original_basename"]
|
|
]
|
|
return {"changed": entry[1] != _mode_bits(module_args["mode"])}
|
|
if module_name == "ansible.legacy.copy":
|
|
return {"changed": True}
|
|
raise AssertionError(f"unexpected module {module_name}")
|
|
|
|
def _ensure_directory(self, path, mode):
|
|
entry = self.entries.get(path)
|
|
if entry is None:
|
|
return {"changed": True, "path": path}
|
|
if not isinstance(entry, _Dir):
|
|
return {"failed": True, "msg": f"{path} already exists as a file"}
|
|
return {
|
|
"changed": mode is not None and entry.mode != _mode_bits(mode),
|
|
"path": path,
|
|
}
|
|
|
|
def _copy_batch(self, args):
|
|
dest = args["dest"]
|
|
files = [entry["path"] for entry in args["files"]]
|
|
refused = [
|
|
dest + rel
|
|
for rel in [*files, *args["settled"]]
|
|
if isinstance(self.entries.get(dest + rel), (_Dir, _Link))
|
|
] + [
|
|
dest + rel
|
|
for rel in args["directories"]
|
|
if dest + rel in self.entries
|
|
and not isinstance(self.entries[dest + rel], _Dir)
|
|
]
|
|
if refused:
|
|
return {"changed": False, "refused": refused}
|
|
archive = self.transfers[-1]
|
|
if list(archive) != files or any(
|
|
_sha1(archive[entry["path"]]) != entry["checksum"]
|
|
for entry in args["files"]
|
|
):
|
|
return {"failed": True, "msg": "archive does not match files"}
|
|
mode = _mode_bits(args["mode"])
|
|
for rel in files:
|
|
self.entries[dest + rel] = (archive[rel], mode)
|
|
for rel in args["settled"]:
|
|
self.entries[dest + rel] = (self.entries[dest + rel][0], mode)
|
|
for rel in args["directories"]:
|
|
self.entries[dest + rel] = _Dir(_mode_bits(args["directory_mode"] or 0o755))
|
|
return {"changed": True}
|
|
|
|
def _state(self, path):
|
|
entry = self.entries.get(path)
|
|
if entry is None or isinstance(entry, _Link):
|
|
return {"isreg": False, "isdir": False, "mode": None, "checksum": None}
|
|
if isinstance(entry, _Dir):
|
|
return {
|
|
"isreg": False,
|
|
"isdir": True,
|
|
"mode": f"{entry.mode:04o}",
|
|
"checksum": None,
|
|
}
|
|
return {
|
|
"isreg": True,
|
|
"isdir": False,
|
|
"mode": f"{entry[1]:04o}",
|
|
"checksum": _sha1(entry[0]),
|
|
}
|
|
|
|
|
|
class _TreeHarness(unittest.TestCase):
|
|
FILES: ClassVar[dict[str, bytes]] = {}
|
|
DEST = "/var/www/public_cdn/roles/app/latest/img/design/"
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.src = Path(self._tmp.name)
|
|
for rel, data in self.FILES.items():
|
|
(self.src / rel).parent.mkdir(parents=True, exist_ok=True)
|
|
(self.src / rel).write_bytes(data)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _synced(self, mode=0o644):
|
|
return {self.DEST + rel: (data, mode) for rel, data in self.FILES.items()}
|
|
|
|
def _run(self, remote, *, check_mode=False, diff=False, loader=None, **extra):
|
|
task = _FakeTask(args={"src": f"{self.src}/", "dest": self.DEST, **extra})
|
|
task.check_mode = check_mode
|
|
task.diff = diff
|
|
action = _make_action(task, _FakeTemplar(is_stack_host=True))
|
|
action._connection = _FakeConnection()
|
|
action._loader = _FakeLoader() if loader is None else loader
|
|
action._play_context = _FakePlayContext()
|
|
action._execute_module = remote
|
|
action._transfer_file = getattr(
|
|
remote, "transfer", lambda local, remote_path: remote_path
|
|
)
|
|
action._fixup_perms2 = getattr(remote, "fixup", lambda *a, **k: None)
|
|
action._remove_tmp_path = lambda *a, **k: None
|
|
action._remote_expand_user = lambda path, **k: path
|
|
|
|
def make_tmp_path(*a, **k):
|
|
action._connection._shell.tmpdir = "/remote/tmp"
|
|
return "/remote/tmp"
|
|
|
|
action._make_tmp_path = make_tmp_path
|
|
return action.run(task_vars={})
|
|
|
|
BATCH: ClassVar[list[str]] = [
|
|
"ansible.legacy.file_checksums",
|
|
"ansible.legacy.copy_batch",
|
|
]
|
|
|
|
@staticmethod
|
|
def _batch_args(remote):
|
|
return remote.args[remote.calls.index("ansible.legacy.copy_batch")]
|
|
|
|
@staticmethod
|
|
def _copies(remote):
|
|
return [
|
|
args
|
|
for name, args in zip(remote.calls, remote.args, strict=True)
|
|
if name == "ansible.legacy.copy"
|
|
]
|
|
|
|
|
|
class TestStackHostCopyTree(_TreeHarness):
|
|
FILES: ClassVar[dict[str, bytes]] = {
|
|
"favicon.ico": b"ico",
|
|
"icon.png": b"png",
|
|
"nested/icon.svg": b"<svg/>",
|
|
}
|
|
|
|
def _changed(self):
|
|
entries = self._synced()
|
|
entries[self.DEST + "icon.png"] = (b"old", 0o644)
|
|
return entries
|
|
|
|
def test_unchanged_tree_costs_one_remote_call(self):
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
self.assertEqual(result["dest"], self.DEST)
|
|
self.assertEqual(result["src"], f"{self.src}/")
|
|
|
|
def test_changed_file_travels_alone_in_one_batch(self):
|
|
entries = self._synced()
|
|
entries[self.DEST + "icon.png"] = (b"old", 0o644)
|
|
remote = _FakeRemote(entries)
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
self.assertEqual(remote.transfers, [{"icon.png": b"png"}])
|
|
self.assertEqual(
|
|
self._batch_args(remote)["files"],
|
|
[{"path": "icon.png", "checksum": _sha1(b"png")}],
|
|
)
|
|
self.assertEqual(sorted(self._batch_args(remote)["tree"]), sorted(self.FILES))
|
|
self.assertEqual(remote.entries, self._synced())
|
|
|
|
def test_missing_file_travels_in_one_batch(self):
|
|
entries = self._synced()
|
|
del entries[self.DEST + "nested/icon.svg"]
|
|
remote = _FakeRemote(entries)
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
self.assertEqual(remote.transfers, [{"nested/icon.svg": b"<svg/>"}])
|
|
|
|
def test_mode_drift_is_settled_without_sending_the_file(self):
|
|
entries = self._synced()
|
|
entries[self.DEST + "favicon.ico"] = (b"ico", 0o600)
|
|
remote = _FakeRemote(entries)
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
self.assertEqual(remote.transfers, [{}])
|
|
self.assertEqual(self._batch_args(remote)["settled"], ["favicon.ico"])
|
|
self.assertEqual(self._batch_args(remote)["files"], [])
|
|
|
|
def test_directory_in_place_of_a_file_is_refused_then_copied_regularly(self):
|
|
entries = self._synced()
|
|
entries[self.DEST + "icon.png"] = _Dir(0o755)
|
|
remote = _FakeRemote(entries)
|
|
|
|
self._run(remote)
|
|
|
|
self.assertEqual(remote.calls[:2], self.BATCH)
|
|
self.assertIn("ansible.legacy.stat", remote.calls)
|
|
self.assertIn("ansible.legacy.copy", remote.calls)
|
|
|
|
def test_symlink_in_place_of_a_file_is_refused_then_copied_regularly(self):
|
|
entries = self._synced()
|
|
entries[self.DEST + "icon.png"] = _Link(b"png")
|
|
remote = _FakeRemote(entries)
|
|
|
|
self._run(remote)
|
|
|
|
self.assertEqual(remote.calls[:2], self.BATCH)
|
|
copies = self._copies(remote)
|
|
self.assertEqual([args["_original_basename"] for args in copies], ["icon.png"])
|
|
self.assertFalse(copies[0]["follow"])
|
|
|
|
def test_check_mode_keeps_the_regular_copy(self):
|
|
remote = _FakeRemote(self._changed())
|
|
|
|
result = self._run(remote, check_mode=True)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertNotIn("ansible.legacy.copy_batch", remote.calls)
|
|
self.assertNotIn("ansible.legacy.copy", remote.calls)
|
|
self.assertIn("ansible.legacy.stat", remote.calls)
|
|
self.assertEqual(remote.transfers, [])
|
|
|
|
def test_diff_mode_keeps_the_regular_copy(self):
|
|
remote = _FakeRemote(self._changed())
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": True, "diff": [{"after": "png"}]},
|
|
) as super_run:
|
|
result = self._run(remote, diff=True)
|
|
|
|
super_run.assert_called_once()
|
|
self.assertEqual(result["diff"], [{"after": "png"}])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
self.assertEqual(remote.transfers, [])
|
|
|
|
def test_failed_batch_stays_failed_without_a_second_copy(self):
|
|
failure = {
|
|
"failed": True,
|
|
"changed": True,
|
|
"msg": f"Failed to write {self.DEST}icon.png: boom",
|
|
}
|
|
|
|
class _FailingBatch(_FakeRemote):
|
|
def _copy_batch(self, args):
|
|
return dict(failure)
|
|
|
|
remote = _FailingBatch(self._changed())
|
|
|
|
with patch("plugins.action.stack_host_copy.CopyActionModule.run") as super_run:
|
|
result = self._run(remote)
|
|
|
|
super_run.assert_not_called()
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn(f"{self.DEST}icon.png", result["msg"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
|
|
def test_batch_result_has_the_keys_of_the_regular_copy(self):
|
|
batch = self._run(_FakeRemote(self._changed()))
|
|
with patch.object(ActionModule, "_write_tree", return_value=None):
|
|
regular = self._run(_FakeRemote(self._changed()))
|
|
|
|
self.assertEqual(set(batch), set(regular))
|
|
self.assertEqual(
|
|
{key: batch[key] for key in ("changed", "dest", "src")},
|
|
{key: regular[key] for key in ("changed", "dest", "src")},
|
|
)
|
|
|
|
def test_unprivileged_become_keeps_the_regular_copy(self):
|
|
remote = _FakeRemote(self._changed())
|
|
|
|
with patch.object(ActionModule, "_is_become_unprivileged", return_value=True):
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertNotIn("ansible.legacy.copy_batch", remote.calls)
|
|
self.assertIn("ansible.legacy.copy", remote.calls)
|
|
|
|
def test_dest_with_a_variable_skips_the_tree_check(self):
|
|
remote = _FakeRemote(self._changed())
|
|
task = _FakeTask(args={"src": f"{self.src}/", "dest": "/srv/$APP/img/"})
|
|
action = _make_action(task, _FakeTemplar(is_stack_host=True))
|
|
action._execute_module = remote
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": True},
|
|
) as super_run:
|
|
action.run(task_vars={})
|
|
|
|
super_run.assert_called_once()
|
|
self.assertEqual(remote.calls, [])
|
|
|
|
def test_archive_over_the_limit_keeps_the_regular_copy(self):
|
|
entries = self._synced()
|
|
entries[self.DEST + "icon.png"] = (b"old", 0o644)
|
|
entries[self.DEST + "favicon.ico"] = (b"old", 0o644)
|
|
remote = _FakeRemote(entries)
|
|
|
|
with patch("plugins.action.stack_host_copy._ARCHIVE_LIMIT", 5):
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertNotIn("ansible.legacy.copy_batch", remote.calls)
|
|
self.assertEqual(sorted(remote.transfers), [b"ico", b"png"])
|
|
|
|
def test_files_travel_in_the_order_the_copy_action_walks(self):
|
|
remote = _FakeRemote({})
|
|
|
|
self._run(remote)
|
|
|
|
walked = [
|
|
rel
|
|
for _, rel in _walk_dirs_without_bytecode_caches(
|
|
f"{self.src}/", local_follow=True
|
|
)["files"]
|
|
]
|
|
sent = [entry["path"] for entry in self._batch_args(remote)["files"]]
|
|
self.assertEqual(sent, walked)
|
|
self.assertEqual(list(remote.transfers[0]), walked)
|
|
|
|
def test_checksum_is_taken_from_the_packed_bytes(self):
|
|
remote = _FakeRemote(self._changed())
|
|
icon = self.src / "icon.png"
|
|
loader = _FakeLoader()
|
|
reads = []
|
|
|
|
def get_real_file(path, decrypt=True):
|
|
reads.append(path)
|
|
if path == str(icon) and reads.count(path) == 2:
|
|
icon.write_bytes(b"edited")
|
|
return path
|
|
|
|
loader.get_real_file = get_real_file
|
|
|
|
self._run(remote, loader=loader)
|
|
|
|
self.assertEqual(remote.transfers, [{"icon.png": b"edited"}])
|
|
self.assertEqual(
|
|
self._batch_args(remote)["files"],
|
|
[{"path": "icon.png", "checksum": _sha1(b"edited")}],
|
|
)
|
|
|
|
def test_requested_mode_is_what_the_tree_check_compares(self):
|
|
remote = _FakeRemote(self._synced(mode=0o600))
|
|
|
|
result = self._run(remote, mode="0600")
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
|
|
def test_attribute_args_skip_the_tree_check(self):
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
self._run(remote, owner="root")
|
|
|
|
self.assertNotIn("ansible.legacy.file_checksums", remote.calls)
|
|
self.assertIn("ansible.legacy.stat", remote.calls)
|
|
|
|
def test_symbolic_mode_skips_the_tree_check(self):
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": False},
|
|
) as super_run:
|
|
self._run(remote, mode="u=rw,g=r,o=r")
|
|
|
|
super_run.assert_called_once()
|
|
self.assertEqual(remote.calls, [])
|
|
|
|
def test_failed_tree_check_falls_back_to_the_regular_copy(self):
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
def failing(module_name=None, **kwargs):
|
|
if module_name == "ansible.legacy.file_checksums":
|
|
remote.calls.append(module_name)
|
|
return {"failed": True, "msg": "boom"}
|
|
return remote(module_name=module_name, **kwargs)
|
|
|
|
result = self._run(failing)
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertIn("ansible.legacy.stat", remote.calls)
|
|
|
|
def test_source_symlink_skips_the_tree_check(self):
|
|
(self.src / "dangling").symlink_to("/nonexistent-target")
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": False},
|
|
) as super_run:
|
|
self._run(remote, local_follow=False)
|
|
|
|
super_run.assert_called_once()
|
|
self.assertEqual(remote.calls, [])
|
|
|
|
LEAVES: ClassVar[tuple[str, ...]] = (
|
|
".claude",
|
|
".claude/.cc-writes",
|
|
"nested/.claude",
|
|
"nested/.claude/.cc-writes",
|
|
)
|
|
|
|
def _add_empty_directories(self):
|
|
for rel in self.LEAVES:
|
|
(self.src / rel).mkdir(parents=True, exist_ok=True)
|
|
|
|
def _with_leaves(self, mode):
|
|
entries = self._synced()
|
|
entries.update({self.DEST + rel: _Dir(mode) for rel in self.LEAVES})
|
|
return entries
|
|
|
|
def _leaf_paths(self):
|
|
return sorted(self.DEST + rel for rel in self.LEAVES)
|
|
|
|
@staticmethod
|
|
def _directory_calls(remote):
|
|
return sorted(
|
|
args["path"]
|
|
for name, args in zip(remote.calls, remote.args, strict=True)
|
|
if name == "ansible.legacy.file" and args["state"] == "directory"
|
|
)
|
|
|
|
def test_existing_empty_directories_settle_in_one_remote_call(self):
|
|
self._add_empty_directories()
|
|
remote = _FakeRemote(self._with_leaves(0o755))
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
self.assertEqual(
|
|
sorted(remote.args[0]["paths"]),
|
|
sorted([*self._synced(), *self._leaf_paths()]),
|
|
)
|
|
|
|
def test_unset_directory_mode_accepts_any_remote_directory_mode(self):
|
|
self._add_empty_directories()
|
|
remote = _FakeRemote(self._with_leaves(0o700))
|
|
regular = _FakeRemote(self._with_leaves(0o700))
|
|
|
|
settled = self._run(remote)
|
|
with patch.object(ActionModule, "_settled_tree", return_value=None):
|
|
copied = self._run(regular)
|
|
|
|
self.assertFalse(settled["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
self.assertFalse(copied["changed"])
|
|
self.assertEqual(self._directory_calls(regular), self._leaf_paths())
|
|
|
|
def test_requested_directory_mode_settles_in_one_remote_call(self):
|
|
self._add_empty_directories()
|
|
remote = _FakeRemote(self._with_leaves(0o750))
|
|
|
|
result = self._run(remote, directory_mode="0750")
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
|
|
def test_missing_remote_directory_is_ensured_in_one_batch(self):
|
|
self._add_empty_directories()
|
|
entries = self._with_leaves(0o755)
|
|
del entries[self.DEST + "nested/.claude/.cc-writes"]
|
|
remote = _FakeRemote(entries)
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
self.assertEqual(
|
|
self._batch_args(remote)["directories"], ["nested/.claude/.cc-writes"]
|
|
)
|
|
self.assertIsNone(self._batch_args(remote)["directory_mode"])
|
|
|
|
def test_directory_mode_drift_is_settled_in_one_batch(self):
|
|
self._add_empty_directories()
|
|
remote = _FakeRemote(self._with_leaves(0o755))
|
|
|
|
result = self._run(remote, directory_mode="0750")
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
self.assertEqual(self._batch_args(remote)["directories"], sorted(self.LEAVES))
|
|
self.assertEqual(self._batch_args(remote)["directory_mode"], "0750")
|
|
|
|
def test_file_in_place_of_a_directory_is_refused_then_fails_regularly(self):
|
|
self._add_empty_directories()
|
|
entries = self._with_leaves(0o755)
|
|
entries[self.DEST + ".claude/.cc-writes"] = (b"", 0o644)
|
|
remote = _FakeRemote(entries)
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertEqual(remote.calls[:2], self.BATCH)
|
|
self.assertTrue(result["failed"])
|
|
self.assertIn(self.DEST + ".claude/.cc-writes", self._directory_calls(remote))
|
|
|
|
def test_symbolic_directory_mode_with_empty_directories_skips_the_tree_check(self):
|
|
self._add_empty_directories()
|
|
remote = _FakeRemote(self._with_leaves(0o755))
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
return_value={"changed": False},
|
|
) as super_run:
|
|
self._run(remote, directory_mode="u=rwx,g=rx,o=")
|
|
|
|
super_run.assert_called_once()
|
|
self.assertEqual(remote.calls, [])
|
|
|
|
def test_symbolic_directory_mode_without_empty_directories_keeps_the_check(self):
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
result = self._run(remote, directory_mode="u=rwx,g=rx,o=")
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
|
|
def test_integer_mode_is_compared_as_octal(self):
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
result = self._run(remote, mode=0o644)
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
|
|
def _add_bytecode_caches(self):
|
|
for rel in (
|
|
"__pycache__/mod.cpython-314.pyc",
|
|
"nested/__pycache__/mod.cpython-313.pyc",
|
|
):
|
|
(self.src / rel).parent.mkdir(parents=True, exist_ok=True)
|
|
(self.src / rel).write_bytes(b"bytecode")
|
|
|
|
def test_bytecode_caches_stay_out_of_the_tree_check(self):
|
|
self._add_bytecode_caches()
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertFalse(result["changed"])
|
|
self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
|
|
self.assertEqual(sorted(remote.args[0]["paths"]), sorted(self._synced()))
|
|
|
|
def test_bytecode_caches_are_not_transferred(self):
|
|
self._add_bytecode_caches()
|
|
entries = self._synced()
|
|
entries[self.DEST + "icon.png"] = (b"old", 0o644)
|
|
remote = _FakeRemote(entries)
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
touched = [str(value) for args in remote.args for value in args.values()]
|
|
self.assertEqual([value for value in touched if "__pycache__" in value], [])
|
|
self.assertEqual(remote.transfers, [{"icon.png": b"png"}])
|
|
|
|
def test_bytecode_outside_a_cache_directory_stays_in_the_copy(self):
|
|
(self.src / "legacy.pyc").write_bytes(b"bytecode")
|
|
remote = _FakeRemote(self._synced())
|
|
|
|
result = self._run(remote)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertIn(self.DEST + "legacy.pyc", remote.args[0]["paths"])
|
|
self.assertEqual(remote.transfers, [{"legacy.pyc": b"bytecode"}])
|
|
|
|
def test_symlinked_root_passes_the_cache_filter_unchanged(self):
|
|
with tempfile.TemporaryDirectory() as links:
|
|
root = Path(links) / "root"
|
|
root.symlink_to(self.src)
|
|
|
|
filtered = _walk_dirs_without_bytecode_caches(str(root), local_follow=False)
|
|
|
|
self.assertEqual(filtered["symlinks"], (str(self.src), "root"))
|
|
|
|
|
|
class TestStackHostCopyFlatTree(_TreeHarness):
|
|
FILES: ClassVar[dict[str, bytes]] = {
|
|
f"slot-{index:02d}.png": f"palette-2-{index}".encode() for index in range(15)
|
|
}
|
|
|
|
def _stale(self, names):
|
|
entries = self._synced()
|
|
entries.update({self.DEST + name: (b"palette-1", 0o644) for name in names})
|
|
return entries
|
|
|
|
def _round_trips(self, entries, *, regular):
|
|
remote = _FakeRemote(entries)
|
|
if regular:
|
|
with patch.object(ActionModule, "_write_tree", return_value=None):
|
|
result = self._run(remote)
|
|
else:
|
|
result = self._run(remote)
|
|
self.assertTrue(result["changed"])
|
|
return {
|
|
"modules": len(remote.calls),
|
|
"transfers": len(remote.transfers),
|
|
"fixups": remote.fixups,
|
|
}
|
|
|
|
def test_all_files_changed_cost_one_transfer_and_two_module_calls(self):
|
|
stale = self._stale(self.FILES)
|
|
|
|
before = self._round_trips(stale, regular=True)
|
|
after = self._round_trips(stale, regular=False)
|
|
|
|
self.assertEqual(before, {"modules": 31, "transfers": 15, "fixups": 15})
|
|
self.assertEqual(after, {"modules": 2, "transfers": 1, "fixups": 1})
|
|
|
|
def test_one_changed_file_costs_one_transfer_and_two_module_calls(self):
|
|
stale = self._stale(["slot-07.png"])
|
|
|
|
before = self._round_trips(stale, regular=True)
|
|
after = self._round_trips(stale, regular=False)
|
|
|
|
self.assertEqual(before, {"modules": 31, "transfers": 1, "fixups": 1})
|
|
self.assertEqual(after, {"modules": 2, "transfers": 1, "fixups": 1})
|
|
|
|
def test_only_the_changed_file_is_sent(self):
|
|
remote = _FakeRemote(self._stale(["slot-07.png"]))
|
|
|
|
self._run(remote)
|
|
|
|
self.assertEqual(remote.transfers, [{"slot-07.png": b"palette-2-7"}])
|
|
self.assertEqual(remote.entries, self._synced())
|
|
|
|
def test_follow_without_symlinks_takes_the_batch(self):
|
|
remote = _FakeRemote(self._stale(["slot-00.png", "slot-01.png"]))
|
|
|
|
result = self._run(remote, follow=True)
|
|
|
|
self.assertTrue(result["changed"])
|
|
self.assertEqual(remote.calls, self.BATCH)
|
|
|
|
def test_follow_onto_a_symlinked_file_keeps_the_copy_semantics(self):
|
|
entries = self._synced()
|
|
entries[self.DEST + "slot-00.png"] = _Link(b"elsewhere")
|
|
remote = _FakeRemote(entries)
|
|
|
|
self._run(remote, follow=True)
|
|
|
|
self.assertEqual(remote.calls[:2], self.BATCH)
|
|
copies = self._copies(remote)
|
|
self.assertEqual(
|
|
[args["_original_basename"] for args in copies], ["slot-00.png"]
|
|
)
|
|
self.assertTrue(copies[0]["follow"])
|
|
|
|
|
|
class TestStackHostCopyWalkSwap(unittest.TestCase):
|
|
def _action(self):
|
|
task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})
|
|
return _make_action(task, _FakeTemplar(is_stack_host=True))
|
|
|
|
def test_copy_action_walks_without_caches_only_during_the_run(self):
|
|
original = copy_action._walk_dirs
|
|
seen = []
|
|
|
|
def copy_run(**kwargs):
|
|
seen.append(copy_action._walk_dirs)
|
|
return {"changed": False}
|
|
|
|
with patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
side_effect=copy_run,
|
|
):
|
|
self._action().run(task_vars={})
|
|
|
|
self.assertEqual(seen, [_walk_dirs_without_bytecode_caches])
|
|
self.assertIs(copy_action._walk_dirs, original)
|
|
|
|
def test_copy_action_walk_is_restored_when_the_copy_raises(self):
|
|
original = copy_action._walk_dirs
|
|
|
|
with (
|
|
patch(
|
|
"plugins.action.stack_host_copy.CopyActionModule.run",
|
|
side_effect=RuntimeError("boom"),
|
|
),
|
|
self.assertRaises(RuntimeError),
|
|
):
|
|
self._action().run(task_vars={})
|
|
|
|
self.assertIs(copy_action._walk_dirs, original)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|