Jan Lindemann
b697b7b754
All checks were successful
CI / Packaging - Kali Linux (pull_request) Successful in 4m51s
CI / Packaging - OpenSUSE Tumbleweed (pull_request) Successful in 4m42s
CI / Packaging test (pull_request) Successful in 0s
CI / Packaging - Kali Linux (push) Successful in 4m53s
CI / Packaging - OpenSUSE Tumbleweed (push) Successful in 4m46s
CI / Packaging test (push) Successful in 0s
copy() copies a single file: it fetches the source root as one content and puts it at the destination, so addressing a directory source fails, and there is no way to mirror a tree between file contexts. Add a recursive mode: - a directory source is mirrored below the destination root, into it if the root is an existing directory, as it if it does not exist yet - entries are enumerated with readdir(), destination directories are created level by level, and every leaf is put with the requested owner, group, and mode - a file source is unaffected, and the destination parent must exist, as before posix copy exposes the mode as -r/--recursive; the help test reference picks up the new option. Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v0.87.1 Signed-off-by: Jan Lindemann <jan@janware.com>
115 lines
4 KiB
Python
115 lines
4 KiB
Python
import asyncio
|
|
import os
|
|
import stat
|
|
import tempfile
|
|
|
|
from jw.pkg.lib.util import copy
|
|
|
|
def uri(path: str) -> str:
|
|
return 'local://localhost' + path
|
|
|
|
async def run_tests() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
# -- Source tree
|
|
#
|
|
# src/a/f1
|
|
# src/a/sub/f2
|
|
# src/a/sub/subsub/f3
|
|
# src/a/empty
|
|
# src/single
|
|
for rel, content in [
|
|
('src/a/f1', 'one\n'),
|
|
('src/a/sub/f2', 'two\n'),
|
|
('src/a/sub/subsub/f3', 'three\n'),
|
|
('src/single', 'solo\n'),
|
|
]:
|
|
path = os.path.join(tmp, rel)
|
|
os.makedirs(os.path.dirname(path), exist_ok = True)
|
|
with open(path, 'w') as f:
|
|
f.write(content)
|
|
os.makedirs(os.path.join(tmp, 'src/a/empty'))
|
|
|
|
# -- A single file to a nonexistent destination path; like cp,
|
|
# the destination's parent must exist
|
|
os.makedirs(os.path.join(tmp, 'dst'))
|
|
ret = await copy(
|
|
uri(os.path.join(tmp, 'src/single')), uri(os.path.join(tmp, 'dst/single'))
|
|
)
|
|
assert ret == os.path.join(tmp, 'dst/single'), ret
|
|
with open(os.path.join(tmp, 'dst/single')) as f:
|
|
assert f.read() == 'solo\n'
|
|
|
|
# -- A single file into an existing directory gets the basename
|
|
os.makedirs(os.path.join(tmp, 'dst/dir'))
|
|
ret = await copy(
|
|
uri(os.path.join(tmp, 'src/single')), uri(os.path.join(tmp, 'dst/dir'))
|
|
)
|
|
assert ret == os.path.join(tmp, 'dst/dir/single'), ret
|
|
with open(os.path.join(tmp, 'dst/dir/single')) as f:
|
|
assert f.read() == 'solo\n'
|
|
|
|
# -- A file with recursive=True behaves like a plain file copy
|
|
ret = await copy(
|
|
uri(os.path.join(tmp, 'src/single')),
|
|
uri(os.path.join(tmp, 'dst/recursive-file')),
|
|
recursive = True,
|
|
)
|
|
assert ret == os.path.join(tmp, 'dst/recursive-file'), ret
|
|
|
|
# -- A tree to a nonexistent destination: the destination becomes
|
|
# the tree, parent directories are created, empty directories
|
|
# survive, and the returned paths are the copied leaves
|
|
dst_tree = os.path.join(tmp, 'dst', 'tree')
|
|
ret = await copy(
|
|
uri(os.path.join(tmp, 'src/a')), uri(dst_tree), recursive = True
|
|
)
|
|
assert isinstance(ret, list), ret
|
|
assert sorted(ret) == sorted([
|
|
os.path.join(dst_tree, 'f1'),
|
|
os.path.join(dst_tree, 'sub/f2'),
|
|
os.path.join(dst_tree, 'sub/subsub/f3'),
|
|
]), ret
|
|
for rel, content in [
|
|
('f1', 'one\n'),
|
|
('sub/f2', 'two\n'),
|
|
('sub/subsub/f3', 'three\n'),
|
|
]:
|
|
with open(os.path.join(dst_tree, rel)) as f:
|
|
assert f.read() == content, rel
|
|
assert os.path.isdir(os.path.join(dst_tree, 'empty'))
|
|
|
|
# -- A tree into an existing directory is copied into it under
|
|
# the source basename
|
|
os.makedirs(os.path.join(tmp, 'dst', 'into'))
|
|
ret = await copy(
|
|
uri(os.path.join(tmp, 'src/a')),
|
|
uri(os.path.join(tmp, 'dst/into')),
|
|
recursive = True,
|
|
)
|
|
assert os.path.isfile(os.path.join(tmp, 'dst/into/a/sub/f2')), ret
|
|
|
|
# -- A mode applies to every copied leaf
|
|
dst_mode = os.path.join(tmp, 'dst', 'mode')
|
|
await copy(
|
|
uri(os.path.join(tmp, 'src/a')),
|
|
uri(dst_mode),
|
|
mode = 0o640,
|
|
recursive = True
|
|
)
|
|
st = os.stat(os.path.join(dst_mode, 'sub/subsub/f3'))
|
|
assert stat.S_IMODE(st.st_mode) == 0o640, st
|
|
|
|
# -- A missing source fails loudly
|
|
try:
|
|
await copy(
|
|
uri(os.path.join(tmp, 'src/missing')), uri(os.path.join(tmp, 'dst/x'))
|
|
)
|
|
raise AssertionError('expected an exception')
|
|
except Exception:
|
|
pass
|
|
|
|
def main() -> None:
|
|
asyncio.run(run_tests())
|
|
print('All util copy tests passed')
|
|
|
|
main()
|