File size: 6,741 Bytes
e479c46 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 | # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Unit tests for ``fast_copy_tree``.
The helper is exercised against ``tmp_path`` so the tests are hermetic
and don't depend on a real ``/shared`` NFS mount. Each test pins one
behavioural property: bulk parity with ``shutil.copytree``, symlink
dereference vs preserve, dirs-exist merging, ``tar``-missing fallback,
and the not-a-directory contract.
"""
from __future__ import annotations
import pathlib
import shutil
import pytest
from test_support import runtime
def _materialise_tree(root: pathlib.Path) -> dict[str, bytes]:
"""Create a small mixed tree under *root*; return path → content map."""
files: dict[str, bytes] = {
"top.txt": b"hello\n",
"pkg/__init__.py": b"",
"pkg/sub/data.bin": b"\x00\x01\x02" * 100,
"pkg/sub/empty": b"",
}
for rel, content in files.items():
path = root / rel
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(content)
return files
def _read_tree(root: pathlib.Path) -> dict[str, bytes]:
"""Return path → content map of all regular files under *root*."""
out: dict[str, bytes] = {}
for path in sorted(root.rglob("*")):
if path.is_file() and not path.is_symlink():
out[str(path.relative_to(root))] = path.read_bytes()
return out
def test_copies_files_and_subdirs(tmp_path):
"""Round-trip: bytes and tree shape match the source after copy."""
src = tmp_path / "src"
expected = _materialise_tree(src)
dst = tmp_path / "dst"
runtime.fast_copy_tree(src, dst)
assert _read_tree(dst) == expected
def test_dereferences_symlinks_by_default(tmp_path):
"""``symlinks=False`` (default) must copy real bytes, not symlinks.
Load-bearing for staged ``/shared`` trees: leaving a symlink in the
staged copy would route subsequent reads back over NFS and defeat
the purpose of staging.
"""
src = tmp_path / "src"
src.mkdir()
target = src / "real.bin"
target.write_bytes(b"payload")
(src / "link.bin").symlink_to(target)
dst = tmp_path / "dst"
runtime.fast_copy_tree(src, dst)
copied_link = dst / "link.bin"
assert copied_link.exists()
assert not copied_link.is_symlink()
assert copied_link.read_bytes() == b"payload"
def test_preserves_symlinks_when_requested(tmp_path):
"""``symlinks=True`` must keep symlinks verbatim (libero venv pattern)."""
src = tmp_path / "src"
src.mkdir()
target = src / "real.bin"
target.write_bytes(b"payload")
(src / "link.bin").symlink_to(target)
dst = tmp_path / "dst"
runtime.fast_copy_tree(src, dst, symlinks=True)
copied_link = dst / "link.bin"
assert copied_link.is_symlink()
def test_merges_into_existing_destination(tmp_path):
"""Existing dst entries are preserved; same-name entries overwritten.
Matches ``shutil.copytree(dirs_exist_ok=True)`` semantics so the new
helper is a drop-in replacement for the prior call sites.
"""
src = tmp_path / "src"
src.mkdir()
(src / "shared.txt").write_bytes(b"new")
(src / "src_only.txt").write_bytes(b"src")
dst = tmp_path / "dst"
dst.mkdir()
(dst / "shared.txt").write_bytes(b"old")
(dst / "dst_only.txt").write_bytes(b"dst")
runtime.fast_copy_tree(src, dst)
assert (dst / "shared.txt").read_bytes() == b"new"
assert (dst / "src_only.txt").read_bytes() == b"src"
assert (dst / "dst_only.txt").read_bytes() == b"dst"
def test_falls_back_to_copytree_when_tar_missing(monkeypatch, tmp_path):
"""When ``tar`` is unavailable we fall back to ``shutil.copytree``.
Asserted via a monkeypatched ``shutil.which`` so the test is
hermetic regardless of host PATH.
"""
src = tmp_path / "src"
expected = _materialise_tree(src)
dst = tmp_path / "dst"
monkeypatch.setattr(runtime.shutil, "which", lambda _: None)
called: list[tuple[str, str]] = []
real_copytree = shutil.copytree
def _spy_copytree(s, d, *args, **kwargs):
# Only the top-level call records — copytree() recurses into
# itself for subdirs, so filter to the call we initiated.
if str(s) == str(src):
called.append((str(s), str(d)))
return real_copytree(s, d, *args, **kwargs)
monkeypatch.setattr(runtime.shutil, "copytree", _spy_copytree)
runtime.fast_copy_tree(src, dst)
assert called == [(str(src), str(dst))]
assert _read_tree(dst) == expected
def test_rejects_non_directory_source(tmp_path):
"""Helper rejects file/non-existent sources up-front rather than mid-pipe.
Prevents the more confusing failure mode where ``tar`` would error
deep inside the pipe and we'd surface a generic non-zero rc.
"""
not_a_dir = tmp_path / "not_a_dir.txt"
not_a_dir.write_bytes(b"")
with pytest.raises(NotADirectoryError):
runtime.fast_copy_tree(not_a_dir, tmp_path / "dst")
def test_reaps_src_proc_when_dst_popen_fails(monkeypatch, tmp_path):
"""If the second Popen raises, src_proc must not be left as a zombie.
Pre-fix, the only cleanup was closing the pipe fd; src_proc kept
running with no reader and was never `wait()`-ed. Under repeated
failures (CI memory pressure) this accumulated zombies.
"""
src = tmp_path / "src"
_materialise_tree(src)
dst = tmp_path / "dst"
real_popen = runtime.subprocess.Popen
started: list[runtime.subprocess.Popen] = []
call_count = {"n": 0}
def _failing_popen(cmd, *args, **kwargs):
call_count["n"] += 1
if call_count["n"] == 1:
proc = real_popen(cmd, *args, **kwargs)
started.append(proc)
return proc
raise OSError("simulated dst Popen failure")
monkeypatch.setattr(runtime.subprocess, "Popen", _failing_popen)
with pytest.raises(OSError, match="simulated dst Popen failure"):
runtime.fast_copy_tree(src, dst)
assert len(started) == 1
src_proc = started[0]
assert src_proc.returncode is not None, "src_proc was not reaped after dst Popen failure"
|